fix(graph-maintenance): restore the review fixes without the birth-time enqueue

Drops the "queue every entity at creation" change and keeps the rest of the
review round (dataclass pass results, the Oracle IN-list chunking on the by-unit
enqueue, the per-site enqueue tests, the migration-seed test, the budget's
follow-up-chain test, the stale-comment sweep).

The birth-time enqueue existed to reclaim an entity created in retain's Phase 1
whose Phase-2 link never landed. It is not worth what it costs: such a row is a
single entry in the registry with no postings and no cooccurrences, and #2662
exists because the retry is *supposed* to adopt it — Phase 2 reasserts resolved
parents under FOR KEY SHARE precisely so a pruner cannot delete one out from
under it. Pointing the pruner at every freshly created entity leans on that race
for a leak that is one row wide. #3408 landing the set-based predicate is what
made the trade obviously bad: the expensive half of this job was never those
rows.

Entities created but never linked are therefore no longer proactively reclaimed.
The migration's one-time seed still clears the population a bank has already
accumulated.

Author: nicoloboschi

hindsight-api-slim/hindsight_api/engine/db/ops.py

@@ -81,9 +81,9 @@ def document_serialization_sql(table: str, alias: str) -> str:
 def graph_maintenance_bank_serialization_sql(table: str, alias: str) -> str:
     """SQL predicate serialising ``graph_maintenance`` claims per bank (#3230).
 
-    Every graph_maintenance run is the same bank-wide sweep — the payload carries
-    only ``bank_id``, and ``run_graph_maintenance_job`` drains the whole queue —
-    so a second concurrent run for one bank adds no work. It is worse than
+    Every graph_maintenance run for a bank is interchangeable — the payload
+    carries only ``bank_id``, and ``run_graph_maintenance_job`` drains that bank's
+    queues — so a second concurrent run for one bank adds no work. It is worse than
     useless: ``claim_graph_maintenance_batch`` locks queue rows ``FOR UPDATE``
     *without* ``SKIP LOCKED`` (it is written assuming a single runner per bank),
     so the runs convoy on each other's row locks while each holds a worker slot.

hindsight-api-slim/hindsight_api/engine/graph_maintenance.py

@@ -36,9 +36,9 @@
 
 That follow-up run is *deferred*, not parallel: ``claim_tasks`` will not claim a
 graph_maintenance row for a bank that already has one in flight (#3230). Two
-concurrent runs would do no extra work anyway — each is this same bank-wide
-sweep — while convoying on each other's row locks and holding a worker slot
-each.
+concurrent runs would do no extra work anyway — each drains the same two
+bank-scoped queues — while convoying on each other's row locks and holding a
+worker slot each.
 """
 
 from __future__ import annotations
@@ -229,8 +229,8 @@ async def run_graph_maintenance_job(
     relink = await store.relink_pass(
         backend=backend, fq_table=fq_table, bank_id=bank_id, config=config, deadline=deadline
     )
-    result.relink_units_processed = relink.get("relink_units_processed", 0)
-    result.relink_links_added = relink.get("relink_links_added", 0)
+    result.relink_units_processed = relink.units_processed
+    result.relink_links_added = relink.links_added
 
     # --- Pass 2: entity prune ---
     # Same shape as Pass 1 and owned by the store for the same reason: a
@@ -241,10 +241,10 @@ async def run_graph_maintenance_job(
     # and this is a no-op. Runs after the relink pass so the remaining budget
     # is whatever Pass 1 left.
     prune = await store.entity_prune_pass(backend=backend, fq_table=fq_table, bank_id=bank_id, deadline=deadline)
-    result.entities_examined = prune.get("entities_examined", 0)
-    result.orphan_entities_pruned = prune.get("orphan_entities_pruned", 0)
-    result.stale_cooccurrences_pruned = prune.get("stale_cooccurrences_pruned", 0)
-    result.queues_drained = relink.get("relink_queue_exhausted", True) and prune.get("entity_queue_exhausted", True)
+    result.entities_examined = prune.entities_examined
+    result.orphan_entities_pruned = prune.orphan_entities_pruned
+    result.stale_cooccurrences_pruned = prune.stale_cooccurrences_pruned
+    result.queues_drained = relink.queue_exhausted and prune.queue_exhausted
 
     elapsed = time.time() - job_start
     if not result.queues_drained:

hindsight-api-slim/hindsight_api/engine/memories/base.py

@@ -359,6 +359,36 @@ def build_fact_records(
     return records
 
 
+@dataclass
+class RelinkPassResult:
+    """What one relink drain got through.
+
+    ``queue_exhausted`` is False when the pass stopped on its deadline (or the
+    runaway-iteration cap) with rows still queued — not a failure, since every
+    batch commits before the next is claimed, but the caller needs to know the
+    queue is not empty so it can arrange for the rest to be picked up.
+    """
+
+    units_processed: int = 0
+    links_added: int = 0
+    queue_exhausted: bool = True
+
+
+@dataclass
+class EntityPrunePassResult:
+    """What one entity-prune drain got through.
+
+    ``entities_examined`` counts candidates claimed, not rows deleted: most
+    candidates turn out to be alive and are kept, which is the pass working as
+    intended rather than wasted effort.
+    """
+
+    entities_examined: int = 0
+    orphan_entities_pruned: int = 0
+    stale_cooccurrences_pruned: int = 0
+    queue_exhausted: bool = True
+
+
 class MemoriesExtension(Extension, ABC):
     """Storage + retrieval for memory units and their links, behind one interface.
 
@@ -1157,9 +1187,11 @@ async def enqueue_relink_victims(
         """
         return 0
 
-    async def relink_pass(self, *, backend, fq_table, bank_id: str, config, deadline: float | None = None) -> dict:
-        """Top up links for queued victims. ``{}`` when there is nothing to relink."""
-        return {}
+    async def relink_pass(
+        self, *, backend, fq_table, bank_id: str, config, deadline: float | None = None
+    ) -> "RelinkPassResult":
+        """Top up links for queued victims. All-zero when there is nothing to relink."""
+        return RelinkPassResult()
 
     async def enqueue_entity_prune_candidates(self, *, conn, fq_table, bank_id: str, affected_unit_ids: list) -> int:
         """Queue the entities ``affected_unit_ids`` reference as prune candidates.
@@ -1169,12 +1201,14 @@ async def enqueue_entity_prune_candidates(self, *, conn, fq_table, bank_id: str,
         """
         return 0
 
-    async def entity_prune_pass(self, *, backend, fq_table, bank_id: str, deadline: float | None = None) -> dict:
+    async def entity_prune_pass(
+        self, *, backend, fq_table, bank_id: str, deadline: float | None = None
+    ) -> "EntityPrunePassResult":
         """Prune queued candidate entities and the co-occurrences they stranded.
 
-        ``{}`` when the store keeps no entity postings and so queues nothing.
+        All-zero when the store keeps no entity postings and so queues nothing.
         """
-        return {}
+        return EntityPrunePassResult()
 
 
 __all__ = [
@@ -1194,10 +1228,12 @@ async def entity_prune_pass(self, *, backend, fq_table, bank_id: str, deadline:
     "META_UPDATED_AT",
     "CausalEdgeRecord",
     "DeletePredicate",
+    "EntityPrunePassResult",
     "FactRecord",
     "MemoriesExtension",
     "MemoryPatch",
     "MemoryTxn",
+    "RelinkPassResult",
     "ScanPage",
     "StoredMemory",
     "build_fact_records",

hindsight-api-slim/hindsight_api/engine/memories/pg/graph.py

@@ -42,6 +42,7 @@
     _normalize_datetime,
     compute_semantic_links_ann,
 )
+from ..base import EntityPrunePassResult, RelinkPassResult
 
 logger = logging.getLogger(__name__)
 
@@ -87,6 +88,11 @@
 # contention window a concurrent retain opens.
 _PRUNE_BATCH_MAX_RETRIES = 3
 
+# Unit ids per candidate-lookup round-trip when enqueueing. Bounded by Oracle's
+# 1000-element IN-list limit (ops_oracle expands ``= ANY(...)`` into a literal
+# list), which a bulk delete would otherwise blow straight through.
+_ENQUEUE_LOOKUP_CHUNK = 500
+
 # Cap at 10k edges — the UI can't usefully render more, and uncapped queries
 # on highly-connected graphs (e.g. 1000 nodes with 500k+ edges) are too slow.
 _GRAPH_MAX_EDGES = 10000
@@ -557,7 +563,7 @@ async def relink_pass(
     bank_id: str,
     config: Any,
     deadline: float | None = None,
-) -> dict:
+) -> RelinkPassResult:
     """Drain ``graph_maintenance_queue`` for ``bank_id``, topping up lost links.
 
     Per-iteration loop: claim → top up → commit. We rely on at most one job per
@@ -583,10 +589,8 @@ async def relink_pass(
     keeps the work already done and leaves the rest queued for the next run.
 
     Returns:
-        ``{"relink_units_processed": int, "relink_links_added": int,
-        "relink_queue_exhausted": bool}``. ``relink_queue_exhausted`` is False
-        when the deadline (or the iteration cap) stopped the drain with rows
-        still queued.
+        A :class:`RelinkPassResult`. ``queue_exhausted`` is False when the
+        deadline (or the iteration cap) stopped the drain with rows still queued.
     """
     del config  # accepted for symmetry with stores that tune their own relinking
     ops = backend.ops
@@ -628,11 +632,11 @@ async def relink_pass(
             drained = False
             break
 
-    return {
-        "relink_units_processed": units_processed,
-        "relink_links_added": links_added,
-        "relink_queue_exhausted": drained,
-    }
+    return RelinkPassResult(
+        units_processed=units_processed,
+        links_added=links_added,
+        queue_exhausted=drained,
+    )
 
 
 async def _relink_batch(
@@ -786,13 +790,23 @@ async def enqueue_entity_prune_candidates(
         return 0
 
     ops = _ops_for(conn)
-    return await ops.enqueue_entity_maintenance(
-        conn,
-        fq_table("entity_maintenance_queue"),
-        fq_table("unit_entities"),
-        bank_id,
-        _as_uuids(list(affected_unit_ids)),
-    )
+    queue_table = fq_table("entity_maintenance_queue")
+    ue_table = fq_table("unit_entities")
+    unit_uuids = _as_uuids(list(affected_unit_ids))
+
+    # Chunked because a bulk delete can hand in thousands of unit ids and the
+    # lookup binds them with `= ANY(...)`, which ops_oracle expands into a
+    # literal IN list — Oracle caps those at 1000 elements.
+    enqueued = 0
+    for start in range(0, len(unit_uuids), _ENQUEUE_LOOKUP_CHUNK):
+        enqueued += await ops.enqueue_entity_maintenance(
+            conn,
+            queue_table,
+            ue_table,
+            bank_id,
+            unit_uuids[start : start + _ENQUEUE_LOOKUP_CHUNK],
+        )
+    return enqueued
 
 
 @dataclass
@@ -810,7 +824,7 @@ async def entity_prune_pass(
     fq_table: Callable[[str], str],
     bank_id: str,
     deadline: float | None = None,
-) -> dict:
+) -> EntityPrunePassResult:
     """Drain ``entity_maintenance_queue`` for ``bank_id``, pruning what died.
 
     Per-iteration loop: claim → prune → commit, mirroring :func:`relink_pass`.
@@ -845,10 +859,8 @@ async def entity_prune_pass(
             ``None`` drains to empty.
 
     Returns:
-        ``{"entities_examined": int, "orphan_entities_pruned": int,
-        "stale_cooccurrences_pruned": int, "entity_queue_exhausted": bool}``.
-        ``entity_queue_exhausted`` is False when the deadline stopped the drain
-        with rows still queued.
+        An :class:`EntityPrunePassResult`. ``queue_exhausted`` is False when the
+        deadline stopped the drain with rows still queued.
     """
     from ...db_utils import retry_with_backoff
     from ...memory_engine import acquire_with_retry
@@ -921,12 +933,12 @@ async def _run_batch() -> _PruneBatch:
         orphans_pruned += batch.orphan_entities_pruned
         stale_pruned += batch.stale_cooccurrences_pruned
 
-    return {
-        "entities_examined": examined,
-        "orphan_entities_pruned": orphans_pruned,
-        "stale_cooccurrences_pruned": stale_pruned,
-        "entity_queue_exhausted": drained,
-    }
+    return EntityPrunePassResult(
+        entities_examined=examined,
+        orphan_entities_pruned=orphans_pruned,
+        stale_cooccurrences_pruned=stale_pruned,
+        queue_exhausted=drained,
+    )
 
 
 __all__ = [

hindsight-api-slim/hindsight_api/engine/memories/postgres.py

@@ -23,7 +23,15 @@
 from datetime import datetime
 from typing import Any
 
-from .base import DeletePredicate, MemoriesExtension, MemoryPatch, ScanPage, StoredMemory
+from .base import (
+    DeletePredicate,
+    EntityPrunePassResult,
+    MemoriesExtension,
+    MemoryPatch,
+    RelinkPassResult,
+    ScanPage,
+    StoredMemory,
+)
 from .pg import counts, curation, graph, reads, writes
 
 
@@ -496,7 +504,9 @@ async def enqueue_relink_victims(
             include_affected_units=include_affected_units,
         )
 
-    async def relink_pass(self, *, backend, fq_table, bank_id: str, config, deadline: float | None = None) -> dict:
+    async def relink_pass(
+        self, *, backend, fq_table, bank_id: str, config, deadline: float | None = None
+    ) -> RelinkPassResult:
         return await graph.relink_pass(
             backend=backend, fq_table=fq_table, bank_id=bank_id, config=config, deadline=deadline
         )
@@ -509,7 +519,9 @@ async def enqueue_entity_prune_candidates(self, *, conn, fq_table, bank_id: str,
             affected_unit_ids=affected_unit_ids,
         )
 
-    async def entity_prune_pass(self, *, backend, fq_table, bank_id: str, deadline: float | None = None) -> dict:
+    async def entity_prune_pass(
+        self, *, backend, fq_table, bank_id: str, deadline: float | None = None
+    ) -> EntityPrunePassResult:
         return await graph.entity_prune_pass(backend=backend, fq_table=fq_table, bank_id=bank_id, deadline=deadline)
 
 

hindsight-api-slim/hindsight_api/engine/memory_engine.py

@@ -8775,9 +8775,15 @@ def _parse_edit_date(value: str | None) -> datetime | None:
                 phase2_committed = True
         finally:
             # Entities were resolved (and possibly autocommitted) in Phase 1 but the edit did not
-            # durably apply (row concurrently invalidated → live2 None, or Phase 2 raised): those
-            # entities may now be orphans, and the edit's own relink-victim enqueue never ran. Force
-            # a bank-wide graph-maintenance sweep to reclaim them.
+            # durably apply (row concurrently invalidated → live2 None, or Phase 2 raised), so the
+            # edit's own enqueues rolled back with it. Kick a job for whatever else the bank has
+            # queued; it short-circuits when there is nothing.
+            #
+            # A Phase-1 entity that never got its posting is deliberately NOT chased here. The
+            # prune is queue-driven (#3222) and reads its candidates out of unit_entities, so such
+            # a row is invisible to it — and that is the safe direction: #2662 is precisely the
+            # race where pruning a just-resolved, not-yet-linked parent turned a retry into silent
+            # memory loss, and the retry is meant to adopt that row rather than re-create it.
             if entities_resolved and not (edit_applied and phase2_committed):
                 try:
                     await self.submit_async_graph_maintenance(

hindsight-api-slim/tests/test_db_utils.py

@@ -83,7 +83,7 @@ def test_backoff_delay_is_jittered_and_bounded():
     """Equal-jitter backoff stays in [ceil/2, ceil] and never exceeds max_delay.
 
     The jitter exists so concurrent deadlock retriers don't wake in lock-step
-    and re-collide (see run_graph_maintenance_job's Pass 2/3 sweep). It must
+    and re-collide (see the entity-prune batch in run_graph_maintenance_job). It must
     still keep a floor (no hot-spin) and honour the max_delay cap once the
     exponential term saturates.
     """