perf(schema): skip meta-schema validation for Pydantic-generated schemas

src/jacobian/schema_registry.py

@@ -92,6 +92,58 @@ def _validated_schema(canonical_schema: bytes) -> Draft202012Validator:
     return Draft202012Validator(normalized, format_checker=FormatChecker())
 
 
+@lru_cache(maxsize=1024)
+def _trusted_validated_schema(canonical_schema: bytes) -> Draft202012Validator:
+    """Compile a schema without re-validating against the meta-schema.
+
+    Pydantic's model_json_schema() produces valid Draft 2020-12 schemas
+    by construction.  Skipping the expensive meta-schema walk for those
+    schemas eliminates the dominant cost of runtime construction (~6.7s
+    across 527 unique schemas at ~12.7ms each).  External references are
+    still rejected; only the meta-schema check is skipped.
+    """
+
+    normalized = loads_strict_json(canonical_schema)
+    _reject_external_references(normalized)
+    return Draft202012Validator(normalized, format_checker=FormatChecker())
+
+
+def _uses_operator_owned_pydantic_schema(model: type[BaseModel]) -> bool:
+    """Return whether Pydantic owns the complete model schema generation path."""
+
+    model_schema_method = getattr(model.model_json_schema, "__func__", None)
+    base_schema_method = getattr(BaseModel.model_json_schema, "__func__", None)
+    model_schema_hook = getattr(model.__get_pydantic_json_schema__, "__func__", None)
+    base_schema_hook = getattr(BaseModel.__get_pydantic_json_schema__, "__func__", None)
+    return (
+        model.__module__.startswith("jacobian.")
+        and model_schema_method is base_schema_method
+        and model_schema_hook is base_schema_hook
+        and model.model_config.get("json_schema_extra") is None
+        and _core_schema_uses_only_pydantic_defaults(model.__pydantic_core_schema__)
+    )
+
+
+def _core_schema_uses_only_pydantic_defaults(value: Any) -> bool:
+    if isinstance(value, list | tuple):
+        return all(_core_schema_uses_only_pydantic_defaults(item) for item in value)
+    if not isinstance(value, dict):
+        return True
+    if value.get("pydantic_js_extra") is not None or value.get(
+        "pydantic_js_annotation_functions"
+    ):
+        return False
+    base_schema_hook = getattr(BaseModel.__get_pydantic_json_schema__, "__func__", None)
+    for hook in value.get("pydantic_js_functions", ()):
+        function = getattr(hook, "__func__", hook)
+        module = getattr(function, "__module__", "")
+        if function is not base_schema_hook and not module.startswith("pydantic."):
+            return False
+    return all(
+        _core_schema_uses_only_pydantic_defaults(item) for item in value.values()
+    )
+
+
 class SchemaRegistry:
     """Store and apply closed local JSON Schemas used by artifact contracts."""
 
@@ -103,9 +155,32 @@ def __init__(self, store: ArtifactRepository) -> None:
         self._registrations: dict[tuple[str, str, bytes], str] = {}
         self._pending: dict[int, _PendingRegistrations] = {}
 
-    def register(self, *, name: str, version: str, schema: dict[str, Any]) -> str:
+    def register(
+        self,
+        *,
+        name: str,
+        version: str,
+        schema: dict[str, Any],
+    ) -> str:
         """Register a schema after rejecting unsupported external references."""
 
+        return self._register(
+            name=name,
+            version=version,
+            schema=schema,
+            skip_meta_validation=False,
+        )
+
+    def _register(
+        self,
+        *,
+        name: str,
+        version: str,
+        schema: dict[str, Any],
+        skip_meta_validation: bool,
+    ) -> str:
+        """Register one schema under an internally established trust policy."""
+
         self._reconcile_pending()
         canonical_schema = canonicalize_json(schema)
         registration = (name, version, canonical_schema)
@@ -132,7 +207,10 @@ def register(self, *, name: str, version: str, schema: dict[str, Any]) -> str:
             version=version,
             definition=schema,
         )
-        _validated_schema(canonical_schema)
+        if skip_meta_validation:
+            _trusted_validated_schema(canonical_schema)
+        else:
+            _validated_schema(canonical_schema)
         schema_uri = self.store.register_descriptor(
             kind="schema",
             name=name,
@@ -164,10 +242,12 @@ def register_model(
         repeats this registration after every restart before accepting writes.
         """
 
-        schema_uri = self.register(
+        schema = model_schema(model)
+        schema_uri = self._register(
             name=name,
             version=version,
-            schema=model_schema(model),
+            schema=schema,
+            skip_meta_validation=_uses_operator_owned_pydantic_schema(model),
         )
         self._bind_model_contract(schema_uri, model)
         if producer_only:

tests/component/schemas/test_schema_registry.py

@@ -1,17 +1,20 @@
 from __future__ import annotations
 
 from pathlib import Path
-from typing import Any, Self
+from typing import Annotated, Any, Self
 
 import pytest
-from pydantic import BaseModel, ConfigDict, model_validator
+from pydantic import BaseModel, ConfigDict, Field, WithJsonSchema, model_validator
 
+import jacobian.schema_registry as schema_registry
+from jacobian.contracts.results import ResultEnvelope
 from jacobian.schema_registry import (
     SchemaRegistry,
     SchemaRegistryError,
     SchemaValidationError,
     model_schema,
 )
+from jacobian.storage.errors import StorageError
 from jacobian.storage.repository import ArtifactRepository
 
 
@@ -36,6 +39,29 @@ def require_order(self) -> Self:
         return self
 
 
+class _MalformedCustomizedSchema(BaseModel):
+    value: int
+
+    @classmethod
+    def model_json_schema(cls, **_kwargs: Any) -> dict[str, Any]:  # type: ignore[override]
+        return {"type": 17}
+
+
+_MalformedCustomizedSchema.__module__ = "jacobian.customized_test_model"
+
+
+class _MalformedFieldExtraSchema(BaseModel):
+    value: int = Field(json_schema_extra={"type": 17})
+
+
+class _MalformedAnnotatedSchema(BaseModel):
+    value: Annotated[int, WithJsonSchema({"type": 17})]
+
+
+for _customized_model in (_MalformedFieldExtraSchema, _MalformedAnnotatedSchema):
+    _customized_model.__module__ = "jacobian.customized_test_model"
+
+
 def test_cached_model_schema_returns_independent_copies(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
@@ -66,6 +92,57 @@ def test_external_dynamic_reference_is_rejected(tmp_path: Path) -> None:
         )
 
 
+@pytest.mark.parametrize(
+    "model",
+    [_MalformedCustomizedSchema, _MalformedFieldExtraSchema, _MalformedAnnotatedSchema],
+)
+def test_customized_model_schema_is_validated_before_persistence(
+    tmp_path: Path,
+    model: type[BaseModel],
+) -> None:
+    store = ArtifactRepository(tmp_path)
+    registry = SchemaRegistry(store)
+
+    with pytest.raises(SchemaRegistryError, match="invalid Draft"):
+        registry.register_model(
+            name="customized-invalid-schema",
+            version="1",
+            model=model,
+        )
+
+    schema = model_schema(model)
+    uri = store.descriptor_uri(
+        kind="schema",
+        name="customized-invalid-schema",
+        version="1",
+        definition=schema,
+    )
+    with pytest.raises(StorageError):
+        store.get_descriptor(uri, expected_kind="schema")
+
+
+def test_operator_owned_default_model_skips_redundant_meta_validation(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    def unexpected_meta_validation(_canonical_schema: bytes) -> None:
+        pytest.fail("default operator-owned Pydantic schema was revalidated")
+
+    monkeypatch.setattr(
+        schema_registry,
+        "_validated_schema",
+        unexpected_meta_validation,
+    )
+
+    uri = SchemaRegistry(ArtifactRepository(tmp_path)).register_model(
+        name="operator-owned-result-envelope",
+        version="1",
+        model=ResultEnvelope,
+    )
+
+    assert uri.startswith("artifact://sha256/")
+
+
 def test_schema_validator_cache_is_bound_to_canonical_schema(
     tmp_path: Path,
 ) -> None: