fix(consolidation): refresh mental models every round, not just the final one (#3411)

[nicoloboschi]

Fixes #3411.

Problem

Mental models with trigger.refresh_after_consolidation: true are silently not refreshed when a bank's consolidation backlog spans more than one round. Only models associated with the final round get refreshed; everything consolidated in preceding rounds is dropped and stays stale indefinitely. This makes auto-refresh effectively non-functional for any bank that bulk-ingests — the only banks whose consolidation needs more than one round.

Root cause

In _run_consolidation_job (consolidation/consolidator.py):

• consolidated_tags accumulates the tags of the memories consolidated in the current call only — reset each round, no state carried across the round-limit re-queue.
• When the round limit is hit, the round skips mental-model refresh entirely and re-queues a successor.
• Only the final round calls _trigger_mental_model_refreshes, prefiltering candidates by that final round's consolidated_tags.

So a tagged, all_strict model whose fact was consolidated in an earlier round is never a candidate (its tag isn't in the final round's consolidated_tags) and stays stale forever. This matches the reported telemetry exactly: across a 36-round drain, rounds 1–35 enqueued zero refreshes; the final round enqueued refreshes only for the last documents ingested.

Fix — accumulate across the chain, flush once at the end

Rather than refresh per-round (which, for a broad/global model during concurrent ingest, could re-refresh it once per round on partial data), the affected tags are accumulated across every round of the round-limited chain and the refresh is flushed a single time on the final round — so every affected model is refreshed exactly once, deduplicated but not dropped.

Mechanics:

• The running tag union is threaded through the re-queue as an internal pending_refresh_tags payload field (mirrors how observation_scopes already flows). Non-final rounds defer refresh and carry the union forward; the final round flushes it via _trigger_mental_model_refreshes.
• Concurrency guard (the lost-update the issue explicitly called out): when the re-queue is deduped into a concurrently-submitted unscoped consolidation — e.g. one a retain enqueued mid-drain, which is exactly the issue's overlapping-consolidation scenario — _submit_async_operation's dedupe-by-bank branch folds the incoming pending_refresh_tags into the surviving op's payload instead of silently dropping them. Safe under the existing FOR NO KEY UPDATE bank-row lock that already serialises submits for a bank.
• submit_async_refresh_mental_model is still called with skip_if_in_flight=True, so two genuinely-parallel chains that both flush at their final rounds don't double-enqueue an overlapping model.

Per-round tag scoping is unchanged (a round still only contributes the tags it actually consolidated); the change is that the union is preserved across the chain and flushed once.

Note on updated_at: mark_consolidated deliberately does not bump updated_at, and the staleness gate keys off updated_at > last_refreshed_at — so a model whose memories were all ingested before the drain is not spuriously re-flagged.

Tests

test_consolidation_multi_round_refresh_3411.py drives a real multi-round consolidation chain through the worker executor (WorkerTaskBackend, so the re-queue lands as a pending op instead of recursing — matching production) with a small round limit, and asserts every entity-scoped model is refreshed exactly once as the backlog drains.

• Before the fix: fails — 7/8 models left stale (11-round drain in the test), only the final round's model refreshed.
• After the fix: passes — 8/8, once each.

test_requeue_dedupe_merges_pending_refresh_tags covers the concurrency guard: a re-queue deduped into an already-pending consolidation must fold its pending_refresh_tags into the survivor's payload.

Full suite green (19 passed): the two new tests plus test_consolidation_round_limit, test_consolidation_reschedule_after_round, test_mental_model_consolidation_refresh_scope, test_mental_model_scheduled_refresh. test_consolidation_round_limit's re-queue assertion was updated to expect the new pending_refresh_tags=None kwarg (that bank is untagged → empty union).

Crash durability (mid-round)

The accumulated union is also durable within a round. Each batch writes its succeeded source facts' tags into the op's task_payload inside that batch's own witness transaction — so the tags share the batch's fate (durable the instant the batch is). If a worker dies after a batch commits but before the round finishes, the op's retry skips those now-consolidated rows, but their tags are still in task_payload, so the final round refreshes their models anyway. The final round reads that durable value as the source of truth. SELECT ... FOR UPDATE serialises the concurrent batches of one op so their unions don't clobber each other.

test_crash_mid_round_preserves_committed_batch_refresh_tags injects a crash right after the first batch commits, asserts its tag survived in task_payload, and that the retry refreshes every model — including the one consolidated pre-crash that the retry never reprocesses.

Commits

1. fix(consolidation): refresh mental models every round … — first cut (per-round + in-flight dedup).
2. refactor(consolidation): accumulate affected tags across rounds, refresh once at drain end — exactly-once design (accumulate + flush once + merge-on-dedupe).
3. fix(consolidation): persist refresh tags per batch … — per-batch crash durability.