graph-service: GET /artifacts/provenance endpoint

apps/python-app/services/graph-service/src/graph_service/app.py

@@ -131,6 +131,19 @@ def artifact_referrers(
             return Response(content=queries.to_mermaid(subgraph), media_type="text/plain")
         return subgraph
 
+    @app.get("/artifacts/provenance")
+    def artifact_provenance(
+        family: str = Query(..., description="Repository family (trailing image name, e.g. python)."),
+        format: str = Query("json", pattern="^(json|cytoscape|mermaid)$"),
+    ) -> Any:
+        with reading() as store:
+            subgraph = queries.provenance(store, family)
+        if format == "cytoscape":
+            return queries.to_cytoscape(subgraph)
+        if format == "mermaid":
+            return Response(content=queries.to_mermaid(subgraph), media_type="text/plain")
+        return subgraph
+
     @app.get("/repositories/tags/history")
     def tag_history(ref: str = Query(...), tag: str = Query(...)) -> dict[str, Any]:
         with reading() as store:

apps/python-app/services/graph-service/tests/test_app.py

@@ -230,6 +230,25 @@ def test_referrers_endpoint_accepts_no_rollup(client: TestClient) -> None:
     )
 
 
+def test_provenance_endpoint_json(client: TestClient) -> None:
+    body = client.get("/artifacts/provenance", params={"family": "python"}).json()
+    assert body["nodes"] and body["edges"]
+    assert any(n.get("state") == "present-golden" for n in body["nodes"])
+    assert "built" in {e["type"] for e in body["edges"]}
+
+
+def test_provenance_endpoint_mermaid(client: TestClient) -> None:
+    resp = client.get("/artifacts/provenance", params={"family": "python", "format": "mermaid"})
+    assert resp.status_code == 200
+    assert resp.text.strip().startswith("flowchart")
+
+
+def test_provenance_endpoint_unknown_family_is_empty(client: TestClient) -> None:
+    body = client.get("/artifacts/provenance", params={"family": "does-not-exist"}).json()
+    assert body == {"nodes": [], "edges": []}
+
+
+
 def test_depth_is_capped(client: TestClient, tmp_path: Path) -> None:
     # A max_depth of 1 must stop bases traversal at the first hop.
     root = _write_data_root(tmp_path / "capped", RECORDS)