unseen: parallelize extraction + batch per-repo blob fetch

Serial extraction at ~147s/repo made a 300-repo pool take days, and
~626 lazy per-file blob faults per clone pushed slow repos past the
240s deadline (a draw reached 21 entries from 347 attempts, 160 of them
extraction_timeout). Both fixed.

Parallel: a bounded ThreadPoolExecutor (--extract-workers, default 6)
runs clones concurrently, each into its own clone with a private counter.
The main thread alone owns the journal, REST (license via finish_license),
counters, and acceptance. Entries are accepted in STRICT SEED ORDER with
the fork-network dedup applied at acceptance, so the pool is byte-
identical to the serial draw (tested: workers=6 == workers=1, including
fork collisions, out-of-order completion, and resume). The systemic guard
and resilient skips are unchanged; a graceful stop drains started work to
the journal for resume.

Batched blobs: repository_changes fetches the fixes together and
Git.materialize fetches all missing blobs for the read revisions in one
`fetch --no-filter --stdin` (rev-list --missing=print), instead of per-
file promisor faults. Content is byte-identical (tested; >=12 fetches
collapsed to 1).

202 unseen tests pass, ruff clean.

Co-Authored-By: Claude Opus 4.8 <[email protected]>

benchmarks/unseen/draft.py

@@ -16,7 +16,8 @@ import random
 import subprocess
 import tempfile
 import time
-from collections import Counter, defaultdict
+from collections import Counter, defaultdict, deque
+from concurrent.futures import ThreadPoolExecutor
 from contextlib import contextmanager
 from dataclasses import asdict
 from pathlib import Path
@@ -121,9 +122,21 @@ def write_draft(root: Path, assigned: list[dict], *, seed: int) -> tuple[Path, P
 
 
 class Clones:
-    """One disposable blobless clone at a time; no name in cache paths or subprocess output."""
+    """Disposable blobless clones; each worker receives its own counter accumulator."""
 
-    def __init__(self, root: Path, cache: Path, *, runner=None, limit_bytes=500 * 1024 * 1024, deadline_seconds: float | None = None):
+    def __init__(
+        self,
+        root: Path,
+        cache: Path,
+        *,
+        runner=None,
+        limit_bytes=500 * 1024 * 1024,
+        deadline_seconds: float | None = None,
+        extract_workers: int = 6,
+    ):
+        if extract_workers < 1:
+            raise ValueError("invalid_extract_workers")
+        self.extract_workers = extract_workers
         self.cache = external(root, cache)
         self.runner = runner or subprocess.run
         self.limit_bytes = limit_bytes
@@ -138,8 +151,8 @@ class Clones:
             path = Path(directory)
             git = Git(path, runner=self.runner, limit_bytes=self.limit_bytes, deadline_seconds=self.deadline_seconds)
             try:
-                # Full commit graph is needed for temporal thirds and OSV brackets; blobs
-                # are fetched lazily. This is a partial clone, not a truncated history.
+                # Full commit graph is needed for temporal thirds and OSV brackets.
+                # repository_changes batches blobs before reading their content.
                 done = self.runner(
                     ["git", "-c", "credential.helper=", "clone", "--quiet", "--filter=blob:none", "--no-checkout", url, str(path)],
                     stdin=subprocess.DEVNULL,
@@ -195,18 +208,21 @@ def advisory_index(rows: list[dict], github) -> dict[str, set[str]]:
 
 
 def repository_changes(git, candidate, fixes: set[str], *, seed: int):
+    # Fetch detached fixes together, then materialize the histories before any diff,
+    # show or blame can trigger a per-file promisor fetch.
+    fixes = sorted(fixes | {candidate.fix})
+    # The candidate fetch in extract_repository retains the original tag behavior.
+    # Do not import other fixes' tags before introducing() checks its version bracket.
+    git.run("fetch", "--quiet", "--no-tags", "--filter=blob:none", "origin", *fixes)
+    git.materialize(["HEAD", *fixes])
     proof = prove_source(git, candidate.parent)
     change = extract.introducing(git, candidate)
     known = set()
     expanded = set()
-    for fix in sorted(fixes | {candidate.fix}):
-        # Explicitly fetch detached advisory commits unavailable on default-branch history.
-        # All git including this possible network call still goes through the injected client.
-        git.run("fetch", "--quiet", "--filter=blob:none", "origin", fix)
+    for fix in fixes:
         full = git.run("rev-parse", fix).strip()
         expanded.add(full)
-        parents = git.parents(full)
-        for parent in parents:
+        for parent in git.parents(full):
             sites = extract.fix_sites(git, full, parent, global_ok=True)
             known.update((site["path"], site["function"]) for site in sites)
     rows = extract.ordinary_history(git, "HEAD")
@@ -255,27 +271,30 @@ def enumerate_advisories(github, journal):
     return list(records.values())
 
 
-def decide(row, github, osv, used, selected_networks):
+def decide(row, github, osv, used, selected_networks=()):
+    # selected_networks is retained for callers; only ordered acceptance consults it.
+    network = {}
     try:
         links = source.fix_links(row)
         if len(links) != 1:
             raise source.Rejected("fix_links")
         name, _ = next(iter(links))
         resolved = eligibility.resolve(name, github)
-        rejected_by = eligibility.rejection(resolved, used, set(selected_networks))
+        rejected_by = eligibility.rejection(resolved, used, set())
         if rejected_by:
             raise source.Rejected("used_" + rejected_by)
+        network = {"network": resolved.network}
         candidate = source.candidate(row, github, osv)
         return {"candidate": asdict(candidate), "network": resolved.network}
     except source.Rejected as exc:
-        return {"rejected": str(exc)}
+        return {"rejected": str(exc), **network}
     except eligibility.RepositoryNotFound:
-        return {"rejected": "repository_or_commit_not_found"}
+        return {"rejected": "repository_or_commit_not_found", **network}
     except ValueError as exc:
         # A per-candidate transport failure (e.g. get_commit on an unresolvable fix ref)
         # skips this advisory; the run loop's systemic guard catches a broken instrument.
         if str(exc) == "http_failure":
-            return {"rejected": "decide_http_failure"}
+            return {"rejected": "decide_http_failure", **network}
         raise
 
 
@@ -285,18 +304,7 @@ def extract_repository(candidate, github, clones, fixes, seed):
         with clones.open(candidate.url) as git:
             git.run("fetch", "--quiet", "--filter=blob:none", "origin", candidate.fix)
             changes, exclusions, proof = repository_changes(git, candidate, fixes, seed=seed)
-            pinned = changes[0]["head"]
-            license_record = github.get_license(candidate.repository, pinned)
-            spdx, license_class = source.license_info(license_record)
-            entry = {
-                **asdict(candidate),
-                "license_spdx": spdx,
-                "license_class": license_class,
-                "license_ref": pinned,
-                "license_path": license_record.get("path", ""),
-                "changes": changes,
-                "instrument": proof,
-            }
+            entry = {**asdict(candidate), "changes": changes, "instrument": proof}
         result = {"entry": entry, "ordinary_exclusions": exclusions}
     except source.Rejected as exc:
         result = {"rejected": str(exc), "ordinary_exclusions": getattr(exc, "counts", {})}
@@ -305,9 +313,104 @@ def extract_repository(candidate, github, clones, fixes, seed):
     except (extract.InstrumentFailure, OSError) as exc:
         result = {"rejected": "extract_" + str(exc), "ordinary_exclusions": {}}
     result["counts"] = {key: value - before.get(key, 0) for key, value in clones.counts.items()}
+    return finish_license(result, github) if github is not None else result
+
+
+def finish_license(result, github):
+    """REST stays on the coordinator, including the license at the extracted pin."""
+    if "entry" not in result:
+        return result
+    try:
+        entry = result["entry"]
+        pinned = entry["changes"][0]["head"]
+        record = github.get_license(entry["repository"], pinned)
+        spdx, license_class = source.license_info(record)
+        entry.update(license_spdx=spdx, license_class=license_class, license_ref=pinned, license_path=record.get("path", ""))
+    except (source.Rejected, eligibility.RepositoryNotFound) as exc:
+        reason = "repository_or_commit_not_found" if isinstance(exc, eligibility.RepositoryNotFound) else str(exc)
+        return {"rejected": reason, "ordinary_exclusions": {}, "counts": result["counts"]}
+    except (extract.InstrumentFailure, OSError) as exc:
+        return {"rejected": "extract_" + str(exc), "ordinary_exclusions": {}, "counts": result["counts"]}
+    except ValueError as exc:
+        if str(exc) != "http_failure":
+            raise
+        return {"rejected": "extract_http_failure", "ordinary_exclusions": {}, "counts": result["counts"]}
     return result
 
 
+@contextmanager
+def ordered_extractions(rows, api, osv, used, clones, fixes, seed, checkpoint, entries, workers):
+    """Bounded speculation, ordered consumption, and coordinator-only durable writes.
+
+    At most `workers` candidates are ahead of the acceptance cursor. Counters are
+    private to each job, including clone cleanup, and merged exactly once here.
+    Graceful STOP/guard/target exits drain started work into the journal for resume;
+    a hard interrupt may lose unjournaled work but cannot change accepted order.
+    """
+    pending = deque()
+    pool = ThreadPoolExecutor(max_workers=workers)
+
+    def collect(item):
+        row, decision, key, future, local = item
+        if future is not None:
+            future.exception()  # Wait for cleanup before reading this job's counters.
+            for name, value in local.counts.items():
+                if name == "peak_clone_bytes":
+                    clones.counts[name] = max(clones.counts[name], value)
+                else:
+                    clones.counts[name] += value
+            result = finish_license(future.result(), api)
+            checkpoint.put("extraction", key, result)
+        return row, decision, checkpoint.rows["extraction"].get(key)
+
+    def iterate():
+        ordered = iter(rows)
+        exhausted = False
+        while pending or not exhausted:
+            while not exhausted and len(pending) < min(workers, 300 - len(entries)):
+                checkpoint.boundary()
+                row = next(ordered, None)
+                if row is None:
+                    exhausted = True
+                    break
+                key = row["ghsa_id"]
+                if key not in checkpoint.rows["eligibility"]:
+                    checkpoint.put("eligibility", key, decide(row, api, osv, used))
+                decision = checkpoint.rows["eligibility"][key]
+                extraction_key, future, local = None, None, None
+                if "candidate" in decision:
+                    candidate = source.Candidate(**decision["candidate"])
+                    extraction_key = digest({"repository": candidate.repository, "advisory": key, "fix": candidate.fix})
+                    if extraction_key not in checkpoint.rows["extraction"]:
+                        checkpoint.boundary()
+                        local = copy.copy(clones)
+                        local.counts = Counter()
+                        future = pool.submit(extract_repository, candidate, None, local, fixes.get(candidate.repository, set()), seed)
+                pending.append((row, decision, extraction_key, future, local))
+            if not pending:
+                return
+            checkpoint.boundary()
+            yield collect(pending.popleft())
+
+    interrupted = False
+    stopped = None
+    try:
+        yield iterate()
+    except BaseException as exc:
+        interrupted = not isinstance(exc, Exception)
+        stopped = str(exc) if isinstance(exc, Stopped) else None
+        raise
+    finally:
+        pool.shutdown(wait=True, cancel_futures=True)
+        if not interrupted:
+            for item in pending:
+                future = item[3]
+                if future is not None and not future.cancelled():
+                    collect(item)
+            if stopped is not None:
+                checkpoint.progress("stopped", reason=stopped)
+
+
 def publish(journal, assigned, seed):
     """Idempotent publication from completed journal data, never inside the checkout."""
     public, private = [], []
@@ -339,10 +442,15 @@ def run(
     journal_path=None,
     max_hours=None,
     clock=None,
+    extract_workers=None,
 ) -> dict:
+    workers = extract_workers if extract_workers is not None else getattr(clones, "extract_workers", 6)
+    if workers < 1:
+        raise ValueError("invalid_extract_workers")
     started = time.monotonic()
     report = {"status": "sourcing", "seed": seed, "candidates": {}, "rejections": {}, "ordinary_exclusions": {}}
     # Operational budget only: changing it must not invalidate journal resume inputs.
+    report["extract_workers"] = workers
     report["extract_deadline_seconds"] = getattr(clones, "deadline_seconds", None)
     counts, rejected, ordinary_counts = Counter(), Counter(), Counter()
     checkpoint = None
@@ -365,41 +473,39 @@ def run(
             fixes_by_repo = advisory_index(rows, api)
             selected_networks, entries = set(), []
             instrument_failures = 0
-            for row in source.ordered_advisories(rows, seed):
-                checkpoint.boundary()
-                key = row["ghsa_id"]
-                counts[source.ecosystem(row)] += 1
-                if key not in checkpoint.rows["eligibility"]:
-                    decision = decide(row, api, osv, used, selected_networks)
-                    checkpoint.put("eligibility", key, decision)
-                decision = checkpoint.rows["eligibility"][key]
-                if "rejected" in decision:
-                    rejected[decision["rejected"]] += 1
-                    instrument_failures += int(decision["rejected"] == "decide_http_failure")
+            with ordered_extractions(
+                source.ordered_advisories(rows, seed),
+                api,
+                osv,
+                used,
+                clones,
+                fixes_by_repo,
+                seed,
+                checkpoint,
+                entries,
+                workers,
+            ) as outcomes:
+                for row, decision, result in outcomes:
+                    counts[source.ecosystem(row)] += 1
+                    if decision.get("network") in selected_networks:
+                        # Serial decide skipped this candidate entirely. Even a failed
+                        # speculative extraction must not affect the systemic guard.
+                        rejected["fork_network"] += 1
+                    elif "rejected" in decision:
+                        rejected[decision["rejected"]] += 1
+                        instrument_failures += int(decision["rejected"] == "decide_http_failure")
+                    else:
+                        ordinary_counts.update(result["ordinary_exclusions"])
+                        if "rejected" in result:
+                            rejected[result["rejected"]] += 1
+                            instrument_failures += int(result["rejected"].startswith("extract_"))
+                        else:
+                            selected_networks.add(decision["network"])
+                            entries.append(result["entry"])
                     if (instrument_failures >= 25 and not entries) or instrument_failures > 200:
                         raise extract.InstrumentFailure("systemic_extraction_failure")
-                    continue
-                candidate = source.Candidate(**decision["candidate"])
-                # Multiple advisories for one repository can have different fix/range inputs.
-                extraction_key = digest({"repository": candidate.repository, "advisory": key, "fix": candidate.fix})
-                if extraction_key not in checkpoint.rows["extraction"]:
-                    checkpoint.boundary()
-                    result = extract_repository(candidate, api, clones, fixes_by_repo.get(candidate.repository, set()), seed)
-                    checkpoint.put("extraction", extraction_key, result)
-                result = checkpoint.rows["extraction"][extraction_key]
-                ordinary_counts.update(result["ordinary_exclusions"])
-                if "rejected" in result:
-                    rejected[result["rejected"]] += 1
-                    instrument_failures += int(result["rejected"].startswith("extract_"))
-                else:
-                    selected_networks.add(decision["network"])
-                    entries.append(result["entry"])
-                # Count per-candidate instrument failures (decide + extract, replay included),
-                # never ordinary eligibility exclusions.
-                if (instrument_failures >= 25 and not entries) or instrument_failures > 200:
-                    raise extract.InstrumentFailure("systemic_extraction_failure")
-                if len(entries) == 300:
-                    break
+                    if len(entries) == 300:
+                        break
             report.update(
                 {
                     "repositories": len(entries),
@@ -463,18 +569,26 @@ def main(argv: list[str] | None = None) -> int:
     parser.add_argument("--cache", type=Path)
     parser.add_argument("--max-hours", type=float, help="Stop between units after this many hours per invocation")
     parser.add_argument("--extract-deadline", type=float, default=240.0, help="Extraction wall-clock budget per repository in seconds")
+    parser.add_argument("--extract-workers", type=int, default=6, help="Maximum concurrent repository extractions (default: 6)")
     parser.add_argument("--seed", type=int, default=20260601)
     parser.add_argument("--local-store")
     parser.add_argument("--s3-store", default="s3://")
     parser.add_argument("--pointer-manifest", type=Path, action="append", default=[])
     parser.add_argument("--known-empty", choices=(*eligibility.SOURCES, "memory_local", "memory_s3"), action="append", default=[])
     args = parser.parse_args(argv)
+    if args.extract_workers < 1:
+        parser.error("--extract-workers must be at least 1")
     try:
         config.load_dotenv(args.root / ".env")
         # Refuse an existing draft before doing any network work.
         if any((args.root / p).exists() for p in ("benchmarks/unseen/draft-p1.toml", "benchmarks/unseen/private/draft-p1.toml")):
             raise FileExistsError("draft_exists")
-        clones = Clones(args.root, args.cache or results_root() / "unseen-draft", deadline_seconds=args.extract_deadline)
+        clones = Clones(
+            args.root,
+            args.cache or results_root() / "unseen-draft",
+            deadline_seconds=args.extract_deadline,
+            extract_workers=args.extract_workers,
+        )
         stores = {
             "local": memory.open_store(args.local_store or "file://" + str(results_root() / "memory"), read_only=True),
             "s3": memory.open_store(args.s3_store, read_only=True),

benchmarks/unseen/extract.py

@@ -53,12 +53,12 @@ class Git:
             raise ExtractionTimeout("extraction_timeout")
         return min(180, remaining)
 
-    def run(self, *args: str, allowed=(0,)) -> str:
+    def run(self, *args: str, allowed=(0,), input_data: bytes | None = None) -> str:
         timeout = self.command_timeout()
         try:
             done = self.runner(
                 ["git", "-c", "core.quotePath=false", "-C", str(self.path), *args],
-                stdin=subprocess.DEVNULL,
+                **({"stdin": subprocess.DEVNULL} if input_data is None else {"input": input_data}),
                 stdout=subprocess.PIPE,
                 stderr=subprocess.PIPE,
                 timeout=timeout,
@@ -75,6 +75,38 @@ class Git:
         self.last_blob_bytes = len(done.stdout)
         return done.stdout.decode("utf-8", errors="replace")
 
+    def materialize(self, revisions: list[str]) -> None:
+        """Fetch missing historical blobs in one pack, without changing content reads.
+
+        ordinary_history inspects the complete first-parent history, including both
+        sides, attributes and non-source diffs. Blame -M/-C and log -L can walk all
+        ancestors and copy sources. Include the reachable trees, not only tip files,
+        so these commands keep exactly their original semantics. This is a safe
+        superset (merge-side trees included), still subject to the disk/deadline caps.
+        Missing-object enumeration itself does not trigger lazy blob fetches.
+        """
+        data = ("\n".join(revisions) + "\n").encode()
+        objects = self.run("rev-list", "--objects", "--missing=print", "--stdin", input_data=data)
+        missing = sorted({line[1:] for line in objects.splitlines() if line.startswith("?")})
+        if not all(SHA.fullmatch(oid) for oid in missing):
+            raise InstrumentFailure("invalid_missing_object")
+        if missing:
+            # stdin avoids argv limits; explicit object wants avoid one promisor
+            # request per show/blame. --no-filter overrides clone's stored filter.
+            self.run(
+                "fetch",
+                "--quiet",
+                "--no-tags",
+                "--no-write-fetch-head",
+                "--no-filter",
+                "--stdin",
+                "origin",
+                input_data=("\n".join(missing) + "\n").encode(),
+            )
+            remaining = self.run("rev-list", "--objects", "--missing=print", "--stdin", input_data=data)
+            if any(line.startswith("?") for line in remaining.splitlines()):
+                raise InstrumentFailure("missing_blobs_after_fetch")
+
     def disk_bytes(self) -> int:
         # git churns .git/objects/pack (incl. transient *.rev) during on-demand blob
         # fetches, so a path rglob just listed can vanish before stat(): skip it.

tests/test_unseen_draft.py

@@ -242,14 +242,14 @@ def test_systemic_extraction_guard_boundaries(tmp_path, monkeypatch, successes,
     expected_failures = min(failures, 201 if successes else 25)
     assert report["status"] == status
     assert report["rejections"] == {"extract_git_exit_128": expected_failures}
-    assert clones.counts["clones"] == successes + expected_failures
+    assert successes + expected_failures <= clones.counts["clones"] <= successes + min(failures, expected_failures + 5)
     assert not (journal_path / "draft-p1.toml").exists()
     if status == "instrument_failure":
         assert report["error_type"] == "InstrumentFailure"
-        # Replayed failures must trip the same guard without attempting more clones.
+        # Replayed failures trip the same guard; only missing speculation may run.
         resumed = d.run(**kwargs)
         assert resumed["status"] == status and resumed["rejections"] == report["rejections"]
-        assert clones.counts["clones"] == successes + expected_failures
+        assert successes + expected_failures <= clones.counts["clones"] <= successes + min(failures, expected_failures + 5)
 
 
 @pytest.mark.parametrize(
@@ -281,7 +281,7 @@ def test_non_instrument_extraction_bug_propagates_and_aborts(tmp_path):
         d.extract_repository(candidate, api, clones, set(), 7)
     report = d.run(tmp_path, github=api, osv=FakeOSV(), clones=clones, used=used, seed=7, known_empty=set(used.sources))
     assert report["status"] == "instrument_failure" and report["error_type"] == "ValueError"
-    assert report["rejections"] == {} and clones.counts["clones"] == 2
+    assert report["rejections"] == {} and 2 <= clones.counts["clones"] <= 3
 
 
 def test_ordinary_rejections_do_not_trip_extraction_guard(tmp_path, monkeypatch):
@@ -345,11 +345,13 @@ def test_systemic_zero_source_is_instrument_failure_with_recorded_rejections(tmp
     report = d.run(tmp_path, github=FakeAPI(30), osv=FakeOSV(), clones=clones, used=used, seed=1, known_empty=set(used.sources))
     assert report["status"] == "instrument_failure"
     assert report["rejections"] == {"extract_zero_source_bytes": 25}
-    assert clones.counts["clones"] == 25
+    assert 25 <= clones.counts["clones"] <= 30
 
 
-@pytest.mark.parametrize("deadline_args,deadline", [([], 240.0), (["--extract-deadline", "12.5"], 12.5)])
-def test_cli_uses_environment_and_injected_stores(tmp_path, monkeypatch, capsys, deadline_args, deadline):
+@pytest.mark.parametrize(
+    "deadline_args,deadline,workers", [([], 240.0, 6), (["--extract-deadline", "12.5", "--extract-workers", "2"], 12.5, 2)]
+)
+def test_cli_uses_environment_and_injected_stores(tmp_path, monkeypatch, capsys, deadline_args, deadline, workers):
     import json
 
     from openultrasast import config
@@ -368,6 +370,7 @@ def test_cli_uses_environment_and_injected_stores(tmp_path, monkeypatch, capsys,
     assert d.main(args + deadline_args) == 1
     output = capsys.readouterr().out
     assert json.loads(output)["extract_deadline_seconds"] == deadline
+    assert json.loads(output)["extract_workers"] == workers
     assert "instrument_failure" in output and "fake-secret" not in output and "fixture/" not in output
     assert seen[-1] == "fake-secret"
     assert all(kwargs == {"read_only": True} for _, kwargs in seen[:-1])
@@ -390,7 +393,16 @@ def test_timeout_skips_repository_and_draft_continues(tmp_path, monkeypatch, tim
     monkeypatch.setattr(d, "repository_changes", lambda *a, **kw: (entries()[0]["changes"], {}, {"bytes": 1, "files": 1}))
     clones = d.Clones(tmp_path / "repo", tmp_path / "cache", runner=runner, deadline_seconds=240.0)
     used = e.UsedSet()
-    report = d.run(tmp_path / "repo", github=FakeAPI(2), osv=FakeOSV(), clones=clones, used=used, seed=1, known_empty=set(used.sources))
+    report = d.run(
+        tmp_path / "repo",
+        github=FakeAPI(2),
+        osv=FakeOSV(),
+        clones=clones,
+        used=used,
+        seed=1,
+        known_empty=set(used.sources),
+        extract_workers=1,
+    )
     assert report["status"] == "insufficient_repositories"
     assert report["rejections"] == {"extraction_timeout": 1}
     assert report["repositories"] == 1 and clone_count == 2
@@ -450,6 +462,9 @@ def test_repository_changes_excludes_all_advisory_sites(monkeypatch):
                 return "app.py\n"
             return args[-1] if args[0] == "rev-parse" else ""
 
+        def materialize(self, revisions):
+            assert revisions == ["HEAD", "a" * 40, "d" * 40]
+
         def blob(self, revision, path):
             return "def handler():\n    return 1\n"
 
@@ -483,7 +498,8 @@ def test_repository_changes_excludes_all_advisory_sites(monkeypatch):
     monkeypatch.setattr(d.extract, "draw_ordinary", draw)
     _, _, proof = d.repository_changes(FakeGit(), candidate, {"d" * 40}, seed=1)
     assert proof["bytes"] > 0
-    assert len([c for c in calls if c[0] == "fetch"]) == 2
+    assert len([c for c in calls if c[0] == "fetch"]) == 1
+    assert "--no-tags" in next(c for c in calls if c[0] == "fetch")
 
 
 def test_candidate_404_is_counted_but_service_failure_aborts(tmp_path):
@@ -911,3 +927,197 @@ def test_decide_http_failure_systemic_guard_aborts(tmp_path):
     report = d.run(tmp_path, github=api, osv=FakeOSV(), clones=FakeClones(), used=used, seed=7, known_empty=set(used.sources))
     assert report["status"] == "instrument_failure" and report["error_type"] == "InstrumentFailure"
     assert report["rejections"].get("decide_http_failure") == 25
+
+
+def test_parallel_pool_is_byte_identical_and_fork_selection_is_seed_ordered(tmp_path, monkeypatch):
+    import threading
+
+    journal_module = importlib.import_module("benchmarks.unseen.journal")
+    monkeypatch.setattr(journal_module.os, "fsync", lambda fd: None)
+    coordinator = threading.get_ident()
+    ordered = s.ordered_advisories(list(FakeAPI(310).advisories()), 7)
+    names = [row["references"][0].split("/commit/")[0].removeprefix("https://github.com/") for row in ordered]
+    ranks = {name: i for i, name in enumerate(names)}
+    networks = {names[1]: names[0], names[3]: names[2], names[5]: names[4]}
+    extracted, selected, snapshots = [], [], []
+    worker_count = [1]
+    second_done = threading.Event()
+    active, peak = [0], [0]
+    lock = threading.Lock()
+    changes = entries()[0]["changes"]
+
+    class API(FakeAPI):
+        def get_repo(self, name):
+            assert threading.get_ident() == coordinator
+            result = super().get_repo(name)
+            if name in networks:
+                result["source"] = {"full_name": networks[name]}
+            return result
+
+        def get_commit(self, name, sha):
+            assert threading.get_ident() == coordinator
+            if ranks.get(name) in {5, 8}:
+                raise ValueError("http_failure")
+            return super().get_commit(name, sha)
+
+        def get_license(self, name, ref):
+            assert threading.get_ident() == coordinator
+            return super().get_license(name, ref)
+
+    def extract_changes(git, candidate, fixes, *, seed):
+        assert threading.get_ident() != coordinator
+        rank = ranks[candidate.repository]
+        with lock:
+            active[0] += 1
+            peak[0] = max(peak[0], active[0])
+        try:
+            if rank == 0 and worker_count[0] > 1:
+                assert second_done.wait(10), "extraction did not overlap"
+            if rank == 1:
+                second_done.set()
+            with lock:
+                extracted.append(rank)
+            if rank in {0, 7}:
+                raise d.InstrumentFailure("git_exit_128")
+            if rank == 6:
+                raise d.extract.ExtractionTimeout("extraction_timeout")
+            return copy.deepcopy(changes), {"eligible": 40}, {"bytes": 12, "files": 1}
+        finally:
+            with lock:
+                active[0] -= 1
+
+    original_assign, original_put = d.assign_slices, d.Journal.put
+
+    def assign(rows, *, seed):
+        selected.append([row["repository"] for row in rows])
+        return original_assign(rows, seed=seed)
+
+    def put(self, *args):
+        assert threading.get_ident() == coordinator
+        return original_put(self, *args)
+
+    monkeypatch.setattr(d, "repository_changes", extract_changes)
+    monkeypatch.setattr(d, "assign_slices", assign)
+    monkeypatch.setattr(d.Journal, "put", put)
+    for workers in (1, 6):
+        worker_count[0] = workers
+        second_done.clear()
+        extracted.clear()
+        peak[0] = 0
+        path = tmp_path / str(workers)
+        used = e.UsedSet()
+        report = d.run(
+            tmp_path / "repo",
+            github=API(310),
+            osv=FakeOSV(),
+            clones=FakeClones(),
+            used=used,
+            seed=7,
+            known_empty=set(used.sources),
+            journal_path=path,
+            extract_workers=workers,
+        )
+        assert report["status"] == "draft", report
+        assert report["rejections"] == {"fork_network": 2, "extract_git_exit_128": 2, "extraction_timeout": 1, "decide_http_failure": 1}
+        assert 1 <= peak[0] <= workers
+        if workers == 6:
+            assert peak[0] > 1 and extracted.index(1) < extracted.index(0)
+        snapshots.append(
+            (report["manifest_sha256"], *(path.joinpath(name).read_bytes() for name in ("draft-p1.toml", "private-draft-p1.toml")))
+        )
+    # Independent serial oracle: first successful repository wins each network.
+    expected = [name for i, name in enumerate(names) if i not in {0, 3, 5, 6, 7, 8}][:300]
+    assert selected == [expected, expected]
+    assert snapshots[0] == snapshots[1]
+
+
+@pytest.mark.parametrize("workers", [0, -1])
+def test_invalid_extract_workers_rejected_before_io(tmp_path, workers):
+    used, api = e.UsedSet(), FakeAPI()
+    with pytest.raises(ValueError, match="invalid_extract_workers"):
+        d.run(tmp_path, github=api, osv=FakeOSV(), clones=FakeClones(), used=used, seed=1, extract_workers=workers)
+    assert api.calls == 0
+
+
+def test_graceful_stop_drains_workers_and_resume_with_different_width_reuses_results(tmp_path, monkeypatch):
+    import json
+
+    journal_module = importlib.import_module("benchmarks.unseen.journal")
+    monkeypatch.setattr(journal_module.os, "fsync", lambda fd: None)
+    changes = entries()[0]["changes"]
+    monkeypatch.setattr(d, "repository_changes", lambda *a, **kw: (changes, {}, {"bytes": 12, "files": 1}))
+    path, full = tmp_path / "resumed", tmp_path / "full"
+    used = e.UsedSet()
+
+    def run(destination, workers, clones):
+        return d.run(
+            tmp_path / "repo",
+            github=FakeAPI(),
+            osv=FakeOSV(),
+            clones=clones,
+            used=used,
+            seed=7,
+            known_empty=set(used.sources),
+            journal_path=destination,
+            extract_workers=workers,
+        )
+
+    assert run(full, 1, FakeClones())["status"] == "draft"
+    original = d.Journal.put
+
+    def stop(self, stage, key, value):
+        original(self, stage, key, value)
+        if stage == "extraction" and len(self.rows[stage]) == 23:
+            (self.path / "STOP").touch()
+
+    monkeypatch.setattr(d.Journal, "put", stop)
+    first = FakeClones()
+    assert run(path, 6, first)["status"] == "stopped"
+    progress = json.loads((path / "progress.json").read_text())
+    completed = progress["counts"]["extraction"]
+    assert progress["status"] == "stopped"
+    assert 23 <= completed <= 28
+    assert first.counts["clones"] == completed
+    (path / "STOP").unlink()
+    monkeypatch.setattr(d.Journal, "put", original)
+    rest = FakeClones()
+    assert run(path, 1, rest)["status"] == "draft"
+    assert rest.counts["clones"] == 300 - completed
+    for name in ("draft-p1.toml", "private-draft-p1.toml"):
+        assert (path / name).read_bytes() == (full / name).read_bytes()
+
+
+@pytest.mark.parametrize(
+    "failure,reason",
+    [
+        (ValueError("http_failure"), "extract_http_failure"),
+        (OSError("read_failed"), "extract_read_failed"),
+        (d.InstrumentFailure("read_failed"), "extract_read_failed"),
+        (e.RepositoryNotFound("not_found"), "repository_or_commit_not_found"),
+    ],
+)
+def test_parallel_license_failure_is_a_resilient_skip(tmp_path, monkeypatch, failure, reason):
+    changes = entries()[0]["changes"]
+    changes[0]["head"] = "c" * 40
+    monkeypatch.setattr(d, "repository_changes", lambda *a, **kw: (changes, {}, {"bytes": 12, "files": 1}))
+
+    class API(FakeAPI):
+        def get_license(self, name, ref):
+            if name == "fixture/project000" and ref == "c" * 40:
+                raise failure
+            return super().get_license(name, ref)
+
+    used = e.UsedSet()
+    report = d.run(
+        tmp_path / "repo",
+        github=API(2),
+        osv=FakeOSV(),
+        clones=FakeClones(),
+        used=used,
+        seed=7,
+        known_empty=set(used.sources),
+        journal_path=tmp_path / "journal",
+        extract_workers=6,
+    )
+    assert report["status"] == "insufficient_repositories"
+    assert report["repositories"] == 1 and report["rejections"] == {reason: 1}

tests/test_unseen_extract.py

@@ -372,3 +372,75 @@ def test_disk_bytes_skips_files_that_vanish_mid_walk(tmp_path):
 
     git.path = type("P", (), {"rglob": lambda self, pattern: iter([*real, Vanished()])})()
     assert git.disk_bytes() == 30
+
+
+def test_partial_clone_batch_matches_lazy_bytes_without_per_file_fetches(history, tmp_path):
+    """Real promisor packs over file:// only: count lazy fetches via Git's trace."""
+    import json
+    import os
+
+    git, commit, candidate, _, _, _, _ = history
+    for revision in range(12):
+        commit(
+            f"def handler(value):\n    return int(value) + {revision}\n",
+            extra={f"file{revision * 5 + i}.py": f"x = {revision * 5 + i}\n" for i in range(5)},
+        )
+    git.run("config", "uploadpack.allowFilter", "true")
+    git.run("config", "uploadpack.allowAnySHA1InWant", "true")
+    baseline = [e.introducing(git, candidate), e.ordinary_history(git, "HEAD")]
+    results, fetches = [], []
+    for batch in (False, True):
+        path = tmp_path / ("batch" if batch else "lazy")
+        trace = tmp_path / (path.name + ".trace")
+        # Local file transport only, no network. No checkout: all blobs start missing.
+        subprocess.run(
+            ["git", "clone", "--quiet", "--filter=blob:none", "--no-checkout", tmp_path.as_uri(), str(path)],
+            check=True,
+            capture_output=True,
+        )
+        commands = []
+
+        def runner(argv, commands=commands, trace=trace, **kwargs):
+            commands.append(argv)
+            return subprocess.run(argv, **kwargs, env={**os.environ, "GIT_TRACE": str(trace), "GIT_ALLOW_PROTOCOL": "file"})
+
+        partial = e.Git(path, runner=runner)
+        missing = partial.run("rev-list", "--objects", "--missing=print", "HEAD")
+        assert sum(line.startswith("?") for line in missing.splitlines()) >= 60
+        if batch:
+            partial.materialize(["HEAD", candidate.fix])
+        results.append([e.introducing(partial, candidate), e.ordinary_history(partial, "HEAD")])
+        assert partial.files_read > 0 and partial.bytes_read > 0
+        fetches.append(sum("built-in: git fetch " in line for line in trace.read_text().splitlines()))
+        if batch:
+            assert sum("fetch" in argv for argv in commands) == 1
+            partial.materialize(["HEAD", candidate.fix])  # Already hydrated: no second fetch.
+            assert sum("fetch" in argv for argv in commands) == 1
+    assert json.dumps(results[0], sort_keys=True).encode() == json.dumps(results[1], sort_keys=True).encode()
+    assert results[1] == baseline
+    assert fetches[0] >= 12 and fetches[1] == 1
+
+
+def test_materialize_timeout_uses_repository_deadline_and_checks_completeness(tmp_path):
+    oid = "a" * 40
+    calls = []
+    now = [0.0]
+
+    def runner(argv, **kwargs):
+        calls.append((argv, kwargs))
+        if "fetch" in argv:
+            assert kwargs["timeout"] == 3
+            assert kwargs["input"] == (oid + "\n").encode()
+            raise subprocess.TimeoutExpired(argv, kwargs["timeout"])
+        now[0] += 7
+        return subprocess.CompletedProcess(argv, 0, ("?" + oid + "\n").encode(), b"")
+
+    with pytest.raises(e.ExtractionTimeout, match="extraction_timeout"):
+        e.Git(tmp_path, runner=runner, deadline_seconds=10, _now=lambda: now[0]).materialize(["HEAD"])
+    assert len(calls) == 2
+
+    def incomplete(argv, **kwargs):
+        return subprocess.CompletedProcess(argv, 0, ("?" + oid + "\n").encode(), b"")
+
+    with pytest.raises(e.InstrumentFailure, match="missing_blobs_after_fetch"):
+        e.Git(tmp_path, runner=incomplete).materialize(["HEAD"])