From d12bc2ea0b6aba44e517aee6d57ee84cba92236c Mon Sep 17 00:00:00 2001 From: Hare Date: Mon, 26 Oct 2026 10:13:36 +0900 Subject: [PATCH] fix: deduplicate live and committed task updates --- .../src/lib/workspace/console/model.test.ts | 122 ++++++++++++++++++ .../src/lib/workspace/console/model.ts | 20 ++- .../console/console-display.browser.spec.ts | 49 +++++++ 3 files changed, 188 insertions(+), 3 deletions(-) diff --git a/web/workspace/src/lib/workspace/console/model.test.ts b/web/workspace/src/lib/workspace/console/model.test.ts index 11cfe60ad..e93362c58 100644 --- a/web/workspace/src/lib/workspace/console/model.test.ts +++ b/web/workspace/src/lib/workspace/console/model.test.ts @@ -3025,6 +3025,128 @@ Deno.test("projectConsole mirrors live TaskCreate and TaskUpdate calls", () => { }]); }); +function taskCallEntry( + callId: string, + name = "TaskCreate", + args: object = { subject: "Inspect tasks", description: "Task detail" }, +): Extract { + return { + kind: "tool_call", entry_id: `entry:${callId}`, timestamp: 1, + provenance: "model_output", derived_from: [], + call_id: callId, name, arguments: JSON.stringify(args), + }; +} + +function taskLiveEvent(entry: ReturnType): Event { + return { event: "tool_call_done", data: { + id: entry.call_id, name: entry.name, arguments: entry.arguments, + } }; +} + +function taskCommitEvent(entry: SessionSnapshotEntry): Event { + return { event: "session_entry_committed", data: { entry } }; +} + +Deno.test("Tasks apply each call once across live, committed and replay events", () => { + const first = taskCallEntry("first"); + const second = taskCallEntry("second"); // Identical arguments, distinct call. + for (const committedFirst of [false, true]) { + const projector = createConsoleProjector(); + for (const entry of [first, second]) { + const events = [taskLiveEvent(entry), taskCommitEvent(entry)]; + if (committedFirst) events.reverse(); + const initial = projector.append([{ eventId: `${entry.call_id}:first`, event: events[0] }]); + const expected = initial.tasks; + const reconciled = projector.append([ + { eventId: `${entry.call_id}:second`, event: events[1] }, + { eventId: `${entry.call_id}:replay-live`, event: taskLiveEvent(entry) }, + { eventId: `${entry.call_id}:replay-commit`, event: taskCommitEvent(entry) }, + ]); + assertEquals(reconciled.tasks, expected); + assertEquals(initial.tasks, expected); + } + const result = projector.snapshot(); + assertEquals(result.tasks.map((task) => task.taskid), [1, 2]); + assertEquals(result.taskNextId, 3); + assertEquals(result.lines.filter((line) => line.kind === "tool").length, 2); + } +}); + +Deno.test("Tasks snapshot and segment rebuild retain call deduplication and reset old state", () => { + const entry = taskCallEntry("retained"); + const discarded = taskCallEntry("discarded"); + const snapshot = snapshotEvent("/repo") as SnapshotEvent; + snapshot.data.session.entries = [entry]; + for (const event of [snapshot, { event: "segment_rotated", data: { session: snapshot.data.session } } satisfies Event]) { + const projector = createConsoleProjector(); + projector.append([ + { eventId: "old", event: taskLiveEvent(discarded) }, + { eventId: "live", event: taskLiveEvent(entry) }, + { eventId: "commit", event: taskCommitEvent(entry) }, + ]); + const rebuilt = projector.append([{ eventId: "rebuild", event }]); + const replayed = projector.append([ + { eventId: "replayed-live", event: taskLiveEvent(entry) }, + { eventId: "replayed-commit", event: taskCommitEvent(entry) }, + ]); + assertEquals(replayed.tasks, rebuilt.tasks); + assertEquals(replayed.tasks.map((task) => task.taskid), [1]); + assertEquals(replayed.taskNextId, 2); + const fresh = projector.append([{ eventId: "new", event: taskLiveEvent(discarded) }]); + assertEquals(fresh.tasks.map((task) => task.taskid), [1, 2]); + } +}); + +Deno.test("Tasks late call replay cannot undo updates or resurrect deleted tasks", () => { + const create = taskCallEntry("create"); + const start = taskCallEntry("start", "TaskUpdate", { taskid: 1, status: "inprogress" }); + const complete = taskCallEntry("complete", "TaskUpdate", { taskid: 1, status: "completed" }); + const remove = taskCallEntry("remove", "TaskUpdate", { taskid: 1, status: "deleted" }); + const projector = createConsoleProjector(); + for (const entry of [create, start, complete]) { + projector.append([{ eventId: entry.call_id, event: taskLiveEvent(entry) }]); + } + const completed = projector.append([{ eventId: "late-start", event: taskCommitEvent(start) }]); + assertEquals(completed.tasks[0].status, "completed"); + const deleted = projector.append([ + { eventId: "remove", event: taskLiveEvent(remove) }, + { eventId: "late-create", event: taskCommitEvent(create) }, + { eventId: "late-complete", event: taskCommitEvent(complete) }, + ]); + assertEquals(deleted.tasks, []); + assertEquals(deleted.taskNextId, 2); +}); + +Deno.test("Tasks deduplication does not suppress committed tool result corrections", () => { + const entry = taskCallEntry("create"); + const result: SessionSnapshotEntry = { + kind: "tool_result", entry_id: "result", timestamp: 2, + provenance: "tool_output", derived_from: [], call_id: entry.call_id, + summary: "Created task", content: "original result", is_error: false, attachments: [], + }; + const projection = projectConsole([ + { eventId: "live", event: taskLiveEvent(entry) }, + { eventId: "commit", event: taskCommitEvent(entry) }, + { eventId: "result", event: taskCommitEvent(result) }, + { eventId: "correction", event: taskCommitEvent({ ...result, content: "corrected result" }) }, + ]); + assertEquals(projection.tasks.length, 1); + assertEquals(projection.lines[0].toolCall?.output, "corrected result"); +}); + +Deno.test("Tasks call identities are scoped to each Worker projection", () => { + const entry = taskCallEntry("shared-call-id"); + const worker = { session_id: "child", name: "child", parent_session_id: "parent", kind: "sub_worker" as const }; + const projection = projectConsole([ + { eventId: "parent-live", event: taskLiveEvent(entry) }, + { eventId: "child-live", event: { event: "internal_worker", data: { worker, event: taskLiveEvent(entry) } } }, + { eventId: "parent-commit", event: taskCommitEvent(entry) }, + { eventId: "child-commit", event: { event: "internal_worker", data: { worker, event: taskCommitEvent(entry) } } }, + ]); + assertEquals(projection.tasks.length, 1); + assertEquals(projection.internalWorkers[0].console.tasks, projection.tasks); +}); + Deno.test("Internal Worker output stays separate and follows parent stream arrival order", () => { const worker = { session_id: "child-session", diff --git a/web/workspace/src/lib/workspace/console/model.ts b/web/workspace/src/lib/workspace/console/model.ts index 195d14bec..168a5481d 100644 --- a/web/workspace/src/lib/workspace/console/model.ts +++ b/web/workspace/src/lib/workspace/console/model.ts @@ -201,6 +201,8 @@ export type ConsoleProjection = { lines: ConsoleLine[]; tasks: ConsoleTask[]; taskNextId: number; + /** Task calls arrive both live and committed; apply each call only once. */ + appliedTaskCallIds: ReadonlySet; status: string | null; workerState: WorkerStateSnapshot | null; compaction: InFlightCompaction | null; @@ -308,6 +310,7 @@ export function emptyConsoleProjection(): ConsoleProjection { lines: [], tasks: [], taskNextId: 1, + appliedTaskCallIds: new Set(), status: null, workerState: null, compaction: null, @@ -1153,6 +1156,7 @@ export function applyProtocolEvent( lines: [...projection.lines], tasks: [...projection.tasks], taskNextId: projection.taskNextId, + appliedTaskCallIds: projection.appliedTaskCallIds, status: projection.status, workerState: projection.workerState, compaction: projection.compaction, @@ -1279,7 +1283,7 @@ export function applyProtocolEvent( argsStream: event.data.arguments, state: "running", }); - applyTaskTool(next, event.data.name, event.data.arguments); + applyTaskTool(next, event.data.id, event.data.name, event.data.arguments); break; case "tool_result": attachToolResult(next, envelope.eventId, event.data.id, { @@ -1326,6 +1330,7 @@ export function applyProtocolEvent( next.lines = snapshot.lines; next.tasks = snapshot.tasks; next.taskNextId = snapshot.taskNextId; + next.appliedTaskCallIds = snapshot.appliedTaskCallIds; appendSnapshotInFlightLines( next, event.data.in_flight?.blocks ?? [], @@ -1416,6 +1421,7 @@ export function applyProtocolEvent( next.lines = [...segment.lines, ...retainedErrors, ...retainedCompaction]; next.tasks = segment.tasks; next.taskNextId = segment.taskNextId; + next.appliedTaskCallIds = segment.appliedTaskCallIds; next.workerMetadata = metadataWithoutContext(next.workerMetadata); break; } @@ -2409,9 +2415,15 @@ function usageText( function applyTaskTool( projection: ConsoleProjection, + callId: string, name: string, argumentsJson: string, ): void { + if (name !== "TaskCreate" && name !== "TaskUpdate") return; + if (projection.appliedTaskCallIds.has(callId)) return; + // Fence task mutations, not tool-row reconciliation or corrected results. + // Copy on write so previously returned projections remain unchanged. + projection.appliedTaskCallIds = new Set(projection.appliedTaskCallIds).add(callId); const state = applyTaskToolCall( { tasks: projection.tasks, nextTaskId: projection.taskNextId }, name, @@ -2448,6 +2460,7 @@ function snapshotProjectionFromSession( lines: [], tasks: [], taskNextId: 1, + appliedTaskCallIds: new Set(), status: null, workerState: null, compaction: null, @@ -2715,10 +2728,11 @@ function applyLoggedItem( case "tool_call": { const name = stringField(item, "name") ?? "Tool"; const argumentsJson = stringField(item, "arguments") ?? ""; + const callId = stringField(item, "call_id") ?? eventId; upsertToolCall( projection, eventId, - stringField(item, "call_id") ?? eventId, + callId, { name, arguments: argumentsJson, @@ -2726,7 +2740,7 @@ function applyLoggedItem( state: "running", }, ); - applyTaskTool(projection, name, argumentsJson); + applyTaskTool(projection, callId, name, argumentsJson); break; } case "tool_result": diff --git a/web/workspace/src/routes/w/[workspaceId]/workers/[workerRef]/console/console-display.browser.spec.ts b/web/workspace/src/routes/w/[workspaceId]/workers/[workerRef]/console/console-display.browser.spec.ts index 207df74bd..32073e3e5 100644 --- a/web/workspace/src/routes/w/[workspaceId]/workers/[workerRef]/console/console-display.browser.spec.ts +++ b/web/workspace/src/routes/w/[workspaceId]/workers/[workerRef]/console/console-display.browser.spec.ts @@ -1015,6 +1015,55 @@ test("mini task summary opens and closes details without a header Tasks button", expect(summary.getAttribute("aria-expanded")).toBe("false"); }); +test.each([false, true])("live and committed TaskCreate render one task (committed first: %s)", async (committedFirst) => { + vi.stubGlobal("fetch", vi.fn(async () => Response.json({ availability: "live_protocol" }))); + render(ConsolePage, { data: pageData() }); + await waitFor(() => expect(multiplexer.subscribe).toHaveBeenCalledOnce()); + latestListener().onFrame(subscribedFrame(emptySession())); + await screen.findByText("No conversation to display"); + const entry = { + kind: "tool_call" as const, entry_id: "task-create", timestamp: 1, + provenance: "model_output" as const, derived_from: [], + call_id: "task-call", name: "TaskCreate", + arguments: JSON.stringify({ subject: "Inspect live tasks", description: "One task per call" }), + }; + const live: ProtocolEvent = { event: "tool_call_done", data: { + id: entry.call_id, name: entry.name, arguments: entry.arguments, + } }; + const committed: ProtocolEvent = { event: "session_entry_committed", data: { entry } }; + protocolEvent({ event: "tool_call_start", data: { id: entry.call_id, name: entry.name } }); + protocolEvent(committedFirst ? committed : live); + const summary = await screen.findByRole("button", { name: /1 task — pending/ }); + await fireEvent.click(summary); + const pane = await screen.findByRole("complementary", { name: "Worker tasks" }); + expect(within(pane).getAllByRole("listitem")).toHaveLength(1); + protocolEvent(committedFirst ? live : committed); + protocolEvent(live); + protocolEvent(committed); + await settleMicrotasks(); + expect(within(pane).getAllByRole("listitem")).toHaveLength(1); + expect(within(pane).getByText("Inspect live tasks")).toBeTruthy(); + expect(screen.getByRole("button", { name: /1 task — pending/ })).toBeTruthy(); + + // Reconnecting from the committed snapshot must produce exactly the same list. + const session = emptySession(); + session.entries.push(entry); + latestListener().onFrame(subscribedFrame(session)); + protocolEvent(committed); + await settleMicrotasks(); + const restoredPane = screen.getByRole("complementary", { name: "Worker tasks" }); + expect(within(restoredPane).getAllByRole("listitem")).toHaveLength(1); + expect(within(restoredPane).getByText("#1")).toBeTruthy(); + + // Identical text in a different call is a genuinely separate task. + protocolEvent({ event: "tool_call_done", data: { ...live.data, id: "second-task-call" } }); + protocolEvent({ event: "session_entry_committed", data: { + entry: { ...entry, entry_id: "second-task-entry", call_id: "second-task-call" }, + } }); + await waitFor(() => expect(within(restoredPane).getAllByRole("listitem")).toHaveLength(2)); + expect(within(restoredPane).getByText("#2")).toBeTruthy(); +}); + test("latest jump is icon-only, hidden at the bottom, and resumes following streamed output", async () => { vi.stubGlobal("fetch", vi.fn(async () => Response.json({ availability: "live_protocol" }))); const view = render(ConsolePage, { data: pageData() });