graph: surface referrers + deletions in the query/CLI core (part of #205)

apps/python-app/libs/cssc_graph/cssc_graph/cli.py

@@ -181,9 +181,10 @@ def _emit_subgraph(subgraph: dict, output_format: str) -> None:
     click.echo(f"{len(subgraph['nodes'])} node(s), {len(subgraph['edges'])} edge(s):")
     for edge in subgraph["edges"]:
         tag = f" ({edge['tag']})" if edge.get("tag") else ""
+        atype = f" [{edge['artifactType']}]" if edge.get("artifactType") else ""
         frm = refs.get(edge["from"], edge["from"])
         to = refs.get(edge["to"], edge["to"])
-        click.echo(f"  {frm}  --{edge['type']}-->  {to}{tag}")
+        click.echo(f"  {frm}  --{edge['type']}-->  {to}{tag}{atype}")
 
 
 @cli.command()
@@ -266,6 +267,27 @@ def derived(database: Path, base: str, depth: int, output_format: str) -> None:
     _emit_subgraph(subgraph, output_format)
 
 
+@cli.command()
+@_database_option
+@click.option("--digest", help="Seed by artifact digest (sha256:...).")
+@click.option("--ref", help="Seed by registry/repository[@digest|:tag].")
+@click.option("--depth", default=3, show_default=True, help="Max referrer depth (referrers-of-referrers).")
+@_format_option
+def referrers(database: Path, digest: str | None, ref: str | None, depth: int, output_format: str) -> None:
+    """List the referrer artifacts (SBOM/provenance/VEX/signatures) attached to an image."""
+
+    if not digest and not ref:
+        raise click.UsageError("provide --digest or --ref")
+    from . import queries
+
+    store = _open_store(database)
+    try:
+        subgraph = queries.referrers(store, digest=digest, ref=ref, depth=depth)
+    finally:
+        store.close()
+    _emit_subgraph(subgraph, output_format)
+
+
 @cli.command()
 @_database_option
 @click.option("--annotation", help="Filter by annotation name=value.")
@@ -319,13 +341,18 @@ def show(database: Path, ref: str, output_format: str) -> None:
         return
     occ = data["occurrence"]
     click.echo(f"occurrence: {occ['key']}")
+    if data.get("deleted"):
+        reason = occ.get("deleteReason") or "unknown"
+        click.echo(f"  deleted: {occ.get('deletedAt')} (reason: {reason})")
     if data["artifact"]:
         art = data["artifact"]
         click.echo(f"  type: {art.get('artifactType') or art.get('mediaType') or '-'}")
     for ann in data["annotations"]:
         click.echo(f"  annotation: {ann['name']}={ann['value']}")
     for tag in data["tags"]:
         click.echo(f"  tag: {tag['tag']} @ {tag['observedAt']}")
+    for r in data.get("referrers", []):
+        click.echo(f"  referrer: {r.get('artifactType') or '-'} ({r['from']})")
 
 
 @cli.command()

apps/python-app/libs/cssc_graph/cssc_graph/queries.py

@@ -17,15 +17,17 @@
 
 _OCC_RETURN = (
     "n.key AS key, n.registry AS registry, n.repository AS repository, "
-    "n.digest AS digest, n.ref AS ref"
+    "n.digest AS digest, n.ref AS ref, n.deletedAt AS deletedAt, "
+    "n.deleteReason AS deleteReason"
 )
 
 
 def get_occurrence(store: GraphStore, key: str) -> dict[str, Any] | None:
     rows = store.query(
         "MATCH (o:Occurrence {key: $k}) "
         "RETURN o.key AS key, o.registry AS registry, o.repository AS repository, "
-        "o.digest AS digest, o.ref AS ref",
+        "o.digest AS digest, o.ref AS ref, o.deletedAt AS deletedAt, "
+        "o.deleteReason AS deleteReason",
         {"k": key},
     )
     return rows[0] if rows else None
@@ -136,6 +138,8 @@ def _record_node(nodes: dict[str, dict[str, Any]], row: dict[str, Any]) -> None:
             "repository": row["repository"],
             "digest": row["digest"],
             "ref": row["ref"],
+            "deletedAt": row.get("deletedAt"),
+            "deleteReason": row.get("deleteReason"),
         }
 
 
@@ -168,6 +172,72 @@ def derived(store: GraphStore, base: str, depth: int = 10) -> dict[str, Any]:
     return traverse(store, seeds, ("BUILT_FROM",), direction="in", max_depth=depth)
 
 
+def referrers(
+    store: GraphStore,
+    *,
+    digest: str | None = None,
+    ref: str | None = None,
+    depth: int = 3,
+) -> dict[str, Any]:
+    """Referrer artifacts of a subject occurrence.
+
+    Returns ``{nodes, edges}`` where each ``REFERS_TO`` edge carries the
+    ``artifactType``. Follows referrers-of-referrers up to *depth* levels (a
+    signature on an SBOM, etc.); ``depth=0`` returns just the subject, matching
+    :func:`traverse`.
+    """
+
+    seeds = resolve_seed(store, digest=digest, ref=ref)
+    nodes: dict[str, dict[str, Any]] = {}
+    edges: list[dict[str, Any]] = []
+    # Keyed like the indexer's REFERS_TO merge so distinct observations and
+    # artifact types between the same pair are all kept, not collapsed.
+    seen_edges: set[tuple[str, str, str | None, str | None]] = set()
+    for seed in seeds:
+        occ = get_occurrence(store, seed)
+        if occ:
+            nodes[seed] = occ
+
+    visited: set[str] = set()
+    frontier = list(dict.fromkeys(seeds))
+    for _ in range(depth):
+        nxt: list[str] = []
+        for key in frontier:
+            if key in visited:
+                continue
+            visited.add(key)
+            rows = store.query(
+                "MATCH (r:Occurrence)-[e:REFERS_TO]->(s:Occurrence {key: $k}) "
+                "RETURN r.key AS key, r.registry AS registry, r.repository AS repository, "
+                "r.digest AS digest, r.ref AS ref, r.deletedAt AS deletedAt, "
+                "r.deleteReason AS deleteReason, e.artifactType AS artifactType, "
+                "e.observedAt AS observedAt ORDER BY e.artifactType, r.key",
+                {"k": key},
+            )
+            for row in rows:
+                _record_node(nodes, row)
+                atype = row.get("artifactType")
+                observed = row.get("observedAt")
+                ekey = (row["key"], key, atype, observed)
+                if ekey not in seen_edges:
+                    seen_edges.add(ekey)
+                    edges.append(
+                        {
+                            "type": "REFERS_TO",
+                            "from": row["key"],
+                            "to": key,
+                            "artifactType": atype,
+                            "observedAt": observed,
+                        }
+                    )
+                nxt.append(row["key"])
+        frontier = [k for k in nxt if k not in visited]
+        if not frontier:
+            break
+
+    return {"nodes": list(nodes.values()), "edges": edges}
+
+
 def tag_history(store: GraphStore, ref: str, tag: str) -> list[dict[str, Any]]:
     registry, repository = split_ref(ref)
     return store.query(
@@ -238,17 +308,31 @@ def show(store: GraphStore, ref: str) -> dict[str, Any] | None:
         "artifact": artifact[0] if artifact else None,
         "annotations": annotations,
         "tags": tags,
+        "referrers": referrers(store, ref=key)["edges"],
+        "deleted": bool(occ.get("deletedAt")),
         "path": traverse(store, [key], PATH_RELS, direction="both", max_depth=2),
     }
 
 
 # -- export -------------------------------------------------------------------
 
 
+def _node_label(node: dict[str, Any]) -> str:
+    """A label that disambiguates occurrences of the same repository by digest."""
+    ref = node.get("ref") or node["key"]
+    digest = node.get("digest") or ""
+    short = digest[7:19] if digest.startswith("sha256:") else digest[:12]
+    label = f"{ref}@{short}" if short else str(ref)
+    if node.get("deletedAt"):
+        label += " (deleted)"
+    return label
+
+
 def to_cytoscape(subgraph: dict[str, Any]) -> dict[str, Any]:
-    elements = [{"data": {"id": n["key"], "label": n.get("ref", n["key"]), **n}} for n in subgraph["nodes"]]
+    elements = [{"data": {"id": n["key"], "label": _node_label(n), **n}} for n in subgraph["nodes"]]
     for i, e in enumerate(subgraph["edges"]):
-        elements.append({"data": {"id": f"e{i}", "source": e["from"], "target": e["to"], "label": e["type"], **e}})
+        label = e.get("artifactType") or e["type"]
+        elements.append({"data": {"id": f"e{i}", "source": e["from"], "target": e["to"], "label": label, **e}})
     return {"elements": elements}
 
 
@@ -261,10 +345,11 @@ def to_mermaid(subgraph: dict[str, Any]) -> str:
     lines = ["flowchart LR"]
     ids = {n["key"]: f"n{i}" for i, n in enumerate(subgraph["nodes"])}
     for n in subgraph["nodes"]:
-        label = _mermaid_label(n.get("ref", n["key"]))
+        label = _mermaid_label(_node_label(n))
         lines.append(f'    {ids[n["key"]]}["{label}"]')
     for e in subgraph["edges"]:
         frm, to = ids.get(e["from"]), ids.get(e["to"])
         if frm and to:
-            lines.append(f'    {frm} -->|{e["type"]}| {to}')
+            rel = _mermaid_label(e.get("artifactType") or e["type"])
+            lines.append(f'    {frm} -->|{rel}| {to}')
     return "\n".join(lines)

apps/python-app/libs/cssc_graph/tests/test_queries.py

@@ -94,3 +94,54 @@ def test_traverse_includes_seed_at_depth_zero(store):
 
 def test_mermaid_label_escaping():
     assert queries._mermaid_label('a"b\nc') == "a#quot;b c"
+
+
+DIGEST3 = "sha256:" + "3" * 64
+QUAR_REF = "ghcr.io/toddysm/quarantine/python"
+
+
+def test_referrers_returns_edge_with_artifact_type(store):
+    sub = queries.referrers(store, ref=f"{GOLDEN_REF}@{DIGEST2}")
+    assert "REFERS_TO" in {e["type"] for e in sub["edges"]}
+    assert "application/vnd.in-toto+json" in {e.get("artifactType") for e in sub["edges"]}
+
+
+def test_referrers_depth_zero_returns_only_subject(store):
+    sub = queries.referrers(store, ref=f"{GOLDEN_REF}@{DIGEST2}", depth=0)
+    assert sub["edges"] == []
+    assert {n["key"] for n in sub["nodes"]} == {f"{GOLDEN_REF}@{DIGEST2}"}
+
+
+def test_show_includes_referrers(store):
+    data = queries.show(store, f"{GOLDEN_REF}@{DIGEST2}")
+    assert data is not None
+    assert data["deleted"] is False
+    assert any(r.get("artifactType") == "application/vnd.in-toto+json" for r in data["referrers"])
+
+
+def test_show_marks_deleted_occurrence(store):
+    data = queries.show(store, f"{QUAR_REF}@{DIGEST3}")
+    assert data is not None
+    assert data["deleted"] is True
+    assert data["occurrence"]["deleteReason"] == "promoted"
+
+
+def test_node_label_disambiguates_by_digest_and_marks_deleted():
+    assert queries._node_label({"ref": GOLDEN_REF, "digest": DIGEST2}) == f"{GOLDEN_REF}@222222222222"
+    lbl = queries._node_label({"ref": QUAR_REF, "digest": DIGEST3, "deletedAt": "2026-08-09T13:10:10Z"})
+    assert lbl.endswith("(deleted)")
+
+
+def test_cli_referrers(tmp_path: Path):
+    from click.testing import CliRunner
+
+    from cssc_graph.cli import cli
+
+    db = tmp_path / "db"
+    with GraphStore(db) as gs:
+        gs.init_schema()
+        index_data(gs, EXAMPLES, SCHEMA_DIR)
+    result = CliRunner().invoke(cli, ["referrers", "-d", str(db), "--ref", f"{GOLDEN_REF}@{DIGEST2}"])
+    assert result.exit_code == 0, result.output
+    assert "REFERS_TO" in result.output
+    assert "application/vnd.in-toto+json" in result.output