summaryrefslogtreecommitdiffhomepage
path: root/packages/frontend
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/frontend
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/frontend')
-rw-r--r--packages/frontend/src/lib/tabs.svelte.ts46
-rw-r--r--packages/frontend/src/lib/types.ts16
-rw-r--r--packages/frontend/tests/chat-store.test.ts82
3 files changed, 142 insertions, 2 deletions
diff --git a/packages/frontend/src/lib/tabs.svelte.ts b/packages/frontend/src/lib/tabs.svelte.ts
index 317de8d..3fd7e5f 100644
--- a/packages/frontend/src/lib/tabs.svelte.ts
+++ b/packages/frontend/src/lib/tabs.svelte.ts
@@ -1223,7 +1223,11 @@ export function createTabStore() {
}
case "message-consumed": {
if (!tabId) break;
- const mcEvent = event as AgentEvent & { tabId: string; messageIds: string[] };
+ const mcEvent = event as AgentEvent & {
+ tabId: string;
+ messageIds: string[];
+ reason?: "interrupt" | "continuation";
+ };
const mcTab = getTabById(tabId);
if (!mcTab) break;
// Track recently consumed IDs so sendMessage can detect early consumption
@@ -1234,6 +1238,46 @@ export function createTabStore() {
updateTab(tabId, {
queuedMessages: mcTab.queuedMessages.filter((m) => !mcEvent.messageIds.includes(m.id)),
});
+
+ // "continuation" — these queued messages are draining BETWEEN turns
+ // to START a fresh turn (the "queue consumed after turn ends" path),
+ // not folding into a running turn's tool result. The backend joins
+ // them into ONE initiating user row, so we collapse the matching
+ // optimistic `queued-` bubbles into a single UNTAGGED user row. It
+ // stays untagged on purpose: the imminent `turn-start` tags it as
+ // this new turn's initiator (exactly like a normal send), and
+ // reconcile then folds it into the sealed turn. Leaving N separate
+ // untagged rows would strand all but the most-recent one (turn-start
+ // only tags one), so collapsing is required.
+ if (mcEvent.reason === "continuation") {
+ updateLive(tabId, (msgs) => {
+ const consumedTexts: string[] = [];
+ const rest: ChatMessage[] = [];
+ let firstConsumedIdx = -1;
+ for (const m of msgs) {
+ if (m.role === "user" && m.id.startsWith("queued-")) {
+ const queuedId = m.id.slice(7);
+ if (mcEvent.messageIds.includes(queuedId)) {
+ if (firstConsumedIdx === -1) firstConsumedIdx = rest.length;
+ const textChunk = m.chunks.find((c) => c.type === "text");
+ consumedTexts.push(textChunk && textChunk.type === "text" ? textChunk.text : "");
+ continue;
+ }
+ }
+ rest.push(m);
+ }
+ if (consumedTexts.length === 0) return msgs;
+ const initiator: ChatMessage = {
+ id: generateId(),
+ role: "user",
+ chunks: [{ type: "text", text: consumedTexts.join("\n---\n") }],
+ };
+ rest.splice(firstConsumedIdx === -1 ? rest.length : firstConsumedIdx, 0, initiator);
+ return rest;
+ });
+ break;
+ }
+
// Split the current assistant message: finalize it, then insert
// the consumed user messages after it. Subsequent streaming events
// will create a NEW assistant message block below.
diff --git a/packages/frontend/src/lib/types.ts b/packages/frontend/src/lib/types.ts
index bede2cc..285b4d2 100644
--- a/packages/frontend/src/lib/types.ts
+++ b/packages/frontend/src/lib/types.ts
@@ -198,7 +198,21 @@ export type AgentEvent =
agentModels?: Array<{ key_id: string; model_id: string }> | null;
}
| { type: "message-queued"; tabId: string; messageId: string; message: string }
- | { type: "message-consumed"; tabId: string; messageIds: string[] }
+ | {
+ type: "message-consumed";
+ tabId: string;
+ messageIds: string[];
+ /**
+ * Why the queue was drained:
+ * - "interrupt": consumed mid-turn, folded into a running turn's tool
+ * result as a [USER INTERRUPT]. The optimistic bubble collapses into
+ * that sealed turn.
+ * - "continuation": consumed between turns to START a new turn. The
+ * optimistic bubble becomes that new turn's initiating user row.
+ * Absent ⇒ treat as "interrupt" (back-compat).
+ */
+ reason?: "interrupt" | "continuation";
+ }
| { type: "message-cancelled"; tabId: string; messageId: string };
export interface TaskItem {
diff --git a/packages/frontend/tests/chat-store.test.ts b/packages/frontend/tests/chat-store.test.ts
index b9d37f2..ea718eb 100644
--- a/packages/frontend/tests/chat-store.test.ts
+++ b/packages/frontend/tests/chat-store.test.ts
@@ -1460,6 +1460,88 @@ describe("tabStore — chunk-native eviction / pagination / reconcile", () => {
expect(tab?.chunks.map((c) => c.seq)).toEqual([0, 1]);
});
+ it("a continuation-consumed queued message becomes the next turn's initiator", async () => {
+ // The turn-end fix: a message queued during turn-a is drained AFTER the
+ // turn ends (reason: "continuation") to START turn-b. Its optimistic
+ // `queued-` bubble must collapse into a single UNTAGGED user row so the
+ // imminent turn-b `turn-start` tags it as that turn's initiator — and it
+ // then folds cleanly into turn-b's sealed chunks (no linger, no dup).
+ const sealedB = [
+ chunkRow("ua", "cc", 0, "turn-a", "user", "text", { text: "first" }),
+ chunkRow("aa", "cc", 1, "turn-a", "assistant", "text", { text: "first answer" }),
+ chunkRow("ub", "cc", 2, "turn-b", "user", "text", { text: "next please" }),
+ chunkRow("ab", "cc", 3, "turn-b", "assistant", "text", { text: "second answer" }),
+ ];
+ vi.stubGlobal(
+ "fetch",
+ vi.fn((url: string) => {
+ if (url.split("?")[0]?.endsWith("/tabs/cc/chunks"))
+ return Promise.resolve(chunksResponse(sealedB, 4));
+ return Promise.reject(new Error(`unexpected ${url}`));
+ }),
+ );
+ const store = createTabStore();
+ store.handleEvent({
+ type: "tab-created",
+ id: "cc",
+ title: "CC",
+ keyId: null,
+ modelId: null,
+ parentTabId: null,
+ });
+ // Turn A streams; user queues a follow-up while it runs.
+ store.handleEvent({ type: "turn-start", turnId: "turn-a", tabId: "cc" });
+ store.handleEvent({ type: "text-delta", delta: "first answer", tabId: "cc" });
+ store.handleEvent({
+ type: "message-queued",
+ tabId: "cc",
+ messageId: "q1",
+ message: "next please",
+ });
+ let tab = store.tabs.find((t) => t.id === "cc");
+ expect(tab?.live.some((m) => m.id === "queued-q1")).toBe(true);
+
+ // Turn A ends. The backend drains the queue as a CONTINUATION (not an
+ // interrupt) and emits message-consumed{reason:"continuation"}.
+ store.handleEvent({
+ type: "message-consumed",
+ tabId: "cc",
+ messageIds: ["q1"],
+ reason: "continuation",
+ });
+ tab = store.tabs.find((t) => t.id === "cc");
+ // The queued- bubble collapsed into ONE plain (untagged, un-prefixed) user row.
+ expect(tab?.live.some((m) => m.id === "queued-q1")).toBe(false);
+ const initiator = tab?.live.find((m) => m.role === "user");
+ expect(initiator).toBeTruthy();
+ expect(initiator?.id.startsWith("queued-")).toBe(false);
+ expect(initiator?.turnId).toBeUndefined();
+ expect(tab?.queuedMessages.some((m) => m.id === "q1")).toBe(false);
+
+ // turn-a seals first (it was the running turn when the queue drained).
+ store.handleEvent({ type: "turn-sealed", turnId: "turn-a", tabId: "cc" });
+ // Now turn-b starts — it must tag the collapsed initiator row.
+ store.handleEvent({ type: "turn-start", turnId: "turn-b", tabId: "cc" });
+ tab = store.tabs.find((t) => t.id === "cc");
+ const taggedInitiator = tab?.live.find((m) => m.role === "user" && m.turnId === "turn-b");
+ expect(taggedInitiator).toBeTruthy();
+
+ store.handleEvent({ type: "text-delta", delta: "second answer", tabId: "cc" });
+ store.handleEvent({ type: "turn-sealed", turnId: "turn-b", tabId: "cc" });
+ await tick();
+ tab = store.tabs.find((t) => t.id === "cc");
+ // Both turns are durable; the live tail is empty (initiator folded into
+ // turn-b, no lingering/duplicated user bubble).
+ expect(tab?.chunks.map((c) => c.seq)).toEqual([0, 1, 2, 3]);
+ expect(tab?.live.length).toBe(0);
+ expect(tab?.renderGroups.map((m) => m.role)).toEqual([
+ "user",
+ "assistant",
+ "user",
+ "assistant",
+ ]);
+ });
+
it("preserves a concurrent newer turn when an earlier deferred reconcile flushes", async () => {
const sealedA = [
chunkRow("ua", "c", 0, "turn-a", "user", "text", { text: "A?" }),