Merge unseen-resolution-cache: cache enumeration + resolution to ousast-corpus

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

benchmarks/unseen/bank.py

@@ -105,6 +105,92 @@ class CorpusBank:
         self.client = client
         self.seed = seed
         self.decisions_prefix = f"{prefix}/{seed}/decisions/"
+        self.resolved_prefix = f"{prefix}/{seed}/resolved/"
+        self.advisories_key = f"{prefix}/{seed}/advisories/list.json"
+        self.repositories_prefix = f"{prefix}/{seed}/repositories/"
+
+    @staticmethod
+    def advisory_key(advisory_id: str) -> str:
+        return hashlib.sha256(advisory_id.encode()).hexdigest()
+
+    def _put(self, key: str, body: dict, *, labels: dict) -> None:
+        existing = _retry(self.client.get, key)
+        if existing is not None:
+            previous = json.loads(existing[0])
+            if {field: value for field, value in previous.items() if field != "ts"} == body:
+                return
+        body = {**body, "ts": datetime.now(UTC).isoformat()}
+        data = json.dumps(body, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode()
+        _retry(self.client.put, key, data, labels=labels)
+
+    def bank_resolution(self, advisory_id: str, result: dict, *, contract: str) -> None:
+        decision = "candidate" if "candidate" in result else "reject"
+        self._put(
+            self.resolved_prefix + self.advisory_key(advisory_id) + ".json",
+            {
+                "advisory": advisory_id,
+                "decision": decision,
+                "result": result,
+                "reason": result.get("rejected"),
+                "resolution_contract": contract,
+                "seed": self.seed,
+            },
+            labels={"decision": decision, "contract": contract},
+        )
+
+    def _listed(self, prefix: str, identity: str) -> dict[str, dict]:
+        results = {}
+        for key in sorted(self.client.keys(prefix)):
+            stored = _retry(self.client.get, key)
+            if stored is None:
+                raise ValueError("corpus_resolution_missing")
+            body = json.loads(stored[0])
+            name = body[identity]
+            if key != prefix + self.advisory_key(name) + ".json" or body["seed"] != self.seed:
+                raise ValueError("corpus_resolution_identity")
+            results[name] = body
+        return results
+
+    def resolved(self) -> dict[str, dict]:
+        results = self._listed(self.resolved_prefix, "advisory")
+        for body in results.values():
+            expected = "candidate" if "candidate" in body["result"] else "reject"
+            if body["decision"] != expected:
+                raise ValueError("corpus_resolution_invalid")
+        return results
+
+    def cache_advisories(self, rows: list, *, contract: str) -> None:
+        if self.cached_advisories(contract=contract) is not None:
+            return
+        self._put(
+            self.advisories_key,
+            {"advisories": rows, "advisories_contract": contract, "seed": self.seed},
+            labels={"contract": contract},
+        )
+
+    def cached_advisories(self, *, contract: str) -> list | None:
+        stored = _retry(self.client.get, self.advisories_key)
+        if stored is None:
+            return None
+        body = json.loads(stored[0])
+        if body["seed"] != self.seed:
+            raise ValueError("corpus_advisories_identity")
+        return body["advisories"] if body["advisories_contract"] == contract else None
+
+    def bank_repository(self, name: str, result: dict | None, *, contract: str) -> None:
+        """Retain redirects, fork identity and 404s needed by the all-fix index."""
+        self._put(
+            self.repositories_prefix + self.advisory_key(name) + ".json",
+            {"repository": name, "result": result, "resolution_contract": contract, "seed": self.seed},
+            labels={"contract": contract},
+        )
+
+    def repositories(self, *, contract: str) -> dict[str, dict | None]:
+        return {
+            name: body["result"]
+            for name, body in self._listed(self.repositories_prefix, "repository").items()
+            if body["resolution_contract"] == contract
+        }
 
     @staticmethod
     def key_for(repository: str, fix: str) -> str:

benchmarks/unseen/draft.py

@@ -26,7 +26,7 @@ from pathlib import Path
 
 from . import eligibility, extract, source
 from .bank import CorpusBank, CorpusObjects, corpus_settings, split_decided
-from .extract import EXTRACTION_CONTRACT, Git, InstrumentFailure
+from .extract import EXTRACTION_CONTRACT, RESOLUTION_CONTRACT, Git, InstrumentFailure
 from .journal import Journal, ResolvedGitHub, ResumeMismatch, Stopped, atomic, digest, external
 
 
@@ -209,6 +209,43 @@ def prove_source(git, revision: str) -> dict:
     raise InstrumentFailure("zero_source_bytes")
 
 
+class BankedGitHub(ResolvedGitHub):
+    """Stream the fix index's repository metadata before continuing the draw."""
+
+    def __init__(self, github, journal, bank):
+        super().__init__(github, journal)
+        self.bank = bank
+        self.resolutions = bank.repositories(contract=RESOLUTION_CONTRACT)
+        journal.rows["resolution"].update({name: body["metadata"] if body is not None else None for name, body in self.resolutions.items()})
+
+    def resolve(self, name):
+        key = eligibility.normalize(name)
+        if key not in self.resolutions:
+            try:
+                resolved = eligibility.resolve(name, self)
+            except eligibility.RepositoryNotFound:
+                body = None
+            else:
+                body = {
+                    "metadata": self.journal.rows["resolution"].get(key),
+                    "full_name": resolved.full_name,
+                    "names": sorted(resolved.names),
+                    "network": resolved.network,
+                }
+            self.bank.bank_repository(key, body, contract=RESOLUTION_CONTRACT)
+            self.resolutions[key] = body
+        body = self.resolutions[key]
+        if body is None:
+            raise eligibility.RepositoryNotFound("not_found")
+        return eligibility.Resolution(body["full_name"], frozenset(body["names"]), body["network"])
+
+
+def resolve_repository(name, github):
+    if isinstance(github, BankedGitHub):
+        return github.resolve(name)
+    return eligibility.resolve(name, github)
+
+
 def advisory_index(rows: list[dict], github) -> dict[str, set[str]]:
     """Include ALL fixes in the enumerated population, even from rejected candidates.
 
@@ -219,7 +256,7 @@ def advisory_index(rows: list[dict], github) -> dict[str, set[str]]:
     for row in rows:
         for name, sha in source.fix_links(row):
             try:
-                resolved = eligibility.resolve(name, github)
+                resolved = resolve_repository(name, github)
             except eligibility.RepositoryNotFound:
                 index[name].add(sha)
                 continue
@@ -305,7 +342,7 @@ def decide(row, github, osv, used, selected_networks=()):
         if len(links) != 1:
             raise source.Rejected("fix_links")
         name, _ = next(iter(links))
-        resolved = eligibility.resolve(name, github)
+        resolved = resolve_repository(name, github)
         rejected_by = eligibility.rejection(resolved, used, set())
         if rejected_by:
             raise source.Rejected("used_" + rejected_by)
@@ -382,6 +419,7 @@ def ordered_extractions(
     pool_size,
     bank: CorpusBank | None = None,
     decided=None,
+    stale_resolutions=(),
 ):
     """Bounded in-flight work, completion journaling, and seed-order acceptance.
 
@@ -400,6 +438,15 @@ def ordered_extractions(
     context = multiprocessing.get_context("fork")
     pool = ProcessPoolExecutor(max_workers=workers, mp_context=context)
 
+    def resolution(row):
+        key = row["ghsa_id"]
+        if key not in checkpoint.rows["eligibility"]:
+            result = decide(row, api, osv, used)
+            if bank is not None:
+                bank.bank_resolution(key, result, contract=RESOLUTION_CONTRACT)
+            checkpoint.put("eligibility", key, result)
+        return checkpoint.rows["eligibility"][key]
+
     def remember(repository, fix, result):
         bank.bank(repository, fix, result, contract=EXTRACTION_CONTRACT)
         # Reuse decisions made during this invocation too: another advisory for
@@ -461,13 +508,15 @@ def ordered_extractions(
                     exhausted = True
                     break
                 key = row["ghsa_id"]
+                if key in stale_resolutions:
+                    resolution(row)
                 identity = None
                 if bank is not None:
                     links = source.fix_links(row)
                     if len(links) == 1:
                         repository, fix = next(iter(links))
                         try:
-                            resolved = eligibility.resolve(repository, api)
+                            resolved = resolve_repository(repository, api)
                             repository = resolved.full_name
                         except eligibility.RepositoryNotFound:
                             resolved = None
@@ -508,9 +557,7 @@ def ordered_extractions(
                                     checkpoint.put("extraction", extraction_key, result)
                                 pending.append((row, decision, extraction_key))
                             break
-                if key not in checkpoint.rows["eligibility"]:
-                    checkpoint.put("eligibility", key, decide(row, api, osv, used))
-                decision = checkpoint.rows["eligibility"][key]
+                decision = resolution(row)
                 extraction_key, future = None, None
                 if "candidate" in decision:
                     candidate = source.Candidate(**decision["candidate"])
@@ -619,6 +666,7 @@ def run(
     scratch = ExitStack()
     try:
         decided = bank.decided() if bank is not None else {}
+        banked = bank.resolved() if bank is not None else {}
         accepted, reusable = split_decided(decided, contract=EXTRACTION_CONTRACT)
         decided = {key: value for key, value in decided.items() if key in reusable}
         known_empty = set(known_empty or ())
@@ -638,16 +686,23 @@ def run(
         )
         with checkpoint.open(inputs(used, seed, known_empty, clones, pool_size=pool_size, slices=slices)):
             checkpoint.boundary()
-            api = ResolvedGitHub(github, checkpoint)
+            api = BankedGitHub(github, checkpoint, bank) if bank is not None else ResolvedGitHub(github, checkpoint)
+            checkpoint.rows["eligibility"].update(
+                {key: body["result"] for key, body in banked.items() if body["resolution_contract"] == RESOLUTION_CONTRACT}
+            )
             used = copy.deepcopy(used)  # canonicalization must not mutate the resume input
             eligibility.canonicalize_used(used, api)
             report["used_set"] = used.report(known_empty)
-            try:
-                rows = enumerate_advisories(api, checkpoint)
-            except InstrumentFailure as exc:
-                if str(exc) != "empty_advisories" or not accepted:
-                    raise
-                rows = []
+            rows = bank.cached_advisories(contract=RESOLUTION_CONTRACT) if bank is not None else None
+            if rows is None:
+                try:
+                    rows = enumerate_advisories(api, checkpoint)
+                except InstrumentFailure as exc:
+                    if str(exc) != "empty_advisories" or not accepted:
+                        raise
+                    rows = []
+                if bank is not None:
+                    bank.cache_advisories(rows, contract=RESOLUTION_CONTRACT)
             if bank is not None:
                 # Retain accepted entries even if their advisory disappeared upstream.
                 present = {row["ghsa_id"] for row in rows}
@@ -680,6 +735,7 @@ def run(
                 pool_size,
                 bank=bank,
                 decided=decided,
+                stale_resolutions={key for key, body in banked.items() if body["resolution_contract"] != RESOLUTION_CONTRACT},
             ) as outcomes:
                 for row, decision, result in outcomes:
                     counts[source.ecosystem(row)] += 1

benchmarks/unseen/extract.py

@@ -21,6 +21,8 @@ from .source import Candidate, Rejected
 
 # Bump only when accept/reject outcomes or accepted entry bytes change.
 EXTRACTION_CONTRACT = "v1"
+# Bump only when enumeration, eligibility or source.candidate outputs change.
+RESOLUTION_CONTRACT = "v1"
 
 LANGUAGES = {".py", ".js", ".jsx", ".mjs", ".cjs", ".ts", ".tsx", ".php", ".java"}
 HUNK = re.compile(r"^@@ -(\d+)(?:,(\d+))? \+(\d+)(?:,(\d+))? @@", re.M)

tests/test_unseen_bank.py

@@ -214,3 +214,104 @@ def test_import_does_not_require_boto3(monkeypatch):
     module = importlib.util.module_from_spec(spec)
     spec.loader.exec_module(module)
     assert module.CorpusBank.key_for("a", "b") == CorpusBank.key_for("a", "b")
+
+
+@pytest.mark.parametrize(
+    "result",
+    [
+        {"candidate": {"repository": "org/é"}, "network": "org/root"},
+        {"rejected": "used_pairs"},
+        {"rejected": "family", "network": "org/root"},
+    ],
+)
+def test_resolution_roundtrip_idempotent_and_contract_replacement(result):
+    objects = FakeObjects()
+    bank = CorpusBank(objects, seed=42, prefix="custom")
+    advisory = "GHSA-test/advisory"
+    key = f"custom/42/resolved/{hashlib.sha256(advisory.encode()).hexdigest()}.json"
+    assert bank.advisory_key(advisory) == hashlib.sha256(advisory.encode()).hexdigest()
+    bank.bank_resolution(advisory, result, contract="old")
+    first = objects.objects.copy()
+    bank.bank_resolution(advisory, result, contract="old")
+    assert objects.objects == first and objects.calls.count("put") == 1
+    row = bank.resolved()[advisory]
+    assert row == {
+        "advisory": advisory,
+        "result": result,
+        "decision": "candidate" if "candidate" in result else "reject",
+        "reason": result.get("rejected"),
+        "resolution_contract": "old",
+        "seed": 42,
+        "ts": row["ts"],
+    }
+    assert list(objects.objects) == [key]
+    assert datetime.fromisoformat(row["ts"]).utcoffset().total_seconds() == 0
+    bank.bank_resolution(advisory, result, contract="v1")
+    assert bank.resolved()[advisory]["resolution_contract"] == "v1"
+    assert objects.calls.count("put") == 2
+
+
+def test_advisory_snapshot_written_once_per_contract():
+    objects = FakeObjects()
+    bank = CorpusBank(objects, seed=7)
+    rows = [{"ghsa_id": "GHSA-é", "references": ["a", "b"]}]
+    assert bank.cached_advisories(contract="v1") is None
+    bank.cache_advisories(rows, contract="v1")
+    original = objects.objects.copy()
+    assert bank.cached_advisories(contract="v1") == rows
+    assert bank.cached_advisories(contract="v2") is None
+    bank.cache_advisories([], contract="v1")
+    assert objects.objects == original and objects.calls.count("put") == 1
+    body = json.loads(objects.objects["unseen/7/advisories/list.json"])
+    assert body == {"advisories": rows, "advisories_contract": "v1", "seed": 7, "ts": body["ts"]}
+    bank.cache_advisories([], contract="v2")
+    assert bank.cached_advisories(contract="v2") == []
+    assert bank.cached_advisories(contract="v1") is None
+
+
+@pytest.mark.parametrize("method", ["resolution", "advisories", "repository"])
+@pytest.mark.parametrize("operation", ["get", "put"])
+@pytest.mark.parametrize("failures", [1, 2])
+def test_new_bank_writes_retry_once(method, operation, failures):
+    objects = FakeObjects()
+    bank = CorpusBank(objects, seed=7)
+    objects.failures[operation] = failures
+
+    def write():
+        if method == "resolution":
+            bank.bank_resolution("GHSA-a", {"rejected": "family"}, contract="v1")
+        elif method == "advisories":
+            bank.cache_advisories([], contract="v1")
+        else:
+            bank.bank_repository("org/repo", None, contract="v1")
+
+    if failures == 2:
+        with pytest.raises(OSError):
+            write()
+        assert objects.calls.count(operation) == 2
+    else:
+        write()
+        assert len(objects.objects) == 1
+        assert objects.failures[operation] == 0
+
+
+@pytest.mark.parametrize("method", ["resolved", "cached_advisories", "repositories"])
+@pytest.mark.parametrize("failures", [1, 2])
+def test_new_bank_reads_retry_once(method, failures):
+    objects = FakeObjects()
+    bank = CorpusBank(objects, seed=7)
+    bank.bank_resolution("GHSA-a", {"rejected": "family"}, contract="v1")
+    bank.cache_advisories([], contract="v1")
+    bank.bank_repository("org/old", {"full_name": "org/new"}, contract="v1")
+    bank.bank_repository("org/missing", None, contract="old")
+    objects.failures["get"] = failures
+    read = getattr(bank, method)
+    kwargs = {} if method == "resolved" else {"contract": "v1"}
+    if failures == 2:
+        with pytest.raises(OSError):
+            read(**kwargs)
+    else:
+        result = read(**kwargs)
+        assert result is not None
+        if method == "repositories":
+            assert result == {"org/old": {"full_name": "org/new"}}

tests/test_unseen_draft.py

@@ -1651,6 +1651,8 @@ def test_bank_crash_resume_and_durability_failure(tmp_path, monkeypatch, failure
         puts = 0
 
         def put(self, key, data, labels):
+            if "/decisions/" not in key:
+                return super().put(key, data, labels)
             self.puts += 1
             if failure == "persistent_put" and not self.interrupted:
                 raise OSError("offline simulated bank outage")
@@ -1687,7 +1689,7 @@ def test_bank_crash_resume_and_durability_failure(tmp_path, monkeypatch, failure
         report = run(tmp_path / "resume", bank)
         assert report["status"] == "instrument_failure"
         assert objects.puts == 2
-        assert not objects.objects
+        assert not bank.decided()
         assert not (tmp_path / "resume" / "draft-p1.toml").exists()
         objects.interrupted = True
     resumed = run(tmp_path / "resume", bank)
@@ -1817,3 +1819,218 @@ def test_bank_duplicate_advisory_cannot_overwrite_pair_decision(tmp_path, monkey
     assert all(report["status"] == "draft" for report in reports)
     assert reports[0]["manifest_sha256"] == reports[1]["manifest_sha256"]
     assert bank.decided()[bank.key_for(repository, fix)]["decision"] == "accept"
+
+
+@pytest.mark.parametrize("workers", [1, 4])
+def test_bank_enumeration_resolution_resume_and_contracts(tmp_path, monkeypatch, workers):
+    from benchmarks.unseen.bank import CorpusBank
+
+    monkeypatch.setattr(importlib.import_module("benchmarks.unseen.journal").os, "fsync", lambda fd: None)
+    rows = s.ordered_advisories(list(FakeAPI(8).advisories()), 7)
+    rows[0]["cwes"] = []  # Bank a rejected advisory as well as candidates.
+    names = [next(iter(s.fix_links(row)))[0] for row in rows]
+    changes = entries()[0]["changes"]
+    monkeypatch.setattr(d, "repository_changes", lambda *a, **kw: (copy.deepcopy(changes), {}, {"bytes": 12, "files": 1}))
+    calls = Counter()
+    original_enumerate, original_decide, original_candidate = d.enumerate_advisories, d.decide, s.candidate
+    original_resolve = e.resolve
+
+    def resolve(*args):
+        calls["resolve"] += 1
+        return original_resolve(*args)
+
+    monkeypatch.setattr(e, "resolve", resolve)
+
+    class API(FakeAPI):
+        def advisories(self):
+            return iter(copy.deepcopy(rows))
+
+        def get_repo(self, name):
+            calls["get_repo"] += 1
+            result = super().get_repo(name)
+            if name == names[2]:
+                result["source"] = {"full_name": names[1]}
+            return result
+
+    def enumerate_rows(*args):
+        calls["enumerate"] += 1
+        return original_enumerate(*args)
+
+    def decide(row, *args):
+        calls[row["ghsa_id"]] += 1
+        return original_decide(row, *args)
+
+    def candidate(*args):
+        calls["candidate"] += 1
+        return original_candidate(*args)
+
+    monkeypatch.setattr(d, "enumerate_advisories", enumerate_rows)
+    monkeypatch.setattr(d, "decide", decide)
+    monkeypatch.setattr(s, "candidate", candidate)
+
+    class Objects(CorpusFakeObjects):
+        def __init__(self):
+            super().__init__()
+            self.puts = Counter()
+
+        def put(self, key, data, labels):
+            self.puts[key] += 1
+            super().put(key, data, labels)
+
+    objects = Objects()
+    bank = CorpusBank(objects, seed=7)
+
+    def run(name, bank):
+        used = e.UsedSet()
+        return d.run(
+            tmp_path / "repo",
+            github=API(),
+            osv=FakeOSV(),
+            clones=FakeClones(),
+            used=used,
+            seed=7,
+            known_empty=set(used.sources),
+            journal_path=tmp_path / name,
+            extract_workers=workers,
+            pool_size=6,
+            slices=1,
+            bank=bank,
+        )
+
+    full = run("full", None)
+    assert full["status"] == "draft", full
+    calls.clear()
+    first = run("first", bank)
+    assert first["manifest_sha256"] == full["manifest_sha256"]
+    assert calls["enumerate"] == 1 and calls["candidate"] == 8
+    resolved = bank.resolved()
+    assert len(resolved) == 8
+    assert {body["decision"] for body in resolved.values()} == {"candidate", "reject"}
+    assert all(objects.puts[key] == 1 for key in objects.objects if "/resolved/" in key or "/advisories/" in key)
+    assert all(body["result"] == original_decide(row, API(), FakeOSV(), e.UsedSet()) for row in rows for body in [resolved[row["ghsa_id"]]])
+    calls.clear()
+    second = run("second", bank)
+    assert not calls  # No enumeration, decide, candidate, or upstream repository lookup.
+    assert second["manifest_sha256"] == first["manifest_sha256"]
+    for filename in ("draft-p1.toml", "private-draft-p1.toml"):
+        assert (tmp_path / "second" / filename).read_bytes() == (tmp_path / "full" / filename).read_bytes()
+
+    # A restart with resolution complete but extraction unfinished also uses
+    # the exact cached candidates, without re-running any prefix operation.
+    for key in list(objects.objects):
+        if "/decisions/" in key:
+            del objects.objects[key]
+    remaining = run("remaining-extraction", bank)
+    assert remaining["manifest_sha256"] == full["manifest_sha256"]
+    assert not calls
+
+    stale_id = rows[1]["ghsa_id"]
+    bank.bank_resolution(stale_id, {"rejected": "old"}, contract="old")
+    stale = run("stale-resolution", bank)
+    assert stale["manifest_sha256"] == full["manifest_sha256"]
+    assert calls == {stale_id: 1, "candidate": 1}
+    assert bank.resolved()[stale_id]["resolution_contract"] == d.RESOLUTION_CONTRACT
+    bank.cache_advisories([{"ghsa_id": "obsolete"}], contract="old")
+    calls.clear()
+    renewed = run("stale-advisories", bank)
+    assert renewed["manifest_sha256"] == full["manifest_sha256"]
+    assert calls == {"enumerate": 1}
+    assert s.ordered_advisories(bank.cached_advisories(contract=d.RESOLUTION_CONTRACT), 7) == s.ordered_advisories(
+        [s.reduce_advisory(row) for row in rows], 7
+    )
+
+
+@pytest.mark.parametrize("stage", ["advisories", "repositories", "resolved"])
+@pytest.mark.parametrize("failure", ["interrupt", "put_failure"])
+def test_prefix_streaming_failure_and_restart(tmp_path, monkeypatch, stage, failure):
+    from benchmarks.unseen.bank import CorpusBank
+
+    monkeypatch.setattr(importlib.import_module("benchmarks.unseen.journal").os, "fsync", lambda fd: None)
+    changes = entries()[0]["changes"]
+    monkeypatch.setattr(d, "repository_changes", lambda *a, **kw: (copy.deepcopy(changes), {}, {"bytes": 1, "files": 1}))
+    calls = Counter()
+    original = d.decide
+
+    def decide(row, *args):
+        calls[row["ghsa_id"]] += 1
+        return original(row, *args)
+
+    monkeypatch.setattr(d, "decide", decide)
+
+    class Objects(CorpusFakeObjects):
+        armed = True
+        attempts = 0
+
+        def put(self, key, data, labels):
+            if self.armed and f"/{stage}/" in key:
+                self.attempts += 1
+                if failure == "put_failure":
+                    raise OSError("offline outage")
+                super().put(key, data, labels)
+                raise KeyboardInterrupt
+            super().put(key, data, labels)
+
+    objects = Objects()
+    bank = CorpusBank(objects, seed=7)
+
+    def run(name, bank):
+        used = e.UsedSet()
+        return d.run(
+            tmp_path / "repo",
+            github=FakeAPI(4),
+            osv=FakeOSV(),
+            clones=FakeClones(),
+            used=used,
+            seed=7,
+            known_empty=set(used.sources),
+            journal_path=tmp_path / name,
+            pool_size=4,
+            slices=1,
+            extract_workers=1,
+            bank=bank,
+        )
+
+    if failure == "interrupt":
+        with pytest.raises(KeyboardInterrupt):
+            run("interrupted", bank)
+        assert objects.attempts == 1
+    else:
+        assert run("interrupted", bank)["status"] == "instrument_failure"
+        assert objects.attempts == 2
+    saved = bank.resolved()
+    objects.armed = False
+    calls.clear()
+    resumed = run("resumed", bank)
+    assert resumed["status"] == "draft", resumed
+    assert all(calls[key] == 0 for key in saved)
+    assert len(bank.resolved()) == 4
+    baseline = run("baseline", None)
+    assert resumed["manifest_sha256"] == baseline["manifest_sha256"]
+
+
+def test_banked_fix_index_preserves_renames_multiple_links_and_404s(tmp_path, monkeypatch):
+    from benchmarks.unseen.bank import CorpusBank
+
+    rows = list(FakeAPI(3).advisories())
+    rows[0]["references"] = ["https://github.com/fixture/old/commit/" + "a" * 40]
+    rows[1]["references"] = ["https://github.com/fixture/missing/commit/" + "b" * 40]
+    rows[2]["references"] += ["https://github.com/fixture/old/commit/" + "c" * 40]
+
+    class API(FakeAPI):
+        def get_repo(self, name):
+            if name == "fixture/missing":
+                raise e.RepositoryNotFound("not_found")
+            return super().get_repo(name)
+
+    bank = CorpusBank(CorpusFakeObjects(), seed=7)
+    with d.Journal(tmp_path / "fresh").open({}) as checkpoint:
+        fresh = d.advisory_index(rows, d.BankedGitHub(API(), checkpoint, bank))
+    assert fresh == {"fixture/project000": {"a" * 40, "c" * 40}, "fixture/missing": {"b" * 40}, "fixture/project002": {"a" * 40}}
+
+    def unexpected(*args):
+        pytest.fail("banked fix index must not repeat resolution")
+
+    monkeypatch.setattr(e, "resolve", unexpected)
+    monkeypatch.setattr(API, "get_repo", unexpected)
+    with d.Journal(tmp_path / "resumed").open({}) as checkpoint:
+        assert d.advisory_index(rows, d.BankedGitHub(API(), checkpoint, bank)) == fresh