feat: graph snapshot + meta sidecar for post-session audit traceability

.gitignore

@@ -50,4 +50,3 @@ dump.rdb
 .dual-graph-context/
 benchmark/dgc-claude/
 benchmark/normal-claude/
-.dual-graph/

README.md

@@ -113,6 +113,8 @@ All data lives in `<project>/.dual-graph/` (gitignored automatically).
 | `chat_action_graph.json` | Session memory: reads, edits, queries, decisions |
 | `context-store.json` | Persistent store for decisions/tasks/facts across sessions |
 | `mcp_server.log` | MCP server logs |
+| `graph_snapshots/info_graph_<ts>.json` | Timestamped snapshot of info_graph.json saved before each rescan. Last 5 retained. |
+| `graph_snapshots/info_graph_<ts>.meta.json` | Sidecar metadata: scan trigger, file count, and action log offset at snapshot time. Links a snapshot directly to the mcp_tool_calls.jsonl entries that followed. |
 
 Global files in `~/.dual-graph/`:
 | File | Description |

bin/dg.ps1

@@ -472,6 +472,9 @@ try {
     Write-Host "[$Tool] Project : $resolvedProject"
     Write-Host "[$Tool] Data    : $DataDir"
     Write-Host ""
+    # Snapshot existing info_graph.json before rescan (fail-safe — never blocks scan)
+    try { & $Python (Join-Path $PSScriptRoot "graph_snapshot.py") $DataDir "auto" 2>$null } catch {}
+
     Write-Host "[$Tool] Scanning project..."
     if ($grapeOk) {
         & (Join-Path $VenvBin "graph-builder.exe") --root $resolvedProject --out (Join-Path $DataDir "info_graph.json") 2> $scanErr

bin/dgc.ps1

@@ -737,6 +737,9 @@ Keep ``CONTEXT.md`` under 20 lines total. Do NOT summarize the full conversation
     # under the global $ErrorActionPreference = "Stop".
     $prevEAPNative = $ErrorActionPreference; $ErrorActionPreference = "Continue"
 
+    # Snapshot existing info_graph.json before rescan (fail-safe — never blocks scan)
+    try { & $Python (Join-Path $PSScriptRoot "graph_snapshot.py") $DataDir "auto" 2>$null } catch {}
+
     Write-Host "[$Tool] Scanning project..."
     if ($grapeOk) {
         & (Join-Path $VenvBin "graph-builder.exe") --root $resolvedProject --out (Join-Path $DataDir "info_graph.json") 2> $scanErr

bin/dual_graph_launch.sh

@@ -1048,6 +1048,10 @@ fi
 
 echo "[$TOOL_LABEL] Scanning project..."
 CURRENT_STEP="Scanning project"
+
+# Snapshot existing info_graph.json before rescan (fail-safe — never blocks scan)
+"$PYTHON" "$SCRIPT_DIR/graph_snapshot.py" "$DATA_DIR" "auto" 2>/dev/null || true
+
 _SCAN_ERR_FILE="$DATA_DIR/scan_error.log"
 rm -f "$_SCAN_ERR_FILE" 2>/dev/null || true
 _SCAN_OK=0

bin/graph_snapshot.py

@@ -0,0 +1,125 @@
+"""
+graph_snapshot.py — Snapshot info_graph.json before each rescan.
+
+Saves a timestamped copy plus a metadata sidecar so that bad graph
+recommendations can be traced post-session.  Rotates to the last 5
+snapshot pairs.  Fails silently on any error — must never block a scan.
+
+Usage (from launch scripts):
+    python graph_snapshot.py <data_dir> [trigger]
+
+    data_dir  — path to .dual-graph/ (contains info_graph.json)
+    trigger   — "manual" (default) or "auto"
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import sys
+from datetime import datetime, timezone
+from pathlib import Path
+
+MAX_SNAPSHOTS = 5
+
+
+def save_graph_snapshot(
+    info_graph_path: str | os.PathLike,
+    data_dir: str | os.PathLike,
+    trigger: str = "manual",
+) -> None:
+    """Save a timestamped snapshot of info_graph.json before it is overwritten.
+
+    Also writes a .meta.json sidecar with:
+    - scan_trigger
+    - file_count (number of nodes in current graph)
+    - action_log_offset (line count of mcp_tool_calls.jsonl at this moment)
+
+    Fails silently on any error — must never block a scan.
+    """
+    try:
+        info_graph = Path(info_graph_path)
+        if not info_graph.is_file():
+            return  # nothing to snapshot
+
+        data = Path(data_dir)
+        snap_dir = data / "graph_snapshots"
+        snap_dir.mkdir(parents=True, exist_ok=True)
+
+        # ISO timestamp safe for filenames (colons replaced with dashes)
+        ts = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H-%M-%S")
+
+        snapshot_path = snap_dir / f"info_graph_{ts}.json"
+        meta_path = snap_dir / f"info_graph_{ts}.meta.json"
+
+        # 1. Copy existing info_graph.json
+        shutil.copy2(str(info_graph), str(snapshot_path))
+
+        # 2. Read file_count from graph
+        file_count = 0
+        try:
+            with open(info_graph, "r", encoding="utf-8") as f:
+                graph_data = json.load(f)
+            if isinstance(graph_data, dict):
+                # count top-level nodes (files) — common structures:
+                # {"nodes": [...]} or {"files": {...}} or flat dict of paths
+                if "nodes" in graph_data:
+                    file_count = len(graph_data["nodes"])
+                elif "files" in graph_data:
+                    file_count = len(graph_data["files"])
+                else:
+                    file_count = len(graph_data)
+            elif isinstance(graph_data, list):
+                file_count = len(graph_data)
+        except Exception:
+            pass
+
+        # 3. Read action_log_offset (line count of mcp_tool_calls.jsonl)
+        action_log_offset = 0
+        tool_calls_path = data / "mcp_tool_calls.jsonl"
+        if tool_calls_path.is_file():
+            try:
+                with open(tool_calls_path, "r", encoding="utf-8") as f:
+                    action_log_offset = sum(1 for _ in f)
+            except Exception:
+                pass
+
+        # 4. Write meta sidecar
+        meta = {
+            "timestamp": ts,
+            "scan_trigger": trigger,
+            "file_count": file_count,
+            "action_log_offset": action_log_offset,
+        }
+        with open(meta_path, "w", encoding="utf-8") as f:
+            json.dump(meta, f, indent=2)
+
+        # 5. Rotate — keep only the last MAX_SNAPSHOTS pairs
+        _rotate_snapshots(snap_dir)
+
+    except Exception:
+        pass  # never block the scan
+
+
+def _rotate_snapshots(snap_dir: Path) -> None:
+    """Keep only the newest MAX_SNAPSHOTS snapshot pairs."""
+    snapshots = sorted(snap_dir.glob("info_graph_*.json"))
+    # Exclude .meta.json from the main list
+    snapshots = [s for s in snapshots if not s.name.endswith(".meta.json")]
+
+    while len(snapshots) > MAX_SNAPSHOTS:
+        oldest = snapshots.pop(0)
+        oldest.unlink(missing_ok=True)
+        meta_path = oldest.parent / oldest.name.replace(".json", ".meta.json")
+        meta_path.unlink(missing_ok=True)
+
+
+# CLI entry point for launch scripts
+if __name__ == "__main__":
+    if len(sys.argv) < 2:
+        sys.exit(0)  # no data dir — silently skip
+    data_dir = sys.argv[1]
+    trigger = sys.argv[2] if len(sys.argv) > 2 else "manual"
+    info_graph = os.path.join(data_dir, "info_graph.json")
+    save_graph_snapshot(info_graph, data_dir, trigger)

install.sh

@@ -269,9 +269,10 @@ curl -sf  "$BASE_URL/bin/version.txt" -o "$INSTALL_DIR/version.txt" 2>/dev/null
   || true
 
 echo "[install] Downloading CLI tools..."
-curl -fsSL "$BASE_URL/bin/dgc"       -o "$INSTALL_DIR/dgc"       && chmod +x "$INSTALL_DIR/dgc"
-curl -fsSL "$BASE_URL/bin/dg"        -o "$INSTALL_DIR/dg"        && chmod +x "$INSTALL_DIR/dg"
-curl -fsSL "$BASE_URL/bin/graperoot" -o "$INSTALL_DIR/graperoot" && chmod +x "$INSTALL_DIR/graperoot"
+curl -fsSL "$BASE_URL/bin/dgc"             -o "$INSTALL_DIR/dgc"             && chmod +x "$INSTALL_DIR/dgc"
+curl -fsSL "$BASE_URL/bin/dg"              -o "$INSTALL_DIR/dg"              && chmod +x "$INSTALL_DIR/dg"
+curl -fsSL "$BASE_URL/bin/graperoot"       -o "$INSTALL_DIR/graperoot"       && chmod +x "$INSTALL_DIR/graperoot"
+curl -fsSL "$BASE_URL/bin/graph_snapshot.py" -o "$INSTALL_DIR/graph_snapshot.py"
 
 echo "[install] Creating Python venv at $VENV ..."
 "$PYTHON" -m venv "$VENV"

tests/test_graph_snapshot.py

@@ -0,0 +1,149 @@
+"""Tests for bin/graph_snapshot.py — snapshot + meta sidecar logic."""
+
+import json
+import os
+import sys
+import time
+from pathlib import Path
+
+import pytest
+
+# Add bin/ to path so we can import graph_snapshot
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "bin"))
+from graph_snapshot import save_graph_snapshot, MAX_SNAPSHOTS
+
+
+@pytest.fixture
+def data_dir(tmp_path):
+    """Create a temporary data directory with a sample info_graph.json."""
+    graph = {
+        "nodes": [
+            {"id": "src/main.py"},
+            {"id": "src/utils.py"},
+            {"id": "src/config.py"},
+        ],
+        "edges": [
+            {"from": "src/main.py", "to": "src/utils.py"},
+        ],
+    }
+    info_graph = tmp_path / "info_graph.json"
+    info_graph.write_text(json.dumps(graph), encoding="utf-8")
+    return tmp_path
+
+
+class TestSnapshotCreation:
+    def test_snapshot_created_before_overwrite(self, data_dir):
+        """Snapshot is created from the current info_graph.json content."""
+        original_content = (data_dir / "info_graph.json").read_text(encoding="utf-8")
+
+        save_graph_snapshot(data_dir / "info_graph.json", data_dir, trigger="manual")
+
+        snap_dir = data_dir / "graph_snapshots"
+        snapshots = list(snap_dir.glob("info_graph_*.json"))
+        snapshots = [s for s in snapshots if not s.name.endswith(".meta.json")]
+        assert len(snapshots) == 1
+
+        snapshot_content = snapshots[0].read_text(encoding="utf-8")
+        assert snapshot_content == original_content
+
+    def test_no_snapshot_when_info_graph_missing(self, tmp_path):
+        """No snapshot or error when info_graph.json doesn't exist."""
+        save_graph_snapshot(tmp_path / "info_graph.json", tmp_path, trigger="auto")
+        snap_dir = tmp_path / "graph_snapshots"
+        assert not snap_dir.exists() or len(list(snap_dir.iterdir())) == 0
+
+
+class TestMetaSidecar:
+    def test_meta_created_with_correct_fields(self, data_dir):
+        """Meta sidecar is created alongside snapshot with all required fields."""
+        # Create a tool calls log with known line count
+        tool_calls = data_dir / "mcp_tool_calls.jsonl"
+        tool_calls.write_text(
+            '{"tool":"graph_read"}\n{"tool":"graph_retrieve"}\n{"tool":"graph_scan"}\n',
+            encoding="utf-8",
+        )
+
+        save_graph_snapshot(data_dir / "info_graph.json", data_dir, trigger="manual")
+
+        snap_dir = data_dir / "graph_snapshots"
+        metas = list(snap_dir.glob("*.meta.json"))
+        assert len(metas) == 1
+
+        meta = json.loads(metas[0].read_text(encoding="utf-8"))
+        assert meta["scan_trigger"] == "manual"
+        assert meta["file_count"] == 3  # 3 nodes in fixture
+        assert meta["action_log_offset"] == 3  # 3 lines in jsonl
+        assert "timestamp" in meta
+
+    def test_action_log_offset_zero_when_no_jsonl(self, data_dir):
+        """action_log_offset is 0 when mcp_tool_calls.jsonl does not exist."""
+        save_graph_snapshot(data_dir / "info_graph.json", data_dir, trigger="auto")
+
+        snap_dir = data_dir / "graph_snapshots"
+        metas = list(snap_dir.glob("*.meta.json"))
+        assert len(metas) == 1
+
+        meta = json.loads(metas[0].read_text(encoding="utf-8"))
+        assert meta["action_log_offset"] == 0
+
+    def test_trigger_auto_recorded(self, data_dir):
+        """scan_trigger correctly records 'auto'."""
+        save_graph_snapshot(data_dir / "info_graph.json", data_dir, trigger="auto")
+
+        snap_dir = data_dir / "graph_snapshots"
+        metas = list(snap_dir.glob("*.meta.json"))
+        meta = json.loads(metas[0].read_text(encoding="utf-8"))
+        assert meta["scan_trigger"] == "auto"
+
+
+class TestRotation:
+    def test_only_five_pairs_retained(self, data_dir):
+        """After 6 scans, only 5 snapshot pairs remain."""
+        for i in range(6):
+            save_graph_snapshot(
+                data_dir / "info_graph.json", data_dir, trigger="manual"
+            )
+            # Ensure unique timestamps by waiting briefly
+            time.sleep(1.1)
+
+        snap_dir = data_dir / "graph_snapshots"
+        snapshots = [
+            s for s in snap_dir.glob("info_graph_*.json")
+            if not s.name.endswith(".meta.json")
+        ]
+        metas = list(snap_dir.glob("*.meta.json"))
+
+        assert len(snapshots) == MAX_SNAPSHOTS
+        assert len(metas) == MAX_SNAPSHOTS
+
+
+class TestFailSafe:
+    def test_write_error_does_not_propagate(self, data_dir):
+        """A write error in save_graph_snapshot does not raise — scan continues."""
+        # Pass a read-only directory to force a write failure
+        readonly_dir = data_dir / "readonly"
+        readonly_dir.mkdir()
+        fake_graph = readonly_dir / "info_graph.json"
+        fake_graph.write_text("{}", encoding="utf-8")
+
+        # Make graph_snapshots path a file so mkdir fails
+        blocker = readonly_dir / "graph_snapshots"
+        blocker.write_text("block", encoding="utf-8")
+
+        # This should NOT raise
+        save_graph_snapshot(fake_graph, readonly_dir, trigger="manual")
+
+    def test_corrupt_graph_json_does_not_propagate(self, tmp_path):
+        """Corrupt info_graph.json doesn't crash the snapshot."""
+        info_graph = tmp_path / "info_graph.json"
+        info_graph.write_text("NOT VALID JSON {{{{", encoding="utf-8")
+
+        # Should not raise — file_count defaults to 0
+        save_graph_snapshot(info_graph, tmp_path, trigger="auto")
+
+        snap_dir = tmp_path / "graph_snapshots"
+        snapshots = [
+            s for s in snap_dir.glob("info_graph_*.json")
+            if not s.name.endswith(".meta.json")
+        ]
+        assert len(snapshots) == 1  # snapshot still created (it's a copy)