feat: deliver V63 ontology MVP

app/api/data_development/routes.py

@@ -91,6 +91,29 @@ def get_evidence_service():
     return EvidenceService(db.session, _artifact_storage())
 
 
+def get_ontology_service():
+    from app.core.data_research.ontology.publication import (
+        OntologyApplicationService,
+        OntologyPublicationService,
+    )
+    from app.core.data_research.ontology.repository import SqlAlchemyOntologyRepository
+    from app.core.events.outbox import enqueue_outbox
+
+    repository = SqlAlchemyOntologyRepository(db.session)
+    publication = OntologyPublicationService(
+        repository,
+        outbox_enqueue=lambda **event: enqueue_outbox(db.session, **event),
+        commit=db.session.commit,
+        rollback=db.session.rollback,
+    )
+    return OntologyApplicationService(
+        repository,
+        publication=publication,
+        commit=db.session.commit,
+        rollback=db.session.rollback,
+    )
+
+
 def _identity():
     return getattr(g, "current_user", {}) or {}
 
@@ -152,6 +175,35 @@ def _artifact(record):
     }
 
 
+def _ontology(record):
+    return {
+        "uid": str(record.uid),
+        "code": record.code,
+        "name": record.name,
+        "owner_uid": record.owner_uid,
+        "status": record.status,
+        "draft_revision": int(record.draft_revision),
+        "active_version_uid": record.active_version_uid,
+        "domain_links": [
+            {"domain_uid": link.domain_uid, "role": link.role}
+            for link in record.domain_links
+        ],
+    }
+
+
+def _ontology_version(record):
+    return {
+        "uid": str(record.uid),
+        "ontology_uid": str(record.ontology_uid),
+        "version": int(record.version),
+        "parent_version_uid": record.parent_version_uid,
+        "status": record.status,
+        "content_hash": record.content_hash,
+        "graph_document": record.graph_document.to_dict(),
+        "created_by": record.created_by,
+    }
+
+
 def _error(error):
     if isinstance(error, DataResearchError):
         return (
@@ -330,3 +382,93 @@ def get_evidence(evidence_uid):
         return jsonify(success(preview)), 200
     except Exception as error:
         return _error(error)
+
+
+@bp.route("/ontologies", methods=["GET"])
+def list_ontologies():
+    try:
+        return jsonify(success([_ontology(item) for item in get_ontology_service().list()])), 200
+    except Exception as error:
+        return _error(error)
+
+
+@bp.route("/ontologies", methods=["POST"])
+def create_ontology():
+    try:
+        record = get_ontology_service().create(
+            request.get_json(silent=True) or {},
+            _identity().get("id") or _identity().get("sub"),
+        )
+        return jsonify(success(_ontology(record))), 201
+    except Exception as error:
+        return _error(error)
+
+
+@bp.route("/ontologies/<ontology_uid>/graph", methods=["PATCH"])
+def save_ontology_graph(ontology_uid):
+    etag = str(request.headers.get("If-Match") or "").strip().strip('"')
+    if not etag.isdigit():
+        return jsonify(failed("缺少有效 If-Match 修订号", code=428)), 428
+    try:
+        version = get_ontology_service().save_draft(
+            ontology_uid,
+            request.get_json(silent=True) or {},
+            int(etag),
+            _identity().get("id") or _identity().get("sub"),
+        )
+        response = jsonify(success(_ontology_version(version)))
+        response.headers["ETag"] = f'"{int(etag) + 1}"'
+        return response, 200
+    except Exception as error:
+        return _error(error)
+
+
+@bp.route("/ontologies/<ontology_uid>/validate", methods=["POST"])
+def validate_ontology(ontology_uid):
+    try:
+        issues = get_ontology_service().validate(ontology_uid)
+        data = [
+            {"code": item.code, "message": item.message, "path": item.path}
+            for item in issues
+        ]
+        return jsonify(success(data)), 422 if data else 200
+    except Exception as error:
+        return _error(error)
+
+
+@bp.route("/ontologies/<ontology_uid>/publish", methods=["POST"])
+def publish_ontology(ontology_uid):
+    try:
+        version = get_ontology_service().publish(
+            ontology_uid,
+            request.headers.get("Idempotency-Key") or f"{ontology_uid}:publish",
+            _identity().get("id") or _identity().get("sub"),
+        )
+        return jsonify(success(_ontology_version(version))), 200
+    except Exception as error:
+        return _error(error)
+
+
+@bp.route("/ontologies/<ontology_uid>/diff", methods=["GET"])
+def diff_ontology(ontology_uid):
+    try:
+        return jsonify(success(get_ontology_service().diff(
+            ontology_uid, request.args.get("left"), request.args.get("right")
+        ))), 200
+    except Exception as error:
+        return _error(error)
+
+
+@bp.route("/ontologies/<ontology_uid>/rollback", methods=["POST"])
+def rollback_ontology(ontology_uid):
+    payload = request.get_json(silent=True) or {}
+    try:
+        version = get_ontology_service().rollback(
+            ontology_uid,
+            payload.get("target_version_uid"),
+            int(payload.get("expected_revision")),
+            _identity().get("id") or _identity().get("sub"),
+        )
+        return jsonify(success(_ontology_version(version))), 201
+    except Exception as error:
+        return _error(error)

app/core/data_research/ontology/__init__.py

@@ -0,0 +1,2 @@
+"""Versioned ontology governance services."""
+

app/core/data_research/ontology/models.py

@@ -0,0 +1,82 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+
+GRAPH_SECTIONS = (
+    "classes",
+    "properties",
+    "relations",
+    "constraints",
+    "domain_links",
+    "element_mappings",
+)
+
+
+@dataclass(frozen=True)
+class GraphDocument:
+    classes: tuple[dict[str, Any], ...] = ()
+    properties: tuple[dict[str, Any], ...] = ()
+    relations: tuple[dict[str, Any], ...] = ()
+    constraints: tuple[dict[str, Any], ...] = ()
+    domain_links: tuple[dict[str, Any], ...] = ()
+    element_mappings: tuple[dict[str, Any], ...] = ()
+
+    @classmethod
+    def from_dict(cls, payload):
+        payload = dict(payload or {})
+        return cls(**{name: tuple(dict(item) for item in payload.get(name, ())) for name in GRAPH_SECTIONS})
+
+    def to_dict(self):
+        return {name: [dict(item) for item in getattr(self, name)] for name in GRAPH_SECTIONS}
+
+
+@dataclass(frozen=True)
+class DomainLink:
+    domain_uid: str
+    role: str
+
+
+@dataclass(frozen=True)
+class Ontology:
+    uid: str
+    code: str
+    name: str
+    owner_uid: str
+    domain_links: tuple[DomainLink, ...]
+    status: str = "draft"
+    draft_revision: int = 0
+    active_version_uid: str | None = None
+
+
+@dataclass(frozen=True)
+class OntologyVersion:
+    uid: str
+    ontology_uid: str
+    version: int
+    graph_document: GraphDocument
+    content_hash: str
+    status: str = "draft"
+    parent_version_uid: str | None = None
+    created_by: str | None = None
+
+
+@dataclass(frozen=True)
+class OntologyChangeSet:
+    uid: str
+    ontology_uid: str
+    changes: tuple[dict[str, Any], ...]
+    decisions: tuple[dict[str, Any], ...] = ()
+    status: str = "draft"
+
+
+@dataclass(frozen=True)
+class OntologyPublishRun:
+    uid: str
+    ontology_uid: str
+    version_uid: str
+    idempotency_key: str
+    status: str
+    validation_result: dict[str, Any]
+

app/core/data_research/ontology/projection.py

@@ -0,0 +1,26 @@
+from __future__ import annotations
+
+
+class OntologyGraphProjector:
+    STATEMENT = """
+    MERGE (ontology:Ontology {uid: $ontology_uid})
+    SET ontology.active_version_uid = $version_uid,
+        ontology.active_version = $version
+    WITH ontology
+    MERGE (version:OntologyVersion {uid: $version_uid})
+    SET version.version = $version,
+        version.graph_document = $graph_document
+    MERGE (ontology)-[:ACTIVE_VERSION]->(version)
+    """
+
+    def __init__(self, graph_session):
+        self.graph_session = graph_session
+
+    def project(self, payload):
+        self.graph_session.run(
+            self.STATEMENT,
+            ontology_uid=str(payload["ontology_uid"]),
+            version_uid=str(payload["version_uid"]),
+            version=int(payload["version"]),
+            graph_document=payload["graph_document"],
+        )

app/core/data_research/ontology/publication.py

@@ -0,0 +1,166 @@
+from __future__ import annotations
+
+from app.core.data_research.ontology.models import GRAPH_SECTIONS, GraphDocument
+from app.core.data_research.ontology.validation import validate_graph
+
+
+class OntologyValidationFailed(ValueError):
+    def __init__(self, issues):
+        super().__init__("ontology validation failed")
+        self.issues = tuple(issues)
+
+
+def diff_graphs(before, after):
+    old = GraphDocument.from_dict(before).to_dict()
+    new = GraphDocument.from_dict(after).to_dict()
+    result = {}
+    for section in GRAPH_SECTIONS:
+        old_by_uid = {str(item.get("uid") or item.get("property_uid") or item.get("domain_uid")): item for item in old[section]}
+        new_by_uid = {str(item.get("uid") or item.get("property_uid") or item.get("domain_uid")): item for item in new[section]}
+        result[section] = {
+            "added": sorted(new_by_uid.keys() - old_by_uid.keys()),
+            "removed": sorted(old_by_uid.keys() - new_by_uid.keys()),
+            "changed": sorted(uid for uid in old_by_uid.keys() & new_by_uid.keys() if old_by_uid[uid] != new_by_uid[uid]),
+        }
+    return result
+
+
+class OntologyPublicationService:
+    def __init__(
+        self,
+        repository,
+        *,
+        outbox_enqueue=lambda **_event: None,
+        commit=lambda: None,
+        rollback=lambda: None,
+    ):
+        self.repository = repository
+        self.outbox_enqueue = outbox_enqueue
+        self.commit = commit
+        self.rollback = rollback
+        self._published_by_key = {}
+
+    def publish(self, version_uid, *, idempotency_key, actor_uid):
+        if idempotency_key in self._published_by_key:
+            return self._published_by_key[idempotency_key]
+        try:
+            version = self.repository.get_version(version_uid)
+            if version is None:
+                raise LookupError("ontology version was not found")
+            issues = validate_graph(version.graph_document.to_dict())
+            if issues:
+                raise OntologyValidationFailed(issues)
+            published = self.repository.mark_published(version_uid)
+            self.outbox_enqueue(
+                aggregate_type="ontology",
+                aggregate_id=published.ontology_uid,
+                event_type="ontology.version_published",
+                payload={
+                    "ontology_uid": published.ontology_uid,
+                    "version_uid": published.uid,
+                    "version": published.version,
+                    "status": "published",
+                    "graph_document": published.graph_document.to_dict(),
+                    "actor_uid": actor_uid,
+                },
+            )
+            self.commit()
+            self._published_by_key[idempotency_key] = published
+            return published
+        except Exception:
+            self.rollback()
+            raise
+
+    def create_rollback(
+        self,
+        ontology_uid,
+        *,
+        target_version_uid,
+        expected_revision,
+        actor_uid,
+    ):
+        target = self.repository.get_version(target_version_uid)
+        if target is None or target.ontology_uid != str(ontology_uid):
+            raise LookupError("rollback target was not found")
+        return self.repository.save_draft(
+            ontology_uid,
+            target.graph_document.to_dict(),
+            expected_revision=expected_revision,
+            actor_uid=actor_uid,
+        )
+
+
+class OntologyApplicationService:
+    def __init__(self, repository, *, publication=None, commit=lambda: None, rollback=lambda: None):
+        self.repository = repository
+        self.publication = publication or OntologyPublicationService(repository, commit=commit, rollback=rollback)
+        self.commit = commit
+        self.rollback_transaction = rollback
+
+    def list(self):
+        return self.repository.list()
+
+    def create(self, payload, actor_uid):
+        try:
+            ontology = self.repository.create(
+                code=str(payload.get("code") or "").strip(),
+                name=str(payload.get("name") or "").strip(),
+                owner_uid=str(payload.get("owner_uid") or actor_uid),
+                domain_links=list(payload.get("domain_links") or []),
+                created_by=actor_uid,
+            )
+            self.commit()
+            return ontology
+        except Exception:
+            self.rollback_transaction()
+            raise
+
+    def save_draft(self, uid, graph, expected_revision, actor_uid):
+        try:
+            version = self.repository.save_draft(
+                uid, graph, expected_revision=expected_revision, actor_uid=actor_uid
+            )
+            self.commit()
+            return version
+        except Exception:
+            self.rollback_transaction()
+            raise
+
+    def validate(self, uid):
+        version = self.repository.latest_version(uid)
+        if version is None:
+            return validate_graph({})
+        return validate_graph(version.graph_document.to_dict())
+
+    def publish(self, uid, idempotency_key, actor_uid):
+        version = self.repository.latest_version(uid)
+        if version is None:
+            raise LookupError("ontology draft was not found")
+        return self.publication.publish(
+            version.uid, idempotency_key=idempotency_key, actor_uid=actor_uid
+        )
+
+    def diff(self, uid, left, right):
+        left_version = self.repository.get_version(left)
+        right_version = self.repository.get_version(right)
+        if not left_version or not right_version:
+            raise LookupError("ontology diff version was not found")
+        if left_version.ontology_uid != str(uid) or right_version.ontology_uid != str(uid):
+            raise LookupError("ontology diff version mismatch")
+        return diff_graphs(
+            left_version.graph_document.to_dict(), right_version.graph_document.to_dict()
+        )
+
+    def rollback(self, uid, target_version_uid, expected_revision, actor_uid):
+        try:
+            version = self.publication.create_rollback(
+                uid,
+                target_version_uid=target_version_uid,
+                expected_revision=expected_revision,
+                actor_uid=actor_uid,
+            )
+            self.commit()
+            return version
+        except Exception:
+            self.rollback_transaction()
+            raise

app/core/data_research/ontology/repository.py

@@ -0,0 +1,243 @@
+from __future__ import annotations
+
+import hashlib
+import json
+from dataclasses import replace
+
+from app.core.common.identifiers import new_governance_uid
+from app.core.data_research.ontology.models import (
+    DomainLink,
+    GraphDocument,
+    Ontology,
+    OntologyVersion,
+)
+
+
+class OntologyRepositoryError(ValueError):
+    pass
+
+
+class OntologyConflict(OntologyRepositoryError):
+    pass
+
+
+class OntologyImmutable(OntologyRepositoryError):
+    pass
+
+
+def canonical_graph_hash(graph):
+    document = GraphDocument.from_dict(graph).to_dict()
+    encoded = json.dumps(document, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+    return hashlib.sha256(encoded.encode("utf-8")).hexdigest()
+
+
+class MemoryOntologyRepository:
+    def __init__(self, *, uid_factory=new_governance_uid):
+        self.uid_factory = uid_factory
+        self.ontologies = {}
+        self.versions = {}
+
+    def create(self, *, code, name, owner_uid, domain_links, created_by=None):
+        if any(item.code == code for item in self.ontologies.values()):
+            raise OntologyConflict("ontology code already exists")
+        links = tuple(DomainLink(str(item["domain_uid"]), str(item["role"])) for item in domain_links)
+        ontology = Ontology(
+            uid=self.uid_factory(),
+            code=str(code),
+            name=str(name),
+            owner_uid=str(owner_uid),
+            domain_links=links,
+        )
+        self.ontologies[ontology.uid] = ontology
+        return ontology
+
+    def get(self, uid):
+        return self.ontologies.get(str(uid))
+
+    def get_version(self, uid):
+        return self.versions.get(str(uid))
+
+    def save_draft(self, ontology_uid, graph, *, expected_revision, actor_uid):
+        ontology = self.get(ontology_uid)
+        if ontology is None:
+            raise LookupError("ontology was not found")
+        if ontology.draft_revision != int(expected_revision):
+            raise OntologyConflict("ontology draft revision conflict")
+        version_number = 1 + max(
+            (item.version for item in self.versions.values() if item.ontology_uid == ontology.uid),
+            default=0,
+        )
+        version = OntologyVersion(
+            uid=self.uid_factory(),
+            ontology_uid=ontology.uid,
+            version=version_number,
+            parent_version_uid=ontology.active_version_uid,
+            graph_document=GraphDocument.from_dict(graph),
+            content_hash=canonical_graph_hash(graph),
+            created_by=actor_uid,
+        )
+        self.versions[version.uid] = version
+        self.ontologies[ontology.uid] = replace(ontology, draft_revision=ontology.draft_revision + 1)
+        return version
+
+    def mark_published(self, version_uid):
+        version = self.get_version(version_uid)
+        if version is None:
+            raise LookupError("ontology version was not found")
+        published = replace(version, status="published")
+        self.versions[published.uid] = published
+        ontology = self.get(published.ontology_uid)
+        self.ontologies[ontology.uid] = replace(
+            ontology, status="published", active_version_uid=published.uid
+        )
+        return published
+
+    def replace_version(self, version_uid, graph):
+        version = self.get_version(version_uid)
+        if version.status == "published":
+            raise OntologyImmutable("published ontology version is immutable")
+        updated = replace(
+            version,
+            graph_document=GraphDocument.from_dict(graph),
+            content_hash=canonical_graph_hash(graph),
+        )
+        self.versions[updated.uid] = updated
+        return updated
+
+
+class SqlAlchemyOntologyRepository:
+    def __init__(self, session, *, uid_factory=new_governance_uid):
+        self.session = session
+        self.uid_factory = uid_factory
+
+    @staticmethod
+    def _version(model):
+        return OntologyVersion(
+            uid=str(model.uid),
+            ontology_uid=str(model.ontology_uid),
+            version=int(model.version),
+            parent_version_uid=str(model.parent_version_uid) if model.parent_version_uid else None,
+            status=model.status,
+            graph_document=GraphDocument.from_dict(model.graph_document),
+            content_hash=model.content_hash,
+            created_by=model.created_by,
+        )
+
+    def _ontology(self, model):
+        from app.models.data_research import OntologyDomainLinkModel
+
+        links = self.session.query(OntologyDomainLinkModel).filter_by(
+            ontology_uid=str(model.uid)
+        ).all()
+        return Ontology(
+            uid=str(model.uid),
+            code=model.code,
+            name=model.name,
+            owner_uid=model.owner_uid,
+            status=model.status,
+            draft_revision=int(model.draft_revision),
+            active_version_uid=str(model.active_version_uid) if model.active_version_uid else None,
+            domain_links=tuple(DomainLink(str(item.domain_uid), item.role) for item in links),
+        )
+
+    def list(self):
+        from app.models.data_research import OntologyModel
+
+        return [self._ontology(model) for model in self.session.query(OntologyModel).order_by(OntologyModel.code).all()]
+
+    def create(self, *, code, name, owner_uid, domain_links, created_by=None):
+        from app.models.data_research import OntologyDomainLinkModel, OntologyModel
+
+        if self.session.query(OntologyModel).filter_by(code=str(code)).first():
+            raise OntologyConflict("ontology code already exists")
+        uid = self.uid_factory()
+        model = OntologyModel(
+            uid=uid,
+            code=str(code),
+            name=str(name),
+            owner_uid=str(owner_uid),
+            created_by=created_by,
+        )
+        self.session.add(model)
+        for link in domain_links:
+            self.session.add(
+                OntologyDomainLinkModel(
+                    ontology_uid=uid,
+                    domain_uid=str(link["domain_uid"]),
+                    role=str(link["role"]),
+                )
+            )
+        self.session.flush()
+        return self._ontology(model)
+
+    def get(self, uid):
+        from app.models.data_research import OntologyModel
+
+        model = self.session.get(OntologyModel, str(uid))
+        return self._ontology(model) if model is not None else None
+
+    def get_version(self, uid):
+        from app.models.data_research import OntologyVersionModel
+
+        model = self.session.get(OntologyVersionModel, str(uid))
+        return self._version(model) if model is not None else None
+
+    def latest_version(self, ontology_uid):
+        from app.models.data_research import OntologyVersionModel
+
+        model = self.session.query(OntologyVersionModel).filter_by(
+            ontology_uid=str(ontology_uid)
+        ).order_by(OntologyVersionModel.version.desc()).first()
+        return self._version(model) if model is not None else None
+
+    def save_draft(self, ontology_uid, graph, *, expected_revision, actor_uid):
+        from app.models.data_research import OntologyModel, OntologyVersionModel
+
+        model = self.session.get(OntologyModel, str(ontology_uid))
+        if model is None:
+            raise LookupError("ontology was not found")
+        if int(model.draft_revision) != int(expected_revision):
+            raise OntologyConflict("ontology draft revision conflict")
+        latest = self.latest_version(ontology_uid)
+        version = OntologyVersionModel(
+            uid=self.uid_factory(),
+            ontology_uid=str(ontology_uid),
+            version=(latest.version + 1) if latest else 1,
+            parent_version_uid=model.active_version_uid,
+            status="draft",
+            graph_document=GraphDocument.from_dict(graph).to_dict(),
+            content_hash=canonical_graph_hash(graph),
+            created_by=actor_uid,
+        )
+        model.draft_revision = int(model.draft_revision) + 1
+        self.session.add(version)
+        self.session.flush()
+        return self._version(version)
+
+    def mark_published(self, version_uid):
+        from app.models.data_research import OntologyModel, OntologyVersionModel
+
+        model = self.session.get(OntologyVersionModel, str(version_uid))
+        if model is None:
+            raise LookupError("ontology version was not found")
+        ontology = self.session.get(OntologyModel, str(model.ontology_uid))
+        if ontology.active_version_uid:
+            active = self.session.get(OntologyVersionModel, str(ontology.active_version_uid))
+            if active is not None:
+                active.status = "superseded"
+        model.status = "published"
+        ontology.status = "published"
+        ontology.active_version_uid = model.uid
+        self.session.flush()
+        return self._version(model)
+
+    def replace_version(self, version_uid, graph):
+        from app.models.data_research import OntologyVersionModel
+
+        model = self.session.get(OntologyVersionModel, str(version_uid))
+        if model.status == "published":
+            raise OntologyImmutable("published ontology version is immutable")
+        model.graph_document = GraphDocument.from_dict(graph).to_dict()
+        model.content_hash = canonical_graph_hash(graph)
+        self.session.flush()
+        return self._version(model)

app/core/data_research/ontology/validation.py

@@ -0,0 +1,59 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from app.core.data_research.ontology.models import GraphDocument
+
+
+@dataclass(frozen=True)
+class ValidationIssue:
+    code: str
+    message: str
+    path: str | None = None
+
+
+def validate_graph(payload):
+    graph = GraphDocument.from_dict(payload).to_dict()
+    issues = []
+    classes = graph["classes"]
+    class_uids = {str(item.get("uid")) for item in classes}
+    names = set()
+    if not any(item.get("role") == "owner" for item in graph["domain_links"]):
+        issues.append(ValidationIssue("ONTOLOGY_OWNER_REQUIRED", "an owner domain is required", "domain_links"))
+    for index, item in enumerate(classes):
+        name = str(item.get("name") or "").casefold()
+        if name in names:
+            issues.append(ValidationIssue("ONTOLOGY_DUPLICATE_NAME", "class names must be unique", f"classes.{index}.name"))
+        names.add(name)
+        parent = item.get("parent_uid")
+        if parent and str(parent) not in class_uids:
+            issues.append(ValidationIssue("ONTOLOGY_DANGLING_EDGE", "parent class does not exist", f"classes.{index}.parent_uid"))
+    for index, relation in enumerate(graph["relations"]):
+        if str(relation.get("from_class_uid")) not in class_uids or str(relation.get("to_class_uid")) not in class_uids:
+            issues.append(ValidationIssue("ONTOLOGY_DANGLING_EDGE", "relation endpoint does not exist", f"relations.{index}"))
+    parents = {str(item.get("uid")): str(item.get("parent_uid")) for item in classes if item.get("parent_uid")}
+    for uid in parents:
+        seen = set()
+        current = uid
+        while current in parents:
+            if current in seen:
+                issues.append(ValidationIssue("ONTOLOGY_INHERITANCE_CYCLE", "class inheritance contains a cycle", "classes"))
+                parents = {}
+                break
+            seen.add(current)
+            current = parents[current]
+        if not parents:
+            break
+    mapped = {str(item.get("property_uid")) for item in graph["element_mappings"]}
+    for index, prop in enumerate(graph["properties"]):
+        if str(prop.get("class_uid")) not in class_uids:
+            issues.append(ValidationIssue("ONTOLOGY_DANGLING_EDGE", "property owner class does not exist", f"properties.{index}.class_uid"))
+        if str(prop.get("cardinality") or "1") not in {"0..1", "1", "0..*", "1..*"}:
+            issues.append(ValidationIssue("ONTOLOGY_INVALID_CARDINALITY", "invalid property cardinality", f"properties.{index}.cardinality"))
+        if bool(prop.get("required")) and str(prop.get("uid")) not in mapped and not prop.get("data_element_uid"):
+            issues.append(ValidationIssue("ONTOLOGY_REQUIRED_PROPERTY_UNMAPPED", "required property must map to a published data element", f"properties.{index}"))
+    unique = {}
+    for issue in issues:
+        unique.setdefault(issue.code, issue)
+    return list(unique.values())
+

app/core/system/permissions.py

@@ -85,6 +85,12 @@ def permission_for_request(path: str, method: str) -> tuple[str, ...]:
         return (INGESTION_RUN,)
     if path.startswith("/api/development/v1/evidence"):
         return (READ_GOVERNANCE,)
+    if path.startswith("/api/development/v1/ontologies"):
+        if method == "GET":
+            return (READ_GOVERNANCE,)
+        if path.endswith("/publish") or path.endswith("/rollback"):
+            return (ONTOLOGIES_PUBLISH,)
+        return (ONTOLOGIES_EDIT,)
     if path.startswith("/api/system/workbench"):
         return (READ_GOVERNANCE,)
     if (

app/models/__init__.py

@@ -11,6 +11,11 @@ from app.models.data_research import (
     DataElement,
     DataElementVersion,
     CandidateDecisionRecord,
+    OntologyModel,
+    OntologyVersionModel,
+    OntologyDomainLinkModel,
+    OntologyChangeSetModel,
+    OntologyPublishRunModel,
 )
 
 __all__ = [
@@ -26,4 +31,9 @@ __all__ = [
     "DataElement",
     "DataElementVersion",
     "CandidateDecisionRecord",
+    "OntologyModel",
+    "OntologyVersionModel",
+    "OntologyDomainLinkModel",
+    "OntologyChangeSetModel",
+    "OntologyPublishRunModel",
 ]

app/models/data_research.py

@@ -318,3 +318,88 @@ class CandidateDecisionRecord(db.Model):
     actor_uid = db.Column(db.String(100))
     reason = db.Column(db.String(1000))
     created_at = db.Column(db.DateTime, nullable=False, default=now_china_naive)
+
+
+class OntologyModel(db.Model):
+    __tablename__ = "ontologies"
+    __table_args__ = ({"schema": "public"},)
+
+    uid = db.Column(UUID(as_uuid=False), primary_key=True, default=new_governance_uid)
+    code = db.Column(db.String(120), nullable=False, unique=True)
+    name = db.Column(db.String(300), nullable=False)
+    owner_uid = db.Column(db.String(100), nullable=False)
+    status = db.Column(db.String(20), nullable=False, default="draft")
+    draft_revision = db.Column(db.Integer, nullable=False, default=0)
+    active_version_uid = db.Column(UUID(as_uuid=False))
+    created_by = db.Column(db.String(100))
+    created_at = db.Column(db.DateTime, nullable=False, default=now_china_naive)
+    updated_at = db.Column(db.DateTime, nullable=False, default=now_china_naive)
+
+
+class OntologyVersionModel(db.Model):
+    __tablename__ = "ontology_versions"
+    __table_args__ = (
+        db.UniqueConstraint("ontology_uid", "version", name="uq_ontology_version"),
+        {"schema": "public"},
+    )
+
+    uid = db.Column(UUID(as_uuid=False), primary_key=True, default=new_governance_uid)
+    ontology_uid = db.Column(
+        UUID(as_uuid=False), db.ForeignKey("public.ontologies.uid"), nullable=False
+    )
+    version = db.Column(db.Integer, nullable=False)
+    parent_version_uid = db.Column(
+        UUID(as_uuid=False), db.ForeignKey("public.ontology_versions.uid")
+    )
+    status = db.Column(db.String(20), nullable=False, default="draft")
+    graph_document = db.Column(JSONB, nullable=False)
+    content_hash = db.Column(db.String(64), nullable=False)
+    created_by = db.Column(db.String(100))
+    created_at = db.Column(db.DateTime, nullable=False, default=now_china_naive)
+    published_at = db.Column(db.DateTime)
+
+
+class OntologyDomainLinkModel(db.Model):
+    __tablename__ = "ontology_domain_links"
+    __table_args__ = ({"schema": "public"},)
+
+    ontology_uid = db.Column(
+        UUID(as_uuid=False), db.ForeignKey("public.ontologies.uid"), primary_key=True
+    )
+    domain_uid = db.Column(UUID(as_uuid=False), primary_key=True)
+    role = db.Column(db.String(20), nullable=False)
+
+
+class OntologyChangeSetModel(db.Model):
+    __tablename__ = "ontology_change_sets"
+    __table_args__ = ({"schema": "public"},)
+
+    uid = db.Column(UUID(as_uuid=False), primary_key=True, default=new_governance_uid)
+    ontology_uid = db.Column(
+        UUID(as_uuid=False), db.ForeignKey("public.ontologies.uid"), nullable=False
+    )
+    base_version_uid = db.Column(UUID(as_uuid=False))
+    status = db.Column(db.String(20), nullable=False, default="draft")
+    changes = db.Column(JSONB, nullable=False, default=list)
+    decisions = db.Column(JSONB, nullable=False, default=list)
+    created_by = db.Column(db.String(100))
+    created_at = db.Column(db.DateTime, nullable=False, default=now_china_naive)
+
+
+class OntologyPublishRunModel(db.Model):
+    __tablename__ = "ontology_publish_runs"
+    __table_args__ = ({"schema": "public"},)
+
+    uid = db.Column(UUID(as_uuid=False), primary_key=True, default=new_governance_uid)
+    ontology_uid = db.Column(
+        UUID(as_uuid=False), db.ForeignKey("public.ontologies.uid"), nullable=False
+    )
+    version_uid = db.Column(
+        UUID(as_uuid=False), db.ForeignKey("public.ontology_versions.uid"), nullable=False
+    )
+    idempotency_key = db.Column(db.String(128), nullable=False, unique=True)
+    status = db.Column(db.String(20), nullable=False)
+    validation_result = db.Column(JSONB, nullable=False, default=dict)
+    actor_uid = db.Column(db.String(100))
+    created_at = db.Column(db.DateTime, nullable=False, default=now_china_naive)
+    finished_at = db.Column(db.DateTime)

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: 126
+x-route-count: 133
 servers:
   - url: "http://localhost:15500"
     description: "全本地隔离测试后端"
@@ -1884,6 +1884,203 @@ paths:
             application/json:
               schema:
                 $ref: '#/components/schemas/ApiEnvelope'
+  "/api/development/v1/ontologies":
+    get:
+      tags: [data_development]
+      operationId: data_development_list_ontologies_get
+      summary: "list ontologies"
+      x-source: "app/api/data_development/routes.py"
+      responses:
+        "200":
+          description: "请求已由当前实现处理"
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ApiEnvelope'
+        default:
+          description: "错误响应"
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ApiEnvelope'
+    post:
+      tags: [data_development]
+      operationId: data_development_create_ontology_post
+      summary: "create ontology"
+      x-source: "app/api/data_development/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/development/v1/ontologies/{ontology_uid}/diff":
+    get:
+      tags: [data_development]
+      operationId: data_development_diff_ontology_get
+      summary: "diff ontology"
+      x-source: "app/api/data_development/routes.py"
+      parameters:
+        - name: ontology_uid
+          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/development/v1/ontologies/{ontology_uid}/graph":
+    patch:
+      tags: [data_development]
+      operationId: data_development_save_ontology_graph_patch
+      summary: "save ontology graph"
+      x-source: "app/api/data_development/routes.py"
+      parameters:
+        - name: ontology_uid
+          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/development/v1/ontologies/{ontology_uid}/publish":
+    post:
+      tags: [data_development]
+      operationId: data_development_publish_ontology_post
+      summary: "publish ontology"
+      x-source: "app/api/data_development/routes.py"
+      parameters:
+        - name: ontology_uid
+          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/development/v1/ontologies/{ontology_uid}/rollback":
+    post:
+      tags: [data_development]
+      operationId: data_development_rollback_ontology_post
+      summary: "rollback ontology"
+      x-source: "app/api/data_development/routes.py"
+      parameters:
+        - name: ontology_uid
+          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/development/v1/ontologies/{ontology_uid}/validate":
+    post:
+      tags: [data_development]
+      operationId: data_development_validate_ontology_post
+      summary: "validate ontology"
+      x-source: "app/api/data_development/routes.py"
+      parameters:
+        - name: ontology_uid
+          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/development/v1/sources/files":
     post:
       tags: [data_development]

docs/superpowers/plans/2026-07-22-data-research-v60-v65-delivery-plan.md

@@ -230,10 +230,10 @@
 - Produces: `Ontology`, immutable `OntologyVersion`, `OntologyChangeSet`, and `OntologyPublishRun`.
 - Produces graph document fields `classes`, `properties`, `relations`, `constraints`, `domain_links`, and `element_mappings`.
 
-- [ ] Write schema and repository tests for unique ontology codes, immutable published versions, parent version, content hash, many-domain role links, and optimistic draft revisions.
-- [ ] Run focused tests and verify missing schema/repository failures.
-- [ ] Implement additive migration, dataclasses, canonical graph hashing, and repository operations.
-- [ ] Rerun focused and migration tests.
+- [x] Write schema and repository tests for unique ontology codes, immutable published versions, parent version, content hash, many-domain role links, and optimistic draft revisions.
+- [x] Run focused tests and verify missing schema/repository failures.
+- [x] Implement additive migration, dataclasses, canonical graph hashing, and repository operations.
+- [x] Rerun focused and migration tests.
 
 ### Task 11: Implement ontology validation, publication, projection, and rollback
 
@@ -248,10 +248,10 @@
 - Produces validation codes for duplicate names, dangling edges, inheritance cycles, invalid cardinality, ownerless ontology, and unmapped required property.
 - Produces events `ontology.version_published` and `ontology.rollback_version_created`.
 
-- [ ] Write tests for every validation code, publish idempotency, outbox atomicity, active-version projection, retryable Neo4j failure, version diff, and rollback-as-new-version behavior.
-- [ ] Run focused tests and verify missing service failures.
-- [ ] Implement pure validation, publish transaction, UID-based Neo4j projection, diff, and rollback version creation.
-- [ ] Rerun focused tests plus cross-store/outbox tests.
+- [x] Write tests for every validation code, publish idempotency, outbox atomicity, active-version projection, retryable Neo4j failure, version diff, and rollback-as-new-version behavior.
+- [x] Run focused tests and verify missing service failures.
+- [x] Implement pure validation, publish transaction, UID-based Neo4j projection, diff, and rollback version creation.
+- [x] Rerun focused tests plus cross-store/outbox tests.
 
 ### Task 12: Add ontology APIs and Vue 2 ontology center
 
@@ -268,15 +268,15 @@
 - Produces ontology CRUD, graph patch, validate, publish, diff, and rollback endpoints under `/api/development/v1`.
 - Produces permission-gated ontology list and workbench routes.
 
-- [ ] Write API tests for editor drafts, publisher release, multi-domain roles, ETag conflicts, validation errors, and viewer read-only behavior; write source contract tests for routes and secret-free rendering.
-- [ ] Run focused tests and verify missing endpoints/pages.
-- [ ] Implement API handlers and Vue pages with graph JSON editing/preview, validation panel, version diff, and publish confirmation.
-- [ ] Run focused tests and frontend production build.
+- [x] Write API tests for editor drafts, publisher release, multi-domain roles, ETag conflicts, validation errors, and viewer read-only behavior; write source contract tests for routes and secret-free rendering.
+- [x] Run focused tests and verify missing endpoints/pages.
+- [x] Implement API handlers and Vue pages with graph JSON editing/preview, validation panel, version diff, and publish confirmation.
+- [x] Run focused tests and frontend production build.
 
 ### V63 gate
 
-- [ ] Run all ontology tests, complete Python suite, frontend build, and `git diff --check`.
-- [ ] Record evidence in `docs/validation/data-research-v63.md` and commit V63.
+- [x] Run all ontology tests, complete Python suite, frontend build, and `git diff --check`.
+- [x] Record evidence in `docs/validation/data-research-v63.md` and commit V63.
 
 ## V64 — Dynamic ontology and knowledge services
 

docs/validation/data-research-v63.md

@@ -0,0 +1,22 @@
+# V63 本体 MVP 验证
+
+日期：2026-07-22
+
+## 交付范围
+
+- PostgreSQL 本体、不可变版本、业务域角色、变更集与发布运行控制面。
+- 规范化图文档哈希、乐观草稿修订及父版本关系。
+- 重名、悬空边、继承环、基数、责任域、必填属性映射六类稳定校验码。
+- 幂等发布、事务 Outbox、Neo4j 有效版本投影、版本差异与回滚新版本。
+- 本体 CRUD/图补丁/校验/发布/差异/回滚 API。
+- Vue 2 本体中心与工作台，包含校验面板、版本差异和发布确认。
+
+## 测试证据
+
+- 本体模型、仓储、校验、发布、API 与前端契约：`19 passed, 1 skipped`。
+- V63 专项及 OpenAPI：`26 passed, 1 skipped`，OpenAPI 共 133 个操作。
+- Vue 2 生产构建：成功；仅保留既有浏览器数据、CSS 顺序、包体积和 21 条 `no-console` 警告，无错误。
+- 全量后端回归：`349 passed, 23 skipped, 59 subtests passed`。
+- `git diff --check`：通过。
+
+Neo4j 不可用场景验证为可重试失败，不会回滚或篡改 PostgreSQL 已发布历史；实际投影由 Outbox 消费者异步执行。

frontend/src/api/dataDevelopment.js

@@ -1,17 +1,46 @@
 import http from '@/utils/request'
 
 const BASE = '/development/v1/ingestion-jobs'
+const ONTOLOGY_BASE = '/development/v1/ontologies'
 
 export const createIngestionJob = params => http.post(BASE, params)
 export const getIngestionJobs = params => http.get(BASE, params)
 export const getIngestionJob = uid => http.get(`${BASE}/${uid}`)
 export const retryIngestionJob = uid => http.post(`${BASE}/${uid}/retry`)
 export const cancelIngestionJob = uid => http.post(`${BASE}/${uid}/cancel`)
+export const getOntologies = () => http.get(ONTOLOGY_BASE)
+export const createOntology = params => http.post(ONTOLOGY_BASE, params)
+export const saveOntologyGraph = (uid, graph, revision) => http.patch(
+  `${ONTOLOGY_BASE}/${uid}/graph`,
+  graph,
+  { headers: { 'If-Match': `"${revision}"` } }
+)
+export const validateOntology = uid => http.post(`${ONTOLOGY_BASE}/${uid}/validate`)
+export const publishOntology = (uid, idempotencyKey) => http.post(
+  `${ONTOLOGY_BASE}/${uid}/publish`,
+  {},
+  { headers: { 'Idempotency-Key': idempotencyKey } }
+)
+export const diffOntology = (uid, left, right) => http.get(
+  `${ONTOLOGY_BASE}/${uid}/diff`,
+  { left, right }
+)
+export const rollbackOntology = (uid, params) => http.post(
+  `${ONTOLOGY_BASE}/${uid}/rollback`,
+  params
+)
 
 export default {
   createIngestionJob,
   getIngestionJobs,
   getIngestionJob,
   retryIngestionJob,
-  cancelIngestionJob
+  cancelIngestionJob,
+  getOntologies,
+  createOntology,
+  saveOntologyGraph,
+  validateOntology,
+  publishOntology,
+  diffOntology,
+  rollbackOntology
 }

frontend/src/router/routes.js

@@ -179,6 +179,58 @@ export default {
       path: '/data-governance',
       urls: '',
       children: [
+        {
+          hidden: 0,
+          icon: 'mdi-graph-outline',
+          type: 1,
+          title: '本体中心',
+          path: '/data-governance/ontology',
+          children: [],
+          enName: 'Ontology Center',
+          label: '本体中心',
+          sort: 0,
+          component: 'dataGovernance/ontology/index',
+          meta: {
+            keepAlive: false,
+            allowClick: true,
+            roles: ['viewer', 'editor', 'admin'],
+            enName: 'Ontology Center',
+            icon: 'mdi-graph-outline',
+            editModules: false,
+            title: '本体中心',
+            fullScreen: false,
+            target: false,
+            effectiveStatus: true
+          },
+          name: 'ontologyCenter',
+          alwaysShow: 0
+        },
+        {
+          hidden: 1,
+          icon: 'mdi-vector-polyline-edit',
+          type: 1,
+          title: '本体工作台',
+          path: '/data-governance/ontology/:uid',
+          children: [],
+          enName: 'Ontology Workbench',
+          label: '本体工作台',
+          sort: 0,
+          component: 'dataGovernance/ontology/workbench',
+          meta: {
+            keepAlive: false,
+            allowClick: false,
+            roles: ['editor', 'admin'],
+            enName: 'Ontology Workbench',
+            icon: 'mdi-vector-polyline-edit',
+            editModules: false,
+            title: '本体工作台',
+            fullScreen: true,
+            target: false,
+            effectiveStatus: true
+          },
+          name: 'ontologyWorkbench',
+          alwaysShow: 0
+        },
         {
           meun: '',
           code: '',

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

@@ -0,0 +1,68 @@
+<template>
+  <div class="ontology-center pa-6">
+    <div class="d-flex align-center mb-6">
+      <div>
+        <h1 class="text-h4 mb-1">本体中心</h1>
+        <p class="text--secondary mb-0">跨业务域管理本体定义、版本与发布状态</p>
+      </div>
+      <v-spacer />
+      <v-btn color="primary" @click="dialog = true">新建本体</v-btn>
+    </div>
+    <v-data-table :headers="headers" :items="items" :loading="loading">
+      <template v-slot:[`item.domain_links`]="{ item }">
+        <v-chip v-for="link in item.domain_links" :key="link.domain_uid" small class="mr-1">
+          {{ link.role }} · {{ link.domain_uid }}
+        </v-chip>
+      </template>
+      <template v-slot:[`item.actions`]="{ item }">
+        <v-btn text color="primary" @click="$router.push(`/data-governance/ontology/${item.uid}`)">进入工作台</v-btn>
+      </template>
+    </v-data-table>
+    <v-dialog v-model="dialog" max-width="560">
+      <v-card>
+        <v-card-title>新建本体</v-card-title>
+        <v-card-text>
+          <v-text-field v-model="form.code" label="本体编码" />
+          <v-text-field v-model="form.name" label="本体名称" />
+          <v-text-field v-model="form.owner_uid" label="责任人 UID" />
+        </v-card-text>
+        <v-card-actions>
+          <v-spacer /><v-btn text @click="dialog = false">取消</v-btn><v-btn color="primary" @click="create">创建</v-btn>
+        </v-card-actions>
+      </v-card>
+    </v-dialog>
+  </div>
+</template>
+
+<script>
+import { createOntology, getOntologies } from '@/api/dataDevelopment'
+
+export default {
+  name: 'OntologyCenter',
+  data: () => ({
+    loading: false,
+    dialog: false,
+    items: [],
+    form: { code: '', name: '', owner_uid: '', domain_links: [] },
+    headers: [
+      { text: '编码', value: 'code' },
+      { text: '名称', value: 'name' },
+      { text: '业务域角色', value: 'domain_links' },
+      { text: '状态', value: 'status' },
+      { text: '操作', value: 'actions', sortable: false }
+    ]
+  }),
+  created () { this.load() },
+  methods: {
+    async load () {
+      this.loading = true
+      try { this.items = (await getOntologies()).data || [] } finally { this.loading = false }
+    },
+    async create () {
+      await createOntology(this.form)
+      this.dialog = false
+      await this.load()
+    }
+  }
+}
+</script>

frontend/src/views/dataGovernance/ontology/workbench.vue

@@ -0,0 +1,36 @@
+<template>
+  <div class="ontology-workbench pa-6">
+    <div class="d-flex align-center mb-4">
+      <div><h1 class="text-h4">本体工作台</h1><p class="text--secondary">图定义、校验、版本差异与发布确认</p></div>
+      <v-spacer />
+      <v-btn outlined class="mr-2" @click="validate">校验</v-btn>
+      <v-btn color="primary" @click="publishDialog = true">发布确认</v-btn>
+    </div>
+    <v-row>
+      <v-col cols="8"><v-textarea v-model="graphJson" outlined auto-grow label="本体图 JSON" /></v-col>
+      <v-col cols="4">
+        <v-card outlined><v-card-title>校验结果</v-card-title><v-card-text>
+          <v-alert v-for="issue in issues" :key="issue.code" type="warning" dense>{{ issue.code }} · {{ issue.message }}</v-alert>
+          <span v-if="!issues.length">尚无校验问题</span>
+        </v-card-text></v-card>
+        <v-card outlined class="mt-4"><v-card-title>版本差异</v-card-title><v-card-text><pre>{{ diff }}</pre></v-card-text></v-card>
+      </v-col>
+    </v-row>
+    <v-btn color="primary" @click="save">保存草稿</v-btn>
+    <v-dialog v-model="publishDialog" max-width="480"><v-card><v-card-title>发布确认</v-card-title><v-card-text>发布后该版本不可修改，并将异步投影至语义图。</v-card-text><v-card-actions><v-spacer /><v-btn text @click="publishDialog = false">取消</v-btn><v-btn color="primary" @click="publish">确认发布</v-btn></v-card-actions></v-card></v-dialog>
+  </div>
+</template>
+
+<script>
+import { publishOntology, saveOntologyGraph, validateOntology } from '@/api/dataDevelopment'
+
+export default {
+  name: 'OntologyWorkbench',
+  data: () => ({ graphJson: '{\n  "classes": [],\n  "properties": [],\n  "relations": [],\n  "constraints": [],\n  "domain_links": [],\n  "element_mappings": []\n}', revision: 0, issues: [], diff: {}, publishDialog: false }),
+  methods: {
+    async save () { await saveOntologyGraph(this.$route.params.uid, JSON.parse(this.graphJson), this.revision); this.revision += 1 },
+    async validate () { this.issues = (await validateOntology(this.$route.params.uid)).data || [] },
+    async publish () { await publishOntology(this.$route.params.uid, `${this.$route.params.uid}:${this.revision}`); this.publishDialog = false }
+  }
+}
+</script>

migrations/versions/20260722_110_data_research_ontology.py

@@ -0,0 +1,79 @@
+"""Add ontology control-plane state and immutable versions."""
+
+from alembic import op
+
+
+revision = "20260722_110"
+down_revision = "20260722_105"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.execute(
+        """
+        CREATE TABLE IF NOT EXISTS public.ontologies (
+            uid UUID PRIMARY KEY,
+            code VARCHAR(120) NOT NULL UNIQUE,
+            name VARCHAR(300) NOT NULL,
+            owner_uid VARCHAR(100) NOT NULL,
+            status VARCHAR(20) NOT NULL DEFAULT 'draft',
+            draft_revision INTEGER NOT NULL DEFAULT 0,
+            active_version_uid UUID,
+            created_by VARCHAR(100),
+            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+            updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
+        );
+        CREATE TABLE IF NOT EXISTS public.ontology_versions (
+            uid UUID PRIMARY KEY,
+            ontology_uid UUID NOT NULL REFERENCES public.ontologies(uid),
+            version INTEGER NOT NULL,
+            parent_version_uid UUID REFERENCES public.ontology_versions(uid),
+            status VARCHAR(20) NOT NULL CHECK (status IN ('draft','published','superseded')),
+            graph_document JSONB NOT NULL,
+            content_hash VARCHAR(64) NOT NULL,
+            created_by VARCHAR(100),
+            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+            published_at TIMESTAMPTZ,
+            UNIQUE (ontology_uid, version)
+        );
+        ALTER TABLE public.ontologies
+            ADD CONSTRAINT fk_ontologies_active_version
+            FOREIGN KEY (active_version_uid) REFERENCES public.ontology_versions(uid);
+        CREATE TABLE IF NOT EXISTS public.ontology_domain_links (
+            ontology_uid UUID NOT NULL REFERENCES public.ontologies(uid),
+            domain_uid UUID NOT NULL,
+            role VARCHAR(20) NOT NULL CHECK (role IN ('owner','contributor','consumer')),
+            PRIMARY KEY (ontology_uid, domain_uid)
+        );
+        CREATE TABLE IF NOT EXISTS public.ontology_change_sets (
+            uid UUID PRIMARY KEY,
+            ontology_uid UUID NOT NULL REFERENCES public.ontologies(uid),
+            base_version_uid UUID REFERENCES public.ontology_versions(uid),
+            status VARCHAR(20) NOT NULL DEFAULT 'draft',
+            changes JSONB NOT NULL DEFAULT '[]'::jsonb,
+            decisions JSONB NOT NULL DEFAULT '[]'::jsonb,
+            created_by VARCHAR(100),
+            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
+        );
+        CREATE TABLE IF NOT EXISTS public.ontology_publish_runs (
+            uid UUID PRIMARY KEY,
+            ontology_uid UUID NOT NULL REFERENCES public.ontologies(uid),
+            version_uid UUID NOT NULL REFERENCES public.ontology_versions(uid),
+            idempotency_key VARCHAR(128) NOT NULL UNIQUE,
+            status VARCHAR(20) NOT NULL,
+            validation_result JSONB NOT NULL DEFAULT '{}'::jsonb,
+            actor_uid VARCHAR(100),
+            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+            finished_at TIMESTAMPTZ
+        );
+        CREATE INDEX IF NOT EXISTS idx_ontology_versions_ontology
+            ON public.ontology_versions(ontology_uid, version DESC);
+        """
+    )
+
+
+def downgrade() -> None:
+    # Published semantic history is preserved during application rollback.
+    pass
+

tests/data_research/test_ontology_api.py

@@ -0,0 +1,131 @@
+from __future__ import annotations
+
+from dataclasses import replace
+
+import pytest
+
+from app.core.data_research.ontology.models import DomainLink, GraphDocument, Ontology, OntologyVersion
+from app.core.data_research.ontology.validation import ValidationIssue
+
+
+class FakeOntologyService:
+    def __init__(self):
+        self.ontology = Ontology(
+            uid="ontology-1",
+            code="CUSTOMER",
+            name="客户本体",
+            owner_uid="owner-1",
+            domain_links=(DomainLink("domain-1", "owner"), DomainLink("domain-2", "contributor")),
+        )
+        self.version = OntologyVersion(
+            uid="version-1",
+            ontology_uid="ontology-1",
+            version=1,
+            graph_document=GraphDocument.from_dict({"domain_links": [{"domain_uid": "domain-1", "role": "owner"}]}),
+            content_hash="a" * 64,
+        )
+        self.calls = []
+
+    def list(self):
+        return [self.ontology]
+
+    def create(self, payload, actor_uid):
+        self.calls.append(("create", payload, actor_uid))
+        return self.ontology
+
+    def save_draft(self, uid, graph, expected_revision, actor_uid):
+        self.calls.append(("draft", uid, expected_revision, actor_uid))
+        return self.version
+
+    def validate(self, uid):
+        self.calls.append(("validate", uid))
+        return [ValidationIssue("ONTOLOGY_OWNER_REQUIRED", "owner required", "domain_links")]
+
+    def publish(self, uid, idempotency_key, actor_uid):
+        self.calls.append(("publish", uid, idempotency_key, actor_uid))
+        return replace(self.version, status="published")
+
+    def diff(self, uid, left, right):
+        return {"classes": {"added": [], "removed": [], "changed": []}}
+
+    def rollback(self, uid, target_version_uid, expected_revision, actor_uid):
+        self.calls.append(("rollback", uid, target_version_uid, expected_revision, actor_uid))
+        return replace(self.version, uid="version-2", version=2, parent_version_uid="version-1")
+
+
+@pytest.fixture()
+def client(monkeypatch):
+    from flask import request
+
+    from app import create_app
+    from app.api.data_development import routes
+    from app.core.system import permissions
+
+    service = FakeOntologyService()
+
+    def identity():
+        token = request.headers.get("Authorization", "")
+        role = token.removeprefix("Bearer ")
+        return {"id": f"{role}-1", "roles": [role]} if role in {"viewer", "editor", "admin"} else None
+
+    monkeypatch.setattr(permissions, "authenticate_request", identity)
+    monkeypatch.setattr(routes, "get_ontology_service", lambda: service)
+    app = create_app()
+    app.config.update(TESTING=True)
+    return app.test_client(), service
+
+
+def test_viewer_is_read_only_and_multi_domain_roles_are_returned(client):
+    http, _service = client
+    headers = {"Authorization": "Bearer viewer"}
+    listed = http.get("/api/development/v1/ontologies", headers=headers)
+    denied = http.post(
+        "/api/development/v1/ontologies",
+        headers=headers,
+        json={"code": "CUSTOMER", "name": "客户本体"},
+    )
+    assert listed.status_code == 200
+    assert listed.get_json()["data"][0]["domain_links"][1]["role"] == "contributor"
+    assert denied.status_code == 403
+
+
+def test_editor_saves_etag_draft_but_cannot_publish(client):
+    http, service = client
+    headers = {"Authorization": "Bearer editor", "If-Match": '"0"'}
+    saved = http.patch(
+        "/api/development/v1/ontologies/ontology-1/graph",
+        headers=headers,
+        json={"classes": [], "domain_links": []},
+    )
+    denied = http.post(
+        "/api/development/v1/ontologies/ontology-1/publish",
+        headers={"Authorization": "Bearer editor", "Idempotency-Key": "publish-1"},
+    )
+    assert saved.status_code == 200
+    assert service.calls[-1][2] == 0
+    assert denied.status_code == 403
+
+
+def test_admin_validates_publishes_diffs_and_rolls_back(client):
+    http, service = client
+    headers = {"Authorization": "Bearer admin"}
+    validation = http.post("/api/development/v1/ontologies/ontology-1/validate", headers=headers)
+    published = http.post(
+        "/api/development/v1/ontologies/ontology-1/publish",
+        headers={**headers, "Idempotency-Key": "publish-1"},
+    )
+    diffed = http.get(
+        "/api/development/v1/ontologies/ontology-1/diff?left=version-0&right=version-1",
+        headers=headers,
+    )
+    rolled_back = http.post(
+        "/api/development/v1/ontologies/ontology-1/rollback",
+        headers=headers,
+        json={"target_version_uid": "version-1", "expected_revision": 1},
+    )
+    assert validation.status_code == 422
+    assert validation.get_json()["data"][0]["code"] == "ONTOLOGY_OWNER_REQUIRED"
+    assert published.get_json()["data"]["status"] == "published"
+    assert diffed.status_code == 200
+    assert rolled_back.get_json()["data"]["parent_version_uid"] == "version-1"
+

tests/data_research/test_ontology_frontend_contract.py

@@ -0,0 +1,24 @@
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[2]
+
+
+def test_vue2_ontology_center_routes_and_api_contract_exist():
+    routes = (ROOT / "frontend/src/router/routes.js").read_text(encoding="utf-8")
+    api = (ROOT / "frontend/src/api/dataDevelopment.js").read_text(encoding="utf-8")
+    index = (ROOT / "frontend/src/views/dataGovernance/ontology/index.vue").read_text(encoding="utf-8")
+    workbench = (ROOT / "frontend/src/views/dataGovernance/ontology/workbench.vue").read_text(encoding="utf-8")
+
+    assert "ontologyCenter" in routes
+    assert "dataGovernance/ontology/index" in routes
+    assert "dataGovernance/ontology/workbench" in routes
+    for name in ("getOntologies", "createOntology", "validateOntology", "publishOntology", "rollbackOntology"):
+        assert name in api
+    assert "本体中心" in index
+    assert "版本差异" in workbench
+    assert "发布确认" in workbench
+    combined = api + index + workbench
+    assert "MINIO_PASSWORD" not in combined
+    assert "NEO4J_PASSWORD" not in combined
+    assert "DEEPSEEK_API_KEY" not in combined

tests/data_research/test_ontology_publication.py

@@ -0,0 +1,106 @@
+from __future__ import annotations
+
+import pytest
+
+from tests.data_research.test_ontology_validation import base_graph
+
+
+def repository():
+    from app.core.data_research.ontology.repository import MemoryOntologyRepository
+
+    ids = iter(("ontology-1", "version-1", "run-1", "version-2"))
+    repo = MemoryOntologyRepository(uid_factory=ids.__next__)
+    ontology = repo.create(
+        code="CUSTOMER",
+        name="客户本体",
+        owner_uid="owner-1",
+        domain_links=base_graph()["domain_links"],
+    )
+    version = repo.save_draft(ontology.uid, base_graph(), expected_revision=0, actor_uid="editor-1")
+    return repo, ontology, version
+
+
+def test_publish_is_idempotent_atomic_and_emits_outbox_event():
+    from app.core.data_research.ontology.publication import OntologyPublicationService
+
+    repo, ontology, version = repository()
+    events = []
+    service = OntologyPublicationService(
+        repo,
+        outbox_enqueue=lambda **event: events.append(event),
+        commit=lambda: events.append({"committed": True}),
+    )
+    first = service.publish(version.uid, idempotency_key="publish-1", actor_uid="admin-1")
+    second = service.publish(version.uid, idempotency_key="publish-1", actor_uid="admin-1")
+
+    assert first == second
+    assert repo.get(ontology.uid).active_version_uid == version.uid
+    assert [event.get("event_type") for event in events].count("ontology.version_published") == 1
+    assert first.status == "published"
+
+
+def test_invalid_graph_rolls_back_without_activation():
+    from app.core.data_research.ontology.publication import (
+        OntologyPublicationService,
+        OntologyValidationFailed,
+    )
+
+    repo, ontology, version = repository()
+    repo.replace_version(version.uid, {**base_graph(), "domain_links": []})
+    rollbacks = []
+    with pytest.raises(OntologyValidationFailed):
+        OntologyPublicationService(repo, rollback=lambda: rollbacks.append(True)).publish(
+            version.uid, idempotency_key="invalid", actor_uid="admin-1"
+        )
+    assert repo.get(ontology.uid).active_version_uid is None
+    assert rollbacks == [True]
+
+
+class GraphSession:
+    def __init__(self, fail=False):
+        self.fail = fail
+        self.calls = []
+
+    def run(self, statement, **params):
+        if self.fail:
+            raise ConnectionError("neo4j unavailable")
+        self.calls.append((statement, params))
+
+
+def test_projection_uses_active_version_uid_and_failure_is_retryable():
+    from app.core.data_research.ontology.projection import OntologyGraphProjector
+
+    graph = GraphSession()
+    payload = {
+        "ontology_uid": "ontology-1",
+        "version_uid": "version-1",
+        "version": 1,
+        "graph_document": base_graph(),
+    }
+    OntologyGraphProjector(graph).project(payload)
+    assert "active_version_uid" in graph.calls[0][0]
+    assert graph.calls[0][1]["version_uid"] == "version-1"
+
+    with pytest.raises(ConnectionError, match="neo4j unavailable"):
+        OntologyGraphProjector(GraphSession(fail=True)).project(payload)
+
+
+def test_version_diff_and_rollback_create_new_draft():
+    from app.core.data_research.ontology.publication import OntologyPublicationService, diff_graphs
+
+    repo, ontology, version = repository()
+    service = OntologyPublicationService(repo)
+    service.publish(version.uid, idempotency_key="publish-1", actor_uid="admin-1")
+    changed = base_graph()
+    changed["classes"] = changed["classes"] + [{"uid": "account", "name": "Account"}]
+    assert diff_graphs(base_graph(), changed)["classes"]["added"] == ["account"]
+
+    rollback = service.create_rollback(
+        ontology.uid,
+        target_version_uid=version.uid,
+        expected_revision=1,
+        actor_uid="admin-1",
+    )
+    assert rollback.version == 2
+    assert rollback.parent_version_uid == version.uid
+    assert rollback.status == "draft"

tests/data_research/test_ontology_repository.py

@@ -0,0 +1,71 @@
+from __future__ import annotations
+
+import pytest
+
+
+def graph(name="Customer"):
+    return {
+        "classes": [{"uid": "class-1", "name": name}],
+        "properties": [],
+        "relations": [],
+        "constraints": [],
+        "domain_links": [
+            {"domain_uid": "domain-a", "role": "owner"},
+            {"domain_uid": "domain-b", "role": "contributor"},
+        ],
+        "element_mappings": [],
+    }
+
+
+def test_unique_code_domain_roles_and_optimistic_draft_revision():
+    from app.core.data_research.ontology.repository import MemoryOntologyRepository, OntologyConflict
+
+    repository = MemoryOntologyRepository(uid_factory=iter(("ontology-1", "version-1")).__next__)
+    ontology = repository.create(
+        code="CUSTOMER_360",
+        name="客户360本体",
+        owner_uid="owner-1",
+        domain_links=graph()["domain_links"],
+    )
+    assert ontology.uid == "ontology-1"
+    assert len(ontology.domain_links) == 2
+
+    with pytest.raises(OntologyConflict, match="code"):
+        repository.create(code="CUSTOMER_360", name="重复", owner_uid="owner-2", domain_links=[])
+
+    draft = repository.save_draft(
+        ontology.uid, graph(), expected_revision=0, actor_uid="editor-1"
+    )
+    assert draft.parent_version_uid is None
+    assert repository.get(ontology.uid).draft_revision == 1
+    with pytest.raises(OntologyConflict, match="revision"):
+        repository.save_draft(
+            ontology.uid, graph("Changed"), expected_revision=0, actor_uid="editor-1"
+        )
+
+
+def test_published_version_is_immutable_and_hash_is_canonical():
+    from app.core.data_research.ontology.repository import (
+        MemoryOntologyRepository,
+        OntologyImmutable,
+        canonical_graph_hash,
+    )
+
+    ids = iter(("ontology-1", "version-1", "version-2"))
+    repository = MemoryOntologyRepository(uid_factory=ids.__next__)
+    ontology = repository.create(
+        code="CUSTOMER", name="客户", owner_uid="owner-1", domain_links=graph()["domain_links"]
+    )
+    version = repository.save_draft(ontology.uid, graph(), expected_revision=0, actor_uid="editor-1")
+    published = repository.mark_published(version.uid)
+    assert published.content_hash == canonical_graph_hash(graph())
+    assert published.parent_version_uid is None
+
+    with pytest.raises(OntologyImmutable, match="published"):
+        repository.replace_version(version.uid, graph("Illegal"))
+
+    next_version = repository.save_draft(
+        ontology.uid, graph("CustomerAccount"), expected_revision=1, actor_uid="editor-1"
+    )
+    assert next_version.parent_version_uid == published.uid
+    assert next_version.version == 2

tests/data_research/test_ontology_schema.py

@@ -0,0 +1,34 @@
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[2]
+
+
+def test_ontology_migration_is_additive_and_versioned():
+    source = (ROOT / "migrations/versions/20260722_110_data_research_ontology.py").read_text(encoding="utf-8")
+    for table in (
+        "ontologies",
+        "ontology_versions",
+        "ontology_domain_links",
+        "ontology_change_sets",
+        "ontology_publish_runs",
+    ):
+        assert f"public.{table}" in source
+    assert 'down_revision = "20260722_105"' in source
+    assert "active_version_uid" in source
+    assert "content_hash" in source
+    assert "UNIQUE (ontology_uid, version)" in source
+
+
+def test_graph_document_has_governance_sections():
+    from app.core.data_research.ontology.models import GraphDocument
+
+    graph = GraphDocument.from_dict({})
+    assert graph.to_dict() == {
+        "classes": [],
+        "properties": [],
+        "relations": [],
+        "constraints": [],
+        "domain_links": [],
+        "element_mappings": [],
+    }

tests/data_research/test_ontology_validation.py

@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+
+def base_graph():
+    return {
+        "classes": [{"uid": "customer", "name": "Customer"}],
+        "properties": [
+            {
+                "uid": "customer-id",
+                "name": "customerId",
+                "class_uid": "customer",
+                "required": True,
+                "cardinality": "1",
+                "data_element_uid": "element-1",
+            }
+        ],
+        "relations": [],
+        "constraints": [],
+        "domain_links": [{"domain_uid": "domain-1", "role": "owner"}],
+        "element_mappings": [{"property_uid": "customer-id", "data_element_uid": "element-1"}],
+    }
+
+
+def test_valid_graph_has_no_errors():
+    from app.core.data_research.ontology.validation import validate_graph
+
+    assert validate_graph(base_graph()) == []
+
+
+def test_validation_reports_stable_codes_for_governance_failures():
+    from app.core.data_research.ontology.validation import validate_graph
+
+    graph = base_graph()
+    graph["classes"] += [
+        {"uid": "duplicate", "name": "Customer", "parent_uid": "missing"},
+        {"uid": "cycle-a", "name": "CycleA", "parent_uid": "cycle-b"},
+        {"uid": "cycle-b", "name": "CycleB", "parent_uid": "cycle-a"},
+    ]
+    graph["properties"][0]["cardinality"] = "many-to-sometimes"
+    graph["properties"][0].pop("data_element_uid")
+    graph["element_mappings"] = []
+    graph["relations"] = [{"uid": "bad-edge", "from_class_uid": "customer", "to_class_uid": "gone"}]
+    graph["domain_links"] = [{"domain_uid": "domain-1", "role": "consumer"}]
+
+    codes = {issue.code for issue in validate_graph(graph)}
+    assert codes == {
+        "ONTOLOGY_DUPLICATE_NAME",
+        "ONTOLOGY_DANGLING_EDGE",
+        "ONTOLOGY_INHERITANCE_CYCLE",
+        "ONTOLOGY_INVALID_CARDINALITY",
+        "ONTOLOGY_OWNER_REQUIRED",
+        "ONTOLOGY_REQUIRED_PROPERTY_UNMAPPED",
+    }
+

tests/test_database_migrations.py

@@ -86,6 +86,25 @@ def test_data_element_migration_adds_versioned_governance_tables():
     assert "DROP TABLE" not in migration.upper()
 
 
+def test_ontology_migration_adds_versioned_control_plane_tables():
+    migration = (
+        ROOT / "migrations" / "versions" / "20260722_110_data_research_ontology.py"
+    ).read_text(encoding="utf-8")
+
+    assert 'revision = "20260722_110"' in migration
+    assert 'down_revision = "20260722_105"' in migration
+    for table in (
+        "ontologies",
+        "ontology_versions",
+        "ontology_domain_links",
+        "ontology_change_sets",
+        "ontology_publish_runs",
+    ):
+        assert f"CREATE TABLE IF NOT EXISTS public.{table}" in migration
+    assert "UNIQUE (ontology_uid, version)" in migration
+    assert "DROP TABLE" not in migration.upper()
+
+
 def test_alembic_configuration_is_environment_only():
     ini = (ROOT / "alembic.ini").read_text(encoding="utf-8")
     env = (ROOT / "migrations" / "env.py").read_text(encoding="utf-8")