From d41530930daa50ba68778a7cae045eafc7a2fa5b Mon Sep 17 00:00:00 2001 From: mikemikimike <13286568797@163.com> Date: Fri, 31 Jul 2026 17:15:36 +0800 Subject: [PATCH] Centralize SKOS cycle validation --- docs/reference/explorer.md | 5 +++++ semantica/context/context_graph.py | 10 +++++++++ semantica/explorer/session.py | 2 +- semantica/{explorer => }/utils/skos.py | 29 +++++++++++++------------- tests/context/test_context.py | 20 ++++++++++++++++++ tests/explorer/test_vocabulary.py | 2 +- 6 files changed, 52 insertions(+), 16 deletions(-) rename semantica/{explorer => }/utils/skos.py (64%) diff --git a/docs/reference/explorer.md b/docs/reference/explorer.md index 34da4708..d0415bdb 100644 --- a/docs/reference/explorer.md +++ b/docs/reference/explorer.md @@ -258,6 +258,11 @@ Full interactive docs at `http://localhost:8000/docs`. All endpoints accept and | `/api/vocabulary/hierarchy` | `GET` | Concept hierarchy tree | | `/api/vocabulary/import` | `POST` | Import SKOS/RDF vocabulary file | + SKOS hierarchy writes reject cycles in both `skos:broader` and + `skos:narrower` relationships. Vocabulary imports validate the complete + batch before adding nodes, while direct graph/session edge writes apply + the same invariant at the graph storage boundary. + **SPARQL:** | Endpoint | Method | Description | diff --git a/semantica/context/context_graph.py b/semantica/context/context_graph.py index ac1b9b59..802295a1 100644 --- a/semantica/context/context_graph.py +++ b/semantica/context/context_graph.py @@ -117,6 +117,7 @@ import uuid from ..utils.logging import get_logger from ..utils.progress_tracker import get_progress_tracker from ..utils.helpers import classify_path_distance +from ..utils.skos import validate_skos_hierarchy from .entity_linker import EntityLinker # Optional imports for advanced features @@ -589,6 +590,9 @@ class ContextGraph: """ count = 0 with self._lock: + # Keep the SKOS hierarchy invariant at the lowest common write + # layer so direct graph users cannot bypass API/session checks. + validate_skos_hierarchy([*self.find_edges(), *edges]) for raw_edge in edges: if not isinstance(raw_edge, dict): continue @@ -948,6 +952,12 @@ class ContextGraph: family_id=explicit_family_id, ) with self._lock: + validate_skos_hierarchy( + [ + *self.find_edges(), + {"source": source_id, "target": target_id, "type": edge_type}, + ] + ) return self._add_internal_edge( ContextEdge( edge_id=edge_id, diff --git a/semantica/explorer/session.py b/semantica/explorer/session.py index 19c3d885..9eafdb8c 100644 --- a/semantica/explorer/session.py +++ b/semantica/explorer/session.py @@ -13,7 +13,7 @@ from typing import Any, Dict, Iterable, List, Optional from ..context.context_graph import ContextGraph, _resolve_edge_identity from .search_index import GraphSearchIndex -from .utils.skos import validate_skos_hierarchy +from ..utils.skos import validate_skos_hierarchy _KG_AVAILABLE = False try: diff --git a/semantica/explorer/utils/skos.py b/semantica/utils/skos.py similarity index 64% rename from semantica/explorer/utils/skos.py rename to semantica/utils/skos.py index 35e6d8f9..cfa48b0f 100644 --- a/semantica/explorer/utils/skos.py +++ b/semantica/utils/skos.py @@ -7,20 +7,27 @@ from typing import Iterable, Mapping _HIERARCHY_EDGE_TYPES = frozenset({"skos:broader", "skos:narrower"}) -def validate_skos_hierarchy( - edges: Iterable[Mapping[str, object]], -) -> None: +def validate_skos_hierarchy(edges: Iterable[Mapping[str, object]]) -> None: """Raise ``ValueError`` when SKOS hierarchy edges contain a cycle. ``skos:broader`` points from a concept to its parent while ``skos:narrower`` expresses the same relationship in the opposite - direction. Both forms are normalized to a child-to-parent adjacency - map before cycle detection. + direction. Both forms are normalized to child-to-parent adjacency before + cycle detection. """ parents: dict[str, set[str]] = defaultdict(set) for edge in edges: - edge_type = edge.get("type") + if not isinstance(edge, Mapping): + continue + edge_type = next( + ( + edge.get(key) + for key in ("type", "edge_type", "relationship", "predicate", "relation") + if edge.get(key) is not None + ), + None, + ) if edge_type not in _HIERARCHY_EDGE_TYPES: continue @@ -29,11 +36,7 @@ def validate_skos_hierarchy( if not source or not target: continue - child, parent = ( - (source, target) - if edge_type == "skos:broader" - else (target, source) - ) + child, parent = ((source, target) if edge_type == "skos:broader" else (target, source)) parents[child].add(parent) visiting: set[str] = set() @@ -41,9 +44,7 @@ def validate_skos_hierarchy( def visit(concept: str) -> None: if concept in visiting: - raise ValueError( - f"SKOS hierarchy contains a cycle involving '{concept}'." - ) + raise ValueError(f"SKOS hierarchy contains a cycle involving '{concept}'.") if concept in visited: return diff --git a/tests/context/test_context.py b/tests/context/test_context.py index 54f705f1..15ec1b49 100644 --- a/tests/context/test_context.py +++ b/tests/context/test_context.py @@ -203,6 +203,26 @@ class TestContextModule(unittest.TestCase): ) self.assertIsNotNone(retriever) + def test_context_graph_rejects_cyclic_skos_single_edge_write(self): + graph = ContextGraph() + graph.add_edge("A", "B", "skos:broader") + + with self.assertRaisesRegex(ValueError, "SKOS hierarchy contains a cycle"): + graph.add_edge("B", "A", "skos:broader") + + self.assertEqual(len(graph.edges), 1) + + def test_context_graph_rejects_cyclic_skos_batch_write(self): + graph = ContextGraph() + + with self.assertRaisesRegex(ValueError, "SKOS hierarchy contains a cycle"): + graph.add_edges([ + {"source": "A", "target": "B", "type": "skos:broader"}, + {"source": "A", "target": "B", "type": "skos:narrower"}, + ]) + + self.assertEqual(len(graph.edges), 0) + # --- AgentContext Tests --- @patch('semantica.context.agent_memory.AgentMemory._generate_embedding') def test_agent_context_end_to_end(self, mock_gen_embedding): diff --git a/tests/explorer/test_vocabulary.py b/tests/explorer/test_vocabulary.py index b148ac3c..ee75ca7a 100644 --- a/tests/explorer/test_vocabulary.py +++ b/tests/explorer/test_vocabulary.py @@ -7,7 +7,7 @@ from fastapi.testclient import TestClient from semantica.explorer.dependencies import get_session from semantica.explorer.routes.vocabulary import router -from semantica.explorer.utils.skos import validate_skos_hierarchy +from semantica.utils.skos import validate_skos_hierarchy app = FastAPI() app.include_router(router)