feat: add parallelism limit for agents and implement task queue

[pikann]

Summary

Fixes #462: assigning several tickets to one agent made it work on all of them at once, and for ACP/environment-backed agents every conversation shares the same on-disk working directory, so concurrent turns stepped on each other's files. Adds a per-agent parallelism limit (default 1) backed by a real, durable task queue, plus a confirm dialog so an interactive chat message that would exceed the limit can be queued or force-sent.

Core feature

• New agents.parallelism_limit column (default 1, CHECK (parallelism_limit >= 1)) and a new agent_pending_triggers table (migration 000053_add_agent_parallelism_queue.sql) that durably stores a trigger's topic + payload when it can't be dispatched immediately, leaving the conversation in status queued until a running slot frees up.
• worker.AgentQueueConsumer reads StreamAgentConversationStatus and, on a terminal status, dequeues and replays the oldest waiting trigger for that agent (and, independently, for that environment folder — see below). Raising an agent's parallelism_limit also triggers a synchronous catch-up dispatch for the newly freed slots.
• Non-interactive triggers (task assignment, comment mention, description write) always queue silently when the agent is at capacity. Interactive chat (StartChatSession, SendChatMessage, and their global-agent equivalents) gain an on_busy request field ("queue" or "force"); when omitted and the agent is at capacity, the request is rejected up front (agent_parallelism_limit_reached, HTTP 409) with the running/limit counts, and the frontend's new busy dialog lets the user pick "Add to Queue" or "Send Now" without ever creating a conversation row for a cancelled request.

Shared environment folders

• Parallelism limit only bounds one agent's own concurrent conversations, but two different agents (or two conversations of the same agent via an explicit per-conversation environment/folder override) can still target the same shared environment folder at once. Added an independent folder-capacity check (agent_environment_folder_busy, HTTP 409) that gates dispatch the same way, backed by environment_id/environment_folder_id columns on agent_pending_triggers and a folder-scoped dequeue path in AdvanceFolderQueue.
• Folder occupancy is hierarchy-aware: a parent folder and any folder nested inside it resolve to the same underlying working directory, so a running conversation in a child folder blocks a new one in the parent (and vice versa), via Postgres starts_with() path-prefix matching rather than exact-folder equality. AdvanceFolderQueue checks each dequeued candidate's own folder freshly (not just the one folder that just freed up), so a blocked sibling can't stall a genuinely-dispatchable one behind it in the queue.

Concurrency safety

• Dispatch is now claim-before-publish: a queued conversation's status is atomically CAS'd from queued to running (ClaimConversationStatus) before its trigger is published, closing two races — Valkey's at-least-once redelivery double-dispatching the same trigger, and a conversation being stopped at the same moment the queue tries to advance it.
• StopConversation/StopGlobalConversation now also delete any pending-trigger row for the conversation being stopped, so a conversation stopped while still waiting in the backlog can't be dequeued and dispatched afterward.
• ACP agents and any agent with a default environment (working-directory-backed) are restricted to parallelism_limit = 1 and always dispatch serially (requiresSerialDispatch/effectiveParallelismLimit) — CreateAgent/UpdateAgent reject a higher limit for these agent types (agent_parallelism_limit_unsupported, HTTP 400), since the underlying ACP session/sandbox model isn't built to run more than one turn at a time against the same working directory.

Frontend

• New parallelism limit field on the agent settings page, disabled with an explanatory hint for ACP/environment-backed agents.
• New AgentBusyDialog/useAgentBusyPrompt shown from the chat composer and new-conversation flow when the API returns either busy error, offering "Add to Queue" or "Send Now" (or just an informational "folder busy" variant when the target folder — not this agent's own limit — is what's occupied).
• New strings translated into all 9 supported locales.

Type of Change

• New feature
• Bug fix (race conditions around duplicate/conflicting dispatch)

Checklist

• The change is focused and scoped to [Feature] Per-agent parallelism limit #462.
• Related documentation is updated (migration doc comments explain the schema's purpose and hierarchy semantics).
• New structure or direction is explained clearly.
• I avoided unnecessary detail or premature abstraction.

🤖 Generated with Claude Code

[pullfrog]

Important

The feature is well-built and the tests are genuinely regression-guarding (they fail against the plain "skip and continue" implementations), but the parallelism guarantee is enforced as a soft limit on three fronts: the capacity checks are read-then-act with no atomic reservation, one reply path never runs the check at all, and a crash between dequeue and publish silently loses queued work. Each is detailed below — none should block a pilot merge, but they should be resolved or explicitly accepted as design trade-offs before this PR is done.

Reviewed changes

• Schema: agents.parallelism_limit (default 1, capped at 10) plus the agent_pending_triggers table, migration 000053 with IF NOT EXISTS idempotency and indexes for the two FIFO dequeue shapes.
• Capacity checks: checkParallelismCapacity / checkFolderCapacity / checkDispatchCapacity with an on_busy value (""/queue/force) threading through all four chat endpoints (StartChatSession, SendChatMessage, and the two global siblings); folder occupancy matches ancestors/descendants via folderOverlapPredicate.
• Task queue: dispatchOrEnqueue and the durable deliverTrigger path, AdvanceQueue / AdvanceFolderQueue with requeueSkipped position preservation and claimQueuedForDispatch closing the stop/redelivery races.
• Worker: new AgentQueueConsumer on StreamAgentConversationStatus (own consumer group, PEL replay at startup) advancing the queue on terminal statuses; StopConversation/StopGlobalConversation now delete a still-queued conversation's pending trigger.
• Web: parallelism_limit input in the agent dialog (disabled for ACP / env-backed agents), the useAgentBusyPrompt dialog wired into the chat-session send paths, and i18n for 9 locales.
• Tests: ~30 new service tests covering requiresSerialDispatch, effectiveParallelismLimit, capacity composition, queue advance/requeue, and the claim-before-dispatch and claim-fail-skip behaviors.

Read-your-own-effort check

For the record I verified the three concerns below straight against the checked-out HEAD source, not just the diff: createConversation always persists status queued, dequeueOldestPendingTrigger deletes the row inside its transaction (committed before dispatch), resumeConversationMessage claims straight to running with no capacity call, agent-runner only serializes per conversation_id (not per agent), and Publisher.AppendFlat never trims the status stream.

⚠️ Reply-in-place resume bypasses both capacity checks

SendConversationMessage (services/api/internal/service/agent/agent_service.go:1733) and its resumeConversationMessage helper (agent_service.go:1768) — the server path a reply takes for exactly the agent types this feature constrains hardest, ACP and environment-attached conversations, which requiresSerialDispatch forces to parallelism_limit = 1 — claim the conversation to running and publish the trigger with no checkDispatchCapacity call at all. A user replying to an env-attached conversation while another conversation of the same agent is still running in the same environment gets a second concurrent turn past the forced limit, and the folder-occupancy constraint is skipped on the same path. The web side mirrors the gap: conversation-view.tsx's !conversation.chat_session_id branch still calls sendConversationMessage directly, so the new busy dialog / on_busy affordance does not exist there either. Note this isn't a regression (the path behaved that way before), it's an enforcement surface the feature leaves open — the fix needs to either route on_busy through sendConversationMessage too, or explicitly document that the limit only applies to chat-session dispatches.

Technical details

# Resume path bypasses parallelism / folder capacity checks

## Affected sites
- services/api/internal/service/agent/agent_service.go:1733 — SendConversationMessage, no capacity check (ACP/env branch delegates to resumeConversationMessage)
- services/api/internal/service/agent/agent_service.go:1768 — resumeConversationMessage, claims to "running" + publishes without checkDispatchCapacity
- apps/web/src/components/projects/agents/conversation-view.tsx (unchanged by this PR) — the !conversation.chat_session_id branch sends via sendConversationMessage without the busy dialog

## Required outcome
- A reply that resumes an env-attached (or ACP, where the bridge itself self-protects) conversation must go through the same dispatch-capacity decision as every other turn, or the parallelism=1 forced on these agents is not actually enforced.
- The still-"running" conversation that would violate the limit must not be able to race a second turn in the shared environment.

## Suggested approach (optional)
- Thread onBusy / the ask-queue-force contract into SendConversationMessage (handler already receives the request; add the field to its DTO), and wire the frontend's !chat_session_id branch through sendWithBusyPrompt.
- At minimum, run checkParallelismCapacity/checkFolderCapacity in resumeConversationMessage before claiming, returning the existing 409 busy codes when the agent/folder is occupied.

## Open questions for the human (optional)
- Is reply-in-place resume intentionally outside the parallelism feature's scope? If so, document it; if not, the fix above.

ℹ️ First deploy replays the entire conversation-status stream

ensureGroup(ctx, "0") creates the new api.agent_queue consumer group at stream position 0, and AppendFlat (publisher.go:83) never trims StreamAgentConversationStatus — so the deploy that first creates the group replays every historical terminal-status event the stream has ever held. Each replayed event is semantically harmless (with agent_pending_triggers empty, every AdvanceQueue is a no-op), but on a long-lived installation this is an unbounded one-time catch-up of DB round-trips, and real-time queue advancement lags until the backlog drains. If the stream has been running for months, consider trimming it (or accepting the one-time cost consciously).

Technical details

# One-time unbounded startup replay from "0"

## Affected sites
- services/api/internal/worker/agent_queue_consumer.go:98 — ensureGroup(ctx, "0") on group creation
- services/api/internal/platform/messaging/publisher.go:83 — AppendFlat, no MAXLEN/trim on the stream

## Required outcome
- Decide explicitly whether the one-time historical replay is acceptable; if not, cap the initial group position or trim the stream (XADD MAXLEN / XTRIM).

## Open questions for the human (optional)
- Is the migration-ever-before-consumer deploy ordering real in this project (which "0" is protecting against), or would "$" be safe?

ℹ️ Nitpicks

• services/api/internal/repository/postgres/agent_repository.go:1352 — the DequeueOldestPendingTriggerForFolder doc comment still says folderID nil is "matched the same IS NOT DISTINCT FROM way CountRunningConversationsInFolder uses"; both queries now use the LEFT JOIN + folderOverlapPredicate form, so the IS NOT DISTINCT FROM reference is stale wording from the first commit.

  ^{｜ Fix it ➔ ｜ View workflow run ｜ Using DeepSeek Flash (free via Pullfrog for OSS) ｜ 𝕏}

[pullfrog]

Supplement to the review just posted — two anchored findings belong inline but are detailed here since the review body only summarized them. Both live in services/api/internal/service/agent/agent_service.go.

⚠️ Capacity checks are read-then-act: the limit is soft under concurrent dispatch

checkParallelismCapacity (agent_service.go:2919) queries CountRunningConversations and returns dispatchNow=true when running < limit (agent_service.go:2932), but nothing re-verifies the count when the conversation is later claimed to running. The claim (claimConversationStatus via claimQueuedForDispatch / the resume branches) is a per-row CAS on the new conversation's own status — it never consults the agent's running count. Same pattern for the folder check.

Consequence: two concurrent dispatches for the same agent can both pass the check before either claim flips a row to running. Realistic triggers:

• a burst of task_assigned triggers (batch-assigning tickets to one agent — the feature's headline scenario) racing the sub-second window between check and claim;
• two users replying to two paused conversations of the same agent at once;
• two API replicas each handling a different terminal-status event for the same agent — both call AdvanceQueue, both count running=0, both dequeue different items (SKIP LOCKED only prevents double-dequeueing the same row) and both dispatch.

With parallelism_limit=1, that's two turns racing the same working directory — exactly the race issue #462 exists to prevent. agent-runner does not save you here: it only serializes per conversation_id, never per agent.

Fix directions (pick one): atomically reserve the slot as part of the claim, e.g. gate dispatch on an UPDATE agent_conversations SET status='running' WHERE id=$1 AND status='queued' AND (SELECT count(*) FROM agent_conversations WHERE agent_id=$2 AND status='running') < $3; take a per-agent pg_advisory_xact_lock around check+claim; or serialize all dispatch through a single consumer so the count and claim can't interleave. If a soft limit is acceptable, document that explicitly — the current doc comments don't.

⚠️ Dequeue deletes the trigger before it is published: a crash silently loses queued work

dequeueOldestPendingTrigger (agent_repository.go:1370) SELECTs and DELETEs the pending-trigger row inside one transaction, and only afterwards does AdvanceQueue → dispatchPendingTrigger resolve the conversation, claim it, and publishTrigger. Every step from the commit to the publish is an unprotected window:

• A crash (or a persistent publishTrigger/resolveWorkdirForConversation failure — e.g. the target environment was deleted while the trigger sat queued) before the publish permanently loses the trigger, and the conversation is stranded at queued forever.
• The consumer's at-least-once machinery cannot recover it: the status message isn't acked, PEL redelivery re-runs AdvanceQueue, finds the table empty, returns (0, nil), and acks. requeueSkipped only covers items that were deliberately skipped, not the in-flight one.

This contradicts the at-least-once guarantee claimQueuedForDispatch is documented to provide (its redelivery fix works precisely because the claim lands before the publish). Fix direction: make the delete conditional on the publish having happened — e.g. dispatch-then-delete inside the same transaction as the claim, or a lease-style dequeue that marks the row in-flight, publishes, and only then deletes (with stale-lease reaping for the crash case).

Both are judgment calls on how hard the enforcement needs to be; the PR is strong as-is, but these two interactions are worth an explicit decision before merge.

  ^{｜ View workflow run ｜ via Pullfrog ｜ Using DeepSeek Flash (free via Pullfrog for OSS) ｜ 𝕏}

[pullfrog]

Important

The new commits resolve the soft-limit (TOCTOU) and server-side resume gaps from the prior review, but the new atomic claim mechanism introduces one silent-loss edge on capacity races, and the web reply-in-place path now receives the new 409s with no way to act on them. Both are detailed inline.

Reviewed changes

• Addressed prior finding #1: ClaimQueuedForDispatch (agent_repository.go:1172) now atomically re-verifies agent capacity inside the claim — agents row lock (FOR UPDATE), a fresh running-count statement, then the queued→running CAS — so concurrent dispatches for the same agent can no longer both win the last slot (resume branches remain an accepted narrow soft edge).
• Addressed prior finding #3 server-side: resumeConversationMessage and sendACPGlobalConversationMessage now run checkDispatchCapacity/checkParallelismCapacity before claiming and thread onBusy through; the same gating was added to SendChatMessage/SendGlobalChatMessage's paused/terminal resume and conv==nil branches.
• Addressed prior finding #2 partially: new revertFailedDispatch reclaims a conversation back to queued and recreates the agent_pending_triggers row when publishTrigger errors — recovers process-surviving publish failures (a hard crash in the same window remains uncovered).
• Threaded on_busy through all four chat-endpoint DTOs (SendMessageRequest, StartChatSessionRequest, SendChatMessageRequest) and handlers; added regression tests for the capacity-gated resume paths, effectiveParallelismLimit wiring, and the revert behavior.

  ^{｜ Fix all ➔ ｜ Fix 👍s ➔ ｜ View workflow run ｜ Using DeepSeek Flash (free via Pullfrog for OSS) ｜ 𝕏}

[pullfrog]

✅ No new issues found.

Commit 2e2298e addresses both open inline findings from the prior review, and I could not find anything new in the delta. Verified against HEAD source and both test suites (go test ./... -race and vitest — 642 web tests) pass.

Reviewed changes

Since the prior pullfrog review (9e58d79), the new commit:

• Distinguishes capacity loss from a lost conversation in the atomic claim — ClaimQueuedForDispatch now reports atCapacity, so a conversation that loses the agent's free-slot race on the claim's re-check is re-queued instead of silently stranded at queued.
• Re-queues every capacity-loss point — deliverTrigger's fresh-dispatch path persists a PendingTrigger on capacity loss, and dispatchPendingTrigger's dequeue-replay path recreates the row preserving its original id/created_at (FIFO position preserved).
• Splits the two queue advances' reactions — AdvanceQueue stops immediately once an item hits atCapacity (same agent, so everything behind it is equally blocked); AdvanceFolderQueue keeps trying the next item, which may belong to a different agent, and requeues the blocked one.
• Wires the reply-in-place path into the busy dialog — sendConversationMessage/sendGlobalConversationMessage gained the onBusy option forwarded as on_busy, and conversation-view's !chat_session_id branch now routes through sendWithBusyPrompt.
• Adds regression tests for each new branch — capacity-race enqueue, identity-preserving requeue, stop-immediately, folder-queue continue, and on_busy forwarding; each fails against the prior behavior.

Both prior inline threads are resolved (replies posted on the threads). The two informational trade-offs from the first review — the one-time replay of StreamAgentConversationStatus when the consumer group is first created, and the crash-between-dequeue-and-publish loss window that revertFailedDispatch can't cover — are unchanged and remain the author's explicit call.

  ^{｜ View workflow run ｜ Using DeepSeek Flash (free via Pullfrog for OSS) ｜ 𝕏}

[pullfrog]

✅ No new issues found.

The two new commits fully match their bar: a defensive validateOnBusy gate with genuine regression tests, and a dedicated migration-applies-cleanly check plus CI job, together with a real schema fix (ON DELETE SET NULL on agent_pending_triggers' environment/folder FKs) that an earlier review flagged and is now locked in by an e2e regression test.

Reviewed changes

• on_busy value validation — validateOnBusy now runs at the top of all six chat entry points, rejecting anything other than ""/queue/force with ErrOnBusyInvalid (HTTP 400 AGENT_ON_BUSY_INVALID) instead of silently falling through to "ask" semantics. Six wiring tests each t.Fatal if validation doesn't run before any repo lookup, so they genuinely guard the wiring.
• Schema fix for folder/environment deletion — migration 000053 now gives agent_pending_triggers.environment_id/environment_folder_id ON DELETE SET NULL (mirroring 000042), so DeleteFolder no longer fails with a 500 FK violation when a queued trigger points at the folder — the primary steady state for an env-backed agent at its parallelism limit.
• New folder-capacity index — idx_agent_conversations_environment_status backs CountRunningConversationsInFolder.
• Migration e2e coverage — TestRealMigrations_ApplyCleanlyFromScratch applies every real .sql file to a fresh per-test database and asserts each lands in schema_migrations, running as its own fast-failing CI job; TestDeleteFolder_SucceedsWithPendingTriggerStillReferencingIt locks in the ON DELETE SET NULL behavior.

Verified go vet across every touched package (including test files), the agent/handler/presenter unit suites, and confirmed the new validateOnBusy tests actually execute and pass. I also traced the ON DELETE SET NULL degrade path: resolveWorkdirForConversation returns (nil,"",nil) for a nulled environment/folder, dispatching to an ephemeral sandbox rather than erroring into a permanent requeue — so the schema change is safe under the documented degradation.

  ^{｜ View workflow run ｜ Using DeepSeek Flash (free via Pullfrog for OSS) ｜ 𝕏}