From ca369ad8690f8aaf0b18fe9d0f3f4efe6173c1a7 Mon Sep 17 00:00:00 2001
From: Toddy Mladenov <toddysm@gmail.com>
Date: Fri, 28 Aug 2026 21:10:52 -0700
Subject: [PATCH] graph: add cssc-graph provenance CLI command (#213)

---
 .../libs/cssc_graph/cssc_graph/cli.py         | 55 +++++++++++++++++++
 .../libs/cssc_graph/tests/test_queries.py     | 52 ++++++++++++++++++
 2 files changed, 107 insertions(+)

diff --git a/apps/python-app/libs/cssc_graph/cssc_graph/cli.py b/apps/python-app/libs/cssc_graph/cssc_graph/cli.py
index 721d4b2..702faf2 100644
--- a/apps/python-app/libs/cssc_graph/cssc_graph/cli.py
+++ b/apps/python-app/libs/cssc_graph/cssc_graph/cli.py
@@ -363,6 +363,61 @@ def show(database: Path, ref: str, output_format: str) -> None:
         click.echo(f"  referrer: {r.get('artifactType') or '-'} ({r['from']}){plat}")
 
 
+def _format_provenance_text(subgraph: dict, family: str) -> str:
+    from . import queries
+
+    lines = [f"family: {family}", f"{len(subgraph['nodes'])} node(s), {len(subgraph['edges'])} edge(s)"]
+    labels: dict[str, str] = {}
+    for n in subgraph["nodes"]:
+        label = queries._node_label(n)
+        labels[n["key"]] = label
+        lines.append(f"  [{n.get('nodeType', '?')}] {label} ({n.get('state', '')})")
+    for e in subgraph["edges"]:
+        date = str(e.get("date") or "")[:10]
+        frm = labels.get(e["from"], e["from"])
+        to = labels.get(e["to"], e["to"])
+        lines.append(f"  {e['type']:<9} {frm} -> {to}  {date}".rstrip())
+    return "\n".join(lines)
+
+
+@cli.command()
+@_database_option
+@click.option("--family", required=True, help="Repository family (trailing image name, e.g. python).")
+@click.option(
+    "--format",
+    "output_format",
+    type=click.Choice(["text", "json", "mermaid", "cytoscape"]),
+    default="text",
+    show_default=True,
+)
+@click.option("--output", "-o", type=click.Path(path_type=Path), default=None, help="Write to a file.")
+def provenance(database: Path, family: str, output_format: str, output: Path | None) -> None:
+    """Show a per-family provenance timeline (imported -> promoted -> built, with referrers)."""
+
+    from . import queries
+
+    store = _open_store(database)
+    try:
+        subgraph = queries.provenance(store, family)
+    finally:
+        store.close()
+
+    if output_format == "json":
+        text = json.dumps(subgraph, indent=2)
+    elif output_format == "mermaid":
+        text = queries.to_mermaid(subgraph)
+    elif output_format == "cytoscape":
+        text = json.dumps(queries.to_cytoscape(subgraph), indent=2)
+    else:
+        text = _format_provenance_text(subgraph, family)
+
+    if output is not None:
+        output.write_text(text + "\n", encoding="utf-8")
+        click.echo(f"Wrote {output_format} provenance to {output}.")
+    else:
+        click.echo(text)
+
+
 @cli.command()
 @_database_option
 @click.option("--digest", help="Seed by artifact digest.")
diff --git a/apps/python-app/libs/cssc_graph/tests/test_queries.py b/apps/python-app/libs/cssc_graph/tests/test_queries.py
index 9618d44..b6eb1be 100644
--- a/apps/python-app/libs/cssc_graph/tests/test_queries.py
+++ b/apps/python-app/libs/cssc_graph/tests/test_queries.py
@@ -288,3 +288,55 @@ def test_non_provenance_subgraph_renders_without_classdefs(store):
     assert "classDef" not in m
 
 
+def _index_db(tmp_path: Path) -> Path:
+    db = tmp_path / "db"
+    with GraphStore(db) as gs:
+        gs.init_schema()
+        index_data(gs, EXAMPLES, SCHEMA_DIR)
+    return db
+
+
+def test_cli_provenance_text(tmp_path: Path):
+    from click.testing import CliRunner
+
+    from cssc_graph.cli import cli
+
+    db = _index_db(tmp_path)
+    result = CliRunner().invoke(cli, ["provenance", "-d", str(db), "--family", "python"])
+    assert result.exit_code == 0, result.output
+    assert "family: python" in result.output
+    assert "[tag-root]" in result.output
+    assert "imported" in result.output
+
+
+def test_cli_provenance_mermaid(tmp_path: Path):
+    from click.testing import CliRunner
+
+    from cssc_graph.cli import cli
+
+    db = _index_db(tmp_path)
+    result = CliRunner().invoke(
+        cli, ["provenance", "-d", str(db), "--family", "python", "--format", "mermaid"]
+    )
+    assert result.exit_code == 0, result.output
+    assert result.output.strip().startswith("flowchart")
+    assert "classDef" in result.output
+
+
+def test_cli_provenance_json(tmp_path: Path):
+    import json
+
+    from click.testing import CliRunner
+
+    from cssc_graph.cli import cli
+
+    db = _index_db(tmp_path)
+    result = CliRunner().invoke(
+        cli, ["provenance", "-d", str(db), "--family", "python", "--format", "json"]
+    )
+    assert result.exit_code == 0, result.output
+    data = json.loads(result.output)
+    assert data["nodes"] and data["edges"]
+
+
+