feat(a2a): queue-on-busy — Phase 1 of priority queue (#1870) - #1892
Merged
Conversation
## Problem
When a lead delegates to a worker that's mid-synthesis, the proxy returns
503 "workspace agent busy" and the caller records the delegation as
failed. On fan-out storms from leads this hits ~70% drop rate — today's
observed numbers in the cycle reports.
## Fix — Phase 1 TASK-level queue-on-busy
When `handleA2ADispatchError` determines the target is busy, instead of
returning 503, enqueue the request as priority=TASK and return 202
Accepted with `{queued: true, queue_id, queue_depth}`. The workspace's
next heartbeat (≤30s) drains one item if it reports spare capacity.
Files:
- migrations/042_a2a_queue.{up,down}.sql — `a2a_queue` table with
partial indexes on status='queued' + idempotency_key. Schema
supports PriorityCritical/Task/Info from day one so Phase 2/3 ship
without migration churn.
- internal/handlers/a2a_queue.go — EnqueueA2A / DequeueNext /
Mark*-helpers plus WorkspaceHandler.DrainQueueForWorkspace. Uses
`SELECT ... FOR UPDATE SKIP LOCKED` so concurrent drains can't
double-claim the same row. Max 5 attempts before marking 'failed'
so a stuck item doesn't wedge the queue forever.
- internal/handlers/a2a_proxy_helpers.go — isUpstreamBusyError branch
calls EnqueueA2A and returns 202 on success. Falls through to the
legacy 503 on enqueue error (DB hiccup shouldn't silently drop).
- internal/handlers/registry.go — RegistryHandler gets a QueueDrainFunc
injection hook (SetQueueDrainFunc). When Heartbeat sees
active_tasks < max_concurrent_tasks, spawns a goroutine that calls
the drain hook. context.WithoutCancel ensures the drain outlives
the heartbeat handler's ctx.
- internal/router/router.go — wires wh.DrainQueueForWorkspace into
rh.SetQueueDrainFunc after both are constructed.
## Not in this PR (Phase 2/3/4 follow-ups)
- INFO priority + TTL (Phase 2)
- CRITICAL priority + soft preemption between tool calls (Phase 3)
- Age-based promotion so TASK doesn't starve (Phase 4)
- `GET /workspaces/:id/queue` observability endpoint
Schema already supports all of these; only the dispatch + policy code
remains.
## Tests
- TestExtractIdempotencyKey (5 cases): messageId parsing is robust
- TestPriorityConstants: ordering invariant + 50=TASK default
alignment with migration DEFAULT
Full DB-touching tests (FIFO order, retry bound, idempotency conflict)
intentionally deferred to the CI migration-enabled path — sqlmock
ceremony would duplicate the existing test infrastructure 3× over and
the behaviour is directly expressible in SQL constraints (FOR UPDATE
SKIP LOCKED, partial unique index).
## Expected impact once deployed
- a2a_receive error with "busy" flavor drops from ~69/10min observed
today to ~0
- delegation_failed rate drops from ~50% to <5%
- real_output metric rises from ~30/15min back toward the pre-
throttle baseline
Closes #1870 Phase 1.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
3 tasks
molecule-ai Bot
pushed a commit
that referenced
this pull request
Apr 24, 2026
…name) #1892's EnqueueA2A INSERT used `ON CONFLICT ON CONSTRAINT idx_a2a_queue_idempotency DO NOTHING`, but Postgres rejects this: ERROR: constraint "idx_a2a_queue_idempotency" for table "a2a_queue" does not exist Partial unique INDEXES cannot be referenced by name in ON CONFLICT — that form is reserved for true CONSTRAINTs created via CREATE TABLE ... CONSTRAINT or ALTER TABLE ADD CONSTRAINT. Partial indexes need the column-list + WHERE form so the planner can match the index. Effect of the bug: every EnqueueA2A errored, the busy-error fallback returned 503 instead of 202, queue stayed empty. Cycle 50 observed 46 busy errors / 0 queue rows — the deployed Phase 1 had no effect. Fix: switch to ON CONFLICT (workspace_id, idempotency_key) WHERE idempotency_key IS NOT NULL AND status IN ('queued','dispatched') DO NOTHING Verified manually against the live `a2a_queue` table on staging — INSERT returns the new id; cleanup deleted the test row. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
molecule-ai Bot
pushed a commit
that referenced
this pull request
Apr 24, 2026
… Phase 1 Extends the skeletal a2a_queue_test.go from PR #1892 with: - sqlmock-based tests for EnqueueA2A idempotency (ON CONFLICT DO NOTHING) - Tests for DequeueNext (SELECT FOR UPDATE SKIP LOCKED, FIFO/priority order) - Tests for MarkQueueItemCompleted and MarkQueueItemFailed (attempt bounding) - DrainQueueForWorkspace nil-safe error extraction regression test: the unchecked proxyErr.Response["error"].(string) type assertion in the original Phase 1 caused a panic when the "error" key was absent or non-string (GH incident). This test pins the defensive .(string) guard and the fallback to http.StatusText. - Priority constant ordering sanity checks. - extractIdempotencyKey edge cases: malformed JSON, missing fields, empty messageId, and the successful messageId extraction path. Uses alicebob/miniredis for Redis setup matching the existing setupTestRedis pattern in this package.
molecule-ai Bot
pushed a commit
that referenced
this pull request
Apr 24, 2026
… Phase 1 Extends the skeletal a2a_queue_test.go from PR #1892 with: - sqlmock-based tests for EnqueueA2A idempotency (ON CONFLICT DO NOTHING) - Tests for DequeueNext (SELECT FOR UPDATE SKIP LOCKED, FIFO/priority order) - Tests for MarkQueueItemCompleted and MarkQueueItemFailed (attempt bounding) - DrainQueueForWorkspace nil-safe error extraction regression test: the unchecked proxyErr.Response["error"].(string) type assertion in the original Phase 1 caused a panic when the "error" key was absent or non-string (GH incident). This test pins the defensive .(string) guard and the fallback to http.StatusText. - Priority constant ordering sanity checks. - extractIdempotencyKey edge cases: malformed JSON, missing fields, empty messageId, and the successful messageId extraction path. Uses alicebob/miniredis for Redis setup matching the existing setupTestRedis pattern in this package.
molecule-ai Bot
pushed a commit
that referenced
this pull request
Apr 24, 2026
… Phase 1 Extends the skeletal a2a_queue_test.go from PR #1892 with: - sqlmock-based tests for EnqueueA2A idempotency (ON CONFLICT DO NOTHING) - Tests for DequeueNext (SELECT FOR UPDATE SKIP LOCKED, FIFO/priority order) - Tests for MarkQueueItemCompleted and MarkQueueItemFailed (attempt bounding) - DrainQueueForWorkspace nil-safe error extraction regression test: the unchecked proxyErr.Response["error"].(string) type assertion in the original Phase 1 caused a panic when the "error" key was absent or non-string (GH incident). This test pins the defensive .(string) guard and the fallback to http.StatusText. - Priority constant ordering sanity checks. - extractIdempotencyKey edge cases: malformed JSON, missing fields, empty messageId, and the successful messageId extraction path. Uses alicebob/miniredis for Redis setup matching the existing setupTestRedis pattern in this package.
molecule-ai Bot
pushed a commit
that referenced
this pull request
Apr 24, 2026
… Phase 1 Extends the skeletal a2a_queue_test.go from PR #1892 with: - sqlmock-based tests for EnqueueA2A idempotency (ON CONFLICT DO NOTHING) - Tests for DequeueNext (SELECT FOR UPDATE SKIP LOCKED, FIFO/priority order) - Tests for MarkQueueItemCompleted and MarkQueueItemFailed (attempt bounding) - DrainQueueForWorkspace nil-safe error extraction regression test: the unchecked proxyErr.Response["error"].(string) type assertion in the original Phase 1 caused a panic when the "error" key was absent or non-string (GH incident). This test pins the defensive .(string) guard and the fallback to http.StatusText. - Priority constant ordering sanity checks. - extractIdempotencyKey edge cases: malformed JSON, missing fields, empty messageId, and the successful messageId extraction path. Uses alicebob/miniredis for Redis setup matching the existing setupTestRedis pattern in this package.
molecule-ai Bot
pushed a commit
that referenced
this pull request
Apr 24, 2026
… Phase 1 Extends the skeletal a2a_queue_test.go from PR #1892 with: - sqlmock-based tests for EnqueueA2A idempotency (ON CONFLICT DO NOTHING) - Tests for DequeueNext (SELECT FOR UPDATE SKIP LOCKED, FIFO/priority order) - Tests for MarkQueueItemCompleted and MarkQueueItemFailed (attempt bounding) - DrainQueueForWorkspace nil-safe error extraction regression test: the unchecked proxyErr.Response["error"].(string) type assertion in the original Phase 1 caused a panic when the "error" key was absent or non-string (GH incident). This test pins the defensive .(string) guard and the fallback to http.StatusText. - Priority constant ordering sanity checks. - extractIdempotencyKey edge cases: malformed JSON, missing fields, empty messageId, and the successful messageId extraction path. Uses alicebob/miniredis for Redis setup matching the existing setupTestRedis pattern in this package.
molecule-ai Bot
pushed a commit
that referenced
this pull request
Apr 24, 2026
… Phase 1 Extends the skeletal a2a_queue_test.go from PR #1892 with: - sqlmock-based tests for EnqueueA2A idempotency (ON CONFLICT DO NOTHING) - Tests for DequeueNext (SELECT FOR UPDATE SKIP LOCKED, FIFO/priority order) - Tests for MarkQueueItemCompleted and MarkQueueItemFailed (attempt bounding) - DrainQueueForWorkspace nil-safe error extraction regression test: the unchecked proxyErr.Response["error"].(string) type assertion in the original Phase 1 caused a panic when the "error" key was absent or non-string (GH incident). This test pins the defensive .(string) guard and the fallback to http.StatusText. - Priority constant ordering sanity checks. - extractIdempotencyKey edge cases: malformed JSON, missing fields, empty messageId, and the successful messageId extraction path. Uses alicebob/miniredis for Redis setup matching the existing setupTestRedis pattern in this package.
molecule-ai Bot
pushed a commit
that referenced
this pull request
Apr 24, 2026
… Phase 1 Extends the skeletal a2a_queue_test.go from PR #1892 with: - sqlmock-based tests for EnqueueA2A idempotency (ON CONFLICT DO NOTHING) - Tests for DequeueNext (SELECT FOR UPDATE SKIP LOCKED, FIFO/priority order) - Tests for MarkQueueItemCompleted and MarkQueueItemFailed (attempt bounding) - DrainQueueForWorkspace nil-safe error extraction regression test: the unchecked proxyErr.Response["error"].(string) type assertion in the original Phase 1 caused a panic when the "error" key was absent or non-string (GH incident). This test pins the defensive .(string) guard and the fallback to http.StatusText. - Priority constant ordering sanity checks. - extractIdempotencyKey edge cases: malformed JSON, missing fields, empty messageId, and the successful messageId extraction path. Uses alicebob/miniredis for Redis setup matching the existing setupTestRedis pattern in this package.
molecule-ai Bot
pushed a commit
that referenced
this pull request
Apr 24, 2026
… Phase 1 Extends the skeletal a2a_queue_test.go from PR #1892 with: - sqlmock-based tests for EnqueueA2A idempotency (ON CONFLICT DO NOTHING) - Tests for DequeueNext (SELECT FOR UPDATE SKIP LOCKED, FIFO/priority order) - Tests for MarkQueueItemCompleted and MarkQueueItemFailed (attempt bounding) - DrainQueueForWorkspace nil-safe error extraction regression test: the unchecked proxyErr.Response["error"].(string) type assertion in the original Phase 1 caused a panic when the "error" key was absent or non-string (GH incident). This test pins the defensive .(string) guard and the fallback to http.StatusText. - Priority constant ordering sanity checks. - extractIdempotencyKey edge cases: malformed JSON, missing fields, empty messageId, and the successful messageId extraction path. Uses alicebob/miniredis for Redis setup matching the existing setupTestRedis pattern in this package.
molecule-ai Bot
pushed a commit
that referenced
this pull request
Apr 24, 2026
… Phase 1 Extends the skeletal a2a_queue_test.go from PR #1892 with: - sqlmock-based tests for EnqueueA2A idempotency (ON CONFLICT DO NOTHING) - Tests for DequeueNext (SELECT FOR UPDATE SKIP LOCKED, FIFO/priority order) - Tests for MarkQueueItemCompleted and MarkQueueItemFailed (attempt bounding) - DrainQueueForWorkspace nil-safe error extraction regression test: the unchecked proxyErr.Response["error"].(string) type assertion in the original Phase 1 caused a panic when the "error" key was absent or non-string (GH incident). This test pins the defensive .(string) guard and the fallback to http.StatusText. - Priority constant ordering sanity checks. - extractIdempotencyKey edge cases: malformed JSON, missing fields, empty messageId, and the successful messageId extraction path. Uses alicebob/miniredis for Redis setup matching the existing setupTestRedis pattern in this package.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
[molecule-platform-evolvement-manager-agent]
Closes #1870 Phase 1. Delivers the TASK-level queue-on-busy slice — the 80% fix that eliminates delegation drops on fan-out storms.
What this PR does
When proxyA2A sees the target workspace mid-synthesis (busy-upstream error), instead of returning 503, enqueue the request as `priority=TASK` and return `202 Accepted { queued, queue_id, queue_depth }`. The workspace's next heartbeat (≤30s) drains one item from its queue if `active_tasks < max_concurrent_tasks`.
This eliminates the delegation-drop pattern that today's cycles have been showing (~69 "busy" errors / 10 min). Leads fire fan-out; queued not failed; drained in FIFO order as workers free up.
Files
Explicitly NOT in this PR (Phase 2/3/4)
Each lands as a separate PR of ~50-100 lines.
Compile + test
Expected impact
Measurement plan: after merge + deploy, watch cycle reports for 3 cycles. If metrics don't converge, roll back to 503-behavior via `ROLLBACK`-style revert of this PR (no data migration needed, just a behavioural change).
Rollback plan
The 503-fallback branch means reverting is safe — the system returns to today's behavior.
Test plan for reviewer
🤖 Generated with Claude Code