fix(agent-runner): drain the parallelism queue when an ACP conversation ends - #483
Conversation
…on ends services/api starts a queued conversation only when AgentQueueConsumer sees a terminal status on StreamAgentConversationStatus (AdvanceQueue). Sandboxed agents publish it from handler.publishTerminalStatus, but none of the three paths that end an ACP conversation did: - acpbridge.Server.handleTurnStatusMessage (the bridge reports finished/failed) - acpbridge.Dispatcher.failOffline (bridge not connected) - acpbridge.Dispatcher.watchdog (turn timed out) They wrote the DB status and a realtime event only. So once an ACP agent was at its parallelism limit, every conversation queued behind it stayed "queued" forever; only StopConversation, which publishes on its own, ever freed one. Publish the terminal status to the stream on all three paths. AdvanceQueue already tolerates a duplicate event (claimQueuedForDispatch), so a late turn_status racing the watchdog is harmless. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
There was a problem hiding this comment.
ℹ️ The fix is correct and well-documented; two rough edges worth a look, both non-blocking.
Reviewed changes
- ACP terminal-status publishing on all three end paths —
handleTurnStatusMessage,failOffline, andwatchdognow append the conversation's terminal status toStreamAgentConversationStatus(paca:agent:conversation_status), the only eventservices/api'sworker.AgentQueueConsumerreacts to. This is the missing half of the parallelism-queue drain for ACP agents that previously only happened viaStopConversation. - New
Dispatcher.publishQueueStatushelper (dispatch.go) — best-effort, warn-only publish; andisTerminalStatusmirroringagentdom.ConversationStatus.IsTerminal. queue_status_test.go— stream-payload round-trip, unreachable-Valkey tolerance, andisTerminalStatuscoverage via miniredis (consistent withregistry_test.go's conventions).
I traced the full producer→consumer contract: the stream payload shape (conversation_id/status) matches AgentQueueConsumer.handle's decode exactly, isTerminalStatus accepts precisely the status set that consumer reacts to, duplicate/racing events are safe (every AdvanceQueue goes through the claimQueuedForDispatch CAS, so re-measured capacity dispatches nothing extra), and the apps/acp-bridge daemon's own terminal vocabulary (finished/failed) is fully covered. A crash between the DB-terminal write and the publish is secondarily caught by reconcileStaleConversations on bridge reconnect. I built and ran the package tests with -race — they pass.
Two suggestions, inline and non-blocking:
DeepSeek Flash (free via Pullfrog for OSS) | 𝕏
| if err := s.Publisher.PublishConversationStatus(ctx, convID, statusStr); err != nil { | ||
| s.Log.Warn("acpbridge: failed to publish conversation status", "conversation_id", convID, "status", statusStr, "error", err) | ||
| } | ||
| } |
There was a problem hiding this comment.
The tests cover the shared helper (publishQueueStatus) and the gate (isTerminalStatus) but not the three call sites this PR actually modified: handleExitStatusMessage's terminal-gate wiring here, failOffline, and the watchdog drain. A future refactor that, say, swaps the arg order or guards the publish differently (or forgets one path) would pass CI while silently re-introducing the canary bug. Consider one integration-style test that drives handleTurnStatusMessage through a fake ConvRepo.UpdateStatus and asserts the stream entry (and a non-terminal status stays absent), which is the risk boundary here.
| // should hear about ("paused" and "running" are not). | ||
| func isTerminalStatus(status string) bool { | ||
| return status == "finished" || status == "failed" || status == "stopped" | ||
| } |
There was a problem hiding this comment.
isTerminalStatus duplicates services/api's agentdom.ConversationStatus.IsTerminal as a hard-coded string list (the same way the stream key strings are hand-synced between the services). Fine as-is, but the daemon drifts-closed coupling means a new terminal status anywhere in the pipeline needs this constant updated in lockstep — worth a one-line comment pointing at apps/acp-bridge's reportStatus call sites as the producer of these strings, or ideally a test listing the exact statuses the daemon emits. The current TestIsTerminalStatus table already documents intent, so this is a nit.
pikann
left a comment
There was a problem hiding this comment.
LGTM! Thank you for the contribution!

Summary
For ACP agents, a conversation queued behind a busy agent is never started. It stays
queueduntil someone presses Stop on a conversation of that agent.services/apistarts queued work only inAdvanceQueue, whichworker.AgentQueueConsumerruns when a terminal status arrives onStreamAgentConversationStatus(paca:agent:conversation_status). Sandboxed agents publish that event fromhandler.publishTerminalStatus. The three paths that end an ACP conversation only write the DB status and a realtime event:acpbridge/server.gohandleTurnStatusMessageacpbridge/dispatch.gofailOfflineacpbridge/dispatch.gowatchdogStopConversationpublishes the event itself, so Stop is the only thing that ever drains an ACP agent's backlog. Because ACP agents dispatch serially (requiresSerialDispatch), a second trigger for a busy ACP agent always lands inagent_pending_triggersand stays there. That covers a task assignment, a comment mention, and a chat withon_busy=queue. On our deployment, conversations satqueuedfor 18–20 hours while the same agent finished other work.Fix: publish the terminal status to the stream on all three ACP paths, as
handler.publishTerminalStatusalready does for sandboxed agents.handleTurnStatusMessage, it publishes after the status is recorded, only for terminal statuses (finished/failed/stopped, mirroringagentdom.ConversationStatus.IsTerminal). It publishes before the realtime-context lookup, so a failure there can't swallow it.failOfflineandwatchdog, it publishes afterfailedis written.A duplicate event is harmless.
AdvanceQueuegoes throughclaimQueuedForDispatch, which is designed for at-least-once delivery. So a lateturn_statusracing the watchdog, or a bridge-reportedstoppedafterStopConversationalready published, re-measures capacity and dispatches nothing extra.Reproduced and verified on a real v0.15.0 deployment:
on_busy=queue. B isqueued.agent-runner, B was stillqueued90 s after A finished.Type of Change
services/agent-runner)Checklist
internal/acpbridge, plus one new test file.internal/acpbridge/queue_status_test.go(miniredis, same setup asregistry_test.go). It checks three things:paca:agent:conversation_statuswith the right fields;isTerminalStatusmatchesIsTerminal.gofmt,go vet ./...,go build ./...,go test -race ./...and the Docker build pass forservices/agent-runner, on this branch and on v0.15.0 with this commit applied.🤖 Generated with Claude Code