fix(provenance): include upstream ancestors + add direction classification and markdown grouping

.gitignore

@@ -110,3 +110,7 @@ sample_data/
 
 # Test Results
 test_results.txt
+
+# Frontend workspace artifacts
+semantica-explorer/
+node_modules/

semantica/explorer/app.py

@@ -85,6 +85,7 @@ def create_app(session: Optional[GraphSession] = None) -> FastAPI:
     from .routes.enrich import router as enrich_router
     from .routes.export_import import router as export_import_router
     from .routes.annotations import router as annotations_router
+    from .routes.provenance import router as provenance_router
 
     app.include_router(graph_router)
     app.include_router(analytics_router)
@@ -93,6 +94,7 @@ def create_app(session: Optional[GraphSession] = None) -> FastAPI:
     app.include_router(enrich_router)
     app.include_router(export_import_router)
     app.include_router(annotations_router)
+    app.include_router(provenance_router)
 
     from fastapi import WebSocket, WebSocketDisconnect
 

semantica/explorer/routes/provenance.py

@@ -0,0 +1,201 @@
+"""
+Provenance routes for lineage visualization and exportable reports.
+"""
+
+import asyncio
+import json
+from typing import Any, Dict, List, Optional
+
+import networkx as nx
+from fastapi import APIRouter, Depends, Query
+from fastapi.responses import PlainTextResponse, Response
+from pydantic import BaseModel
+
+from ..dependencies import get_session
+from ..session import GraphSession
+
+router = APIRouter(prefix="/api/provenance", tags=["Power User Tools"])
+
+
+class ProvenanceNode(BaseModel):
+    id: str
+    label: str
+    prov_type: str
+    parent_id: str
+
+
+class ProvenanceEdge(BaseModel):
+    id: str
+    source: str
+    target: str
+    label: str
+    direction: str  # "upstream" | "downstream" | "lateral"
+
+
+class ProvenanceResponse(BaseModel):
+    nodes: List[ProvenanceNode]
+    edges: List[ProvenanceEdge]
+
+
+_AGENT_TYPES = {"person", "organization", "system", "agent"}
+_ACTIVITY_TYPES = {"action", "event", "process", "activity", "decision", "publication"}
+
+
+def _classify_prov(node_type: str) -> tuple[str, str]:
+    lowered = node_type.lower()
+    if lowered in _AGENT_TYPES:
+        return "Agent", "group_agent"
+    if lowered in _ACTIVITY_TYPES:
+        return "Activity", "group_activity"
+    return "Entity", "group_entity"
+
+
+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": []}
+
+    graph = nx.DiGraph()
+    graph.add_node(node_id)
+
+    hop_nodes = {node_id}
+    for edge in session.graph.edges:
+        if edge.source_id == node_id or edge.target_id == node_id:
+            graph.add_edge(edge.source_id, edge.target_id, label=edge.edge_type)
+            hop_nodes.add(edge.source_id)
+            hop_nodes.add(edge.target_id)
+
+    for edge in session.graph.edges:
+        if edge.source_id in hop_nodes or edge.target_id in hop_nodes:
+            graph.add_edge(edge.source_id, edge.target_id, label=edge.edge_type)
+
+    subgraph = nx.ego_graph(graph, node_id, radius=2, undirected=True)
+    provenance_nodes: List[Dict[str, Any]] = []
+    for graph_node_id in subgraph.nodes():
+        node = session.graph.nodes.get(graph_node_id)
+        if node is None:
+            continue
+        prov_type, parent_id = _classify_prov(node.node_type)
+        provenance_nodes.append(
+            {
+                "id": graph_node_id,
+                "label": node.content or graph_node_id,
+                "prov_type": prov_type,
+                "parent_id": parent_id,
+            }
+        )
+
+    provenance_edges: List[Dict[str, Any]] = []
+    for source, target, data in subgraph.edges(data=True):
+        if target == node_id:
+            direction = "upstream"
+        elif source == node_id:
+            direction = "downstream"
+        else:
+            direction = "lateral"
+
+        provenance_edges.append(
+            {
+                "id": f"{source}-{target}",
+                "source": source,
+                "target": target,
+                "label": data.get("label", "related_to"),
+                "direction": direction,
+            }
+        )
+
+    return {"nodes": provenance_nodes, "edges": provenance_edges}
+
+
+def _build_report(session: GraphSession, node_id: str) -> Dict[str, Any]:
+    node = session.get_node(node_id)
+    provenance = _build_provenance(session, node_id)
+    return {
+        "node_id": node_id,
+        "label": node.get("content", node_id) if node else node_id,
+        "type": node.get("type", "entity") if node else "entity",
+        "properties": node.get("properties", {}) if node else {},
+        "lineage": provenance,
+    }
+
+
+def _render_markdown(report: Dict[str, Any]) -> str:
+    lines = [
+        f"# Provenance Report: {report['label']}",
+        "",
+        f"- Node ID: `{report['node_id']}`",
+        f"- Type: `{report['type']}`",
+        "",
+        "## Properties",
+    ]
+    properties = report.get("properties", {})
+    if properties:
+        for key, value in properties.items():
+            lines.append(f"- **{key}**: {value}")
+    else:
+        lines.append("- No properties recorded")
+
+    lines.extend(["", "## Lineage Nodes"])
+    for node in report.get("lineage", {}).get("nodes", []):
+        lines.append(f"- `{node['id']}` ({node['prov_type']}): {node['label']}")
+
+    edges = report.get("lineage", {}).get("edges", [])
+    grouped_edges = {
+        "upstream": [],
+        "downstream": [],
+        "lateral": [],
+    }
+    for edge in edges:
+        direction = edge.get("direction", "lateral")
+        grouped_edges.setdefault(direction, []).append(edge)
+
+    if grouped_edges.get("upstream"):
+        lines.extend(["", "## Upstream"])
+        for edge in grouped_edges["upstream"]:
+            lines.append(f"- `{edge['source']}` -[{edge['label']}]-> `{edge['target']}`")
+
+    if grouped_edges.get("downstream"):
+        lines.extend(["", "## Downstream"])
+        for edge in grouped_edges["downstream"]:
+            lines.append(f"- `{edge['source']}` -[{edge['label']}]-> `{edge['target']}`")
+
+    if grouped_edges.get("lateral"):
+        lines.extend(["", "## Lateral"])
+        for edge in grouped_edges["lateral"]:
+            lines.append(f"- `{edge['source']}` -[{edge['label']}]-> `{edge['target']}`")
+
+    return "\n".join(lines)
+
+
+@router.get("", response_model=ProvenanceResponse)
+@router.get("/", response_model=ProvenanceResponse, include_in_schema=False)
+async def get_provenance_lineage(
+    node_id: Optional[str] = None,
+    session: GraphSession = Depends(get_session),
+):
+    data = await asyncio.to_thread(_build_provenance, session, node_id)
+    return ProvenanceResponse(
+        nodes=[ProvenanceNode(**node) for node in data["nodes"]],
+        edges=[ProvenanceEdge(**edge) for edge in data["edges"]],
+    )
+
+
+@router.get("/report")
+async def export_provenance_report(
+    node_id: str = Query(..., description="Node ID to export"),
+    format: str = Query("json", description="json or markdown"),
+    session: GraphSession = Depends(get_session),
+):
+    report = await asyncio.to_thread(_build_report, session, node_id)
+    if format.lower() in {"md", "markdown"}:
+        content = _render_markdown(report)
+        return PlainTextResponse(
+            content,
+            headers={"Content-Disposition": f'attachment; filename="{node_id}_provenance.md"'},
+        )
+
+    content = json.dumps(report, indent=2, default=str)
+    return Response(
+        content=content,
+        media_type="application/json",
+        headers={"Content-Disposition": f'attachment; filename="{node_id}_provenance.json"'},
+    )

tests/explorer/test_provenance_route.py

@@ -0,0 +1,74 @@
+"""Unit tests for explorer provenance route helpers."""
+
+from types import SimpleNamespace
+
+from semantica.explorer.routes.provenance import _build_provenance, _render_markdown
+
+
+def _make_session_with_chain() -> SimpleNamespace:
+    """Build a minimal session-like object for Source -> Intermediate -> node_id."""
+    nodes = {
+        "Source": SimpleNamespace(node_type="entity", content="Source"),
+        "Intermediate": SimpleNamespace(node_type="entity", content="Intermediate"),
+        "node_id": SimpleNamespace(node_type="entity", content="Target"),
+    }
+    edges = [
+        SimpleNamespace(source_id="Source", target_id="Intermediate", edge_type="related_to"),
+        SimpleNamespace(source_id="Intermediate", target_id="node_id", edge_type="related_to"),
+    ]
+    graph = SimpleNamespace(nodes=nodes, edges=edges)
+    return SimpleNamespace(graph=graph)
+
+
+def test_build_provenance_direction_classification_chain():
+    session = _make_session_with_chain()
+
+    data = _build_provenance(session, "node_id")
+
+    node_ids = {node["id"] for node in data["nodes"]}
+    assert "Source" in node_ids
+    assert "Intermediate" in node_ids
+
+    edge_by_pair = {(edge["source"], edge["target"]): edge for edge in data["edges"]}
+
+    assert edge_by_pair[("Intermediate", "node_id")]["direction"] == "upstream"
+    assert edge_by_pair[("Source", "Intermediate")]["direction"] != "downstream"
+
+
+def test_render_markdown_groups_edges_by_direction():
+    report = {
+        "node_id": "node_id",
+        "label": "Target",
+        "type": "entity",
+        "properties": {},
+        "lineage": {
+            "nodes": [
+                {"id": "Source", "prov_type": "Entity", "label": "Source"},
+                {"id": "Intermediate", "prov_type": "Entity", "label": "Intermediate"},
+                {"id": "node_id", "prov_type": "Entity", "label": "Target"},
+            ],
+            "edges": [
+                {
+                    "id": "Intermediate-node_id",
+                    "source": "Intermediate",
+                    "target": "node_id",
+                    "label": "related_to",
+                    "direction": "upstream",
+                },
+                {
+                    "id": "Source-Intermediate",
+                    "source": "Source",
+                    "target": "Intermediate",
+                    "label": "related_to",
+                    "direction": "lateral",
+                },
+            ],
+        },
+    }
+
+    markdown = _render_markdown(report)
+
+    assert "## Upstream" in markdown
+    assert "## Lateral" in markdown
+    assert "`Intermediate` -[related_to]-> `node_id`" in markdown
+    assert "`Source` -[related_to]-> `Intermediate`" in markdown