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/tests | |
| 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/tests')
| -rw-r--r-- | packages/api/tests/agent-manager.test.ts | 125 |
1 files changed, 125 insertions, 0 deletions
diff --git a/packages/api/tests/agent-manager.test.ts b/packages/api/tests/agent-manager.test.ts index 1358eb1..b9b4510 100644 --- a/packages/api/tests/agent-manager.test.ts +++ b/packages/api/tests/agent-manager.test.ts @@ -1129,6 +1129,131 @@ describe("AgentManager", () => { }); }); + describe("queue continuation after a turn ends", () => { + // A run generator that enqueues `msg` (as if a user/agent sent it mid-turn) + // exactly once, then streams a normal short reply. Used to simulate a + // message landing on the queue while the agent is busy. + function runThatEnqueues(manager: AgentManager, tabId: string, msg: string): RunGen { + let enqueued = false; + return async function* () { + yield { type: "status", status: "running" } as const; + if (!enqueued) { + enqueued = true; + manager.queueMessage(tabId, msg); + } + yield { type: "text-delta", delta: "reply" } as const; + yield { + type: "done", + message: { role: "assistant", chunks: [{ type: "text", text: "reply" }] }, + } as const; + yield { type: "status", status: "idle" } as const; + }; + } + + it("starts a NEW turn for a message queued during the turn (the bug fix)", async () => { + const manager = new AgentManager(); + const processSpy = vi.spyOn(manager, "processMessage"); + setRunImpl(runThatEnqueues(manager, "tab-cont", "follow-up question")); + + await manager.processMessage("tab-cont", "first"); + // Let the fire-and-forget continuation turn run to completion. + await new Promise<void>((r) => setTimeout(r, 50)); + + // processMessage called twice: the original turn + the continuation. + expect(processSpy).toHaveBeenCalledTimes(2); + expect(processSpy.mock.calls[1]?.[0]).toBe("tab-cont"); + expect(processSpy.mock.calls[1]?.[1]).toBe("follow-up question"); + + // Queue is drained and the tab is idle again. + const inner = manager as unknown as { + tabAgents: Map<string, { messageQueue: unknown[] }>; + }; + expect(inner.tabAgents.get("tab-cont")?.messageQueue).toHaveLength(0); + expect(manager.getTabStatus("tab-cont")).toBe("idle"); + }); + + it('emits message-consumed with reason "continuation" when draining between turns', async () => { + const manager = new AgentManager(); + const events: AgentEvent[] = []; + manager.onEvent((e) => events.push(e)); + setRunImpl(runThatEnqueues(manager, "tab-evt", "next")); + + await manager.processMessage("tab-evt", "first"); + await new Promise<void>((r) => setTimeout(r, 50)); + + const consumed = events.find((e) => e.type === "message-consumed") as + | (AgentEvent & { reason?: string }) + | undefined; + expect(consumed).toBeDefined(); + expect(consumed?.reason).toBe("continuation"); + }); + + it("does NOT continue when the queue is empty after a clean turn", async () => { + const manager = new AgentManager(); + const processSpy = vi.spyOn(manager, "processMessage"); + + await manager.processMessage("tab-noqueue", "only message"); + await new Promise<void>((r) => setTimeout(r, 30)); + + expect(processSpy).toHaveBeenCalledTimes(1); // no continuation + }); + + it("does NOT continue a turn the user stopped (queue is preserved)", async () => { + const manager = new AgentManager(); + const processSpy = vi.spyOn(manager, "processMessage"); + // Run that enqueues then aborts itself via stopTab to mimic a user stop. + setRunImpl(async function* () { + yield { type: "status", status: "running" } as const; + manager.queueMessage("tab-stop", "should wait"); + manager.stopTab("tab-stop"); + yield { + type: "done", + message: { role: "assistant", chunks: [] }, + } as const; + }); + + await manager.processMessage("tab-stop", "go"); + await new Promise<void>((r) => setTimeout(r, 30)); + + // Only the original turn ran; the queued message is preserved, unanswered. + expect(processSpy).toHaveBeenCalledTimes(1); + const inner = manager as unknown as { + tabAgents: Map<string, { messageQueue: unknown[] }>; + }; + expect(inner.tabAgents.get("tab-stop")?.messageQueue).toHaveLength(1); + }); + + it("bounds runaway agent<->agent continuation via the auto-wake budget", async () => { + const manager = new AgentManager(); + // A run that ALWAYS enqueues another message → would loop forever + // without the budget cap. + setRunImpl(async function* () { + yield { type: "status", status: "running" } as const; + manager.queueMessage("tab-loop", "again and again"); + yield { + type: "done", + message: { role: "assistant", chunks: [{ type: "text", text: "r" }] }, + } as const; + yield { type: "status", status: "idle" } as const; + }); + const processSpy = vi.spyOn(manager, "processMessage"); + + await manager.processMessage("tab-loop", "kick off"); + await new Promise<void>((r) => setTimeout(r, 120)); + + // 1 original + at most MAX_AGENT_AUTO_WAKES (6) continuations = 7. + // Crucially BOUNDED, not infinite. + expect(processSpy.mock.calls.length).toBeLessThanOrEqual(7); + expect(processSpy.mock.calls.length).toBeGreaterThan(1); + // Budget spent; the last queued message is held, not answered. + const inner = manager as unknown as { + tabAgents: Map<string, { autoWakeBudget: number; messageQueue: unknown[] }>; + }; + expect(inner.tabAgents.get("tab-loop")?.autoWakeBudget).toBe(0); + expect(inner.tabAgents.get("tab-loop")?.messageQueue.length).toBeGreaterThan(0); + }); + }); + describe("getLastTabResponse", () => { it("returns the most recent assistant turn's text and current status", () => { const manager = new AgentManager(); |
