mengzhx/semantica
1
0
Fork 0
mirror of https://github.com/semantica-agi/semantica.git synced 2026-08-29 04:26:20 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/main' into fix/779-record-decision-logging

Browse Source 
# Conflicts:
#	CHANGELOG.md
This commit is contained in:
KaifAhmad1
2026-08-02 13:10:20 +05:30
parent aa58b46d4d 0d447560bc
commit 46dcbbe731
19 changed files with 547 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+13
View File
@@ -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
docs/reference/explorer.md
+5
View File
@@ -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 |
docs/reference/provenance.md
+4 -4
View File
@@ -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
semantica/conflicts/source_tracker.py
+24 -9
View File
@@ -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(
semantica/context/context_graph.py
+15
View File
@@ -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,
semantica/explorer/routes/export_import.py
+8 -4
View File
@@ -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(
semantica/explorer/routes/ontology.py
+27 -10
View File
@@ -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(
semantica/explorer/routes/vocabulary.py
+7 -3
View File
@@ -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",
semantica/explorer/session.py
+35
View File
@@ -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,
semantica/provenance/manager.py
+28 -14
View File
@@ -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
semantica/split/provenance_tracker.py
+8 -4
View File
@@ -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)
semantica/utils/skos.py
+96
View File
@@ -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)
tests/conflicts/test_conflicts.py
+24
View File
@@ -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()
tests/context/test_context.py
+33
View File
@@ -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):
tests/explorer/test_ontology_subissue3.py
+56 -1
View File
@@ -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: <http://www.w3.org/2004/02/skos/core#> .
@prefix ex: <http://example.org/> .
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: <http://example.org/> . ex:A a ex:Thing .", "format": "turtle"},
)
assert response.status_code == 422
assert "cycle" in response.json()["detail"].lower()
fallback_parse.assert_not_called()
tests/explorer/test_vocabulary.py
+30 -6
View File
@@ -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"""<?xml version=\"1.0\"?>
"""
CYCLIC_TTL = b"""
@prefix skos: <http://www.w3.org/2004/02/skos/core#> .
@prefix ex: <http://example.org/> .
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()
tests/provenance/test_backward_compat.py
+16
View File
@@ -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."""
tests/provenance/test_manager.py
+54 -13
View File
@@ -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
tests/utils/test_skos.py
+64
View File
@@ -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()