fix(maintenance): stop the scheduled mental-model refresh from enqueueing duplicates (#3210)

[nicoloboschi]

Fixes #3210

Why the duplicates happened

The in-flight check the issue asks for already exists — but in the wrong place to
be safe. mental_models_with_cron() excludes models that have a pending/processing
refresh_mental_model op, and that exclusion is a read. The maintenance loop runs
independently in every API/worker process with no leader election, so every process
evaluated the same snapshot ("nothing in flight, due, stale") and inserted its own
operation. That is the ~8× amplification reported: one wave per process, not one wave
per stalled interval.

The fix

The check now rides on the INSERT. submit_async_refresh_mental_model(skip_if_in_flight=True)
inserts the operation row only WHERE NOT EXISTS a pending/processing
refresh_mental_model op for the same (bank_id, mental_model_id) — one statement, so
the check can't be separated from the write:

INSERT INTO async_operations (...)
SELECT $1::uuid, $2, $3, $4::jsonb, $5::text, $6::jsonb
WHERE NOT EXISTS (
    SELECT 1 FROM async_operations
    WHERE bank_id = $2 AND operation_type = $3
      AND status IN ('pending', 'processing')
      AND task_payload->>$7 = $8
)
RETURNING operation_id

Nothing returned ⇒ a refresh was already scheduled ⇒ the caller gets that operation's
id with deduplicated=True, and the loop logs it as "already in flight". The submit
also takes the FOR NO KEY UPDATE bank-row lock that dedupe_by_bank already uses
(#1842) — it is a modifier on the bank-exists SELECT that runs anyway, no extra
round-trip — so two simultaneous submits serialize rather than both passing the READ
COMMITTED snapshot.

No new table, no schema change, no scheduler coordination.

Only the cron scheduler opts in. Explicit user-triggered refreshes (HTTP, MCP,
consolidation-triggered) keep today's behaviour — an on-demand refresh is never
swallowed.

Tests

• test_concurrent_scheduled_submits_queue_one_refresh — two schedulers enqueue the
  same due model with completions stalled → exactly one queued op, and the loser
  reports the winner's id. Verified it fails without the fix (assert 2 == 1).
• test_user_triggered_refresh_is_not_deduplicated — an explicit refresh still queues
  while a scheduled one is in flight.

test_mental_model_scheduled_refresh.py, test_mental_model_delta.py,
test_consolidation_submit_atomic_dedup.py, test_async_submit_bank_not_found.py,
test_maintenance_*, test_worker.py, test_operation_status.py,
test_admin_backup_restore.py and test_migration_shape.py pass locally;
lint.sh + ty clean.

Not changed

MemoryEngine.initialize()'s if self._initialized: return guard is not
concurrency-safe, so two concurrent calls could register two loops in one process —
which would explain the issue's "same scan twice within seconds". Left alone as a
separate concern; with the enqueue now idempotent, an extra scan queues nothing extra.