From 2ee4da0b47e3d2e2241cb2598f9af20134b2622d Mon Sep 17 00:00:00 2001 From: Sameer6305 Date: Tue, 28 Jul 2026 16:23:02 +0530 Subject: [PATCH] fix(explorer): wire ProvenanceManager into provenance routes (closes #792) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Explorer's /api/provenance now queries the audit-grade ProvenanceManager (SQLite-backed, checksummed) first, falling back to the naive 2-hop graph traversal when no audit records exist for a node. - Fixed a process-global mutable-state risk in the initial approach: provenance storage path is threaded per-session via GraphSession, not via ProvenanceManager's global set_default_storage_path classmethod. - Added source: 'audit' | 'graph_traversal' to the response so callers can distinguish which path served the data. - Documented a known limitation: ProvenanceManager currently only traces upstream/ancestor lineage, not descendants — the naive fallback remains the only source for downstream relationships until ProvenanceManager gains a reverse lookup (tracked separately). - Warns (rather than silently no-ops) if a provided session's provenance_manager was already constructed before create_app() applied a provenance_storage_path. - Never lets a provenance-manager failure crash the route; degrades to the naive path with a logged warning instead. Tests: 5 new tests in test_provenance_manager_wiring.py covering the audit path, empty-record fallback, storage-failure degradation, app startup wiring, and cross-session storage isolation. Full tests/explorer/ + tests/provenance/ suite passing, order-invariant. --- semantica/explorer/app.py | 35 ++++- semantica/explorer/routes/provenance.py | 119 ++++++++++++++++- semantica/explorer/schemas.py | 1 + semantica/explorer/session.py | 20 ++- .../test_provenance_manager_wiring.py | 124 ++++++++++++++++++ 5 files changed, 292 insertions(+), 7 deletions(-) create mode 100644 tests/explorer/test_provenance_manager_wiring.py diff --git a/semantica/explorer/app.py b/semantica/explorer/app.py index 96d66b14..7189fade 100644 --- a/semantica/explorer/app.py +++ b/semantica/explorer/app.py @@ -3,6 +3,7 @@ Semantica Explorer FastAPI application factory. """ import asyncio +import logging import os from contextlib import asynccontextmanager from pathlib import Path @@ -18,6 +19,8 @@ from ..context.context_graph import ContextGraph from .session import GraphSession from .ws import ConnectionManager +logger = logging.getLogger(__name__) + def _read_int_env(name: str, default: int) -> int: raw_value = os.environ.get(name) @@ -45,6 +48,10 @@ def _read_explorer_settings() -> dict: # ContextGraph and does not open a network connection to FalkorDB. "falkordb_host": os.environ.get("FALKORDB_HOST", "localhost"), "falkordb_port": _read_int_env("FALKORDB_PORT", 6379), + "provenance_storage_path": os.environ.get( + "SEMANTICA_PROVENANCE_DB", + os.environ.get("EXPLORER_PROVENANCE_DB"), + ), } @@ -75,9 +82,33 @@ def _install_mutation_bridge(app: FastAPI, session: GraphSession) -> None: session.graph.mutation_callback = on_mutation -def create_app(session: Optional[GraphSession] = None) -> FastAPI: - active_session = session or GraphSession(ContextGraph(advanced_analytics=False)) +def create_app( + session: Optional[GraphSession] = None, + provenance_storage_path: Optional[str] = None, +) -> FastAPI: settings = _read_explorer_settings() + prov_path = provenance_storage_path or settings.get("provenance_storage_path") + if session is None: + active_session = GraphSession( + ContextGraph(advanced_analytics=False), + provenance_storage_path=prov_path, + ) + else: + active_session = session + if prov_path is not None: + if active_session._provenance_storage_path is None: + if getattr(active_session, "_provenance_manager", None) is not None: + logger.warning( + "provenance_storage_path=%s was supplied to create_app(), but " + "the given session's provenance_manager was already constructed " + "(likely accessed before create_app() ran) and will keep using " + "its original storage. Pass provenance_storage_path when " + "constructing the GraphSession instead, or access " + "session.provenance_manager only after create_app().", + prov_path, + ) + else: + active_session._provenance_storage_path = prov_path @asynccontextmanager async def lifespan(app: FastAPI): diff --git a/semantica/explorer/routes/provenance.py b/semantica/explorer/routes/provenance.py index 8b1b1f03..a3f72337 100644 --- a/semantica/explorer/routes/provenance.py +++ b/semantica/explorer/routes/provenance.py @@ -4,6 +4,7 @@ Provenance routes for lineage visualization and exportable reports. import asyncio import json +import logging from typing import Any, Dict, List, Optional import networkx as nx @@ -14,6 +15,7 @@ from ..dependencies import get_session from ..schemas import ProvenanceEdge, ProvenanceNode, ProvenanceResponse from ..session import GraphSession +logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/provenance", tags=["Power User Tools"]) _AGENT_TYPES = {"person", "organization", "system", "agent"} @@ -29,9 +31,118 @@ def _classify_prov(node_type: str) -> tuple[str, str]: return "Entity", "group_entity" +def _transform_audit_lineage(lineage: Dict[str, Any], node_id: str) -> Dict[str, Any]: + # Mapping decision (W3C PROV-O to frontend swim-lanes): + # Every ProvenanceEntry maps to a node classified by _classify_prov(entry["entity_type"]), + # placing documents/chunks/entities in 'group_entity' (prov_type='Entity'), persons/systems + # in 'group_agent', and actions/processes in 'group_activity'. + # For derivation relationships (parent_entity_id and used_entities), we connect + # parent -> child directly with edge label set to activity_id (or 'wasDerivedFrom'), + # keeping the lineage graph scannable without cluttering it with intermediate activity + # nodes when activity_id is an operational label. + nodes: List[Dict[str, Any]] = [] + edges: List[Dict[str, Any]] = [] + seen_nodes = set() + seen_edges = set() + + chain = lineage.get("lineage_chain") or lineage.get("entries") or [] + for entry in chain: + if not isinstance(entry, dict): + continue + eid = str(entry.get("entity_id") or "") + if not eid or eid in seen_nodes: + continue + seen_nodes.add(eid) + metadata = entry.get("metadata") if isinstance(entry.get("metadata"), dict) else {} + label = str(metadata.get("title") or metadata.get("label") or eid) + prov_type, parent_id = _classify_prov(str(entry.get("entity_type", "entity"))) + nodes.append({ + "id": eid, + "label": label, + "prov_type": prov_type, + "parent_id": parent_id, + }) + + for entry in chain: + if not isinstance(entry, dict): + continue + eid = str(entry.get("entity_id") or "") + if not eid: + continue + parents = [] + if entry.get("parent_entity_id"): + parents.append(str(entry.get("parent_entity_id"))) + used = entry.get("used_entities") + if isinstance(used, list): + for u in used: + if u and str(u) not in parents: + parents.append(str(u)) + + activity = str(entry.get("activity_id") or "wasDerivedFrom") + for src in parents: + edge_key = (src, eid) + if edge_key in seen_edges: + continue + seen_edges.add(edge_key) + + # NOTE: ProvenanceManager.get_lineage() currently only traces upstream + # ancestor chains via parent_entity_id and used_entities. It does not perform + # reverse lookups for downstream descendants. Consequently, 'direction = "downstream"' + # is unreachable in practice for this audit path until reverse lookup is supported + # by ProvenanceManager. + if eid == node_id: + direction = "upstream" + elif src == node_id: + direction = "downstream" + else: + direction = "lateral" + + edges.append({ + "id": f"{src}-{eid}", + "source": src, + "target": eid, + "label": activity, + "direction": direction, + }) + + for edge in edges: + for endpoint in (edge["source"], edge["target"]): + if endpoint not in seen_nodes: + seen_nodes.add(endpoint) + prov_type, parent_id = _classify_prov("entity") + nodes.append({ + "id": endpoint, + "label": endpoint, + "prov_type": prov_type, + "parent_id": parent_id, + }) + + return {"nodes": nodes, "edges": edges, "source": "audit"} + + def _build_provenance(session: GraphSession, node_id: Optional[str] = None) -> dict: - if not node_id or node_id not in session.graph.nodes: - return {"nodes": [], "edges": []} + """Build provenance lineage for a node, attempting the audit-grade store first. + + NOTE: The audit path (source='audit') shows verified upstream lineage only. + For descendant/downstream relationships, the naive graph-traversal fallback + remains the only source until ProvenanceManager gains a reverse lookup. + """ + if not node_id: + return {"nodes": [], "edges": [], "source": "graph_traversal"} + + try: + manager = getattr(session, "provenance_manager", None) + if manager is not None: + lineage = manager.get_lineage(node_id) + if lineage and lineage.get("entity_count", 0) > 0: + return _transform_audit_lineage(lineage, node_id) + except Exception as exc: + logger.warning( + f"ProvenanceManager get_lineage failed for {node_id}, falling back to graph traversal: {exc}" + ) + + if node_id not in session.graph.nodes: + return {"nodes": [], "edges": [], "source": "graph_traversal"} graph = nx.DiGraph() graph.add_node(node_id) @@ -81,7 +192,7 @@ def _build_provenance(session: GraphSession, node_id: Optional[str] = None) -> d } ) - return {"nodes": provenance_nodes, "edges": provenance_edges} + return {"nodes": provenance_nodes, "edges": provenance_edges, "source": "graph_traversal"} def _build_report(session: GraphSession, node_id: str) -> Dict[str, Any]: @@ -93,6 +204,7 @@ def _build_report(session: GraphSession, node_id: str) -> Dict[str, Any]: "type": node.get("type", "entity") if node else "entity", "properties": node.get("metadata", node.get("properties", {})) if node else {}, "lineage": provenance, + "source": provenance.get("source", "graph_traversal"), } @@ -152,6 +264,7 @@ async def get_provenance_lineage( return ProvenanceResponse( nodes=[ProvenanceNode(**node) for node in data["nodes"]], edges=[ProvenanceEdge(**edge) for edge in data["edges"]], + source=data.get("source", "graph_traversal"), ) diff --git a/semantica/explorer/schemas.py b/semantica/explorer/schemas.py index 226cd41e..65418433 100644 --- a/semantica/explorer/schemas.py +++ b/semantica/explorer/schemas.py @@ -324,6 +324,7 @@ class ProvenanceEdge(BaseModel): class ProvenanceResponse(BaseModel): nodes: List[ProvenanceNode] edges: List[ProvenanceEdge] + source: Optional[str] = None # --------------------------------------------------------------------------- diff --git a/semantica/explorer/session.py b/semantica/explorer/session.py index b2ef8c57..1ea6c082 100644 --- a/semantica/explorer/session.py +++ b/semantica/explorer/session.py @@ -1,4 +1,4 @@ -""" +""" Semantica Explorer session helpers. """ @@ -37,8 +37,13 @@ logger = logging.getLogger(__name__) class GraphSession: """Thread-safe session wrapper around a loaded ``ContextGraph``.""" - def __init__(self, graph: ContextGraph) -> None: + def __init__( + self, + graph: ContextGraph, + provenance_storage_path: Optional[str] = None, + ) -> None: self.graph = graph + self._provenance_storage_path = provenance_storage_path self._lock = threading.RLock() self._search_index = GraphSearchIndex() @@ -52,6 +57,7 @@ class GraphSession: self._similarity: Any = None self._link_predictor: Any = None self._validator: Any = None + self._provenance_manager: Any = None self._graph_revision: int = 0 self._cached_embeddings: Optional[Dict[str, List[float]]] = None @@ -178,6 +184,16 @@ class GraphSession: self._validator = GraphValidator() return self._validator + @property + def provenance_manager(self) -> Any: + with self._lock: + if self._provenance_manager is None: + from ..provenance import ProvenanceManager + self._provenance_manager = ProvenanceManager( + storage_path=self._provenance_storage_path + ) + return self._provenance_manager + def normalize_node(self, node: Dict[str, Any]) -> Dict[str, Any]: meta: Dict[str, Any] = {} meta.update(node.get("metadata", {}) or {}) diff --git a/tests/explorer/test_provenance_manager_wiring.py b/tests/explorer/test_provenance_manager_wiring.py new file mode 100644 index 00000000..dec8c1b8 --- /dev/null +++ b/tests/explorer/test_provenance_manager_wiring.py @@ -0,0 +1,124 @@ +""" +Tests for ProvenanceManager wiring into Explorer routes and application startup. +""" + +from unittest.mock import patch + +import pytest +from starlette.testclient import TestClient + +from semantica.context.context_graph import ContextGraph +from semantica.explorer.app import create_app +from semantica.explorer.session import GraphSession +from semantica.provenance import ProvenanceManager +from semantica.provenance.storage import SQLiteStorage + + +@pytest.fixture +def session(): + sess = GraphSession(ContextGraph(advanced_analytics=False)) + return sess + + +@pytest.fixture +def client(session): + app = create_app(session=session) + with TestClient(app) as tc: + yield tc + + +def test_provenance_manager_wiring_audit_path(client, session): + """Test that a multi-hop track_entity chain is returned from /api/provenance with source='audit'.""" + pm = session.provenance_manager + pm.track_entity(entity_id="grandparent", source="doc_1", entity_type="document") + pm.track_entity( + entity_id="parent", + source="doc_1", + metadata={"derived_from": "grandparent"}, + entity_type="chunk", + ) + pm.track_entity( + entity_id="child", + source="doc_1", + metadata={"derived_from": "parent"}, + entity_type="named_entity", + ) + + response = client.get("/api/provenance", params={"node_id": "child"}) + assert response.status_code == 200 + data = response.json() + + assert data.get("source") == "audit" + node_ids = {n["id"] for n in data["nodes"]} + assert {"grandparent", "parent", "child"}.issubset(node_ids) + edge_pairs = {(e["source"], e["target"]) for e in data["edges"]} + assert ("grandparent", "parent") in edge_pairs + assert ("parent", "child") in edge_pairs + + # Confirm /api/provenance/report also uses audit path + rep_response = client.get("/api/provenance/report", params={"node_id": "child"}) + assert rep_response.status_code == 200 + report_data = rep_response.json() + assert report_data.get("source") == "audit" + assert report_data["lineage"].get("source") == "audit" + + +def test_provenance_manager_wiring_fallback_no_records(client, session): + """Test that a node with no audit records falls back cleanly to source='graph_traversal'.""" + session.add_node("orphan_node", content="Orphan Node Content", node_type="entity") + + response = client.get("/api/provenance", params={"node_id": "orphan_node"}) + assert response.status_code == 200 + data = response.json() + + assert data.get("source") == "graph_traversal" + assert len(data["nodes"]) == 1 + assert data["nodes"][0]["id"] == "orphan_node" + + +def test_provenance_manager_wiring_error_graceful_degradation(client, session): + """Test that a simulated ProvenanceManager error degrades gracefully to naive traversal (200, not 500).""" + session.add_node("some_node", content="Some Node", node_type="entity") + + with patch.object( + session.provenance_manager, + "get_lineage", + side_effect=RuntimeError("Simulated storage failure"), + ): + response = client.get("/api/provenance", params={"node_id": "some_node"}) + assert response.status_code == 200 + data = response.json() + assert data.get("source") == "graph_traversal" + assert len(data["nodes"]) == 1 + assert data["nodes"][0]["id"] == "some_node" + + +def test_create_app_provenance_storage_path_wiring(tmp_path): + """Test that create_app(provenance_storage_path=...) sets per-instance storage without global mutation.""" + test_path = str(tmp_path / "test_explorer_prov.db") + app = create_app(provenance_storage_path=test_path) + + with TestClient(app): + assert app.state.session._provenance_storage_path == test_path + assert isinstance(app.state.session.provenance_manager.storage, SQLiteStorage) + + +def test_provenance_storage_isolation_between_sessions(tmp_path): + """Test that two GraphSessions with different storage paths do not leak state across instances.""" + path1 = str(tmp_path / "session1.db") + path2 = str(tmp_path / "session2.db") + sess1 = GraphSession(ContextGraph(advanced_analytics=False), provenance_storage_path=path1) + sess2 = GraphSession(ContextGraph(advanced_analytics=False), provenance_storage_path=path2) + + pm1 = sess1.provenance_manager + pm2 = sess2.provenance_manager + + assert pm1.storage.db_path == path1 + assert pm2.storage.db_path == path2 + assert pm1 is not pm2 + assert pm1.storage is not pm2.storage + + # Write an entity to sess1 and confirm it does NOT appear in sess2 + pm1.track_entity(entity_id="node_in_1", source="doc_A", entity_type="entity") + assert pm1.storage.retrieve("node_in_1") is not None + assert pm2.storage.retrieve("node_in_1") is None