🔴 Critical — per-row fire has no try/except; one bad row aborts the whole batch and poisons the connection.

The fire path (INSERT + UPDATE + conn.commit(), lines 147–167) runs with no exception handling. If any statement or the commit fails (serialization failure, check-constraint violation, transient socket drop, oversized payload), the exception propagates out of dispatch_due_schedules immediately:

• All remaining due rows in due are dropped this cycle — rows after the failing one never fire (they re-fire next tick, but the batch is non-isolated).
• The connection is left in an aborted-transaction state (InFailedSqlTransaction on every subsequent statement) — there is no conn.rollback() anywhere in this function. The owned-conn path is rescued by _discard_owned_sweep_conn(), but an injected conn (the documented caller-supplied path, and how the real-PG tests drive it) is handed back poisoned.

The docstring (lines 85–87) already promises per-row isolation, but that only covers the cron-parse step, not the DB step. The sibling recover_expired_barriers does exactly the right thing: per-row try/except → conn.rollback() → logger.exception(...) → continue.

Fix: wrap the per-row DB work (both the fire branch here and the baseline UPDATE at lines 124–139) in try/except that rolls back and continues:

try:
    with conn.cursor() as cur:
        cur.execute("INSERT INTO pg_queue_message ...", (...))
        cur.execute("UPDATE pg_periodic_schedule SET last_run_at=%s, next_run_at=%s WHERE pipeline_id=%s", (...))
    conn.commit()
except Exception:
    with contextlib.suppress(Exception):
        conn.rollback()
    logger.exception("PG scheduler: failed to fire pipeline %s — leaving for next tick", pipeline_id)
    continue
fired += 1

Fixed in 4c25ae5 — per-row fire (INSERT+UPDATE+commit) and the baseline UPDATE are now wrapped in try/except → conn.rollback() + logger.exception + continue, exactly like recover_expired_barriers. A row-level failure no longer poisons the (injected or owned) connection or drops the rest of the batch. +atomicity test forcing the advance UPDATE to fail post-INSERT and asserting the enqueue rolled back.

🟠 High — a permanently-invalid cron wedges the row and emits a full traceback every tick, forever.

When compute_next_run raises (croniter raises at construction for empty/whitespace/malformed crons, and the model allows cron_string = ""), this branch logs and continues before touching next_run_at. Because the row is selected on next_run_at IS NULL OR next_run_at <= now() and its next_run_at is never advanced, the same row is re-selected on every tick. At the default ~5s reaper interval that is a logger.exception (full traceback to Sentry) every ~5s per bad row, indefinitely — burying real errors and re-parsing the bad cron each cycle. The schedule also silently never fires, with no durable surfacing to an operator.

Fix: make the bad-cron state self-quiescing and actionable — disable the row (or push next_run_at far out) so it stops being re-selected, and log once at the transition:

except Exception:
    logger.exception("PG scheduler: invalid cron %r for pipeline %s — disabling row", cron_string, pipeline_id)
    with contextlib.suppress(Exception):
        with conn.cursor() as cur:
            cur.execute("UPDATE pg_periodic_schedule SET enabled = FALSE WHERE pipeline_id = %s", (pipeline_id,))
        conn.commit()
    continue

If auto-disable is undesirable by product policy, at minimum dedupe the logging (log at error only on first observation) — re-erroring every 5s is not acceptable.

Fixed in 4c25ae5 — _quiesce_invalid_cron now sets enabled = FALSE on the bad row (+ logs once at the transition), so it stops being re-selected and stops re-emitting a traceback every tick. +test asserting the bad-cron row ends up disabled while the good row still fires.

🟡 Medium — the read step (SELECT now() + due scan) has no rollback on error, deviating from the sibling pattern.

The directly-analogous recover_expired_barriers wraps its read in try/except: conn.rollback(); raise precisely so the connection isn't left in an aborted-txn state. Here, if the SELECT fails mid-read the exception propagates with the transaction still open. As with the fire path, the owned-conn case is rescued by _discard_owned_sweep_conn() but the injected-conn case is left poisoned.

Fix — mirror recover_expired_barriers:

try:
    with conn.cursor() as cur:
        cur.execute("SELECT now()")
        base = cur.fetchone()[0]
        cur.execute("SELECT ... WHERE ...", (base,))
        due = cur.fetchall()
    conn.commit()
except Exception:
    with contextlib.suppress(Exception):
        conn.rollback()
    raise

Fixed in 4c25ae5 — the read step (SELECT now() + due scan) now wraps try/except: conn.rollback(); raise, mirroring recover_expired_barriers, so the connection is never handed back in an aborted-txn state.

🟠 High — the "Never double-fires" property is stated as already-true but currently has no enforcement.

The guarantee depends on the Beat PeriodicTask being disabled for pg_owned rows — done "by the ramp control, next slice", a component that does not exist in the repo yet (confirmed: nothing sets pg_owned=True and no ramp control exists). Until it lands, a row manually flipped to pg_owned=True while its Beat PeriodicTask is still enabled will double-fire. The only thing actually preventing double-fires today is pg_owned defaulting to False (nothing is owned).

This is a cross-table invariant, so a CheckConstraint can't express it. Suggestions: (1) soften this docstring + the models.py comment to state the guarantee is conditional on the ramp control disabling Beat and that, pre-ramp, safety rests on pg_owned=False; (2) before/with the ramp control, add a reconciliation check or integration test asserting no pg_owned=True row has a correspondingly-enabled PeriodicTask.

Fixed in 4c25ae5 — softened both the docstring and the models.py comment: the no-double-fire guarantee is now stated as CONDITIONAL on the ②c ramp control disabling the matching Beat PeriodicTask, and that until it lands safety rests on pg_owned defaulting to False (a manual flip while the PeriodicTask is still enabled would double-fire). An integration check that no pg_owned row has an enabled PeriodicTask will land with the ramp control.

🟡 Medium — type contract is weaker than what the code already has in hand.

• -> dict discards a type that exists: this returns to_payload(...)'s TaskPayload (a TypedDict). The rest of the package returns typed rows everywhere (PgQueueClient.read -> list[QueueMessage], PgReaper.tick -> TickOutcome); this is the one place that drops it. Annotate -> TaskPayload.
• workflow_id: object / pipeline_id: object (lines 53, 55) say "literally anything," the opposite of the real contract. At runtime these are uuid.UUID (psycopg2) or str (tests). Use str | uuid.UUID (and str | uuid.UUID | None for workflow_id).
• Minor asymmetry: workflow_id is truthiness-guarded (line 66) but pipeline_id is unconditionally str(pipeline_id) (line 70) → would render "None" if ever NULL. Safe today (PK), but consider asserting non-None to make the asymmetry intentional.

Fixed in 4c25ae5 — _build_trigger_payload -> TaskPayload; workflow_id: str | uuid.UUID | None, pipeline_id: str | uuid.UUID. (pipeline_id is the PK so never None; left it unconditional str() since the NamedTuple types it non-optional.)

🔵 Low (comment-rot) — "done by the ramp control — next slice".

Two rot risks: (1) it states the Beat-disable as already-arranged, but that component doesn't exist yet (see the no-double-fire comment on pg_scheduler.py); (2) "next slice" is a transient delivery-plan artifact with no anchor once merged to main. Replace "next slice" with a stable reference (the ramp-control ticket, e.g. UN-xxxx, or "a later change") and make the dependency explicit so the absence of enforcement reads as intentional.

Fixed in 4c25ae5 — dropped 'next slice' and the already-arranged phrasing; the comment now states the dependency explicitly (the ②c ramp control must disable Beat atomically) and that pre-ramp safety rests on the False default.

🟡 Medium (test rigor) — assertions weak enough to pass under the exact regressions this module guards against.

• assert fired >= 1 (line 116): with a single seeded row this also passes if the row fired twice — the very double-fire bug the module claims to prevent. Tighten to == 1.
• assert next_run > past (line 124): only checks monotonicity, not correctness — passes even if the next run were computed wrong (e.g. +1 minute instead of next 09:00). Assert equality against the cron-computed value.
• test_null_next_run_baselines_without_firing should also assert dispatch_due_schedules(conn) == 0 to pin "baseline counts as not-fired."

Fixed in 4c25ae5 — fired == 1 (catches a double-fire on a single seeded row); next_run asserted at the cron's 09:00 match (hour/minute/second), not just monotonic; the baseline test now asserts dispatch_due_schedules(conn) == 0.

🟡 Medium (test gaps) — the two headline correctness properties and the most-likely prod bug are untested.

1. Atomicity / no-re-fire (highest value): every test hits the happy commit path. Add a real-PG test that forces the advance UPDATE to fail after the INSERT, expects the exception, then asserts _queued_messages(conn) == [] and next_run_at unchanged — proving the INSERT rolled back with the UPDATE (defends "a crash between them can't re-fire").
2. Timezone: compute_next_run's unit tests use naive datetimes, but production base is tz-aware (SELECT now()). Add a pure test with an aware base asserting the result is aware+correct, and in test_due_owned_row_fires_and_advances assert the persisted next_run_at equals the cron-computed value. This is the single most likely place for a real prod bug.
3. Multi-row: no test that two due, valid, owned rows both fire (fired == 2, two distinct messages). This is the normal production case and the per-row-transaction design specifically supports it.

Fixed in 4c25ae5 — added: (1) atomicity test — a conn wrapper fails the advance UPDATE after the INSERT; asserts no message persisted AND next_run_at unchanged (INSERT rolled back with the UPDATE); (2) tz-aware compute_next_run test (aware base → aware+correct result); (3) multi-row test (two due owned rows → fired == 2, two distinct messages).

🟡 Medium (test gap) — the autouse stub is reasonable, but it hides the scheduler error-isolation wiring.

Stubbing dispatch_due_schedules module-wide is the right call for the leadership/recovery/connection tests. But nothing covers the try/except at reaper.py:495-499 for the scheduler call: test_failed_sweep_discards_owned_conn only exercises the recover_expired_barriers path. If someone deleted the try/except around the scheduler call (or dropped the _discard_owned_sweep_conn()), no test would fail.

Fix (in TestSchedulerTick): set stub_scheduler_tick.side_effect = psycopg2.OperationalError(...) with recover_expired_barriers patched to return [], assert reaper.tick() raises and the owned conn was discarded.

Fixed in 4c25ae5 — added test_scheduler_error_discards_owned_conn in TestSchedulerTick: stub_scheduler_tick.side_effect = psycopg2.OperationalError(...) with recovery patched → asserts tick() raises AND the owned sweep conn was discarded (set to None).

🔵 Info — croniter lock drift. Both backend and workers pin croniter>=3.0.3, but the workers uv.lock resolved to 6.2.2 while the backend uv.lock is at 6.0.0. Same constraint and major version, so behavior is consistent — flagging only so a future lock refresh aligns them.

Acknowledged — lock drift noted (workers resolved croniter 6.2.2 vs backend 6.0.0; same >=3.0.3 constraint + major, so behaviour is consistent). Left as-is for this PR; flagging for a future aligned lock refresh.