fix: deduplicate live and committed task updates

2026-10-26 10:13:36 +09:00
parent a1df3742b1
commit d12bc2ea0b
3 changed files with 188 additions and 3 deletions
@@ -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<SessionSnapshotEntry, { kind: "tool_call" }> {
+  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<typeof taskCallEntry>): 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",
@@ -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<string>;
  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":
@@ -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() });