Reduce writer contention during vector repair discovery

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.

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. |

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}