UN-4176 [FIX] Cut idle PG-queue polling to one autocommit claim per cycle

[muhammad-ali-e]

What

• Idle PG-queue consumers now cost one SQL statement per poll cycle, instead of up to 21.
• Each claim is a single autocommit statement, with no BEGIN/COMMIT around it.
• A worker that drains several queues claims across all of them in one statement, instead of one claim per queue.

Why

• Idle polling is the largest steady source of database statements from the PG queue.
• Every poll cycle sent one claim per queue, and each claim was wrapped in BEGIN + COMMIT, so it cost 3 statements. A 7-queue executor process therefore sent 21 statements every cycle, even with nothing to do.
• Per-statement overhead dominates Cloud SQL CPU, and this polling is the floor we see in production even when no work is running.

How

• PgQueueClient: a connection the client opens itself now runs in autocommit. Every client operation (claim, ack, lease renewal, send) is one statement, so the transaction wrapper only added round trips.
  • Injected connections are never changed: they may be the caller's open transaction, e.g. the agentic fan-out outbox.
  • create_pg_connection itself stays non-autocommit, which pg_barrier relies on.
• New PgQueueClient.read_any(): claims from several queues in one statement.
  • Each queue gets its own indexed top-N probe (LATERAL on the existing partial claim index), so cost does not grow with the backlog.
  • Same claim contract as read(): SKIP LOCKED, lease, read_ct.
  • Duplicate queue names are ignored (a repeated queue would otherwise take batch slots twice).
• QueueMessage.queue_name (new, required field): the queue a message was claimed from. Both read() and read_any() fill it in.
• PgQueueConsumer.poll_once():
  • Multi-queue consumers make one read_any() call per cycle.
  • The next claim starts one past the queue just served (one further along after an empty cycle), so busy queues take equal turns and none starves.
  • Priority still orders messages within a queue.
  • Single-queue consumers keep the existing claim SQL unchanged.
• The poll interval and back-off are unchanged: 0.1 s doubling to 2 s while idle. This PR makes each poll cheaper; it does not make polling less frequent.

Can this PR break any existing features. If yes, please list possible items. If no, please explain why. (PS: Admins do not merge the PR without this section filled)

• No queue lists, routing, task handling, lease/retry or reaper logic change. Every consumer drains exactly the same queues as before.
• Behaviour differences, both intended:
  • A multi-queue consumer runs one message per cycle instead of one per queue; when it finds work it loops again immediately, so throughput is unaffected.
  • A failed claim counts as an empty cycle (run() backs off). A handler failure stops the batch, as each queue's batch did before; the messages already handled are still counted, and the msg_ids left unstarted are logged (they stay claimed until their lease lapses and the reaper re-arms them). Batches above 1 only happen when the lease equals the max runtime (batch_size is otherwise forced to 1).
  • A message whose payload is not a JSON object is now dropped and acked as malformed (logged with its raw content), instead of raising out of _handle on every delivery.
• Autocommit: limited to connections PgQueueClient creates itself. Injected connections and pg_barrier's connection are untouched, so callers that rely on a shared transaction behave as before (e.g. unstract-cloud's agentic fan-out claim, which relies on send() committing its claim INSERT together with the enqueue).
• Dead-connection detection: on a client-owned (autocommit) connection rollback() sends nothing, so it can no longer prove a connection dead. Termination surfaces as OperationalError (dropped and reconnected, covered by a pg_terminate_backend test), and, as a heuristic, a bare psycopg2.DatabaseError is now treated as dead as well, keeping what the failed-rollback path did before (a rare unmapped statement error costs only a needless reconnect).
• Autocommit, accepted trade-off (documented on read_any): a claim commits at statement end, so if the connection dies after the claim but before its rows are read back, those messages stay claimed until their lease lapses (≤ lease_seconds, 120 s by default), then the reaper re-arms them; that delivery counts toward read_ct. Before, the claim would have rolled back. The window is one statement's reply on a dying connection.
• Claim ordering (accepted trade-off, documented on read_any): under contention, the probe locks up to qty ready rows in every queue for the duration of the single statement; candidates that aren't chosen stay ready. A reader racing that statement skips them, so if each queue's only ready row is locked it gets [] and takes the normal empty-poll back-off (starting at poll_interval, 0.1 s). This only happens while several consumers contend for the last rows; idle polls lock nothing.

Database Migrations

• None. read_any() uses the existing partial claim index.

Env Config

• None.

Relevant Docs

• Docstrings on PgQueueClient.read_any, PgQueueClient.conn and PgQueueConsumer.poll_once.

Related Issues or PRs

• UN-4176

Dependencies Versions

• None.

Notes on Testing

• New tests:
  • Unit tests for autocommit on owned connections only, the single claim per cycle and the rotating start queue.
  • Real-Postgres tests for read_any: claiming from whichever queue has work (with its lease and queue_name), rotation and wrap-around, priority within a queue and within a qty>1 batch, a batch spanning queues, candidates that aren't chosen staying ready, a held claim being skipped rather than waited on, two concurrent readers never double-claiming, an owned autocommit connection persisting the claim, and reconnecting after pg_terminate_backend, and duplicate queue names still filling the batch.
  • Real-Postgres consumer tests: multi-queue claim → execute → ack end to end, and two busy queues with an empty one between them alternate.
  • The new guards were mutation-checked (each mutant fails a test): SKIP LOCKED removed from the multi-queue claim, rotation advancing by one regardless of the queue served, rotation following the first instead of the last message of a batch, queue-name de-duplication removed, message/queue_name swapped in the claim's final SELECT, a handler failure continuing past the failed message or returning 0, the non-object payload guard removed, the unstarted-msg_id log slice, the bare-DatabaseError rule removed or widened to statement errors, the read_any lease set to now(), and priority dropped from the batch order.
• Worker test suite: passed against real Postgres (REQUIRE_PG_TESTS=1 for the PG queue client, consumer, dispatch, supervisor, barrier and reaper tests).
• Local Docker stack: all PG workers idle for 5 minutes, measured with pg_stat_statements:
  • queue-polling statements went from 89.0/s to 9.2/s, and transactions from 30.1/s to 9.2/s;
  • BEGIN/COMMIT from the queue client disappeared.
• Functional check: an API deployment completed with the same output and timing before and after the change.

Screenshots

• N/A

Checklist

I have read and understood the Contribution Guidelines.

🤖 Generated with Claude Code

https://claude.ai/code/session_017Ymcknu7CYVmAt7QoRTTax

---

Generated by Claude Code

---

Generated by Claude Code

[CLAassistant]

Thank you for your submission! We really appreciate it. Like many open source projects, we ask that you sign our Contributor License Agreement before we can accept your contribution.
_{You have signed the CLA already but the status is still pending? Let us recheck it.}

[greptile-apps]

via Greptile

Confidence Score: 5/5

[High impact] The PR appears safe to merge; no new actionable issues were found.

Summary

This PR reduces idle PostgreSQL queue polling to one statement per cycle.

• Queue claims use one statement on client-owned connections.
• Multi-queue consumers give busy queues turns across poll cycles.
• Non-object JSON payloads are dropped as malformed messages.

Diagram

%%{init: {'theme': 'neutral'}}%%
flowchart TD
  A[Start poll cycle] --> B{How many queues?}
  B -->|One| C[Claim from that queue]
  B -->|Several| D[Claim across queues in one statement]
  D --> E[Rotate past the last queue claimed]
  C --> F[Run the claimed batch]
  E --> F
  F --> G[Acknowledge completed messages]

Loading

_{Reviews (4) · Last reviewed commit: "UN-4176 Revert out-of-scope review sugge..." · Reviewed by Greptile}

[muhammad-ali-e]

Self-review — round 1

• Tip: 36b4596dd07822895002878fa650a8d151f92b8d (delta: merge base 6da4590 → tip, i.e. the whole PR)
• Review run: /unstract:review-deep. /unstract:review-shallow refused first on its concurrency_primitive gate (eligibility.sh exit 11).
• Run on the branch's local diff against main, because GraphQL (gh pr view) is blocked in this environment; same code as this PR's head.

Mode: INITIAL. This is a local working diff of claude/loving-curie-qdqwqj at 36b4596 against merge-base 6da4590. It was not resolved as a PR, so there is no changedFiles count to check against. The base fetch succeeded.

Signals: lenses 6 and 8 raised (the claim and poll logic in client.py and consumer.py). All 19 lenses were assessed.
Security deep-dive: not triggered, because the diff has no attack surface.
boundary-sweep: 3 sinks traced (the new SQL parameters, the poll-failure log line, the queue-name list), clean.

Verdict: REQUEST CHANGES

Summary: Critical 0 · High 1 · Medium 6 · Low 5 · Lenses run 19/19 (13 examined, 6 N/A)

Findings

1. [High] [Lens 6] Rotating start by one per cycle splits throughput by list position, not equally among busy queues

• Location: workers/queue_backend/pg_queue/consumer.py:402 (the _next_queue advance) and workers/queue_backend/pg_queue/client.py:161 (the ORDER BY). The prose that hides it is at client.py:458-461, consumer.py:313-318 and consumer.py:378-382.
• Failure mode: _next_queue advances by one whichever queue was served. A start that lands on an empty queue falls through to the next non-empty queue in list order, so a busy queue gets one extra turn per empty queue just before it. The old loop gave every busy queue one claim per cycle. The compose consumers drain 3, 5 and 7 queues (docker/docker-compose.yaml:282,323,494,587).
  • On the 7-queue executor consumer with work only at index 0 and index 6, the serve order was q0 q6 q6 q6 q6 q6 q6 q0 …, a 1:6 split. With index 0 and 1 busy, it is 6:1.
  • On file_processing,file_processing_priority,api_file_processing with one of the other two empty, the split is 2:1 instead of 1:1. Over 300 cycles the code-reviewer got {'api_file_processing': 200, 'file_processing': 100}, against 150/150 before.
  • Executor tasks are long LLM calls, so the queue in the unlucky position waits minutes longer. Nothing starves, but the docstrings' "as with the per-queue loop this replaces" is false.
• Evidence: _next_queue = (self._next_queue + 1) % len(self.queue_names). The claimed CTE already returns ord.
• Suggested fix:
  • Advance _next_queue to the index after the last queue actually served. Return ord or the queue name from read_any, and add 1 only on an empty cycle.
  • Add a test with an empty queue between two busy ones. test_start_picks_the_first_queue_probed and test_start_queue_rotates_every_cycle can't see this.
• Confidence: High.
• Verification: VERIFIED. Two agents independently ran the real SQL on a scratch Postgres 16. The code-reviewer also ran poll_once with a stubbed _handle.

2. [Medium] [Lens 3] The cycle-wide try in poll_once returns 0 after partial work and strands the rest of a mixed-queue batch

• Location: workers/queue_backend/pg_queue/consumer.py:403-410. run() consumes it at consumer.py:1080-1092.
• Failure mode: read_any claims up to batch_size rows across queues (each gets vt and read_ct+1), and the loop handles them inside one try.
  • If _handle raises on message k (for example an ack delete fails), messages k+1..n stay claimed and unstarted until the lease expires.
  • read_ct climbs for work that never ran, toward the poison-drop threshold.
  • return 0 makes run() take the idle backoff even though messages 1..k-1 were processed.
  • The removed docstring promised "the work already done this cycle still counts". The removed test_one_queue_failing_does_not_abort_the_others pinned per-queue isolation.
  • Stranding within one queue is pre-existing. This diff adds stranding across queues and the lost partial count.
  • It also collapses claim errors, handle errors and programming errors into one log line. The except in run() becomes unreachable for ordinary exceptions.
• Suggested fix: separate the claim from the per-message handling, log the claimed msg_ids on failure, and return the handled count from the failure path. Return 0 only when the claim itself failed.
• Confidence: High on the control flow, Medium on how often it happens.
• Verification: UNVERIFIED. A test where read returns 3 messages and _handle raises on the 2nd would confirm the return value and that the 3rd is never started.

3. [Medium] [Lens 9] The per-queue probe locks up to n×qty rows to claim qty, and a concurrent reader that hits them sees an empty queue and backs off

• Location: workers/queue_backend/pg_queue/client.py:150-160. The docstring is at client.py:463-466.
• Failure mode: every multi-queue claim on busy queues FOR UPDATE-locks the head of all n queues (7 for the executor consumer). That means heap and WAL writes on a change meant to cut load. A second reader skips those rows. It takes a lower-priority row or, if the locked row was that queue's only ready row, gets [] and enters exponential backoff up to backoff_max. The docstring says "a few milliseconds", but the cost is a backoff sleep. The old per-queue read would have claimed that row immediately.
• Suggested fix: probe in rotation order and stop locking once qty is reached (separate arms with their own LIMIT, or an in-statement loop). At minimum, state the empty-result and backoff consequence in the docstring.
• Confidence: Medium.
• Verification: UNVERIFIED. Does a reader racing a multi-queue claim on a queue with one ready row get []? test_held_claim_is_skipped_not_waited_on shows the skip but not the empty case.

4. [Medium] [Lens 8] Autocommit removes the failed-rollback dead-connection signal that _cursor relies on

• Location: workers/queue_backend/pg_queue/client.py:264-288 (_cursor).
• Made stale by: client.py:250 (conn.autocommit = True).
• Failure mode: the comment says a failed rollback() proves the connection is unusable, even when a server termination surfaces as a bare DatabaseError. On an autocommit connection rollback() doesn't touch the socket, so it can't fail. Liveness then rests on is_connection_dead(exc) and conn.closed. If a termination surfaces as a bare DatabaseError without closed set, the dead handle stays cached. Every later poll then fails and logs without reconnecting.
• Suggested fix: on the owned-connection error path, also treat an unclassified error as suspect. Check conn.closed or conn.status, or run a cheap SELECT 1 probe.
• Confidence: Medium.
• Verification: UNVERIFIED. An integration test that runs pg_terminate_backend on the owned connection mid-statement and checks the next call reconnects would confirm it.

5. [Medium] [Lens 13] read_any's lease (vt) is not pinned by any added test

• Location: workers/tests/test_pg_queue_client.py:904-913, against client.py:158-163.
• Failure mode: the tests check only state == 'claimed' and read_ct. Replacing make_interval(secs => %s) with … * 0 (so vt = now()) left all 11 TestReadAnyIntegration tests green. The existing read() lease tests cover only the single-queue SQL. A lease bug in the new SQL would let the reaper re-arm messages immediately and double-process them.
• Suggested fix: assert vt > now() on the claimed row. Or assert that a second read_any returns [] and rearm_expired_claims leaves the row alone.
• Confidence: High.
• Verification: VERIFIED. The mutant was run through a pytest plugin that patches _dequeue_any_sql.

6. [Medium] [Lens 13] Batch order within a qty>1 claim is not pinned

• Location: workers/tests/test_pg_queue_client.py:935-940, against client.py:161,174.
• Failure mode: test_priority_orders_within_a_queue uses the default qty=1, so it only exercises the inner per-queue ORDER BY. Dropping priority DESC from either outer ORDER BY, which sets the order the consumer processes a batch in, left all 11 tests green.
• Suggested fix: send low and high priority messages into one queue, call read_any(..., qty=2), and assert ["high", "low"].
• Confidence: High.
• Verification: VERIFIED, by mutation.

7. [Medium] [Lens 16] The module and _cursor docstrings still promise transaction semantics that owned connections no longer have

• Location: workers/queue_backend/pg_queue/client.py:21-24 and client.py:256-262.
• Made stale by: client.py:250.
• Failure mode: "every operation rolls back on error" and "commit on success, roll back … a failed statement leaves the connection in an aborted transaction" are false for a client-created connection, where each statement is already durable. A maintainer who adds a second execute inside one _cursor() block would expect all-or-nothing and get partial commits. The one-statement invariant is stated only at the conn property.
• Suggested fix: drop the rollback clause from the module docstring and keep the reconnect half. Add to _cursor: "on an owned autocommit connection each statement commits on its own; keep each operation to one statement".
• Confidence: High.
• Verification: VERIFIED. Every _cursor caller issues a single execute today.

8. [Low] [Lens 16] "The queue probed first rotates" is inaccurate, because every queue is probed on every call

• Location: workers/queue_backend/pg_queue/client.py:140-146, consumer.py:378-382, and the test name test_start_picks_the_first_queue_probed (tests/test_pg_queue_client.py:915).
• Failure mode: the LATERAL probes and locks all queues each call. Rotation only decides which queue is served first. The read_any docstring (client.py:463) says this correctly, and these contradict it.
• Suggested fix: say "the queue served first".
• Confidence: High.
• Verification: VERIFIED, by reading the SQL.

9. [Low] [Lens 17] "Stamped before each queue read" is stale in three places, and the new comment misstates the constant

• Location: workers/queue_backend/pg_queue/consumer.py:99, workers/pg_queue_consumer/supervisor.py:373, workers/tests/test_pg_queue_consumer.py:669, and consumer.py:386.
• Made stale by: workers/queue_backend/pg_queue/consumer.py:387.
• Failure mode: the stamp is now once per claim. consumer.py:386 says HEALTH_STALE is "sized for one task", while consumer.py:105-108 says to size it for batch_size × worst_case_task_seconds.
• Suggested fix: say "each claim", and "sized for one batch (see the constant)".
• Confidence: High.
• Verification: VERIFIED, by grep and by reading poll_once.

10. [Low] [Lens 16] Comments that cite removed behaviour, or a caller that was not found

• Location:
  • client.py:246-248 and tests/test_pg_queue_client.py:228-229 (the "agentic fan-out outbox" that injects its own connection).
  • consumer.py:318-320 (the dedup reason "double-read a queue per cycle").
  • client.py:451-461 and consumer.py:381-384 ("as with the per-queue loop this replaces", "as a failed queue did before").
• Failure mode: nothing in the repo constructs PgQueueClient(conn=…) outside tests, so the outbox premise can't be checked. The dedup reason no longer applies, because a duplicate would now skew the rotation. The "as before" comparisons have no "before" once merged.
• Suggested fix: state the current guarantee only, and drop or name the outbox caller.
• Confidence: Medium.
• Verification: VERIFIED, with grep -rniE outbox and grep 'PgQueueClient(conn'.

11. [Low] [Lens 16] "One statement per cycle" holds only on a client-owned autocommit connection

• Location: workers/queue_backend/pg_queue/client.py:241-244 and client.py:452-454.
• Failure mode: with an injected non-autocommit connection, read_any is still BEGIN, claim, COMMIT. "Most of the idle-poll load" is unmeasured.
• Suggested fix: add "on a client-owned connection", or drop the load claim.
• Confidence: Medium.
• Verification: VERIFIED, traced from the conn property.

12. [Low] [Lens 13] test_rejects_bad_args needs a database, and the fixture docstring says "by prefix" but deletes by name

• Location: workers/tests/test_pg_queue_client.py:1053 and :888-889.
• Failure mode: pure argument validation takes pg_conn, so it is skipped on lanes without Postgres. REQUIRE_PG_TESTS unset skips silently.
• Suggested fix: drop the fixture from that test, and say "cleaned up by name".
• Confidence: High.
• Verification: VERIFIED, by reading the test.

Unanchored findings

• No PR was resolved, so the title and description were not assessed.
• Lens 15 gate: the repo pins ruff v0.3.4 but only 0.16.8 is installed, so the pinned gate was not run. At 0.16.8 its complaints fall on lines this diff did not add, so none is attributed to it.
  • I001 and UP047 in consumer.py.
  • Format hunks in the two test files.

Lens checklist

1. Correctness vs. intent: See findings fix: Prompt studio index and run issues #1, Set server_tokens to off in the nginx configuration #8
2. Architecture fit: Clean (assessed by code-reviewer)
3. Error handling: See findings Profile manager migration fix #2, Basic workflow and adapter exception messages #4
4. Security: Clean (assessed by me; SQL parameters are bound, the f-string takes only constants and the schema name, the new log line carries queue names only)
5. Data integrity & migrations: Clean (assessed by me; the claim stays one atomic statement, no schema change)
6. Concurrency: See findings fix: Prompt studio index and run issues #1, connectors #3 (assessed by me and the cold reader)
7. API compatibility: N/A — no wire format, status code or schema change; read_any is internal (the only PgQueueClient callers use send, set_vt or read)
8. Reliability & resilience: See findings Profile manager migration fix #2, connectors #3, Basic workflow and adapter exception messages #4
9. Performance & cost: See findings connectors #3
10. Observability: Clean
11. Operational safety: N/A — no manifests, flags or rollout config touched
12. LLM/agent: N/A — no prompt, model or agent-loop change
13. Testing: See findings Add support for configuring and using OCR adapters #5, Corrected alignment of the OnBoard page to accommodate various screen sizes #6, Connector metadata encryption/decryption on connector-instance model #12 (167 tests passed, 0 skipped, against a live migrated Postgres)
14. Dependencies & build: N/A — no manifests changed
15. Code quality: Clean (see the ruff caveat under Unanchored findings)
16. Documentation accuracy: See findings fix: Prompt studio tool delete fix #7, Set server_tokens to off in the nginx configuration #8, Connector metadata encryption/decryption on connector-instance model - 2 #10, removed translator, PII redactor and workflow settings #11
17. Cross-document consistency: See findings Prompt Studio - Summarize, Output Analyzer and other UI Improvements #9 (assessed by me; the facts.sh merge-sweep hits on "drain" are unrelated noise and were dismissed)
18. On-prem deployment experience: N/A — no chart, schema or on-prem surface touched
19. Scaling & workload lifecycle: N/A — no scale or lifecycle values changed

Open questions for the author

• Is a position-dependent throughput split among busy queues acceptable? The old comment said throughput was never equalized, but per-cycle fairness did hold before.
• Does an "agentic fan-out outbox" caller that injects a connection exist on another branch?

Assumptions made

• _handle can raise, because an ack delete is unguarded in some branches. If every path inside it is guarded, Profile manager migration fix #2 shrinks to Low.
• The tree is the author's own, not loop-authored, so the self-authored-lens dispatch was not applied.

cost: agents=5 wall=06:16 lenses-examined=13 of 19 files=4 lines=443

---

Generated by Claude Code

[muhammad-ali-e]

Self-review — round 2

• Tip: 351b04c93e9ff12da9a75496cf5a54ce92468784; delta: 36b4596..351b04c (the round-1 fixes)
• Review run: /unstract:review-deep (review-shallow refuses this PR on its concurrency_primitive gate). Local diff against main; GraphQL is blocked here.
• Coverage caveat (as reported by the review): its scratch directory was removed mid-run, so the code-reviewer and cold-read agents ran without the rubric file; see Not covered below.

Mode: INITIAL, on the local diff. The base is origin/main, merge-base 6da4590, HEAD 351b04c. The base fetch succeeded. There was no changedFiles count to check against, so the file count (5) is unverified. The diff is 5 files and 618 lines. The round delta (36b4596..351b04c) is 5 files and 221 insertions.

Signals: lens 6 raised (claim, lease and queue-consumer logic), lens 8 (retries and connection recovery), lens 13 (tests changed), lens 16 (comments changed) and lens 17 (the queue_name / read semantics are restated in several comments). Lens 5 stayed at baseline: no schema or migration files changed, and the claim SQL change is not a schema change. Security deep-dive: unavailable. Facts showed no attack-surface trigger, so I did not run it. Boundary sweep: 0 outbound sinks added or changed (log lines carry msg_ids and queue names only), clean.

Verdict: REQUEST-CHANGES

Summary: 0 Critical, 0 High, 3 Medium, 5 Low (8 findings).

Findings

1. [Medium] [Lens 3, 6, 8] poll_once abandons the rest of a batch on any handler escape, burning their read_ct

• Location: workers/queue_backend/pg_queue/consumer.py:398-408.
• Failure: with batch_size > 1 (only possible when lease equals vt), a failure that escapes _handle returns at the first failure. The later messages stay claimed until vt lapses, then the reaper re-arms them.
• Each claim does read_ct = read_ct + 1, and the re-arm doesn't reset it. A tail that never started therefore pays one attempt per round. If the head fails deterministically it is reclaimed first, so the same tail is stranded every round. After max_attempts (default 5) the tail is poison-dropped, and the error is sent to its reply_key/on_error for work that never ran.
• One deterministic trigger: a non-dict JSONB payload (a foreign writer's list or string) raises at payload.get in _handle before the drop guard.
• Fix: continue past the failed message and count successes, or stop the batch only on connection-dead errors.
• Flagged by silent-failure-hunter. I confirmed the control flow (the return done at line 408) but did not run the read_ct climb.

2. [Medium] [Lens 3, 8, 11] Owned-connection autocommit makes a statement that failed in flight ambiguous, which send() and the claims weren't written for

• Location: client.py:257 (autocommit), client.py:359-384 (send() retry), client.py:446-450 and 498-505 (claims).
• Failure: before, a connection death mid-statement rolled the transaction back, so "nothing landed" held. Now each statement commits at statement end.
  • A reply lost after the INSERT ran makes the send() retry write a duplicate. The docstring at client.py:338-347 still describes that as a rare post-commit case.
  • A claim that committed but whose rows were never read leaves the message claimed for lease_seconds (120s default) with read_ct raised. That counts toward the poison cap at consumer.py:556.
• The new reconnect test kills the backend between calls (the safe idle-reap case) and does not cover death mid-statement.
• Fix: either document this openly in the send() docstring and the conn comment, or log a WARNING on a connection-dead error from a claim so stalled messages can be diagnosed. The poison-attempt cost of a lost claim is worth stating.
• Flagged independently by the silent-failure hunter and the cold read. My own read agrees.

3. [Medium] [Lens 16] The _cursor NOTE contradicts the lines added above it

• Location: client.py:285-290, contradicted by client.py:281-284.
• Failure: the NOTE says a bare DatabaseError "is still treated as dead here (via the failed-rollback branch)". The added lines say rollback() "sends nothing and cannot fail" on an owned autocommit connection. The discard only runs when the client owns the connection (_owns_conn), and owned means autocommit. So the sentence is true only via conn.closed.
• The new test comment also repeats "cannot fail". On an already-closed connection it raises InterfaceError, but the code discards correctly anyway.
• Fix: rewrite the NOTE to say a bare DatabaseError is dropped only if conn.closed is set.

4. [Medium] [Lens 13] Untested _claim rotation branches

• Location: consumer.py:425-428 (the queue_name fallback and the messages[-1] rule), with the gap in tests/test_pg_queue_consumer.py:916-953.
• Failure: replacing messages[-1] with messages[0], or the fallback with served = 0, leaves every consumer test green.
• Fix: add a batch_size=3 mock case returning qa then qc and assert the next start is 0. Add an unknown-name case with _next_queue=1 and assert the next start is 2.
• Flagged by pr-test-analyzer. The analyzer did not run mutants.

5. [Low] [Lens 16, 17] HEALTH_STALE comment still says "single task"

• Location: consumer.py:98-103. The heartbeat is now stamped once per claim, so it is frozen for the whole batch (consumer.py:389 says "one batch").
• Fix: the bound is a batch's wall-clock. The supervisor's per-task wording (supervisor.py:139-143) predates this diff but repeats the framing.
• Rated Low rather than the analyzer's Medium: the constant's last line already gives the per-batch formula, and batches above 1 are rare.

6. [Low] [Lens 3, 10] QueueMessage.queue_name defaults to ""

• Location: client.py:199 and the fallback at consumer.py:428.
• Failure: this is the same silent-default shape the read_ct docstring rejects. A constructor that omits the field falls back to rotate-by-one with no log, quietly bringing back the 2:1 skew around empty queues. The tests use the default via _msg.
• Fix: make the field required, or log at WARNING in the fallback when the batch is non-empty.

7. [Low] [Lens 8] run() backoff resets after a partially failed batch

• Location: consumer.py:408 returns done, and run() resets the backoff on a non-zero count. During a DB outage that costs one extra fast cycle per round.

8. [Low] [Lens 16] Two inaccurate test comments

• tests/test_pg_queue_consumer.py:875: "used to hand the queue after it a second turn (2:1)" refers to an intermediate version of this PR, not the base. Drop it.
• tests/test_pg_queue_client.py:986: "nor return one" has no clear referent. Write "nor return a row reader A has locked".

Not covered / caveats

• Agent coverage was partly degraded. The scratch directory with the rubric files was deleted mid-run. The silent-failure, test, and comment agents were re-dispatched with the contracts at a new path and ran under the full contract. The first-round code-reviewer and the cold-read agent had already started and ran without the contract file. They worked from their default formats and reviewed the diff directly. Their coverage of lenses 1, 2, 9, 10 and 15 is real but not rubric-scoped, and the cold read had no cold-read payload.
• The lens17-merge-sweep returned only generic claim identifier hits in files this PR doesn't touch. I read it as noise, with no lens 17 finding.
• Nothing was run: no pytest, and no live-Postgres checks. Findings 1 and 2 rest on code reading.

Lens checklist (19)

1. Intent and PR hygiene: Clean (code-reviewer, no contract; PR title and description not fetched, local diff)
2. Correctness: Clean (code-reviewer, no contract; rotation, SQL param order and queue_name propagation traced). See fix: Prompt studio index and run issues #1 for the batch-abort issue, filed under lens 3.
3. Error handling and silent failure: See findings fix: Prompt studio index and run issues #1, Profile manager migration fix #2, Corrected alignment of the OnBoard page to accommodate various screen sizes #6
4. Security: Clean (no attack surface; boundary sweep 0 sinks; I assessed this myself and did not run the deep-dive)
5. Data integrity and migrations: Clean (no schema change; claim UPDATE semantics read directly, I assessed this myself)
6. Concurrency: See findings fix: Prompt studio index and run issues #1, Profile manager migration fix #2. The SKIP LOCKED locks (accepted trade-off) and rotation fairness are Clean (I assessed this myself).
7. API and contract compatibility: Clean (I checked the consumers: QueueMessage gained an optional field, and read/read_any signatures are unchanged. Type-design agent not dispatched, so not independently covered.)
8. Reliability and resilience: See findings Profile manager migration fix #2, fix: Prompt studio tool delete fix #7
9. Performance and cost: Clean (code-reviewer, no contract; one statement per claim, per-queue LIMIT bounds the lock count)
10. Maintainability: See finding Corrected alignment of the OnBoard page to accommodate various screen sizes #6
11. Operational safety: See finding Profile manager migration fix #2
12. LLM and agent: N/A (no prompt, model or tool config)
13. Testing: See finding Basic workflow and adapter exception messages #4. The four round-delta tests would fail without their fixes.
14. Dependencies and build: N/A (no dependency changes)
15. Code style and conventions: Clean (code-reviewer, no contract). The pinned lint gate was not run, and I did not run ruff; this is not checked.
16. Documentation accuracy: See findings connectors #3, Add support for configuring and using OCR adapters #5, Set server_tokens to off in the nginx configuration #8
17. Cross-document consistency: See finding Add support for configuring and using OCR adapters #5 (the constant against the batch wording). The merge sweep was noise; I assessed this myself.
18. On-prem deployment experience: N/A (no chart, values, schema, or on-prem files)
19. Scaling and workload lifecycle: Clean (no HPA, replicas or termination values changed; I assessed this myself). See fix: Prompt studio index and run issues #1 for the batch lease interaction, filed under lens 3.

cost: agents=8 (5 first-round, 3 re-dispatched) wall=? files=5 lines=618 lenses-examined=15 of 19

---

Generated by Claude Code

[muhammad-ali-e]

Self-review — round 3

• Tip: 7202b32175b7a56e84f857c30bed81777c83a390; delta: 351b04c..7202b32 (the round-2 fixes)
• Review run: /unstract:review-deep (review-shallow refuses this PR on its concurrency_primitive gate). Local diff against main; GraphQL is blocked here.

Verdict: APPROVE WITH NITS.

Summary: Critical 0 · High 0 · Medium 2 · Low 3 · Lenses run 19/19 (11, 12, 14, 18 and 19 are N/A, with reasons below).

I didn't run the security deep-dive. The diff has no endpoint, parser, authn/authz, secret or new external-input boundary. The boundary sweep traced 6 log sinks and found them clean.

Findings

1. [Medium] [Lens 3] The dead-connection batch abort rests on a premise the client contradicts. workers/queue_backend/pg_queue/consumer.py:405-417, docstring at :385-392.

   • Premise: The docstring says the remaining messages "could not be acked either". But _cursor has already dropped the dead handle (client.py:298-301), and delete() retries once. The next ack would reconnect.
   • Failure mode, batch_size>1 only: The stranded tail waits out lease_seconds plus the reaper interval. Each re-arm burns one read_ct toward the poison cap.
   • Classifier: is_connection_dead (connection.py:67-82) also matches QueryCanceled and deadlock errors, where the connection is fine.
   • Fix: Narrow the abort to conn.closed, drop the special case, or reword the docstring.
   • Confidence: Medium. The agents disagreed: silent-failure and the cold read flagged it, the correctness reviewer did not.
   • Verification: UNVERIFIED, from reading the code only. Does the set_vt or first-delete path hit a reaped connection, or does QueryCanceled reach poll_once, often enough to matter?

2. [Medium] [Lens 3] After the autocommit switch, a bare DatabaseError on a dead owned connection no longer discards the handle. client.py:257 with :281-301.

   • Failure mode: rollback() no longer probes the connection, so a handle with conn.closed == 0 is kept. For read and read_any, which have no retry, that costs one failed cycle and a back-off.
   • Fix: On an owned connection, treat a non-payload DatabaseError as dead and drop the handle.
   • Confidence: Medium.
   • Verification: UNVERIFIED. Kill a backend so it surfaces as a bare DatabaseError with closed == 0, and see whether the next call recovers.

3. [Low] [Lens 16] "rollback() … cannot fail" is false on a closed connection. client.py:267-270 and :281-285, repeated at workers/tests/test_pg_queue_client.py:1089.

   • Why: psycopg2 raises InterfaceError there. The guard under the comment handles it, but the comment invites its removal.
   • Fix: Delete the clause.
   • Verification: UNVERIFIED. This relies on psycopg2 behaviour, and psycopg2 wasn't importable in the reviewer's sandbox.

4. [Low] [Lens 16] The read_any trade-off text understates who loses rows to its probe locks. client.py:484-491.

   • Why: A dedicated single-queue consumer on one of the listed queues can also get [] and back off, not only contending multi-queue consumers.
   • Verification: VERIFIED from _dequeue_any_sql.

5. [Low] [Lens 13] The new owned-connection recovery has no test, and an old test models an impossible state. workers/tests/test_pg_queue_client.py:208-216 against client.py:272-297.

   • Why: test_owned_conn_is_recovered_when_rollback_fails only passes because a MagicMock raises. Changing (conn_dead or conn.closed) to conn_dead at client.py:292 fails no test.
   • Fix: Add closed=0 and closed=1 cases with a bare DatabaseError.
   • Verification: UNVERIFIED. The mutation was not run.

Unanchored findings

• Test environment (untallied): workers/tests/conftest.py puts every non-xdist run in one shared schema. Concurrent runs by the review agents caused 2–7 spurious failures. With an isolated schema, 175 of 175 tests pass against a real Postgres 16 with REQUIRE_PG_TESTS=1. Removing either rotation ORDER BY in _dequeue_any_sql was caught by a test. I did not run the suites myself.
• PR title: not checked, because this is a local diff.

Lens checklist

1. Spec & intent: Clean (no PR description to judge against; title item skipped)
2. Architectural fit & precedent: Clean
3. Correctness & edge cases: See findings fix: Prompt studio index and run issues #1, Profile manager migration fix #2
4. Security: Clean (SQL fully parameterised; no new boundary; assessed by me)
5. Data integrity & migrations: Clean (no schema change; the 10 %s placeholders match the 10 bound params; assessed by me)
6. Concurrency: See findings fix: Prompt studio index and run issues #1, Basic workflow and adapter exception messages #4 (no double-claim path found; assessed by me)
7. API & contract compatibility: Clean (queue_name is appended with a default; the only other repo caller, unstract-cloud fanout_claim.py, calls send() on an injected connection and is unaffected)
8. Reliability & resilience: See findings fix: Prompt studio index and run issues #1, Profile manager migration fix #2
9. Performance & cost: Clean
10. Observability: Clean
11. Operational safety: N/A (no deployment, flag or rollout config touched)
12. LLM/agent-specific: N/A (no model or agent path)
13. Testing: See findings Add support for configuring and using OCR adapters #5
14. Dependencies & build: N/A (no manifest or lockfile touched)
15. Code quality: Clean (I did not run the pinned formatter gate, so no formatting finding is claimed)
16. Doc & comment accuracy: See findings connectors #3, Basic workflow and adapter exception messages #4
17. Cross-document rule consistency: Clean (grep for the old "before each queue read" / "read once each" / "one task from EACH" prose finds no remaining restatement; the comment analyzer found the README, OPERATIONS.md and docker-compose still correct at batch_size=1; the merge-sweep script's hits were unrelated noise on the generic word "claim")
18. On-prem deployment experience: N/A (no chart, values or customer docs touched)
19. Scaling & workload lifecycle: N/A (no concurrency, replica, grace or pool-sizing config touched)

Open questions for the author

• Finding 1: is the abort meant to protect only against a real outage? If so, should it key on conn.closed instead of is_connection_dead(exc)?
• Finding 2: can you reproduce a server kill that surfaces as a bare DatabaseError with closed == 0 on the autocommit connection?

Assumptions made

• The default batch_size=1 is the production path, so I weighted findings 1 and 2 down.
• No ruff or black gate was run. A formatting failure on added lines would be a further Low under lens 15.

cost: agents=6 wall=? lenses-examined=14 of 19 files=5 lines=731

Addressed in 5c25907: #1 (batch no longer stops on connection-classified errors), #2 (bare DatabaseError treated as dead on owned connections), #3 and #4 (comments), #5 (tests; each new guard mutation-checked).

---

Generated by Claude Code

[muhammad-ali-e]

Self-review — round 4

• Tip: 5c25907e6bd3ae0fd5bb075c95f8d68be3e9a107; delta: 7202b32..5c25907 (the round-3 fixes)
• Review run: /unstract:review-deep (review-shallow refuses this PR on its concurrency_primitive gate). Local diff against main; GraphQL is blocked here.

Verdict: REQUEST CHANGES

Summary: Critical: 0 · High: 0 · Medium: 1 · Low: 1 · Lenses run: 19/19 (lens 4 read directly, no security deep-dive needed)

• Target: local working diff of claude/loving-curie-qdqwqj, HEAD 5c25907e6bd3ae0fd5bb075c95f8d68be3e9a107, against origin/main. Merge-base is 6da459096d67be1ae89e47621f83a15f332bfefa, and base_fetch was ok.
• Size: 5 files, 749 lines. There is no PR to reconcile the file count against, so the count is unverified.
• Mode: no PR was resolved (gh pr view is blocked here), so this is a local diff and always INITIAL.
• Round delta: git diff 7202b32 5c25907 touches client.py, consumer.py and their two test files.
• Signals: lenses 6, 8 and 3 raised (retry and queue-consumer logic, FOR UPDATE SKIP LOCKED claims, autocommit). concurrency_primitive and tests_changed fired.
• Not run: the Step 2a security deep-dive. No attack-surface trigger fired, so Security deep-dive: not triggered.
• Dispatched: code-reviewer (lenses 1, 2, 3, 9, 10, 15), silent-failure-hunter, pr-test-analyzer, comment-analyzer, and a cold-read whole-diff agent.
• Not dispatched: type-design-analyzer. The only type change is QueueMessage.queue_name with a default, so I took lens 7 myself.

Findings

1. [Medium] [Lens 3, 6] — A batch now keeps running tasks while the DB is unreachable, so finished work is redelivered and runs twice

• Location: workers/queue_backend/pg_queue/consumer.py:402-411 (the poll_once handling loop). The docstring claim is at consumer.py:385-390. The ack paths are consumer.py:651 and :709, and PgQueueClient.delete is at client.py:550-589.
• Failure mode:
  1. Round 4 removed the is_connection_dead stop. The loop now does except Exception: log; continue for every message in the batch.
  2. _handle resolves ordinary task failures itself. What escapes is a failure to reach the DB: the ack delete, or _fail_dispatch plus delete. delete() retries once after a reconnect.
  3. During a real DB or PgBouncer outage that retry also fails, so the exception escapes _handle.
  4. The loop then runs the next message's task to completion and its ack fails the same way.
  5. Every tail message is redelivered at lease expiry and runs again.
  6. delete()'s own docstring (client.py:559-566) names the costly case: a re-fired aggregating callback re-sends webhooks and subscription-usage billing.
  7. The new docstring ("the client reconnects for their acks if the failure was a dropped connection") is true for a one-off blip and not for an outage.
• Scope: only with batch_size > 1. The constructor (consumer.py:351-358) forces it back to 1 unless lease_seconds == vt_seconds. At the default of 1 there is no tail. The docker-compose services the cold-read agent checked do not set WORKER_PG_QUEUE_CONSUMER_BATCH_SIZE.
• Origin: this is a reversal of the round-3 fix. Round 3 added the batch stop and round 4 removed it. The trade-off is real in both directions:
  • With the stop: the tail burns a read_ct attempt toward the poison cap.
  • Without the stop: the tail's work runs twice in an outage.
• Evidence: silent-failure-hunter and the cold-read agent raised this independently. I confirmed the loop shape against the code. Both agents verified the ack-retry path only by reading it.
• Suggested fix: keep per-message isolation for task failures. Stop the batch only when the escaping exception satisfies is_connection_dead(exc) and the client could not reconnect. A simple form is to stop after a second consecutive ack or connection failure and log the unrun msg_ids. If you keep the current behaviour deliberately, state the outage double-run in the poll_once docstring.
• Confidence: Medium. It rises to High if a production consumer runs with batch_size > 1 and lease == vt.
• Verification: UNVERIFIED. Question for the author: with batch_size > 1, is it intended that a DB outage makes every message after the first failed ack run and then redeliver? A test would confirm it: a real PgQueueClient whose delete raises OperationalError for every message of a batch of 3, counting task executions before and after the lease lapses. test_connection_error_does_not_stop_the_batch (workers/tests/test_pg_queue_consumer.py:958) cannot show this, because its MagicMock client never reconnects.

2. [Low] [Lens 16] — "no SQLSTATE subclass, so not a statement error" is stated more strongly than psycopg2 guarantees

• Location: workers/queue_backend/pg_queue/client.py:283-285, and the paired comments in workers/tests/test_pg_queue_client.py:219-222 and :230-232.
• Failure mode: the comment presents a bare DatabaseError as proof the connection died. psycopg2 falls back to the base class for any SQLSTATE class it does not map. A statement error from such a class would drop and reconnect the owned connection. The cost is one needless reconnect. The comment misleads only a maintainer who relies on "cannot happen".
• Evidence: the type(exc) is psycopg2.DatabaseError check at client.py:292. The code-reviewer agent ran psycopg2 2.9.9 and found the specific error classes are all subclasses, so the unmapped case is narrow.
• Suggested fix: drop the claim and say the exact-type check is a heuristic: on an autocommit connection rollback() proves nothing, so a bare DatabaseError is treated as dead. Deleting the sentence is also fine, since the code states it.
• Confidence: Medium.
• Verification: UNVERIFIED. Does any SQLSTATE class that these statements can raise fall outside psycopg2's mapping? psycopg2.errors.lookup() on a few codes would answer it.

Unanchored findings

• Lens 15 formatter gate not run at the pinned version. .pre-commit-config.yaml pins ruff-pre-commit v0.3.4, and only ruff 0.16.8 is available here.
  • ruff format --diff at 0.16.8 flags workers/tests/test_pg_queue_client.py:588 and three hunks in workers/tests/test_pg_queue_consumer.py. Only the client-test hunk (blamed to a post-merge-base commit) may sit on a line this PR added.
  • I could not tell whether the version difference explains it, so treat it as a question: does pre-commit run ruff-format at v0.3.4 pass on these files?
  • ruff check at 0.16.8 also reports I001 import-sorting on client.py:33 and consumer.py:24. Both are rules from a different version and unverified.
  • No line is attributed here, so there is no severity.

Checks that came back clean

• Round-3 tests: the new _cursor tests are not vacuous. test_owned_conn_dropped_on_bare_database_error fails without the new clause, and test_owned_conn_kept_on_statement_error fails if the check is loosened to isinstance. The rewritten consumer test is a legitimate expectation change. No stale references remain.
• Test run: the cold-read agent ran both test files, 178 passed, with the real-Postgres integration tests executing rather than skipping. This run does not cover finding 1.
• Boundary sweep: 4 log sinks traced (claim failed, handling failed, unknown queue_name, and the loop's log). They carry only msg_id, queue names and exception text, with no PII or credentials. Clean.
• Lens 17 merge sweep: hits were noise from the token "claim" in unrelated pg_barrier and orchestration code, and none restates a rule this diff changed. The cold-read agent also found no other text describing the old one-read-per-queue behaviour.

Addressed in 233e66b: #1 by keeping main's batch-stop semantics (rounds 2–4 showed both alternatives cost something; the trade-off is written in poll_once), #2 by rewording the comment as a heuristic. Both batch-failure guards mutation-checked.

---

Generated by Claude Code

[muhammad-ali-e]

Self-review — round 5

• Tip: 233e66b27599a921a32bf569940dd050f3164f15; delta: 5c25907..233e66b (the round-4 fixes)
• Review run: /unstract:review-deep (review-shallow refuses this PR on its concurrency_primitive gate). Local diff against main; GraphQL is blocked here.

Verdict: APPROVE WITH NITS

Summary: Critical 0 · High 0 · Medium 0 · Low 4 · Lenses run 19/19 (5, 11, 12, 14, 18 and 19 are N/A, see the checklist).

Scope: 5 files, 730 lines under workers/. The file count is unverified because gh is blocked and there's no PR. base_fetch was ok.

Signals raised: lenses 6, 8, 13 and 16 (queue-consumer batch logic, tests, comments).

Security: the deep-dive wasn't triggered because no attack surface fired. The boundary sweep traced 2 log sinks and found them clean: they carry only msg_ids.

Fan-out: five agents ran.

• The code-reviewer covered lenses 1, 2, 3, 9, 10 and 15.
• The silent-failure-hunter covered lens 3.
• The test-analyzer covered lens 13.
• The comment-analyzer covered lens 16.
• A cold whole-diff reader had no PR description.

Test evidence: the cold reader ran both changed test files: 177 passed, including the 35 integration tests against real Postgres. The code-reviewer also checked the new read_any claim query against a throwaway Postgres table.

Round-4 delta: the delta is correct. In poll_once, enumerate plus return handled counts exactly the messages finished before the failure, and the log slice messages[handled+1:] lists exactly the unstarted ones. Stopping the batch matches origin/main, and the new code also keeps the partial count that main dropped. The replaced test would fail against the old continue loop, so it isn't vacuous.

Findings

1. [Low] [Lens 16] — _cursor NOTE says an injected connection is dropped, but the code only drops an owned one.

   • Location: workers/queue_backend/pg_queue/client.py:290-293.
   • The NOTE says a bare DatabaseError is dropped "via the failed-rollback branch on an injected connection". The discard gate at client.py:301 is if self._owns_conn and (conn_dead or conn.closed), so an injected connection is never dropped. The _cursor docstring says the opposite of the NOTE.
   • The clause was added in 5c25907 and survived the round-5 rewrite of the paragraph above it. The comment-analyzer, the cold reader and I each confirmed it by reading.
   • Fix: delete "or via the failed-rollback branch on an injected connection".
   • Verified. I rated it Low because the comment has no runtime effect; the two agents rated it Medium.

2. [Low] [Lens 3, also 6 and 8] — A deterministic exception out of _handle now strands the batch tail, and a non-dict payload can raise on every delivery.

   • Location: workers/queue_backend/pg_queue/consumer.py:405-415, exposed by consumer.py:593 (payload.get("task_name"), outside any try).
   • If a row's JSON message is a list or string written by a raw writer, payload.get raises before the malformed-message drop and the poison-cap check at consumer.py:573. Before round 4, poll_once continued past that row. Now the batch stops there every cycle, and each tail message loses a read_ct attempt per lease lapse without running. After max_attempts (default 5) the tail is dropped as poison without having run, and the bad row is never dropped.
   • backend/pg_queue/models.py:30 is a plain JSONField. The only DB constraints are on priority and state (models.py:97-108), so nothing enforces an object.
   • Exposure is narrow: it needs a non-dict row from a raw writer, and batch_size > 1, which is opt-in (default 1) and forced back to 1 unless the lease equals the max runtime.
   • Fix: guard isinstance(payload, dict) at the top of _handle and drop and ack the row as malformed. Or stop the batch only on connection-class errors. The new docstring justifies stopping only for DB-outage cases.
   • Unverified. Run poll_once with a non-dict head message, a good tail and batch_size=3, and check the tail's read_ct climbs toward the poison cap.
   • The general version of this trade-off, where any deterministic head failure strands healthy tail messages, is documented as accepted in the docstring. I did not file it separately.

3. [Low] [Lens 10, also 16] — The handling-failure log reads left claimed until their lease lapses: [] in the default case and omits the queue name.

   • Location: workers/queue_backend/pg_queue/consumer.py:408-414.
   • With batch_size=1 the unstarted list is always empty, so every handling failure logs a dangling [] with no subject for "their". It also doesn't say what happens to the failed message itself. _handle mostly raises when the ack fails after the task already ran, which means a re-run on redelivery. The old log named the queue; message.queue_name is available now.
   • Fix: log message.queue_name and append the unstarted-ids clause only when the list is non-empty.
   • Verified by reading the code.

4. [Low] [Lens 13] — Only a mid-batch failure is pinned; other failure positions, rotation and the unstarted-id log are untested.

   • Location: workers/tests/test_pg_queue_consumer.py:932-953 against consumer.py:405-416.
   • The only case is a failure at index 1 of 3. The 0 returned on a first-message failure drives the run() backoff, and no test covers it. The last-message case is uncovered too.
   • _claim advances _next_queue past the last claimed message, which after a stopped batch was never started. test_rotation_follows_the_last_message_of_a_batch uses a handler that never fails.
   • No test asserts the unstarted-id log, so a wrong slice such as messages[handled:] would pass.
   • Fix: parametrise the failing msg over 1, 2 and 3 (expected returns 0, 1 and 2). Assert _next_queue and the ids in caplog.text.
   • Unverified; no mutations were run.

Unanchored findings

• Lens 15 format gate: the repo pins ruff v0.3.4 and only 0.16.8 is installed, so I couldn't run the pinned gate. The newer ruff's check-only diffs touch only lines that already exist on origin/main, so I report no lens-15 finding.
• Lens 17 merge-result sweep: the sweep facts.sh ran matched the generic word "claim" and returned noise. I didn't re-run it with identifiers for the changed rules. The comment-analyzer instead checked consumer.py, client.py and supervisor.py for stale restatements ("read once per queue", "continuing with the rest of the batch") and found none.
• Verification gap: the code-reviewer's own run had no migrated pg_queue_message table, so it exercised the claim logic against a temp table. The cold reader's run hit real Postgres, and all TestReadAnyIntegration cases passed.

Addressed in 8c78d0f: #1 (NOTE), #2 (non-object payloads dropped as malformed), #3 (log), #4 (parametrised tests). New guards mutation-checked.

---

Generated by Claude Code

[muhammad-ali-e]

Self-review — round 6 (loop converged)

• Tip: 8c78d0fee76bbb77e4f1c99882b773b1645a4473; delta: 233e66b..8c78d0f (the round-5 fixes)
• Review run: /unstract:review-deep (review-shallow refuses this PR on its concurrency_primitive gate). Local diff against main; GraphQL is blocked here.

Verdict: APPROVE, with 3 Low findings and nothing above Low.

I reviewed HEAD 8c78d0f against merge-base 6da4590. The base fetch worked. This is a local diff, so there was no GitHub file count to check it against. I read the round-6 delta (233e66b..8c78d0f) on its own as well as the whole branch. tests/test_pg_queue_consumer.py passes (105 tests). I did not run test_pg_queue_client.py: its read_any integration tests need a Postgres, and none was reachable (finding 3).

Summary: Critical 0 · High 0 · Medium 0 · Low 3 · Lenses run 19/19 (5 N/A with reasons)

Signals: lenses 6 and 8 (concurrency, queue-consumer logic) and 13 (tests changed) were raised. The security deep-dive did not trigger because no attack-surface trigger fired. The boundary sweep traced one sink, the malformed-payload log, which is finding 2.

Findings

1. [Low] [Lens 13, Tests] The new non-object-payload test never checks the drop's error log.

   • Location: workers/tests/test_pg_queue_consumer.py:966.
   • The test takes no caplog and asserts only the delete calls and the count. Deleting or breaking the logger.error at consumer.py:595-605 would still pass. The sibling malformed-drop test (around line 95) does assert its log text.
   • Fix: add caplog and assert "not a JSON object" in caplog.text.
   • The test does fail without the fix itself: the old code raised AttributeError and stopped the batch.

2. [Low] [Lens 4, Security / boundary sweep] The malformed-payload drop logs the whole raw payload (%r) at ERROR.

   • Location: workers/queue_backend/pg_queue/consumer.py:598.
   • A raw writer can store arbitrarily large or sensitive JSON, so the line is unbounded.
   • It matches the existing missing-task_name and poison drops in _resolve_runnable_task, so it follows the file's pattern rather than adding a regression.
   • Optional fix: log the type and size, or truncate. Two review passes flagged this.

3. [Low] [Lens 13 / 6, Verification gap] The read_any integration tests were not executed.

   • Location: workers/tests/test_pg_queue_client.py:933, with the SQL at client.py:143.
   • The per-queue FOR UPDATE SKIP LOCKED claim, the rotation ORDER BY and RETURNING locked.ord were checked by reading only. I found no defect, and I rated this Low because no wrong behaviour was shown.
   • To settle it, run pytest workers/tests/test_pg_queue_client.py -k ReadAny --no-cov against a TEST_DB_* Postgres.

Notes on the round-6 delta

• Failure log (consumer.py:406-417): the reworded log is accurate. unstarted is computed before the return. A first-message failure returns 0, so run() backs off. The parametrized test pins the counts, the log text and _next_queue == 0.
• Non-dict drop (consumer.py:595-605): it acks via delete. It needs no _fail_dispatch because a non-dict payload has no reply_key. If delete raises, the batch loop logs it, stops, and the row redelivers.
• Comment fix (client.py:288-292): it now matches the code. A bare DatabaseError drops only a connection the client owns.
• Lens 15 (code quality): I did not run its pinned format/lint gate.
• Scratch directory: the review's scratch directory disappeared mid-run, so I validated the report from the session scratchpad instead. The checklist validator passed and the review record is closed.

Author's disposition: nothing above the severity floor, and this round reviewed the round-5 fixes, so the self-review loop is complete.

• connectors #3: the read_any integration tests were run against a real migrated Postgres at this tip (REQUIRE_PG_TESTS=1; 489 passed across the PG queue suites; full workers suite 1599 passed).
• fix: Prompt studio index and run issues #1 and Profile manager migration fix #2 are below the floor and left as recorded here (Profile manager migration fix #2 matches the existing malformed/poison drop logging in _resolve_runnable_task).

---

Generated by Claude Code

[vishnuszipstack]

Ran the PR Review Toolkit agents (code-reviewer, silent-failure-hunter, type-design-analyzer, pr-test-analyzer, comment-analyzer, code-simplifier) against 8c78d0f.

Correctness: no blocking issues found. Checked the rotation key in _dequeue_any_sql (never negative, same key in both ORDER BYs), the order of the 10 placeholders, FOR UPDATE SKIP LOCKED LIMIT inside the LATERAL, autocommit set only on owned connections, every client op being one statement, and no outside caller using PgQueueClient.conn for a multi-statement transaction. The handler-failure slice and count are also right. The accepted trade-offs in the description hold as written.

Inline comments, highest priority first: the queue_name default and the multi-queue E2E test gap (medium), then type/robustness, simplification, a test flake, and comment-accuracy nits. Nothing here blocks the merge.

[sonarqubecloud]

Quality Gate passed

Issues
1 New issue
0 Accepted issues

Measures
0 Security Hotspots
0.0% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube Cloud

[github-actions]

Unstract test results

Per-group results

Status	Group	Tier	Passed	Failed	Errors	Skipped	Duration (s)
✅	e2e-api-deployment	e2e	3	0	0	0	14.5
✅	e2e-coowners	e2e	1	0	0	0	1.3
✅	e2e-etl	e2e	1	0	0	0	14.4
✅	e2e-login	e2e	2	0	0	0	1.2
✅	e2e-prompt-studio	e2e	1	0	0	0	4.6
✅	e2e-smoke	e2e	2	0	0	0	1.4
✅	e2e-workflow	e2e	1	0	0	0	14.3
✅	frontend	unit	625	0	0	0	14.0
✅	integration-backend	integration	638	0	0	26	60.6
✅	integration-connectors	integration	1	0	0	7	9.4
✅	integration-workers	integration	179	0	0	1	40.8
❌	ui	e2e	0	1	0	0	0.0
✅	unit-backend	unit	1567	0	0	1	39.8
✅	unit-connectors	unit	82	0	0	0	10.3
✅	unit-core	unit	281	0	0	0	2.5
✅	unit-platform-service	unit	18	0	0	0	2.1
✅	unit-rig	unit	120	0	0	0	3.8
✅	unit-runner	unit	12	0	0	0	2.5
✅	unit-sdk1	unit	843	0	0	0	39.8
✅	unit-tool-sidecar	unit	8	0	0	0	1.4
✅	unit-workers	unit	1445	0	0	1	113.5
	TOTAL		5830	1	0	36	392.1

Critical paths

⚠️ Critical paths not yet covered

• workflow-execution-fan-out — Multi-file workflow execution fans out to file-processing workers and rejoins. (declared coverage: no groups declared)

✅ Covered critical paths

• auth-login — covered by e2e-login
• adapter-register-llm — covered by integration-backend
• workflow-author — covered by integration-backend
• co-owner-manage — covered by integration-backend, e2e-coowners
• workflow-create-execute — covered by e2e-workflow
• api-deployment-provision — covered by integration-backend
• api-deployment-auth — covered by integration-backend
• api-deployment-run — covered by e2e-api-deployment
• mcp-server-auth — covered by integration-backend
• mcp-platform-auth — covered by integration-backend
• platform-key-whoami — covered by integration-backend
• prompt-studio-author — covered by integration-backend
• prompt-studio-fetch-response — covered by e2e-prompt-studio
• connector-register-test — covered by integration-backend
• pipeline-etl-execute — covered by e2e-etl
• usage-aggregate-read — covered by integration-backend
• usage-token-tracking — covered by e2e-api-deployment
• callback-result-delivery — covered by e2e-api-deployment