fix(email): stop holding a transaction across model and provider calls

apps/services/gateway/gateway/routes/email/automation/cleanup.py

@@ -557,6 +557,13 @@ async def sweep_uncategorized(
             except HTTPException:
                 summary["error"] = "account not found"
                 return summary
+            # Close the read transaction the loader just opened before the
+            # network round-trip. A session left `idle in transaction` across a
+            # provider call holds ACCESS SHARE on everything it touched, which
+            # is how a migration's ALTER TABLE ends up queued behind a sweep —
+            # see the 2026-08-06 write-up. Nothing is pending, so this only
+            # ends the transaction.
+            await db.commit()
             if not await provider.authenticate():
                 # ABORT — do NOT continue with provider=None. A local-only label
                 # is logged APPLIED but Outlook (categories-authoritative) wipes

apps/services/gateway/gateway/routes/email/automation/replyzero.py

@@ -1152,6 +1152,28 @@ async def _maybe_classify_threads(account_id: str) -> None:
                 # CC send. Capped per cycle; threads past the cap are left
                 # UNWRITTEN (not blind-AWAITING) so they retry next cycle.
                 if sent_handled < _REPLY_DETERMINE_CAP:
+                    # End OUR transaction before the model call.
+                    #
+                    # `_mark_thread_replied` opens its own session and spends a
+                    # full LLM determination in it. Ours has been in a
+                    # transaction since the reads above, so without this commit
+                    # it sits `idle in transaction` — holding ACCESS SHARE on
+                    # every table it touched — for the length of that call,
+                    # times up to _REPLY_DETERMINE_CAP consecutive sent threads.
+                    #
+                    # That is the exact shape that took production down on
+                    # 2026-08-06: a session parked mid-LLM-call, a migration's
+                    # ALTER TABLE queued behind its lock, and Postgres's FIFO
+                    # lock queue then stalling every later reader of that table.
+                    #
+                    # The acb_llm wall-clock ceiling and the 600s
+                    # idle_in_transaction_session_timeout bound that damage;
+                    # this removes its cause. The ceiling alone is NOT enough
+                    # here — 40 capped calls idle well past 600s, at which point
+                    # Postgres kills this session and the upserts accumulated
+                    # below are lost with it. Committing first is what makes the
+                    # backstop and this loop safe together.
+                    await db.commit()
                     await _mark_thread_replied(account_id, r.thread_id)
                     sent_handled += 1
                 # else: overflow — leave it for the next cycle, never guess.

apps/services/gateway/gateway/routes/email/automation/runner.py

@@ -1503,6 +1503,11 @@ async def _process_past_emails_job(
                 store = get_key_store()
                 creds = json.loads(store.decrypt(acc.credentials_encrypted))
                 provider = _instantiate_provider(acc.provider, creds)
+                # End the transaction the SELECT above opened before the network
+                # round-trip — a session parked `idle in transaction` across a
+                # provider call is what queues a migration's ALTER TABLE behind
+                # it (2026-08-06). Nothing pending; this only ends it.
+                await db.commit()
                 if not await provider.authenticate():
                     provider = None
 

tests/unit/test_email_reply_zero.py

@@ -612,3 +612,57 @@ async def fake_upsert(_db, _aid, tid, status, _mid, _mat, reason, **kw):
     assert cap["preserve_done"] is True          # inbound must not clobber DONE
     assert str(cap["reason"]).endswith("· auto")  # fallback tagged for re-check
     assert det.await_args.kwargs["user_sent_last"] is False
+
+
+async def test_backfill_commits_before_every_model_call() -> None:
+    """The backfill must not sit `idle in transaction` across an LLM call.
+
+    `_mark_thread_replied` opens its OWN session and spends a full LLM
+    determination in it. This function's session has been in a transaction since
+    the reads at the top, so without an explicit commit it sits `idle in
+    transaction` — holding ACCESS SHARE on every table it touched — for the
+    length of that call, times up to `_REPLY_DETERMINE_CAP` sent threads.
+
+    That is the shape that took production down on 2026-08-06: a parked session,
+    a migration's ALTER TABLE queued behind its lock, and Postgres's FIFO lock
+    queue then stalling every later reader of the table — including the one the
+    send path makes. The acb_llm ceiling and the 600s
+    idle_in_transaction_session_timeout bound the damage; committing first
+    removes the cause. The ceiling alone is NOT sufficient here: 40 capped calls
+    idle well past 600s, at which point Postgres kills the session and the
+    upserts accumulated in this loop are lost with it.
+
+    Asserts ORDER, not merely that commit was called — a commit after the model
+    call would satisfy a presence check while leaving the lock held throughout.
+    """
+    latest = [
+        _row("t1", "m1", "me@x.com", "sent"),
+        _row("t2", "m2", "me@x.com", "sent"),
+    ]
+    db = _backfill_db(latest, [])
+    order: list[str] = []
+
+    async def _commit() -> None:
+        order.append("commit")
+
+    async def _mark(*_a, **_kw) -> None:
+        order.append("model-call")
+
+    db.commit = AsyncMock(side_effect=_commit)
+
+    with patch.object(_rz, "_get_db", AsyncMock(return_value=db)), \
+            patch.object(_rz, "_load_assistant_about",
+                         AsyncMock(return_value=("", ""))), \
+            patch.object(_rz, "_mark_thread_replied",
+                         AsyncMock(side_effect=_mark)), \
+            patch.object(_rz, "_upsert_thread_status", AsyncMock()), \
+            patch.object(_rz, "_reconcile_thread_labels", AsyncMock()):
+        await m._maybe_classify_threads("acc-1")
+
+    assert "model-call" in order, "the sent-thread path never ran"
+    for i, event in enumerate(order):
+        if event == "model-call":
+            assert i > 0 and order[i - 1] == "commit", (
+                f"model call at position {i} is not immediately preceded by a "
+                f"commit — the transaction is held across it. Order: {order}"
+            )