From d25f2fb90f288b8b3fd0f9e30dac83d2f9df3339 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=A0=D0=BE=D0=B1=D0=BE=D1=82?= Date: Sat, 26 Sep 2026 16:32:50 +0200 Subject: [PATCH] feat(index): a full rebuild interrupted part-way resumes, against its own tree (PRJ-06 T12a) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every full rebuild used to start over: the route created a fresh checkpoint whenever force_full was set, and CheckpointService.create deletes the old one. A restart at 63% of a ~12 000 s rebuild paid for all of it again — observed 2026-09-09 as three consecutive runs orphaned at 63%, 33%, 33%. - repos.resume_checkpoint: like continues like. Full continues full, an incremental checkpoint an incremental run; neither crosses (KNOW-08); an unknown (pre-column) checkpoint is never resumed. Tested as a matrix. - pipeline_runner.pin_to_tree (S-11): a resume is checked out at its checkpoint's head_sha, because clone_or_pull moved the clone to the branch's current head while the restored changed files, profile and cross-file analysis describe the checkpoint's; ast_parse re-runs on every resume and would otherwise parse a newer tree. A commit that is gone (force-push) resets the progress (CheckpointService.reset_progress) and the run starts from the top, logged at WARNING. Tested on a real repo. - The rebuild stays clean: ast_parse and graph_build are ungated, so a resumed full run re-parses the whole tree and saves the whole graph. - CLAUDE.md: the paragraph that said a full rebuild "has no checkpoint to resume from" now says what happens. Co-Authored-By: Claude Opus 5.5 (1M context) --- CLAUDE.md | 30 ++++-- backend/app/api/routes/repos.py | 34 ++++-- backend/app/knowledge/pipeline_runner.py | 51 +++++++++ backend/app/services/checkpoint_service.py | 31 ++++++ .../knowledge/test_a_full_rebuild_resumes.py | 100 ++++++++++++++++++ 5 files changed, 225 insertions(+), 21 deletions(-) create mode 100644 backend/tests/unit/knowledge/test_a_full_rebuild_resumes.py diff --git a/CLAUDE.md b/CLAUDE.md index 9ab86c6f..b44bd5ec 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -1030,17 +1030,25 @@ as a code change does. On 2026-09-09 a commit whose own text said "shipping rele through one is a losing race" orphaned the third consecutive rebuild. Before starting a full rebuild, finish merging. -**It prevents loss, not repetition, and that is deliberate.** The replacement starts at -`clone_or_pull`, because `force_full` is inherited and a full rebuild has no checkpoint to -resume from — only a clean run reconciles what `save_incremental` merges by FILE, which is -the whole reason the rebuild was full. So a full rebuild still needs an uninterrupted -window, and shipping releases through one is a losing race: observed 2026-09-09, three -consecutive runs orphaned at 63%, 33% and 33% of the way. What blunts the cost is the -document cache — each attempt regenerates fewer documents than the last, because every one -it finished recorded its `content_hash`. Verified in production: `1 put back` at 07:27:41 -and again at 08:04:45, both carrying `ORPHAN_ERROR` so neither spent the reaper's budget, -and the replacement started 36 s after the first — against 300 s of waiting followed by a -refusal. +**Since 2026-09-26 it prevents repetition too (PRJ-06 T12a).** Until then the replacement +started at `clone_or_pull`: the route created a fresh checkpoint whenever `force_full` was +set, and `CheckpointService.create` deletes the old one, so a full rebuild had nothing to +resume from — observed 2026-09-09, three consecutive runs orphaned at 63%, 33% and 33% of +the way, each paying for the whole rebuild again (the document cache blunted it: each +attempt regenerated fewer documents than the last). Now `repos.resume_checkpoint` lets +like continue like — a full run continues a full checkpoint, an incremental one an +incremental checkpoint, and neither the other (KNOW-08) — and the gated steps +(`detect_changes`, `cross_file_analysis`, `code_symbol_embed`, per-document +`generate_docs`) are skipped as they already were for incremental resumes. The rebuild +stays CLEAN: `ast_parse` and `graph_build` are ungated, so a resumed full run re-parses the +whole tree and saves the whole graph, which is what `save_incremental`'s merge-by-file +cannot do. A resume is pinned to the tree its checkpoint describes +(`pipeline_runner.pin_to_tree`, S-11): `clone_or_pull` moves the clone to the branch's +current head, and the restored changed-file list, profile and cross-file analysis describe +the checkpoint's; if that commit is gone, the recorded progress is discarded +(`CheckpointService.reset_progress`) and the run starts from the top. The orphan sweep's +`ORPHAN_ERROR` requeue is what reaches this path — verified in production earlier as `1 put +back` 36 s after a restart. Second, `_requeue_attempts` no longer counts a reap whose run started under a different release. An unstamped run still counts — unknown means counted, or the bound stops diff --git a/backend/app/api/routes/repos.py b/backend/app/api/routes/repos.py index a4c7aece..025cc1ca 100644 --- a/backend/app/api/routes/repos.py +++ b/backend/app/api/routes/repos.py @@ -38,6 +38,23 @@ logger = logging.getLogger(__name__) + +def resume_checkpoint(checkpoint_force_full: bool | None, run_force_full: bool) -> bool: + """Whether a run continues the checkpoint it finds (KNOW-08, PRJ-06). + + Only like continues like. An incremental run over a FULL checkpoint would read its + whole-tree `changed_files` as "everything changed" and redo a full rebuild while + calling itself incremental; a full run over an INCREMENTAL checkpoint would inherit + a diff it did not ask for. A full run over a full checkpoint continues it — the + rebuild interrupted at 63% resumes at 63% (PRJ-06; before 2026-09-26 every full + rebuild started over, so a restart mid-rebuild paid for the whole thing again). + `None` is a checkpoint written before the column existed: unknown, never resumed. + """ + if checkpoint_force_full is None: + return False + return checkpoint_force_full == run_force_full + + router = APIRouter() _project_svc = ProjectService() _ssh_key_svc = SshKeyService() @@ -717,21 +734,18 @@ async def _run_index_background( # 7 200 s ceiling, while calling itself incremental. `force_full is None` # is a checkpoint written before this column existed: unknown, and treated # as full, because resuming an unknown as incremental is the mistake here. - if ( - existing_cp is not None - and existing_cp.force_full is not False - and not body.force_full + if existing_cp is not None and not resume_checkpoint( + existing_cp.force_full, bool(body.force_full) ): logger.info( - "index: not resuming project %s — its checkpoint came from a full " - "rebuild (force_full=%r), which cannot be continued incrementally", + "index: not resuming project %s — its checkpoint (force_full=%r) " + "cannot be continued by a %s run", project_id[:8], existing_cp.force_full, + "full" if body.force_full else "incremental", ) - if ( - existing_cp is not None - and existing_cp.force_full is False - and not body.force_full + if existing_cp is not None and resume_checkpoint( + existing_cp.force_full, bool(body.force_full) ): existing_cp.workflow_id = wf_id existing_cp.status = "running" diff --git a/backend/app/knowledge/pipeline_runner.py b/backend/app/knowledge/pipeline_runner.py index 30c9cb59..e9eebe6f 100644 --- a/backend/app/knowledge/pipeline_runner.py +++ b/backend/app/knowledge/pipeline_runner.py @@ -177,6 +177,28 @@ class _PipelineState: } +def pin_to_tree(repo_dir: Any, sha: str) -> bool: + """Check the clone out at *sha* (detached); False when that commit is unavailable. + + The next `clone_or_pull` checks the branch out again before pulling + (`repo_analyzer.clone_or_pull`), so a detached head left here does not outlive the + run. Nothing else writes to the clone between the two. + """ + from git.exc import GitCommandError, InvalidGitRepositoryError, NoSuchPathError + + try: + repo = Repo(str(repo_dir)) + if repo.head.is_valid() and repo.head.commit.hexsha == sha: + return True + repo.git.checkout("--detach", sha) + return repo.head.commit.hexsha == sha + except (GitCommandError, InvalidGitRepositoryError, NoSuchPathError, ValueError) as exc: + # The caller logs the consequence at WARNING and restarts the run; here the + # reason is enough (an unknown commit raises GitCommandError). + logger.info("pin_to_tree: cannot check out %s: %s", sha[:12], exc) + return False + + class IndexingPipelineRunner: def __init__( self, @@ -344,6 +366,35 @@ async def _run_steps( ssh_key_passphrase=state.ssh_key_passphrase, ) + # PRJ-06 S-11: a resume continues against the tree its checkpoint was computed + # from. `clone_or_pull` just moved the clone to the branch's CURRENT head, while + # the restored `changed_files`, profile and cross-file analysis describe the + # checkpoint's head — and `ast_parse` re-runs on every resume, so without this + # it would parse a newer tree than the rest of the run describes. If that commit + # is gone (a force-push), the recorded progress describes nothing we can check + # out: it is discarded and the run starts again from the top. + if resuming and "detect_changes" in done and checkpoint.head_sha: + if not await asyncio.to_thread(pin_to_tree, state.repo_dir, checkpoint.head_sha): + logger.warning( + "index: project %s cannot resume — commit %s from its checkpoint is not " + "in the clone; starting again from the top", + project_id[:8], + checkpoint.head_sha[:12], + ) + await tracker.emit( + wf_id, + "pipeline_resume", + "warning", + f"Commit {checkpoint.head_sha[:12]} from the interrupted run is gone; " + "starting again from the top", + ) + await self._cp_svc.reset_progress(db, cp_id) + await db.refresh(checkpoint) + done.clear() + resuming = False + result.resumed = False + result.resumed_from_step = None + # --- Step 3: detect_changes --- if "detect_changes" in done: state.head_sha = checkpoint.head_sha diff --git a/backend/app/services/checkpoint_service.py b/backend/app/services/checkpoint_service.py index 1cdbb217..4deea7b0 100644 --- a/backend/app/services/checkpoint_service.py +++ b/backend/app/services/checkpoint_service.py @@ -88,6 +88,37 @@ async def create( await session.refresh(cp) return cp + async def reset_progress(self, session: AsyncSession, checkpoint_id: str) -> None: + """Forget every recorded step and document, keeping the checkpoint row. + + For a resume that cannot continue against the tree it started on (PRJ-06 S-11): + the recorded progress describes a commit that is no longer available, so the run + starts again from the top under the same checkpoint rather than reusing work + computed from a different tree. Both stores are cleared — the step/doc rows and + the legacy JSON columns `get_completed_steps` falls back to. + """ + await session.execute( + delete(IndexingCheckpointStep).where( + IndexingCheckpointStep.checkpoint_id == checkpoint_id + ) + ) + await session.execute( + delete(IndexingCheckpointDoc).where( + IndexingCheckpointDoc.checkpoint_id == checkpoint_id + ) + ) + cp = await session.get(IndexingCheckpoint, checkpoint_id) + if cp is not None: + cp.completed_steps = "[]" + cp.processed_doc_paths = "[]" + cp.head_sha = "" + cp.last_sha = None + cp.changed_files_json = "[]" + cp.deleted_files_json = "[]" + cp.profile_json = "{}" + cp.knowledge_json = "{}" + await session.commit() + async def complete_step( self, session: AsyncSession, diff --git a/backend/tests/unit/knowledge/test_a_full_rebuild_resumes.py b/backend/tests/unit/knowledge/test_a_full_rebuild_resumes.py new file mode 100644 index 00000000..7a54cc3f --- /dev/null +++ b/backend/tests/unit/knowledge/test_a_full_rebuild_resumes.py @@ -0,0 +1,100 @@ +"""PRJ-06 (T12a): a full rebuild interrupted part-way continues, against its own tree. + +Before 2026-09-26 every full rebuild started over: the route created a fresh checkpoint +whenever `force_full` was set, and `CheckpointService.create` deletes the old one. So a +dyno restart at 63% of a 12 000 s rebuild paid for the whole rebuild again; observed on +2026-09-09 as three consecutive rebuilds orphaned at 63%, 33% and 33%. + +A resume is only sound against the tree its checkpoint describes (S-11): `clone_or_pull` +moves the clone to the branch's current head, while the restored changed-file list, +profile and cross-file analysis describe the checkpoint's head. +""" + +from __future__ import annotations + +import pytest +from git import Repo +from sqlalchemy import select +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + +import app.models # noqa: F401 +from app.api.routes.repos import resume_checkpoint +from app.knowledge.pipeline_runner import pin_to_tree +from app.models.base import Base +from app.models.indexing_checkpoint import IndexingCheckpointDoc, IndexingCheckpointStep +from app.models.project import Project +from app.services.checkpoint_service import CheckpointService + + +@pytest.mark.parametrize( + ("checkpoint", "run", "resumes"), + [ + (True, True, True), # the fix: a full rebuild continues a full rebuild + (False, False, True), # unchanged: incremental continues incremental + (True, False, False), # KNOW-08: a full checkpoint is not an incremental diff + (False, True, False), # a full run must not inherit an incremental diff + (None, True, False), # written before the column existed: unknown + (None, False, False), + ], +) +def test_only_like_continues_like(checkpoint, run, resumes): + assert resume_checkpoint(checkpoint, run) is resumes + + +def _repo_with_two_commits(tmp_path): + repo = Repo.init(tmp_path) + with repo.config_writer() as cw: + cw.set_value("user", "name", "t").set_value("user", "email", "t@t") + (tmp_path / "a.py").write_text("x = 1\n") + repo.index.add(["a.py"]) + first = repo.index.commit("first").hexsha + (tmp_path / "b.py").write_text("y = 2\n") + repo.index.add(["b.py"]) + second = repo.index.commit("second").hexsha + return repo, first, second + + +def test_a_resume_is_pinned_to_the_checkpoint_tree(tmp_path): + repo, first, second = _repo_with_two_commits(tmp_path) + assert repo.head.commit.hexsha == second + assert pin_to_tree(tmp_path, first) is True + assert repo.head.commit.hexsha == first + assert not (tmp_path / "b.py").exists(), "the working tree must be the checkpoint's" + + +def test_pinning_to_the_current_head_changes_nothing(tmp_path): + repo, _first, second = _repo_with_two_commits(tmp_path) + assert pin_to_tree(tmp_path, second) is True + assert not repo.head.is_detached + + +def test_a_commit_that_is_gone_refuses_the_pin(tmp_path): + repo, _first, second = _repo_with_two_commits(tmp_path) + assert pin_to_tree(tmp_path, "0" * 40) is False + assert repo.head.commit.hexsha == second, "a failed pin must leave the clone as it was" + + +@pytest.mark.asyncio +async def test_reset_progress_forgets_the_work_and_keeps_the_checkpoint(): + engine = create_async_engine("sqlite+aiosqlite:///:memory:") + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + sm = async_sessionmaker(engine, expire_on_commit=False) + svc = CheckpointService() + async with sm() as s: + s.add(Project(id="p1", name="x", repo_url="https://x/y.git")) + await s.commit() + cp = await svc.create(s, "p1", "wf", head_sha="", force_full=True) + await svc.complete_step(s, cp.id, "detect_changes", head_sha="a" * 40) + await svc.mark_docs_batch_processed(s, cp.id, ["a.py", "b.py"]) + + await svc.reset_progress(s, cp.id) + + assert await svc.get_completed_steps(s, cp.id) == set() + assert await svc.get_processed_doc_paths(s, cp.id) == set() + steps = (await s.execute(select(IndexingCheckpointStep))).scalars().all() + docs = (await s.execute(select(IndexingCheckpointDoc))).scalars().all() + assert steps == [] and docs == [] + await s.refresh(cp) + assert cp.head_sha == "" and cp.force_full is True + await engine.dispose()