fix(sessions): fire cloud completion notification once per turn

packages/core/src/cloud-task/cloud-task.test.ts

@@ -283,6 +283,71 @@ describe("CloudTaskService", () => {
     );
   });
 
+  it("drops a re-delivered log entry with a duplicate stream id", async () => {
+    const updates: unknown[] = [];
+    service.on(CloudTaskEvent.Update, (payload) => updates.push(payload));
+
+    mockNetFetch
+      .mockResolvedValueOnce(
+        createJsonResponse({
+          id: "run-1",
+          status: "in_progress",
+          stage: null,
+          output: null,
+          error_message: null,
+          branch: "main",
+          updated_at: "2026-01-01T00:00:00Z",
+        }),
+      )
+      .mockResolvedValueOnce(
+        createJsonResponse([], 200, { "X-Has-More": "false" }),
+      );
+
+    const consoleEntry = (message: string) =>
+      `data: {"type":"notification","timestamp":"2026-01-01T00:00:01Z","notification":{"jsonrpc":"2.0","method":"_posthog/console","params":{"sessionId":"run-1","level":"info","message":"${message}"}}}`;
+
+    // The durable stream re-sends the tail by id on reconnect/replay. Here id 1
+    // arrives twice; the second copy must be dropped so the entry is delivered
+    // — and counted — exactly once (the fix for duplicate transcript entries and
+    // back-to-back completion notifications).
+    mockStreamFetch.mockResolvedValueOnce(
+      createOpenSseResponse(
+        `id: 1\n${consoleEntry("first")}\n\nid: 1\n${consoleEntry("first")}\n\nid: 2\n${consoleEntry("second")}\n\n`,
+      ),
+    );
+
+    service.watch({
+      taskId: "task-1",
+      runId: "run-1",
+      apiHost: "https://app.example.com",
+      teamId: 2,
+    });
+
+    await waitFor(() =>
+      updates.some(
+        (u) =>
+          (u as { kind?: string; totalEntryCount?: number }).kind === "logs" &&
+          ((u as { totalEntryCount?: number }).totalEntryCount ?? 0) >= 2,
+      ),
+    );
+
+    const messages = updates
+      .filter((u) => (u as { kind?: string }).kind === "logs")
+      .flatMap(
+        (u) =>
+          (u as { newEntries: Array<{ notification?: { params?: unknown } }> })
+            .newEntries,
+      )
+      .map(
+        (entry) =>
+          (entry.notification?.params as { message?: string } | undefined)
+            ?.message,
+      );
+
+    // id 1 delivered once (not twice), id 2 delivered once, order preserved.
+    expect(messages).toEqual(["first", "second"]);
+  });
+
   it("reconnects with Last-Event-ID after a stream error", async () => {
     vi.useFakeTimers();
 

packages/core/src/cloud-task/cloud-task.ts

@@ -119,6 +119,11 @@ interface WatcherState {
   // Leg that issued lastEventId, and the leg of the connection currently being read.
   lastEventIdLeg: StreamLeg | null;
   streamLeg: StreamLeg | null;
+  // Ids of log entries already ingested on the current leg. The durable stream
+  // re-sends the tail by id on reconnect/replay, so dropping a seen id here is
+  // what stops a re-delivered entry (e.g. a `turn_complete`) from being counted
+  // and emitted twice. Cleared on a leg switch, where the id space changes.
+  seenEventIds: Set<string>;
   lastStatus: TaskRunStatus | null;
   lastStage: string | null;
   lastOutput: Record<string, unknown> | null;
@@ -538,6 +543,7 @@ export class CloudTaskService extends TypedEventEmitter<CloudTaskEvents> {
       lastEventId: null,
       lastEventIdLeg: null,
       streamLeg: null,
+      seenEventIds: new Set(),
       lastStatus: null,
       lastStage: null,
       lastOutput: null,
@@ -790,6 +796,9 @@ export class CloudTaskService extends TypedEventEmitter<CloudTaskEvents> {
       });
       watcher.lastEventId = null;
       watcher.lastEventIdLeg = null;
+      // Proxy and Django ids are unrelated, so a retained id could false-match a
+      // different entry on the new leg. Drop them; the snapshot covers the gap.
+      watcher.seenEventIds.clear();
     }
     watcher.streamLeg = leg;
 
@@ -1136,6 +1145,20 @@ export class CloudTaskService extends TypedEventEmitter<CloudTaskEvents> {
       return null;
     }
 
+    // Drop a re-delivered log entry by its stream id. The durable stream
+    // re-sends the tail on reconnect/replay, and each resend would otherwise be
+    // counted as a new entry (advancing totalEntryCount past the renderer's
+    // processedLineCount guard) and emitted again — the root cause of duplicate
+    // transcript entries and back-to-back completion notifications. Entries
+    // without an id (legacy servers) fall through and are handled downstream.
+    const eventId = event.id;
+    if (eventId !== undefined) {
+      if (watcher.seenEventIds.has(eventId)) {
+        return null;
+      }
+      watcher.seenEventIds.add(eventId);
+    }
+
     watcher.pendingLogEntries.push(event.data as StoredLogEntry);
     if (watcher.pendingLogEntries.length >= EVENT_BATCH_MAX_SIZE) {
       this.flushLogBatch(key);

packages/core/src/sessions/cloudTaskUpdateNotifications.test.ts

@@ -16,6 +16,20 @@ function turnComplete(): StoredLogEntry {
   };
 }
 
+// The `session/prompt` request that opens a turn. Its arrival is what arms the
+// turn's single completion notification, so realistic sequences pair it with a
+// later `turn_complete`.
+function sessionPrompt(id: number): StoredLogEntry {
+  return {
+    type: "notification",
+    notification: {
+      id,
+      method: "session/prompt",
+      params: { sessionId: RUN_ID, prompt: [] },
+    },
+  };
+}
+
 function permissionRequest(
   requestId: string,
   toolCallId: string,
@@ -33,6 +47,32 @@ function permissionRequest(
   };
 }
 
+function logsUpdate(
+  newEntries: StoredLogEntry[],
+  totalEntryCount: number,
+): CloudTaskUpdatePayload {
+  return {
+    taskId: TASK_ID,
+    runId: RUN_ID,
+    kind: "logs",
+    newEntries,
+    totalEntryCount,
+  };
+}
+
+function snapshotUpdate(
+  newEntries: StoredLogEntry[],
+  totalEntryCount: number,
+): CloudTaskUpdatePayload {
+  return {
+    taskId: TASK_ID,
+    runId: RUN_ID,
+    kind: "snapshot",
+    newEntries,
+    totalEntryCount,
+  };
+}
+
 function createHarness() {
   const sessions: Record<string, AgentSession> = {};
   const store = {
@@ -155,25 +195,48 @@ describe("cloud task update notifications", () => {
     expect(harness.markActivity).not.toHaveBeenCalled();
   });
 
-  it("notifies once for a live turn_complete delta after the snapshot", () => {
-    const harness = createHarness();
-    harness.sendUpdate({
-      taskId: TASK_ID,
-      runId: RUN_ID,
-      kind: "snapshot",
-      newEntries: [turnComplete(), turnComplete()],
-      totalEntryCount: 2,
-    });
+  // Each case applies a sequence of updates to a fresh harness; `expected` is
+  // the resulting notify count. Snapshots never ring; each live turn_complete
+  // rings once. Re-delivered stream entries are dropped upstream in
+  // CloudTaskService by their event id (see cloud-task.test.ts), so a replay
+  // never reaches this layer.
+  it.each([
+    {
+      label: "a live turn that starts and completes",
+      updates: [logsUpdate([sessionPrompt(1), turnComplete()], 2)],
+      expected: 1,
+    },
+    {
+      label: "several turns each completing",
+      updates: [
+        logsUpdate([sessionPrompt(1), turnComplete()], 2),
+        logsUpdate([sessionPrompt(2), turnComplete()], 4),
+      ],
+      expected: 2,
+    },
+    {
+      // Opening a task mid-turn: its session/prompt is already in history and
+      // only the turn_complete arrives live. The completion must still ring.
+      label: "a prompt seen only in the snapshot, completing live",
+      updates: [
+        snapshotUpdate([sessionPrompt(1)], 1),
+        logsUpdate([turnComplete()], 2),
+      ],
+      expected: 1,
+    },
+  ])(
+    "fires the completion notification once per turn: $label",
+    ({ updates, expected }) => {
+      const harness = createHarness();
+      for (const update of updates) harness.sendUpdate(update);
+      expect(harness.notifyPromptComplete).toHaveBeenCalledTimes(expected);
+    },
+  );
 
-    harness.sendUpdate({
-      taskId: TASK_ID,
-      runId: RUN_ID,
-      kind: "logs",
-      newEntries: [turnComplete()],
-      totalEntryCount: 3,
-    });
+  it("notifies with the task title and stop reason, and marks activity", () => {
+    const harness = createHarness();
+    harness.sendUpdate(logsUpdate([sessionPrompt(1), turnComplete()], 2));
 
-    expect(harness.notifyPromptComplete).toHaveBeenCalledTimes(1);
     expect(harness.notifyPromptComplete).toHaveBeenCalledWith(
       "Cloud Task",
       "end_turn",