diff --git a/CHANGELOG.md b/CHANGELOG.md index c74b0dda..22e57a25 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -43,6 +43,13 @@ 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 + - 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 + - **Agno `_AgentScopedStore.upsert_memory` silently swallowed decision recording failures** (#779) - `upsert_memory()` now logs `logger.warning("[%s] record_decision failed: %s", self._role, exc, exc_info=True)` when `record_decision()` fails, matching the error-logging convention used for `store()` in the same method with traceback context preserved - Preserves graceful fallback behavior: `record_decision()` remains optional and `upsert_memory()` continues without propagating the exception @@ -55,6 +62,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - This is a behavior change for callers that construct these toolkits expecting instantiation to always succeed — audited: no in-repo call site relies on the old silent-failure behavior - Expanded `tests/integrations/agno/test_decision_kit.py` and `test_kg_toolkit.py` with coverage for registration invocation counts, failure propagation, graceful degradation, and no-duplicate-`_tools` assertions +- **`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 + - **MCP `handle_get_causal_chain` returned an empty-but-valid-looking response when both `CausalChainAnalyzer` and the graph fallback were unavailable** (#781, #817) by @Sameer6305 and @KaifAhmad1 - Returns an explicit `{"error": "Causal chain analysis is not supported on this graph backend", "chain": []}` instead of `{"chain": [], "count": 0, "direction": ...}`, letting clients distinguish "unsupported" from a legitimately empty chain - The fallback path now introspects `graph.get_causal_chain`'s signature to forward `direction`/`max_depth` (or a `depth` kwarg, or nothing, depending on what the backend accepts) instead of always calling with just `decision_id`, matching the primary analyzer path's behavior 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/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 diff --git a/semantica/conflicts/source_tracker.py b/semantica/conflicts/source_tracker.py index 3a00567f..9ae19aa9 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 + ok = self.track_entity_source(entity_id, source_ref, **metadata) + if ok: + 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( + ok = self.track_property_source( entity_id, property_name, value, source_ref, **metadata ) - stats["properties_tracked"] += 1 - stats["total_tracked"] += 1 + if ok: + 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( + ok = self.track_relationship_source( relationship_id, source_ref, **metadata ) - stats["relationships_tracked"] += 1 - stats["total_tracked"] += 1 + if ok: + 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/context/context_graph.py b/semantica/context/context_graph.py index ac1b9b59..ebfc0464 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 is_skos_hierarchy_edge, validate_skos_hierarchy from .entity_linker import EntityLinker # Optional imports for advanced features @@ -589,6 +590,14 @@ 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. + 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(hierarchy_edges, existing_edges) for raw_edge in edges: if not isinstance(raw_edge, dict): continue @@ -948,6 +957,12 @@ class ContextGraph: family_id=explicit_family_id, ) with self._lock: + 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([candidate], existing_edges) 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..a9cce469 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) @@ -1334,21 +1339,29 @@ 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 # 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 +1558,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 4e992229..17739b67 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,8 +224,12 @@ async def import_vocabulary( 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) + 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 return VocabularyImportResponse( status="success", diff --git a/semantica/explorer/session.py b/semantica/explorer/session.py index 8692f59e..1d0a60f9 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 is_skos_hierarchy_edge, 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,38 @@ 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.""" + 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(hierarchy_edges, existing_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, node_id: str, @@ -771,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/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/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/semantica/utils/skos.py b/semantica/utils/skos.py new file mode 100644 index 00000000..dd105bca --- /dev/null +++ b/semantica/utils/skos.py @@ -0,0 +1,96 @@ +"""Validation helpers for SKOS graph relationships.""" + +from collections import defaultdict +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 _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 existing_edges: + pair = _child_parent(edge) + if pair is not None: + parents[pair[0]].add(pair[1]) + + touched: set[str] = set() + for edge in new_edges: + pair = _child_parent(edge) + if pair is None: + continue + child, parent = pair + parents[child].add(parent) + touched.add(child) + touched.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 touched: + visit(concept) diff --git a/tests/conflicts/test_conflicts.py b/tests/conflicts/test_conflicts.py index bee5c828..34dad4d2 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=False): + 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/context/test_context.py b/tests/context/test_context.py index 54f705f1..73edaa4e 100644 --- a/tests/context/test_context.py +++ b/tests/context/test_context.py @@ -203,6 +203,39 @@ 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) + + 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 2d6f7d9d..a106d6d7 100644 --- a/tests/explorer/test_ontology_subissue3.py +++ b/tests/explorer/test_ontology_subissue3.py @@ -636,7 +636,62 @@ 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() + + +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/explorer/test_vocabulary.py b/tests/explorer/test_vocabulary.py index 7b91d17f..3de0f3e9 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.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() @@ -114,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", @@ -129,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", @@ -141,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", @@ -158,3 +165,20 @@ 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(): + 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", + files={"file": ("cyclic.ttl", CYCLIC_TTL, "text/turtle")}, + ) + + assert response.status_code == 422 + assert "cycle" in response.json()["detail"].lower() + mock_session.add_nodes_and_edges.assert_called_once() 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.""" 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 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()