Centralize SKOS cycle validation

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 |

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,

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:

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
 

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):

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)