Checkpoint semantic cache per chunk so interrupted runs resume

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

@@ -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

graphify/llm.py

@@ -1903,6 +1903,27 @@ def extract_corpus_parallel(
     # 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 extract_corpus_parallel(
                 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 extract_corpus_parallel(
                     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):