diff --git a/CHANGELOG.md b/CHANGELOG.md index c33d626a..f6a7e809 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/semantica/explorer/app.py b/semantica/explorer/app.py index 96d66b14..c214021a 100644 --- a/semantica/explorer/app.py +++ b/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): diff --git a/semantica/explorer/routes/provenance.py b/semantica/explorer/routes/provenance.py index 8b1b1f03..4a6ac5da 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 @@ -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"), ) diff --git a/semantica/explorer/schemas.py b/semantica/explorer/schemas.py index 226cd41e..638fa82e 100644 --- a/semantica/explorer/schemas.py +++ b/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 # --------------------------------------------------------------------------- diff --git a/semantica/explorer/session.py b/semantica/explorer/session.py index b2ef8c57..8692f59e 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,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 {}) diff --git a/semantica/provenance/integrity.py b/semantica/provenance/integrity.py index 5680d728..e5327089 100644 --- a/semantica/provenance/integrity.py +++ b/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 diff --git a/semantica/provenance/manager.py b/semantica/provenance/manager.py index ab0abb26..9e2679bf 100644 --- a/semantica/provenance/manager.py +++ b/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]: diff --git a/tests/explorer/test_provenance_manager_wiring.py b/tests/explorer/test_provenance_manager_wiring.py new file mode 100644 index 00000000..282a48cc --- /dev/null +++ b/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