feat(index): a full rebuild interrupted part-way resumes, against its own tree (PRJ-06 T12a)

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

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"

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

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,

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