fix: wire governed schema catalog

app/api/data_rules/routes.py

@@ -23,7 +23,10 @@ from app.core.data_rules.contracts import (
 from app.core.data_rules.production_line import resolve_production_line
 from app.core.data_rules.release import ProductionLineReleaseService
 from app.core.data_rules.repository import DataRuleRepository
-from app.core.data_rules.schema_resolver import SchemaResolver
+from app.core.data_rules.schema_resolver import (
+    Neo4jSchemaMetadataCatalog,
+    SchemaResolver,
+)
 from app.models.result import failed, success
 
 _VALIDATORS = {
@@ -66,8 +69,7 @@ def _release_service() -> ProductionLineReleaseService:
     resolver = current_app.extensions.get("data_rule_schema_resolver")
     if resolver is None:
         catalog = current_app.extensions.get("data_rule_metadata_catalog")
-        if catalog is not None:
-            resolver = SchemaResolver(catalog, repository)
+        resolver = SchemaResolver(catalog or Neo4jSchemaMetadataCatalog(), repository)
     return ProductionLineReleaseService(repository, schema_resolver=resolver)
 
 

app/core/data_rules/repository.py

@@ -2,8 +2,8 @@
 
 from __future__ import annotations
 
-import hashlib
 import json
+import hashlib
 import re
 from typing import Any
 
@@ -28,6 +28,7 @@ from app.core.data_rules.execution_contracts import (
     validate_schema_snapshot,
 )
 
+
 RULE_CATEGORIES = {
     "general",
     "reusable",
@@ -178,11 +179,15 @@ class DataRuleRepository:
                 "id": generation_id,
                 "rule_version_id": version_id,
                 "authoring_surface": surface,
-                "source_text_hash": hashlib.sha256(source.encode("utf-8")).hexdigest(),
+                "source_text_hash": hashlib.sha256(
+                    source.encode("utf-8")
+                ).hexdigest(),
                 "model_provider": _text(
                     evidence.get("model_provider"), "model_provider", 80
                 ),
-                "model_name": _text(evidence.get("model_name"), "model_name", 120),
+                "model_name": _text(
+                    evidence.get("model_name"), "model_name", 120
+                ),
                 "prompt_version": _text(
                     evidence.get("prompt_version"), "prompt_version", 80
                 ),
@@ -196,7 +201,9 @@ class DataRuleRepository:
                 "ambiguities": _json(candidate["ambiguities"]),
                 "repair_attempts": repair_attempts,
                 "decision": decision,
-                "decision_detail": _json({"explanation": candidate["explanation"]}),
+                "decision_detail": _json(
+                    {"explanation": candidate["explanation"]}
+                ),
                 "correlation_id": correlation_id,
             },
         )
@@ -234,20 +241,16 @@ class DataRuleRepository:
         rule_uid = spec["rule_uid"]
         digest = rule_spec_hash(spec)
         self._lock(rule_uid)
-        existing = (
-            self.session.execute(
-                text(
-                    "SELECT id::text AS id, version_no, status "
-                    "FROM public.data_rule_versions "
-                    "WHERE rule_uid = CAST(:rule_uid AS uuid) "
-                    "AND spec_hash = :spec_hash "
-                    "/* existing_version */"
-                ),
-                {"rule_uid": rule_uid, "spec_hash": digest},
-            )
-            .mappings()
-            .one_or_none()
-        )
+        existing = self.session.execute(
+            text(
+                "SELECT id::text AS id, version_no, status "
+                "FROM public.data_rule_versions "
+                "WHERE rule_uid = CAST(:rule_uid AS uuid) "
+                "AND spec_hash = :spec_hash "
+                "/* existing_version */"
+            ),
+            {"rule_uid": rule_uid, "spec_hash": digest},
+        ).mappings().one_or_none()
         if existing is not None:
             return {
                 **dict(existing),
@@ -321,21 +324,17 @@ class DataRuleRepository:
     ) -> dict[str, Any]:
         version = _uid(version_id, "version_id")
         _uid(published_by, "published_by")
-        row = (
-            self.session.execute(
-                text(
-                    "UPDATE public.data_rule_versions "
-                    "SET status = 'published', published_at = CURRENT_TIMESTAMP "
-                    "WHERE id = CAST(:version_id AS uuid) "
-                    "AND status = 'validated' "
-                    "RETURNING id::text AS id, rule_uid::text AS rule_uid, "
-                    "version_no, status, spec_hash"
-                ),
-                {"version_id": version},
-            )
-            .mappings()
-            .one_or_none()
-        )
+        row = self.session.execute(
+            text(
+                "UPDATE public.data_rule_versions "
+                "SET status = 'published', published_at = CURRENT_TIMESTAMP "
+                "WHERE id = CAST(:version_id AS uuid) "
+                "AND status = 'validated' "
+                "RETURNING id::text AS id, rule_uid::text AS rule_uid, "
+                "version_no, status, spec_hash"
+            ),
+            {"version_id": version},
+        ).mappings().one_or_none()
         if row is not None:
             return dict(row)
         status = self.session.execute(
@@ -368,39 +367,33 @@ class DataRuleRepository:
             {clause["rule_version_id"] for clause in spec["clauses"]}
         )
 
-        published_rows = (
-            self.session.execute(
-                text(
-                    "SELECT id::text AS id "
-                    "FROM public.data_rule_versions "
-                    "WHERE id = ANY(CAST(:rule_version_ids AS uuid[])) "
-                    "AND status = 'published' "
-                    "/* published_rule_reference */"
-                ),
-                {"rule_version_ids": rule_version_ids},
-            )
-            .mappings()
-            .all()
-        )
+        published_rows = self.session.execute(
+            text(
+                "SELECT id::text AS id "
+                "FROM public.data_rule_versions "
+                "WHERE id = ANY(CAST(:rule_version_ids AS uuid[])) "
+                "AND status = 'published' "
+                "/* published_rule_reference */"
+            ),
+            {"rule_version_ids": rule_version_ids},
+        ).mappings().all()
         published_ids = {str(row["id"]) for row in published_rows}
         if published_ids != set(rule_version_ids):
-            raise ValueError("standard clauses require published rule versions")
+            raise ValueError(
+                "standard clauses require published rule versions"
+            )
 
         self._lock(standard_uid)
-        existing = (
-            self.session.execute(
-                text(
-                    "SELECT id::text AS id, version_no, status "
-                    "FROM public.data_standard_versions "
-                    "WHERE standard_uid = CAST(:standard_uid AS uuid) "
-                    "AND spec_hash = :spec_hash "
-                    "/* existing_version */"
-                ),
-                {"standard_uid": standard_uid, "spec_hash": digest},
-            )
-            .mappings()
-            .one_or_none()
-        )
+        existing = self.session.execute(
+            text(
+                "SELECT id::text AS id, version_no, status "
+                "FROM public.data_standard_versions "
+                "WHERE standard_uid = CAST(:standard_uid AS uuid) "
+                "AND spec_hash = :spec_hash "
+                "/* existing_version */"
+            ),
+            {"standard_uid": standard_uid, "spec_hash": digest},
+        ).mappings().one_or_none()
         if existing is not None:
             return {
                 **dict(existing),
@@ -477,7 +470,9 @@ class DataRuleRepository:
                     },
                 )
         except IntegrityError as exc:
-            raise ValueError("standard version conflicts with existing data") from exc
+            raise ValueError(
+                "standard version conflicts with existing data"
+            ) from exc
         return {
             "id": version_id,
             "standard_uid": standard_uid,
@@ -492,21 +487,17 @@ class DataRuleRepository:
     ) -> dict[str, Any]:
         version = _uid(version_id, "version_id")
         _uid(published_by, "published_by")
-        row = (
-            self.session.execute(
-                text(
-                    "UPDATE public.data_standard_versions "
-                    "SET status = 'published', published_at = CURRENT_TIMESTAMP "
-                    "WHERE id = CAST(:version_id AS uuid) "
-                    "AND status = 'validated' "
-                    "RETURNING id::text AS id, standard_uid::text AS standard_uid, "
-                    "version_no, status, spec_hash"
-                ),
-                {"version_id": version},
-            )
-            .mappings()
-            .one_or_none()
-        )
+        row = self.session.execute(
+            text(
+                "UPDATE public.data_standard_versions "
+                "SET status = 'published', published_at = CURRENT_TIMESTAMP "
+                "WHERE id = CAST(:version_id AS uuid) "
+                "AND status = 'validated' "
+                "RETURNING id::text AS id, standard_uid::text AS standard_uid, "
+                "version_no, status, spec_hash"
+            ),
+            {"version_id": version},
+        ).mappings().one_or_none()
         if row is not None:
             return dict(row)
         status = self.session.execute(
@@ -518,7 +509,9 @@ class DataRuleRepository:
             {"version_id": version},
         ).scalar_one_or_none()
         if status == "published":
-            raise ValueError("standard version is already published and immutable")
+            raise ValueError(
+                "standard version is already published and immutable"
+            )
         if status is None:
             raise ValueError("standard version was not found")
         raise ValueError("only validated standard versions may be published")
@@ -538,29 +531,25 @@ class DataRuleRepository:
         )
         standard_rows = []
         if standard_ids:
-            standard_rows = (
-                self.session.execute(
-                    text(
-                        "SELECT sv.id::text AS id, sv.status, "
-                        "jsonb_agg(jsonb_build_object("
-                        "'clause_id', b.clause_id, "
-                        "'rule_version_id', b.rule_version_id::text, "
-                        "'severity', b.severity, "
-                        "'exception_policy', b.exception_policy) "
-                        "ORDER BY b.clause_id) AS clauses "
-                        "FROM public.data_standard_versions sv "
-                        "JOIN public.standard_rule_bindings b "
-                        "ON b.standard_version_id = sv.id "
-                        "WHERE sv.id = ANY(CAST(:standard_ids AS uuid[])) "
-                        "AND sv.status = 'published' "
-                        "/* published_standard_assets */ "
-                        "GROUP BY sv.id, sv.status"
-                    ),
-                    {"standard_ids": standard_ids},
-                )
-                .mappings()
-                .all()
-            )
+            standard_rows = self.session.execute(
+                text(
+                    "SELECT sv.id::text AS id, sv.status, "
+                    "jsonb_agg(jsonb_build_object("
+                    "'clause_id', b.clause_id, "
+                    "'rule_version_id', b.rule_version_id::text, "
+                    "'severity', b.severity, "
+                    "'exception_policy', b.exception_policy) "
+                    "ORDER BY b.clause_id) AS clauses "
+                    "FROM public.data_standard_versions sv "
+                    "JOIN public.standard_rule_bindings b "
+                    "ON b.standard_version_id = sv.id "
+                    "WHERE sv.id = ANY(CAST(:standard_ids AS uuid[])) "
+                    "AND sv.status = 'published' "
+                    "/* published_standard_assets */ "
+                    "GROUP BY sv.id, sv.status"
+                ),
+                {"standard_ids": standard_ids},
+            ).mappings().all()
         standards = {
             str(row["id"]): {
                 "id": str(row["id"]),
@@ -581,25 +570,22 @@ class DataRuleRepository:
         }
         for standard in standards.values():
             rule_ids.update(
-                str(clause["rule_version_id"]) for clause in standard["clauses"]
+                str(clause["rule_version_id"])
+                for clause in standard["clauses"]
             )
         rule_rows = []
         if rule_ids:
-            rule_rows = (
-                self.session.execute(
-                    text(
-                        "SELECT rv.id::text AS id, rv.status, rv.rule_spec, "
-                        "rv.spec_hash "
-                        "FROM public.data_rule_versions rv "
-                        "WHERE rv.id = ANY(CAST(:rule_ids AS uuid[])) "
-                        "AND rv.status = 'published' "
-                        "/* published_rule_assets */"
-                    ),
-                    {"rule_ids": sorted(rule_ids)},
-                )
-                .mappings()
-                .all()
-            )
+            rule_rows = self.session.execute(
+                text(
+                    "SELECT rv.id::text AS id, rv.status, rv.rule_spec, "
+                    "rv.spec_hash "
+                    "FROM public.data_rule_versions rv "
+                    "WHERE rv.id = ANY(CAST(:rule_ids AS uuid[])) "
+                    "AND rv.status = 'published' "
+                    "/* published_rule_assets */"
+                ),
+                {"rule_ids": sorted(rule_ids)},
+            ).mappings().all()
         rules = {
             str(row["id"]): {
                 "id": str(row["id"]),
@@ -610,7 +596,9 @@ class DataRuleRepository:
             for row in rule_rows
         }
         if set(rules) != rule_ids:
-            raise ValueError("dataflow references missing or unpublished rule versions")
+            raise ValueError(
+                "dataflow references missing or unpublished rule versions"
+            )
         return standards, rules
 
     def begin_dataflow_release(
@@ -832,7 +820,9 @@ class DataRuleRepository:
             raise ValueError("compiled plan contains unsupported fields")
         if plan["backend"] not in PLAN_BACKENDS:
             raise ValueError("unsupported plan backend")
-        compiler_version = _text(plan["compiler_version"], "compiler_version", 80)
+        compiler_version = _text(
+            plan["compiler_version"], "compiler_version", 80
+        )
         plan_body = _object(plan["plan"], "compiled plan body")
         plan_hash = _digest(plan["plan_hash"], "plan_hash")
         if _canonical_hash(plan_body) != plan_hash:
@@ -863,7 +853,9 @@ class DataRuleRepository:
                 "rule_version_id": rule_id,
                 "stage": _text(stage, "stage", 30),
                 "order_no": order_no,
-                "idempotency": _json(idempotency) if idempotency is not None else None,
+                "idempotency": _json(idempotency)
+                if idempotency is not None
+                else None,
                 "provenance": _json(provenance),
             },
         )
@@ -897,26 +889,22 @@ class DataRuleRepository:
         if not isinstance(package, dict):
             raise ValueError("production-line package must be an object")
         package_hash = _digest(package.get("package_hash"), "package_hash")
-        row = (
-            self.session.execute(
-                text(
-                    "UPDATE public.dataflow_versions "
-                    "SET package = CAST(:package AS jsonb), "
-                    "package_hash = :package_hash, status = 'released', "
-                    "released_at = CURRENT_TIMESTAMP "
-                    "WHERE id = CAST(:version_id AS uuid) "
-                    "AND status = 'validated' "
-                    "RETURNING id::text AS id, version_no, status, package_hash"
-                ),
-                {
-                    "version_id": version_id,
-                    "package": _json(package),
-                    "package_hash": package_hash,
-                },
-            )
-            .mappings()
-            .one_or_none()
-        )
+        row = self.session.execute(
+            text(
+                "UPDATE public.dataflow_versions "
+                "SET package = CAST(:package AS jsonb), "
+                "package_hash = :package_hash, status = 'released', "
+                "released_at = CURRENT_TIMESTAMP "
+                "WHERE id = CAST(:version_id AS uuid) "
+                "AND status = 'validated' "
+                "RETURNING id::text AS id, version_no, status, package_hash"
+            ),
+            {
+                "version_id": version_id,
+                "package": _json(package),
+                "package_hash": package_hash,
+            },
+        ).mappings().one_or_none()
         if row is None:
             raise ValueError("dataflow release is not in validated state")
         return {**dict(row), "package": package}
@@ -940,29 +928,25 @@ class DataRuleRepository:
 
         standard_rows = []
         if standard_ids:
-            standard_rows = (
-                self.session.execute(
-                    text(
-                        "SELECT sv.id::text AS id, sv.status, "
-                        "jsonb_agg(jsonb_build_object("
-                        "'clause_id', b.clause_id, "
-                        "'rule_version_id', b.rule_version_id::text, "
-                        "'severity', b.severity, "
-                        "'exception_policy', b.exception_policy) "
-                        "ORDER BY b.clause_id) AS clauses "
-                        "FROM public.data_standard_versions sv "
-                        "JOIN public.standard_rule_bindings b "
-                        "ON b.standard_version_id = sv.id "
-                        "WHERE sv.id = ANY(CAST(:standard_ids AS uuid[])) "
-                        "AND sv.status = 'published' "
-                        "/* catalog_standard_versions */ "
-                        "GROUP BY sv.id, sv.status"
-                    ),
-                    {"standard_ids": standard_ids},
-                )
-                .mappings()
-                .all()
-            )
+            standard_rows = self.session.execute(
+                text(
+                    "SELECT sv.id::text AS id, sv.status, "
+                    "jsonb_agg(jsonb_build_object("
+                    "'clause_id', b.clause_id, "
+                    "'rule_version_id', b.rule_version_id::text, "
+                    "'severity', b.severity, "
+                    "'exception_policy', b.exception_policy) "
+                    "ORDER BY b.clause_id) AS clauses "
+                    "FROM public.data_standard_versions sv "
+                    "JOIN public.standard_rule_bindings b "
+                    "ON b.standard_version_id = sv.id "
+                    "WHERE sv.id = ANY(CAST(:standard_ids AS uuid[])) "
+                    "AND sv.status = 'published' "
+                    "/* catalog_standard_versions */ "
+                    "GROUP BY sv.id, sv.status"
+                ),
+                {"standard_ids": standard_ids},
+            ).mappings().all()
         standards = {
             str(row["id"]): {
                 "id": str(row["id"]),
@@ -979,32 +963,29 @@ class DataRuleRepository:
         rule_ids = set(direct_rule_ids)
         for standard in standards.values():
             rule_ids.update(
-                str(clause["rule_version_id"]) for clause in standard["clauses"]
+                str(clause["rule_version_id"])
+                for clause in standard["clauses"]
             )
         rule_rows = []
         if rule_ids:
-            rule_rows = (
-                self.session.execute(
-                    text(
-                        "SELECT DISTINCT ON (rv.id) "
-                        "rv.id::text AS id, rv.status, rv.rule_spec, rv.spec_hash, "
-                        "p.backend, p.plan_hash "
-                        "FROM public.data_rule_versions rv "
-                        "JOIN public.dataflow_component_bindings cb "
-                        "ON cb.rule_version_id = rv.id "
-                        "JOIN public.rule_execution_plans p "
-                        "ON p.component_binding_id = cb.id "
-                        "WHERE rv.id = ANY(CAST(:rule_ids AS uuid[])) "
-                        "AND rv.status = 'published' "
-                        "AND p.status = 'published' "
-                        "/* catalog_rule_versions */ "
-                        "ORDER BY rv.id, p.created_at DESC"
-                    ),
-                    {"rule_ids": sorted(rule_ids)},
-                )
-                .mappings()
-                .all()
-            )
+            rule_rows = self.session.execute(
+                text(
+                    "SELECT DISTINCT ON (rv.id) "
+                    "rv.id::text AS id, rv.status, rv.rule_spec, rv.spec_hash, "
+                    "p.backend, p.plan_hash "
+                    "FROM public.data_rule_versions rv "
+                    "JOIN public.dataflow_component_bindings cb "
+                    "ON cb.rule_version_id = rv.id "
+                    "JOIN public.rule_execution_plans p "
+                    "ON p.component_binding_id = cb.id "
+                    "WHERE rv.id = ANY(CAST(:rule_ids AS uuid[])) "
+                    "AND rv.status = 'published' "
+                    "AND p.status = 'published' "
+                    "/* catalog_rule_versions */ "
+                    "ORDER BY rv.id, p.created_at DESC"
+                ),
+                {"rule_ids": sorted(rule_ids)},
+            ).mappings().all()
         rules = {
             str(row["id"]): {
                 "id": str(row["id"]),

app/core/data_rules/schema_resolver.py

@@ -14,6 +14,7 @@ from app.core.data_rules.execution_contracts import (
     validate_dataset_binding,
     validate_schema_snapshot,
 )
+from app.services.neo4j_driver import neo4j_driver
 
 
 class SchemaMetadataCatalog(Protocol):
@@ -56,6 +57,34 @@ _FIELD_TYPE_ALIASES = {
     "varchar": "string",
 }
 _IDENTIFIER = re.compile(r"^[A-Za-z_][A-Za-z0-9_]{0,127}$")
+_SCHEMA_REF = re.compile(r"^bd:([a-z][a-z0-9_-]{0,199}):(v[1-9][0-9]*)$")
+_TYPE_WITH_PARAMETERS = re.compile(
+    r"^([a-z ]+?)(?:\((\d+)(?:\s*,\s*(\d+))?\))?$",
+    re.IGNORECASE,
+)
+_GRAPH_FIELD_TYPES = {
+    "bigint": "integer",
+    "binary": "binary",
+    "bool": "boolean",
+    "boolean": "boolean",
+    "character": "string",
+    "character varying": "string",
+    "date": "date",
+    "datetime": "timestamp",
+    "decimal": "decimal",
+    "double": "double",
+    "float": "float",
+    "int": "integer",
+    "integer": "integer",
+    "json": "json",
+    "numeric": "decimal",
+    "string": "string",
+    "text": "string",
+    "timestamp": "timestamp",
+    "timestamp with time zone": "timestamptz",
+    "timestamptz": "timestamptz",
+    "varchar": "string",
+}
 
 
 def _uid(value: Any, label: str) -> str:
@@ -104,6 +133,134 @@ def _normalized_fields(fields: Any) -> list[dict[str, Any]]:
     )["fields"]
 
 
+def _schema_ref(value: str) -> tuple[str, str]:
+    match = _SCHEMA_REF.fullmatch(value)
+    if match is None:
+        raise ValueError("schema_ref must use bd:<stable-domain-key>:<revision>")
+    return match.group(1), match.group(2)
+
+
+def _record_value(record: Any, key: str) -> Any:
+    if isinstance(record, dict):
+        return record.get(key)
+    return record.get(key)
+
+
+def _nullable(value: Any) -> bool:
+    if isinstance(value, bool):
+        return value
+    if isinstance(value, str):
+        normalized = value.strip().lower()
+        if normalized in {"yes", "true"}:
+            return True
+        if normalized in {"no", "false"}:
+            return False
+    raise ValueError("governance field nullability is missing or invalid")
+
+
+def _non_negative_integer(value: Any, label: str) -> int | None:
+    if value is None:
+        return None
+    if isinstance(value, bool) or not isinstance(value, int) or value < 0:
+        raise ValueError(f"governance field {label} is invalid")
+    return value
+
+
+class Neo4jSchemaMetadataCatalog:
+    """Lazy, credential-free lookup of governed BusinessDomain schemas."""
+
+    def __init__(self, driver=None):
+        self.driver = driver or neo4j_driver
+
+    @staticmethod
+    def _field(record: Any) -> dict[str, Any]:
+        name = _required_string(
+            _record_value(record, "field_name"), "governance field name", 200
+        )
+        raw_type = _required_string(
+            _record_value(record, "data_type"), "governance field type", 100
+        )
+        match = _TYPE_WITH_PARAMETERS.fullmatch(raw_type.strip())
+        if match is None:
+            raise ValueError("unsupported governance field type")
+        field_type = _GRAPH_FIELD_TYPES.get(match.group(1).strip().lower())
+        if field_type is None:
+            raise ValueError("unsupported governance field type")
+        precision = _non_negative_integer(
+            _record_value(record, "precision"), "precision"
+        )
+        scale = _non_negative_integer(_record_value(record, "scale"), "scale")
+        if precision is None and match.group(2) is not None:
+            precision = int(match.group(2))
+        if scale is None and match.group(3) is not None:
+            scale = int(match.group(3))
+        field = {
+            "name": name,
+            "type": field_type,
+            "nullable": _nullable(_record_value(record, "nullable")),
+        }
+        if precision is not None:
+            field["precision"] = precision
+        if scale is not None:
+            field["scale"] = scale
+        timezone = _record_value(record, "timezone")
+        if timezone is not None:
+            field["timezone"] = _required_string(
+                timezone, "governance field timezone", 100
+            )
+        return field
+
+    def load_schema(self, schema_ref: str) -> dict[str, Any] | None:
+        domain_key, revision = _schema_ref(schema_ref)
+        query = """
+        MATCH (bd:BusinessDomain)-[:INCLUDES]->(m:DataMeta)
+        WHERE bd.name_en = $domain_key
+        RETURN id(bd) AS domain_id,
+               bd.name_en AS domain_key,
+               coalesce(bd.schema_revision, bd.revision, bd.version) AS revision,
+               coalesce(m.name_en, m.name_zh) AS field_name,
+               m.data_type AS data_type,
+               m.nullable AS nullable,
+               m.precision AS precision,
+               m.scale AS scale,
+               m.timezone AS timezone
+        ORDER BY field_name
+        """
+        with self.driver.get_session() as session:
+            records = list(
+                session.run(
+                    query,
+                    {"domain_key": domain_key, "revision": revision},
+                )
+            )
+        if not records:
+            raise ValueError("governance schema domain or fields were not found")
+        domains = {
+            (
+                _record_value(record, "domain_id"),
+                _record_value(record, "domain_key"),
+                _record_value(record, "revision"),
+            )
+            for record in records
+        }
+        if len(domains) != 1:
+            raise ValueError("governance schema domain is ambiguous")
+        domain_id, returned_key, returned_revision = domains.pop()
+        if returned_key != domain_key:
+            raise ValueError("governance schema domain key mismatch")
+        if returned_revision != revision:
+            raise ValueError("governance schema revision mismatch")
+        if domain_id is None:
+            raise ValueError("governance schema domain is missing")
+        fields = [self._field(record) for record in records]
+        if not fields:
+            raise ValueError("governance schema fields are missing")
+        return {
+            "source_revision": f"neo4j:{domain_id}:{revision}",
+            "fields": sorted(fields, key=lambda field: field["name"]),
+        }
+
+
 class SchemaResolver:
     """Resolve stable schema references without trusting request payloads."""
 

tests/core/data_rules/test_schema_resolver.py

@@ -29,6 +29,112 @@ class SnapshotRepository:
         return copy.deepcopy(saved)
 
 
+class FakeGraphSession:
+    def __init__(self, records):
+        self.records = records
+        self.calls = []
+
+    def run(self, query, parameters):
+        self.calls.append((query, parameters))
+        return list(self.records)
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *_args):
+        return False
+
+
+class FakeGraphDriver:
+    def __init__(self, records):
+        self.session = FakeGraphSession(records)
+        self.session_requests = 0
+
+    def get_session(self):
+        self.session_requests += 1
+        return self.session
+
+
+def test_neo4j_catalog_loads_stable_domain_schema_without_credentials():
+    from app.core.data_rules.schema_resolver import Neo4jSchemaMetadataCatalog
+
+    driver = FakeGraphDriver(
+        [
+            {
+                "domain_id": 42,
+                "domain_key": "customer",
+                "revision": "v7",
+                "field_name": "mobile",
+                "data_type": "varchar(32)",
+                "nullable": True,
+                "precision": None,
+                "scale": None,
+                "timezone": None,
+            },
+            {
+                "domain_id": 42,
+                "domain_key": "customer",
+                "revision": "v7",
+                "field_name": "updated_at",
+                "data_type": "timestamp",
+                "nullable": False,
+                "precision": None,
+                "scale": None,
+                "timezone": "Asia/Shanghai",
+            },
+        ]
+    )
+
+    metadata = Neo4jSchemaMetadataCatalog(driver).load_schema("bd:customer:v7")
+
+    assert metadata == {
+        "source_revision": "neo4j:42:v7",
+        "fields": [
+            {
+                "name": "mobile",
+                "type": "string",
+                "nullable": True,
+                "precision": 32,
+            },
+            {
+                "name": "updated_at",
+                "type": "timestamp",
+                "nullable": False,
+                "timezone": "Asia/Shanghai",
+            },
+        ],
+    }
+    assert driver.session_requests == 1
+    assert driver.session.calls[0][1] == {
+        "domain_key": "customer",
+        "revision": "v7",
+    }
+    assert "password" not in str(metadata)
+
+
+def test_neo4j_catalog_fails_closed_for_revision_mismatch():
+    from app.core.data_rules.schema_resolver import Neo4jSchemaMetadataCatalog
+
+    driver = FakeGraphDriver(
+        [
+            {
+                "domain_id": 42,
+                "domain_key": "customer",
+                "revision": "v6",
+                "field_name": "mobile",
+                "data_type": "string",
+                "nullable": True,
+                "precision": None,
+                "scale": None,
+                "timezone": None,
+            }
+        ]
+    )
+
+    with pytest.raises(ValueError, match="revision"):
+        Neo4jSchemaMetadataCatalog(driver).load_schema("bd:customer:v7")
+
+
 def test_schema_resolver_hashes_server_metadata_not_request_values():
     from app.core.data_rules.schema_resolver import SchemaResolver
 

tests/test_data_rule_api.py

@@ -108,6 +108,54 @@ class FakeReleaseService:
         }
 
 
+class FakeGraphSession:
+    def __init__(self):
+        self.calls = []
+
+    def run(self, query, parameters):
+        self.calls.append((query, parameters))
+        return [
+            {
+                "domain_id": 9,
+                "domain_key": "customer_raw",
+                "revision": "v2",
+                "field_name": "customer_id",
+                "data_type": "string",
+                "nullable": False,
+                "precision": None,
+                "scale": None,
+                "timezone": None,
+            }
+        ]
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *_args):
+        return False
+
+
+class FakeGraphDriver:
+    def __init__(self):
+        self.session = FakeGraphSession()
+
+    def get_session(self):
+        return self.session
+
+
+class SnapshotOnlyRepository:
+    def __init__(self):
+        self.snapshots = {}
+
+    def find_schema_snapshot(self, *, schema_ref, schema_hash):
+        return self.snapshots.get((schema_ref, schema_hash))
+
+    def persist_schema_snapshot(self, *, snapshot):
+        value = {"id": new_governance_uid(), **snapshot}
+        self.snapshots[(snapshot["schema_ref"], snapshot["schema_hash"])] = value
+        return value
+
+
 def _headers(app, role):
     token = issue_access_token(
         user_id=new_governance_uid(),
@@ -407,3 +455,26 @@ def test_dataflow_release_rejects_client_authored_schema_hashes(monkeypatch):
 
     assert response.status_code == 409
     assert service.calls == []
+
+
+def test_default_release_service_uses_lazy_neo4j_schema_catalog(monkeypatch):
+    from app import create_app
+    from app.api.data_rules.routes import _release_service
+
+    app = create_app()
+    repository = SnapshotOnlyRepository()
+    driver = FakeGraphDriver()
+    monkeypatch.setattr(
+        "app.core.data_rules.schema_resolver.neo4j_driver", driver
+    )
+    app.extensions["data_rule_repository"] = repository
+
+    with app.app_context():
+        service = _release_service()
+        snapshot = service.schema_resolver.resolve("bd:customer_raw:v2")
+
+    assert snapshot["source_revision"] == "neo4j:9:v2"
+    assert driver.session.calls[0][1] == {
+        "domain_key": "customer_raw",
+        "revision": "v2",
+    }