Merge pull request #809 from Sameer6305/fix/792-provenance-manager-wiring

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

CHANGELOG.md

@@ -35,6 +35,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- **Explorer's Provenance UI used a naive 2-hop graph traversal instead of the audit-grade `ProvenanceManager` backend** (#792, #809) by @Sameer6305
+  - `semantica/explorer/routes/provenance.py` never imported or called `ProvenanceManager` (`semantica/provenance/manager.py`); `/api/provenance` and `/api/provenance/report` built their lineage response entirely from a naive 2-hop networkx traversal over the live graph instead of querying the SQLite-backed, checksummed audit log. Both endpoints now query `session.provenance_manager.get_lineage(node_id)` first, and a new `_transform_audit_lineage()` maps the W3C PROV-O entries into the exact `{"nodes": [...], "edges": [...]}` shape `LineageDiagram.tsx` already expects — no frontend changes required
+  - Falls back to the original 2-hop traversal, never a 500: no audit records for a node, a `ProvenanceManager` storage failure (corrupted DB, permissions), or a failed SHA-256 integrity check on any entry in the lineage chain all degrade cleanly to the naive path. A new `source: "audit" | "graph_traversal"` field on the response discloses which path actually served the data
+  - `ProvenanceManager.get_lineage()` now returns `integrity_verified`, computed by re-verifying every entry's checksum before it's trusted; a single tampered or corrupted entry anywhere in the lineage chain now falls the *entire* response back to graph traversal rather than serving partially-verified audit data
+  - Replaced an initial classmethod-based `ProvenanceManager.set_default_storage_path()` approach (caught in review before merge — it would have let any two sessions/apps in the same process silently share and overwrite each other's storage path, including across unrelated test runs) with `provenance_storage_path` threaded through `GraphSession.__init__` and `create_app(...)`, so each session's `ProvenanceManager` is independently scoped
+  - Disclosed limitation: `ProvenanceManager.trace_lineage()`/`get_lineage()` only walk `parent_entity_id`/`used_entities` backward, so the audit path currently surfaces upstream lineage only — the naive fallback remains the only source for downstream/descendant relationships until `ProvenanceManager` gains a reverse lookup
+  - New `tests/explorer/test_provenance_manager_wiring.py` (8 tests): the audit path via a real multi-hop `track_entity()` chain, empty-record fallback, simulated storage-failure degradation (asserts `200`, not `500`), checksum-tamper fallback, evidence-field preservation, `create_app()` storage-path wiring, and cross-session storage isolation, confirmed order-invariant across `tests/explorer/` and `tests/provenance/` in both execution orders
+
 - **`POST /shacl/validate` and the `/health` SHACL dimension never ran live SHACL validation** (#772, #804) by @Sameer6305 and @KaifAhmad1
   - `/shacl/validate` had no data graph to validate submitted shapes against — only a Turtle syntax check. Added `_data_graph_turtle_for_uri()`, which serializes the loaded ontology's nodes/edges into an RDF/Turtle instance graph (CURIE resolution across owl/rdfs/skos/dct/dc, arbitrary node-property projection, typed individuals) and wires both `/shacl/validate` and the `/health` SHACL dimension to `OntologyEngine.validate_graph()` via pySHACL, returning real `conforms`/violations instead of a hardcoded `status="unavailable"` stub
   - Fixed a cross-ontology namespace leak in `_node_belongs_to_ontology`: its prefix fallback (`_extract_namespace()`) split only on the last `/`, so sibling ontologies sharing a domain (e.g. `.../onto-a` and `.../onto-b`) could match entities across ontologies that shouldn't be related; fixed by comparing against the full URI stem via the new `_ontology_namespace()` helper

semantica/explorer/app.py

@@ -45,6 +45,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 +79,21 @@ 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:
+            active_session.set_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
@@ -13,7 +14,9 @@ from fastapi.responses import PlainTextResponse, Response
 from ..dependencies import get_session
 from ..schemas import ProvenanceEdge, ProvenanceNode, ProvenanceResponse
 from ..session import GraphSession
+from ...provenance.integrity import verify_checksum
 
+logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/api/provenance", tags=["Power User Tools"])
 
 _AGENT_TYPES = {"person", "organization", "system", "agent"}
@@ -29,9 +32,134 @@ 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,
+            "source_document": entry.get("source_document") or None,
+            "source_location": entry.get("source_location") or None,
+            "source_quote": entry.get("source_quote") or None,
+            "confidence": entry.get("confidence"),
+            "checksum": entry.get("checksum") or None,
+        })
+
+    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. All ancestor derivation edges are upstream lineage.
+            if src == node_id:
+                direction = "downstream"
+            else:
+                direction = "upstream"
+
+            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,
+                    "source_document": None,
+                    "source_location": None,
+                    "source_quote": None,
+                    "confidence": None,
+                    "checksum": None,
+                })
+
+    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:
+                integrity_ok = lineage.get("integrity_verified")
+                if integrity_ok is None:
+                    entries = lineage.get("lineage_chain") or lineage.get("entries") or []
+                    integrity_ok = all(verify_checksum(entry) for entry in entries)
+                if integrity_ok:
+                    return _transform_audit_lineage(lineage, node_id)
+                logger.warning(
+                    f"Provenance integrity verification failed for {node_id}, falling back to graph traversal"
+                )
+    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 +209,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 +221,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"),
     }
 
 
@@ -114,7 +243,14 @@ def _render_markdown(report: Dict[str, Any]) -> str:
 
     lines.extend(["", "## Lineage Nodes"])
     for node in report.get("lineage", {}).get("nodes", []):
-        lines.append(f"- `{node['id']}` ({node['prov_type']}): {node['label']}")
+        line = f"- `{node['id']}` ({node['prov_type']}): {node['label']}"
+        if node.get("source_document"):
+            line += f" [source: {node['source_document']}]"
+        if node.get("confidence") is not None:
+            line += f" (confidence: {node['confidence']})"
+        if node.get("checksum"):
+            line += f" (checksum: {node['checksum'][:8]}...)"
+        lines.append(line)
 
     edges = report.get("lineage", {}).get("edges", [])
     grouped_edges: Dict[str, List] = {"upstream": [], "downstream": [], "lateral": []}
@@ -152,6 +288,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

@@ -311,6 +311,11 @@ class ProvenanceNode(BaseModel):
     label: str
     prov_type: str
     parent_id: Optional[str] = None
+    source_document: Optional[str] = None
+    source_location: Optional[str] = None
+    source_quote: Optional[str] = None
+    confidence: Optional[float] = None
+    checksum: Optional[str] = None
 
 
 class ProvenanceEdge(BaseModel):
@@ -324,6 +329,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,38 @@ 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 set_provenance_storage_path(self, storage_path: Optional[str]) -> None:
+        """Set or reconfigure the provenance storage path for this session.
+
+        Raises a ValueError if a conflicting storage path is already configured or
+        if the provenance manager has already been constructed with a different path.
+        """
+        with self._lock:
+            if self._provenance_storage_path == storage_path:
+                return
+            if self._provenance_manager is not None:
+                raise ValueError(
+                    f"Cannot change provenance_storage_path to '{storage_path}': "
+                    f"provenance_manager is already initialized with "
+                    f"'{self._provenance_storage_path}'."
+                )
+            if self._provenance_storage_path is not None and storage_path is not None:
+                raise ValueError(
+                    f"Conflicting provenance_storage_path: session is already configured "
+                    f"with '{self._provenance_storage_path}', cannot overwrite with '{storage_path}'."
+                )
+            self._provenance_storage_path = storage_path
+
     def normalize_node(self, node: Dict[str, Any]) -> Dict[str, Any]:
         meta: Dict[str, Any] = {}
         meta.update(node.get("metadata", {}) or {})

semantica/provenance/integrity.py

@@ -24,7 +24,7 @@ from typing import Any, Dict, Optional
 from .schemas import ProvenanceEntry
 
 
-def compute_checksum(entry: ProvenanceEntry) -> str:
+def compute_checksum(entry: Any) -> str:
     """
     Compute SHA-256 checksum for a provenance entry.
     
@@ -32,7 +32,7 @@ def compute_checksum(entry: ProvenanceEntry) -> str:
     to detect any tampering or corruption of provenance data.
     
     Args:
-        entry: ProvenanceEntry to compute checksum for
+        entry: ProvenanceEntry or dict to compute checksum for
         
     Returns:
         SHA-256 checksum as hexadecimal string
@@ -49,19 +49,29 @@ def compute_checksum(entry: ProvenanceEntry) -> str:
         'a3b2c1d4e5f6...'
     """
     # Concatenate critical fields for checksum
-    data = (
-        f"{entry.entity_id}"
-        f"{entry.entity_type}"
-        f"{entry.activity_id}"
-        f"{entry.source_document}"
-        f"{entry.timestamp}"
-        f"{entry.confidence}"
-    )
+    if isinstance(entry, dict):
+        data = (
+            f"{entry.get('entity_id') or ''}"
+            f"{entry.get('entity_type') or ''}"
+            f"{entry.get('activity_id') or ''}"
+            f"{entry.get('source_document') or ''}"
+            f"{entry.get('timestamp') or ''}"
+            f"{entry.get('confidence') if entry.get('confidence') is not None else 1.0}"
+        )
+    else:
+        data = (
+            f"{entry.entity_id}"
+            f"{entry.entity_type}"
+            f"{entry.activity_id}"
+            f"{entry.source_document}"
+            f"{entry.timestamp}"
+            f"{entry.confidence}"
+        )
     
     return hashlib.sha256(data.encode('utf-8')).hexdigest()
 
 
-def verify_checksum(entry: ProvenanceEntry, expected_checksum: Optional[str] = None) -> bool:
+def verify_checksum(entry: Any, expected_checksum: Optional[str] = None) -> bool:
     """
     Verify checksum for a provenance entry.
     
@@ -69,7 +79,7 @@ def verify_checksum(entry: ProvenanceEntry, expected_checksum: Optional[str] = N
     to detect tampering or corruption.
     
     Args:
-        entry: ProvenanceEntry to verify
+        entry: ProvenanceEntry or dict to verify
         expected_checksum: Expected checksum (uses entry.checksum if None)
         
     Returns:
@@ -83,7 +93,10 @@ def verify_checksum(entry: ProvenanceEntry, expected_checksum: Optional[str] = N
         True
     """
     if expected_checksum is None:
-        expected_checksum = entry.checksum
+        if isinstance(entry, dict):
+            expected_checksum = entry.get("checksum")
+        else:
+            expected_checksum = getattr(entry, "checksum", None)
     
     if expected_checksum is None:
         return False

semantica/provenance/manager.py

@@ -29,11 +29,12 @@ from collections.abc import Mapping
 from datetime import datetime
 from contextlib import contextmanager
 import copy
+import json
 import threading
 
 from .schemas import ProvenanceEntry, SourceReference
 from .storage import ProvenanceStorage, InMemoryStorage, SQLiteStorage
-from .integrity import compute_checksum
+from .integrity import compute_checksum, verify_checksum
 
 
 @contextmanager
@@ -553,7 +554,6 @@ class ProvenanceManager:
                 meta = entry.metadata
                 if isinstance(meta, str):
                     try:
-                        import json
                         meta = json.loads(meta)
                     except (json.JSONDecodeError, TypeError):
                         pass
@@ -561,6 +561,7 @@ class ProvenanceManager:
                 if isinstance(meta, dict):
                     aggregated_metadata.update(meta)
         
+        integrity_verified = all(verify_checksum(entry) for entry in lineage_entries)
         chain_dicts = [entry.to_dict() for entry in lineage_entries]
         return {
             "entity_id": entity_id,
@@ -579,7 +580,8 @@ class ProvenanceManager:
                 default=None
             ),
             "entity_count": len(lineage_entries),
-            "metadata": aggregated_metadata  # Add metadata key
+            "metadata": aggregated_metadata,
+            "integrity_verified": integrity_verified,
         }
     
     def trace_lineage(self, entity_id: str) -> List[ProvenanceEntry]:

tests/explorer/test_provenance_manager_wiring.py

@@ -0,0 +1,202 @@
+"""
+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
+    assert all(e["direction"] == "upstream" for e in data["edges"])
+
+    # 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"
+
+    # Confirm markdown export classifies multi-hop ancestor edges under Upstream, not Lateral
+    md_response = client.get(
+        "/api/provenance/report", params={"node_id": "child", "format": "markdown"}
+    )
+    assert md_response.status_code == 200
+    md_text = md_response.text
+    assert "## Upstream" in md_text
+    assert "grandparent" in md_text and "parent" in md_text
+    assert "## Lateral" not in md_text
+
+
+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
+
+
+def test_create_app_rejects_conflicting_provenance_storage_path(tmp_path):
+    """Test that create_app raises ValueError when given a session with a conflicting preconfigured path."""
+    path1 = str(tmp_path / "orig.db")
+    path2 = str(tmp_path / "conflict.db")
+    sess = GraphSession(ContextGraph(advanced_analytics=False), provenance_storage_path=path1)
+    with pytest.raises(ValueError, match="Conflicting provenance_storage_path"):
+        create_app(session=sess, provenance_storage_path=path2)
+
+
+def test_create_app_rejects_path_when_manager_already_initialized(tmp_path):
+    """Test that create_app raises ValueError when given a session whose manager was already constructed."""
+    sess = GraphSession(ContextGraph(advanced_analytics=False))
+    _ = sess.provenance_manager  # construct and cache
+    with pytest.raises(ValueError, match="provenance_manager is already initialized"):
+        create_app(session=sess, provenance_storage_path=str(tmp_path / "late.db"))
+
+
+def test_create_app_allows_matching_provenance_storage_path(tmp_path):
+    """Test that create_app succeeds when the supplied path matches the session's existing path."""
+    path1 = str(tmp_path / "same.db")
+    sess = GraphSession(ContextGraph(advanced_analytics=False), provenance_storage_path=path1)
+    app = create_app(session=sess, provenance_storage_path=path1)
+    with TestClient(app):
+        assert app.state.session._provenance_storage_path == path1
+
+
+def test_provenance_manager_wiring_checksum_failure_falls_back(client, session):
+    """Test that if any lineage entry fails checksum verification, Explorer falls back to graph traversal."""
+    pm = session.provenance_manager
+    entry = pm.track_entity(entity_id="tampered_node", source="doc_1", entity_type="entity")
+    # Simulate tampering by altering confidence in storage without updating checksum
+    entry.confidence = 0.1
+    pm.storage.store(entry)
+
+    session.add_node("tampered_node", content="Tampered Node", node_type="entity")
+
+    response = client.get("/api/provenance", params={"node_id": "tampered_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"] == "tampered_node"
+
+
+def test_provenance_audit_evidence_fields_preserved(client, session):
+    """Test that audit evidence fields (source_document, confidence, checksum) are preserved in JSON and markdown."""
+    pm = session.provenance_manager
+    pm.track_entity(entity_id="ev_node", source="DOI:10.1234/test", entity_type="entity", metadata={"label": "Evidence Node"})
+
+    response = client.get("/api/provenance", params={"node_id": "ev_node"})
+    assert response.status_code == 200
+    data = response.json()
+    assert data["source"] == "audit"
+    assert len(data["nodes"]) == 1
+    node = data["nodes"][0]
+    assert node["source_document"] == "DOI:10.1234/test"
+    assert node["confidence"] == 1.0
+    assert node["checksum"] is not None
+
+    rep_json = client.get("/api/provenance/report", params={"node_id": "ev_node", "format": "json"}).json()
+    assert rep_json["lineage"]["nodes"][0]["source_document"] == "DOI:10.1234/test"
+
+    rep_md = client.get("/api/provenance/report", params={"node_id": "ev_node", "format": "markdown"}).text
+    assert "[source: DOI:10.1234/test]" in rep_md
+    assert "(confidence: 1.0)" in rep_md