fix(explorer): wire ProvenanceManager into provenance routes (closes #792)

- 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

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

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"),
     )
 
 

semantica/explorer/schemas.py

@@ -324,6 +324,7 @@ class ProvenanceEdge(BaseModel):
 class ProvenanceResponse(BaseModel):
     nodes: List[ProvenanceNode]
     edges: List[ProvenanceEdge]
+    source: Optional[str] = None
 
 
 # ---------------------------------------------------------------------------

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 {})

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