Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 21 additions & 4 deletions graphify/llm.py
Original file line number Diff line number Diff line change
Expand Up @@ -2888,6 +2888,26 @@ def _checkpoint_chunk(result: dict, chunk: "list[Path | FileSlice]") -> None:
# Deep-mode results checkpoint into their own namespace
# (cache/semantic-deep/) so a deep run never overwrites standard
# entries — and a later standard run never serves deep ones (#1894).
# A truncated/partial chunk must not be checkpointed as
# authoritative, nor may a checkpoint that covers only some of a
# sliced file's slices (#4943). An oversized splittable file is
# dispatched as several FileSlice chunks; if the run is interrupted
# before every slice's chunk returns (crash, kill, KeyboardInterrupt),
# the slices that did finish describe an incomplete parent file, and
# a clean-stamped entry would let the next run serve the truncated
# content forever. Stamp every sliced file whose slice set is not
# fully inside this chunk as partial; the final full save of a
# completed run overwrites the stamp with a clean entry.
_partial_files: set = set(_partial_source_files(result) or [])
_covered: dict = {}
_slice_totals: dict = {}
for item in chunk:
if isinstance(item, FileSlice):
_covered.setdefault(item.path, set()).add(item.index)
_slice_totals[item.path] = item.total
for _sliced_path, _indexes in _covered.items():
if len(_indexes) < _slice_totals[_sliced_path]:
_partial_files.add(_sliced_path)
_scs(
result.get("nodes", []),
result.get("edges", []),
Expand All @@ -2901,10 +2921,7 @@ def _checkpoint_chunk(result: dict, chunk: "list[Path | FileSlice]") -> None:
# that changes _EXTRACTION_SYSTEM re-extracts instead of replaying
# this vintage forever (#1939).
prompt=_extraction_system(deep=deep_mode),
# A truncated/partial chunk must not be checkpointed as
# authoritative: pass the partial file set so its entry is
# stamped ``partial: True`` and re-dispatched next run.
partial_source_files=_partial_source_files(result) or None,
partial_source_files=_partial_files or None,
)
except Exception as _exc: # noqa: BLE001 — checkpoint is best-effort
print(f"[graphify] incremental cache checkpoint failed: {_exc}", file=sys.stderr)
Expand Down
39 changes: 39 additions & 0 deletions tests/test_chunking.py
Original file line number Diff line number Diff line change
Expand Up @@ -1004,3 +1004,42 @@ def test_pack_chunks_with_special_token_doc_does_not_crash(tmp_path):
code = tmp_path / "code.py"; code.write_text("def f():\n return 1\n")
chunks = _pack_chunks_by_tokens([doc, code], token_budget=60_000)
assert chunks # produced at least one chunk, no exception


def test_interrupted_multi_slice_checkpoint_is_partial_and_missed_on_reload(tmp_path):
"""#4943: an oversized splittable file is dispatched as several FileSlice
chunks. If a later slice's chunk never returns (crash, kill,
KeyboardInterrupt), the checkpoint written by the completed slice must not
cache the parent file as complete — the entry is stamped ``partial``, so a
reload is a miss and the file re-dispatches next run instead of serving the
truncated content forever."""
from graphify.llm import extract_corpus_parallel, _extraction_system, _FILE_CHAR_CAP
from graphify.cache import load_cached

doc = tmp_path / "doc.md"
doc.write_text("# Head\n" + "filler " * 6000 + "\n# Tail\nTail assertion.\n")
assert len(doc.read_text()) > _FILE_CHAR_CAP # the corpus really slices

calls = []

def interrupted_first_only(chunk, **kwargs):
calls.append(len(chunk))
if len(calls) > 1:
raise KeyboardInterrupt()
return {
"nodes": [{"id": "head", "source_file": "doc.md", "file_type": "document"}],
"edges": [], "hyperedges": [],
"input_tokens": 1, "output_tokens": 1,
"finish_reason": "stop",
}

with patch("graphify.llm.extract_files_direct", side_effect=interrupted_first_only):
with pytest.raises(KeyboardInterrupt):
extract_corpus_parallel(
[doc], backend="kimi", root=tmp_path,
token_budget=1, max_concurrency=1, max_retry_depth=0,
)

assert len(calls) >= 2 # the file was sliced into at least two chunks
# The checkpoint of the completed slice must not read as a complete file.
assert load_cached(doc, tmp_path, kind="semantic", prompt=_extraction_system()) is None
Loading