fix(sessions): dedupe cloud completion notifications (#3572)

Co-authored-by: posthog[bot] <206114724+posthog[bot]@users.noreply.github.com>
Co-authored-by: richardsolomou <2622273+richardsolomou@users.noreply.github.com>

Author: richardsolomou

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

@@ -206,11 +206,14 @@ describe("cloud task update notifications", () => {
   });
 
   // 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.
+  // the resulting notify count. Snapshots never ring; each armed turn rings at
+  // most once even if the producer writes duplicate completion entries.
   it.each([
+    {
+      label: "a completion event without an armed turn",
+      updates: [logsUpdate([turnComplete()], 1)],
+      expected: 0,
+    },
     {
       label: "a live turn that starts and completes",
       updates: [logsUpdate([sessionPrompt(1), turnComplete()], 2)],
@@ -224,6 +227,14 @@ describe("cloud task update notifications", () => {
       ],
       expected: 2,
     },
+    {
+      label: "duplicate completion events for one turn",
+      updates: [
+        logsUpdate([sessionPrompt(1), turnComplete()], 2),
+        logsUpdate([turnComplete()], 3),
+      ],
+      expected: 1,
+    },
     {
       // Opening a task mid-turn: its session/prompt is already in history and
       // only the turn_complete arrives live. The completion must still ring.

packages/core/src/sessions/sessionService.ts

@@ -1947,6 +1947,9 @@ export class SessionService {
         // above. Cloud sessions never see that response.
         const session = this.getSessionByRunId(taskRunId);
         if (session?.isCloud) {
+          const completedActiveTurn =
+            session.currentPromptId !== null &&
+            session.currentPromptId !== undefined;
           const turnStartedAtTs =
             this.liveTurnContent.get(taskRunId)?.startedAtTs ??
             session.promptStartedAt;
@@ -1957,7 +1960,7 @@ export class SessionService {
           });
           if (isLive) {
             // Queued messages will start a new turn — suppress the "done" notification in that case.
-            if (session.messageQueue.length === 0) {
+            if (completedActiveTurn && session.messageQueue.length === 0) {
               this.d.notifyPromptComplete(
                 session.taskTitle,
                 "end_turn",