From 495e29d5438d4a35bb793debcc2e64bec465272d Mon Sep 17 00:00:00 2001 From: Sameer6305 Date: Fri, 31 Jul 2026 15:00:23 +0530 Subject: [PATCH 1/7] fix(provenance): log tracking failures and return None on storage error (closes #783) --- semantica/conflicts/source_tracker.py | 33 +++++++++---- semantica/provenance/manager.py | 42 +++++++++++------ tests/conflicts/test_conflicts.py | 24 ++++++++++ tests/provenance/test_manager.py | 67 +++++++++++++++++++++------ 4 files changed, 130 insertions(+), 36 deletions(-) diff --git a/semantica/conflicts/source_tracker.py b/semantica/conflicts/source_tracker.py index 3a00567f..d2035c66 100644 --- a/semantica/conflicts/source_tracker.py +++ b/semantica/conflicts/source_tracker.py @@ -614,29 +614,44 @@ class SourceTracker: if item_type == "entity": entity_id = item.get("entity_id") if entity_id: - self.track_entity_source(entity_id, source_ref, **metadata) - stats["entities_tracked"] += 1 - stats["total_tracked"] += 1 + entry = self.track_entity_source(entity_id, source_ref, **metadata) + if entry is not None: + stats["entities_tracked"] += 1 + stats["total_tracked"] += 1 + else: + self.logger.warning( + f"Failed to track entity source for '{entity_id}' in item {i}" + ) elif item_type == "property": entity_id = item.get("entity_id") property_name = item.get("property_name") value = item.get("value") if entity_id and property_name is not None: - self.track_property_source( + entry = self.track_property_source( entity_id, property_name, value, source_ref, **metadata ) - stats["properties_tracked"] += 1 - stats["total_tracked"] += 1 + if entry is not None: + stats["properties_tracked"] += 1 + stats["total_tracked"] += 1 + else: + self.logger.warning( + f"Failed to track property source for '{entity_id}.{property_name}' in item {i}" + ) elif item_type == "relationship": relationship_id = item.get("relationship_id") if relationship_id: - self.track_relationship_source( + entry = self.track_relationship_source( relationship_id, source_ref, **metadata ) - stats["relationships_tracked"] += 1 - stats["total_tracked"] += 1 + if entry is not None: + stats["relationships_tracked"] += 1 + stats["total_tracked"] += 1 + else: + self.logger.warning( + f"Failed to track relationship source for '{relationship_id}' in item {i}" + ) else: self.logger.warning( diff --git a/semantica/provenance/manager.py b/semantica/provenance/manager.py index 2eafb937..b2e9874c 100644 --- a/semantica/provenance/manager.py +++ b/semantica/provenance/manager.py @@ -144,8 +144,8 @@ class ProvenanceManager: entry: ProvenanceEntry, _conn: Optional[Any] = None, _raise_on_error: bool = False, - ) -> ProvenanceEntry: - """Compute checksum, store entry persistently, and gracefully ignore storage errors.""" + ) -> Optional[ProvenanceEntry]: + """Compute checksum, store entry persistently, and log/handle storage errors (#783).""" entry.checksum = compute_checksum(entry) try: @@ -153,13 +153,19 @@ class ProvenanceManager: self.storage._store_with_conn(_conn, entry) else: self.storage.store(entry) - except Exception: + except Exception as e: + self.logger.error( + "Failed to save provenance entry for entity '%s': %s", + entry.entity_id, + e, + exc_info=True, + ) # Propagate when called from a batch's shared transaction so the # caller's per-item try/except can skip counting this item instead # of reporting an unpersisted entry as tracked (#807). if _raise_on_error: raise - pass # Graceful failure - don't break main functionality + return None # Graceful failure - don't return unpersisted entry (#783) return entry @@ -297,7 +303,7 @@ class ProvenanceManager: source: str, metadata: Optional[Dict[str, Any]] = None, **kwargs - ) -> ProvenanceEntry: + ) -> Optional[ProvenanceEntry]: """ Track relationship provenance (kg.ProvenanceTracker compatible). @@ -308,7 +314,7 @@ class ProvenanceManager: **kwargs: Additional fields Returns: - ProvenanceEntry object + Optional[ProvenanceEntry]: ProvenanceEntry on success, or None if storage fails Example: >>> prov_mgr.track_relationship( @@ -343,7 +349,7 @@ class ProvenanceManager: parent_chunk_id: Optional[str] = None, _conn: Optional[Any] = None, **metadata - ) -> ProvenanceEntry: + ) -> Optional[ProvenanceEntry]: """ Track chunk provenance (split.ProvenanceTracker compatible). @@ -357,7 +363,7 @@ class ProvenanceManager: **metadata: Additional metadata Returns: - ProvenanceEntry object + Optional[ProvenanceEntry]: ProvenanceEntry on success, or None if storage fails Example: >>> prov_mgr.track_chunk( @@ -392,7 +398,7 @@ class ProvenanceManager: value: Any, source: SourceReference, **metadata - ) -> ProvenanceEntry: + ) -> Optional[ProvenanceEntry]: """ Track property source (conflicts.SourceTracker compatible). @@ -404,7 +410,7 @@ class ProvenanceManager: **metadata: Additional metadata Returns: - ProvenanceEntry object + Optional[ProvenanceEntry]: ProvenanceEntry on success, or None if storage fails Example: >>> source = SourceReference( @@ -489,8 +495,12 @@ class ProvenanceManager: # Add batch_count to tracked_count only after the transaction context # exits successfully and commits (#807). tracked_count += batch_count - except Exception: - pass # Partial failure: if block-level storage transaction fails, continue to next block + except Exception as e: + self.logger.error( + "Block-level storage transaction failed in track_entities_batch: %s", + e, + exc_info=True, + ) return tracked_count @@ -544,8 +554,12 @@ class ProvenanceManager: # Add batch_count to tracked_count only after the transaction context # exits successfully and commits (#807). tracked_count += batch_count - except Exception: - pass + except Exception as e: + self.logger.error( + "Block-level storage transaction failed in track_chunks_batch: %s", + e, + exc_info=True, + ) return tracked_count diff --git a/tests/conflicts/test_conflicts.py b/tests/conflicts/test_conflicts.py index bee5c828..8018cb4c 100644 --- a/tests/conflicts/test_conflicts.py +++ b/tests/conflicts/test_conflicts.py @@ -84,6 +84,30 @@ class TestConflictsModule(unittest.TestCase): sources = tracker.get_entity_sources("e1") self.assertTrue(len(sources) >= 1) + def test_track_sources_batch_failure_not_counted(self): + """Test that track_sources_batch does not increment stats when tracking fails (#783).""" + tracker = SourceTracker() + source_data = [ + { + "type": "property", + "entity_id": "e1", + "property_name": "age", + "value": 30, + "source": self.source1, + }, + { + "type": "property", + "entity_id": "e2", + "property_name": "age", + "value": 25, + "source": self.source2, + }, + ] + with patch.object(tracker, "track_property_source", return_value=None): + stats = tracker.track_sources_batch(source_data) + self.assertEqual(stats["properties_tracked"], 0) + self.assertEqual(stats["total_tracked"], 0) + def test_conflict_detector(self): detector = ConflictDetector() diff --git a/tests/provenance/test_manager.py b/tests/provenance/test_manager.py index 87eab374..77f61b79 100644 --- a/tests/provenance/test_manager.py +++ b/tests/provenance/test_manager.py @@ -7,7 +7,8 @@ chunk tracking, source tracking, and lineage tracing. import pytest from unittest.mock import patch -from semantica.provenance import ProvenanceManager, SourceReference +from datetime import datetime +from semantica.provenance import ProvenanceManager, SourceReference, ProvenanceEntry from semantica.provenance.storage import InMemoryStorage, SQLiteStorage @@ -611,16 +612,18 @@ class TestProvenanceManager: assert entry is None def test_track_relationship_storage_error_swallowed(self): - """Test that track_relationship swallows storage.store() exceptions and still returns a ProvenanceEntry.""" + """Test that track_relationship returns None when storage.store() fails, + since nothing was persisted (#783).""" prov_mgr = ProvenanceManager() with patch.object(prov_mgr.storage, "store", side_effect=RuntimeError("storage error")): entry = prov_mgr.track_relationship("r_test", source="doc_1") - assert entry is not None - assert entry.entity_id == "r_test" - assert entry.checksum is not None + # Returning None is correct because nothing was actually persisted, + # and the old assertion was encoding the bug this issue was filed to fix. + assert entry is None def test_track_chunk_storage_error_swallowed(self): - """Test that track_chunk swallows storage.store() exceptions and still returns a ProvenanceEntry.""" + """Test that track_chunk returns None when storage.store() fails, + since nothing was persisted (#783).""" prov_mgr = ProvenanceManager() with patch.object(prov_mgr.storage, "store", side_effect=RuntimeError("storage error")): entry = prov_mgr.track_chunk( @@ -630,12 +633,13 @@ class TestProvenanceManager: start_index=0, end_index=100, ) - assert entry is not None - assert entry.entity_id == "c_test" - assert entry.checksum is not None + # Returning None is correct because nothing was actually persisted, + # and the old assertion was encoding the bug this issue was filed to fix. + assert entry is None def test_track_property_source_storage_error_swallowed(self): - """Test that track_property_source swallows storage.store() exceptions and still returns a ProvenanceEntry.""" + """Test that track_property_source returns None when storage.store() fails, + since nothing was persisted (#783).""" prov_mgr = ProvenanceManager() source = SourceReference(document="doc_1", page=1, confidence=0.9) with patch.object(prov_mgr.storage, "store", side_effect=RuntimeError("storage error")): @@ -645,9 +649,46 @@ class TestProvenanceManager: value="val", source=source, ) - assert entry is not None - assert entry.entity_id == "e_test_prop_test" - assert entry.checksum is not None + # Returning None is correct because nothing was actually persisted, + # and the old assertion was encoding the bug this issue was filed to fix. + assert entry is None + + def test_save_entry_logs_on_every_failure_path(self): + """Test that _save_entry logs on storage failures for both raising and swallowing paths (#783).""" + prov_mgr = ProvenanceManager() + entry = ProvenanceEntry( + entity_id="log_test_id", + entity_type="entity", + activity_id="test", + source_document="doc_1", + first_seen=datetime.utcnow().isoformat(), + last_updated=datetime.utcnow().isoformat(), + ) + with patch.object(prov_mgr.storage, "store", side_effect=RuntimeError("storage error")), \ + patch.object(prov_mgr.logger, "error") as mock_log_error: + # Swallowing branch (_raise_on_error=False) + res = prov_mgr._save_entry(entry, _raise_on_error=False) + assert res is None + mock_log_error.assert_called_once() + assert "log_test_id" in mock_log_error.call_args[0][1] + + mock_log_error.reset_mock() + + # Raising branch (_raise_on_error=True) + with pytest.raises(RuntimeError, match="storage error"): + prov_mgr._save_entry(entry, _raise_on_error=True) + mock_log_error.assert_called_once() + assert "log_test_id" in mock_log_error.call_args[0][1] + + def test_track_entities_batch_logs_per_item_failure(self): + """Test that track_entities_batch emits item-level logs via _save_entry when items fail (#783).""" + prov_mgr = ProvenanceManager() + entities = [{"id": "e_fail_1"}, {"id": "e_fail_2"}] + with patch.object(prov_mgr.storage, "_store_with_conn", side_effect=RuntimeError("batch store error")), \ + patch.object(prov_mgr.logger, "error") as mock_log_error: + count = prov_mgr.track_entities_batch(entities, "doc_1") + assert count == 0 + assert mock_log_error.call_count == 2 def test_track_entity_pre_build_failure_fallback_skips_store(self): """Test that when track_entity fails before the entry is built (e.g. a From 1ae1e6d57a8c02e5098beed293abd662fcdf3df8 Mon Sep 17 00:00:00 2001 From: Sameer6305 Date: Fri, 31 Jul 2026 15:01:33 +0530 Subject: [PATCH 2/7] docs(provenance): document Optional return types and failure behavior (#783) --- CHANGELOG.md | 6 ++++++ docs/reference/provenance.md | 8 ++++---- 2 files changed, 10 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 78138cc4..3d9e94ea 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -43,6 +43,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- **`ProvenanceManager` tracking methods silently swallowed failures without logging and returned fabricated entries** (#783) + - `track_relationship()`, `track_chunk()`, and `track_property_source()` now return `Optional[ProvenanceEntry]` (`None` on storage failure, consistent with #782's `track_entity` fix) instead of a fabricated populated object + - `_save_entry()` now always logs on any storage failure, including previously-silent per-item batch failures + - `track_entities_batch()` and `track_chunks_batch()`'s rare block-level transaction failures are now logged too + - `source_tracker.py`'s `track_sources_batch()` no longer counts failed tracking calls in its stats + - **`ProvenanceManager.track_entity` persisted partial history and returned fabricated entries on storage failure** (#782, #816) by @Sameer6305 and @KaifAhmad1 - `track_entity()`'s two-step write (history archive + primary update) is now atomic — if either write fails, the whole operation rolls back via the existing #807 `transaction()` mechanism, instead of silently persisting a partial state - `track_entity()`'s return type is now `Optional[ProvenanceEntry]`: on failure it returns a safe deep copy of the pre-failure existing entry (if one existed) or `None` (if this was a brand-new, never-successfully-tracked entity) — never a fabricated object claiming values that were never actually persisted diff --git a/docs/reference/provenance.md b/docs/reference/provenance.md index 5c1cf3bb..6a7e4ced 100644 --- a/docs/reference/provenance.md +++ b/docs/reference/provenance.md @@ -222,9 +222,9 @@ cleared = manager.clear() | Method | Returns | Description | | :------ | :------- | :----------- | | `track_entity(entity_id, source, metadata, **kwargs)` | `Optional[ProvenanceEntry]` | Record entity provenance atomically; returns `ProvenanceEntry` on success, or `None`/existing entry on storage failure | -| `track_relationship(relationship_id, source, metadata, **kwargs)` | `ProvenanceEntry` | Record relationship provenance | -| `track_chunk(chunk_id, source_document, ...)` | `ProvenanceEntry` | Record chunk provenance with char offsets | -| `track_property_source(entity_id, property_name, value, source)` | `ProvenanceEntry` | Record property-level source attribution | +| `track_relationship(relationship_id, source, metadata, **kwargs)` | `Optional[ProvenanceEntry]` | Record relationship provenance; returns `ProvenanceEntry` on success, or `None` on storage failure | +| `track_chunk(chunk_id, source_document, ...)` | `Optional[ProvenanceEntry]` | Record chunk provenance with char offsets; returns `ProvenanceEntry` on success, or `None` on storage failure | +| `track_property_source(entity_id, property_name, value, source)` | `Optional[ProvenanceEntry]` | Record property-level source attribution; returns `ProvenanceEntry` on success, or `None` on storage failure | | `track_entities_batch(entities, source)` | `int` | Batch-track entities; returns success count | | `track_chunks_batch(chunks, source_document)` | `int` | Batch-track chunks; returns success count | | `get_lineage(entity_id)` | `Dict[str, Any]` | Full lineage as aggregated dict | @@ -236,7 +236,7 @@ cleared = manager.clear() ## ProvenanceEntry Fields -`ProvenanceEntry` is the core dataclass. Every tracking method returns one: +`ProvenanceEntry` is the core dataclass. Every tracking method returns one on success (or `None` on storage failure): ```python from semantica.provenance import ProvenanceEntry From 4dea295f0d0ba2e049347ebbf54e0bacde5ba2ea Mon Sep 17 00:00:00 2001 From: Sameer6305 Date: Fri, 31 Jul 2026 16:05:37 +0530 Subject: [PATCH 3/7] fixed qodo reviews - split.ProvenanceTracker: Check _unified_manager.track_chunk() return value and fall back to legacy storage when None is returned (storage failure) - split.ProvenanceTracker: Initialize legacy stores (_provenance_store and _chunk_registry) unconditionally in __init__ so legacy fallback storage works safely even when initialized with use_unified=True - split.ProvenanceTracker: Update get_provenance() to check legacy storage when no record is found in unified storage, ensuring fallback-tracked chunks remain retrievable - SourceTracker: Change track_sources_batch() condition from if entry is not None: to if ok: to respect the boolean return contract (-> bool) of track_entity_source(), track_property_source(), and track_relationship_source() - Tests: Add regression test test_unified_tracking_returns_none_triggers_fallback and update test_track_sources_batch_failure_not_counted to test boolean failure return_value=False --- semantica/conflicts/source_tracker.py | 12 ++++++------ semantica/split/provenance_tracker.py | 12 ++++++++---- tests/conflicts/test_conflicts.py | 2 +- tests/provenance/test_backward_compat.py | 16 ++++++++++++++++ 4 files changed, 31 insertions(+), 11 deletions(-) diff --git a/semantica/conflicts/source_tracker.py b/semantica/conflicts/source_tracker.py index d2035c66..9ae19aa9 100644 --- a/semantica/conflicts/source_tracker.py +++ b/semantica/conflicts/source_tracker.py @@ -614,8 +614,8 @@ class SourceTracker: if item_type == "entity": entity_id = item.get("entity_id") if entity_id: - entry = self.track_entity_source(entity_id, source_ref, **metadata) - if entry is not None: + ok = self.track_entity_source(entity_id, source_ref, **metadata) + if ok: stats["entities_tracked"] += 1 stats["total_tracked"] += 1 else: @@ -628,10 +628,10 @@ class SourceTracker: property_name = item.get("property_name") value = item.get("value") if entity_id and property_name is not None: - entry = self.track_property_source( + ok = self.track_property_source( entity_id, property_name, value, source_ref, **metadata ) - if entry is not None: + if ok: stats["properties_tracked"] += 1 stats["total_tracked"] += 1 else: @@ -642,10 +642,10 @@ class SourceTracker: elif item_type == "relationship": relationship_id = item.get("relationship_id") if relationship_id: - entry = self.track_relationship_source( + ok = self.track_relationship_source( relationship_id, source_ref, **metadata ) - if entry is not None: + if ok: stats["relationships_tracked"] += 1 stats["total_tracked"] += 1 else: diff --git a/semantica/split/provenance_tracker.py b/semantica/split/provenance_tracker.py index e5d0edb4..e4cd2096 100644 --- a/semantica/split/provenance_tracker.py +++ b/semantica/split/provenance_tracker.py @@ -97,6 +97,9 @@ class ProvenanceTracker: self.store_metadata = config.get("store_metadata", True) self.track_versions = config.get("track_versions", False) + self._provenance_store: Dict[str, ProvenanceInfo] = {} + self._chunk_registry: Dict[str, str] = {} # chunk_id -> provenance_id + # Determine whether to use unified backend use_unified = config.get("use_unified", True) and UNIFIED_AVAILABLE @@ -110,8 +113,6 @@ class ProvenanceTracker: # Fallback to legacy in-memory storage self._unified_manager = None self._use_unified = False - self._provenance_store: Dict[str, ProvenanceInfo] = {} - self._chunk_registry: Dict[str, str] = {} # chunk_id -> provenance_id self.logger.debug("Chunk provenance tracker initialized with legacy backend") def track_chunk( @@ -147,7 +148,7 @@ class ProvenanceTracker: # Delegate to unified manager try: chunk_metadata = {**chunk.metadata, **metadata, "chunk_size": len(chunk.text)} if self.store_metadata else metadata - self._unified_manager.track_chunk( + entry = self._unified_manager.track_chunk( chunk_id=chunk_id, source_document=source_document, source_path=source_path, @@ -156,6 +157,9 @@ class ProvenanceTracker: parent_chunk_id=parent_chunk_id, **chunk_metadata ) + if entry is None: + self.logger.warning("Unified tracking failed (returned None), using fallback") + return self._track_chunk_legacy(chunk, source_document, source_path, parent_chunk_id, **metadata) return chunk_id # Return chunk_id for compatibility except Exception as e: self.logger.warning(f"Unified tracking failed, using fallback: {e}") @@ -303,7 +307,7 @@ class ProvenanceTracker: version=prov.get("version", "1.0"), timestamp=prov.get("timestamp") ) - return None + return self._get_provenance_legacy(chunk_id) except Exception as e: self.logger.warning(f"Unified retrieval failed, using fallback: {e}") return self._get_provenance_legacy(chunk_id) diff --git a/tests/conflicts/test_conflicts.py b/tests/conflicts/test_conflicts.py index 8018cb4c..34dad4d2 100644 --- a/tests/conflicts/test_conflicts.py +++ b/tests/conflicts/test_conflicts.py @@ -103,7 +103,7 @@ class TestConflictsModule(unittest.TestCase): "source": self.source2, }, ] - with patch.object(tracker, "track_property_source", return_value=None): + with patch.object(tracker, "track_property_source", return_value=False): stats = tracker.track_sources_batch(source_data) self.assertEqual(stats["properties_tracked"], 0) self.assertEqual(stats["total_tracked"], 0) diff --git a/tests/provenance/test_backward_compat.py b/tests/provenance/test_backward_compat.py index 16268aeb..cb785e5b 100644 --- a/tests/provenance/test_backward_compat.py +++ b/tests/provenance/test_backward_compat.py @@ -141,6 +141,22 @@ class TestSplitProvenanceBackwardCompat: assert len(prov_ids) == 3 + def test_unified_tracking_returns_none_triggers_fallback(self): + """Test that when _unified_manager.track_chunk returns None, fallback provenance is stored (#783).""" + from unittest.mock import patch + tracker = SplitProvenanceTracker() + chunk = Chunk(text="Test chunk for none fallback", start_index=0, end_index=10, metadata={}) + + with patch.object(tracker._unified_manager, "track_chunk", return_value=None): + prov_id = tracker.track_chunk(chunk, source_document="doc_1") + assert prov_id is not None + # Confirm it fell back to legacy store and is retrievable + assert chunk.id in tracker._chunk_registry + assert prov_id in tracker._provenance_store + prov_info = tracker.get_provenance(chunk.id) + assert prov_info is not None + assert prov_info.source_document == "doc_1" + class TestNoProvenanceOverhead: """Test that provenance has zero overhead when not used.""" From 692260cc768f384b4c186aa4774b5275305896dd Mon Sep 17 00:00:00 2001 From: mikemikimike <13286568797@163.com> Date: Fri, 31 Jul 2026 17:08:03 +0800 Subject: [PATCH 4/7] Reject cyclic SKOS hierarchies --- semantica/explorer/routes/vocabulary.py | 7 ++- semantica/explorer/session.py | 7 +++ semantica/explorer/utils/skos.py | 57 +++++++++++++++++++++++++ tests/explorer/test_vocabulary.py | 23 ++++++++++ 4 files changed, 93 insertions(+), 1 deletion(-) create mode 100644 semantica/explorer/utils/skos.py diff --git a/semantica/explorer/routes/vocabulary.py b/semantica/explorer/routes/vocabulary.py index 4e992229..7ba6bbc4 100644 --- a/semantica/explorer/routes/vocabulary.py +++ b/semantica/explorer/routes/vocabulary.py @@ -6,7 +6,7 @@ import asyncio from collections import defaultdict from typing import Dict, List, Optional -from fastapi import APIRouter, Depends, File, Form, HTTPException, Query, Request, UploadFile +from fastapi import APIRouter, Depends, File, Form, HTTPException, Query, UploadFile from ..dependencies import get_session from ..schemas import ConceptNode, ConceptSummary, VocabularyImportResponse, VocabularyScheme @@ -224,6 +224,11 @@ async def import_vocabulary( except ValueError as exc: raise HTTPException(status_code=422, detail=str(exc)) from exc + try: + await asyncio.to_thread(session.validate_skos_hierarchy, edges) + except ValueError as exc: + raise HTTPException(status_code=422, detail=str(exc)) from exc + nodes_added = await asyncio.to_thread(session.add_nodes, nodes) edges_added = await asyncio.to_thread(session.add_edges, edges) diff --git a/semantica/explorer/session.py b/semantica/explorer/session.py index 8692f59e..19c3d885 100644 --- a/semantica/explorer/session.py +++ b/semantica/explorer/session.py @@ -13,6 +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 _KG_AVAILABLE = False try: @@ -736,6 +737,7 @@ class GraphSession: def add_edges(self, edges: List[Dict[str, Any]]) -> int: with self._lock: + self.validate_skos_hierarchy(edges) added = self.graph.add_edges(edges) has_mutation_callback = callable(getattr(self.graph, "mutation_callback", None)) if added and not has_mutation_callback: @@ -744,6 +746,11 @@ class GraphSession: self.rebuild_search_index() return added + def validate_skos_hierarchy(self, edges: List[Dict[str, Any]]) -> None: + """Validate new SKOS hierarchy edges against the current graph.""" + existing_edges = self.graph.find_edges() + validate_skos_hierarchy([*existing_edges, *edges]) + def add_node( self, node_id: str, diff --git a/semantica/explorer/utils/skos.py b/semantica/explorer/utils/skos.py new file mode 100644 index 00000000..35e6d8f9 --- /dev/null +++ b/semantica/explorer/utils/skos.py @@ -0,0 +1,57 @@ +"""Validation helpers for SKOS graph relationships.""" + +from collections import defaultdict +from typing import Iterable, Mapping + + +_HIERARCHY_EDGE_TYPES = frozenset({"skos:broader", "skos:narrower"}) + + +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. + """ + + parents: dict[str, set[str]] = defaultdict(set) + for edge in edges: + edge_type = edge.get("type") + if edge_type not in _HIERARCHY_EDGE_TYPES: + continue + + source = str(edge.get("source", edge.get("source_id", ""))) + target = str(edge.get("target", edge.get("target_id", ""))) + if not source or not target: + continue + + child, parent = ( + (source, target) + if edge_type == "skos:broader" + else (target, source) + ) + parents[child].add(parent) + + visiting: set[str] = set() + visited: set[str] = set() + + def visit(concept: str) -> None: + if concept in visiting: + raise ValueError( + f"SKOS hierarchy contains a cycle involving '{concept}'." + ) + if concept in visited: + return + + visiting.add(concept) + for parent in parents.get(concept, ()): + visit(parent) + visiting.remove(concept) + visited.add(concept) + + for concept in parents: + visit(concept) diff --git a/tests/explorer/test_vocabulary.py b/tests/explorer/test_vocabulary.py index 7b91d17f..b148ac3c 100644 --- a/tests/explorer/test_vocabulary.py +++ b/tests/explorer/test_vocabulary.py @@ -7,6 +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 app = FastAPI() app.include_router(router) @@ -34,6 +35,15 @@ MINIMAL_RDF_XML = b""" """ +CYCLIC_TTL = b""" +@prefix skos: . +@prefix ex: . +ex:S a skos:ConceptScheme ; skos:prefLabel \"Scheme\" . +ex:A a skos:Concept ; skos:prefLabel \"Alpha\" ; skos:inScheme ex:S ; skos:broader ex:B . +ex:B a skos:Concept ; skos:prefLabel \"Beta\" ; skos:inScheme ex:S ; skos:broader ex:A . +""" + + def setup_function(): mock_session.reset_mock() @@ -158,3 +168,16 @@ def test_import_invalid_file_returns_422(): files={"file": ("bad.ttl", b"this is not valid RDF!", "text/turtle")}, ) assert response.status_code == 422 + + +def test_import_rejects_cyclic_hierarchy_before_writing_nodes(): + mock_session.validate_skos_hierarchy.side_effect = lambda edges: validate_skos_hierarchy(edges) + + response = client.post( + "/api/vocabulary/import", + files={"file": ("cyclic.ttl", CYCLIC_TTL, "text/turtle")}, + ) + + assert response.status_code == 422 + assert "cycle" in response.json()["detail"].lower() + mock_session.add_nodes.assert_not_called() 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 5/7] 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) From f992504227f1b8a7120a8989f47d9b91a5836854 Mon Sep 17 00:00:00 2001 From: mikemikimike <13286568797@163.com> Date: Fri, 31 Jul 2026 17:42:21 +0800 Subject: [PATCH 6/7] Make SKOS hierarchy imports atomic --- CHANGELOG.md | 5 ++++ semantica/context/context_graph.py | 19 +++++++----- semantica/explorer/routes/export_import.py | 12 +++++--- semantica/explorer/routes/ontology.py | 27 ++++++++++++----- semantica/explorer/routes/vocabulary.py | 7 ++--- semantica/explorer/session.py | 34 ++++++++++++++++++++-- semantica/utils/skos.py | 22 +++++++++++--- tests/explorer/test_ontology_subissue3.py | 20 ++++++++++++- tests/explorer/test_vocabulary.py | 17 ++++++----- 9 files changed, 125 insertions(+), 38 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8723c746..bc00cef6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -43,6 +43,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- **No cycle detection for SKOS concepts at write time** (#774, #819) by @mikemikimike, reviewed by @Sameer6305 and @KaifAhmad1 + - Added cycle detection (`validate_skos_hierarchy`) for `skos:broader` and `skos:narrower` relationships in `ContextGraph.add_edge()` and `ContextGraph.add_edges()`, preventing direct 2-node cycles, self-loops, and multi-hop hierarchy cycles + - Added `GraphSession.add_nodes_and_edges()` to validate SKOS hierarchy edges upfront under lock before node insertion, preventing partial-write leaks where nodes remain after a cyclic edge is rejected + - Updated vocabulary, ontology (`/api/ontology/load`, `/api/ontology/create`), and JSON/CSV import routes to use `add_nodes_and_edges()` and return HTTP 422 with actionable error messages when a cycle is detected + - **`AgnoDecisionKit`/`AgnoKGToolkit` silently swallowed Agno tool registration failures** (#780, #818) by @Sameer6305 and @KaifAhmad1 - Removed the `try/except: pass` wrapped around `self.register(fn)` in both toolkits' `__init__`; when Agno is installed, a registration failure now propagates immediately instead of leaving the toolkit half-registered with no signal to the caller - Graceful degradation when Agno isn't installed (`AGNO_AVAILABLE=False`) is unchanged — `_tools` is still populated so callers can introspect available tools without the package diff --git a/semantica/context/context_graph.py b/semantica/context/context_graph.py index 802295a1..f5ef322e 100644 --- a/semantica/context/context_graph.py +++ b/semantica/context/context_graph.py @@ -117,7 +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 ..utils.skos import is_skos_hierarchy_edge, validate_skos_hierarchy from .entity_linker import EntityLinker # Optional imports for advanced features @@ -592,7 +592,12 @@ class ContextGraph: 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]) + hierarchy_edges = [edge for edge in edges if is_skos_hierarchy_edge(edge)] + if hierarchy_edges: + existing_edges = [ + edge for edge in self.find_edges() if is_skos_hierarchy_edge(edge) + ] + validate_skos_hierarchy([*existing_edges, *hierarchy_edges]) for raw_edge in edges: if not isinstance(raw_edge, dict): continue @@ -952,12 +957,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}, + candidate = {"source": source_id, "target": target_id, "type": edge_type} + if is_skos_hierarchy_edge(candidate): + existing_edges = [ + edge for edge in self.find_edges() if is_skos_hierarchy_edge(edge) ] - ) + validate_skos_hierarchy([*existing_edges, candidate]) return self._add_internal_edge( ContextEdge( edge_id=edge_id, diff --git a/semantica/explorer/routes/export_import.py b/semantica/explorer/routes/export_import.py index 3529590a..6beda930 100644 --- a/semantica/explorer/routes/export_import.py +++ b/semantica/explorer/routes/export_import.py @@ -112,8 +112,10 @@ async def import_file( } ) - nodes_added = session.add_nodes(nodes) - edges_added = session.add_edges(edges) + try: + nodes_added, edges_added = session.add_nodes_and_edges(nodes, edges) + except ValueError as exc: + raise HTTPException(status_code=422, detail=str(exc)) from exc return _import_response(nodes_added, edges_added) if filename.endswith(".csv"): @@ -184,8 +186,10 @@ async def import_file( detail="No valid nodes or edges could be parsed from the CSV payload.", ) - nodes_added = session.add_nodes(nodes) - edges_added = session.add_edges(edges) + try: + nodes_added, edges_added = session.add_nodes_and_edges(nodes, edges) + except ValueError as exc: + raise HTTPException(status_code=422, detail=str(exc)) from exc return _import_response(nodes_added, edges_added) raise HTTPException( diff --git a/semantica/explorer/routes/ontology.py b/semantica/explorer/routes/ontology.py index 31c21afa..2fac4c15 100644 --- a/semantica/explorer/routes/ontology.py +++ b/semantica/explorer/routes/ontology.py @@ -1289,6 +1289,8 @@ async def load_ontology( temp_path, format=fmt ) + if not ontology_data.data.get("classes") and not ontology_data.data.get("properties"): + raise ValueError("No OWL classes or properties found by OntologyIngestor") # Convert to graph nodes/edges using ontology data nodes, edges = await asyncio.to_thread( @@ -1296,9 +1298,12 @@ async def load_ontology( ontology_data.data ) - # Add nodes and edges to session - nodes_added = await asyncio.to_thread(session.add_nodes, nodes) - edges_added = await asyncio.to_thread(session.add_edges, edges) + try: + nodes_added, edges_added = await asyncio.to_thread( + session.add_nodes_and_edges, nodes, edges + ) + except ValueError as exc: + raise HTTPException(status_code=422, detail=str(exc)) from exc # Register in registry registry = _get_registry(request) @@ -1347,8 +1352,12 @@ async def load_ontology( raise HTTPException(status_code=422, detail=f"Could not parse ontology: {exc}") from exc # Fallback path - use basic parsing - nodes_added = await asyncio.to_thread(session.add_nodes, nodes) - edges_added = await asyncio.to_thread(session.add_edges, edges) + try: + nodes_added, edges_added = await asyncio.to_thread( + session.add_nodes_and_edges, nodes, edges + ) + except ValueError as exc: + raise HTTPException(status_code=422, detail=str(exc)) from exc registry = _get_registry(request) ontology_uri = metadata.get("uri", f"temp:{uuid.uuid4().hex[:12]}") @@ -1545,8 +1554,12 @@ async def create_ontology( logger.exception("Failed to generate ontology from schema text; aborting ontology creation.") raise HTTPException(status_code=500, detail=f"Ontology generation failed: {exc}") from exc - nodes_added = await asyncio.to_thread(session.add_nodes, nodes) - edges_added = await asyncio.to_thread(session.add_edges, edges) + try: + nodes_added, edges_added = await asyncio.to_thread( + session.add_nodes_and_edges, nodes, edges + ) + except ValueError as exc: + raise HTTPException(status_code=422, detail=str(exc)) from exc registry = _get_registry(request) registry[onto_uri] = OntologyEntry( diff --git a/semantica/explorer/routes/vocabulary.py b/semantica/explorer/routes/vocabulary.py index 7ba6bbc4..17739b67 100644 --- a/semantica/explorer/routes/vocabulary.py +++ b/semantica/explorer/routes/vocabulary.py @@ -225,13 +225,12 @@ async def import_vocabulary( raise HTTPException(status_code=422, detail=str(exc)) from exc try: - await asyncio.to_thread(session.validate_skos_hierarchy, edges) + nodes_added, edges_added = await asyncio.to_thread( + session.add_nodes_and_edges, nodes, edges + ) except ValueError as exc: raise HTTPException(status_code=422, detail=str(exc)) from exc - nodes_added = await asyncio.to_thread(session.add_nodes, nodes) - edges_added = await asyncio.to_thread(session.add_edges, edges) - return VocabularyImportResponse( status="success", filename=filename, diff --git a/semantica/explorer/session.py b/semantica/explorer/session.py index 9eafdb8c..369d326f 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 is_skos_hierarchy_edge, validate_skos_hierarchy _KG_AVAILABLE = False try: @@ -748,8 +748,35 @@ class GraphSession: def validate_skos_hierarchy(self, edges: List[Dict[str, Any]]) -> None: """Validate new SKOS hierarchy edges against the current graph.""" - existing_edges = self.graph.find_edges() - validate_skos_hierarchy([*existing_edges, *edges]) + hierarchy_edges = [edge for edge in edges if is_skos_hierarchy_edge(edge)] + if not hierarchy_edges: + return + existing_edges = [ + edge for edge in self.graph.find_edges() if is_skos_hierarchy_edge(edge) + ] + validate_skos_hierarchy([*existing_edges, *hierarchy_edges]) + + def add_nodes_and_edges( + self, + nodes: List[Dict[str, Any]], + edges: List[Dict[str, Any]], + ) -> tuple[int, int]: + """ + Validate SKOS hierarchy edges upfront and add nodes and edges under lock. + + Note: This provides lock-based mutual exclusion and pre-write validation, + not transactional rollback atomicity. + """ + with self._lock: + self.validate_skos_hierarchy(edges) + nodes_added = self.graph.add_nodes(nodes) + edges_added = self.graph.add_edges(edges) + has_mutation_callback = callable(getattr(self.graph, "mutation_callback", None)) + if (nodes_added or edges_added) and not has_mutation_callback: + self._bump_graph_revision_locked() + if (nodes_added or edges_added) and not has_mutation_callback: + self.rebuild_search_index() + return nodes_added, edges_added def add_node( self, @@ -778,6 +805,7 @@ class GraphSession: **properties: Any, ) -> bool: with self._lock: + self.validate_skos_hierarchy([{"source": source_id, "target": target_id, "type": edge_type}]) added = self.graph.add_edge( source_id, target_id, diff --git a/semantica/utils/skos.py b/semantica/utils/skos.py index cfa48b0f..8c24bba5 100644 --- a/semantica/utils/skos.py +++ b/semantica/utils/skos.py @@ -7,6 +7,16 @@ from typing import Iterable, Mapping _HIERARCHY_EDGE_TYPES = frozenset({"skos:broader", "skos:narrower"}) +def is_skos_hierarchy_edge(edge: Mapping[str, object]) -> bool: + """Return whether an edge uses a SKOS hierarchy predicate.""" + if not isinstance(edge, Mapping): + return False + return any( + edge.get(key) in _HIERARCHY_EDGE_TYPES + for key in ("type", "edge_type", "relationship", "predicate", "relation") + ) + + def validate_skos_hierarchy(edges: Iterable[Mapping[str, object]]) -> None: """Raise ``ValueError`` when SKOS hierarchy edges contain a cycle. @@ -18,21 +28,25 @@ def validate_skos_hierarchy(edges: Iterable[Mapping[str, object]]) -> None: parents: dict[str, set[str]] = defaultdict(set) for edge in edges: - if not isinstance(edge, Mapping): + if not isinstance(edge, Mapping) or not is_skos_hierarchy_edge(edge): continue edge_type = next( ( edge.get(key) for key in ("type", "edge_type", "relationship", "predicate", "relation") - if edge.get(key) is not None + if edge.get(key) in _HIERARCHY_EDGE_TYPES ), None, ) if edge_type not in _HIERARCHY_EDGE_TYPES: continue - source = str(edge.get("source", edge.get("source_id", ""))) - target = str(edge.get("target", edge.get("target_id", ""))) + raw_source = edge.get("source", edge.get("source_id")) + raw_target = edge.get("target", edge.get("target_id")) + if raw_source is None or raw_target is None: + continue + source = str(raw_source).strip() + target = str(raw_target).strip() if not source or not target: continue diff --git a/tests/explorer/test_ontology_subissue3.py b/tests/explorer/test_ontology_subissue3.py index 2d6f7d9d..40cb6bbe 100644 --- a/tests/explorer/test_ontology_subissue3.py +++ b/tests/explorer/test_ontology_subissue3.py @@ -636,7 +636,25 @@ def test_health_shacl_dimension_returns_critical_for_truncated_graph(client): payload = client.get("/api/ontology/health?uri=http%3A%2F%2Fexample.org%2Fonto-a").json() shacl_dim = next(d for d in payload["dimensions"] if d["key"] == "shacl") assert shacl_dim["status"] == "critical" - assert shacl_dim["score"] == 0.0 assert "exceeds maximum analysis limit" in shacl_dim["detail"] +def test_ontology_load_rejects_cyclic_skos_hierarchy(client): + cyclic_ttl = """ +@prefix skos: . +@prefix ex: . +ex:S a skos:ConceptScheme ; skos:prefLabel "Scheme" . +ex:A a skos:Concept ; skos:prefLabel "Alpha" ; skos:inScheme ex:S ; skos:broader ex:B . +ex:B a skos:Concept ; skos:prefLabel "Beta" ; skos:inScheme ex:S ; skos:broader ex:A . +""" + response = client.post( + "/api/ontology/load", + json={ + "content": cyclic_ttl, + "format": "turtle", + }, + ) + assert response.status_code == 422 + assert "cycle" in response.json()["detail"].lower() + + diff --git a/tests/explorer/test_vocabulary.py b/tests/explorer/test_vocabulary.py index ee75ca7a..3de0f3e9 100644 --- a/tests/explorer/test_vocabulary.py +++ b/tests/explorer/test_vocabulary.py @@ -124,8 +124,7 @@ def test_hierarchy_cycle_does_not_hang(): def test_import_ttl_success(): - mock_session.add_nodes.return_value = 2 - mock_session.add_edges.return_value = 1 + mock_session.add_nodes_and_edges.return_value = (2, 1) response = client.post( "/api/vocabulary/import", @@ -139,8 +138,7 @@ def test_import_ttl_success(): def test_import_raw_text_success(): - mock_session.add_nodes.return_value = 1 - mock_session.add_edges.return_value = 0 + mock_session.add_nodes_and_edges.return_value = (1, 0) response = client.post( "/api/vocabulary/import", @@ -151,8 +149,7 @@ def test_import_raw_text_success(): def test_import_rdf_xml_success(): - mock_session.add_nodes.return_value = 1 - mock_session.add_edges.return_value = 0 + mock_session.add_nodes_and_edges.return_value = (1, 0) response = client.post( "/api/vocabulary/import", @@ -171,7 +168,11 @@ def test_import_invalid_file_returns_422(): def test_import_rejects_cyclic_hierarchy_before_writing_nodes(): - mock_session.validate_skos_hierarchy.side_effect = lambda edges: validate_skos_hierarchy(edges) + def add_nodes_and_edges(nodes, edges): + validate_skos_hierarchy(edges) + return 0, 0 + + mock_session.add_nodes_and_edges.side_effect = add_nodes_and_edges response = client.post( "/api/vocabulary/import", @@ -180,4 +181,4 @@ def test_import_rejects_cyclic_hierarchy_before_writing_nodes(): assert response.status_code == 422 assert "cycle" in response.json()["detail"].lower() - mock_session.add_nodes.assert_not_called() + mock_session.add_nodes_and_edges.assert_called_once() From bc75768afe1f3a991fc5e1d0e8919ae8f3a10a34 Mon Sep 17 00:00:00 2001 From: KaifAhmad1 Date: Sat, 1 Aug 2026 11:46:13 +0530 Subject: [PATCH 7/7] Fix two review findings in SKOS cycle validation - validate_skos_hierarchy() re-walked every existing hierarchy edge in the graph on each write, so one pre-existing cycle anywhere would block all unrelated future SKOS writes. It now only traverses concepts touched by the edges actually being written, while still checking those against existing edges for cross-boundary cycles. - In /api/ontology/load, `except HTTPException: raise` sat after a broader `except Exception` clause that already matched HTTPException, so a 422 raised after a successful OntologyIngestor parse was silently swallowed and retried via the fallback RDF parser instead of reaching the caller. Reordered the except clauses. Co-authored-by: mikemikimike <13286568797@163.com> --- CHANGELOG.md | 2 + semantica/context/context_graph.py | 4 +- semantica/explorer/routes/ontology.py | 10 ++- semantica/explorer/session.py | 2 +- semantica/utils/skos.py | 76 +++++++++++++++-------- tests/context/test_context.py | 13 ++++ tests/explorer/test_ontology_subissue3.py | 37 +++++++++++ tests/utils/test_skos.py | 64 +++++++++++++++++++ 8 files changed, 176 insertions(+), 32 deletions(-) create mode 100644 tests/utils/test_skos.py diff --git a/CHANGELOG.md b/CHANGELOG.md index bc00cef6..b9649c8b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -47,6 +47,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Added cycle detection (`validate_skos_hierarchy`) for `skos:broader` and `skos:narrower` relationships in `ContextGraph.add_edge()` and `ContextGraph.add_edges()`, preventing direct 2-node cycles, self-loops, and multi-hop hierarchy cycles - Added `GraphSession.add_nodes_and_edges()` to validate SKOS hierarchy edges upfront under lock before node insertion, preventing partial-write leaks where nodes remain after a cyclic edge is rejected - Updated vocabulary, ontology (`/api/ontology/load`, `/api/ontology/create`), and JSON/CSV import routes to use `add_nodes_and_edges()` and return HTTP 422 with actionable error messages when a cycle is detected + - Follow-up fix by @KaifAhmad1: `validate_skos_hierarchy()` previously re-walked *every* SKOS hierarchy edge already in the graph on each write, so one pre-existing cycle anywhere (e.g. legacy data) blocked all unrelated future writes; it now only traverses concepts touched by the edges being written, while still checking against existing edges for cycles that span old and new data + - Follow-up fix by @KaifAhmad1: in `/api/ontology/load`, `except HTTPException: raise` was unreachable because a broader `except Exception` clause above it already matched `HTTPException`, so a 422 raised after a successful `OntologyIngestor` parse was silently swallowed and reprocessed via the fallback RDF parser; reordered the clauses so the deliberate 422 always propagates - **`AgnoDecisionKit`/`AgnoKGToolkit` silently swallowed Agno tool registration failures** (#780, #818) by @Sameer6305 and @KaifAhmad1 - Removed the `try/except: pass` wrapped around `self.register(fn)` in both toolkits' `__init__`; when Agno is installed, a registration failure now propagates immediately instead of leaving the toolkit half-registered with no signal to the caller diff --git a/semantica/context/context_graph.py b/semantica/context/context_graph.py index f5ef322e..ebfc0464 100644 --- a/semantica/context/context_graph.py +++ b/semantica/context/context_graph.py @@ -597,7 +597,7 @@ class ContextGraph: existing_edges = [ edge for edge in self.find_edges() if is_skos_hierarchy_edge(edge) ] - validate_skos_hierarchy([*existing_edges, *hierarchy_edges]) + validate_skos_hierarchy(hierarchy_edges, existing_edges) for raw_edge in edges: if not isinstance(raw_edge, dict): continue @@ -962,7 +962,7 @@ class ContextGraph: existing_edges = [ edge for edge in self.find_edges() if is_skos_hierarchy_edge(edge) ] - validate_skos_hierarchy([*existing_edges, candidate]) + validate_skos_hierarchy([candidate], existing_edges) return self._add_internal_edge( ContextEdge( edge_id=edge_id, diff --git a/semantica/explorer/routes/ontology.py b/semantica/explorer/routes/ontology.py index 2fac4c15..a9cce469 100644 --- a/semantica/explorer/routes/ontology.py +++ b/semantica/explorer/routes/ontology.py @@ -1339,15 +1339,19 @@ async def load_ontology( except OSError as cleanup_exc: logger.debug("Failed to remove temporary ontology file: %s", cleanup_exc) + except HTTPException: + # Re-raise HTTPExceptions we deliberately raised above (e.g. the 422 from + # SKOS cycle validation) instead of letting the broad `except Exception` + # below mask them as an ingestor failure and silently retry via the + # fallback parser. + raise except Exception as ingest_exc: logger.warning(f"OntologyIngestor failed, falling back to basic parsing: {ingest_exc}") - + # Fallback to basic parsing nodes, edges, metadata = await asyncio.to_thread( _parse_rdf_sync, content_str.encode("utf-8"), fmt ) - except HTTPException: - raise except Exception as exc: raise HTTPException(status_code=422, detail=f"Could not parse ontology: {exc}") from exc diff --git a/semantica/explorer/session.py b/semantica/explorer/session.py index 369d326f..1d0a60f9 100644 --- a/semantica/explorer/session.py +++ b/semantica/explorer/session.py @@ -754,7 +754,7 @@ class GraphSession: existing_edges = [ edge for edge in self.graph.find_edges() if is_skos_hierarchy_edge(edge) ] - validate_skos_hierarchy([*existing_edges, *hierarchy_edges]) + validate_skos_hierarchy(hierarchy_edges, existing_edges) def add_nodes_and_edges( self, diff --git a/semantica/utils/skos.py b/semantica/utils/skos.py index 8c24bba5..dd105bca 100644 --- a/semantica/utils/skos.py +++ b/semantica/utils/skos.py @@ -17,41 +17,65 @@ def is_skos_hierarchy_edge(edge: Mapping[str, object]) -> bool: ) -def validate_skos_hierarchy(edges: Iterable[Mapping[str, object]]) -> None: - """Raise ``ValueError`` when SKOS hierarchy edges contain a cycle. +def _child_parent(edge: Mapping[str, object]) -> tuple[str, str] | None: + """Normalize a SKOS hierarchy edge to a ``(child, parent)`` pair, or ``None``.""" + if not isinstance(edge, Mapping) or not is_skos_hierarchy_edge(edge): + return None + edge_type = next( + ( + edge.get(key) + for key in ("type", "edge_type", "relationship", "predicate", "relation") + if edge.get(key) in _HIERARCHY_EDGE_TYPES + ), + None, + ) + if edge_type not in _HIERARCHY_EDGE_TYPES: + return None + + raw_source = edge.get("source", edge.get("source_id")) + raw_target = edge.get("target", edge.get("target_id")) + if raw_source is None or raw_target is None: + return None + source = str(raw_source).strip() + target = str(raw_target).strip() + if not source or not target: + return None + + return (source, target) if edge_type == "skos:broader" else (target, source) + + +def validate_skos_hierarchy( + new_edges: Iterable[Mapping[str, object]], + existing_edges: Iterable[Mapping[str, object]] = (), +) -> None: + """Raise ``ValueError`` when adding ``new_edges`` would introduce 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 child-to-parent adjacency before cycle detection. + + ``existing_edges`` supplies the SKOS hierarchy edges already persisted in + the graph so that cycles spanning old and new edges are still caught. + Only the concepts touched by ``new_edges`` are checked, though: a cycle + that already exists entirely within ``existing_edges`` must not block an + unrelated write elsewhere in the graph. """ - parents: dict[str, set[str]] = defaultdict(set) - for edge in edges: - if not isinstance(edge, Mapping) or not is_skos_hierarchy_edge(edge): - continue - edge_type = next( - ( - edge.get(key) - for key in ("type", "edge_type", "relationship", "predicate", "relation") - if edge.get(key) in _HIERARCHY_EDGE_TYPES - ), - None, - ) - if edge_type not in _HIERARCHY_EDGE_TYPES: - continue + for edge in existing_edges: + pair = _child_parent(edge) + if pair is not None: + parents[pair[0]].add(pair[1]) - raw_source = edge.get("source", edge.get("source_id")) - raw_target = edge.get("target", edge.get("target_id")) - if raw_source is None or raw_target is None: + touched: set[str] = set() + for edge in new_edges: + pair = _child_parent(edge) + if pair is None: continue - source = str(raw_source).strip() - target = str(raw_target).strip() - if not source or not target: - continue - - child, parent = ((source, target) if edge_type == "skos:broader" else (target, source)) + child, parent = pair parents[child].add(parent) + touched.add(child) + touched.add(parent) visiting: set[str] = set() visited: set[str] = set() @@ -68,5 +92,5 @@ def validate_skos_hierarchy(edges: Iterable[Mapping[str, object]]) -> None: visiting.remove(concept) visited.add(concept) - for concept in parents: + for concept in touched: visit(concept) diff --git a/tests/context/test_context.py b/tests/context/test_context.py index 15ec1b49..73edaa4e 100644 --- a/tests/context/test_context.py +++ b/tests/context/test_context.py @@ -223,6 +223,19 @@ class TestContextModule(unittest.TestCase): self.assertEqual(len(graph.edges), 0) + def test_context_graph_preexisting_unrelated_cycle_does_not_block_new_write(self): + """A cycle already persisted elsewhere in the graph (e.g. legacy data + written before cycle detection existed) must not poison unrelated + SKOS hierarchy writes for concepts it doesn't touch.""" + from semantica.context.context_graph import ContextEdge + + graph = ContextGraph() + graph._add_internal_edge(ContextEdge(source_id="X", target_id="Y", edge_type="skos:broader")) + graph._add_internal_edge(ContextEdge(source_id="Y", target_id="X", edge_type="skos:broader")) + + self.assertTrue(graph.add_edge("C", "D", "skos:broader")) + self.assertEqual(len(graph.edges), 3) + # --- 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_ontology_subissue3.py b/tests/explorer/test_ontology_subissue3.py index 40cb6bbe..a106d6d7 100644 --- a/tests/explorer/test_ontology_subissue3.py +++ b/tests/explorer/test_ontology_subissue3.py @@ -658,3 +658,40 @@ ex:B a skos:Concept ; skos:prefLabel "Beta" ; skos:inScheme ex:S ; skos:broader assert "cycle" in response.json()["detail"].lower() +def test_ontology_load_does_not_swallow_422_from_ingestor_success_path(client): + """A ValueError raised by add_nodes_and_edges() after OntologyIngestor + succeeds must surface as its own 422, not be masked by the broad + `except Exception` fallback-to-basic-parsing handler and silently + retried under a different parser.""" + from semantica.ingest.ontology_ingestor import OntologyData + + fake_data = OntologyData( + data={ + "uri": "http://example.org/onto-fake", + "name": "Fake Ontology", + "classes": [{"uri": "http://example.org/onto-fake#A", "name": "A"}], + "properties": [], + }, + source_path="fake.ttl", + format="turtle", + ) + + with patch( + "semantica.ingest.ontology_ingestor.OntologyIngestor.ingest_ontology", + return_value=fake_data, + ), patch( + "semantica.explorer.session.GraphSession.add_nodes_and_edges", + side_effect=ValueError("SKOS hierarchy contains a cycle involving 'A'."), + ), patch( + "semantica.explorer.routes.ontology._parse_rdf_sync" + ) as fallback_parse: + response = client.post( + "/api/ontology/load", + json={"content": "@prefix ex: . ex:A a ex:Thing .", "format": "turtle"}, + ) + + assert response.status_code == 422 + assert "cycle" in response.json()["detail"].lower() + fallback_parse.assert_not_called() + + diff --git a/tests/utils/test_skos.py b/tests/utils/test_skos.py new file mode 100644 index 00000000..ed28a6f0 --- /dev/null +++ b/tests/utils/test_skos.py @@ -0,0 +1,64 @@ +import unittest + +from semantica.utils.skos import is_skos_hierarchy_edge, validate_skos_hierarchy + + +class TestIsSkosHierarchyEdge(unittest.TestCase): + + def test_recognizes_broader_and_narrower(self): + self.assertTrue(is_skos_hierarchy_edge({"source": "A", "target": "B", "type": "skos:broader"})) + self.assertTrue(is_skos_hierarchy_edge({"source": "A", "target": "B", "type": "skos:narrower"})) + + def test_ignores_other_edge_types(self): + self.assertFalse(is_skos_hierarchy_edge({"source": "A", "target": "B", "type": "rdfs:subClassOf"})) + + def test_ignores_non_mapping(self): + self.assertFalse(is_skos_hierarchy_edge("not-a-dict")) + + +class TestValidateSkosHierarchy(unittest.TestCase): + + def test_accepts_acyclic_chain(self): + validate_skos_hierarchy([ + {"source": "A", "target": "B", "type": "skos:broader"}, + {"source": "B", "target": "C", "type": "skos:broader"}, + ]) + + def test_rejects_self_loop(self): + with self.assertRaisesRegex(ValueError, "cycle"): + validate_skos_hierarchy([{"source": "A", "target": "A", "type": "skos:broader"}]) + + def test_rejects_direct_two_node_cycle(self): + with self.assertRaisesRegex(ValueError, "cycle"): + validate_skos_hierarchy([ + {"source": "A", "target": "B", "type": "skos:broader"}, + {"source": "B", "target": "A", "type": "skos:broader"}, + ]) + + def test_rejects_cycle_spanning_existing_and_new_edges(self): + existing = [ + {"source": "A", "target": "B", "type": "skos:broader"}, + {"source": "B", "target": "C", "type": "skos:broader"}, + ] + with self.assertRaisesRegex(ValueError, "cycle"): + validate_skos_hierarchy([{"source": "C", "target": "A", "type": "skos:broader"}], existing) + + def test_preexisting_unrelated_cycle_does_not_block_new_write(self): + """A cycle already persisted elsewhere in the graph (e.g. legacy data + written before cycle detection existed) must not poison unrelated + writes for concepts it doesn't touch.""" + existing = [ + {"source": "X", "target": "Y", "type": "skos:broader"}, + {"source": "Y", "target": "X", "type": "skos:broader"}, + ] + validate_skos_hierarchy([{"source": "C", "target": "D", "type": "skos:broader"}], existing) + + def test_none_and_blank_endpoints_are_ignored(self): + validate_skos_hierarchy([ + {"source": None, "target": "B", "type": "skos:broader"}, + {"source": " ", "target": "B", "type": "skos:broader"}, + ]) + + +if __name__ == "__main__": + unittest.main()