yashraj4 · Sep 24, 2026
diff --git a/src/codegraphcontext/cli/main.py b/src/codegraphcontext/cli/main.py
@@ -47,6 +47,8 @@
 )
 from .hook_manager import HookError, get_hook_status, install_hooks, uninstall_hooks
 from codegraphcontext.utils.tool_limits import get_tool_result_limit
+from codegraphcontext.core import graph_snapshot
+from . import snapshot_commands
 
 # Set the log level for the noisy neo4j, asyncio, and urllib3 loggers to keep the output clean.
 # Get the log level from config, defaulting to WARNING
@@ -1078,6 +1080,155 @@ def _read_bytes(p: Path) -> bytes:
     )
     raise typer.Exit(code=0)
 
+# ============================================================================
+# SNAPSHOT COMMAND GROUP - Save/list index snapshots for `cgc diff` (#1312)
+# ============================================================================
+
+snapshot_app = typer.Typer(help="Save and list graph snapshots for `cgc diff`")
+app.add_typer(snapshot_app, name="snapshot")
+
+
+def _init_snapshot_services(context: Optional[str]):
+    """Resolve credentials and services with stdout reserved for the product.
+
+    `cli_helpers` prints its init chatter on stdout, and every command in this
+    section writes its result there, so the chatter is rerouted to stderr for
+    the init window only and restored afterwards — a permanent rebind leaks a
+    closed stream into later in-process invocations (CliRunner-based tests).
+    """
+    import sys as _sys
+    from . import cli_helpers as _cli_helpers
+    # NB: the .file property GETTER materializes the current (possibly
+    # transient, CliRunner-captured) stream; the dynamic state lives in
+    # ._file, which is None while the console follows sys.stdout live.
+    _prev_console_file = _cli_helpers.console._file
+    _cli_helpers.console.file = _sys.stderr
+    try:
+        _load_credentials()
+        services = _initialize_services(context)
+    finally:
+        _cli_helpers.console._file = _prev_console_file
+    if not all(services[:3]):
+        raise typer.Exit(code=1)
+    return services
+
+
+def _snapshot_fail(exc: Exception):
+    """Render a snapshot/diff failure on stderr and exit: 2 for bad input, 1 otherwise."""
+    console.print(f"[bold red]Error:[/bold red] {exc}")
+    raise typer.Exit(code=2 if isinstance(exc, graph_snapshot.SnapshotUsageError) else 1)
+
+
+@snapshot_app.command("save")
+def snapshot_save(
+    name: Optional[str] = typer.Option(None, "--name", "-n", help="Snapshot name (default: snapshot-YYYYmmdd-HHMMSS)"),
+    force: bool = typer.Option(False, "--force", "-f", help="Overwrite an existing snapshot with this name"),
+    context: Optional[str] = typer.Option(None, "--context", "-c", help="Specific context to use"),
+    json_output: bool = typer.Option(False, "--json", help="Print a JSON receipt on stdout instead of the summary line"),
+):
+    """
+    Save the current index state as a lightweight snapshot.
+
+    The snapshot is a node/edge manifest with content digests — enough for
+    `cgc diff` to say what changed later, small enough to keep around.
+
+    Example:
+        cgc snapshot save --name before-refactor
+        cgc snapshot save --name ci-42 --json > snapshot.json
+    """
+    try:
+        snapshot_name = (
+            graph_snapshot.validate_name(name) if name else graph_snapshot.default_snapshot_name()
+        )
+    except graph_snapshot.SnapshotUsageError as exc:
+        _snapshot_fail(exc)
+    services = _init_snapshot_services(context)
+    db_manager, _graph_builder, _code_finder, resolved_context = services
+    try:
+        snapshot_commands.run_snapshot_save(
+            db_manager,
+            name=snapshot_name,
+            force=force,
+            ctx=resolved_context,
+            as_json=json_output,
+        )
+    except graph_snapshot.SnapshotError as exc:
+        _snapshot_fail(exc)
+    finally:
+        db_manager.close_driver()
+
+
+@snapshot_app.command("list")
+def snapshot_list(
+    json_output: bool = typer.Option(False, "--json", help="Print the snapshot inventory as JSON on stdout"),
+):
+    """
+    List snapshots saved by `cgc snapshot save`.
+
+    Newest first. Does not open a database connection.
+
+    Example:
+        cgc snapshot list
+        cgc snapshot list --json | jq -r '.snapshots[].name'
+    """
+    try:
+        snapshot_commands.run_snapshot_list(as_json=json_output)
+    except graph_snapshot.SnapshotError as exc:
+        _snapshot_fail(exc)
+
+
+@app.command("diff")
+def diff_command(
+    against: Optional[str] = typer.Option(None, "--against", "-a", help="Snapshot to compare against (default: most recent snapshot)"),
+    json_output: bool = typer.Option(False, "--json", help="Print the full diff document as JSON on stdout"),
+    limit: Optional[int] = typer.Option(None, "--limit", "-l", help="Maximum change lines to print (text output only; JSON is never truncated)"),
+    fail_on_changes: bool = typer.Option(False, "--fail-on-changes", help="Exit 1 when the diff is non-empty (CI gate)"),
+    context: Optional[str] = typer.Option(None, "--context", "-c", help="Specific context to use"),
+):
+    """
+    Show what changed in the graph since a snapshot was saved.
+
+    Exit codes: 0 = no differences (or no gate requested), 1 = differences
+    found with --fail-on-changes, or a snapshot/initialisation error,
+    2 = bad usage.
+
+    Example:
+        cgc snapshot save --name before-refactor
+        # ... edit code, re-index ...
+        cgc diff --against before-refactor
+        cgc diff --against before-refactor --json > changes.json
+        cgc diff --fail-on-changes --json
+    """
+    if against:
+        try:
+            against = graph_snapshot.validate_name(against)
+        except graph_snapshot.SnapshotUsageError as exc:
+            _snapshot_fail(exc)
+    if limit is not None and limit < 0:
+        _snapshot_fail(graph_snapshot.SnapshotUsageError("--limit must be zero or greater"))
+    try:
+        base = snapshot_commands.load_base_snapshot(against)
+    except graph_snapshot.SnapshotError as exc:
+        _snapshot_fail(exc)
+
+    services = _init_snapshot_services(context)
+    db_manager, _graph_builder, _code_finder, resolved_context = services
+    try:
+        changed = snapshot_commands.run_diff(
+            db_manager,
+            base=base,
+            ctx=resolved_context,
+            as_json=json_output,
+            limit=limit,
+        )
+    except graph_snapshot.SnapshotError as exc:
+        _snapshot_fail(exc)
+    finally:
+        db_manager.close_driver()
+
+    if changed and fail_on_changes:
+        raise typer.Exit(code=1)
+
 # ============================================================================
 # HOOK COMMAND GROUP - Git integration
 # ============================================================================

diff --git a/src/codegraphcontext/cli/snapshot_commands.py b/src/codegraphcontext/cli/snapshot_commands.py
@@ -0,0 +1,130 @@
+# src/codegraphcontext/cli/snapshot_commands.py
+"""`cgc snapshot save|list` and `cgc diff` (#1312).
+
+This module owns the operational half of the feature: where snapshots live,
+which one `--against` resolves to, and what each command prints. The Typer
+surface (option parsing, service init, stream rebinding, exit codes) stays in
+``cli/main.py``, and all document/diff/render logic stays in
+``core/graph_snapshot.py``.
+
+Stream contract (#1725): the *product* of every command here is written with
+plain ``print()`` so it lands on stdout — `cgc diff --json > changes.json` must
+yield a parseable file. Diagnostics are raised as exceptions and rendered on
+the stderr console by the caller.
+"""
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any, Dict, Optional
+
+from ..core.graph_snapshot import (
+    Snapshot,
+    SnapshotError,
+    build_snapshot,
+    diff_snapshots,
+    dump_json,
+    list_snapshots,
+    load_snapshot,
+    render_diff_text,
+    render_snapshot_list_text,
+    save_result_document,
+    snapshot_list_document,
+)
+from ..core.graph_snapshot import save_snapshot as persist_snapshot
+from . import config_manager
+
+#: Snapshots live beside the rest of the per-user configuration (#1312).
+SNAPSHOT_DIR_NAME = "snapshots"
+
+
+def snapshot_directory() -> Path:
+    return Path(config_manager.CONFIG_DIR) / SNAPSHOT_DIR_NAME
+
+
+def snapshot_file(name: str) -> Path:
+    return snapshot_directory() / f"{name}.json"
+
+
+def context_info(ctx: Any) -> Dict[str, Any]:
+    """The slice of a resolved context worth stamping into a snapshot.
+
+    ``db_path`` is left out on purpose: it is machine-specific, and a snapshot
+    shared between two checkouts should still diff cleanly.
+    """
+    return {
+        "mode": str(getattr(ctx, "mode", "") or ""),
+        "name": str(getattr(ctx, "context_name", "") or ""),
+        "database": str(getattr(ctx, "database", "") or ""),
+    }
+
+
+def run_snapshot_save(
+    db_manager: Any,
+    *,
+    name: str,
+    force: bool,
+    ctx: Any,
+    as_json: bool = False,
+) -> Snapshot:
+    """Serialise the live index and write it to the snapshot directory."""
+    snapshot = build_snapshot(db_manager, name=name, context=context_info(ctx))
+    path = persist_snapshot(snapshot, snapshot_file(snapshot.name), force=force)
+    if as_json:
+        print(dump_json(save_result_document(snapshot, path)))
+    else:
+        counts = snapshot.counts
+        print(
+            f'Saved snapshot "{snapshot.name}" '
+            f"({counts['nodes']} nodes, {counts['edges']} edges) to {path}"
+        )
+    return snapshot
+
+
+def run_snapshot_list(as_json: bool = False) -> None:
+    """List saved snapshots; never touches the database."""
+    entries = list_snapshots(snapshot_directory())
+    if as_json:
+        print(dump_json(snapshot_list_document(entries)))
+    else:
+        print("\n".join(render_snapshot_list_text(entries)))
+
+
+def load_base_snapshot(against: Optional[str]) -> Snapshot:
+    """Resolve ``--against`` to a snapshot: an explicit name, else the newest."""
+    directory = snapshot_directory()
+    if against:
+        path = snapshot_file(against)
+        if not path.is_file():
+            available = [entry.name for entry in list_snapshots(directory) if entry.snapshot]
+            hint = f" Available: {', '.join(available)}" if available else " No snapshots saved yet."
+            raise SnapshotError(f"Snapshot {against!r} not found in {directory}.{hint}")
+        return load_snapshot(path)
+
+    entries = [entry for entry in list_snapshots(directory) if entry.snapshot]
+    if not entries:
+        raise SnapshotError(
+            f"No snapshots saved in {directory}. Run `cgc snapshot save --name <name>` first."
+        )
+    return entries[0].snapshot
+
+
+def run_diff(
+    db_manager: Any,
+    *,
+    base: Snapshot,
+    ctx: Any,
+    as_json: bool = False,
+    limit: Optional[int] = None,
+) -> bool:
+    """Diff the live index against *base*. Returns True when anything changed.
+
+    *base* is resolved by the caller first so a bad ``--against`` fails before
+    a database connection is opened.
+    """
+    current = build_snapshot(db_manager, name="current", context=context_info(ctx))
+    diff = diff_snapshots(base, current)
+    if as_json:
+        print(dump_json(diff))
+    else:
+        print("\n".join(render_diff_text(diff, limit)))
+    return diff["summary"]["total"] > 0