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