Fix stale commit extraction resume failures

scripts/streaming_conveyor.py

@@ -606,8 +606,11 @@ def manifest_complete_commit_ranges(
     """
     if range_size <= 0:
         raise ValueError(f"range_size must be positive, got {range_size}")
-    if any(k == f"{repo}::commits" or k.startswith(f"{repo}::r")
-           for k in manifest.failed):
+    # A repo-level failure records an earlier extraction attempt, not a unit of
+    # range coverage. Once the authoritative plan and every range prove exact
+    # coverage it is stale. A failed range remains authoritative and must keep
+    # completion fail-closed until that same range key is marked done.
+    if any(k.startswith(f"{repo}::r") for k in manifest.failed):
         return None
 
     plan = manifest.done.get(commit_plan_key(repo))
@@ -660,6 +663,52 @@ def manifest_complete_commit_ranges(
     return tuple(ranges)
 
 
+def mark_commit_stream_complete(
+    repo: str,
+    manifest: Manifest,
+    manifest_lock: threading.Lock | None,
+    complete_ranges: Sequence[tuple[int, int]],
+) -> dict:
+    """Persist aggregate completion proven by the plan and exact range coverage.
+
+    ``Manifest.mark_done`` also removes an earlier aggregate failure. Keeping
+    this as a derived summary avoids treating ``repo::commits`` as a second,
+    independent source of completion truth.
+    """
+    if not complete_ranges:
+        raise ValueError(f"cannot mark {repo}::commits complete without ranges")
+    plan = manifest.done.get(commit_plan_key(repo))
+    if not isinstance(plan, dict) or plan.get("source") != "commit_plan":
+        raise RuntimeError(
+            f"cannot mark {repo}::commits complete without authoritative commit_plan"
+        )
+    try:
+        n_records = int(plan["n_records"])
+    except (KeyError, TypeError, ValueError) as exc:
+        raise RuntimeError(
+            f"cannot mark {repo}::commits complete: invalid commit_plan n_records"
+        ) from exc
+    if complete_ranges[0][0] != 0 or complete_ranges[-1][1] != n_records:
+        raise RuntimeError(
+            f"cannot mark {repo}::commits complete: ranges do not cover "
+            f"[0, {n_records})"
+        )
+    info = {
+        "source": "commits",
+        "repo": repo,
+        "complete": True,
+        "completion_proof": "commit_plan_exact_range_coverage",
+        "n_records": n_records,
+        "range_count": len(complete_ranges),
+    }
+    if manifest_lock is None:
+        manifest.mark_done(f"{repo}::commits", info)
+    else:
+        with manifest_lock:
+            manifest.mark_done(f"{repo}::commits", info)
+    return info
+
+
 def manifest_done_commit_intervals(repo: str, manifest: Manifest) -> tuple[tuple[int, int], ...]:
     """Return validated done commit intervals for ``repo``.
 
@@ -1295,6 +1344,7 @@ def should_stage_repo_from_manifest(
     manifest: Manifest,
     range_size: int,
     only_repos: set[str] | None,
+    manifest_lock: threading.Lock | None = None,
 ) -> bool:
     """Return False when the manifest proves this repo needs no extraction.
 
@@ -1310,10 +1360,20 @@ def should_stage_repo_from_manifest(
         return False
 
     code_needed = streams in {"both", "code"} and not manifest.is_done(code_key(repo))
-    commits_needed = (
-        streams in {"both", "commits"}
-        and manifest_complete_commit_ranges(repo, manifest, range_size) is None
-    )
+    commits_needed = False
+    if streams in {"both", "commits"}:
+        complete_ranges = manifest_complete_commit_ranges(repo, manifest, range_size)
+        commits_needed = complete_ranges is None
+        if complete_ranges is not None:
+            # This callback can skip extraction entirely, so reconcile the
+            # derived aggregate sentinel here instead of waiting for
+            # run_commits_half, which will never be called for this repo.
+            mark_commit_stream_complete(
+                repo,
+                manifest,
+                manifest_lock,
+                complete_ranges,
+            )
     return code_needed or commits_needed
 
 
@@ -1872,6 +1932,12 @@ def run_commits_half(
         if complete_ranges is not None:
             first = complete_ranges[0][0]
             last = complete_ranges[-1][1]
+            mark_commit_stream_complete(
+                repo,
+                manifest,
+                manifest_lock,
+                complete_ranges,
+            )
             _log(
                 f"SKIP (done) {repo}::commits: manifest covers "
                 f"{len(complete_ranges)} range(s) [{first}:{last}); "
@@ -2246,7 +2312,15 @@ def handle_commit_range(
     # True iff EVERY range for this repo is now marked done in the manifest
     # (covers resume-skipped + newly-done; excludes cancelled/failed). Drives
     # temp + extract-cache retention in process_one_repo.
-    all_ranges_done = manifest_covers_commit_span(repo, manifest, n_records)
+    complete_ranges = manifest_complete_commit_ranges(repo, manifest, range_size)
+    all_ranges_done = complete_ranges is not None
+    if complete_ranges is not None:
+        mark_commit_stream_complete(
+            repo,
+            manifest,
+            manifest_lock,
+            complete_ranges,
+        )
     return done, failed, all_ranges_done
 
 
@@ -2987,6 +3061,7 @@ def should_process(repo: str) -> bool:
             manifest=manifest,
             range_size=args.range_size,
             only_repos=only_repos,
+            manifest_lock=manifest_lock,
         )
         if should_stage:
             ensure_min_free_disk(

tests/test_streaming_conveyor_progress.py

@@ -164,6 +164,52 @@ def test_manifest_complete_commit_ranges_uses_authoritative_record_count(tmp_pat
     ) == ((0, 500), (500, 823))
 
 
+def test_manifest_complete_commit_ranges_ignores_stale_aggregate_failure(tmp_path):
+    import streaming_conveyor
+
+    manifest = streaming_conveyor.Manifest(
+        path=tmp_path / "_done.json",
+        done={
+            "repo::commit_plan": {"source": "commit_plan", "n_records": 823},
+            "repo::r0": {"range": [0, 500], "source": "commits"},
+            "repo::r500": {"range": [500, 823], "source": "commits"},
+        },
+        failed={
+            "repo::commits": {
+                "stage": "extract_git_history",
+                "detail": "stale failure from an earlier attempt",
+            }
+        },
+    )
+
+    assert streaming_conveyor.manifest_complete_commit_ranges(
+        "repo", manifest, 500
+    ) == ((0, 500), (500, 823))
+
+
+def test_manifest_complete_commit_ranges_rejects_failed_range(tmp_path):
+    import streaming_conveyor
+
+    manifest = streaming_conveyor.Manifest(
+        path=tmp_path / "_done.json",
+        done={
+            "repo::commit_plan": {"source": "commit_plan", "n_records": 823},
+            "repo::r0": {"range": [0, 500], "source": "commits"},
+            "repo::r500": {"range": [500, 823], "source": "commits"},
+        },
+        failed={
+            "repo::r500": {
+                "stage": "pack",
+                "detail": "range is not authoritatively complete",
+            }
+        },
+    )
+
+    assert streaming_conveyor.manifest_complete_commit_ranges(
+        "repo", manifest, 500
+    ) is None
+
+
 def test_manifest_complete_commit_ranges_refuses_exact_multiple(tmp_path):
     import streaming_conveyor
 
@@ -740,7 +786,12 @@ def test_run_commits_half_skips_extract_when_manifest_proves_complete(tmp_path):
             "repo::r0": {"range": [0, 500], "source": "commits"},
             "repo::r500": {"range": [500, 612], "source": "commits"},
         },
-        failed={},
+        failed={
+            "repo::commits": {
+                "stage": "extract_git_history",
+                "detail": "stale failure from an earlier attempt",
+            }
+        },
     )
 
     with ThreadPoolExecutor(max_workers=1) as pool:
@@ -764,6 +815,15 @@ def test_run_commits_half_skips_extract_when_manifest_proves_complete(tmp_path):
         )
 
     assert (done, failed, all_done) == (0, 0, True)
+    assert manifest.done["repo::commits"] == {
+        "source": "commits",
+        "repo": "repo",
+        "complete": True,
+        "completion_proof": "commit_plan_exact_range_coverage",
+        "n_records": 612,
+        "range_count": 2,
+    }
+    assert "repo::commits" not in manifest.failed
 
 
 def test_process_one_repo_cleans_partial_intermediates_by_default(tmp_path, monkeypatch):
@@ -1060,6 +1120,15 @@ def runner(
     range_keys = sorted(key for key in manifest.done if key.startswith("repo::r"))
     assert range_keys == ["repo::r0", "repo::r1", "repo::r2"]
     assert manifest.done["repo::commit_plan"]["n_records"] == 3
+    assert manifest.done["repo::commits"] == {
+        "source": "commits",
+        "repo": "repo",
+        "complete": True,
+        "completion_proof": "commit_plan_exact_range_coverage",
+        "n_records": 3,
+        "range_count": 3,
+    }
+    assert "repo::commits" not in manifest.failed
     assert len(observed_kwargs) == 3
     reader = DedupStore(str(db), near=False, commit_every=1)
     try:
@@ -1302,7 +1371,12 @@ def test_should_stage_repo_skips_manifest_proven_complete_both_streams(tmp_path)
             "repo::r0": {"range": [0, 500], "source": "commits"},
             "repo::r500": {"range": [500, 612], "source": "commits"},
         },
-        failed={},
+        failed={
+            "repo::commits": {
+                "stage": "extract_git_history",
+                "detail": "stale failure from an earlier attempt",
+            }
+        },
     )
 
     assert not streaming_conveyor.should_stage_repo_from_manifest(
@@ -1313,6 +1387,15 @@ def test_should_stage_repo_skips_manifest_proven_complete_both_streams(tmp_path)
         range_size=500,
         only_repos=None,
     )
+    assert manifest.done["repo::commits"] == {
+        "source": "commits",
+        "repo": "repo",
+        "complete": True,
+        "completion_proof": "commit_plan_exact_range_coverage",
+        "n_records": 612,
+        "range_count": 2,
+    }
+    assert "repo::commits" not in manifest.failed
     assert streaming_conveyor.should_stage_repo_from_manifest(
         "repo",
         streams="both",