fix(agent-runner): drain the parallelism queue when an ACP conversation ends

[HCH-hash]

Summary

For ACP agents, a conversation queued behind a busy agent is never started. It stays queued until someone presses Stop on a conversation of that agent.

services/api starts queued work only in AdvanceQueue, which worker.AgentQueueConsumer runs when a terminal status arrives on StreamAgentConversationStatus (paca:agent:conversation_status). Sandboxed agents publish that event from handler.publishTerminalStatus. The three paths that end an ACP conversation only write the DB status and a realtime event:

path	file	status
bridge reports the turn result	acpbridge/server.go handleTurnStatusMessage	finished / failed / …
bridge not connected at dispatch	acpbridge/dispatch.go failOffline	failed
no result before the timeout	acpbridge/dispatch.go watchdog	failed

StopConversation publishes 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 in agent_pending_triggers and stays there. That covers a task assignment, a comment mention, and a chat with on_busy=queue. On our deployment, conversations sat queued for 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.publishTerminalStatus already does for sandboxed agents.

• In handleTurnStatusMessage, it publishes after the status is recorded, only for terminal statuses (finished/failed/stopped, mirroring agentdom.ConversationStatus.IsTerminal). It publishes before the realtime-context lookup, so a failure there can't swallow it.
• In failOffline and watchdog, it publishes after failed is written.

A duplicate event is harmless. AdvanceQueue goes through claimQueuedForDispatch, which is designed for at-least-once delivery. So a late turn_status racing the watchdog, or a bridge-reported stopped after StopConversation already published, re-measures capacity and dispatches nothing extra.

Reproduced and verified on a real v0.15.0 deployment:

1. Send chat A to an idle ACP agent.
2. While A runs, send chat B to the same agent with on_busy=queue. B is queued.

• On the stock agent-runner, B was still queued 90 s after A finished.
• With this patch (v0.15.0 plus this commit only), B started 0 s after A finished and then finished normally.

Type of Change

• Other: bug fix (services/agent-runner)

Checklist

• The change is focused and scoped. It touches two files in internal/acpbridge, plus one new test file.
• Tests added: internal/acpbridge/queue_status_test.go (miniredis, same setup as registry_test.go). It checks three things:
  • the entry lands on paca:agent:conversation_status with the right fields;
  • a Valkey outage is logged and doesn't panic;
  • isTerminalStatus matches IsTerminal.
• gofmt, go vet ./..., go build ./..., go test -race ./... and the Docker build pass for services/agent-runner, on this branch and on v0.15.0 with this commit applied.
• No docs change: this restores the documented behaviour of the parallelism queue for ACP agents.
• Not in scope: the ACP serial-dispatch rule itself.

🤖 Generated with Claude Code

[pullfrog]

ℹ️ 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, and watchdog now append the conversation's terminal status to StreamAgentConversationStatus (paca:agent:conversation_status), the only event services/api's worker.AgentQueueConsumer reacts to. This is the missing half of the parallelism-queue drain for ACP agents that previously only happened via StopConversation.
• New Dispatcher.publishQueueStatus helper (dispatch.go) — best-effort, warn-only publish; and isTerminalStatus mirroring agentdom.ConversationStatus.IsTerminal.
• queue_status_test.go — stream-payload round-trip, unreachable-Valkey tolerance, and isTerminalStatus coverage via miniredis (consistent with registry_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:

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

[pullfrog]

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.

[pullfrog]

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.