diff --git a/apps/web/src/components/ThreadNotificationCoordinator.badge.browser.test.ts b/apps/web/src/components/ThreadNotificationCoordinator.badge.browser.test.ts index f253dd6be22c7..0d04fcc76085d 100644 --- a/apps/web/src/components/ThreadNotificationCoordinator.badge.browser.test.ts +++ b/apps/web/src/components/ThreadNotificationCoordinator.badge.browser.test.ts @@ -245,3 +245,73 @@ it("holds a completion through a continuation gap and notifies once it settles", await vi.advanceTimersByTimeAsync(1); expect(TestNotification.sent.map((sent) => sent.title)).toEqual(["Thread completed"]); }); + +it.each(["hasPendingApprovals", "hasPendingUserInput"] as const)( + "notifies for offline completion and %s once across reconnects", + async (flag) => { + await using app = setup(); + const asking = { id: "asking", hasPendingApprovals: true }; + const offline = { id: "offline", [flag]: true }; + const done = { + id: "done", + latestTurn: { + turnId: "done-turn", + state: "completed", + completedAt: "2026-09-13T07:00:00.000Z", + }, + }; + state.shells.set("one", shell({ id: "working" }, { id: "asking" }, { id: "offline" }, done)); + app.start(); + using _timers = fakeTimers(); + state.shells.set("one", shell({ id: "working" }, asking, { id: "offline" }, done)); + state.shells.set("one", { + ...shell({ id: "working" }, asking, { id: "offline" }, done), + status: "cached", + }); + + const completed = { + id: "working", + latestTurn: { turnId: "turn", state: "completed", completedAt: "2026-09-13T07:05:00.000Z" }, + }; + state.shells.set("one", { + ...shell(completed, asking, offline, done), + status: "synchronizing", + }); + state.shells.set("one", shell(completed, asking, offline, done)); + + expect(TestNotification.sent.map(({ title }) => title)).toEqual([ + "Approval needed", + flag === "hasPendingApprovals" ? "Approval needed" : "Input needed", + ]); + await vi.advanceTimersByTimeAsync(WORKING_EXIT_GRACE_MS); + state.shells.set("one", { ...shell(completed, asking, offline, done), status: "cached" }); + state.shells.set("one", shell(completed, asking, offline, done)); + await vi.advanceTimersByTimeAsync(WORKING_EXIT_GRACE_MS); + expect(TestNotification.sent.map(({ title }) => title)).toEqual([ + "Approval needed", + flag === "hasPendingApprovals" ? "Approval needed" : "Input needed", + "Thread completed", + ]); + }, +); + +it.each([false, true])( + "rechecks a pending completion after reconnect (continued: %s)", + async (continued) => { + await using app = setup(); + app.start(); + using _timers = fakeTimers(); + const completed = shell({ + latestTurn: { turnId: "turn", state: "completed", completedAt: "2026-09-13T07:05:00.000Z" }, + }); + state.shells.set("one", completed); + state.shells.set("one", { ...completed, status: "cached" }); + await vi.advanceTimersByTimeAsync(WORKING_EXIT_GRACE_MS); + expect(TestNotification.sent).toEqual([]); + state.shells.set("one", continued ? shell() : completed); + await vi.advanceTimersByTimeAsync(WORKING_EXIT_GRACE_MS); + expect(TestNotification.sent.map(({ title }) => title)).toEqual( + continued ? [] : ["Thread completed"], + ); + }, +); diff --git a/apps/web/src/components/ThreadNotificationCoordinator.logic.ts b/apps/web/src/components/ThreadNotificationCoordinator.logic.ts index 77edccfbe7741..fdbfb7ae1f49d 100644 --- a/apps/web/src/components/ThreadNotificationCoordinator.logic.ts +++ b/apps/web/src/components/ThreadNotificationCoordinator.logic.ts @@ -69,6 +69,9 @@ export function createThreadNotificationGate( for (const key of handled.keys()) if (!keys.has(key)) handled.delete(key); for (const key of lastSettled.keys()) if (!keys.has(key)) lastSettled.delete(key); }, + cancelPending(): void { + for (const key of pending.keys()) cancel(key); + }, clear(): void { for (const key of pending.keys()) cancel(key); handled.clear(); diff --git a/apps/web/src/components/ThreadNotificationCoordinator.tsx b/apps/web/src/components/ThreadNotificationCoordinator.tsx index a782e24da6522..5e75d42cd1385 100644 --- a/apps/web/src/components/ThreadNotificationCoordinator.tsx +++ b/apps/web/src/components/ThreadNotificationCoordinator.tsx @@ -46,6 +46,7 @@ interface NotificationState extends ThreadNotificationObservation { interface WatchedEnvironment { unsubscribe: () => void; previous: Map; + reconnecting: boolean; readonly gate: ReturnType>; } @@ -179,7 +180,11 @@ export function startThreadNotifications( if (rawThread.lineage.relationshipToParent === "subagent") continue; const prior = previous.get(rawThread.id); const subagentRequest = subagentRequests.get(rawThread.id) ?? null; - if (prior?.raw === rawThread && prior.subagentRequest === subagentRequest) { + if ( + !entry.reconnecting && + prior?.raw === rawThread && + prior.subagentRequest === subagentRequest + ) { next.set(rawThread.id, prior); continue; } @@ -208,6 +213,7 @@ export function startThreadNotifications( next.set(thread.id, state); changed.push([prior, state]); } + entry.reconnecting = false; entry.previous = next; entry.gate.retain(new Set(next.keys())); for (const [prior, state] of changed) { @@ -221,6 +227,7 @@ export function startThreadNotifications( const watchEnvironment = (environmentId: EnvironmentId): WatchedEnvironment => { const entry: WatchedEnvironment = { previous: new Map(), + reconnecting: false, unsubscribe: () => {}, gate: createThreadNotificationGate((threadId, kind) => { const state = entry.previous.get(threadId); @@ -238,8 +245,8 @@ export function startThreadNotifications( observeThreads(environmentId, entry, threads); return; } - entry.gate.clear(); - entry.previous = new Map(); + entry.gate.cancelPending(); + entry.reconnecting = true; }, { immediate: true }, );