feat: add phase 2 domain template contracts

app/api/meta_data/__init__.py

@@ -1,5 +1,10 @@
+# ruff: noqa: E402, F401
+
 from flask import Blueprint
 
 bp = Blueprint("meta_data", __name__)
 
-from app.api.meta_data import routes  # noqa: E402, F401
+from app.api.meta_data import (
+    domain_templates,
+    routes,
+)

app/api/meta_data/domain_templates.py

@@ -0,0 +1,95 @@
+"""Management API for generic governance domain templates."""
+
+from __future__ import annotations
+
+from flask import g, jsonify, request
+
+from app import db
+from app.api.meta_data import bp
+from app.core.governance.domain_template_repository import (
+    SqlAlchemyDomainTemplateRepository,
+)
+from app.core.governance.domain_templates import (
+    DomainTemplateNotFound,
+    DomainTemplateService,
+    DomainTemplateValidationError,
+)
+from app.models.result import failed, success
+
+
+def _service():
+    return DomainTemplateService(SqlAlchemyDomainTemplateRepository(db.session))
+
+
+@bp.route("/domain-templates", methods=["GET"])
+def list_domain_templates():
+    return jsonify(success(_service().list_templates()))
+
+
+@bp.route("/domain-templates/<template_code>", methods=["GET"])
+def get_domain_template(template_code):
+    try:
+        return jsonify(success(_service().get_template(template_code)))
+    except DomainTemplateNotFound as exc:
+        return jsonify(failed(str(exc), code=404)), 404
+    except DomainTemplateValidationError as exc:
+        return jsonify(failed(str(exc), code=400)), 400
+
+
+@bp.route("/domain-templates/<template_code>/imports", methods=["GET"])
+def list_domain_template_imports(template_code):
+    try:
+        return jsonify(success(_service().list_imports(template_code)))
+    except DomainTemplateValidationError as exc:
+        return jsonify(failed(str(exc), code=400)), 400
+
+
+@bp.route("/domain-templates/dry-run", methods=["POST"])
+def dry_run_domain_template():
+    try:
+        return jsonify(success(_service().dry_run(request.get_json(silent=True))))
+    except DomainTemplateValidationError as exc:
+        return jsonify(failed(str(exc), code=400)), 400
+
+
+@bp.route("/domain-templates/import", methods=["POST"])
+def import_domain_template():
+    try:
+        result = _service().import_template(
+            request.get_json(silent=True),
+            actor_uid=g.current_user["id"],
+        )
+        db.session.commit()
+        return jsonify(success(result, "领域模板已导入"))
+    except DomainTemplateValidationError as exc:
+        db.session.rollback()
+        return jsonify(failed(str(exc), code=400)), 400
+    except Exception:
+        db.session.rollback()
+        raise
+
+@bp.route("/domain-templates/<template_code>/rollback", methods=["POST"])
+def rollback_domain_template(template_code):
+    try:
+        body = request.get_json(silent=True) or {}
+        target_version = body.get("target_version")
+        if isinstance(target_version, bool):
+            target_version = None
+        if isinstance(target_version, str) and target_version.isdigit():
+            target_version = int(target_version)
+        result = _service().rollback(
+            template_code,
+            target_version=target_version,
+            actor_uid=g.current_user["id"],
+        )
+        db.session.commit()
+        return jsonify(success(result, "领域模板已回滚为新版本"))
+    except DomainTemplateNotFound as exc:
+        db.session.rollback()
+        return jsonify(failed(str(exc), code=404)), 404
+    except DomainTemplateValidationError as exc:
+        db.session.rollback()
+        return jsonify(failed(str(exc), code=400)), 400
+    except Exception:
+        db.session.rollback()
+        raise

app/core/governance/domain_template_repository.py

@@ -0,0 +1,303 @@
+"""PostgreSQL persistence for versioned governance domain templates."""
+
+from __future__ import annotations
+
+import json
+
+from sqlalchemy import text
+
+from app.core.common.identifiers import new_governance_uid
+
+
+def _json(value):
+    return json.dumps(value, ensure_ascii=False, sort_keys=True)
+
+
+class SqlAlchemyDomainTemplateRepository:
+    def __init__(self, session):
+        self.session = session
+
+    @staticmethod
+    def _definition(row):
+        if row is None:
+            return None
+        definition = dict(row["definition"])
+        definition["uid"] = str(row["uid"])
+        definition["current_version"] = int(row["current_version"])
+        definition["created_at"] = (
+            row["created_at"].isoformat() if row.get("created_at") else None
+        )
+        definition["updated_at"] = (
+            row["updated_at"].isoformat() if row.get("updated_at") else None
+        )
+        return definition
+
+    def list_templates(self):
+        rows = (
+            self.session.execute(
+                text(
+                    """
+                    SELECT uid, current_version, definition, created_at, updated_at
+                    FROM public.governance_domain_templates
+                    ORDER BY template_code
+                    """
+                )
+            )
+            .mappings()
+            .all()
+        )
+        return [self._definition(row) for row in rows]
+
+    def get_template(self, template_code):
+        row = (
+            self.session.execute(
+                text(
+                    """
+                    SELECT uid, current_version, definition, created_at, updated_at
+                    FROM public.governance_domain_templates
+                    WHERE template_code = :template_code
+                    """
+                ),
+                {"template_code": template_code},
+            )
+            .mappings()
+            .one_or_none()
+        )
+        return self._definition(row)
+
+    def get_version(self, template_code, version):
+        row = (
+            self.session.execute(
+                text(
+                    """
+                    SELECT v.definition
+                    FROM public.governance_domain_template_versions v
+                    JOIN public.governance_domain_templates t
+                      ON t.uid = v.template_uid
+                    WHERE t.template_code = :template_code
+                      AND v.version = :version
+                    """
+                ),
+                {"template_code": template_code, "version": version},
+            )
+            .mappings()
+            .one_or_none()
+        )
+        if row is None:
+            return None
+        return dict(row["definition"])
+
+    def list_imports(self, template_code):
+        rows = (
+            self.session.execute(
+                text(
+                    """
+                    SELECT i.uid, t.template_code, i.operation, i.status,
+                           i.version, i.target_version, i.diff,
+                           i.actor_uid, i.created_at
+                    FROM public.governance_domain_template_imports i
+                    JOIN public.governance_domain_templates t
+                      ON t.uid = i.template_uid
+                    WHERE t.template_code = :template_code
+                    ORDER BY i.created_at DESC, i.uid DESC
+                    """
+                ),
+                {"template_code": template_code},
+            )
+            .mappings()
+            .all()
+        )
+        return [
+            {
+                **dict(row),
+                "uid": str(row["uid"]),
+                "actor_uid": str(row["actor_uid"]),
+                "created_at": row["created_at"].isoformat(),
+            }
+            for row in rows
+        ]
+
+    def apply_import(self, record):
+        template = record["template"]
+        template_code = record["template_code"]
+        current = (
+            self.session.execute(
+                text(
+                    """
+                    SELECT uid, current_version
+                    FROM public.governance_domain_templates
+                    WHERE template_code = :template_code
+                    FOR UPDATE
+                    """
+                ),
+                {"template_code": template_code},
+            )
+            .mappings()
+            .one_or_none()
+        )
+        if current is None:
+            template_uid = record["uid"]
+            self.session.execute(
+                text(
+                    """
+                    INSERT INTO public.governance_domain_templates (
+                        uid, template_code, name, lifecycle_status,
+                        current_version, content_hash, definition
+                    ) VALUES (
+                        CAST(:uid AS uuid), :template_code, :name,
+                        :lifecycle_status, :current_version, :content_hash,
+                        CAST(:definition AS jsonb)
+                    )
+                    """
+                ),
+                {
+                    "uid": template_uid,
+                    "template_code": template_code,
+                    "name": template["name"],
+                    "lifecycle_status": template["lifecycle_status"],
+                    "current_version": record["version"],
+                    "content_hash": template["content_hash"],
+                    "definition": _json(template),
+                },
+            )
+        else:
+            template_uid = str(current["uid"])
+            expected = int(current["current_version"]) + 1
+            if record["version"] != expected:
+                raise RuntimeError("domain template version conflict")
+            self.session.execute(
+                text(
+                    """
+                    UPDATE public.governance_domain_templates
+                    SET name = :name,
+                        lifecycle_status = :lifecycle_status,
+                        current_version = :current_version,
+                        content_hash = :content_hash,
+                        definition = CAST(:definition AS jsonb),
+                        updated_at = CURRENT_TIMESTAMP
+                    WHERE uid = CAST(:uid AS uuid)
+                    """
+                ),
+                {
+                    "uid": template_uid,
+                    "name": template["name"],
+                    "lifecycle_status": template["lifecycle_status"],
+                    "current_version": record["version"],
+                    "content_hash": template["content_hash"],
+                    "definition": _json(template),
+                },
+            )
+
+        self.session.execute(
+            text(
+                """
+                INSERT INTO public.governance_domain_template_versions (
+                    uid, template_uid, version, content_hash,
+                    definition, actor_uid
+                ) VALUES (
+                    CAST(:uid AS uuid), CAST(:template_uid AS uuid), :version,
+                    :content_hash, CAST(:definition AS jsonb),
+                    CAST(:actor_uid AS uuid)
+                )
+                """
+            ),
+            {
+                "uid": new_governance_uid(),
+                "template_uid": template_uid,
+                "version": record["version"],
+                "content_hash": template["content_hash"],
+                "definition": _json(template),
+                "actor_uid": record["audit"]["actor_uid"],
+            },
+        )
+
+        active_type_codes = []
+        for object_type in template["object_types"]:
+            active_type_codes.append(object_type["type_code"])
+            self.session.execute(
+                text(
+                    """
+                    INSERT INTO public.governance_object_types (
+                        uid, template_uid, type_code, name, lifecycle_status,
+                        current_version, stable_uid_prefix,
+                        source_identity_fields, definition
+                    ) VALUES (
+                        CAST(:uid AS uuid), CAST(:template_uid AS uuid),
+                        :type_code, :name, :lifecycle_status, :current_version,
+                        :stable_uid_prefix,
+                        CAST(:source_identity_fields AS jsonb),
+                        CAST(:definition AS jsonb)
+                    )
+                    ON CONFLICT (template_uid, type_code) DO UPDATE SET
+                        name = EXCLUDED.name,
+                        lifecycle_status = EXCLUDED.lifecycle_status,
+                        current_version = EXCLUDED.current_version,
+                        stable_uid_prefix = EXCLUDED.stable_uid_prefix,
+                        source_identity_fields = EXCLUDED.source_identity_fields,
+                        definition = EXCLUDED.definition,
+                        updated_at = CURRENT_TIMESTAMP
+                    """
+                ),
+                {
+                    "uid": new_governance_uid(),
+                    "template_uid": template_uid,
+                    "type_code": object_type["type_code"],
+                    "name": object_type["name"],
+                    "lifecycle_status": object_type["lifecycle_status"],
+                    "current_version": record["version"],
+                    "stable_uid_prefix": object_type["stable_uid_prefix"],
+                    "source_identity_fields": _json(
+                        object_type["source_identity_fields"]
+                    ),
+                    "definition": _json(object_type),
+                },
+            )
+        self.session.execute(
+            text(
+                """
+                UPDATE public.governance_object_types
+                SET lifecycle_status = 'retired',
+                    current_version = :current_version,
+                    updated_at = CURRENT_TIMESTAMP
+                WHERE template_uid = CAST(:template_uid AS uuid)
+                  AND NOT (type_code = ANY(:active_type_codes))
+                """
+            ),
+            {
+                "template_uid": template_uid,
+                "current_version": record["version"],
+                "active_type_codes": active_type_codes,
+            },
+        )
+
+        audit = record["audit"]
+        self.session.execute(
+            text(
+                """
+                INSERT INTO public.governance_domain_template_imports (
+                    uid, template_uid, operation, status, version,
+                    target_version, before_state, after_state, diff, actor_uid
+                ) VALUES (
+                    CAST(:uid AS uuid), CAST(:template_uid AS uuid),
+                    :operation, :status, :version, :target_version,
+                    CAST(:before_state AS jsonb), CAST(:after_state AS jsonb),
+                    CAST(:diff AS jsonb), CAST(:actor_uid AS uuid)
+                )
+                """
+            ),
+            {
+                "uid": audit["uid"],
+                "template_uid": template_uid,
+                "operation": audit["operation"],
+                "status": audit["status"],
+                "version": audit["version"],
+                "target_version": audit["target_version"],
+                "before_state": _json(audit["before_state"])
+                if audit["before_state"] is not None
+                else None,
+                "after_state": _json(audit["after_state"]),
+                "diff": _json(audit["diff"]),
+                "actor_uid": audit["actor_uid"],
+            },
+        )
+        return self.get_template(template_code)

app/core/governance/domain_templates.py

@@ -0,0 +1,432 @@
+"""Generic governance object contracts and versioned domain templates."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import re
+import uuid
+from collections.abc import Callable
+from copy import deepcopy
+from dataclasses import dataclass
+from typing import Any
+
+from app.core.common.identifiers import new_governance_uid
+
+IDENTIFIER_RE = re.compile(r"^[a-z][a-z0-9_]{1,63}$")
+CODE_RE = re.compile(r"^[A-Za-z][A-Za-z0-9_.-]{1,79}$")
+LIFECYCLE_STATUSES = frozenset({"draft", "active", "retired"})
+FIELD_TYPES = frozenset(
+    {"string", "integer", "number", "boolean", "date", "datetime", "object", "array"}
+)
+SECRET_MARKERS = ("password", "secret", "token", "credential", "api_key", "private_key")
+MAX_OBJECT_TYPES = 100
+MAX_COLLECTION_ITEMS = 500
+
+
+class DomainTemplateValidationError(ValueError):
+    pass
+
+
+class DomainTemplateNotFound(LookupError):
+    pass
+
+
+@dataclass(frozen=True)
+class GovernanceObjectType:
+    """Runtime contract for one template-defined governance object type."""
+
+    template_code: str
+    type_code: str
+    name: str
+    stable_uid_prefix: str
+    source_identity_fields: tuple[str, ...]
+    fields: tuple[dict[str, Any], ...]
+    lifecycle_status: str = "active"
+    current_version: int = 1
+
+    @classmethod
+    def from_definition(
+        cls,
+        template_code: str,
+        definition: dict[str, Any],
+        *,
+        current_version: int = 1,
+    ):
+        normalized = _normalize_object_types([definition])[0]
+        return cls(
+            template_code=_require_identifier(template_code, "template_code"),
+            type_code=normalized["type_code"],
+            name=normalized["name"],
+            stable_uid_prefix=normalized["stable_uid_prefix"],
+            source_identity_fields=tuple(normalized["source_identity_fields"]),
+            fields=tuple(normalized["fields"]),
+            lifecycle_status=normalized["lifecycle_status"],
+            current_version=int(current_version),
+        )
+
+    def stable_uid(self, source_identity: dict[str, Any]) -> str:
+        missing = [
+            field
+            for field in self.source_identity_fields
+            if source_identity.get(field) is None
+            or str(source_identity.get(field)).strip() == ""
+        ]
+        if missing:
+            raise DomainTemplateValidationError(
+                "source_identity is missing required fields: " + ", ".join(missing)
+            )
+        canonical_identity = {
+            field: source_identity[field] for field in self.source_identity_fields
+        }
+        return stable_governance_object_uid(
+            self.template_code,
+            self.type_code,
+            canonical_identity,
+        )
+
+
+def _canonical_json(value: Any) -> str:
+    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+
+
+def _content_hash(value: Any) -> str:
+    return hashlib.sha256(_canonical_json(value).encode("utf-8")).hexdigest()
+
+
+def _require_identifier(value: Any, field: str) -> str:
+    normalized = str(value or "").strip()
+    if not IDENTIFIER_RE.fullmatch(normalized):
+        raise DomainTemplateValidationError(
+            f"{field} must be a lowercase snake_case identifier"
+        )
+    return normalized
+
+
+def _require_code(value: Any, field: str) -> str:
+    normalized = str(value or "").strip()
+    if not CODE_RE.fullmatch(normalized):
+        raise DomainTemplateValidationError(f"{field} must be a stable code")
+    return normalized
+
+
+def _require_text(value: Any, field: str, max_length: int = 200) -> str:
+    normalized = str(value or "").strip()
+    if not normalized or len(normalized) > max_length:
+        raise DomainTemplateValidationError(
+            f"{field} must be between 1 and {max_length} characters"
+        )
+    return normalized
+
+
+def _reject_secrets(value: Any, path: str = "template") -> None:
+    if isinstance(value, dict):
+        for key, nested in value.items():
+            normalized_key = str(key).lower().replace("-", "_")
+            if any(marker in normalized_key for marker in SECRET_MARKERS):
+                raise DomainTemplateValidationError(
+                    f"secret-bearing field is not allowed at {path}.{key}"
+                )
+            _reject_secrets(nested, f"{path}.{key}")
+    elif isinstance(value, list):
+        for index, nested in enumerate(value):
+            _reject_secrets(nested, f"{path}[{index}]")
+
+
+def _normalize_fields(fields: Any, object_code: str) -> list[dict[str, Any]]:
+    if fields is None:
+        fields = []
+    if not isinstance(fields, list) or len(fields) > MAX_COLLECTION_ITEMS:
+        raise DomainTemplateValidationError(f"{object_code}.fields must be a bounded list")
+    normalized = []
+    seen = set()
+    for item in fields:
+        if not isinstance(item, dict):
+            raise DomainTemplateValidationError(f"{object_code}.fields items must be objects")
+        code = _require_identifier(item.get("code"), f"{object_code}.field.code")
+        if any(marker in code for marker in SECRET_MARKERS):
+            raise DomainTemplateValidationError(
+                f"secret-bearing field is not allowed at {object_code}.{code}"
+            )
+        if code in seen:
+            raise DomainTemplateValidationError(f"duplicate field {object_code}.{code}")
+        seen.add(code)
+        field_type = str(item.get("type") or "string").strip().lower()
+        if field_type not in FIELD_TYPES:
+            raise DomainTemplateValidationError(
+                f"{object_code}.{code}.type is unsupported"
+            )
+        normalized.append(
+            {
+                "code": code,
+                "name": _require_text(
+                    item.get("name") or code, f"{object_code}.{code}.name"
+                ),
+                "type": field_type,
+                "required": bool(item.get("required", False)),
+                "description": str(item.get("description") or "").strip(),
+            }
+        )
+    return sorted(normalized, key=lambda item: item["code"])
+
+
+def _normalize_object_types(items: Any) -> list[dict[str, Any]]:
+    if not isinstance(items, list) or not items or len(items) > MAX_OBJECT_TYPES:
+        raise DomainTemplateValidationError("object_types must be a non-empty bounded list")
+    normalized = []
+    seen = set()
+    for item in items:
+        if not isinstance(item, dict):
+            raise DomainTemplateValidationError("object type must be an object")
+        code = _require_identifier(item.get("type_code"), "object_types.type_code")
+        if code in seen:
+            raise DomainTemplateValidationError(f"duplicate object type: {code}")
+        seen.add(code)
+        source_identity_fields = item.get("source_identity_fields")
+        if not isinstance(source_identity_fields, list) or not source_identity_fields:
+            raise DomainTemplateValidationError(
+                f"{code}.source_identity_fields must be a non-empty list"
+            )
+        source_identity_fields = [
+            _require_identifier(value, f"{code}.source_identity_fields")
+            for value in source_identity_fields
+        ]
+        if any(
+            marker in field
+            for field in source_identity_fields
+            for marker in SECRET_MARKERS
+        ):
+            raise DomainTemplateValidationError(
+                f"{code}.source_identity_fields cannot contain secret-bearing fields"
+            )
+        if len(set(source_identity_fields)) != len(source_identity_fields):
+            raise DomainTemplateValidationError(
+                f"{code}.source_identity_fields contains duplicates"
+            )
+        normalized.append(
+            {
+                "type_code": code,
+                "name": _require_text(item.get("name"), f"{code}.name"),
+                "description": str(item.get("description") or "").strip(),
+                "stable_uid_prefix": _require_text(
+                    item.get("stable_uid_prefix"), f"{code}.stable_uid_prefix", 16
+                ).upper(),
+                "source_identity_fields": source_identity_fields,
+                "fields": _normalize_fields(item.get("fields"), code),
+                "lifecycle_status": str(
+                    item.get("lifecycle_status") or "active"
+                ).strip().lower(),
+                "extension": deepcopy(item.get("extension") or {}),
+            }
+        )
+        if normalized[-1]["lifecycle_status"] not in LIFECYCLE_STATUSES:
+            raise DomainTemplateValidationError(
+                f"{code}.lifecycle_status is unsupported"
+            )
+    return sorted(normalized, key=lambda item: item["type_code"])
+
+
+def _normalize_named_collection(value: Any, name: str) -> list[dict[str, Any]]:
+    if value is None:
+        return []
+    if not isinstance(value, list) or len(value) > MAX_COLLECTION_ITEMS:
+        raise DomainTemplateValidationError(f"{name} must be a bounded list")
+    normalized = []
+    seen = set()
+    for item in value:
+        if not isinstance(item, dict):
+            raise DomainTemplateValidationError(f"{name} items must be objects")
+        code = _require_code(item.get("code"), f"{name}.code")
+        if code in seen:
+            raise DomainTemplateValidationError(f"duplicate {name} code: {code}")
+        seen.add(code)
+        normalized.append({**deepcopy(item), "code": code})
+    return sorted(normalized, key=lambda item: item["code"])
+
+
+def normalize_domain_template(definition: Any) -> dict[str, Any]:
+    if not isinstance(definition, dict):
+        raise DomainTemplateValidationError("template definition must be an object")
+    _reject_secrets(definition)
+    lifecycle_status = str(definition.get("lifecycle_status") or "draft").strip().lower()
+    if lifecycle_status not in LIFECYCLE_STATUSES:
+        raise DomainTemplateValidationError("lifecycle_status is unsupported")
+    seed_data = deepcopy(definition.get("seed_data") or [])
+    if not isinstance(seed_data, list) or len(seed_data) > MAX_COLLECTION_ITEMS:
+        raise DomainTemplateValidationError("seed_data must be a bounded list")
+    normalized = {
+        "template_code": _require_identifier(
+            definition.get("template_code"), "template_code"
+        ),
+        "name": _require_text(definition.get("name"), "name"),
+        "description": str(definition.get("description") or "").strip(),
+        "lifecycle_status": lifecycle_status,
+        "object_types": _normalize_object_types(definition.get("object_types")),
+        "responsibility_roles": _normalize_named_collection(
+            definition.get("responsibility_roles"), "responsibility_roles"
+        ),
+        "rules": _normalize_named_collection(definition.get("rules"), "rules"),
+        "metrics": _normalize_named_collection(definition.get("metrics"), "metrics"),
+        "seed_data": seed_data,
+        "source_identity_contract": {
+            "strategy": "uuid5",
+            "canonicalization": "sorted-json",
+            "required": True,
+        },
+        "version_contract": {
+            "strategy": "append-only",
+            "current_version_field": "current_version",
+            "content_hash": "sha256",
+        },
+        "lifecycle_contract": {
+            "statuses": sorted(LIFECYCLE_STATUSES),
+            "transitions": {
+                "draft": ["active", "retired"],
+                "active": ["retired"],
+                "retired": [],
+            },
+        },
+    }
+    normalized["content_hash"] = _content_hash(
+        {key: value for key, value in normalized.items() if key != "content_hash"}
+    )
+    return normalized
+
+
+def stable_governance_object_uid(
+    template_code: str,
+    object_type: str,
+    source_identity: dict[str, Any],
+) -> str:
+    template_code = _require_identifier(template_code, "template_code")
+    object_type = _require_identifier(object_type, "object_type")
+    if not isinstance(source_identity, dict) or not source_identity:
+        raise DomainTemplateValidationError("source_identity must be a non-empty object")
+    if any(value is None or str(value).strip() == "" for value in source_identity.values()):
+        raise DomainTemplateValidationError("source_identity values must be non-empty")
+    identity = f"{template_code}/{object_type}/{_canonical_json(source_identity)}"
+    return str(uuid.uuid5(uuid.NAMESPACE_URL, f"dataops-governance:{identity}"))
+
+
+def diff_domain_templates(
+    before: dict[str, Any] | None,
+    after: dict[str, Any],
+) -> dict[str, Any]:
+    before = before or {}
+    before_types = {
+        item["type_code"]: item for item in before.get("object_types", [])
+    }
+    after_types = {item["type_code"]: item for item in after.get("object_types", [])}
+    shared = set(before_types) & set(after_types)
+    sections = ("responsibility_roles", "rules", "metrics", "seed_data")
+    return {
+        "added_object_types": sorted(set(after_types) - set(before_types)),
+        "removed_object_types": sorted(set(before_types) - set(after_types)),
+        "changed_object_types": sorted(
+            code for code in shared if before_types[code] != after_types[code]
+        ),
+        "changed_sections": sorted(
+            section
+            for section in sections
+            if before.get(section, []) != after.get(section, [])
+        ),
+        "lifecycle_changed": (
+            bool(before)
+            and before.get("lifecycle_status") != after.get("lifecycle_status")
+        ),
+    }
+
+
+class DomainTemplateService:
+    def __init__(
+        self,
+        repository,
+        *,
+        uid_factory: Callable[[], str] = new_governance_uid,
+    ):
+        self.repository = repository
+        self.uid_factory = uid_factory
+
+    def list_templates(self):
+        return self.repository.list_templates()
+
+    def get_template(self, template_code: str):
+        template_code = _require_identifier(template_code, "template_code")
+        result = self.repository.get_template(template_code)
+        if result is None:
+            raise DomainTemplateNotFound(f"domain template {template_code} was not found")
+        return result
+
+    def list_imports(self, template_code: str):
+        template_code = _require_identifier(template_code, "template_code")
+        return self.repository.list_imports(template_code)
+
+    def dry_run(self, definition: Any):
+        normalized = normalize_domain_template(definition)
+        before = self.repository.get_template(normalized["template_code"])
+        return {
+            "valid": True,
+            "template": normalized,
+            "diff": diff_domain_templates(before, normalized),
+        }
+
+    def import_template(
+        self,
+        definition: Any,
+        *,
+        actor_uid: str,
+        operation: str = "import",
+        target_version: int | None = None,
+    ):
+        normalized = normalize_domain_template(definition)
+        before = self.repository.get_template(normalized["template_code"])
+        version = int(before.get("current_version", 0)) + 1 if before else 1
+        persisted = {**normalized, "current_version": version}
+        audit = {
+            "uid": self.uid_factory(),
+            "template_code": normalized["template_code"],
+            "operation": operation,
+            "status": "applied",
+            "version": version,
+            "target_version": target_version,
+            "before_state": deepcopy(before),
+            "after_state": deepcopy(persisted),
+            "diff": diff_domain_templates(before, normalized),
+            "actor_uid": str(actor_uid),
+        }
+        return self.repository.apply_import(
+            {
+                "uid": self.uid_factory(),
+                "template_code": normalized["template_code"],
+                "version": version,
+                "template": persisted,
+                "audit": audit,
+            }
+        )
+
+    def rollback(
+        self,
+        template_code: str,
+        *,
+        target_version: int,
+        actor_uid: str,
+    ):
+        template_code = _require_identifier(template_code, "template_code")
+        if not isinstance(target_version, int) or target_version < 1:
+            raise DomainTemplateValidationError("target_version must be a positive integer")
+        target = self.repository.get_version(template_code, target_version)
+        if target is None:
+            raise DomainTemplateNotFound(
+                f"domain template {template_code} version {target_version} was not found"
+            )
+        definition = {
+            key: deepcopy(value)
+            for key, value in target.items()
+            if key not in {"current_version", "content_hash"}
+        }
+        return self.import_template(
+            definition,
+            actor_uid=actor_uid,
+            operation="rollback",
+            target_version=target_version,
+        )

app/core/system/permissions.py

@@ -46,10 +46,18 @@ QUALITY_ISSUES_REVIEW = "quality-issues:review"
 DEVICE_OBSERVABILITY_EDIT = "device-observability:edit"
 GOVERNANCE_AUDIT_READ = "governance-audit:read"
 GOVERNANCE_AUDIT_SEAL = "governance-audit:seal"
+DOMAIN_TEMPLATES_READ = "domain-templates:read"
+DOMAIN_TEMPLATES_PREVIEW = "domain-templates:preview"
+DOMAIN_TEMPLATES_MANAGE = "domain-templates:manage"
 
 ROLE_PERMISSIONS = {
     "viewer": frozenset(
-        {READ_GOVERNANCE, RULES_READ, RESPONSIBILITIES_READ}
+        {
+            READ_GOVERNANCE,
+            RULES_READ,
+            RESPONSIBILITIES_READ,
+            DOMAIN_TEMPLATES_READ,
+        }
     ),
     "editor": frozenset(
         {
@@ -71,6 +79,8 @@ ROLE_PERMISSIONS = {
             DEVICE_QUALITY_EXECUTE,
             QUALITY_ISSUES_EDIT,
             DEVICE_OBSERVABILITY_EDIT,
+            DOMAIN_TEMPLATES_READ,
+            DOMAIN_TEMPLATES_PREVIEW,
         }
     ),
     "admin": frozenset(
@@ -114,6 +124,9 @@ ROLE_PERMISSIONS = {
             DEVICE_OBSERVABILITY_EDIT,
             GOVERNANCE_AUDIT_READ,
             GOVERNANCE_AUDIT_SEAL,
+            DOMAIN_TEMPLATES_READ,
+            DOMAIN_TEMPLATES_PREVIEW,
+            DOMAIN_TEMPLATES_MANAGE,
         }
     ),
 }
@@ -134,6 +147,12 @@ def permission_for_request(path: str, method: str) -> tuple[str, ...]:
         if method == "GET":
             return (GOVERNANCE_AUDIT_READ,)
         return (GOVERNANCE_AUDIT_SEAL,)
+    if path.startswith("/api/meta/domain-templates"):
+        if method == "GET":
+            return (DOMAIN_TEMPLATES_READ,)
+        if path == "/api/meta/domain-templates/dry-run":
+            return (DOMAIN_TEMPLATES_PREVIEW,)
+        return (DOMAIN_TEMPLATES_MANAGE,)
     if path in {"/api/knowledge/search", "/api/knowledge/ask"}:
         return (READ_GOVERNANCE,)
     if path.startswith("/api/rules"):

app/models/__init__.py

@@ -1,28 +1,38 @@
 # Models package initialization
 
 from app.models.data_product import DataOrder, DataProduct
-from app.models.metadata_review import MetadataReviewRecord, MetadataVersionHistory
 from app.models.data_research import (
+    CandidateDecisionRecord,
+    DataElement,
+    DataElementVersion,
     EvidenceFragment,
     ExtractionCandidate,
     IngestionJob,
     IngestionSource,
-    SourceArtifact,
-    DataElement,
-    DataElementVersion,
-    CandidateDecisionRecord,
-    OntologyModel,
-    OntologyVersionModel,
-    OntologyDomainLinkModel,
     OntologyChangeSetModel,
+    OntologyDomainLinkModel,
+    OntologyModel,
     OntologyPublishRunModel,
+    OntologyVersionModel,
+    SourceArtifact,
 )
+from app.models.governance_template import (
+    GovernanceDomainTemplate,
+    GovernanceDomainTemplateImport,
+    GovernanceDomainTemplateVersion,
+    GovernanceObjectType,
+)
+from app.models.metadata_review import MetadataReviewRecord, MetadataVersionHistory
 
 __all__ = [
     "DataOrder",
     "DataProduct",
     "MetadataReviewRecord",
     "MetadataVersionHistory",
+    "GovernanceDomainTemplate",
+    "GovernanceDomainTemplateVersion",
+    "GovernanceObjectType",
+    "GovernanceDomainTemplateImport",
     "IngestionSource",
     "IngestionJob",
     "SourceArtifact",

app/models/governance_template.py

@@ -0,0 +1,88 @@
+"""SQLAlchemy mappings for governance domain templates."""
+
+from __future__ import annotations
+
+from sqlalchemy.dialects.postgresql import JSONB, UUID
+
+from app import db
+from app.core.common.identifiers import new_governance_uid
+
+
+class GovernanceDomainTemplate(db.Model):
+    __tablename__ = "governance_domain_templates"
+    __table_args__ = {"schema": "public"}
+
+    uid = db.Column(UUID(as_uuid=False), primary_key=True, default=new_governance_uid)
+    template_code = db.Column(db.String(64), unique=True, nullable=False)
+    name = db.Column(db.String(200), nullable=False)
+    lifecycle_status = db.Column(db.String(20), nullable=False)
+    current_version = db.Column(db.Integer, nullable=False)
+    content_hash = db.Column(db.String(64), nullable=False)
+    definition = db.Column(JSONB, nullable=False)
+    created_at = db.Column(db.DateTime(timezone=True), nullable=False)
+    updated_at = db.Column(db.DateTime(timezone=True), nullable=False)
+
+
+class GovernanceDomainTemplateVersion(db.Model):
+    __tablename__ = "governance_domain_template_versions"
+    __table_args__ = (
+        db.UniqueConstraint("template_uid", "version"),
+        {"schema": "public"},
+    )
+
+    uid = db.Column(UUID(as_uuid=False), primary_key=True, default=new_governance_uid)
+    template_uid = db.Column(
+        UUID(as_uuid=False),
+        db.ForeignKey("public.governance_domain_templates.uid"),
+        nullable=False,
+    )
+    version = db.Column(db.Integer, nullable=False)
+    content_hash = db.Column(db.String(64), nullable=False)
+    definition = db.Column(JSONB, nullable=False)
+    actor_uid = db.Column(UUID(as_uuid=False), nullable=False)
+    created_at = db.Column(db.DateTime(timezone=True), nullable=False)
+
+
+class GovernanceObjectType(db.Model):
+    __tablename__ = "governance_object_types"
+    __table_args__ = (
+        db.UniqueConstraint("template_uid", "type_code"),
+        {"schema": "public"},
+    )
+
+    uid = db.Column(UUID(as_uuid=False), primary_key=True, default=new_governance_uid)
+    template_uid = db.Column(
+        UUID(as_uuid=False),
+        db.ForeignKey("public.governance_domain_templates.uid"),
+        nullable=False,
+    )
+    type_code = db.Column(db.String(64), nullable=False)
+    name = db.Column(db.String(200), nullable=False)
+    lifecycle_status = db.Column(db.String(20), nullable=False)
+    current_version = db.Column(db.Integer, nullable=False)
+    stable_uid_prefix = db.Column(db.String(16), nullable=False)
+    source_identity_fields = db.Column(JSONB, nullable=False)
+    definition = db.Column(JSONB, nullable=False)
+    created_at = db.Column(db.DateTime(timezone=True), nullable=False)
+    updated_at = db.Column(db.DateTime(timezone=True), nullable=False)
+
+
+class GovernanceDomainTemplateImport(db.Model):
+    __tablename__ = "governance_domain_template_imports"
+    __table_args__ = {"schema": "public"}
+
+    uid = db.Column(UUID(as_uuid=False), primary_key=True, default=new_governance_uid)
+    template_uid = db.Column(
+        UUID(as_uuid=False),
+        db.ForeignKey("public.governance_domain_templates.uid"),
+        nullable=False,
+    )
+    operation = db.Column(db.String(20), nullable=False)
+    status = db.Column(db.String(20), nullable=False)
+    version = db.Column(db.Integer, nullable=False)
+    target_version = db.Column(db.Integer)
+    before_state = db.Column(JSONB)
+    after_state = db.Column(JSONB, nullable=False)
+    diff = db.Column(JSONB, nullable=False)
+    actor_uid = db.Column(UUID(as_uuid=False), nullable=False)
+    created_at = db.Column(db.DateTime(timezone=True), nullable=False)

deployment/app/api/meta_data/__init__.py

@@ -1,5 +1,10 @@
+# ruff: noqa: E402, F401
+
 from flask import Blueprint
 
 bp = Blueprint("meta_data", __name__)
 
-from app.api.meta_data import routes  # noqa: E402, F401
+from app.api.meta_data import (
+    domain_templates,
+    routes,
+)

deployment/app/api/meta_data/domain_templates.py

@@ -0,0 +1,95 @@
+"""Management API for generic governance domain templates."""
+
+from __future__ import annotations
+
+from flask import g, jsonify, request
+
+from app import db
+from app.api.meta_data import bp
+from app.core.governance.domain_template_repository import (
+    SqlAlchemyDomainTemplateRepository,
+)
+from app.core.governance.domain_templates import (
+    DomainTemplateNotFound,
+    DomainTemplateService,
+    DomainTemplateValidationError,
+)
+from app.models.result import failed, success
+
+
+def _service():
+    return DomainTemplateService(SqlAlchemyDomainTemplateRepository(db.session))
+
+
+@bp.route("/domain-templates", methods=["GET"])
+def list_domain_templates():
+    return jsonify(success(_service().list_templates()))
+
+
+@bp.route("/domain-templates/<template_code>", methods=["GET"])
+def get_domain_template(template_code):
+    try:
+        return jsonify(success(_service().get_template(template_code)))
+    except DomainTemplateNotFound as exc:
+        return jsonify(failed(str(exc), code=404)), 404
+    except DomainTemplateValidationError as exc:
+        return jsonify(failed(str(exc), code=400)), 400
+
+
+@bp.route("/domain-templates/<template_code>/imports", methods=["GET"])
+def list_domain_template_imports(template_code):
+    try:
+        return jsonify(success(_service().list_imports(template_code)))
+    except DomainTemplateValidationError as exc:
+        return jsonify(failed(str(exc), code=400)), 400
+
+
+@bp.route("/domain-templates/dry-run", methods=["POST"])
+def dry_run_domain_template():
+    try:
+        return jsonify(success(_service().dry_run(request.get_json(silent=True))))
+    except DomainTemplateValidationError as exc:
+        return jsonify(failed(str(exc), code=400)), 400
+
+
+@bp.route("/domain-templates/import", methods=["POST"])
+def import_domain_template():
+    try:
+        result = _service().import_template(
+            request.get_json(silent=True),
+            actor_uid=g.current_user["id"],
+        )
+        db.session.commit()
+        return jsonify(success(result, "领域模板已导入"))
+    except DomainTemplateValidationError as exc:
+        db.session.rollback()
+        return jsonify(failed(str(exc), code=400)), 400
+    except Exception:
+        db.session.rollback()
+        raise
+
+@bp.route("/domain-templates/<template_code>/rollback", methods=["POST"])
+def rollback_domain_template(template_code):
+    try:
+        body = request.get_json(silent=True) or {}
+        target_version = body.get("target_version")
+        if isinstance(target_version, bool):
+            target_version = None
+        if isinstance(target_version, str) and target_version.isdigit():
+            target_version = int(target_version)
+        result = _service().rollback(
+            template_code,
+            target_version=target_version,
+            actor_uid=g.current_user["id"],
+        )
+        db.session.commit()
+        return jsonify(success(result, "领域模板已回滚为新版本"))
+    except DomainTemplateNotFound as exc:
+        db.session.rollback()
+        return jsonify(failed(str(exc), code=404)), 404
+    except DomainTemplateValidationError as exc:
+        db.session.rollback()
+        return jsonify(failed(str(exc), code=400)), 400
+    except Exception:
+        db.session.rollback()
+        raise

deployment/app/core/governance/domain_template_repository.py

@@ -0,0 +1,303 @@
+"""PostgreSQL persistence for versioned governance domain templates."""
+
+from __future__ import annotations
+
+import json
+
+from sqlalchemy import text
+
+from app.core.common.identifiers import new_governance_uid
+
+
+def _json(value):
+    return json.dumps(value, ensure_ascii=False, sort_keys=True)
+
+
+class SqlAlchemyDomainTemplateRepository:
+    def __init__(self, session):
+        self.session = session
+
+    @staticmethod
+    def _definition(row):
+        if row is None:
+            return None
+        definition = dict(row["definition"])
+        definition["uid"] = str(row["uid"])
+        definition["current_version"] = int(row["current_version"])
+        definition["created_at"] = (
+            row["created_at"].isoformat() if row.get("created_at") else None
+        )
+        definition["updated_at"] = (
+            row["updated_at"].isoformat() if row.get("updated_at") else None
+        )
+        return definition
+
+    def list_templates(self):
+        rows = (
+            self.session.execute(
+                text(
+                    """
+                    SELECT uid, current_version, definition, created_at, updated_at
+                    FROM public.governance_domain_templates
+                    ORDER BY template_code
+                    """
+                )
+            )
+            .mappings()
+            .all()
+        )
+        return [self._definition(row) for row in rows]
+
+    def get_template(self, template_code):
+        row = (
+            self.session.execute(
+                text(
+                    """
+                    SELECT uid, current_version, definition, created_at, updated_at
+                    FROM public.governance_domain_templates
+                    WHERE template_code = :template_code
+                    """
+                ),
+                {"template_code": template_code},
+            )
+            .mappings()
+            .one_or_none()
+        )
+        return self._definition(row)
+
+    def get_version(self, template_code, version):
+        row = (
+            self.session.execute(
+                text(
+                    """
+                    SELECT v.definition
+                    FROM public.governance_domain_template_versions v
+                    JOIN public.governance_domain_templates t
+                      ON t.uid = v.template_uid
+                    WHERE t.template_code = :template_code
+                      AND v.version = :version
+                    """
+                ),
+                {"template_code": template_code, "version": version},
+            )
+            .mappings()
+            .one_or_none()
+        )
+        if row is None:
+            return None
+        return dict(row["definition"])
+
+    def list_imports(self, template_code):
+        rows = (
+            self.session.execute(
+                text(
+                    """
+                    SELECT i.uid, t.template_code, i.operation, i.status,
+                           i.version, i.target_version, i.diff,
+                           i.actor_uid, i.created_at
+                    FROM public.governance_domain_template_imports i
+                    JOIN public.governance_domain_templates t
+                      ON t.uid = i.template_uid
+                    WHERE t.template_code = :template_code
+                    ORDER BY i.created_at DESC, i.uid DESC
+                    """
+                ),
+                {"template_code": template_code},
+            )
+            .mappings()
+            .all()
+        )
+        return [
+            {
+                **dict(row),
+                "uid": str(row["uid"]),
+                "actor_uid": str(row["actor_uid"]),
+                "created_at": row["created_at"].isoformat(),
+            }
+            for row in rows
+        ]
+
+    def apply_import(self, record):
+        template = record["template"]
+        template_code = record["template_code"]
+        current = (
+            self.session.execute(
+                text(
+                    """
+                    SELECT uid, current_version
+                    FROM public.governance_domain_templates
+                    WHERE template_code = :template_code
+                    FOR UPDATE
+                    """
+                ),
+                {"template_code": template_code},
+            )
+            .mappings()
+            .one_or_none()
+        )
+        if current is None:
+            template_uid = record["uid"]
+            self.session.execute(
+                text(
+                    """
+                    INSERT INTO public.governance_domain_templates (
+                        uid, template_code, name, lifecycle_status,
+                        current_version, content_hash, definition
+                    ) VALUES (
+                        CAST(:uid AS uuid), :template_code, :name,
+                        :lifecycle_status, :current_version, :content_hash,
+                        CAST(:definition AS jsonb)
+                    )
+                    """
+                ),
+                {
+                    "uid": template_uid,
+                    "template_code": template_code,
+                    "name": template["name"],
+                    "lifecycle_status": template["lifecycle_status"],
+                    "current_version": record["version"],
+                    "content_hash": template["content_hash"],
+                    "definition": _json(template),
+                },
+            )
+        else:
+            template_uid = str(current["uid"])
+            expected = int(current["current_version"]) + 1
+            if record["version"] != expected:
+                raise RuntimeError("domain template version conflict")
+            self.session.execute(
+                text(
+                    """
+                    UPDATE public.governance_domain_templates
+                    SET name = :name,
+                        lifecycle_status = :lifecycle_status,
+                        current_version = :current_version,
+                        content_hash = :content_hash,
+                        definition = CAST(:definition AS jsonb),
+                        updated_at = CURRENT_TIMESTAMP
+                    WHERE uid = CAST(:uid AS uuid)
+                    """
+                ),
+                {
+                    "uid": template_uid,
+                    "name": template["name"],
+                    "lifecycle_status": template["lifecycle_status"],
+                    "current_version": record["version"],
+                    "content_hash": template["content_hash"],
+                    "definition": _json(template),
+                },
+            )
+
+        self.session.execute(
+            text(
+                """
+                INSERT INTO public.governance_domain_template_versions (
+                    uid, template_uid, version, content_hash,
+                    definition, actor_uid
+                ) VALUES (
+                    CAST(:uid AS uuid), CAST(:template_uid AS uuid), :version,
+                    :content_hash, CAST(:definition AS jsonb),
+                    CAST(:actor_uid AS uuid)
+                )
+                """
+            ),
+            {
+                "uid": new_governance_uid(),
+                "template_uid": template_uid,
+                "version": record["version"],
+                "content_hash": template["content_hash"],
+                "definition": _json(template),
+                "actor_uid": record["audit"]["actor_uid"],
+            },
+        )
+
+        active_type_codes = []
+        for object_type in template["object_types"]:
+            active_type_codes.append(object_type["type_code"])
+            self.session.execute(
+                text(
+                    """
+                    INSERT INTO public.governance_object_types (
+                        uid, template_uid, type_code, name, lifecycle_status,
+                        current_version, stable_uid_prefix,
+                        source_identity_fields, definition
+                    ) VALUES (
+                        CAST(:uid AS uuid), CAST(:template_uid AS uuid),
+                        :type_code, :name, :lifecycle_status, :current_version,
+                        :stable_uid_prefix,
+                        CAST(:source_identity_fields AS jsonb),
+                        CAST(:definition AS jsonb)
+                    )
+                    ON CONFLICT (template_uid, type_code) DO UPDATE SET
+                        name = EXCLUDED.name,
+                        lifecycle_status = EXCLUDED.lifecycle_status,
+                        current_version = EXCLUDED.current_version,
+                        stable_uid_prefix = EXCLUDED.stable_uid_prefix,
+                        source_identity_fields = EXCLUDED.source_identity_fields,
+                        definition = EXCLUDED.definition,
+                        updated_at = CURRENT_TIMESTAMP
+                    """
+                ),
+                {
+                    "uid": new_governance_uid(),
+                    "template_uid": template_uid,
+                    "type_code": object_type["type_code"],
+                    "name": object_type["name"],
+                    "lifecycle_status": object_type["lifecycle_status"],
+                    "current_version": record["version"],
+                    "stable_uid_prefix": object_type["stable_uid_prefix"],
+                    "source_identity_fields": _json(
+                        object_type["source_identity_fields"]
+                    ),
+                    "definition": _json(object_type),
+                },
+            )
+        self.session.execute(
+            text(
+                """
+                UPDATE public.governance_object_types
+                SET lifecycle_status = 'retired',
+                    current_version = :current_version,
+                    updated_at = CURRENT_TIMESTAMP
+                WHERE template_uid = CAST(:template_uid AS uuid)
+                  AND NOT (type_code = ANY(:active_type_codes))
+                """
+            ),
+            {
+                "template_uid": template_uid,
+                "current_version": record["version"],
+                "active_type_codes": active_type_codes,
+            },
+        )
+
+        audit = record["audit"]
+        self.session.execute(
+            text(
+                """
+                INSERT INTO public.governance_domain_template_imports (
+                    uid, template_uid, operation, status, version,
+                    target_version, before_state, after_state, diff, actor_uid
+                ) VALUES (
+                    CAST(:uid AS uuid), CAST(:template_uid AS uuid),
+                    :operation, :status, :version, :target_version,
+                    CAST(:before_state AS jsonb), CAST(:after_state AS jsonb),
+                    CAST(:diff AS jsonb), CAST(:actor_uid AS uuid)
+                )
+                """
+            ),
+            {
+                "uid": audit["uid"],
+                "template_uid": template_uid,
+                "operation": audit["operation"],
+                "status": audit["status"],
+                "version": audit["version"],
+                "target_version": audit["target_version"],
+                "before_state": _json(audit["before_state"])
+                if audit["before_state"] is not None
+                else None,
+                "after_state": _json(audit["after_state"]),
+                "diff": _json(audit["diff"]),
+                "actor_uid": audit["actor_uid"],
+            },
+        )
+        return self.get_template(template_code)

deployment/app/core/governance/domain_templates.py

@@ -0,0 +1,432 @@
+"""Generic governance object contracts and versioned domain templates."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import re
+import uuid
+from collections.abc import Callable
+from copy import deepcopy
+from dataclasses import dataclass
+from typing import Any
+
+from app.core.common.identifiers import new_governance_uid
+
+IDENTIFIER_RE = re.compile(r"^[a-z][a-z0-9_]{1,63}$")
+CODE_RE = re.compile(r"^[A-Za-z][A-Za-z0-9_.-]{1,79}$")
+LIFECYCLE_STATUSES = frozenset({"draft", "active", "retired"})
+FIELD_TYPES = frozenset(
+    {"string", "integer", "number", "boolean", "date", "datetime", "object", "array"}
+)
+SECRET_MARKERS = ("password", "secret", "token", "credential", "api_key", "private_key")
+MAX_OBJECT_TYPES = 100
+MAX_COLLECTION_ITEMS = 500
+
+
+class DomainTemplateValidationError(ValueError):
+    pass
+
+
+class DomainTemplateNotFound(LookupError):
+    pass
+
+
+@dataclass(frozen=True)
+class GovernanceObjectType:
+    """Runtime contract for one template-defined governance object type."""
+
+    template_code: str
+    type_code: str
+    name: str
+    stable_uid_prefix: str
+    source_identity_fields: tuple[str, ...]
+    fields: tuple[dict[str, Any], ...]
+    lifecycle_status: str = "active"
+    current_version: int = 1
+
+    @classmethod
+    def from_definition(
+        cls,
+        template_code: str,
+        definition: dict[str, Any],
+        *,
+        current_version: int = 1,
+    ):
+        normalized = _normalize_object_types([definition])[0]
+        return cls(
+            template_code=_require_identifier(template_code, "template_code"),
+            type_code=normalized["type_code"],
+            name=normalized["name"],
+            stable_uid_prefix=normalized["stable_uid_prefix"],
+            source_identity_fields=tuple(normalized["source_identity_fields"]),
+            fields=tuple(normalized["fields"]),
+            lifecycle_status=normalized["lifecycle_status"],
+            current_version=int(current_version),
+        )
+
+    def stable_uid(self, source_identity: dict[str, Any]) -> str:
+        missing = [
+            field
+            for field in self.source_identity_fields
+            if source_identity.get(field) is None
+            or str(source_identity.get(field)).strip() == ""
+        ]
+        if missing:
+            raise DomainTemplateValidationError(
+                "source_identity is missing required fields: " + ", ".join(missing)
+            )
+        canonical_identity = {
+            field: source_identity[field] for field in self.source_identity_fields
+        }
+        return stable_governance_object_uid(
+            self.template_code,
+            self.type_code,
+            canonical_identity,
+        )
+
+
+def _canonical_json(value: Any) -> str:
+    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+
+
+def _content_hash(value: Any) -> str:
+    return hashlib.sha256(_canonical_json(value).encode("utf-8")).hexdigest()
+
+
+def _require_identifier(value: Any, field: str) -> str:
+    normalized = str(value or "").strip()
+    if not IDENTIFIER_RE.fullmatch(normalized):
+        raise DomainTemplateValidationError(
+            f"{field} must be a lowercase snake_case identifier"
+        )
+    return normalized
+
+
+def _require_code(value: Any, field: str) -> str:
+    normalized = str(value or "").strip()
+    if not CODE_RE.fullmatch(normalized):
+        raise DomainTemplateValidationError(f"{field} must be a stable code")
+    return normalized
+
+
+def _require_text(value: Any, field: str, max_length: int = 200) -> str:
+    normalized = str(value or "").strip()
+    if not normalized or len(normalized) > max_length:
+        raise DomainTemplateValidationError(
+            f"{field} must be between 1 and {max_length} characters"
+        )
+    return normalized
+
+
+def _reject_secrets(value: Any, path: str = "template") -> None:
+    if isinstance(value, dict):
+        for key, nested in value.items():
+            normalized_key = str(key).lower().replace("-", "_")
+            if any(marker in normalized_key for marker in SECRET_MARKERS):
+                raise DomainTemplateValidationError(
+                    f"secret-bearing field is not allowed at {path}.{key}"
+                )
+            _reject_secrets(nested, f"{path}.{key}")
+    elif isinstance(value, list):
+        for index, nested in enumerate(value):
+            _reject_secrets(nested, f"{path}[{index}]")
+
+
+def _normalize_fields(fields: Any, object_code: str) -> list[dict[str, Any]]:
+    if fields is None:
+        fields = []
+    if not isinstance(fields, list) or len(fields) > MAX_COLLECTION_ITEMS:
+        raise DomainTemplateValidationError(f"{object_code}.fields must be a bounded list")
+    normalized = []
+    seen = set()
+    for item in fields:
+        if not isinstance(item, dict):
+            raise DomainTemplateValidationError(f"{object_code}.fields items must be objects")
+        code = _require_identifier(item.get("code"), f"{object_code}.field.code")
+        if any(marker in code for marker in SECRET_MARKERS):
+            raise DomainTemplateValidationError(
+                f"secret-bearing field is not allowed at {object_code}.{code}"
+            )
+        if code in seen:
+            raise DomainTemplateValidationError(f"duplicate field {object_code}.{code}")
+        seen.add(code)
+        field_type = str(item.get("type") or "string").strip().lower()
+        if field_type not in FIELD_TYPES:
+            raise DomainTemplateValidationError(
+                f"{object_code}.{code}.type is unsupported"
+            )
+        normalized.append(
+            {
+                "code": code,
+                "name": _require_text(
+                    item.get("name") or code, f"{object_code}.{code}.name"
+                ),
+                "type": field_type,
+                "required": bool(item.get("required", False)),
+                "description": str(item.get("description") or "").strip(),
+            }
+        )
+    return sorted(normalized, key=lambda item: item["code"])
+
+
+def _normalize_object_types(items: Any) -> list[dict[str, Any]]:
+    if not isinstance(items, list) or not items or len(items) > MAX_OBJECT_TYPES:
+        raise DomainTemplateValidationError("object_types must be a non-empty bounded list")
+    normalized = []
+    seen = set()
+    for item in items:
+        if not isinstance(item, dict):
+            raise DomainTemplateValidationError("object type must be an object")
+        code = _require_identifier(item.get("type_code"), "object_types.type_code")
+        if code in seen:
+            raise DomainTemplateValidationError(f"duplicate object type: {code}")
+        seen.add(code)
+        source_identity_fields = item.get("source_identity_fields")
+        if not isinstance(source_identity_fields, list) or not source_identity_fields:
+            raise DomainTemplateValidationError(
+                f"{code}.source_identity_fields must be a non-empty list"
+            )
+        source_identity_fields = [
+            _require_identifier(value, f"{code}.source_identity_fields")
+            for value in source_identity_fields
+        ]
+        if any(
+            marker in field
+            for field in source_identity_fields
+            for marker in SECRET_MARKERS
+        ):
+            raise DomainTemplateValidationError(
+                f"{code}.source_identity_fields cannot contain secret-bearing fields"
+            )
+        if len(set(source_identity_fields)) != len(source_identity_fields):
+            raise DomainTemplateValidationError(
+                f"{code}.source_identity_fields contains duplicates"
+            )
+        normalized.append(
+            {
+                "type_code": code,
+                "name": _require_text(item.get("name"), f"{code}.name"),
+                "description": str(item.get("description") or "").strip(),
+                "stable_uid_prefix": _require_text(
+                    item.get("stable_uid_prefix"), f"{code}.stable_uid_prefix", 16
+                ).upper(),
+                "source_identity_fields": source_identity_fields,
+                "fields": _normalize_fields(item.get("fields"), code),
+                "lifecycle_status": str(
+                    item.get("lifecycle_status") or "active"
+                ).strip().lower(),
+                "extension": deepcopy(item.get("extension") or {}),
+            }
+        )
+        if normalized[-1]["lifecycle_status"] not in LIFECYCLE_STATUSES:
+            raise DomainTemplateValidationError(
+                f"{code}.lifecycle_status is unsupported"
+            )
+    return sorted(normalized, key=lambda item: item["type_code"])
+
+
+def _normalize_named_collection(value: Any, name: str) -> list[dict[str, Any]]:
+    if value is None:
+        return []
+    if not isinstance(value, list) or len(value) > MAX_COLLECTION_ITEMS:
+        raise DomainTemplateValidationError(f"{name} must be a bounded list")
+    normalized = []
+    seen = set()
+    for item in value:
+        if not isinstance(item, dict):
+            raise DomainTemplateValidationError(f"{name} items must be objects")
+        code = _require_code(item.get("code"), f"{name}.code")
+        if code in seen:
+            raise DomainTemplateValidationError(f"duplicate {name} code: {code}")
+        seen.add(code)
+        normalized.append({**deepcopy(item), "code": code})
+    return sorted(normalized, key=lambda item: item["code"])
+
+
+def normalize_domain_template(definition: Any) -> dict[str, Any]:
+    if not isinstance(definition, dict):
+        raise DomainTemplateValidationError("template definition must be an object")
+    _reject_secrets(definition)
+    lifecycle_status = str(definition.get("lifecycle_status") or "draft").strip().lower()
+    if lifecycle_status not in LIFECYCLE_STATUSES:
+        raise DomainTemplateValidationError("lifecycle_status is unsupported")
+    seed_data = deepcopy(definition.get("seed_data") or [])
+    if not isinstance(seed_data, list) or len(seed_data) > MAX_COLLECTION_ITEMS:
+        raise DomainTemplateValidationError("seed_data must be a bounded list")
+    normalized = {
+        "template_code": _require_identifier(
+            definition.get("template_code"), "template_code"
+        ),
+        "name": _require_text(definition.get("name"), "name"),
+        "description": str(definition.get("description") or "").strip(),
+        "lifecycle_status": lifecycle_status,
+        "object_types": _normalize_object_types(definition.get("object_types")),
+        "responsibility_roles": _normalize_named_collection(
+            definition.get("responsibility_roles"), "responsibility_roles"
+        ),
+        "rules": _normalize_named_collection(definition.get("rules"), "rules"),
+        "metrics": _normalize_named_collection(definition.get("metrics"), "metrics"),
+        "seed_data": seed_data,
+        "source_identity_contract": {
+            "strategy": "uuid5",
+            "canonicalization": "sorted-json",
+            "required": True,
+        },
+        "version_contract": {
+            "strategy": "append-only",
+            "current_version_field": "current_version",
+            "content_hash": "sha256",
+        },
+        "lifecycle_contract": {
+            "statuses": sorted(LIFECYCLE_STATUSES),
+            "transitions": {
+                "draft": ["active", "retired"],
+                "active": ["retired"],
+                "retired": [],
+            },
+        },
+    }
+    normalized["content_hash"] = _content_hash(
+        {key: value for key, value in normalized.items() if key != "content_hash"}
+    )
+    return normalized
+
+
+def stable_governance_object_uid(
+    template_code: str,
+    object_type: str,
+    source_identity: dict[str, Any],
+) -> str:
+    template_code = _require_identifier(template_code, "template_code")
+    object_type = _require_identifier(object_type, "object_type")
+    if not isinstance(source_identity, dict) or not source_identity:
+        raise DomainTemplateValidationError("source_identity must be a non-empty object")
+    if any(value is None or str(value).strip() == "" for value in source_identity.values()):
+        raise DomainTemplateValidationError("source_identity values must be non-empty")
+    identity = f"{template_code}/{object_type}/{_canonical_json(source_identity)}"
+    return str(uuid.uuid5(uuid.NAMESPACE_URL, f"dataops-governance:{identity}"))
+
+
+def diff_domain_templates(
+    before: dict[str, Any] | None,
+    after: dict[str, Any],
+) -> dict[str, Any]:
+    before = before or {}
+    before_types = {
+        item["type_code"]: item for item in before.get("object_types", [])
+    }
+    after_types = {item["type_code"]: item for item in after.get("object_types", [])}
+    shared = set(before_types) & set(after_types)
+    sections = ("responsibility_roles", "rules", "metrics", "seed_data")
+    return {
+        "added_object_types": sorted(set(after_types) - set(before_types)),
+        "removed_object_types": sorted(set(before_types) - set(after_types)),
+        "changed_object_types": sorted(
+            code for code in shared if before_types[code] != after_types[code]
+        ),
+        "changed_sections": sorted(
+            section
+            for section in sections
+            if before.get(section, []) != after.get(section, [])
+        ),
+        "lifecycle_changed": (
+            bool(before)
+            and before.get("lifecycle_status") != after.get("lifecycle_status")
+        ),
+    }
+
+
+class DomainTemplateService:
+    def __init__(
+        self,
+        repository,
+        *,
+        uid_factory: Callable[[], str] = new_governance_uid,
+    ):
+        self.repository = repository
+        self.uid_factory = uid_factory
+
+    def list_templates(self):
+        return self.repository.list_templates()
+
+    def get_template(self, template_code: str):
+        template_code = _require_identifier(template_code, "template_code")
+        result = self.repository.get_template(template_code)
+        if result is None:
+            raise DomainTemplateNotFound(f"domain template {template_code} was not found")
+        return result
+
+    def list_imports(self, template_code: str):
+        template_code = _require_identifier(template_code, "template_code")
+        return self.repository.list_imports(template_code)
+
+    def dry_run(self, definition: Any):
+        normalized = normalize_domain_template(definition)
+        before = self.repository.get_template(normalized["template_code"])
+        return {
+            "valid": True,
+            "template": normalized,
+            "diff": diff_domain_templates(before, normalized),
+        }
+
+    def import_template(
+        self,
+        definition: Any,
+        *,
+        actor_uid: str,
+        operation: str = "import",
+        target_version: int | None = None,
+    ):
+        normalized = normalize_domain_template(definition)
+        before = self.repository.get_template(normalized["template_code"])
+        version = int(before.get("current_version", 0)) + 1 if before else 1
+        persisted = {**normalized, "current_version": version}
+        audit = {
+            "uid": self.uid_factory(),
+            "template_code": normalized["template_code"],
+            "operation": operation,
+            "status": "applied",
+            "version": version,
+            "target_version": target_version,
+            "before_state": deepcopy(before),
+            "after_state": deepcopy(persisted),
+            "diff": diff_domain_templates(before, normalized),
+            "actor_uid": str(actor_uid),
+        }
+        return self.repository.apply_import(
+            {
+                "uid": self.uid_factory(),
+                "template_code": normalized["template_code"],
+                "version": version,
+                "template": persisted,
+                "audit": audit,
+            }
+        )
+
+    def rollback(
+        self,
+        template_code: str,
+        *,
+        target_version: int,
+        actor_uid: str,
+    ):
+        template_code = _require_identifier(template_code, "template_code")
+        if not isinstance(target_version, int) or target_version < 1:
+            raise DomainTemplateValidationError("target_version must be a positive integer")
+        target = self.repository.get_version(template_code, target_version)
+        if target is None:
+            raise DomainTemplateNotFound(
+                f"domain template {template_code} version {target_version} was not found"
+            )
+        definition = {
+            key: deepcopy(value)
+            for key, value in target.items()
+            if key not in {"current_version", "content_hash"}
+        }
+        return self.import_template(
+            definition,
+            actor_uid=actor_uid,
+            operation="rollback",
+            target_version=target_version,
+        )

deployment/app/core/system/permissions.py

@@ -46,10 +46,18 @@ QUALITY_ISSUES_REVIEW = "quality-issues:review"
 DEVICE_OBSERVABILITY_EDIT = "device-observability:edit"
 GOVERNANCE_AUDIT_READ = "governance-audit:read"
 GOVERNANCE_AUDIT_SEAL = "governance-audit:seal"
+DOMAIN_TEMPLATES_READ = "domain-templates:read"
+DOMAIN_TEMPLATES_PREVIEW = "domain-templates:preview"
+DOMAIN_TEMPLATES_MANAGE = "domain-templates:manage"
 
 ROLE_PERMISSIONS = {
     "viewer": frozenset(
-        {READ_GOVERNANCE, RULES_READ, RESPONSIBILITIES_READ}
+        {
+            READ_GOVERNANCE,
+            RULES_READ,
+            RESPONSIBILITIES_READ,
+            DOMAIN_TEMPLATES_READ,
+        }
     ),
     "editor": frozenset(
         {
@@ -71,6 +79,8 @@ ROLE_PERMISSIONS = {
             DEVICE_QUALITY_EXECUTE,
             QUALITY_ISSUES_EDIT,
             DEVICE_OBSERVABILITY_EDIT,
+            DOMAIN_TEMPLATES_READ,
+            DOMAIN_TEMPLATES_PREVIEW,
         }
     ),
     "admin": frozenset(
@@ -114,6 +124,9 @@ ROLE_PERMISSIONS = {
             DEVICE_OBSERVABILITY_EDIT,
             GOVERNANCE_AUDIT_READ,
             GOVERNANCE_AUDIT_SEAL,
+            DOMAIN_TEMPLATES_READ,
+            DOMAIN_TEMPLATES_PREVIEW,
+            DOMAIN_TEMPLATES_MANAGE,
         }
     ),
 }
@@ -134,6 +147,12 @@ def permission_for_request(path: str, method: str) -> tuple[str, ...]:
         if method == "GET":
             return (GOVERNANCE_AUDIT_READ,)
         return (GOVERNANCE_AUDIT_SEAL,)
+    if path.startswith("/api/meta/domain-templates"):
+        if method == "GET":
+            return (DOMAIN_TEMPLATES_READ,)
+        if path == "/api/meta/domain-templates/dry-run":
+            return (DOMAIN_TEMPLATES_PREVIEW,)
+        return (DOMAIN_TEMPLATES_MANAGE,)
     if path in {"/api/knowledge/search", "/api/knowledge/ask"}:
         return (READ_GOVERNANCE,)
     if path.startswith("/api/rules"):

deployment/app/models/__init__.py

@@ -1,28 +1,38 @@
 # Models package initialization
 
 from app.models.data_product import DataOrder, DataProduct
-from app.models.metadata_review import MetadataReviewRecord, MetadataVersionHistory
 from app.models.data_research import (
+    CandidateDecisionRecord,
+    DataElement,
+    DataElementVersion,
     EvidenceFragment,
     ExtractionCandidate,
     IngestionJob,
     IngestionSource,
-    SourceArtifact,
-    DataElement,
-    DataElementVersion,
-    CandidateDecisionRecord,
-    OntologyModel,
-    OntologyVersionModel,
-    OntologyDomainLinkModel,
     OntologyChangeSetModel,
+    OntologyDomainLinkModel,
+    OntologyModel,
     OntologyPublishRunModel,
+    OntologyVersionModel,
+    SourceArtifact,
 )
+from app.models.governance_template import (
+    GovernanceDomainTemplate,
+    GovernanceDomainTemplateImport,
+    GovernanceDomainTemplateVersion,
+    GovernanceObjectType,
+)
+from app.models.metadata_review import MetadataReviewRecord, MetadataVersionHistory
 
 __all__ = [
     "DataOrder",
     "DataProduct",
     "MetadataReviewRecord",
     "MetadataVersionHistory",
+    "GovernanceDomainTemplate",
+    "GovernanceDomainTemplateVersion",
+    "GovernanceObjectType",
+    "GovernanceDomainTemplateImport",
     "IngestionSource",
     "IngestionJob",
     "SourceArtifact",

deployment/app/models/governance_template.py

@@ -0,0 +1,88 @@
+"""SQLAlchemy mappings for governance domain templates."""
+
+from __future__ import annotations
+
+from sqlalchemy.dialects.postgresql import JSONB, UUID
+
+from app import db
+from app.core.common.identifiers import new_governance_uid
+
+
+class GovernanceDomainTemplate(db.Model):
+    __tablename__ = "governance_domain_templates"
+    __table_args__ = {"schema": "public"}
+
+    uid = db.Column(UUID(as_uuid=False), primary_key=True, default=new_governance_uid)
+    template_code = db.Column(db.String(64), unique=True, nullable=False)
+    name = db.Column(db.String(200), nullable=False)
+    lifecycle_status = db.Column(db.String(20), nullable=False)
+    current_version = db.Column(db.Integer, nullable=False)
+    content_hash = db.Column(db.String(64), nullable=False)
+    definition = db.Column(JSONB, nullable=False)
+    created_at = db.Column(db.DateTime(timezone=True), nullable=False)
+    updated_at = db.Column(db.DateTime(timezone=True), nullable=False)
+
+
+class GovernanceDomainTemplateVersion(db.Model):
+    __tablename__ = "governance_domain_template_versions"
+    __table_args__ = (
+        db.UniqueConstraint("template_uid", "version"),
+        {"schema": "public"},
+    )
+
+    uid = db.Column(UUID(as_uuid=False), primary_key=True, default=new_governance_uid)
+    template_uid = db.Column(
+        UUID(as_uuid=False),
+        db.ForeignKey("public.governance_domain_templates.uid"),
+        nullable=False,
+    )
+    version = db.Column(db.Integer, nullable=False)
+    content_hash = db.Column(db.String(64), nullable=False)
+    definition = db.Column(JSONB, nullable=False)
+    actor_uid = db.Column(UUID(as_uuid=False), nullable=False)
+    created_at = db.Column(db.DateTime(timezone=True), nullable=False)
+
+
+class GovernanceObjectType(db.Model):
+    __tablename__ = "governance_object_types"
+    __table_args__ = (
+        db.UniqueConstraint("template_uid", "type_code"),
+        {"schema": "public"},
+    )
+
+    uid = db.Column(UUID(as_uuid=False), primary_key=True, default=new_governance_uid)
+    template_uid = db.Column(
+        UUID(as_uuid=False),
+        db.ForeignKey("public.governance_domain_templates.uid"),
+        nullable=False,
+    )
+    type_code = db.Column(db.String(64), nullable=False)
+    name = db.Column(db.String(200), nullable=False)
+    lifecycle_status = db.Column(db.String(20), nullable=False)
+    current_version = db.Column(db.Integer, nullable=False)
+    stable_uid_prefix = db.Column(db.String(16), nullable=False)
+    source_identity_fields = db.Column(JSONB, nullable=False)
+    definition = db.Column(JSONB, nullable=False)
+    created_at = db.Column(db.DateTime(timezone=True), nullable=False)
+    updated_at = db.Column(db.DateTime(timezone=True), nullable=False)
+
+
+class GovernanceDomainTemplateImport(db.Model):
+    __tablename__ = "governance_domain_template_imports"
+    __table_args__ = {"schema": "public"}
+
+    uid = db.Column(UUID(as_uuid=False), primary_key=True, default=new_governance_uid)
+    template_uid = db.Column(
+        UUID(as_uuid=False),
+        db.ForeignKey("public.governance_domain_templates.uid"),
+        nullable=False,
+    )
+    operation = db.Column(db.String(20), nullable=False)
+    status = db.Column(db.String(20), nullable=False)
+    version = db.Column(db.Integer, nullable=False)
+    target_version = db.Column(db.Integer)
+    before_state = db.Column(JSONB)
+    after_state = db.Column(JSONB, nullable=False)
+    diff = db.Column(JSONB, nullable=False)
+    actor_uid = db.Column(UUID(as_uuid=False), nullable=False)
+    created_at = db.Column(db.DateTime(timezone=True), nullable=False)

docs/DATAOPS_PHASE2_3_MONTH_DEVELOPMENT_PLAN_20260730.md

@@ -195,11 +195,11 @@ P2-WP10 和 P2-WP11 为贯穿性工作包，从第 1 周开始建立门禁，在
 
 **主要工作：**
 
-- [ ] 定义 `GovernanceObjectType`、稳定 UID、来源身份、版本和生命周期契约。
-- [ ] 定义领域模板的对象类型、字段、责任角色、规则、指标和初始化数据格式。
-- [ ] 保持设备域现有 API 与数据兼容，不复制第二套资产服务。
-- [ ] 提供模板 dry-run、导入、差异、失败回滚和审计。
-- [ ] 建立设备域与第二业务域的契约测试。
+- [x] 定义 `GovernanceObjectType`、稳定 UID、来源身份、版本和生命周期契约。
+- [x] 定义领域模板的对象类型、字段、责任角色、规则、指标和初始化数据格式。
+- [x] 保持设备域现有 API 与数据兼容，不复制第二套资产服务。
+- [x] 提供模板 dry-run、导入、差异、失败回滚和审计。
+- [x] 建立设备域与第二业务域的契约测试。
 
 **主要文件区域：**
 

docs/architecture/DATA_MODEL.md

@@ -174,6 +174,10 @@ flowchart LR
 | `ontology_domain_links` | `ontology_uid`, `domain_uid`, `role` | 多业务域 owner/contributor/consumer 关系 |
 | `ontology_change_sets` | `base_version_uid`, `changes`, `decisions`, `status` | 动态建议及人工决策 |
 | `ontology_publish_runs` | `version_uid`, `idempotency_key`, `validation_result`, `status` | 幂等发布运行 |
+| `governance_domain_templates` | `template_code`, `lifecycle_status`, `current_version`, `content_hash`, `definition` | 领域模板稳定身份和当前规范化定义 |
+| `governance_domain_template_versions` | `template_uid`, `version`, `content_hash`, `definition`, `actor_uid` | 追加式领域模板历史版本 |
+| `governance_object_types` | `template_uid`, `type_code`, `source_identity_fields`, `stable_uid_prefix`, `current_version` | 可被业务域初始化的通用治理对象类型 |
+| `governance_domain_template_imports` | `template_uid`, `operation`, `version`, `target_version`, `before_state`, `after_state`, `diff` | 模板导入和回滚审计 |
 
 环境级唯一生效约束：`UNIQUE (dataflow_uid, environment) WHERE status = 'active'`，保证同一环境只允许一个当前生效版本。
 
@@ -261,6 +265,23 @@ SHA-256，再按稳定顺序生成根摘要，最后对封存元数据签名；
 不只检查顶层字段。五年保留目前是设计目标；定时归档、外部时间戳/不可变存储、法务保全
 以及备份恢复证明仍需企业制度和 WP13 交付流程补齐。
 
+## 4.3 P2-WP01 通用治理对象与领域模板
+
+领域模板只定义业务域的对象类型、字段、来源身份、责任角色、规则、指标和初始化数据，
+不创建第二套领域资产服务。每个对象类型通过 `template_code + type_code + canonical
+source_identity` 计算 UUIDv5 稳定身份；来源身份字段必须在模板中显式声明且值非空。
+模板定义使用 canonical JSON 和 SHA-256 内容哈希，版本只追加，不覆盖历史。
+
+`dry-run` 只执行规范化、秘密字段检查和差异计算，不写数据库。正式导入在同一事务内更新
+模板当前版本、追加不可变版本、物化对象类型并写入导入审计；任一步骤失败由 API 回滚整个
+事务。回滚读取目标历史版本，但以新版本重新导入，保留原版本和回滚审计。被新模板移除的
+对象类型只标记为 `retired`，不删除类型或业务数据。
+
+P2-WP00 默认的“备品备件/物料主数据”模板包含物料、物料分类、仓库与库位、库存余额、
+供应商物料映射、设备备件适配关系六类对象，以及三类责任角色、八条质量规则、三项运营
+指标和来源/术语/流程初始化契约。企业真实连接、样本和人员仍需在 P2-WP12 前绑定，模板
+不保存密码、令牌、连接串或真实网络地址。
+
 ## 5. 所有权与删除规则
 
 - PostgreSQL 是身份、权限、映射、任务状态、布局和一致性事件的源真相。
@@ -277,6 +298,7 @@ SHA-256，再按稳定顺序生成根摘要，最后对封存元数据签名；
 - 设备知识检索直接读取授权后的 PostgreSQL canonical 资产、来源映射和运行事件；不建立第二份设备主数据，不读取来源配置和事件原始证据。问答不能替代 WP-09 的证据路径或设备专家根因结论。
 - 治理运营指标是 PostgreSQL canonical 数据的实时只读查询投影；不保存人工覆盖值。指标汇总与明细必须使用同一业务域授权边界，跨域合并只有两端均可见时才能计入非管理员结果。
 - 审计中心只读取六类现有权威记录的安全投影；`governance_audit_seals` 只追加封存摘要和签名，不接收原始问题、凭据、来源配置、自由文本备注或证据正文。
+- 领域模板、模板版本、通用对象类型和导入审计以 PostgreSQL 为源真相；模板只描述对象契约，不替代设备台账或复制第二套资产服务。模板回滚追加新版本，被移除对象类型只退役、不删除。
 - 设备本体、故障/原因/措施代码身份、不可变代码版本和审批记录以 PostgreSQL 为源真相；Neo4j 只接收通过发布门禁的本体投影。
 - `DEVICE_SEMANTIC` 本体发布必须同时通过通用图校验、设备语义覆盖度校验和设备资产负责人校验；代码审批复用同一责任矩阵门禁。
 - 本轮只清理代码和建库脚本。生产表必须在数据核查、备份和依赖确认后以独立变更单下线。

docs/architecture/OPENAPI.yaml

@@ -3,7 +3,7 @@ info:
   title: "DataOps Platform API（当前代码基线）"
   version: "2026-07-16"
   description: "由 scripts/generate_openapi.py 从 app/api/*/routes.py 生成。请求体与响应细节仍以现有专项 API 文档和代码为准。"
-x-route-count: 219
+x-route-count: 225
 servers:
   - url: "http://localhost:15500"
     description: "全本地隔离测试后端"
@@ -4600,6 +4600,159 @@ paths:
             application/json:
               schema:
                 $ref: '#/components/schemas/ApiEnvelope'
+  "/api/meta/domain-templates":
+    get:
+      tags: [meta_data]
+      operationId: meta_data_list_domain_templates_get
+      summary: "list domain templates"
+      x-source: "app/api/meta_data/routes.py"
+      responses:
+        "200":
+          description: "请求已由当前实现处理"
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ApiEnvelope'
+        default:
+          description: "错误响应"
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ApiEnvelope'
+  "/api/meta/domain-templates/dry-run":
+    post:
+      tags: [meta_data]
+      operationId: meta_data_dry_run_domain_template_post
+      summary: "dry run domain template"
+      x-source: "app/api/meta_data/routes.py"
+      requestBody:
+        required: false
+        content:
+          application/json:
+            schema:
+              type: object
+              additionalProperties: true
+      responses:
+        "200":
+          description: "请求已由当前实现处理"
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ApiEnvelope'
+        default:
+          description: "错误响应"
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ApiEnvelope'
+  "/api/meta/domain-templates/import":
+    post:
+      tags: [meta_data]
+      operationId: meta_data_import_domain_template_post
+      summary: "import domain template"
+      x-source: "app/api/meta_data/routes.py"
+      requestBody:
+        required: false
+        content:
+          application/json:
+            schema:
+              type: object
+              additionalProperties: true
+      responses:
+        "200":
+          description: "请求已由当前实现处理"
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ApiEnvelope'
+        default:
+          description: "错误响应"
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ApiEnvelope'
+  "/api/meta/domain-templates/{template_code}":
+    get:
+      tags: [meta_data]
+      operationId: meta_data_get_domain_template_get
+      summary: "get domain template"
+      x-source: "app/api/meta_data/routes.py"
+      parameters:
+        - name: template_code
+          in: path
+          required: true
+          schema:
+            type: string
+      responses:
+        "200":
+          description: "请求已由当前实现处理"
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ApiEnvelope'
+        default:
+          description: "错误响应"
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ApiEnvelope'
+  "/api/meta/domain-templates/{template_code}/imports":
+    get:
+      tags: [meta_data]
+      operationId: meta_data_list_domain_template_imports_get
+      summary: "list domain template imports"
+      x-source: "app/api/meta_data/routes.py"
+      parameters:
+        - name: template_code
+          in: path
+          required: true
+          schema:
+            type: string
+      responses:
+        "200":
+          description: "请求已由当前实现处理"
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ApiEnvelope'
+        default:
+          description: "错误响应"
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ApiEnvelope'
+  "/api/meta/domain-templates/{template_code}/rollback":
+    post:
+      tags: [meta_data]
+      operationId: meta_data_rollback_domain_template_post
+      summary: "rollback domain template"
+      x-source: "app/api/meta_data/routes.py"
+      parameters:
+        - name: template_code
+          in: path
+          required: true
+          schema:
+            type: string
+      requestBody:
+        required: false
+        content:
+          application/json:
+            schema:
+              type: object
+              additionalProperties: true
+      responses:
+        "200":
+          description: "请求已由当前实现处理"
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ApiEnvelope'
+        default:
+          description: "错误响应"
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ApiEnvelope'
   "/api/meta/full/text/query":
     post:
       tags: [meta_data]

docs/phase2/P2_WP01_SPARE_PARTS_DOMAIN_TEMPLATE.json

@@ -0,0 +1,126 @@
+{
+  "template_code": "spare_parts",
+  "name": "备品备件/物料主数据",
+  "description": "P2-WP00 默认第二业务域的可导入领域模板；企业真实样本和人员在复制验收前绑定。",
+  "lifecycle_status": "active",
+  "object_types": [
+    {
+      "type_code": "material",
+      "name": "物料主数据",
+      "stable_uid_prefix": "MAT",
+      "source_identity_fields": ["source_system", "material_code"],
+      "fields": [
+        {"code": "material_code", "name": "物料编码", "type": "string", "required": true},
+        {"code": "name", "name": "物料名称", "type": "string", "required": true},
+        {"code": "specification", "name": "规格", "type": "string", "required": false},
+        {"code": "uom_code", "name": "计量单位", "type": "string", "required": true},
+        {"code": "category_code", "name": "分类编码", "type": "string", "required": true},
+        {"code": "status", "name": "生命周期状态", "type": "string", "required": true}
+      ]
+    },
+    {
+      "type_code": "material_category",
+      "name": "物料分类",
+      "stable_uid_prefix": "MCT",
+      "source_identity_fields": ["source_system", "category_code"],
+      "fields": [
+        {"code": "category_code", "name": "分类编码", "type": "string", "required": true},
+        {"code": "name", "name": "分类名称", "type": "string", "required": true},
+        {"code": "parent_code", "name": "上级分类编码", "type": "string", "required": false}
+      ]
+    },
+    {
+      "type_code": "storage_location",
+      "name": "仓库与库位",
+      "stable_uid_prefix": "LOC",
+      "source_identity_fields": ["source_system", "warehouse_code", "location_code"],
+      "fields": [
+        {"code": "warehouse_code", "name": "仓库编码", "type": "string", "required": true},
+        {"code": "location_code", "name": "库位编码", "type": "string", "required": true},
+        {"code": "name", "name": "库位名称", "type": "string", "required": true}
+      ]
+    },
+    {
+      "type_code": "inventory_balance",
+      "name": "库存余额",
+      "stable_uid_prefix": "INV",
+      "source_identity_fields": ["source_system", "material_code", "warehouse_code", "location_code", "snapshot_at"],
+      "fields": [
+        {"code": "material_code", "name": "物料编码", "type": "string", "required": true},
+        {"code": "warehouse_code", "name": "仓库编码", "type": "string", "required": true},
+        {"code": "location_code", "name": "库位编码", "type": "string", "required": true},
+        {"code": "quantity", "name": "库存数量", "type": "number", "required": true},
+        {"code": "snapshot_at", "name": "快照时间", "type": "datetime", "required": true}
+      ]
+    },
+    {
+      "type_code": "supplier_material_mapping",
+      "name": "供应商物料映射",
+      "stable_uid_prefix": "SUM",
+      "source_identity_fields": ["source_system", "supplier_code", "supplier_material_code"],
+      "fields": [
+        {"code": "supplier_code", "name": "供应商编码", "type": "string", "required": true},
+        {"code": "supplier_material_code", "name": "供应商物料编码", "type": "string", "required": true},
+        {"code": "material_code", "name": "平台物料编码", "type": "string", "required": true}
+      ]
+    },
+    {
+      "type_code": "equipment_compatibility",
+      "name": "设备备件适配关系",
+      "stable_uid_prefix": "ECP",
+      "source_identity_fields": ["source_system", "material_code", "equipment_type"],
+      "fields": [
+        {"code": "material_code", "name": "物料编码", "type": "string", "required": true},
+        {"code": "equipment_type", "name": "设备类型", "type": "string", "required": true},
+        {"code": "evidence_ref", "name": "适配证据引用", "type": "string", "required": true}
+      ]
+    }
+  ],
+  "responsibility_roles": [
+    {"code": "material_domain_owner", "name": "物料域负责人", "raci_role": "accountable"},
+    {"code": "material_data_steward", "name": "物料数据管理员", "raci_role": "responsible"},
+    {"code": "phase2_acceptance_owner", "name": "第二阶段验收负责人", "raci_role": "consulted"}
+  ],
+  "rules": [
+    {"code": "QR-MAT-001", "name": "物料编码必填", "dimension": "COMPLETENESS", "target_object": "material"},
+    {"code": "QR-MAT-002", "name": "物料编码全局唯一", "dimension": "UNIQUENESS", "target_object": "material"},
+    {"code": "QR-MAT-003", "name": "物料名称与规格完整", "dimension": "COMPLETENESS", "target_object": "material"},
+    {"code": "QR-MAT-004", "name": "计量单位符合代码集", "dimension": "VALIDITY", "target_object": "material"},
+    {"code": "QR-MAT-005", "name": "物料分类引用有效", "dimension": "REFERENTIAL_INTEGRITY", "target_object": "material"},
+    {"code": "QR-MAT-006", "name": "库存快照新鲜度", "dimension": "FRESHNESS", "target_object": "inventory_balance"},
+    {"code": "QR-MAT-007", "name": "库存数量非负", "dimension": "VALIDITY", "target_object": "inventory_balance"},
+    {"code": "QR-MAT-008", "name": "设备备件适配关系可追溯", "dimension": "TRACEABILITY", "target_object": "equipment_compatibility"}
+  ],
+  "metrics": [
+    {"code": "material_completeness", "name": "物料主数据完整率", "unit": "%"},
+    {"code": "inventory_freshness", "name": "库存快照新鲜率", "unit": "%"},
+    {"code": "responsibility_coverage", "name": "责任覆盖率", "unit": "%"}
+  ],
+  "seed_data": [
+    {
+      "kind": "source_contract",
+      "records": [
+        {"id": "SRC-MATERIAL-POSTGRES", "type": "PostgreSQL", "binding_status": "ENTERPRISE_SAMPLE_REQUIRED"},
+        {"id": "SRC-INVENTORY-MYSQL", "type": "MySQL", "binding_status": "ENTERPRISE_SAMPLE_REQUIRED"},
+        {"id": "SRC-SUPPLIER-FILE", "type": "controlled_file", "binding_status": "ENTERPRISE_SAMPLE_REQUIRED"}
+      ]
+    },
+    {
+      "kind": "term_and_code_set",
+      "records": [
+        {"id": "TERM-MATERIAL-CATEGORY", "name": "物料分类术语与代码集"},
+        {"id": "CODE-UOM", "name": "计量单位代码集"},
+        {"id": "CODE-MATERIAL-STATUS", "name": "物料生命周期状态代码集"},
+        {"id": "TERM-SPARE-PART-CRITICALITY", "name": "备件关键度术语"}
+      ]
+    },
+    {
+      "kind": "workflow_contract",
+      "records": [
+        {"id": "WF-MAT-001", "name": "物料主数据纠错审批"},
+        {"id": "WF-MAT-002", "name": "物料质量问题整改"},
+        {"id": "WF-MAT-003", "name": "术语与代码集变更"}
+      ]
+    }
+  ]
+}

docs/validation/P2_WP01_DOMAIN_TEMPLATE_EVIDENCE.md

@@ -0,0 +1,70 @@
+# P2-WP01 通用治理对象与领域模板验证记录
+
+## 1. 工程范围
+
+P2-WP01 在现有设备台账、治理责任、质量和指标能力之外增加一层通用定义契约：
+
+1. `GovernanceObjectType` 的稳定类型编码、UUIDv5 对象身份、来源身份字段、版本和生命周期；
+2. 领域模板的对象类型、字段、责任角色、规则、指标和初始化数据格式；
+3. PostgreSQL 模板当前态、不可变历史版本、物化对象类型和导入/回滚审计；
+4. 模板列表、详情、审计、差异预检、导入和追加式回滚 API；
+5. viewer 只读、editor 差异预检、admin 导入/回滚的管理页面；
+6. 与 P2-WP00 对齐的备品备件/物料主数据默认模板。
+
+本工作包没有新建设备或物料资产 CRUD 服务。设备台账 API、数据表和前端页面未修改，
+第二业务域通过领域模板描述六类对象，后续工作包复用平台能力。
+
+## 2. 事务与安全边界
+
+- 模板编码和对象类型编码采用稳定代码；对象来源身份字段必须非空并去重；
+- 稳定对象 UID 基于模板、对象类型和排序后的来源身份 JSON 计算 UUIDv5；
+- 模板拒绝包含 password、secret、token、credential、API Key 或私钥语义的字段；
+- dry-run 只做规范化和差异计算，不写模板、版本或审计；
+- 导入在同一事务内更新当前模板、追加版本、物化对象类型和写审计，失败时整体回滚；
+- 回滚不覆盖旧版本，而是将目标历史内容作为新版本导入并记录目标版本；
+- 模板移除对象类型时只将已物化类型标记为 retired，不删除类型或业务数据；
+- 只有 admin 可以导入和回滚，editor 仅可预检，viewer 仅可读取。
+
+## 3. 默认第二业务域
+
+`docs/phase2/P2_WP01_SPARE_PARTS_DOMAIN_TEMPLATE.json` 固化 P2-WP00 的默认第二业务域：
+
+- 六类对象：物料、物料分类、仓库与库位、库存余额、供应商物料映射、设备备件适配关系；
+- 三类责任角色；
+- 八条质量规则；
+- 三项运营指标；
+- 三个来源契约、四类术语/代码集和三类流程初始化契约。
+
+模板中的企业样本和责任人仍为待绑定状态，不虚构真实姓名、凭据或网络地址。
+
+## 4. 定向验证范围
+
+遵循工作包级验证策略，本次不执行全量回归。定向验证覆盖：
+
+- 模板规范化、稳定 UID、差异、版本、追加式回滚和秘密字段拒绝；
+- 默认第二业务域六对象与八规则契约；
+- viewer/editor/admin API 权限矩阵；
+- 新增迁移的四张表、约束、追加性和 Alembic 头；
+- 管理页面 API、路由和关键操作契约；
+- 既有设备台账服务可导入且设备相关 API 权限与专项测试保持通过；
+- OpenAPI 重新生成并包含六个模板操作；
+- `app/` 与 `deployment/app/` 的新增和触达后端文件一致；
+- 前端生产构建成功。
+
+## 5. 本地工程验证结果（2026-07-30）
+
+本轮只执行 P2-WP01 与直接兼容范围，没有执行全量回归：
+
+- P2-WP01 核心、API、迁移、前端契约、RBAC、设备台账兼容和 OpenAPI 定向测试
+  51 项通过；
+- 本地 PostgreSQL 已由 Alembic `20260730_360` 升级到 `20260730_370`，领域模板与既有
+  设备台账两项真实数据库集成测试通过；
+- 集成测试验证六类默认对象一次性物化、模板 V1/V2/回滚 V3、三条追加式审计，以及在
+  持久化中途模拟异常后事务回滚仍停留在 V1；
+- 本次触达 Python 文件 Ruff 检查通过，新增和触达前端文件 ESLint 检查通过；
+- 前端生产构建成功；仍有项目既有的 Browserslist 数据过期、旧文件 `console`、CSS
+  顺序和包体积警告，本次新增页面没有 ESLint 错误；
+- OpenAPI 由当前路由重新生成，共 225 个操作，包含六个领域模板操作；
+- `app/` 与 `deployment/app/` 的新增和触达后端文件逐字节一致。
+
+以上是本地工程验证，不代表企业第二业务域样本、真实人员、生产权限和发布已经验收。

frontend/src/api/domainTemplates.js

@@ -0,0 +1,13 @@
+import http from '@/utils/request'
+
+const BASE = '/meta/domain-templates'
+
+export const getDomainTemplates = () => http.get(BASE)
+export const getDomainTemplate = code => http.get(`${BASE}/${code}`)
+export const getDomainTemplateImports = code => http.get(`${BASE}/${code}/imports`)
+export const dryRunDomainTemplate = definition => http.post(`${BASE}/dry-run`, definition)
+export const importDomainTemplate = definition => http.post(`${BASE}/import`, definition)
+export const rollbackDomainTemplate = (code, targetVersion) => http.post(
+  `${BASE}/${code}/rollback`,
+  { target_version: targetVersion }
+)

frontend/src/router/routes.js

@@ -228,6 +228,18 @@ export default {
           name: 'dataResearchDeviceAssets',
           alwaysShow: 0
         },
+        {
+          hidden: 1,
+          type: 1,
+          title: '领域模板',
+          path: '/data-governance/development/domain-templates',
+          children: [],
+          label: '领域模板',
+          component: 'dataGovernance/development/domainTemplates',
+          meta: { roles: ['viewer', 'editor', 'admin'], title: '领域模板', readOnly: 'viewer' },
+          name: 'dataResearchDomainTemplates',
+          alwaysShow: 0
+        },
         {
           hidden: 1,
           type: 1,

frontend/src/views/dataGovernance/development/domainTemplates.vue

@@ -0,0 +1,282 @@
+<template>
+  <div class="pa-6 domain-templates">
+    <div class="d-flex flex-wrap align-center mb-5">
+      <div>
+        <h1 class="text-h4 mb-1">领域模板</h1>
+        <div class="text--secondary">
+          用通用对象契约配置业务域，不复制设备域专用服务和页面。
+        </div>
+      </div>
+      <v-spacer />
+      <v-btn outlined color="primary" :loading="loading" @click="load">
+        刷新
+      </v-btn>
+    </div>
+
+    <v-alert type="info" outlined>
+      编辑人员可以执行差异预检；仅管理员可以导入和回滚。导入与回滚均追加新版本并保留审计记录。
+    </v-alert>
+
+    <v-row>
+      <v-col cols="12" lg="5">
+        <v-card outlined>
+          <v-card-title>已登记模板</v-card-title>
+          <v-data-table
+            :headers="headers"
+            :items="templates"
+            :loading="loading"
+            hide-default-footer
+          >
+            <template v-slot:[`item.lifecycle_status`]="{ item }">
+              <v-chip small outlined :color="item.lifecycle_status === 'active' ? 'success' : 'grey'">
+                {{ lifecycleLabel(item.lifecycle_status) }}
+              </v-chip>
+            </template>
+            <template v-slot:[`item.current_version`]="{ item }">
+              V{{ item.current_version }}
+            </template>
+            <template v-slot:[`item.actions`]="{ item }">
+              <v-btn text small color="primary" @click="selectTemplate(item)">
+                查看
+              </v-btn>
+            </template>
+            <template v-slot:no-data>
+              <div class="py-8 text--secondary">尚未导入领域模板</div>
+            </template>
+          </v-data-table>
+        </v-card>
+      </v-col>
+
+      <v-col cols="12" lg="7">
+        <v-card outlined>
+          <v-card-title>模板 JSON</v-card-title>
+          <v-card-text>
+            <v-textarea
+              v-model="definitionText"
+              outlined
+              rows="20"
+              spellcheck="false"
+              label="对象类型、字段、责任角色、规则、指标与初始化数据"
+            />
+            <div class="d-flex flex-wrap">
+              <v-btn
+                color="primary"
+                outlined
+                class="mr-3 mb-2"
+                :loading="previewing"
+                :disabled="readOnly"
+                @click="preview"
+              >
+                差异预检
+              </v-btn>
+              <v-btn
+                color="primary"
+                class="mr-3 mb-2"
+                :loading="importing"
+                :disabled="!isAdmin"
+                @click="importDefinition"
+              >
+                导入并生成新版本
+              </v-btn>
+              <v-btn
+                color="warning"
+                text
+                class="mb-2"
+                :disabled="!isAdmin || !selectedCode"
+                @click="rollbackDialog = true"
+              >
+                回滚为历史版本
+              </v-btn>
+            </div>
+          </v-card-text>
+        </v-card>
+      </v-col>
+    </v-row>
+
+    <v-card v-if="previewResult" outlined class="mt-5">
+      <v-card-title>差异结果</v-card-title>
+      <v-card-text>
+        <v-row>
+          <v-col cols="12" md="4">
+            <div class="field-label">新增对象类型</div>
+            <div>{{ joinValues(previewResult.added_object_types) }}</div>
+          </v-col>
+          <v-col cols="12" md="4">
+            <div class="field-label">变更对象类型</div>
+            <div>{{ joinValues(previewResult.changed_object_types) }}</div>
+          </v-col>
+          <v-col cols="12" md="4">
+            <div class="field-label">移除对象类型</div>
+            <div>{{ joinValues(previewResult.removed_object_types) }}</div>
+          </v-col>
+        </v-row>
+      </v-card-text>
+    </v-card>
+
+    <v-dialog v-model="rollbackDialog" max-width="480">
+      <v-card>
+        <v-card-title>回滚领域模板</v-card-title>
+        <v-card-text>
+          <v-text-field
+            v-model.number="targetVersion"
+            type="number"
+            min="1"
+            label="目标历史版本"
+          />
+          回滚不会覆盖历史记录，而是以目标内容追加一个新版本。
+        </v-card-text>
+        <v-card-actions>
+          <v-spacer />
+          <v-btn text @click="rollbackDialog = false">取消</v-btn>
+          <v-btn color="warning" :loading="rollingBack" @click="rollback">
+            确认回滚
+          </v-btn>
+        </v-card-actions>
+      </v-card>
+    </v-dialog>
+  </div>
+</template>
+
+<script>
+import {
+  dryRunDomainTemplate,
+  getDomainTemplate,
+  getDomainTemplates,
+  importDomainTemplate,
+  rollbackDomainTemplate
+} from '@/api/domainTemplates'
+
+export default {
+  name: 'DomainTemplates',
+  data: () => ({
+    templates: [],
+    definitionText: '',
+    selectedCode: '',
+    previewResult: null,
+    loading: false,
+    previewing: false,
+    importing: false,
+    rollingBack: false,
+    rollbackDialog: false,
+    targetVersion: 1,
+    headers: [
+      { text: '模板', value: 'name' },
+      { text: '编码', value: 'template_code' },
+      { text: '状态', value: 'lifecycle_status' },
+      { text: '版本', value: 'current_version' },
+      { text: '操作', value: 'actions', sortable: false }
+    ]
+  }),
+  computed: {
+    roles () {
+      return this.$store.state.user.userInfo.roles || []
+    },
+    isAdmin () {
+      return this.roles.includes('admin')
+    },
+    readOnly () {
+      return !this.roles.includes('editor') && !this.isAdmin
+    }
+  },
+  created () {
+    this.load()
+  },
+  methods: {
+    async load () {
+      this.loading = true
+      try {
+        const response = await getDomainTemplates()
+        this.templates = response.data || []
+      } catch (error) {
+        this.$snackbar.error(error)
+      } finally {
+        this.loading = false
+      }
+    },
+    async selectTemplate (item) {
+      try {
+        const response = await getDomainTemplate(item.template_code)
+        const detail = response.data
+        this.selectedCode = item.template_code
+        this.targetVersion = Math.max(1, Number(item.current_version) - 1)
+        this.definitionText = JSON.stringify(detail, null, 2)
+        this.previewResult = null
+      } catch (error) {
+        this.$snackbar.error(error)
+      }
+    },
+    parseDefinition () {
+      try {
+        const parsed = JSON.parse(this.definitionText)
+        delete parsed.uid
+        delete parsed.current_version
+        delete parsed.created_at
+        delete parsed.updated_at
+        delete parsed.content_hash
+        return parsed
+      } catch (error) {
+        this.$snackbar.error('模板 JSON 格式不正确')
+        return null
+      }
+    },
+    async preview () {
+      const definition = this.parseDefinition()
+      if (!definition) return
+      this.previewing = true
+      try {
+        const result = await dryRunDomainTemplate(definition)
+        this.previewResult = result.data.diff
+        this.$snackbar.success('差异预检完成，未写入数据')
+      } catch (error) {
+        this.$snackbar.error(error)
+      } finally {
+        this.previewing = false
+      }
+    },
+    async importDefinition () {
+      const definition = this.parseDefinition()
+      if (!definition) return
+      this.importing = true
+      try {
+        const result = await importDomainTemplate(definition)
+        this.selectedCode = result.data.template_code
+        this.$snackbar.success(`模板已导入为 V${result.data.current_version}`)
+        await this.load()
+      } catch (error) {
+        this.$snackbar.error(error)
+      } finally {
+        this.importing = false
+      }
+    },
+    async rollback () {
+      if (!this.selectedCode || Number(this.targetVersion) < 1) return
+      this.rollingBack = true
+      try {
+        const result = await rollbackDomainTemplate(this.selectedCode, Number(this.targetVersion))
+        this.rollbackDialog = false
+        this.$snackbar.success(`已追加回滚版本 V${result.data.current_version}`)
+        await this.load()
+        await this.selectTemplate(result.data)
+      } catch (error) {
+        this.$snackbar.error(error)
+      } finally {
+        this.rollingBack = false
+      }
+    },
+    joinValues (values) {
+      return values && values.length ? values.join('、') : '无'
+    },
+    lifecycleLabel (status) {
+      return { draft: '草稿', active: '生效', retired: '退役' }[status] || status
+    }
+  }
+}
+</script>
+
+<style scoped>
+.field-label {
+  color: rgba(0, 0, 0, 0.6);
+  font-size: 12px;
+  margin-bottom: 4px;
+}
+</style>

frontend/src/views/dataGovernance/development/index.vue

@@ -24,6 +24,7 @@ export default {
       { title: '研发任务', description: '查看解析进度、失败阶段与重试状态', icon: 'mdi-progress-clock', path: '/data-governance/development/tasks' },
       { title: '治理评审', description: '证据预览、批量决策与数据元素生命周期', icon: 'mdi-clipboard-check-outline', path: '/data-governance/development/review', writeOnly: true },
       { title: '设备台账', description: '设备、部件、测点、告警与维护记录统一追溯', icon: 'mdi-factory', path: '/data-governance/development/device-assets' },
+      { title: '领域模板', description: '配置通用对象类型、责任、规则、指标与初始化数据', icon: 'mdi-shape-plus-outline', path: '/data-governance/development/domain-templates' },
       { title: '实体匹配', description: '跨来源匹配候选、审核、非破坏性合并与回滚', icon: 'mdi-vector-link', path: '/data-governance/development/entity-resolution' },
       { title: '设备质量', description: '规则版本、质量检查、违规样本与资产评分', icon: 'mdi-shield-check-outline', path: '/data-governance/development/device-quality' },
       { title: '设备关系与根因', description: '追溯设备运行事件关系，查看有证据约束的根因候选', icon: 'mdi-vector-polyline', path: '/data-governance/development/device-observability' },

migrations/versions/20260730_370_governance_domain_templates.py

@@ -0,0 +1,101 @@
+"""Add generic governance object types and versioned domain templates."""
+
+from alembic import op
+
+revision = "20260730_370"
+down_revision = "20260730_360"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.execute(
+        """
+        CREATE TABLE public.governance_domain_templates (
+            uid UUID PRIMARY KEY,
+            template_code VARCHAR(64) NOT NULL UNIQUE,
+            name VARCHAR(200) NOT NULL,
+            lifecycle_status VARCHAR(20) NOT NULL
+                CHECK (lifecycle_status IN ('draft','active','retired')),
+            current_version INTEGER NOT NULL CHECK (current_version >= 1),
+            content_hash CHAR(64) NOT NULL,
+            definition JSONB NOT NULL,
+            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+            updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+            CHECK (jsonb_typeof(definition) = 'object')
+        );
+
+        CREATE TABLE public.governance_domain_template_versions (
+            uid UUID PRIMARY KEY,
+            template_uid UUID NOT NULL
+                REFERENCES public.governance_domain_templates(uid),
+            version INTEGER NOT NULL CHECK (version >= 1),
+            content_hash CHAR(64) NOT NULL,
+            definition JSONB NOT NULL,
+            actor_uid UUID NOT NULL REFERENCES public.users(id),
+            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+            UNIQUE (template_uid, version),
+            CHECK (jsonb_typeof(definition) = 'object')
+        );
+
+        CREATE TABLE public.governance_object_types (
+            uid UUID PRIMARY KEY,
+            template_uid UUID NOT NULL
+                REFERENCES public.governance_domain_templates(uid),
+            type_code VARCHAR(64) NOT NULL,
+            name VARCHAR(200) NOT NULL,
+            lifecycle_status VARCHAR(20) NOT NULL
+                CHECK (lifecycle_status IN ('draft','active','retired')),
+            current_version INTEGER NOT NULL CHECK (current_version >= 1),
+            stable_uid_prefix VARCHAR(16) NOT NULL,
+            source_identity_fields JSONB NOT NULL,
+            definition JSONB NOT NULL,
+            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+            updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+            UNIQUE (template_uid, type_code),
+            CHECK (jsonb_typeof(source_identity_fields) = 'array'),
+            CHECK (jsonb_array_length(source_identity_fields) > 0),
+            CHECK (jsonb_typeof(definition) = 'object')
+        );
+
+        CREATE TABLE public.governance_domain_template_imports (
+            uid UUID PRIMARY KEY,
+            template_uid UUID NOT NULL
+                REFERENCES public.governance_domain_templates(uid),
+            operation VARCHAR(20) NOT NULL
+                CHECK (operation IN ('import','rollback')),
+            status VARCHAR(20) NOT NULL
+                CHECK (status IN ('applied','failed')),
+            version INTEGER NOT NULL CHECK (version >= 1),
+            target_version INTEGER,
+            before_state JSONB,
+            after_state JSONB NOT NULL,
+            diff JSONB NOT NULL,
+            actor_uid UUID NOT NULL REFERENCES public.users(id),
+            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+            CHECK (before_state IS NULL OR jsonb_typeof(before_state) = 'object'),
+            CHECK (jsonb_typeof(after_state) = 'object'),
+            CHECK (jsonb_typeof(diff) = 'object')
+        );
+
+        CREATE INDEX idx_governance_domain_template_versions_created
+            ON public.governance_domain_template_versions(
+                template_uid, version DESC
+            );
+        CREATE INDEX idx_governance_object_types_status
+            ON public.governance_object_types(
+                template_uid, lifecycle_status, type_code
+            );
+        CREATE INDEX idx_governance_domain_template_imports_created
+            ON public.governance_domain_template_imports(
+                template_uid, created_at DESC
+            );
+        """
+    )
+
+
+def downgrade() -> None:
+    raise RuntimeError(
+        "governance domain template history is append-only; "
+        "schema downgrade requires an approved archival migration"
+    )

tests/core/governance/test_domain_template_frontend_contract.py

@@ -0,0 +1,26 @@
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[3]
+
+
+def test_domain_template_management_surface_is_registered():
+    api = (ROOT / "frontend/src/api/domainTemplates.js").read_text(encoding="utf-8")
+    view = (
+        ROOT
+        / "frontend/src/views/dataGovernance/development/domainTemplates.vue"
+    ).read_text(encoding="utf-8")
+    routes = (ROOT / "frontend/src/router/routes.js").read_text(encoding="utf-8")
+    center = (
+        ROOT / "frontend/src/views/dataGovernance/development/index.vue"
+    ).read_text(encoding="utf-8")
+
+    assert "/meta/domain-templates" in api
+    assert "dry-run" in api
+    assert "rollback" in api
+    assert "差异预检" in view
+    assert "导入并生成新版本" in view
+    assert "仅管理员可以导入和回滚" in view
+    assert "response.data" in view
+    assert "result.data.diff" in view
+    assert "/data-governance/development/domain-templates" in routes
+    assert "领域模板" in center

tests/core/governance/test_domain_templates.py

@@ -0,0 +1,271 @@
+from __future__ import annotations
+
+import json
+from copy import deepcopy
+from pathlib import Path
+
+import pytest
+
+ACTOR_UID = "01900000-0000-7000-8000-000000000101"
+ROOT = Path(__file__).resolve().parents[3]
+
+
+def spare_parts_template():
+    return {
+        "template_code": "spare_parts",
+        "name": "备品备件/物料主数据",
+        "description": "第二业务域默认模板",
+        "lifecycle_status": "active",
+        "object_types": [
+            {
+                "type_code": "spare_part",
+                "name": "备件",
+                "stable_uid_prefix": "SP",
+                "source_identity_fields": ["source_system", "part_code"],
+                "fields": [
+                    {"code": "part_code", "name": "备件编码", "type": "string", "required": True},
+                    {"code": "name", "name": "备件名称", "type": "string", "required": True},
+                ],
+            },
+            {
+                "type_code": "warehouse",
+                "name": "仓库",
+                "stable_uid_prefix": "WH",
+                "source_identity_fields": ["source_system", "warehouse_code"],
+                "fields": [
+                    {"code": "warehouse_code", "name": "仓库编码", "type": "string", "required": True},
+                ],
+            },
+            {
+                "type_code": "inventory_balance",
+                "name": "库存余额",
+                "stable_uid_prefix": "IB",
+                "source_identity_fields": ["source_system", "warehouse_code", "part_code"],
+                "fields": [
+                    {"code": "quantity", "name": "数量", "type": "number", "required": True},
+                ],
+            },
+        ],
+        "responsibility_roles": [
+            {"code": "material_owner", "name": "物料负责人", "raci_role": "accountable"}
+        ],
+        "rules": [{"code": "SP_CODE_REQUIRED", "name": "备件编码必填", "severity": "error"}],
+        "metrics": [{"code": "inventory_completeness", "name": "库存完整率", "unit": "%"}],
+        "seed_data": [{"object_type": "warehouse", "records": []}],
+    }
+
+
+class MemoryRepository:
+    def __init__(self):
+        self.templates = {}
+        self.versions = {}
+        self.imports = []
+
+    def list_templates(self):
+        return list(self.templates.values())
+
+    def get_template(self, template_code):
+        return self.templates.get(template_code)
+
+    def get_version(self, template_code, version):
+        return self.versions.get((template_code, version))
+
+    def apply_import(self, record):
+        self.templates[record["template_code"]] = deepcopy(record["template"])
+        self.versions[(record["template_code"], record["version"])] = deepcopy(
+            record["template"]
+        )
+        self.imports.append(deepcopy(record["audit"]))
+        return deepcopy(record["template"])
+
+    def list_imports(self, template_code):
+        return [
+            item for item in self.imports if item["template_code"] == template_code
+        ]
+
+
+def test_template_normalization_initializes_second_domain_without_device_code():
+    from app.core.governance.domain_templates import (
+        DomainTemplateService,
+        GovernanceObjectType,
+        stable_governance_object_uid,
+    )
+
+    repository = MemoryRepository()
+    service = DomainTemplateService(repository)
+    preview = service.dry_run(spare_parts_template())
+
+    assert preview["valid"] is True
+    assert preview["diff"]["added_object_types"] == [
+        "inventory_balance",
+        "spare_part",
+        "warehouse",
+    ]
+
+    imported = service.import_template(
+        spare_parts_template(),
+        actor_uid=ACTOR_UID,
+    )
+    assert imported["current_version"] == 1
+    assert [item["type_code"] for item in imported["object_types"]] == [
+        "inventory_balance",
+        "spare_part",
+        "warehouse",
+    ]
+    assert imported["source_identity_contract"]["strategy"] == "uuid5"
+
+    uid_a = stable_governance_object_uid(
+        "spare_parts",
+        "spare_part",
+        {"source_system": "erp", "part_code": "P-001"},
+    )
+    uid_b = stable_governance_object_uid(
+        "spare_parts",
+        "spare_part",
+        {"part_code": "P-001", "source_system": "erp"},
+    )
+    assert uid_a == uid_b
+    assert len(uid_a) == 36
+
+    object_type = GovernanceObjectType.from_definition(
+        "spare_parts",
+        imported["object_types"][1],
+        current_version=imported["current_version"],
+    )
+    assert object_type.type_code == "spare_part"
+    assert object_type.stable_uid(
+        {"source_system": "erp", "part_code": "P-001", "ignored": "value"}
+    ) == uid_a
+
+
+def test_dry_run_reports_object_type_and_section_differences_without_writes():
+    from app.core.governance.domain_templates import DomainTemplateService
+
+    repository = MemoryRepository()
+    service = DomainTemplateService(repository)
+    service.import_template(spare_parts_template(), actor_uid=ACTOR_UID)
+    before_imports = len(repository.imports)
+
+    changed = spare_parts_template()
+    changed["object_types"].append(
+        {
+            "type_code": "supplier",
+            "name": "供应商",
+            "stable_uid_prefix": "SU",
+            "source_identity_fields": ["source_system", "supplier_code"],
+            "fields": [],
+        }
+    )
+    changed["metrics"].append(
+        {"code": "stockout_rate", "name": "缺货率", "unit": "%"}
+    )
+
+    preview = service.dry_run(changed)
+    assert preview["diff"]["added_object_types"] == ["supplier"]
+    assert preview["diff"]["removed_object_types"] == []
+    assert preview["diff"]["changed_sections"] == ["metrics"]
+    assert len(repository.imports) == before_imports
+
+
+def test_import_is_versioned_audited_and_rollback_is_append_only():
+    from app.core.governance.domain_templates import DomainTemplateService
+
+    repository = MemoryRepository()
+    service = DomainTemplateService(repository)
+    first = service.import_template(spare_parts_template(), actor_uid=ACTOR_UID)
+
+    changed = spare_parts_template()
+    changed["description"] = "第二业务域默认模板 V2"
+    changed["object_types"][0]["fields"].append(
+        {"code": "specification", "name": "规格", "type": "string", "required": False}
+    )
+    second = service.import_template(changed, actor_uid=ACTOR_UID)
+    rolled_back = service.rollback(
+        "spare_parts",
+        target_version=1,
+        actor_uid=ACTOR_UID,
+    )
+
+    assert first["current_version"] == 1
+    assert second["current_version"] == 2
+    assert rolled_back["current_version"] == 3
+    assert rolled_back["description"] == "第二业务域默认模板"
+    assert [item["operation"] for item in repository.imports] == [
+        "import",
+        "import",
+        "rollback",
+    ]
+    assert repository.imports[-1]["target_version"] == 1
+
+
+@pytest.mark.parametrize(
+    ("mutation", "message"),
+    [
+        (lambda data: data.update(template_code="Device Assets"), "template_code"),
+        (
+            lambda data: data["object_types"][0].update(source_identity_fields=[]),
+            "source_identity_fields",
+        ),
+        (
+            lambda data: data["object_types"].append(deepcopy(data["object_types"][0])),
+            "duplicate object type",
+        ),
+        (
+            lambda data: data.update(seed_data=[{"password": "secret"}]),
+            "secret",
+        ),
+        (
+            lambda data: data["object_types"][0]["fields"].append(
+                {
+                    "code": "api_key",
+                    "name": "API Key",
+                    "type": "string",
+                    "required": False,
+                }
+            ),
+            "secret",
+        ),
+    ],
+)
+def test_template_validation_rejects_unsafe_or_ambiguous_contracts(mutation, message):
+    from app.core.governance.domain_templates import (
+        DomainTemplateValidationError,
+        normalize_domain_template,
+    )
+
+    definition = spare_parts_template()
+    mutation(definition)
+    with pytest.raises(DomainTemplateValidationError, match=message):
+        normalize_domain_template(definition)
+
+
+def test_device_asset_contract_remains_usable_beside_generic_template():
+    from app.core.data_research.device_assets import DeviceAssetService
+    from app.core.governance.domain_templates import normalize_domain_template
+
+    assert DeviceAssetService is not None
+    normalized = normalize_domain_template(spare_parts_template())
+    assert normalized["template_code"] == "spare_parts"
+    assert all(item["type_code"] != "device" for item in normalized["object_types"])
+
+
+def test_default_second_domain_template_matches_wp00_baseline():
+    from app.core.governance.domain_templates import normalize_domain_template
+
+    definition = json.loads(
+        (
+            ROOT / "docs/phase2/P2_WP01_SPARE_PARTS_DOMAIN_TEMPLATE.json"
+        ).read_text(encoding="utf-8")
+    )
+    normalized = normalize_domain_template(definition)
+
+    assert len(normalized["object_types"]) == 6
+    assert len(normalized["rules"]) == 8
+    assert {item["type_code"] for item in normalized["object_types"]} == {
+        "material",
+        "material_category",
+        "storage_location",
+        "inventory_balance",
+        "supplier_material_mapping",
+        "equipment_compatibility",
+    }

tests/integration/test_domain_template_postgres.py

@@ -0,0 +1,139 @@
+from __future__ import annotations
+
+import json
+import os
+import uuid
+from pathlib import Path
+
+import pytest
+from sqlalchemy import text
+
+pytestmark = pytest.mark.integration
+ROOT = Path(__file__).resolve().parents[2]
+
+
+def test_domain_template_import_diff_and_rollback_are_atomic(monkeypatch):
+    database_url = os.environ.get("TEST_DATABASE_URL")
+    if not database_url:
+        pytest.skip("TEST_DATABASE_URL is required")
+
+    monkeypatch.setenv("DATABASE_URL", database_url)
+    from app import create_app, db
+    from app.core.governance.domain_template_repository import (
+        SqlAlchemyDomainTemplateRepository,
+    )
+    from app.core.governance.domain_templates import DomainTemplateService
+
+    app = create_app()
+    app.config.update(TESTING=True)
+    actor_uid = str(uuid.uuid4())
+    definition = json.loads(
+        (
+            ROOT / "docs/phase2/P2_WP01_SPARE_PARTS_DOMAIN_TEMPLATE.json"
+        ).read_text(encoding="utf-8")
+    )
+    definition["template_code"] = f"spare_parts_{uuid.uuid4().hex[:8]}"
+    definition["name"] = f"集成测试模板 {definition['template_code']}"
+    template_uid = None
+    try:
+        with app.app_context():
+            db.session.execute(
+                text(
+                    """
+                    INSERT INTO public.users (
+                        id, username, display_name, password_hash, status
+                    ) VALUES (
+                        CAST(:id AS uuid), :username, :username,
+                        'p2-wp01-integration-hash', 'active'
+                    )
+                    """
+                ),
+                {"id": actor_uid, "username": f"p2-wp01-{actor_uid[:8]}"},
+            )
+            db.session.commit()
+
+            service = DomainTemplateService(
+                SqlAlchemyDomainTemplateRepository(db.session)
+            )
+            first = service.import_template(definition, actor_uid=actor_uid)
+            db.session.commit()
+            template_uid = first["uid"]
+
+            changed = dict(definition)
+            changed["description"] = "集成测试第二版"
+            preview = service.dry_run(changed)
+            assert preview["diff"]["added_object_types"] == []
+
+            class FailingRepository(SqlAlchemyDomainTemplateRepository):
+                def apply_import(self, record):
+                    super().apply_import(record)
+                    raise RuntimeError("simulated failure after persistence")
+
+            failing_service = DomainTemplateService(FailingRepository(db.session))
+            with pytest.raises(RuntimeError, match="simulated failure"):
+                failing_service.import_template(changed, actor_uid=actor_uid)
+            db.session.rollback()
+            assert service.get_template(
+                definition["template_code"]
+            )["current_version"] == 1
+
+            second = service.import_template(changed, actor_uid=actor_uid)
+            db.session.commit()
+            rolled_back = service.rollback(
+                definition["template_code"],
+                target_version=1,
+                actor_uid=actor_uid,
+            )
+            db.session.commit()
+
+            assert first["current_version"] == 1
+            assert second["current_version"] == 2
+            assert rolled_back["current_version"] == 3
+            assert rolled_back["description"] == definition["description"]
+            assert len(rolled_back["object_types"]) == 6
+            assert [item["operation"] for item in service.list_imports(
+                definition["template_code"]
+            )] == ["rollback", "import", "import"]
+
+            counts = db.session.execute(
+                text(
+                    """
+                    SELECT
+                      (SELECT count(*) FROM public.governance_object_types
+                       WHERE template_uid = CAST(:uid AS uuid)) AS object_types,
+                      (SELECT count(*) FROM public.governance_domain_template_versions
+                       WHERE template_uid = CAST(:uid AS uuid)) AS versions,
+                      (SELECT count(*) FROM public.governance_domain_template_imports
+                       WHERE template_uid = CAST(:uid AS uuid)) AS imports
+                    """
+                ),
+                {"uid": template_uid},
+            ).mappings().one()
+            assert dict(counts) == {
+                "object_types": 6,
+                "versions": 3,
+                "imports": 3,
+            }
+    finally:
+        with app.app_context():
+            db.session.rollback()
+            if template_uid:
+                for statement in (
+                    "DELETE FROM public.governance_domain_template_imports "
+                    "WHERE template_uid = CAST(:uid AS uuid)",
+                    "DELETE FROM public.governance_object_types "
+                    "WHERE template_uid = CAST(:uid AS uuid)",
+                    "DELETE FROM public.governance_domain_template_versions "
+                    "WHERE template_uid = CAST(:uid AS uuid)",
+                    "DELETE FROM public.governance_domain_templates "
+                    "WHERE uid = CAST(:uid AS uuid)",
+                ):
+                    db.session.execute(
+                        text(statement),
+                        {"uid": template_uid},
+                    )
+            db.session.execute(
+                text("DELETE FROM public.users WHERE id = CAST(:id AS uuid)"),
+                {"id": actor_uid},
+            )
+            db.session.commit()

tests/test_domain_template_api.py

@@ -0,0 +1,106 @@
+from __future__ import annotations
+
+ACTOR_UID = "01900000-0000-7000-8000-000000000101"
+
+
+class FakeService:
+    def __init__(self):
+        self.calls = []
+
+    def list_templates(self):
+        self.calls.append(("list",))
+        return [{"template_code": "spare_parts", "current_version": 1}]
+
+    def get_template(self, template_code):
+        self.calls.append(("get", template_code))
+        return {"template_code": template_code, "current_version": 1}
+
+    def dry_run(self, definition):
+        self.calls.append(("dry_run", definition))
+        return {"valid": True, "diff": {"added_object_types": ["spare_part"]}}
+
+    def import_template(self, definition, *, actor_uid):
+        self.calls.append(("import", definition, actor_uid))
+        return {"template_code": definition["template_code"], "current_version": 1}
+
+    def rollback(self, template_code, *, target_version, actor_uid):
+        self.calls.append(("rollback", template_code, target_version, actor_uid))
+        return {"template_code": template_code, "current_version": 2}
+
+    def list_imports(self, template_code):
+        self.calls.append(("audits", template_code))
+        return [{"operation": "import", "template_code": template_code}]
+
+
+def _headers(role):
+    return {"Authorization": f"Bearer {role}"}
+
+
+def test_domain_template_api_is_readable_and_admin_managed(monkeypatch):
+    from app import create_app
+    from app.api.meta_data import domain_templates
+
+    service = FakeService()
+    monkeypatch.setattr(domain_templates, "_service", lambda: service)
+    monkeypatch.setattr(
+        "app.core.system.auth.load_identity_from_token",
+        lambda token, secret: (
+            {"id": ACTOR_UID, "username": token, "roles": [token]}
+            if token in {"viewer", "editor", "admin"}
+            else None
+        ),
+    )
+    app = create_app()
+    app.config.update(TESTING=True)
+    client = app.test_client()
+
+    listed = client.get("/api/meta/domain-templates", headers=_headers("viewer"))
+    assert listed.status_code == 200
+
+    preview = client.post(
+        "/api/meta/domain-templates/dry-run",
+        json={"template_code": "spare_parts"},
+        headers=_headers("editor"),
+    )
+    assert preview.status_code == 200
+
+    forbidden = client.post(
+        "/api/meta/domain-templates/import",
+        json={"template_code": "spare_parts"},
+        headers=_headers("editor"),
+    )
+    assert forbidden.status_code == 403
+
+    imported = client.post(
+        "/api/meta/domain-templates/import",
+        json={"template_code": "spare_parts"},
+        headers=_headers("admin"),
+    )
+    assert imported.status_code == 200
+    assert service.calls[-1][2] == ACTOR_UID
+
+    rolled_back = client.post(
+        "/api/meta/domain-templates/spare_parts/rollback",
+        json={"target_version": 1},
+        headers=_headers("admin"),
+    )
+    assert rolled_back.status_code == 200
+
+
+def test_domain_template_paths_have_dedicated_permissions():
+    from app.core.system.permissions import (
+        DOMAIN_TEMPLATES_MANAGE,
+        DOMAIN_TEMPLATES_PREVIEW,
+        DOMAIN_TEMPLATES_READ,
+        permission_for_request,
+    )
+
+    assert permission_for_request("/api/meta/domain-templates", "GET") == (
+        DOMAIN_TEMPLATES_READ,
+    )
+    assert permission_for_request("/api/meta/domain-templates/dry-run", "POST") == (
+        DOMAIN_TEMPLATES_PREVIEW,
+    )
+    assert permission_for_request("/api/meta/domain-templates/import", "POST") == (
+        DOMAIN_TEMPLATES_MANAGE,
+    )

tests/test_phase2_wp01_migration_contract.py

@@ -0,0 +1,40 @@
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def test_domain_template_migration_is_additive_versioned_and_auditable():
+    migration = (
+        ROOT
+        / "migrations"
+        / "versions"
+        / "20260730_370_governance_domain_templates.py"
+    ).read_text(encoding="utf-8")
+
+    assert 'revision = "20260730_370"' in migration
+    assert 'down_revision = "20260730_360"' in migration
+    for table in (
+        "governance_domain_templates",
+        "governance_domain_template_versions",
+        "governance_object_types",
+        "governance_domain_template_imports",
+    ):
+        assert f"CREATE TABLE public.{table}" in migration
+    assert "definition JSONB NOT NULL" in migration
+    assert "source_identity_fields JSONB NOT NULL" in migration
+    assert "before_state JSONB" in migration
+    assert "after_state JSONB NOT NULL" in migration
+    assert "operation IN ('import','rollback')" in migration
+    assert "DROP TABLE" not in migration.upper()
+
+
+def test_release_copy_will_include_domain_template_runtime():
+    for relative in (
+        "core/governance/domain_templates.py",
+        "core/governance/domain_template_repository.py",
+        "api/meta_data/domain_templates.py",
+        "models/governance_template.py",
+    ):
+        assert (ROOT / "app" / relative).read_bytes() == (
+            ROOT / "deployment" / "app" / relative
+        ).read_bytes()