diff --git a/CHANGELOG.md b/CHANGELOG.md
index bef748b5..f8e37364 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,6 +1,6 @@
 # Changelog
 
-All notable changes to Engraphis are documented here. Format loosely follows
+All notable changes to Engraphis are documented here. Format loosely follows
 [Keep a Changelog](https://keepachangelog.com/); versions use SemVer.
 
 ## [Unreleased]
diff --git a/docs/INDEX_REPAIR_MAINTENANCE.md b/docs/INDEX_REPAIR_MAINTENANCE.md
new file mode 100644
index 00000000..31dfc4aa
--- /dev/null
+++ b/docs/INDEX_REPAIR_MAINTENANCE.md
@@ -0,0 +1,108 @@
+# Repair discovery and writer occupancy
+
+This incremental change depends on the reliability candidate at
+`31da32c06a5ade989608504472e00dfca6f13f94` (PR #203). It changes candidate
+discovery for separate vector indexes. Public entrypoints, repair return fields,
+ranking defaults, schema 18, and transaction-sharing native indexes are unchanged.
+
+## Reproduced problem and acceptance
+
+Repair prioritizes erasure/quarantine cleanup before vector updates. Previously,
+finding one erasure behind 1,000 queued updates acquired SQLite's writer 1,002
+times: registration, 1,000 skipped updates, and one deletion. An independent
+connection could not finish a write while classification held that reservation.
+
+Discovery now reads 100-row pages containing IDs, generation, canonical existence,
+and provenance/metadata. It avoids memory text and vector payloads. Canonical JSON
+decoding and quarantine rules remain shared with the store. Each page is fetched
+before yielding; no read transaction or reader lease spans publication. The keyset
+advances past the last scanned row even if the page yields no matching candidate.
+The store's canonical workspace predicate applies to the memory join, with temporal
+filtering disabled. Out-of-binding and missing records remain cleanup candidates;
+allowed historical records remain indexable. This matches publication's `get_memory`
+view without exposing another workspace's metadata during discovery.
+
+Classification is a hint. Publication still reserves the writer, verifies the
+selected generation, rereads current canonical existence, eligibility and vector
+identity, applies the provider operation, and acknowledges only that generation.
+Stale hints leave recoverable debt. Once the provider-attempt budget is spent,
+iteration stops before requesting another filtered candidate, including after a
+failed deletion. Otherwise the iterator could scan an irrelevant tail after the
+last permitted attempt.
+
+`tests/test_vector_repair_discovery.py` exercises real sync/store sequences for:
+
+- Erasure and quarantine after 105 and 1,000 queued updates; one deletion needs at
+  most two writer reservations, independent of the stable update backlog.
+- An independent writer completing while discovery is deliberately paused.
+- Erasure, same-generation quarantine, vector replacement and restoration between
+  discovery and publication, without stale publication or lost repair debt.
+- Canonical decoding of malformed and legacy metadata/provenance.
+- Workspace-bound cleanup, multiple allowed workspaces and retained historical
+  canonical records; external cleanup does not erase the canonical memory.
+- Early cleanup, both successful and failing, without scanning newer updates after
+  the provider budget is exhausted.
+
+Existing sync and storage tests retain coverage of delayed publication, newer
+generations during provider callbacks, process/restart recovery and native rollback.
+
+## Reproducible measurement
+
+Run the repair-only probe from the checkout being measured:
+
+```sh
+python -m eval.repair_discovery --backlog 1000 --repetition 1 --output repair-1000-1.json
+python -m eval.repair_discovery --backlog 10000 --repetition 1 --output repair-10000-1.json
+```
+
+For a baseline that predates the driver, run the same driver file with `runpy` from
+the baseline checkout. The imported Engraphis package identifies the measured
+source; every report records that source's revision and file hashes independently
+of the driver hash. Use the same Python/dependencies, machine, storage location and
+driver for both sources. Alternate their order across five independent process
+repetitions per backlog. Retain every raw result, including failures.
+The CLI records failure type and attempted configuration with source/driver identity
+and exits nonzero if setup, repair or verification fails; failed work has no success
+measurement. An unwritable output location or forced process termination requires
+the invoking runner to retain its own exit-status/log record.
+
+The synthetic dataset has fixed 32-dimensional vectors and one erasure after the
+declared update backlog. Bulk setup uses canonical store APIs and queue triggers
+on a disposable file-backed SQLite database. Setup is excluded from timing. The
+measured invocation includes discovery, writer acquisition, synchronous fixture
+publication and pending counts. Writer timing includes commit/release overhead;
+nested acknowledgement does not acquire or count a second reservation. The same
+timing instrumentation applies to both sources.
+
+The adapter makes no network calls. No embedding, recall, tokenizer or answer
+generation is measured. This probe does not establish the 100k operating target,
+mixed-workload contention, semantic quality or a provider latency guarantee. The
+complete-engine protocol and hardware gates remain in
+[ENGINE_CAPACITY_PROTOCOL.md](ENGINE_CAPACITY_PROTOCOL.md).
+
+## Compatibility, backout and next dependencies
+
+There is no new migration, policy, service or default. Backout restores the previous
+discovery implementation while retaining the canonical database, durable queue and
+generation checks. Never delete pending repair work to recover availability.
+
+The following remain separate work:
+
+1. Discovery can scan the whole queue, and repeated calls can repeat that scan.
+   Pages bound row count, not metadata bytes or total time. Pending counts also
+   traverse the queue. Measure these costs before adding indexed scheduling state.
+2. A permanently failing oldest deletion can consume repeated small attempt
+   budgets. Durable fairness/backoff and coordination must preserve cleanup priority
+   without acknowledging unapplied work or fabricating canonical generations.
+3. Provider calls still occupy the writer. Moving an arbitrary provider outside it
+   lets a delayed old upsert recreate an erased vector, even if acknowledgement is
+   rejected. Hard deadlines need an adapter-level cancellation/fencing contract;
+   current tests do not prove remote completion safety after a process dies.
+4. Legacy resource imports still perform filesystem/extraction/embedding preparation
+   inside a service writer boundary. A prepared batch must preserve whole-batch
+   rollback, per-file outcomes, provenance, and caller-owned transactions before
+   replacing that path. Removing its transaction decorator alone is insufficient.
+
+An optional background repair worker must coordinate ownership and shut down its
+own connections cleanly. The dependency-light offline library continues to work
+without one. This change does not introduce background scheduling.
diff --git a/docs/REWORK_EXECUTION.md b/docs/REWORK_EXECUTION.md
index 2970b7fd..f705baa4 100644
--- a/docs/REWORK_EXECUTION.md
+++ b/docs/REWORK_EXECUTION.md
@@ -18,6 +18,7 @@ The original checkout's active graph/layout changes remain separate.
 | --- | --- | --- | --- |
 | P1 reproduced defect | Delayed sync publication restored an erased or outdated external vector. | `core/vector_repair.py` publishes current canonical state under the writer reservation and acknowledges the applied generation. `tests/test_sync_index_repair.py` covers delayed publication, erasure, newer updates, provider failures and native rollback. | Arbitrary synchronous providers can still occupy the writer while publishing. |
 | P1 reproduced defect | A blocked vector update prevented later queued erasures from being repaired. | Repair traversal prioritizes canonical deletions and makes bounded progress past deferred updates. Focused regressions cover a one-operation budget, blocked embedding spaces, provider failures and later erasure. | A provider that cannot delete still leaves durable repair debt; deletion is not falsely acknowledged. |
+| P2 reproduced contention | Finding one erasure behind 1,000 updates acquired 1,002 writer reservations. | [Repair discovery](INDEX_REPAIR_MAINTENANCE.md) classifies paged canonical headers before reserving the writer, then revalidates inside it. Tests check independent writer progress, stale hints, and immediate stopping after the attempt budget. | Total discovery, repeated scans, failed-deletion fairness and provider latency remain separate scheduling work. |
 | P1 reproduced defect | Separate engines accepted multiple governed successors of one record. | `core/mutations.py` validates prepared versions and source claims inside the transaction; schema 18 retains content-free command receipts. `tests/test_governed_concurrency.py` exercises corrections, approvals, promotions and merges through independent engines and spawned processes. | Receipts coordinate processes sharing the canonical database; they are not a new distributed multi-database transaction protocol. |
 | P2 reproduced defect | A completed promotion or merge could not be retried after its session closed. | Existing receipts replay before transient active-session and embedding requirements. Tests reopen the engine, disable embedding, replay the result and reject removed successors; new writes still recheck session activity under the writer. | Changed requests are new operations and remain subject to current session and source guards. |
 | P1 reproduced defect | A shared claim key or consolidation lineage collapsed distinct repository facts. | Packing deduplicates repeated canonical IDs, preserves full ownership attribution and budgets it. `tests/test_context_scope_grounding.py` retains distinct repositories, values, conditions and title-bound subjects. | Stronger semantic compression remains an experiment; no ranking default changed. |
diff --git a/engraphis/core/vector_repair.py b/engraphis/core/vector_repair.py
index 44ce768d..3dabf581 100644
--- a/engraphis/core/vector_repair.py
+++ b/engraphis/core/vector_repair.py
@@ -12,7 +12,7 @@
     vector_index_shares_store_transaction,
 )
 from engraphis.core.poisoning import inspection_eligible
-from engraphis.core.store import _is_memory_database_path
+from engraphis.core.store import _is_memory_database_path, _loads
 
 if TYPE_CHECKING:
     from engraphis.core.store import Store
@@ -57,29 +57,50 @@ def canonical_search_required(index, store: "Store", *,
 
 
 def _repair_candidates(store: "Store", target: str, memory_id: Optional[str],
-                       ceiling: tuple[int, str]) -> Iterator[tuple[str, int]]:
-    """Page queue identities without loading vectors or revisiting failed work."""
+                       ceiling: tuple[int, str], *,
+                       cleanup_only: bool) -> Iterator[tuple[str, int]]:
+    """Read bounded header pages; classification is only a publication hint.
+
+    Materialize each page before yielding, without retaining a read transaction.
+    No vector payload or memory text is needed to skip work for the other phase.
+    The publisher still revalidates current canonical state under the writer.
+    """
     after: Optional[tuple[int, str]] = None
     while True:
+        # Match get_memory's instance boundary, including for historical rows.
+        # Keep the predicate on the LEFT JOIN so hidden/orphaned queue entries
+        # remain cleanup candidates instead of disappearing from discovery.
+        scope_where, scope_params = store._where(None, include_invalid=True, alias="m")
+        memory_join = " AND ".join(["m.id=r.memory_id", *scope_where])
         sql = (
-            "SELECT memory_id,generation FROM vector_index_repairs WHERE identity=? "
-            "AND (generation,memory_id)<=(?,?)"
+            "SELECT r.memory_id,r.generation,m.id AS canonical_id,v.id AS vector_id,"
+            "m.provenance,m.metadata FROM vector_index_repairs r "
+            f"LEFT JOIN memories m ON {memory_join} "
+            "LEFT JOIN mem_vectors v ON v.id=r.memory_id "
+            "WHERE r.identity=? AND (r.generation,r.memory_id)<=(?,?)"
         )
-        params: list[Any] = [target, *ceiling]
+        params: list[Any] = [*scope_params, target, *ceiling]
         if memory_id is not None:
-            sql += " AND memory_id=?"
+            sql += " AND r.memory_id=?"
             params.append(memory_id)
         if after is not None:
-            sql += " AND (generation,memory_id)>(?,?)"
+            sql += " AND (r.generation,r.memory_id)>(?,?)"
             params.extend(after)
         rows = store.conn.execute(
-            sql + " ORDER BY generation,memory_id LIMIT 100", params,
+            sql + " ORDER BY r.generation,r.memory_id LIMIT 100", params,
         ).fetchall()
         if not rows:
             return
         after = (int(rows[-1]["generation"]), str(rows[-1]["memory_id"]))
         for row in rows:
-            yield str(row["memory_id"]), int(row["generation"])
+            needs_upsert = (
+                row["canonical_id"] is not None and row["vector_id"] is not None
+                and inspection_eligible(
+                    _loads(row["provenance"], {}), _loads(row["metadata"], {}),
+                )
+            )
+            if needs_upsert != cleanup_only:
+                yield str(row["memory_id"]), int(row["generation"])
 
 
 def repair_vector_index(store: "Store", index: Any, *, embedding_space: str,
@@ -93,7 +114,8 @@ def repair_vector_index(store: "Store", index: Any, *, embedding_space: str,
     public engine's compatibility adapter without coupling this coordinator to it.
     Cleanup precedes upserts, including when ``limit=1``. The limit bounds provider
     attempts; finding cleanup may inspect the whole pending queue in 100-row
-    pages. Repeated calls can rescan pending upserts; this is not a latency bound.
+    read-only header pages. Skipped candidates do not acquire writer reservations.
+    Repeated calls can rescan pending upserts; this is not a latency bound.
     """
     if isinstance(limit, bool) or not isinstance(limit, int) or not 1 <= limit <= 1000:
         raise ValueError("repair limit must be an integer between 1 and 1000")
@@ -125,7 +147,9 @@ def repair_vector_index(store: "Store", index: Any, *, embedding_space: str,
     for cleanup_only in (True, False):
         if attempted >= limit or (not cleanup_only and not vector_writes_ready):
             break
-        for selected_id, generation in _repair_candidates(store, target, memory_id, ceiling):
+        for selected_id, generation in _repair_candidates(
+            store, target, memory_id, ceiling, cleanup_only=cleanup_only,
+        ):
             if attempted >= limit:
                 break
             operation = "delete" if cleanup_only else "upsert"
@@ -176,5 +200,9 @@ def repair_vector_index(store: "Store", index: Any, *, embedding_space: str,
                     # Cleanup has already had its turn; retain fail-fast publication
                     # during an upsert outage instead of repeatedly calling the provider.
                     break
+            # A filtered iterator may scan a long tail before yielding again. Stop
+            # here after success or failure, before asking for another candidate.
+            if attempted >= limit:
+                break
     return {"attempted": attempted, "repaired": repaired,
             "pending": store.vector_index_pending(target) or 0}
diff --git a/eval/repair_discovery.py b/eval/repair_discovery.py
new file mode 100644
index 00000000..a3dc4676
--- /dev/null
+++ b/eval/repair_discovery.py
@@ -0,0 +1,179 @@
+"""Offline repair-only contention probe; not an engine-capacity or quality result.
+
+Setup bulk-seeds synthetic canonical rows on a disposable file-backed database.
+The measured operation repairs one erased external ID after a queue of updates,
+with embedding-space readiness deliberately unavailable. The adapter is a local
+synchronous fixture, so no model, network, provider latency or paid calls occur.
+"""
+from __future__ import annotations
+
+import argparse
+from contextlib import contextmanager
+from pathlib import Path
+import sqlite3
+import tempfile
+import time
+
+import numpy as np
+
+import engraphis
+from engraphis.core.interfaces import MemoryRecord, Scope
+from engraphis.core import vector_repair
+from engraphis.factory import create_memory_engine
+from eval.benchmark import (
+    canonical_json, environment_provenance, git_provenance, sha256_file,
+)
+from eval.vector_scale_storage import _disk, _hardware
+
+
+class _Index:
+    index_identity = "repair-discovery-probe-v1"
+
+    def __init__(self):
+        self.present = {"mem_probe_erased"}
+        self.calls = 0
+
+    def upsert(self, *args, **kwargs):
+        raise AssertionError("this probe must only publish cleanup")
+
+    def delete(self, ids, **kwargs):
+        self.calls += 1
+        self.present.difference_update(ids)
+
+
+def run_probe(backlog: int, repetition: int, *, directory=None) -> dict:
+    """Measure one warm, uncontented invocation, including queue scans and counts."""
+    if type(backlog) is not int or backlog < 1:
+        raise ValueError("backlog must be a positive integer")
+    with tempfile.TemporaryDirectory(prefix="repair-probe-", dir=directory) as temporary:
+        path = Path(temporary) / "probe.db"
+        engine = create_memory_engine(
+            str(path), embed_dim=32, auto_evolve=False, extractor="none",
+            graph_extractor="none", require_exact_backends=True,
+        )
+        try:
+            store = engine.store
+            workspace = store.get_or_create_workspace("repair-probe")
+            index = _Index()
+            target = vector_repair.index_repair_identity(index, store)
+            assert target is not None
+            store.register_vector_index(target)
+            vector = np.zeros(32, dtype=np.float32)
+            vector[0] = 1.0
+            with store.write_transaction():
+                for number in range(backlog + 1):
+                    mid = f"mem_probe_{number:08d}" if number < backlog else "mem_probe_erased"
+                    store.add_memory(MemoryRecord(
+                        id=mid, content=f"Synthetic record {number}.", workspace_id=workspace,
+                        scope=Scope.WORKSPACE, provenance={"source": "offline-repair-probe"},
+                    ), audit=False, commit=False)
+                    store.put_vector(mid, vector, model=engine.embedding_space)
+                # A disposable fixture deletion creates real trigger-maintained debt.
+                store.conn.execute("DELETE FROM memories WHERE id=?", ("mem_probe_erased",))
+            assert store.vector_index_pending(target) == backlog + 1
+            last = store.conn.execute(
+                "SELECT memory_id FROM vector_index_repairs WHERE identity=? "
+                "ORDER BY generation DESC,memory_id DESC LIMIT 1", (target,),
+            ).fetchone()
+            assert last[0] == "mem_probe_erased"
+            pragmas = {name: store.conn.execute(f"PRAGMA {name}").fetchone()[0]
+                       for name in ("journal_mode", "synchronous", "page_size")}
+            disk_before = _disk(path)
+            original = store.write_transaction
+            measured = {"writer_reservations": 0, "writer_acquisition_ms": 0.0,
+                        "writer_reserved_ms": 0.0}
+
+            @contextmanager
+            def timed_writer():
+                # Nested acknowledgement shares the reservation; never count it twice.
+                if store.conn.transaction_owned_by_current_thread():
+                    with original():
+                        yield
+                    return
+                started = time.perf_counter()
+                acquired = None
+                try:
+                    with original():
+                        acquired = time.perf_counter()
+                        measured["writer_reservations"] += 1
+                        measured["writer_acquisition_ms"] += (acquired - started) * 1000
+                        yield
+                finally:
+                    if acquired is not None:
+                        # Includes commit/rollback and the wrapper's release overhead.
+                        measured["writer_reserved_ms"] += (time.perf_counter() - acquired) * 1000
+
+            store.write_transaction = timed_writer
+            try:
+                started = time.perf_counter()
+                result = vector_repair.repair_vector_index(
+                    store, index, embedding_space="unavailable-space", dim=0, limit=1,
+                )
+                elapsed_ms = (time.perf_counter() - started) * 1000
+            finally:
+                store.write_transaction = original
+            assert result == {"attempted": 1, "repaired": 1, "pending": backlog}
+            assert index.calls == 1 and not index.present
+            assert store.get_memory("mem_probe_erased") is None
+            assert store.vector_index_repair_generations(target, ["mem_probe_erased"]) == {}
+            return {"backlog_updates": backlog, "repetition": repetition,
+                    "elapsed_ms": elapsed_ms, **measured, "result": result,
+                    "provider_calls": index.calls, "erasure_verified": True,
+                    "sqlite_pragmas": pragmas, "disk_before_bytes": disk_before}
+        finally:
+            engine.close()
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--backlog", type=int, default=1000)
+    parser.add_argument("--repetition", type=int, default=1)
+    parser.add_argument("--temp-root", type=Path)
+    parser.add_argument("--output", type=Path, required=True)
+    args = parser.parse_args()
+    root = Path(engraphis.__file__).resolve().parents[1]
+    sources = {path.relative_to(root).as_posix(): sha256_file(path) for path in sorted([
+        *root.joinpath("engraphis/core").glob("*.py"),
+        *root.joinpath("engraphis/backends").glob("*.py"),
+        root / "engraphis/factory.py", root / "engraphis/__init__.py",
+        root / "eval/benchmark.py", root / "eval/vector_scale_storage.py",
+    ])}
+    payload = {
+        "schema": "engraphis-repair-discovery-probe/v1",
+        "configuration": {"backlog": args.backlog, "repetition": args.repetition},
+        "source": git_provenance(root), "source_files": sources,
+        "driver_sha256": sha256_file(__file__), "environment": environment_provenance(),
+        "sqlite_version": sqlite3.sqlite_version, "hardware": _hardware(),
+        "boundary": {
+            "operation": "one repair(limit=1), erasure after queued updates",
+            "storage": "file-backed SQLite", "external_index": "synchronous fixture",
+            "vector": "fixed one-hot float32, dimension 32, bulk-seeded",
+            "embedding": "none measured; readiness deliberately unavailable",
+            "tokenizer": "not used", "concurrency": 1, "temperature": "warm after setup",
+            "includes": ["discovery", "writer acquisition", "publication", "pending count",
+                         "same transaction-timing instrumentation on both sources"],
+            "excludes": ["setup", "embedding", "recall", "network", "assertions", "teardown"],
+            "limitations": ["not target hardware qualification", "not a contention load test",
+                            "not a total scan or provider latency bound", "synthetic data"],
+        },
+    }
+    try:
+        payload["measurement"] = run_probe(
+            args.backlog, args.repetition, directory=args.temp_root,
+        )
+        payload["status"] = "ok"
+    except (Exception, KeyboardInterrupt) as exc:
+        # Retain attempted-cell identity even when setup or verification fails.
+        # Do not represent missing measurements as successful zero-cost work.
+        payload["status"] = "failed"
+        payload["failure"] = {"type": type(exc).__name__}
+    args.output.parent.mkdir(parents=True, exist_ok=True)
+    args.output.write_text(canonical_json(payload) + "\n", encoding="utf-8")
+    if payload["status"] != "ok":
+        print(canonical_json(payload["failure"]))
+        raise SystemExit(1)
+    print(canonical_json(payload["measurement"]))
+
+
+if __name__ == "__main__":
+    main()
diff --git a/tests/test_history_scope.py b/tests/test_history_scope.py
index 60b5f82f..c566767b 100644
--- a/tests/test_history_scope.py
+++ b/tests/test_history_scope.py
@@ -1,3 +1,4 @@
+"""Record history retains promoted ancestors without widening caller access."""
 import time
 
 import pytest
diff --git a/tests/test_repair_discovery_probe.py b/tests/test_repair_discovery_probe.py
new file mode 100644
index 00000000..7c169daa
--- /dev/null
+++ b/tests/test_repair_discovery_probe.py
@@ -0,0 +1,35 @@
+"""Failed probe attempts retain source identity instead of disappearing from evidence."""
+import hashlib
+import json
+from pathlib import Path
+import subprocess
+import sys
+
+import pytest
+
+
+@pytest.mark.parametrize("backlog", [0, 5])
+def test_probe_cli_retains_success_and_failure_artifacts(tmp_path, backlog):
+    root = Path(__file__).resolve().parents[1]
+    output = tmp_path / "result.json"
+    result = subprocess.run(
+        [sys.executable, "-m", "eval.repair_discovery", "--backlog", str(backlog),
+         "--output", str(output), "--temp-root", str(tmp_path)],
+        cwd=root, capture_output=True, text=True, timeout=60,
+    )
+    report = json.loads(output.read_text())
+    assert report["configuration"] == {"backlog": backlog, "repetition": 1}
+    assert report["driver_sha256"] == hashlib.sha256(
+        (root / "eval/repair_discovery.py").read_bytes(),
+    ).hexdigest()
+    assert report["source_files"]["engraphis/core/vector_repair.py"] == hashlib.sha256(
+        (root / "engraphis/core/vector_repair.py").read_bytes(),
+    ).hexdigest()
+    if backlog:
+        assert result.returncode == 0 and report["status"] == "ok"
+        assert report["measurement"]["result"] == {"attempted": 1, "repaired": 1, "pending": 5}
+        assert report["measurement"]["erasure_verified"] is True
+    else:
+        assert result.returncode != 0 and report["status"] == "failed"
+        assert report["failure"]["type"] == "ValueError"
+        assert "measurement" not in report
diff --git a/tests/test_vector_repair_discovery.py b/tests/test_vector_repair_discovery.py
new file mode 100644
index 00000000..a5757d17
--- /dev/null
+++ b/tests/test_vector_repair_discovery.py
@@ -0,0 +1,275 @@
+"""Repair discovery must not reserve the writer for unrelated queued updates."""
+from concurrent.futures import ThreadPoolExecutor
+import json
+import threading
+
+import numpy as np
+import pytest
+
+from engraphis.core.sync import SyncEngine
+from engraphis.core.store import Store
+from engraphis.core import vector_repair
+from engraphis.factory import create_memory_engine
+from tests.test_sync_index_repair import ExternalIndex, _bundle, _queue_blocked_upserts_and_cleanup
+
+
+@pytest.fixture
+def queued_index(tmp_path):
+    engine = create_memory_engine(str(tmp_path / "discovery.db"), auto_evolve=False)
+    index = ExternalIndex()
+    sync = SyncEngine(engine.store, embedder=engine.embedder, vector_index=index)
+    try:
+        yield engine, index, sync
+    finally:
+        engine.close()
+
+
+@pytest.mark.parametrize("cleanup", ["erase", "quarantine"])
+@pytest.mark.parametrize("count", [105, 1000])
+def test_cleanup_discovery_writer_cost_does_not_grow_with_upsert_backlog(
+    queued_index, monkeypatch, cleanup, count,
+):
+    engine, index, sync = queued_index
+    target, victim = _queue_blocked_upserts_and_cleanup(
+        engine, index, sync, cleanup=cleanup, count=count,
+    )
+    connection = engine.store.conn
+    execute = type(connection).execute
+    writer_reservations = []
+
+    def measured_execute(conn, sql, *args, **kwargs):
+        if conn is connection and sql.strip().upper() == "BEGIN IMMEDIATE":
+            writer_reservations.append(sql)
+        return execute(conn, sql, *args, **kwargs)
+
+    monkeypatch.setattr(type(connection), "execute", measured_execute)
+    result = vector_repair.repair_vector_index(
+        engine.store, index, embedding_space="unavailable-space", dim=0, limit=1,
+    )
+    assert result == {"attempted": 1, "repaired": 1, "pending": count}
+    assert victim not in index.rows
+    assert engine.store.vector_index_repair_generations(target, [victim]) == {}
+    # Registration plus the actual publication. Inspecting skipped candidates must
+    # not repeatedly contend with other processes for SQLite's single writer.
+    assert len(writer_reservations) <= 1 + result["attempted"]
+
+
+@pytest.mark.parametrize("delete_fails", [False, True])
+def test_repair_stops_discovery_when_the_provider_budget_is_spent(
+    queued_index, monkeypatch, delete_fails,
+):
+    engine, index, sync = queued_index
+    count = 105
+    target, victim = _queue_blocked_upserts_and_cleanup(
+        engine, index, sync, cleanup="erase", count=count,
+    )
+    # Put newer updates after the pending erasure using real generation triggers.
+    with engine.store.write_transaction():
+        for mid, values in engine.store.get_vectors(
+            [f"mem_sync_{number}" for number in range(count)],
+        ).items():
+            engine.store.put_vector(mid, values, model=engine.embedding_space)
+    first = engine.store.conn.execute(
+        "SELECT memory_id FROM vector_index_repairs WHERE identity=? "
+        "ORDER BY generation,memory_id LIMIT 1", (target,),
+    ).fetchone()
+    assert first[0] == victim
+    index.fail = delete_fails
+    eligible = vector_repair.inspection_eligible
+    inspected = []
+
+    def measured_eligibility(provenance, metadata):
+        inspected.append(True)
+        return eligible(provenance, metadata)
+
+    monkeypatch.setattr(vector_repair, "inspection_eligible", measured_eligibility)
+    result = vector_repair.repair_vector_index(
+        engine.store, index, embedding_space="unavailable-space", dim=0, limit=1,
+    )
+    assert result == {"attempted": 1, "repaired": int(not delete_fails),
+                      "pending": count + int(delete_fails)}
+    assert (victim in index.rows) == delete_fails
+    assert bool(engine.store.vector_index_repair_generations(target, [victim])) == delete_fails
+    # The erased first candidate requires no eligibility call. Do not resume the
+    # filtered iterator and scan every later update just to discover the limit.
+    assert inspected == []
+
+
+def test_independent_writer_can_complete_while_discovery_is_paused(queued_index, monkeypatch):
+    engine, index, sync = queued_index
+    _queue_blocked_upserts_and_cleanup(engine, index, sync, cleanup="erase")
+    other = create_memory_engine(engine.store.path, auto_evolve=False)
+    workspace = other.store.get_or_create_workspace("other-project")
+    entered, release = threading.Event(), threading.Event()
+    eligible = vector_repair.inspection_eligible
+    discovery_owned_writer = []
+
+    def pause_discovery(provenance, metadata):
+        if not entered.is_set():
+            discovery_owned_writer.append(engine.store.conn.transaction_owned_by_current_thread())
+            entered.set()
+            assert release.wait(10), "discovery release timed out"
+        return eligible(provenance, metadata)
+
+    monkeypatch.setattr(vector_repair, "inspection_eligible", pause_discovery)
+    try:
+        with ThreadPoolExecutor(max_workers=2) as pool:
+            repair = pool.submit(
+                vector_repair.repair_vector_index, engine.store, index,
+                embedding_space="unavailable-space", dim=0, limit=1,
+            )
+            try:
+                assert entered.wait(10), "discovery did not reach an eligible update"
+                write = pool.submit(
+                    other.remember, "Independent project evidence.", workspace_id=workspace,
+                )
+                memory_id = write.result(timeout=3)
+                assert other.store.get_memory(memory_id) is not None
+            finally:
+                release.set()
+            assert repair.result(timeout=10)["repaired"] == 1
+        assert discovery_owned_writer == [False]
+    finally:
+        release.set()
+        other.close()
+
+
+@pytest.mark.parametrize("transition", ["erase", "quarantine", "replace_vector", "restore_vector"])
+def test_classification_hint_cannot_override_new_canonical_state(
+    queued_index, monkeypatch, transition,
+):
+    engine, index, sync = queued_index
+    assert sync.apply_bundle(_bundle())["added"] == 1
+    mid = "mem_sync_0"
+    target = vector_repair.index_repair_identity(index, engine.store)
+    if transition == "restore_vector":
+        engine.store.conn.execute("DELETE FROM mem_vectors WHERE id=?", (mid,))
+        engine.store.conn.commit()
+    else:
+        engine.store.queue_vector_index_repairs(target, [mid])
+    candidates = vector_repair._repair_candidates
+    changed = []
+    replacement = engine.embedder.embed(["Current canonical replacement."])[0]
+
+    def change_after_classification(*args, **kwargs):
+        for item in candidates(*args, **kwargs):
+            if item[0] == mid and not changed:
+                changed.append(True)
+                if transition == "erase":
+                    engine.store.conn.execute("DELETE FROM memories WHERE id=?", (mid,))
+                    engine.store.conn.commit()
+                elif transition == "quarantine":
+                    # Deliberately keep the vector generation unchanged. Eligibility
+                    # must also be revalidated under the writer, not just the queue ID.
+                    engine.store.conn.execute(
+                        "UPDATE memories SET metadata=? WHERE id=?",
+                        (json.dumps({"quarantine": {"state": "quarantined"}}), mid),
+                    )
+                    engine.store.conn.commit()
+                else:
+                    engine.store.put_vector(mid, replacement, model=engine.embedding_space)
+                    engine.store.conn.commit()
+            yield item
+
+    monkeypatch.setattr(vector_repair, "_repair_candidates", change_after_classification)
+    before = len(index.published)
+    vector_repair.repair_vector_index(
+        engine.store, index, embedding_space=engine.embedding_space, dim=engine.embedder.dim,
+    )
+    assert changed == [True]
+    assert len(index.published) == before
+    pending = engine.store.vector_index_pending(target)
+    assert pending == 1 or (transition == "quarantine" and pending == 0)
+    monkeypatch.setattr(vector_repair, "_repair_candidates", candidates)
+    result = vector_repair.repair_vector_index(
+        engine.store, index, embedding_space=engine.embedding_space, dim=engine.embedder.dim,
+    )
+    assert result == {"attempted": pending, "repaired": pending, "pending": 0}
+    if transition in {"erase", "quarantine"}:
+        assert mid not in index.rows
+    else:
+        np.testing.assert_array_equal(index.rows[mid], engine.store.get_vectors([mid])[mid])
+
+
+@pytest.mark.parametrize("column,raw,cleanup", [
+    ("metadata", "not-json", False),
+    ("metadata", "[]", False),
+    ("metadata", '{"quarantine":{"state":"quarantined"}}', True),
+    ("metadata", '{"provenance":{"quarantined":true}}', True),
+    ("provenance", "not-json", False),
+    ("provenance", "null", False),
+    ("provenance", '{"quarantined":true}', True),
+])
+def test_discovery_uses_canonical_legacy_metadata_decoding(queued_index, column, raw, cleanup):
+    engine, index, sync = queued_index
+    sync.apply_bundle(_bundle())
+    mid = "mem_sync_0"
+    target = vector_repair.index_repair_identity(index, engine.store)
+    # column is a closed, test-owned parameter set, never supplied by a request.
+    engine.store.conn.execute(f"UPDATE memories SET {column}=? WHERE id=?", (raw, mid))
+    engine.store.conn.commit()
+    engine.store.queue_vector_index_repairs(target, [mid])
+    result = vector_repair.repair_vector_index(
+        engine.store, index, embedding_space=engine.embedding_space, dim=engine.embedder.dim,
+    )
+    assert result == {"attempted": 1, "repaired": 1, "pending": 0}
+    assert (mid not in index.rows) == cleanup
+
+
+@pytest.mark.parametrize("allowed", [{"sync-index"}, {"unrelated"}, {"sync-index", "unrelated"}])
+@pytest.mark.parametrize("temporal_column", ["valid_from", "valid_to", "expired_at"])
+def test_discovery_matches_bound_store_visibility_without_hiding_history(
+    queued_index, allowed, temporal_column,
+):
+    engine, index, sync = queued_index
+    sync.apply_bundle(_bundle())
+    mid = "mem_sync_0"
+    target = vector_repair.index_repair_identity(index, engine.store)
+    # Historical canonical vectors remain indexable; only the instance binding
+    # determines whether this target is allowed to retain the row.
+    anchor = 4_000_000_000 if temporal_column == "valid_from" else 1
+    engine.store.conn.execute(
+        f"UPDATE memories SET {temporal_column}=? WHERE id=?", (anchor, mid),
+    )
+    engine.store.conn.commit()
+    engine.store.queue_vector_index_repairs(target, [mid])
+    bound = Store(engine.store.path, allowed_workspaces=allowed)
+    try:
+        visible = bound.get_memory(mid) is not None
+        result = vector_repair.repair_vector_index(
+            bound, index, embedding_space=engine.embedding_space, dim=engine.embedder.dim,
+        )
+        assert result == {"attempted": 1, "repaired": 1, "pending": 0}
+        assert (mid in index.rows) == visible
+        assert bound.vector_index_repair_generations(target, [mid]) == {}
+        assert engine.store.get_memory(mid) is not None  # Cleanup never erases canonical data.
+    finally:
+        bound.close()
+
+
+def test_bound_cleanup_preserves_allowed_work_until_embedding_recovers(queued_index):
+    engine, index, sync = queued_index
+    sync.apply_bundle(_bundle(count=2))
+    allowed = engine.store.get_or_create_workspace("allowed")
+    engine.store.conn.execute(
+        "UPDATE memories SET workspace_id=? WHERE id=?", (allowed, "mem_sync_1"),
+    )
+    engine.store.conn.commit()
+    target = vector_repair.index_repair_identity(index, engine.store)
+    engine.store.queue_vector_index_repairs(target, ["mem_sync_0", "mem_sync_1"])
+    bound = Store(engine.store.path, allowed_workspaces={"allowed"})
+    try:
+        result = vector_repair.repair_vector_index(
+            bound, index, embedding_space="unavailable-space", dim=0, limit=1,
+        )
+        assert result == {"attempted": 1, "repaired": 1, "pending": 1}
+        assert "mem_sync_0" not in index.rows and "mem_sync_1" in index.rows
+        assert engine.store.get_memory("mem_sync_0") is not None
+        assert "mem_sync_0" in engine.store.get_vectors(["mem_sync_0"])
+        result = vector_repair.repair_vector_index(
+            bound, index, embedding_space=engine.embedding_space, dim=engine.embedder.dim,
+        )
+        assert result == {"attempted": 1, "repaired": 1, "pending": 0}
+        assert "mem_sync_0" not in index.rows and "mem_sync_1" in index.rows
+    finally:
+        bound.close()