Checkpoint semantic cache per chunk so interrupted runs resume

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 _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):