Worker & background jobs: id collisions, session rollback, email robustness (4 PRs)

[ahueb]

Four verified defects in the worker container: the job loop, the inbound-email poller, and the notification sweeps. src/worker/main.py is at 0 % test coverage — these will be its first tests.

Line numbers current as of origin/main @ b7edcbc (2026-07-30).

Suggested order: V11 (one-liner, prevents unrecoverable data loss) → V2 (unblocks the job loop) → V3 → V4.

PR V11 — Claim a canonical-id writer slot in the worker process (trivial)

The R1 work quantizes minted ids into per-process residue classes so two processes can never mint the same canonical id (src/agent/ids.py; WRITER_SLOT_MODULUS = 100). specs/local-db-conversations.md:65 enumerates the writers as "Three processes … the engine, the web app and GrantBot" — but docker-compose.yml runs four application processes: app (:18), worker (:32), agent (:46), grantbot (:65).

set_default_writer_id is called in src/main.py:107 (WEB), src/agent/main.py:46 (ENGINE_AUX) and src/agent/grantbot.py:633/666 (GRANTBOT) — but not in src/worker/main.py.

The worker does mint: worker/main.py:165-166 → poll_inbound_emails → process_inbound_email → _handle_instruction → record_pi_message (email_inbound.py:512-515) → mint_local_ts(), which falls through to the module default TsMinter(WRITER_WEB) (ids.py:111). So the worker and app containers share residue class 1 — the exact cross-process collision R1 exists to prevent, whose documented resolution is the uq_agent_messages_run_ts conflict handler dropping one message, unrecoverable now that the DB is the only durable store.

Reproduced 2026-07-30: two TsMinter(WRITER_WEB) instances minted the identical id (1785421880.733001). Four distinct writer slots minting 200 ids each produced zero collisions, confirming the mechanism works — the worker is simply not using it.

(The backfill scripts are not affected: they use Slack's ts as the canonical message_ts and never mint.)

Fix: add a WRITER_WORKER id to src/agent/ids.py and claim it at src/worker/main.py entry; correct the "Three processes" count at specs/local-db-conversations.md:65.

PR V2 — Worker session rollback + stale-job reaper (small)

• COR-17 — worker/main.py:100-111 except Exception: … await db.commit() with no await db.rollback() first: when the original error was a DB op, the failure-path commit itself raises.

  Reproduced by driving the real process_job with a job body whose failure is a DB error:

  *** process_job itself raised: DBAPIError:
      (sqlalchemy.dialects.postgresql.asyncpg.Error) <class 'asyncpg.exceptions.InFailedSQLTransactionError'>:
      current transaction is aborted, commands ignored until end of transaction block
      job row after the failure handler: status='processing' last_error=None
  

  Fix: await db.rollback() before the failure-path commit.

• COR-18 — no reaper for processing rows orphaned by a worker crash (started_at is written, never read); failed jobs re-queue with no backoff (:108); completed_at set on failure (:110). (Retries are already bounded by attempts >= max_attempts.) Confirmed in the same run above: after COR-17 fired, the row was left status='processing' with nothing to reclaim it. Fix: a stale-processing reaper + exponential backoff.

PR V3 — Inbound-email poison-pill & instruction-drop hardening (small)

• COR-19 — S3 mail is deleted only on success (email_inbound.py:111; the except at :113 logs without deleting). Two poison pills reproduced 2026-07-30:
  • charset=unknown-8bit → LookupError: unknown encoding: unknown-8bit at .decode(charset, errors="replace") (:271/:275) — the codec lookup fails before errors= applies, so the errors="replace" only looks like a fix;
  • a string rating → TypeError: '<' not supported between instances of 'str' and 'int' at :208.
    Neither object is deleted → reprocessed forever. Also: MAX_REPLIES_PER_TOKEN_PER_HOUR (:27) is defined but never referenced anywhere else in the file; list_objects_v2(MaxKeys=50) (:94) is unpaginated. Fix: guard decode + rating coercion; enforce the rate limit; quarantine/delete poison objects; paginate.
• COR-32 (remainder) — with Slack on, when _handle_instruction can't resolve a bot token/channel it returns False (:527-529,:540-542) yet the caller still calls mark_notification_responded (:234) and the poller deletes the S3 object (:111) → the PI instruction is silently consumed with no post. (The Slack-off subcase now records to the DB inbox at :512-515.) Fix: on a failed post, don't mark-responded and don't delete the object.

PR V4 — Email-notification transactional safety (medium)

None of the three sweeps' per-user except blocks call db.rollback() — the string rollback does not appear anywhere in email_notifications.py (:208/683/874) → one IntegrityError poisons the session and the final commit aborts the cycle for every later user. The row is flushed status="sent" before the SES send (:329, :963) → phantom-sent on SES failure (permanently blocks proposal_review, loses new_proposal). "expired" is never written. consecutive_missed is bumped only at :295, which is unreachable once an unanswered status="sent" row exists (early return at :247-250) — so it caps at 1 and never reaches MISSED_THRESHOLD=3 (:42), leaving the auto-downgrade ladder dead. Fix: db.rollback() per sweep; commit-after-send (or a claim); write an expiry status.

Definition of done: each PR ships a test that fails against the pre-fix code. This issue takes worker/main.py from 0 % coverage.