fix(web): add turn-diff-completed fallback for turn-end notifications

The prior widened session-set check still missed turn-end notifications in
practice. The session-set signal is apparently unreliable in the user's
environment — possibly lost to subscription races, snapshot overwrites, or
missed batches.

Add thread.turn-diff-completed as a fallback completion signal. This event
is dispatched by CheckpointReactor strictly in response to an actual
turn.completed runtime event, so its presence is a strong guarantee that a
turn ended. It is already in the thread-detail subscription stream and
flows through the same applyRecoveredEventBatch path.

Deduplication: a completion fired via the primary session-set path in the
same batch marks the thread in completionFiredThreadIds, preventing the
turn-diff-completed branch from firing a second notification for the same
turn. Both paths honor the 5-second user-stop suppression window and the
in-batch userInitiated guard.

Tests cover: fallback firing when session-set is absent, dedup when both
signals arrive together, and suppression respect on the fallback path.

Also renames the in-batch running-transition tracker to make its intent
clearer in the guarded-completion branch.

Author: tyulyukov

apps/web/src/turnNotification.test.ts

@@ -55,6 +55,16 @@ function makeActivityAppendedEvent(threadId: string, kind: string): Orchestratio
   } as unknown as OrchestrationEvent;
 }
 
+function makeTurnDiffCompletedEvent(threadId: string): OrchestrationEvent {
+  return {
+    type: "thread.turn-diff-completed",
+    payload: {
+      threadId: threadId as ThreadId,
+      turnId: `turn-${threadId}`,
+    },
+  } as unknown as OrchestrationEvent;
+}
+
 describe("BUILT_IN_SOUNDS", () => {
   it("has 4 entries with valid id, label, and src", () => {
     expect(BUILT_IN_SOUNDS).toHaveLength(4);
@@ -255,6 +265,72 @@ describe("deriveTurnNotificationTriggers", () => {
 
     expect(triggers).toHaveLength(0);
   });
+
+  it("falls back to turn-diff-completed when session-set signal is missing", () => {
+    // The session-set path can be lost (subscription race, snapshot overwrite)
+    // but the CheckpointReactor reliably dispatches thread.turn-diff-completed
+    // after an actual turn.completed runtime event, so the user still gets
+    // notified.
+    const thread = makeThread({
+      session: { orchestrationStatus: "ready" },
+    } as Partial<Thread>);
+    const project = makeProject();
+    const events = [makeTurnDiffCompletedEvent("thread-1")];
+
+    const triggers = deriveTurnNotificationTriggers(
+      events,
+      () => thread,
+      () => project,
+    );
+
+    expect(triggers).toHaveLength(1);
+    expect(triggers[0]!.reason).toBe("turn-completed");
+  });
+
+  it("dedupes session-set + turn-diff-completed for the same thread", () => {
+    // Both the primary session-set signal and the turn-diff-completed fallback
+    // arrive together in the happy path. Only one notification should fire.
+    const thread = makeThread({
+      session: { orchestrationStatus: "running" },
+    } as Partial<Thread>);
+    const project = makeProject();
+    const events = [
+      makeSessionSetEvent("thread-1", "ready"),
+      makeTurnDiffCompletedEvent("thread-1"),
+    ];
+
+    const triggers = deriveTurnNotificationTriggers(
+      events,
+      () => thread,
+      () => project,
+    );
+
+    expect(triggers).toHaveLength(1);
+    expect(triggers[0]!.reason).toBe("turn-completed");
+  });
+
+  it("honors user-initiated stops on the turn-diff-completed fallback", () => {
+    // If the user stopped the turn themselves, the fallback must respect the
+    // 5-second suppression window so we don't notify for a stop they just did.
+    const thread = makeThread({
+      session: { orchestrationStatus: "running" },
+    } as Partial<Thread>);
+    const project = makeProject();
+
+    vi.spyOn(Date, "now").mockReturnValue(1000);
+    markThreadUserStopped("thread-1" as ThreadId);
+
+    vi.spyOn(Date, "now").mockReturnValue(2000);
+    const events = [makeTurnDiffCompletedEvent("thread-1")];
+
+    const triggers = deriveTurnNotificationTriggers(
+      events,
+      () => thread,
+      () => project,
+    );
+
+    expect(triggers).toHaveLength(0);
+  });
 });
 
 describe("locally interrupted turn tracker", () => {

apps/web/src/turnNotification.ts

@@ -116,10 +116,13 @@ export function deriveTurnNotificationTriggers(
     }
   }
 
-  // Track threads observed transitioning to "running" earlier in this batch so
-  // that a subsequent completion event still fires a notification even if the
-  // stored session hasn't been updated yet (e.g. on reconnect/replay when
-  // turn.started and turn.completed arrive in the same batch).
+  // Threads that already produced a completion trigger in this batch. Used to
+  // dedupe between the primary session-set signal and the turn-diff-completed
+  // fallback so we never fire two notifications for the same turn end.
+  const completionFiredThreadIds = new Set<ThreadId>();
+  // Threads observed transitioning to "running" earlier in this batch, used
+  // when reconnect/replay delivers turn.started + turn.completed together and
+  // the stored session still reflects the pre-batch state.
   const threadTransitionedToRunning = new Set<ThreadId>();
 
   for (const event of events) {
@@ -140,19 +143,18 @@ export function deriveTurnNotificationTriggers(
 
       const thread = getThread(threadId);
       if (!thread) continue;
-      // The thread is considered to have been actively turning if the stored
-      // session is still "running", the current batch already transitioned it
-      // into "running", or the latest turn is still marked as running. The
-      // activeTurnId check catches the common case where completion notifications
-      // would otherwise be dropped because the session orchestrationStatus has
-      // already been written ahead of this derivation (e.g. when the server
-      // replays both start + completion events as a single batch).
-      const wasRunning =
+
+      // Skip session-bootstrap completions (e.g. provider "session.started"
+      // maps to status:ready when no turn is active). A real turn completion
+      // always has one of: a session previously in running state, an
+      // activeTurnId on the prior session, a latestTurn still flagged
+      // running, or an in-batch running transition.
+      const turnWasActive =
         thread.session?.orchestrationStatus === "running" ||
         thread.session?.activeTurnId !== undefined ||
         thread.latestTurn?.state === "running" ||
         threadTransitionedToRunning.has(threadId);
-      if (!wasRunning) continue;
+      if (!turnWasActive) continue;
 
       const project = getProject(thread.projectId);
       triggers.push({
@@ -161,6 +163,30 @@ export function deriveTurnNotificationTriggers(
         threadTitle: thread.title || "Untitled",
         projectName: project?.name || "Unknown project",
       });
+      completionFiredThreadIds.add(threadId);
+      continue;
+    }
+
+    // Fallback signal: a turn-diff capture completing is a strong guarantee
+    // that a turn ended — CheckpointReactor only fires this after an actual
+    // turn.completed runtime event. If the session-set path was dropped
+    // (subscription race, snapshot overwrite, missed batch), this still
+    // notifies the user and is deduped against the primary path above.
+    if (event.type === "thread.turn-diff-completed") {
+      const { threadId } = event.payload;
+      if (completionFiredThreadIds.has(threadId)) continue;
+      if (isThreadSuppressed(threadId)) continue;
+      if (userInitiatedThreadIds.has(threadId)) continue;
+      const thread = getThread(threadId);
+      if (!thread) continue;
+      const project = getProject(thread.projectId);
+      triggers.push({
+        threadId,
+        reason: "turn-completed",
+        threadTitle: thread.title || "Untitled",
+        projectName: project?.name || "Unknown project",
+      });
+      completionFiredThreadIds.add(threadId);
       continue;
     }