From e7e2459e443989dbd061c9d01d297637996725f2 Mon Sep 17 00:00:00 2001
From: "A.Levin" <abdullahlevin@gmail.com>
Date: Tue, 7 Jul 2026 16:30:16 +0300
Subject: [PATCH] Checkpoint semantic cache per chunk so interrupted runs
 resume
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Semantic extraction only wrote to the cache once, at the very end of the
run (save_semantic_cache in __main__ after extract_corpus_parallel returns).
A run interrupted partway — a crash, a kill, or a claude-cli/API run that
exits when it hits a rate limit — therefore lost every completed chunk and
restarted from scratch. On a large corpus with a slow local backend this can
throw away many hours of work.

Persist each chunk's results to the semantic cache as soon as it completes,
in both the serial and threaded paths of extract_corpus_parallel. Add a
merge_existing option to save_semantic_cache so a file split into slices
across several chunks accumulates its slices instead of the later chunk
overwriting the earlier one. The checkpoint is best-effort (a cache write
error never aborts extraction) and can be disabled with
GRAPHIFY_NO_INCREMENTAL_CACHE. Default behaviour of save_semantic_cache is
unchanged (merge_existing defaults to False).
---
 graphify/cache.py | 14 ++++++++++++++
 graphify/llm.py   | 23 +++++++++++++++++++++++
 2 files changed, 37 insertions(+)

diff --git a/graphify/cache.py b/graphify/cache.py
index bb3f9d593..35dc7cc11 100644
--- a/graphify/cache.py
+++ b/graphify/cache.py
@@ -533,12 +533,18 @@ def save_semantic_cache(
     edges: list[dict],
     hyperedges: list[dict] | None = None,
     root: Path = Path("."),
+    merge_existing: bool = False,
 ) -> int:
     """Save semantic extraction results to cache, keyed by source_file.
 
     Groups nodes and edges by source_file, then saves one cache entry per file
     under cache/semantic/ (separate from AST entries in cache/ast/) to prevent
     hash-key collisions (#582).
+
+    When ``merge_existing`` is True, any already-cached entry for a file is
+    unioned with the new results before saving instead of being overwritten.
+    This lets callers checkpoint incrementally (e.g. once per chunk) without
+    dropping a prior slice of a large file that was split across chunks.
     Returns the number of files cached.
     """
     from collections import defaultdict
@@ -563,6 +569,14 @@ def save_semantic_cache(
         if not p.is_absolute():
             p = Path(root) / p
         if p.is_file():
+            if merge_existing:
+                prev = load_cached(p, root, kind="semantic")
+                if prev:
+                    result = {
+                        "nodes": (prev.get("nodes", []) or []) + result["nodes"],
+                        "edges": (prev.get("edges", []) or []) + result["edges"],
+                        "hyperedges": (prev.get("hyperedges", []) or []) + result["hyperedges"],
+                    }
             save_cached(p, result, root, kind="semantic")
             saved += 1
     return saved
diff --git a/graphify/llm.py b/graphify/llm.py
index f00ed0c84..560bbcb88 100644
--- a/graphify/llm.py
+++ b/graphify/llm.py
@@ -1903,6 +1903,27 @@ def _run_one(idx: int, chunk: list[Path]) -> tuple[int, dict | None, Exception |
     # over session state. Force serial unless the user explicitly opts in.
     if backend == "claude-cli" and os.environ.get("GRAPHIFY_CLAUDE_CLI_PARALLEL", "").strip() != "1":
         max_concurrency = 1
+    def _checkpoint_chunk(result: dict) -> None:
+        # Persist each chunk's semantic results to the cache as soon as it
+        # completes. Without this, the semantic cache is only written once, at
+        # the very end of the run (in __main__), so a run interrupted partway
+        # — a crash, a kill, or a claude-cli/API run that exits on a rate
+        # limit — loses every completed chunk and restarts from scratch. This
+        # is best-effort: a cache write failure must never abort extraction.
+        if os.environ.get("GRAPHIFY_NO_INCREMENTAL_CACHE"):
+            return
+        try:
+            from .cache import save_semantic_cache as _scs
+            _scs(
+                result.get("nodes", []),
+                result.get("edges", []),
+                result.get("hyperedges", []),
+                root=root,
+                merge_existing=True,
+            )
+        except Exception as _exc:  # noqa: BLE001 — checkpoint is best-effort
+            print(f"[graphify] incremental cache checkpoint failed: {_exc}", file=sys.stderr)
+
     workers = max(1, min(max_concurrency, total))
     if workers == 1:
         # Avoid thread pool overhead for single-worker runs (and keep
@@ -1915,6 +1936,7 @@ def _run_one(idx: int, chunk: list[Path]) -> tuple[int, dict | None, Exception |
                 continue
             assert result is not None
             _merge_into(merged, result)
+            _checkpoint_chunk(result)
             if callable(on_chunk_done):
                 on_chunk_done(idx, total, result)
     else:
@@ -1939,6 +1961,7 @@ def _run_one(idx: int, chunk: list[Path]) -> tuple[int, dict | None, Exception |
                     continue
                 assert result is not None
                 results_by_idx[idx] = result
+                _checkpoint_chunk(result)
                 if callable(on_chunk_done):
                     on_chunk_done(idx, total, result)
         for idx in sorted(results_by_idx):