diff options
| author | Adam Malczewski <[email protected]> | 2026-06-01 07:51:23 +0900 |
|---|---|---|
| committer | Adam Malczewski <[email protected]> | 2026-06-01 07:51:23 +0900 |
| commit | dbc3b36f6d94d719cb1a07074e5d74ce22d2fad3 (patch) | |
| tree | 8454bd10191ce1ccbe9740b5778ee1e9679f6339 /packages/api/src | |
| parent | 8b9533c22a47bbf6f916667e2c25d8e8e419da37 (diff) | |
| download | dispatch-dbc3b36f6d94d719cb1a07074e5d74ce22d2fad3.tar.gz dispatch-dbc3b36f6d94d719cb1a07074e5d74ce22d2fad3.zip | |
fix(queue): consume queued messages after a turn ends (start a new turn)
A message queued while the agent was mid-turn was only handled if it
arrived DURING a tool batch (injected as a [USER INTERRUPT]). If it
landed after the last tool call — or the turn had no tools — the agent
silently appended it to history and ended the turn with no response, so
it sat there unanswered. This affected both user-queued messages and
agent-queued ones (send_to_tab).
- agent.ts: stop the end-of-turn drain that swallowed trailing queued
messages into history. They now stay on the queue.
- agent-manager: after a CLEAN turn settles, continueFromQueue() drains
the queue and starts a fresh turn to answer it. Skipped on a
user-stopped or errored turn (queue preserved for the next send).
- Loop safety: continuation draws from the existing autoWakeBudget, so a
runaway agent<->agent chain is bounded; human sends refill it, so human
conversations are never throttled.
- dequeueMessages now tags message-consumed with reason
"interrupt" | "continuation"; the frontend collapses continuation-
consumed queued bubbles into the next turn's initiator row (avoids the
linger/dup traps documented in queue-interrupt-reconcile-edge-cases.md).
- Tests: agent (no-swallow + interrupt regression), agent-manager
(continuation, no-op when empty, user-stop preserves queue, bounded
loop), frontend (continuation bubble becomes next initiator).
- wishlist: remove the now-fixed item.
Diffstat (limited to 'packages/api/src')
| -rw-r--r-- | packages/api/src/agent-manager.ts | 79 |
1 files changed, 77 insertions, 2 deletions
diff --git a/packages/api/src/agent-manager.ts b/packages/api/src/agent-manager.ts index 517c661..5a0ffdf 100644 --- a/packages/api/src/agent-manager.ts +++ b/packages/api/src/agent-manager.ts @@ -1628,6 +1628,75 @@ export class AgentManager { } else { tabAgent.completionResolve?.({ status: "error", error: processError }); } + + // The turn has fully settled. If messages piled up on the queue during it + // and were NOT injected as a mid-turn interrupt (they arrived after the + // last tool call, or this turn had no tool calls), kick off a fresh turn + // to answer them instead of letting them sit unanswered — the queue is + // consumed, not just appended. Only on a clean finish: a turn the user + // explicitly stopped, or one that errored out, leaves its queue intact + // for the next deliberate send (see continueFromQueue). + if (processError === null) { + this.continueFromQueue(tabId); + } + } + + /** + * Start a new turn for any messages that accumulated on `tabId`'s queue + * during the turn that just finished. This is what makes a queued message + * (from a user OR another agent via send_to_tab) actually get a response + * after the agent's current turn ends, rather than waiting forever. + * + * Loop safety: a queued-then-continued turn draws from the SAME + * `autoWakeBudget` that bounds agent-to-agent wakes. Every human-originated + * message refills that budget when it is delivered (see deliverMessage), so + * human conversations are never throttled; only a runaway agent<->agent + * chain (A queues B, B queues A, ...) is capped. When the budget is spent + * the messages stay queued and a notice is emitted; the next human message + * refills the budget and starts their turn. + */ + private continueFromQueue(tabId: string): void { + const tabAgent = this.tabAgents.get(tabId); + if (!tabAgent) return; + if (tabAgent.messageQueue.length === 0) return; + // Never auto-continue a turn the user stopped or one that errored. + if (tabAgent.status === "error") return; + if (tabAgent.abortController?.signal.aborted) return; + + if (tabAgent.autoWakeBudget <= 0) { + // Budget spent — hold the queued messages (don't drop them) until a + // human message refills the budget. Prevents unbounded agent loops. + const notice = + `Automatic continuation limit reached for this tab ` + + `(${MAX_AGENT_AUTO_WAKES} consecutive turns). Queued messages are held ` + + `until you send a message here.`; + this.emit({ type: "notice", message: notice }, tabId); + this.routeSystemEventToTab(tabId, "notice", notice); + return; + } + tabAgent.autoWakeBudget -= 1; + + // Drain the queue as a "continuation" so the frontend folds the pending + // queued bubbles into this NEW turn's initiating user row (rather than + // into a running turn's tool result, which is the "interrupt" case). + const drained = this.dequeueMessages(tabId, "continuation"); + if (drained.length === 0) return; + const message = drained.map((m) => m.message).join("\n---\n"); + + // Reuse the tab's resolved key/model/fallback chain — the continuation is + // the same conversation, just a new turn. Fire-and-forget: if more + // messages arrive during it, its own tail will continue the chain. + this.processMessage( + tabId, + message, + tabAgent.keyId ?? undefined, + tabAgent.modelId ?? undefined, + undefined, + undefined, + tabAgent.agentModels, + ).catch((err) => { + console.error(`[dispatch] continueFromQueue processMessage error for tab ${tabId}:`, err); + }); } private buildFallbackSequence( @@ -1673,13 +1742,19 @@ export class AgentManager { return true; } - dequeueMessages(tabId: string): QueuedMessage[] { + dequeueMessages( + tabId: string, + reason: "interrupt" | "continuation" = "interrupt", + ): QueuedMessage[] { const tabAgent = this.tabAgents.get(tabId); if (!tabAgent) return []; const messages = [...tabAgent.messageQueue]; tabAgent.messageQueue = []; if (messages.length > 0) { - this.emit({ type: "message-consumed", tabId, messageIds: messages.map((m) => m.id) }, tabId); + this.emit( + { type: "message-consumed", tabId, messageIds: messages.map((m) => m.id), reason }, + tabId, + ); } return messages; } |
