summaryrefslogtreecommitdiffhomepage
path: root/packages/api/src
diff options
context:
space:
mode:
authorAdam Malczewski <[email protected]>2026-06-01 07:51:23 +0900
committerAdam Malczewski <[email protected]>2026-06-01 07:51:23 +0900
commitdbc3b36f6d94d719cb1a07074e5d74ce22d2fad3 (patch)
tree8454bd10191ce1ccbe9740b5778ee1e9679f6339 /packages/api/src
parent8b9533c22a47bbf6f916667e2c25d8e8e419da37 (diff)
downloaddispatch-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.ts79
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;
}