feat: deliver V64 dynamic ontology services

app/api/data_development/routes.py

@@ -114,6 +114,35 @@ def get_ontology_service():
     )
 
 
+def get_ontology_dynamic_service():
+    from app.core.data_research.ontology.change_sets import (
+        SqlAlchemyDynamicOntologyService,
+    )
+
+    return SqlAlchemyDynamicOntologyService(db.session)
+
+
+def get_ontology_exchange_service():
+    from app.core.data_research.ontology.exchange import OntologyExchangeService
+    from app.core.data_research.ontology.repository import SqlAlchemyOntologyRepository
+
+    return OntologyExchangeService(
+        SqlAlchemyOntologyRepository(db.session),
+        commit=db.session.commit,
+        rollback=db.session.rollback,
+    )
+
+
+def get_semantic_query_service():
+    from app.core.data_research.ontology.query import (
+        Neo4jSemanticRepository,
+        SemanticQueryService,
+    )
+    from app.services.neo4j_driver import neo4j_driver
+
+    return SemanticQueryService(Neo4jSemanticRepository(neo4j_driver))
+
+
 def _identity():
     return getattr(g, "current_user", {}) or {}
 
@@ -472,3 +501,97 @@ def rollback_ontology(ontology_uid):
         return jsonify(success(_ontology_version(version))), 201
     except Exception as error:
         return _error(error)
+
+
+@bp.route("/ontologies/<ontology_uid>/suggestions", methods=["POST"])
+def generate_ontology_suggestions(ontology_uid):
+    try:
+        records = get_ontology_dynamic_service().generate(
+            ontology_uid,
+            request.get_json(silent=True) or {},
+            _identity().get("id") or _identity().get("sub"),
+        )
+        data = [
+            {
+                "uid": item.uid,
+                "kind": item.kind,
+                "payload": item.payload,
+                "evidence_uids": list(item.evidence_uids),
+                "confidence": item.confidence,
+                "source": item.source,
+                "model_version": item.model_version,
+                "prompt_version": item.prompt_version,
+            }
+            if not isinstance(item, dict)
+            else item
+            for item in records
+        ]
+        return jsonify(success(data)), 201
+    except Exception as error:
+        return _error(error)
+
+
+@bp.route(
+    "/ontologies/<ontology_uid>/change-sets/<change_set_uid>/decisions",
+    methods=["POST"],
+)
+def decide_ontology_change_set(ontology_uid, change_set_uid):
+    payload = request.get_json(silent=True) or {}
+    try:
+        result = get_ontology_dynamic_service().decide(
+            ontology_uid,
+            change_set_uid,
+            payload.get("decisions") or [],
+            _identity().get("id") or _identity().get("sub"),
+        )
+        return jsonify(success(result)), 200
+    except Exception as error:
+        return _error(error)
+
+
+@bp.route("/ontologies/<ontology_uid>/export", methods=["GET"])
+def export_ontology(ontology_uid):
+    try:
+        content, media_type, filename = get_ontology_exchange_service().export(
+            ontology_uid, request.args.get("format") or "json"
+        )
+        return send_file(
+            io.BytesIO(content),
+            mimetype=media_type,
+            as_attachment=True,
+            download_name=filename,
+        )
+    except Exception as error:
+        return _error(error)
+
+
+@bp.route("/ontologies/import", methods=["POST"])
+def import_ontology():
+    try:
+        result = get_ontology_exchange_service().import_document(
+            request.get_data(cache=False),
+            request.args.get("format") or "json",
+            _identity().get("id") or _identity().get("sub"),
+        )
+        return jsonify(success(result)), 201
+    except Exception as error:
+        return _error(error)
+
+
+@bp.route("/semantic/properties/<property_uid>", methods=["GET"])
+def query_semantic_property(property_uid):
+    domain = str(request.args.get("business_domain_uid") or "").strip()
+    identity = _identity()
+    scoped = set(identity.get("business_domains") or [domain])
+    if "*" not in scoped and domain not in scoped:
+        return jsonify(failed("权限不足", code=403)), 403
+    try:
+        result = get_semantic_query_service().trace_property(
+            property_uid,
+            allowed_domains={domain},
+            limit=int(request.args.get("limit") or 20),
+            after_uid=request.args.get("after_uid"),
+        )
+        return jsonify(success(result)), 200
+    except Exception as error:
+        return _error(error)

app/core/data_research/ontology/change_sets.py

@@ -0,0 +1,170 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, replace
+
+from app.core.common.identifiers import new_governance_uid
+
+
+class ChangeSetConflict(ValueError):
+    pass
+
+
+@dataclass(frozen=True)
+class ChangeDecision:
+    suggestion_uid: str
+    decision: str
+    actor_uid: str
+    reason: str
+    payload: dict | None = None
+
+
+@dataclass(frozen=True)
+class ChangeSet:
+    uid: str
+    ontology_uid: str
+    suggestions: tuple
+    decisions: tuple[ChangeDecision, ...]
+    created_by: str
+
+
+class ChangeSetService:
+    def __init__(self, *, uid_factory=new_governance_uid):
+        self.uid_factory = uid_factory
+
+    def create(self, ontology_uid, suggestions, *, actor_uid):
+        targets = {}
+        for suggestion in suggestions:
+            target = (suggestion.kind, str(suggestion.payload.get("uid")))
+            if target in targets and targets[target].payload != suggestion.payload:
+                raise ChangeSetConflict("multiple suggestions conflict on the same target")
+            targets[target] = suggestion
+        return ChangeSet(self.uid_factory(), str(ontology_uid), tuple(suggestions), (), str(actor_uid))
+
+    def decide(self, change_set, decisions, *, actor_uid):
+        known = {item.uid: item for item in change_set.suggestions}
+        audited = list(change_set.decisions)
+        decided_uids = {item.suggestion_uid for item in audited}
+        for item in decisions:
+            uid = str(item.get("suggestion_uid") or "")
+            decision = str(item.get("decision") or "")
+            if uid not in known or decision not in {"accept", "reject", "edit"}:
+                raise ValueError("invalid change-set decision")
+            if uid in decided_uids:
+                raise ValueError("change-set decision is immutable")
+            payload = dict(item.get("payload") or {}) if decision == "edit" else None
+            if decision == "edit" and not payload:
+                raise ValueError("edited suggestion payload is required")
+            audited.append(
+                ChangeDecision(
+                    uid,
+                    decision,
+                    str(actor_uid),
+                    str(item.get("reason") or "").strip(),
+                    payload,
+                )
+            )
+            decided_uids.add(uid)
+        return replace(change_set, decisions=tuple(audited))
+
+    def apply(self, graph, change_set):
+        result = {
+            "classes": list(graph.get("classes") or []),
+            "properties": list(graph.get("properties") or []),
+            "relations": list(graph.get("relations") or []),
+            "constraints": list(graph.get("constraints") or []),
+            "domain_links": list(graph.get("domain_links") or []),
+            "element_mappings": list(graph.get("element_mappings") or []),
+        }
+        suggestions = {item.uid: item for item in change_set.suggestions}
+        sections = {"class": "classes", "property": "properties", "relation": "relations", "constraint": "constraints"}
+        for decision in change_set.decisions:
+            if decision.decision == "reject":
+                continue
+            suggestion = suggestions[decision.suggestion_uid]
+            payload = dict(decision.payload or suggestion.payload)
+            section = sections.get(suggestion.kind)
+            if section:
+                result[section].append(payload)
+        return result
+
+
+class SqlAlchemyDynamicOntologyService:
+    def __init__(self, session, *, suggestion_engine=None, change_set_service=None):
+        from app.core.data_research.ontology.suggestions import OntologySuggestionEngine
+
+        self.session = session
+        self.suggestion_engine = suggestion_engine or OntologySuggestionEngine()
+        self.change_set_service = change_set_service or ChangeSetService()
+
+    @staticmethod
+    def _suggestion_dict(item):
+        return {
+            "uid": item.uid,
+            "kind": item.kind,
+            "payload": item.payload,
+            "evidence_uids": list(item.evidence_uids),
+            "confidence": item.confidence,
+            "source": item.source,
+            "model_version": item.model_version,
+            "prompt_version": item.prompt_version,
+        }
+
+    def generate(self, ontology_uid, payload, actor_uid):
+        from app.models.data_research import OntologyChangeSetModel
+
+        suggestions = self.suggestion_engine.generate(
+            domains=list(payload.get("domains") or []),
+            data_elements=list(payload.get("data_elements") or []),
+            foreign_keys=list(payload.get("foreign_keys") or []),
+            include_ai=bool(payload.get("include_ai", False)),
+        )
+        change_set = self.change_set_service.create(
+            ontology_uid, suggestions, actor_uid=actor_uid
+        )
+        self.session.add(
+            OntologyChangeSetModel(
+                uid=change_set.uid,
+                ontology_uid=str(ontology_uid),
+                changes=[self._suggestion_dict(item) for item in suggestions],
+                decisions=[],
+                created_by=actor_uid,
+            )
+        )
+        self.session.commit()
+        return suggestions
+
+    def decide(self, ontology_uid, change_set_uid, decisions, actor_uid):
+        from app.core.data_research.ontology.suggestions import Suggestion
+        from app.models.data_research import OntologyChangeSetModel
+
+        model = self.session.get(OntologyChangeSetModel, str(change_set_uid))
+        if model is None or str(model.ontology_uid) != str(ontology_uid):
+            raise LookupError("ontology change set was not found")
+        suggestions = tuple(
+            Suggestion(
+                item["uid"],
+                item["kind"],
+                dict(item["payload"]),
+                tuple(item.get("evidence_uids") or []),
+                float(item["confidence"]),
+                item["source"],
+                item.get("model_version"),
+                item.get("prompt_version"),
+            )
+            for item in model.changes
+        )
+        change_set = ChangeSet(
+            str(model.uid), str(model.ontology_uid), suggestions, (), model.created_by
+        )
+        decided = self.change_set_service.decide(
+            change_set, decisions, actor_uid=actor_uid
+        )
+        model.decisions = [item.__dict__ for item in decided.decisions]
+        model.status = "reviewed"
+        self.session.commit()
+        return {
+            "uid": str(model.uid),
+            "ontology_uid": str(model.ontology_uid),
+            "decisions": list(model.decisions),
+            "actor_uid": actor_uid,
+        }

app/core/data_research/ontology/exchange.py

@@ -0,0 +1,147 @@
+from __future__ import annotations
+
+import json
+import xml.etree.ElementTree as ET
+
+from app.core.data_research.ontology.models import GRAPH_SECTIONS, GraphDocument
+
+
+class ExchangeInvalid(ValueError):
+    pass
+
+
+SECRET_KEYS = {"password", "secret", "token", "api_key", "authorization", "credentials"}
+OWL_NS = "http://www.w3.org/2002/07/owl#"
+RDF_NS = "http://www.w3.org/1999/02/22-rdf-syntax-ns#"
+DATAOPS_NS = "https://dataops.local/ontology#"
+
+
+def _safe(value, key=""):
+    if key.casefold() in SECRET_KEYS:
+        return "[redacted]"
+    if isinstance(value, dict):
+        return {name: _safe(item, name) for name, item in sorted(value.items())}
+    if isinstance(value, list):
+        return [_safe(item) for item in value]
+    return value
+
+
+class OntologyExchange:
+    def __init__(self, *, max_import_bytes=5 * 1024 * 1024):
+        self.max_import_bytes = int(max_import_bytes)
+
+    def _bounded(self, content):
+        if len(content) > self.max_import_bytes:
+            raise ExchangeInvalid("ontology import exceeds size limit")
+
+    def export_json(self, ontology_uid, version, graph):
+        payload = {
+            "format": "dataops-ontology-json-v1",
+            "ontology_uid": str(ontology_uid),
+            "version": int(version),
+            "graph_document": _safe(GraphDocument.from_dict(graph).to_dict()),
+        }
+        return json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
+
+    def import_json(self, content):
+        self._bounded(content)
+        try:
+            payload = json.loads(content.decode("utf-8"))
+            return {
+                "ontology_uid": str(payload["ontology_uid"]),
+                "version": int(payload["version"]),
+                "graph_document": GraphDocument.from_dict(payload["graph_document"]).to_dict(),
+            }
+        except (KeyError, TypeError, ValueError, UnicodeDecodeError) as exc:
+            raise ExchangeInvalid("invalid ontology JSON") from exc
+
+    def export_owl(self, ontology_uid, version, graph):
+        ET.register_namespace("owl", OWL_NS)
+        ET.register_namespace("rdf", RDF_NS)
+        ET.register_namespace("dataops", DATAOPS_NS)
+        root = ET.Element(f"{{{RDF_NS}}}RDF")
+        ontology = ET.SubElement(root, f"{{{OWL_NS}}}Ontology")
+        ontology.set(f"{{{RDF_NS}}}about", f"urn:dataops:ontology:{ontology_uid}")
+        ontology.set(f"{{{DATAOPS_NS}}}uid", str(ontology_uid))
+        ontology.set(f"{{{DATAOPS_NS}}}version", str(int(version)))
+        safe_graph = _safe(GraphDocument.from_dict(graph).to_dict())
+        for section in GRAPH_SECTIONS:
+            node = ET.SubElement(ontology, f"{{{DATAOPS_NS}}}{section}")
+            node.text = json.dumps(safe_graph[section], ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+        return ET.tostring(root, encoding="utf-8", xml_declaration=True)
+
+    def import_owl(self, content):
+        self._bounded(content)
+        upper = content.upper()
+        if b"<!DOCTYPE" in upper or b"<!ENTITY" in upper:
+            raise ExchangeInvalid("unsafe XML declarations are forbidden")
+        try:
+            root = ET.fromstring(content)
+            ontology = root.find(f"{{{OWL_NS}}}Ontology")
+            if ontology is None:
+                raise ValueError("missing ontology")
+            graph = {}
+            for section in GRAPH_SECTIONS:
+                node = ontology.find(f"{{{DATAOPS_NS}}}{section}")
+                graph[section] = json.loads(node.text or "[]") if node is not None else []
+            return {
+                "ontology_uid": str(ontology.attrib[f"{{{DATAOPS_NS}}}uid"]),
+                "version": int(ontology.attrib[f"{{{DATAOPS_NS}}}version"]),
+                "graph_document": GraphDocument.from_dict(graph).to_dict(),
+            }
+        except (ET.ParseError, KeyError, TypeError, ValueError, json.JSONDecodeError) as exc:
+            raise ExchangeInvalid("invalid ontology RDF/OWL XML") from exc
+
+
+class OntologyExchangeService:
+    def __init__(self, repository, *, exchange=None, commit=lambda: None, rollback=lambda: None):
+        self.repository = repository
+        self.exchange = exchange or OntologyExchange()
+        self.commit = commit
+        self.rollback = rollback
+
+    def export(self, ontology_uid, format_name):
+        version = self.repository.latest_version(ontology_uid)
+        if version is None or version.status not in {"published", "superseded"}:
+            raise LookupError("published ontology version was not found")
+        graph = version.graph_document.to_dict()
+        if format_name == "json":
+            content = self.exchange.export_json(ontology_uid, version.version, graph)
+            return content, "application/json", f"{ontology_uid}-v{version.version}.json"
+        if format_name in {"rdf", "owl"}:
+            content = self.exchange.export_owl(ontology_uid, version.version, graph)
+            return content, "application/rdf+xml", f"{ontology_uid}-v{version.version}.owl"
+        raise ExchangeInvalid("unsupported ontology exchange format")
+
+    def import_document(self, content, format_name, actor_uid):
+        try:
+            imported = (
+                self.exchange.import_json(content)
+                if format_name == "json"
+                else self.exchange.import_owl(content)
+            )
+            code = "IMPORTED_" + imported["ontology_uid"].replace("-", "_").upper()
+            ontology = self.repository.create(
+                code=code,
+                name=code,
+                owner_uid=actor_uid,
+                domain_links=imported["graph_document"].get("domain_links") or [],
+                created_by=actor_uid,
+            )
+            version = self.repository.save_draft(
+                ontology.uid,
+                imported["graph_document"],
+                expected_revision=0,
+                actor_uid=actor_uid,
+            )
+            self.commit()
+            return {
+                "ontology_uid": ontology.uid,
+                "version_uid": version.uid,
+                "version": version.version,
+                "format": format_name,
+                "actor_uid": actor_uid,
+            }
+        except Exception:
+            self.rollback()
+            raise

app/core/data_research/ontology/knowledge_sync.py

@@ -0,0 +1,30 @@
+from __future__ import annotations
+
+from app.core.knowledge.document_builder import build_document, chunk_document
+
+
+class OntologyKnowledgeSync:
+    def __init__(self, repository, *, max_chunk_chars=1200):
+        self.repository = repository
+        self.max_chunk_chars = int(max_chunk_chars)
+
+    def sync(self, published_version):
+        if published_version.get("status") != "published":
+            raise ValueError("only a published ontology version can be synchronized")
+        source = {
+            "uid": str(published_version["uid"]),
+            "name": published_version.get("name") or published_version["uid"],
+            "version": int(published_version["version"]),
+            "status": "published",
+            "graph_document": published_version.get("graph_document") or {},
+            "token": published_version.get("token"),
+        }
+        document = build_document("ontology", source)
+        changed = self.repository.upsert_document(document)
+        key = (document["object_uid"], document["object_version"])
+        if changed or key not in getattr(self.repository, "chunks", {}):
+            self.repository.replace_chunks(
+                key,
+                chunk_document(document["content"], self.max_chunk_chars),
+            )
+        return {"changed": changed, "document": document}

app/core/data_research/ontology/query.py

@@ -0,0 +1,72 @@
+from __future__ import annotations
+
+
+SAFE_FIELDS = (
+    "path_uid",
+    "ontology_uid",
+    "property_uid",
+    "data_element_uid",
+    "field_uid",
+    "evidence_uid",
+    "business_domain_uid",
+)
+
+
+class SemanticQueryService:
+    def __init__(self, repository):
+        self.repository = repository
+
+    def trace_property(self, property_uid, *, allowed_domains, limit=20, after_uid=None):
+        if isinstance(limit, bool) or not isinstance(limit, int) or limit < 1 or limit > 50:
+            raise ValueError("limit must be between 1 and 50")
+        rows = self.repository.find_property_paths(
+            str(property_uid), depth=3, limit=limit + 1, after_uid=after_uid
+        )
+        safe = []
+        for row in sorted(rows, key=lambda item: str(item.get("path_uid"))):
+            if row.get("ontology_status") != "published":
+                continue
+            domain = str(row.get("business_domain_uid") or "")
+            if "*" not in allowed_domains and domain not in allowed_domains:
+                continue
+            safe.append({key: row.get(key) for key in SAFE_FIELDS if row.get(key) is not None})
+        truncated = len(safe) > limit
+        items = safe[:limit]
+        return {
+            "items": items,
+            "count": len(items),
+            "truncated": truncated,
+            "next_cursor": items[-1]["path_uid"] if truncated and items else None,
+        }
+
+
+class Neo4jSemanticRepository:
+    STATEMENT = """
+    MATCH (ontology:Ontology)-[:ACTIVE_VERSION]->(:OntologyVersion)
+    MATCH (ontology)-[:HAS_PROPERTY]->(property:OntologyProperty {uid: $property_uid})
+    OPTIONAL MATCH (property)-[:MAPS_TO]->(element:DataElement)
+    OPTIONAL MATCH (element)-[:IMPLEMENTED_BY]->(field:PhysicalField)
+    OPTIONAL MATCH (field)-[:HAS_EVIDENCE]->(evidence:Evidence)
+    OPTIONAL MATCH (ontology)-[:SERVES_DOMAIN]->(domain:BusinessDomain)
+    WHERE $after_uid IS NULL OR property.uid > $after_uid
+    RETURN property.uid AS path_uid, ontology.uid AS ontology_uid,
+           'published' AS ontology_status, property.uid AS property_uid,
+           element.uid AS data_element_uid, field.uid AS field_uid,
+           evidence.uid AS evidence_uid, domain.uid AS business_domain_uid
+    ORDER BY path_uid LIMIT $limit
+    """
+
+    def __init__(self, driver):
+        self.driver = driver
+
+    def find_property_paths(self, property_uid, *, depth, limit, after_uid):
+        if int(depth) != 3:
+            raise ValueError("semantic traversal depth is fixed")
+        with self.driver.get_session() as session:
+            result = session.run(
+                self.STATEMENT,
+                property_uid=str(property_uid),
+                after_uid=after_uid,
+                limit=int(limit),
+            )
+            return [dict(record) for record in result]

app/core/data_research/ontology/suggestions.py

@@ -0,0 +1,78 @@
+from __future__ import annotations
+
+import hashlib
+import json
+from dataclasses import dataclass
+
+
+class SuggestionInvalid(ValueError):
+    pass
+
+
+@dataclass(frozen=True)
+class Suggestion:
+    uid: str
+    kind: str
+    payload: dict
+    evidence_uids: tuple[str, ...]
+    confidence: float
+    source: str
+    model_version: str | None = None
+    prompt_version: str | None = None
+
+
+def _uid(kind, payload):
+    value = json.dumps([kind, payload], ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+    return "suggestion-" + hashlib.sha256(value.encode()).hexdigest()[:16]
+
+
+class OntologySuggestionEngine:
+    def __init__(self, *, ai_port=None, model_version=None, prompt_version=None):
+        self.ai_port = ai_port
+        self.model_version = model_version
+        self.prompt_version = prompt_version
+
+    def generate(self, *, domains, data_elements, foreign_keys, include_ai=False):
+        suggestions = []
+        for domain in sorted(domains, key=lambda item: str(item.get("uid"))):
+            payload = {"uid": f"class-{domain['uid']}", "name": domain.get("name") or domain["uid"]}
+            suggestions.append(Suggestion(_uid("class", payload), "class", payload, (str(domain["uid"]),), 0.98, "rule"))
+        for element in sorted(data_elements, key=lambda item: str(item.get("uid"))):
+            if element.get("status") != "published":
+                continue
+            payload = {
+                "uid": f"property-{element['uid']}",
+                "name": element.get("name") or element["uid"],
+                "data_element_uid": str(element["uid"]),
+                "aliases": list(element.get("aliases") or []),
+            }
+            suggestions.append(Suggestion(_uid("property", payload), "property", payload, (str(element["uid"]),), 0.92, "rule"))
+        for relation in sorted(foreign_keys, key=lambda item: (str(item.get("from_uid")), str(item.get("to_uid")))):
+            payload = {
+                "uid": f"relation-{relation['from_uid']}-{relation['to_uid']}",
+                "from_class_uid": str(relation["from_uid"]),
+                "to_class_uid": str(relation["to_uid"]),
+            }
+            suggestions.append(Suggestion(_uid("relation", payload), "relation", payload, (str(relation["evidence_uid"]),), 0.85, "rule"))
+        if include_ai:
+            if self.ai_port is None:
+                raise SuggestionInvalid("AI suggestion provider is not configured")
+            for item in self.ai_port({"domains": domains, "data_elements": data_elements, "foreign_keys": foreign_keys}):
+                evidence = tuple(str(value) for value in item.get("evidence_uids") or ())
+                if not evidence:
+                    raise SuggestionInvalid("AI suggestion evidence is required")
+                payload = dict(item.get("payload") or {})
+                suggestions.append(
+                    Suggestion(
+                        _uid(str(item["kind"]), payload),
+                        str(item["kind"]),
+                        payload,
+                        evidence,
+                        float(item.get("confidence") or 0),
+                        "ai",
+                        self.model_version,
+                        self.prompt_version,
+                    )
+                )
+        return sorted(suggestions, key=lambda item: (-item.confidence, item.uid))
+

app/core/mcp/context.py

@@ -285,3 +285,32 @@ class ContextService:
                 _untrusted(value) for value in list(row.get("differences") or [])[:50]
             ],
         }
+
+    def get_semantic_property_context(
+        self,
+        identity,
+        property_uid,
+        *,
+        business_domain,
+        limit=20,
+    ):
+        identity.require_domain(business_domain)
+        if isinstance(limit, bool) or not isinstance(limit, int) or limit < 1 or limit > 50:
+            raise ValueError("limit must be between 1 and 50")
+        fields = (
+            "path_uid",
+            "property_uid",
+            "ontology_uid",
+            "business_domain_uid",
+            "data_element_uid",
+            "field_uid",
+            "evidence_uid",
+        )
+        safe = []
+        for row in self.repository.list_semantic_property_context(str(property_uid)):
+            if row.get("ontology_status") != "published":
+                continue
+            if row.get("business_domain_uid") != business_domain:
+                continue
+            safe.append({key: row.get(key) for key in fields if row.get(key) is not None})
+        return _page(safe, limit)

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: 133
+x-route-count: 138
 servers:
   - url: "http://localhost:15500"
     description: "全本地隔离测试后端"
@@ -1928,6 +1928,69 @@ paths:
             application/json:
               schema:
                 $ref: '#/components/schemas/ApiEnvelope'
+  "/api/development/v1/ontologies/import":
+    post:
+      tags: [data_development]
+      operationId: data_development_import_ontology_post
+      summary: "import 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}/change-sets/{change_set_uid}/decisions":
+    post:
+      tags: [data_development]
+      operationId: data_development_decide_ontology_change_set_post
+      summary: "decide ontology change set"
+      x-source: "app/api/data_development/routes.py"
+      parameters:
+        - name: ontology_uid
+          in: path
+          required: true
+          schema:
+            type: string
+        - name: change_set_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}/diff":
     get:
       tags: [data_development]
@@ -1953,6 +2016,31 @@ paths:
             application/json:
               schema:
                 $ref: '#/components/schemas/ApiEnvelope'
+  "/api/development/v1/ontologies/{ontology_uid}/export":
+    get:
+      tags: [data_development]
+      operationId: data_development_export_ontology_get
+      summary: "export 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]
@@ -2049,6 +2137,38 @@ paths:
             application/json:
               schema:
                 $ref: '#/components/schemas/ApiEnvelope'
+  "/api/development/v1/ontologies/{ontology_uid}/suggestions":
+    post:
+      tags: [data_development]
+      operationId: data_development_generate_ontology_suggestions_post
+      summary: "generate ontology suggestions"
+      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]
@@ -2081,6 +2201,31 @@ paths:
             application/json:
               schema:
                 $ref: '#/components/schemas/ApiEnvelope'
+  "/api/development/v1/semantic/properties/{property_uid}":
+    get:
+      tags: [data_development]
+      operationId: data_development_query_semantic_property_get
+      summary: "query semantic property"
+      x-source: "app/api/data_development/routes.py"
+      parameters:
+        - name: property_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/sources/files":
     post:
       tags: [data_development]

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

@@ -293,10 +293,10 @@
 - Produces deterministic suggestions from business domains, published data elements, foreign-key evidence, and aliases before optional AI suggestions.
 - Produces decisions `accept`, `reject`, `edit` with actor, reason, evidence UIDs, confidence, and model/prompt version.
 
-- [ ] Write tests for deterministic class/property/relation proposals, evidence-required AI proposals, confidence ordering, conflict detection, partial decisions, and the guarantee that undecided suggestions never publish.
-- [ ] Run focused tests and verify missing services.
-- [ ] Implement rule suggestions, injected AI suggestion port, immutable decision audit, and accepted-change application to a draft version.
-- [ ] Rerun focused tests and ontology publication regressions.
+- [x] Write tests for deterministic class/property/relation proposals, evidence-required AI proposals, confidence ordering, conflict detection, partial decisions, and the guarantee that undecided suggestions never publish.
+- [x] Run focused tests and verify missing services.
+- [x] Implement rule suggestions, injected AI suggestion port, immutable decision audit, and accepted-change application to a draft version.
+- [x] Rerun focused tests and ontology publication regressions.
 
 ### Task 14: Add JSON/RDF/OWL exchange and governance knowledge synchronization
 
@@ -311,10 +311,10 @@
 - Produces deterministic JSON export and bounded RDF/OWL XML export/import with stable platform UIDs.
 - Consumes: `app/core/knowledge/document_builder.py`; produces versioned governance documents only for published ontology versions.
 
-- [ ] Write round-trip tests for classes/properties/relations/domain links, unsafe XML rejection, import size limits, deterministic hashes, secret redaction, and knowledge-document version/hash behavior.
-- [ ] Run focused tests and verify missing exchange/sync modules.
-- [ ] Implement bounded import/export, canonical namespaces, publication-triggered knowledge documents, and idempotent chunk sync.
-- [ ] Rerun focused tests plus governance document builder tests.
+- [x] Write round-trip tests for classes/properties/relations/domain links, unsafe XML rejection, import size limits, deterministic hashes, secret redaction, and knowledge-document version/hash behavior.
+- [x] Run focused tests and verify missing exchange/sync modules.
+- [x] Implement bounded import/export, canonical namespaces, publication-triggered knowledge documents, and idempotent chunk sync.
+- [x] Rerun focused tests plus governance document builder tests.
 
 ### Task 15: Expose read-only semantic queries and MCP context
 
@@ -329,15 +329,15 @@
 - Produces bounded read-only queries from ontology property to data element, physical field, source evidence, and business domain.
 - Produces MCP context tools that return published data only and enforce permission scope.
 
-- [ ] Write tests for bounded traversal, unpublished exclusion, permission filtering, stable pagination, no arbitrary Cypher input, and MCP result redaction.
-- [ ] Run focused tests and verify missing query/tool behavior.
-- [ ] Implement allowlisted query methods and MCP context adapters.
-- [ ] Rerun focused tests and existing MCP security/runtime suites.
+- [x] Write tests for bounded traversal, unpublished exclusion, permission filtering, stable pagination, no arbitrary Cypher input, and MCP result redaction.
+- [x] Run focused tests and verify missing query/tool behavior.
+- [x] Implement allowlisted query methods and MCP context adapters.
+- [x] Rerun focused tests and existing MCP security/runtime suites.
 
 ### V64 gate
 
-- [ ] Run all V64-focused, governance knowledge, MCP, complete Python, and frontend build tests.
-- [ ] Record evidence in `docs/validation/data-research-v64.md` and commit V64.
+- [x] Run all V64-focused, governance knowledge, MCP, complete Python, and frontend build tests.
+- [x] Record evidence in `docs/validation/data-research-v64.md` and commit V64.
 
 ## V65 — Production hardening and acceptance
 

docs/validation/data-research-v64.md

@@ -0,0 +1,24 @@
+# V64 动态本体与知识服务验证
+
+日期：2026-07-22
+
+## 交付范围
+
+- 基于业务域、已发布数据元素、别名和外键证据的确定性本体建议。
+- 可注入 AI 建议端口，强制证据、置信度、模型版本与提示版本记录。
+- `accept/reject/edit` 不可变决策审计；未决建议不应用到草稿。
+- 确定性 JSON 和受限 RDF/OWL XML 交换，危险 XML 与超限导入拒绝。
+- 已发布本体版本到治理知识文档的幂等分块同步。
+- 只读、固定深度、固定 Cypher 的语义追溯及 MCP 权限上下文。
+- 动态建议、变更集决策、导入导出与语义查询 API。
+
+## 测试证据
+
+- V64 新增建议、决策、交换、知识同步、查询与 API：`14 passed`。
+- 全部本体、治理知识、MCP 与 OpenAPI 专项：`68 passed`。
+- OpenAPI 共 138 个操作，生成器可重复性通过。
+- 全量后端回归：`363 passed, 23 skipped, 59 subtests passed`。
+- Vue 2 生产构建：成功；仅既有依赖年龄、CSS 顺序、包体积和 `no-console` 警告。
+- `git diff --check`：通过。
+
+语义查询不接受任意 Cypher；只返回已发布本体、调用者业务域范围内的稳定字段，并移除令牌、密码等非白名单字段。

tests/data_research/test_ontology_change_sets.py

@@ -0,0 +1,56 @@
+from __future__ import annotations
+
+import pytest
+
+
+def suggestions():
+    from app.core.data_research.ontology.suggestions import Suggestion
+
+    return [
+        Suggestion("s-1", "class", {"uid": "customer", "name": "Customer"}, ("domain-1",), 0.95, "rule"),
+        Suggestion("s-2", "property", {"uid": "name", "name": "name", "class_uid": "customer"}, ("element-1",), 0.9, "rule"),
+    ]
+
+
+def test_partial_decisions_apply_only_accepted_or_edited_changes():
+    from app.core.data_research.ontology.change_sets import ChangeSetService
+
+    service = ChangeSetService(uid_factory=lambda: "change-set-1")
+    change_set = service.create("ontology-1", suggestions(), actor_uid="editor-1")
+    decided = service.decide(
+        change_set,
+        [{"suggestion_uid": "s-1", "decision": "accept", "reason": "符合业务定义"}],
+        actor_uid="reviewer-1",
+    )
+    graph = service.apply({"classes": [], "properties": []}, decided)
+
+    assert graph["classes"][0]["uid"] == "customer"
+    assert graph["properties"] == []  # undecided suggestions never apply
+    assert decided.decisions[0].actor_uid == "reviewer-1"
+    assert decided.decisions[0].reason == "符合业务定义"
+
+
+def test_edit_reject_and_conflict_detection_are_audited():
+    from app.core.data_research.ontology.change_sets import ChangeSetConflict, ChangeSetService
+
+    service = ChangeSetService(uid_factory=lambda: "change-set-1")
+    change_set = service.create("ontology-1", suggestions(), actor_uid="editor-1")
+    decided = service.decide(
+        change_set,
+        [
+            {"suggestion_uid": "s-1", "decision": "edit", "payload": {"uid": "customer", "name": "Customer360"}, "reason": "统一命名"},
+            {"suggestion_uid": "s-2", "decision": "reject", "reason": "定义不充分"},
+        ],
+        actor_uid="reviewer-1",
+    )
+    assert service.apply({"classes": [], "properties": []}, decided)["classes"][0]["name"] == "Customer360"
+
+    duplicate = suggestions() + [replace_suggestion(suggestions()[0], uid="s-3")]
+    with pytest.raises(ChangeSetConflict, match="target"):
+        service.create("ontology-1", duplicate, actor_uid="editor-1")
+
+
+def replace_suggestion(value, *, uid):
+    from dataclasses import replace
+
+    return replace(value, uid=uid, payload={**value.payload, "name": "Conflicting"})

tests/data_research/test_ontology_dynamic_api.py

@@ -0,0 +1,82 @@
+from __future__ import annotations
+
+import pytest
+
+
+class DynamicService:
+    def generate(self, ontology_uid, payload, actor_uid):
+        return [{"uid": "suggestion-1", "kind": "class", "confidence": 0.95, "evidence_uids": ["domain-1"]}]
+
+    def decide(self, ontology_uid, change_set_uid, decisions, actor_uid):
+        return {"uid": change_set_uid, "ontology_uid": ontology_uid, "decisions": decisions, "actor_uid": actor_uid}
+
+
+class ExchangeService:
+    def export(self, ontology_uid, format_name):
+        assert (ontology_uid, format_name) == ("ontology-1", "json")
+        return b'{"ontology_uid":"ontology-1"}', "application/json", "ontology-1.json"
+
+    def import_document(self, content, format_name, actor_uid):
+        return {"ontology_uid": "ontology-2", "version": 1, "format": format_name, "actor_uid": actor_uid}
+
+
+class SemanticService:
+    def trace_property(self, uid, *, allowed_domains, limit, after_uid):
+        return {"items": [{"property_uid": uid, "business_domain_uid": next(iter(allowed_domains))}], "count": 1, "truncated": False, "next_cursor": None}
+
+
+@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
+
+    def identity():
+        role = request.headers.get("Authorization", "").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_dynamic_service", lambda: DynamicService())
+    monkeypatch.setattr(routes, "get_ontology_exchange_service", lambda: ExchangeService())
+    monkeypatch.setattr(routes, "get_semantic_query_service", lambda: SemanticService())
+    app = create_app()
+    app.config.update(TESTING=True)
+    return app.test_client()
+
+
+def test_dynamic_suggestion_and_audited_decision_endpoints(client):
+    headers = {"Authorization": "Bearer editor"}
+    generated = client.post(
+        "/api/development/v1/ontologies/ontology-1/suggestions",
+        headers=headers,
+        json={"domains": [], "data_elements": [], "foreign_keys": []},
+    )
+    decided = client.post(
+        "/api/development/v1/ontologies/ontology-1/change-sets/change-1/decisions",
+        headers=headers,
+        json={"decisions": [{"suggestion_uid": "suggestion-1", "decision": "accept"}]},
+    )
+    assert generated.get_json()["data"][0]["evidence_uids"] == ["domain-1"]
+    assert decided.get_json()["data"]["actor_uid"] == "editor-1"
+
+
+def test_exchange_and_read_only_semantic_query_endpoints(client):
+    exported = client.get(
+        "/api/development/v1/ontologies/ontology-1/export?format=json",
+        headers={"Authorization": "Bearer viewer"},
+    )
+    imported = client.post(
+        "/api/development/v1/ontologies/import?format=json",
+        headers={"Authorization": "Bearer editor"},
+        data=b'{"ontology_uid":"ontology-2"}',
+        content_type="application/json",
+    )
+    queried = client.get(
+        "/api/development/v1/semantic/properties/property-1?business_domain_uid=sales&limit=20",
+        headers={"Authorization": "Bearer viewer"},
+    )
+    assert exported.status_code == 200 and exported.mimetype == "application/json"
+    assert imported.status_code == 201
+    assert queried.get_json()["data"]["items"][0]["business_domain_uid"] == "sales"

tests/data_research/test_ontology_exchange.py

@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+import pytest
+
+from tests.data_research.test_ontology_validation import base_graph
+
+
+def test_json_and_owl_round_trip_preserve_stable_uids_and_hashes():
+    from app.core.data_research.ontology.exchange import OntologyExchange
+    from app.core.data_research.ontology.repository import canonical_graph_hash
+
+    exchange = OntologyExchange(max_import_bytes=100_000)
+    graph = base_graph()
+    json_bytes = exchange.export_json("ontology-1", 3, graph)
+    owl_bytes = exchange.export_owl("ontology-1", 3, graph)
+
+    imported_json = exchange.import_json(json_bytes)
+    imported_owl = exchange.import_owl(owl_bytes)
+    assert imported_json["ontology_uid"] == "ontology-1"
+    assert imported_owl["ontology_uid"] == "ontology-1"
+    assert imported_owl["graph_document"] == graph
+    assert canonical_graph_hash(imported_json["graph_document"]) == canonical_graph_hash(graph)
+    assert canonical_graph_hash(imported_owl["graph_document"]) == canonical_graph_hash(graph)
+
+
+def test_owl_import_rejects_unsafe_xml_and_size_limits():
+    from app.core.data_research.ontology.exchange import ExchangeInvalid, OntologyExchange
+
+    exchange = OntologyExchange(max_import_bytes=80)
+    with pytest.raises(ExchangeInvalid, match="unsafe XML"):
+        exchange.import_owl(b'<!DOCTYPE x [<!ENTITY e SYSTEM "file:///etc/passwd">]><x>&e;</x>')
+    with pytest.raises(ExchangeInvalid, match="size limit"):
+        exchange.import_json(b"{" + b" " * 100 + b"}")
+
+
+def test_exports_redact_secret_like_fields_deterministically():
+    from app.core.data_research.ontology.exchange import OntologyExchange
+
+    graph = base_graph()
+    graph["classes"][0]["password"] = "must-not-export"
+    first = OntologyExchange().export_json("ontology-1", 1, graph)
+    second = OntologyExchange().export_json("ontology-1", 1, graph)
+    assert first == second
+    assert b"must-not-export" not in first
+    assert b"[redacted]" in first
+

tests/data_research/test_ontology_knowledge_sync.py

@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+import pytest
+
+from tests.data_research.test_ontology_validation import base_graph
+
+
+class Repository:
+    def __init__(self):
+        self.documents = {}
+        self.chunks = {}
+
+    def upsert_document(self, document):
+        key = (document["object_uid"], document["object_version"])
+        existing = self.documents.get(key)
+        self.documents[key] = document
+        return existing is None or existing["content_hash"] != document["content_hash"]
+
+    def replace_chunks(self, key, chunks):
+        self.chunks[key] = list(chunks)
+
+
+def test_only_published_versions_sync_as_versioned_idempotent_documents():
+    from app.core.data_research.ontology.knowledge_sync import OntologyKnowledgeSync
+
+    repository = Repository()
+    service = OntologyKnowledgeSync(repository, max_chunk_chars=80)
+    payload = {
+        "uid": "ontology-1",
+        "name": "客户本体",
+        "version": 3,
+        "status": "published",
+        "graph_document": base_graph(),
+        "token": "must-not-sync",
+    }
+    first = service.sync(payload)
+    second = service.sync(payload)
+
+    assert first["changed"] is True
+    assert second["changed"] is False
+    assert first["document"]["object_version"] == 3
+    assert "must-not-sync" not in first["document"]["content"]
+    assert len(repository.chunks[("ontology-1", 3)]) > 1
+
+    with pytest.raises(ValueError, match="published"):
+        service.sync({**payload, "status": "draft"})

tests/data_research/test_ontology_suggestions.py

@@ -0,0 +1,56 @@
+from __future__ import annotations
+
+import pytest
+
+
+def test_rule_suggestions_are_deterministic_explainable_and_confidence_sorted():
+    from app.core.data_research.ontology.suggestions import OntologySuggestionEngine
+
+    engine = OntologySuggestionEngine()
+    suggestions = engine.generate(
+        domains=[{"uid": "domain-customer", "name": "客户"}],
+        data_elements=[
+            {"uid": "element-name", "name": "客户名称", "status": "published", "aliases": ["姓名"]},
+            {"uid": "element-id", "name": "客户编号", "status": "published", "aliases": []},
+            {"uid": "draft", "name": "未发布", "status": "draft"},
+        ],
+        foreign_keys=[
+            {"from_uid": "customer", "to_uid": "account", "evidence_uid": "evidence-fk"}
+        ],
+    )
+
+    assert [item.kind for item in suggestions[:3]] == ["class", "property", "property"]
+    assert suggestions == engine.generate(
+        domains=[{"uid": "domain-customer", "name": "客户"}],
+        data_elements=[
+            {"uid": "element-name", "name": "客户名称", "status": "published", "aliases": ["姓名"]},
+            {"uid": "element-id", "name": "客户编号", "status": "published", "aliases": []},
+            {"uid": "draft", "name": "未发布", "status": "draft"},
+        ],
+        foreign_keys=[{"from_uid": "customer", "to_uid": "account", "evidence_uid": "evidence-fk"}],
+    )
+    assert all(item.evidence_uids for item in suggestions)
+    assert [item.confidence for item in suggestions] == sorted(
+        [item.confidence for item in suggestions], reverse=True
+    )
+
+
+def test_ai_suggestion_requires_evidence_and_records_model_versions():
+    from app.core.data_research.ontology.suggestions import (
+        OntologySuggestionEngine,
+        SuggestionInvalid,
+    )
+
+    ai = lambda _context: [
+        {
+            "kind": "relation",
+            "payload": {"uid": "rel-ai", "from_class_uid": "a", "to_class_uid": "b"},
+            "evidence_uids": [],
+            "confidence": 0.7,
+        }
+    ]
+    with pytest.raises(SuggestionInvalid, match="evidence"):
+        OntologySuggestionEngine(ai_port=ai, model_version="model-v1", prompt_version="prompt-v2").generate(
+            domains=[], data_elements=[], foreign_keys=[], include_ai=True
+        )
+

tests/data_research/test_semantic_query.py

@@ -0,0 +1,65 @@
+from __future__ import annotations
+
+import pytest
+
+
+class Repository:
+    def __init__(self):
+        self.calls = []
+
+    def find_property_paths(self, property_uid, *, depth, limit, after_uid):
+        self.calls.append((property_uid, depth, limit, after_uid))
+        return [
+            {
+                "path_uid": "path-1",
+                "ontology_uid": "ontology-1",
+                "ontology_status": "published",
+                "property_uid": property_uid,
+                "data_element_uid": "element-1",
+                "field_uid": "field-1",
+                "evidence_uid": "evidence-1",
+                "business_domain_uid": "sales",
+                "password": "must-not-leak",
+            },
+            {
+                "path_uid": "path-2",
+                "ontology_uid": "ontology-draft",
+                "ontology_status": "draft",
+                "property_uid": property_uid,
+                "business_domain_uid": "sales",
+            },
+            {
+                "path_uid": "path-3",
+                "ontology_uid": "ontology-hr",
+                "ontology_status": "published",
+                "property_uid": property_uid,
+                "business_domain_uid": "hr",
+            },
+        ]
+
+
+def test_semantic_query_is_bounded_published_permission_filtered_and_redacted():
+    from app.core.data_research.ontology.query import SemanticQueryService
+
+    repository = Repository()
+    result = SemanticQueryService(repository).trace_property(
+        "property-1", allowed_domains={"sales"}, limit=10, after_uid=None
+    )
+    assert [item["path_uid"] for item in result["items"]] == ["path-1"]
+    assert "password" not in repr(result).lower()
+    assert repository.calls == [("property-1", 3, 11, None)]
+
+    with pytest.raises(ValueError, match="limit"):
+        SemanticQueryService(repository).trace_property(
+            "property-1", allowed_domains={"sales"}, limit=51
+        )
+
+
+def test_semantic_query_has_stable_cursor_and_no_arbitrary_cypher_interface():
+    from app.core.data_research.ontology.query import SemanticQueryService
+
+    service = SemanticQueryService(Repository())
+    result = service.trace_property("property-1", allowed_domains={"*"}, limit=1)
+    assert result["next_cursor"] == "path-1"
+    assert not hasattr(service, "execute_cypher")
+

tests/mcp/test_data_research_context.py

@@ -0,0 +1,59 @@
+from __future__ import annotations
+
+import pytest
+
+from app.core.mcp.context import ContextService
+from app.core.mcp.identity import AgentIdentity
+
+
+class Repository:
+    def list_semantic_property_context(self, property_uid):
+        assert property_uid == "property-1"
+        return [
+            {
+                "path_uid": "path-1",
+                "property_uid": property_uid,
+                "ontology_uid": "ontology-1",
+                "ontology_status": "published",
+                "business_domain_uid": "sales",
+                "data_element_uid": "element-1",
+                "field_uid": "field-1",
+                "evidence_uid": "evidence-1",
+                "token": "must-not-leak",
+            },
+            {
+                "path_uid": "path-2",
+                "property_uid": property_uid,
+                "ontology_uid": "ontology-2",
+                "ontology_status": "draft",
+                "business_domain_uid": "sales",
+            },
+        ]
+
+
+def identity(domains=frozenset({"sales"})):
+    return AgentIdentity(
+        subject="agent-viewer",
+        roles=frozenset({"viewer"}),
+        business_domains=domains,
+        environments=frozenset({"test"}),
+        correlation_id="correlation-1",
+    )
+
+
+def test_mcp_context_returns_only_published_secret_free_scoped_semantics():
+    result = ContextService(Repository()).get_semantic_property_context(
+        identity(), "property-1", business_domain="sales", limit=10
+    )
+    assert result["count"] == 1
+    assert result["items"][0]["evidence_uid"] == "evidence-1"
+    assert "must-not-leak" not in repr(result)
+    assert "token" not in repr(result).lower()
+
+
+def test_mcp_context_rejects_cross_domain_and_unbounded_limit():
+    service = ContextService(Repository())
+    with pytest.raises(PermissionError):
+        service.get_semantic_property_context(identity(), "property-1", business_domain="hr")
+    with pytest.raises(ValueError, match="limit"):
+        service.get_semantic_property_context(identity(), "property-1", business_domain="sales", limit=51)