From 3eb6c2d50b5eab824ded39a430c8ad76a761b89b Mon Sep 17 00:00:00 2001 From: Omar <76748259+omar9848492@users.noreply.github.com> Date: Tue, 20 Oct 2026 15:02:49 -0400 Subject: [PATCH 1/2] fix: give Working shelf a two-second exit grace --- apps/mobile/src/features/home/HomeScreen.tsx | 4 +- .../threads/ThreadNavigationSidebar.tsx | 4 +- .../src/features/threads/threadListV2.test.ts | 31 +++- .../src/features/threads/threadListV2.ts | 3 +- apps/web/src/components/Sidebar.tsx | 17 +- .../src/state/threadInbox.test.ts | 150 +++++++++++++++++- .../client-runtime/src/state/threadInbox.ts | 133 +++++++++++++--- 7 files changed, 313 insertions(+), 29 deletions(-) diff --git a/apps/mobile/src/features/home/HomeScreen.tsx b/apps/mobile/src/features/home/HomeScreen.tsx index f20a8fee636ad..457069d67499c 100644 --- a/apps/mobile/src/features/home/HomeScreen.tsx +++ b/apps/mobile/src/features/home/HomeScreen.tsx @@ -17,7 +17,7 @@ import { import { type EnvironmentId, type SidebarProjectGroupingMode } from "@t3tools/contracts"; import { useAtomValue } from "@effect/atom-react"; import { useFocusEffect } from "@react-navigation/native"; -import { use, useEffect, useRef, useState, type ComponentProps } from "react"; +import { use, useEffect, useRef, useState, useSyncExternalStore, type ComponentProps } from "react"; import { Platform, ScrollView, @@ -488,6 +488,7 @@ export function HomeScreen(props: HomeScreenProps) { ...(workingShelfEnabled ? [] : sectionAvailability("active")), ]); })(); + useSyncExternalStore(threadListInboxReturns.subscribe, threadListInboxReturns.getSnapshot); const threadListV2Layout = (() => { threadListInboxReturns.observe(workingShelfEnabled ? props.threads : null); return buildThreadListV2Items({ @@ -505,6 +506,7 @@ export function HomeScreen(props: HomeScreenProps) { workingShelfEnabled, workingShelfExpanded, inboxReturnAt: threadListInboxReturns.returnedAt, + isWorking: threadListInboxReturns.isWorking, snoozedShelfExpanded, settledShelfExpanded, selectedThreadKey, diff --git a/apps/mobile/src/features/threads/ThreadNavigationSidebar.tsx b/apps/mobile/src/features/threads/ThreadNavigationSidebar.tsx index 13ce478929d25..69a436262acf8 100644 --- a/apps/mobile/src/features/threads/ThreadNavigationSidebar.tsx +++ b/apps/mobile/src/features/threads/ThreadNavigationSidebar.tsx @@ -16,7 +16,7 @@ import { AnimatedLegendList } from "@legendapp/list/reanimated"; import { THREAD_SIDEBAR_ROW_LAYOUT } from "./threadListAnimations"; import type { MenuAction } from "@react-native-menu/menu"; import { useAtomValue } from "@effect/atom-react"; -import { useEffect, useMemo, useRef, useState } from "react"; +import { useEffect, useMemo, useRef, useState, useSyncExternalStore } from "react"; import type { LayoutChangeEvent, TextInputInstance } from "react-native"; import { Platform, StyleSheet, TextInput, View } from "react-native"; import { GestureDetector, useNativeGesture } from "react-native-gesture-handler"; @@ -320,6 +320,7 @@ function ThreadNavigationSidebarPane( ...(workingShelfEnabled ? [] : sectionAvailability("active")), ]); })(); + useSyncExternalStore(threadListInboxReturns.subscribe, threadListInboxReturns.getSnapshot); const threadListV2Layout = (() => { threadListInboxReturns.observe(workingShelfEnabled ? threads : null); return buildThreadListV2Items({ @@ -337,6 +338,7 @@ function ThreadNavigationSidebarPane( workingShelfEnabled, workingShelfExpanded, inboxReturnAt: threadListInboxReturns.returnedAt, + isWorking: threadListInboxReturns.isWorking, snoozedShelfExpanded, settledShelfExpanded, selectedThreadKey: props.selectedThreadKey ?? null, diff --git a/apps/mobile/src/features/threads/threadListV2.test.ts b/apps/mobile/src/features/threads/threadListV2.test.ts index 2c150561fde6c..2899daa35b5b7 100644 --- a/apps/mobile/src/features/threads/threadListV2.test.ts +++ b/apps/mobile/src/features/threads/threadListV2.test.ts @@ -1,4 +1,5 @@ import { threadSearchMatchKey } from "@t3tools/client-runtime/state/thread-search"; +import { createInboxReturnTracker } from "@t3tools/client-runtime/state/thread-inbox"; import { presentThreadShell } from "@t3tools/client-runtime/state/models"; import * as DateTime from "effect/DateTime"; import * as Schema from "effect/Schema"; @@ -24,7 +25,7 @@ import { RunId, ThreadId, } from "@t3tools/contracts"; -import { describe, expect, it } from "vite-plus/test"; +import { describe, expect, it, vi } from "vite-plus/test"; import type { PendingNewTask } from "../../state/use-pending-new-tasks"; import { makeRawThreadShell, makeThreadShellFixture } from "../../test-fixtures"; @@ -1549,6 +1550,34 @@ describe("Working section beta", () => { expect(off.workingCount).toBe(0); }); + it("keeps an inactive row in Working during the shared grace and returns it on expiry", () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date(NOW)); + const tracker = createInboxReturnTracker(); + const working = makeThread({ id: ThreadId.make("grace"), title: "grace", runtime: running }); + const inactive = { ...working, runtime: null }; + const render = () => + build({ + threads: [inactive], + workingShelfExpanded: true, + isWorking: tracker.isWorking, + inboxReturnAt: tracker.returnedAt, + }); + try { + tracker.observe([working]); + tracker.observe([inactive]); + expect(render().workingCount).toBe(1); + expect(render().items[0]?.thread.runtime).toBeNull(); + vi.advanceTimersByTime(2_000); + expect(render().workingCount).toBe(0); + expect(ids(render())).toEqual(["grace"]); + expect(tracker.returnedAt(inactive)).toBe(Date.parse(NOW) + 2_000); + } finally { + tracker.observe(null); + vi.useRealTimers(); + } + }); + it("orders the inbox by the latest return this device observed", () => { const layout = build({ inboxReturnAt: (thread) => diff --git a/apps/mobile/src/features/threads/threadListV2.ts b/apps/mobile/src/features/threads/threadListV2.ts index fe0683498d301..52acd1799b165 100644 --- a/apps/mobile/src/features/threads/threadListV2.ts +++ b/apps/mobile/src/features/threads/threadListV2.ts @@ -589,6 +589,7 @@ export function buildThreadListV2Items(input: { readonly now: string; readonly workingShelfEnabled?: boolean; readonly workingShelfExpanded?: boolean; + readonly isWorking?: (thread: EnvironmentThreadShell) => boolean; readonly inboxReturnAt?: (thread: EnvironmentThreadShell) => number | undefined; readonly snoozedShelfExpanded?: boolean; readonly settledShelfExpanded?: boolean; @@ -671,7 +672,7 @@ export function buildThreadListV2Items(input: { settled.push(thread); } else if (thread.pinnedAt != null) { pinned.push(thread); - } else if (workingShelfEnabled && isThreadWorking(thread)) { + } else if (workingShelfEnabled && (input.isWorking ?? isThreadWorking)(thread)) { working.push(thread); } else { active.push(thread); diff --git a/apps/web/src/components/Sidebar.tsx b/apps/web/src/components/Sidebar.tsx index 827f62ee61198..f059463d13597 100644 --- a/apps/web/src/components/Sidebar.tsx +++ b/apps/web/src/components/Sidebar.tsx @@ -28,7 +28,10 @@ import { effectiveSnoozed, threadWokeAt, } from "@t3tools/client-runtime/state/thread-settled"; -import { createInboxReturnTracker } from "@t3tools/client-runtime/state/thread-inbox"; +import { + createInboxReturnTracker, + isThreadWorking, +} from "@t3tools/client-runtime/state/thread-inbox"; import { sortSettledThreads } from "@t3tools/client-runtime/state/thread-sort"; import { nextShownOrder, @@ -61,6 +64,7 @@ import { useReducer, useRef, useState, + useSyncExternalStore, type KeyboardEvent as ReactKeyboardEvent, type MouseEvent as ReactMouseEvent, type ReactNode, @@ -151,7 +155,6 @@ import { filterSidebarProjectScopeItems, firstValidTimestampMs, isSidebarNestedLinkClick, - isSidebarThreadWorking, isTrailingDoubleClick, orderItemsByPreferredIds, planSidebarThreadDrop, @@ -383,6 +386,7 @@ function useSidebarThreads(input: { ), ); const workingShelfEnabled = useClientSettings((s) => s.sidebarWorkingShelfEnabled); + useSyncExternalStore(inboxReturns.subscribe, inboxReturns.getSnapshot); const nowMinute = useNowMinute(); // Snooze wake times are second-precise, so classifying with the quantized // minute would hold a woken thread on the shelf for up to a minute. @@ -520,8 +524,11 @@ function partitionSidebarThreads(input: SidebarThreadPartitionInput) { const pinned: EnvironmentThreadShell[] = []; const active: EnvironmentThreadShell[] = []; const working: EnvironmentThreadShell[] = []; - const inbox = (thread: EnvironmentThreadShell) => - workingShelfEnabled && isSidebarThreadWorking(thread) ? working : active; + const inbox = (thread: EnvironmentThreadShell, manuallyMoved = false) => + workingShelfEnabled && + (manuallyMoved ? isThreadWorking(thread) : inboxReturns.isWorking(thread)) + ? working + : active; const snoozed: EnvironmentThreadShell[] = []; const settled: EnvironmentThreadShell[] = []; const draggable = new Set(); @@ -546,7 +553,7 @@ function partitionSidebarThreads(input: SidebarThreadPartitionInput) { ? pinned : optimisticDrop.section === "settled" ? settled - : inbox(projected) + : inbox(projected, true) ).push( optimisticDrop.clearsSnooze ? projected diff --git a/packages/client-runtime/src/state/threadInbox.test.ts b/packages/client-runtime/src/state/threadInbox.test.ts index 86516983a2fe4..0c4b7a81b93d8 100644 --- a/packages/client-runtime/src/state/threadInbox.test.ts +++ b/packages/client-runtime/src/state/threadInbox.test.ts @@ -1,7 +1,12 @@ import { EnvironmentId, ProviderInstanceId, RunId, ThreadId } from "@t3tools/contracts"; -import { describe, expect, it } from "vite-plus/test"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vite-plus/test"; -import { createInboxReturnTracker, sortWorkingThreadsBySend } from "./threadInbox.ts"; +import { + createInboxReturnTracker, + isThreadWorking, + sortInboxThreadsByReturn, + sortWorkingThreadsBySend, +} from "./threadInbox.ts"; const environmentId = EnvironmentId.make("environment-1"); @@ -30,6 +35,14 @@ function thread(id: string, working: boolean) { } describe("createInboxReturnTracker", () => { + beforeEach(() => { + vi.useFakeTimers(); + vi.setSystemTime(10_000); + }); + afterEach(() => { + vi.clearAllTimers(); + vi.useRealTimers(); + }); it("stamps a thread when it stops working, but never on the first observation", () => { const tracker = createInboxReturnTracker(); tracker.observe([thread("a", true), thread("b", false)]); @@ -37,9 +50,142 @@ describe("createInboxReturnTracker", () => { expect(tracker.returnedAt(thread("b", false))).toBeUndefined(); tracker.observe([thread("a", false), thread("b", false)]); + vi.advanceTimersByTime(2_000); expect(tracker.returnedAt(thread("a", false))).toBeDefined(); expect(tracker.returnedAt(thread("b", false))).toBeUndefined(); }); + + it("holds only placement, then expires and notifies without another shell event", () => { + const tracker = createInboxReturnTracker(); + const notify = vi.fn(); + const unsubscribe = tracker.subscribe(notify); + tracker.observe([thread("a", true)]); + tracker.observe([thread("a", false)]); + expect(isThreadWorking(thread("a", false))).toBe(false); + expect(tracker.isWorking(thread("a", false))).toBe(true); + vi.advanceTimersByTime(1_999); + tracker.observe([thread("a", false)]); + expect(tracker.isWorking(thread("a", false))).toBe(true); + expect(tracker.returnedAt(thread("a", false))).toBeUndefined(); + vi.advanceTimersByTime(1); + expect(tracker.isWorking(thread("a", false))).toBe(false); + expect(tracker.returnedAt(thread("a", false))).toBe(12_000); + expect(notify).toHaveBeenCalledTimes(1); + unsubscribe(); + }); + + it("cancels a transient return on resume and grants a fresh grace after the next finish", () => { + const tracker = createInboxReturnTracker(); + tracker.observe([thread("a", false)]); + expect(tracker.isWorking(thread("a", false))).toBe(false); + tracker.observe([thread("a", true)]); + expect(tracker.isWorking(thread("a", true))).toBe(true); + tracker.observe([thread("a", false)]); + vi.advanceTimersByTime(1_500); + tracker.observe([thread("a", true)]); + vi.advanceTimersByTime(1_000); + expect(tracker.returnedAt(thread("a", true))).toBeUndefined(); + tracker.observe([thread("a", false)]); + vi.advanceTimersByTime(1_999); + expect(tracker.isWorking(thread("a", false))).toBe(true); + vi.advanceTimersByTime(1); + expect(tracker.returnedAt(thread("a", false))).toBe(14_500); + }); + + it.each([ + { hasPendingApprovals: true }, + { hasPendingUserInput: true }, + { pendingSubagentRequest: "approval" as const }, + { pendingSubagentRequest: "user_input" as const }, + { interactionMode: "plan" as const, hasActionableProposedPlan: true }, + { runtime: { ...thread("a", true).runtime!, status: "failed" as const, lastError: "failed" } }, + { pinnedAt: "2026-06-01T00:00:01Z" }, + { settledOverride: "settled" as const }, + { snoozedUntil: "2026-06-02T00:00:00Z" }, + { activeOrderKey: "manual-order" }, + ])("ends a grace immediately for attention or manual placement: %j", (change) => { + const tracker = createInboxReturnTracker(); + tracker.observe([thread("a", true)]); + tracker.observe([thread("a", false)]); + tracker.observe([{ ...thread("a", false), ...change }]); + expect(tracker.isWorking(thread("a", false))).toBe(false); + expect(tracker.returnedAt(thread("a", false))).toBe(10_000); + expect(vi.getTimerCount()).toBe(0); + }); + + it("grants grace after a retained expired snooze", () => { + const tracker = createInboxReturnTracker(); + const retained = { + snoozedAt: "1970-01-01T00:00:01.000Z", + snoozedUntil: "1970-01-01T00:00:05.000Z", + latestRun: { + runId: RunId.make("completed-wake"), + status: "completed" as const, + requestedAt: "1970-01-01T00:00:02.000Z", + startedAt: "1970-01-01T00:00:02.000Z", + completedAt: "1970-01-01T00:00:09.000Z", + assistantMessageId: null, + }, + }; + const active = { ...thread("a", true), ...retained }; + const finished = { ...thread("a", false), ...retained }; + tracker.observe([active]); + tracker.observe([finished]); + expect(tracker.isWorking(finished)).toBe(true); + expect(tracker.returnedAt(finished)).toBeUndefined(); + vi.advanceTimersByTime(1_999); + expect(tracker.isWorking(finished)).toBe(true); + vi.advanceTimersByTime(1); + expect(tracker.isWorking(finished)).toBe(false); + expect(tracker.returnedAt(finished)).toBe(12_000); + }); + + it("does not grant grace to a future snooze after background completion", () => { + const tracker = createInboxReturnTracker(); + const retained = { + snoozedAt: "1970-01-01T00:00:01.000Z", + snoozedUntil: "1970-01-01T00:00:30.000Z", + }; + tracker.observe([{ ...thread("a", true), ...retained }]); + const finished = { ...thread("a", false), ...retained }; + tracker.observe([finished]); + expect(tracker.isWorking(finished)).toBe(false); + expect(vi.getTimerCount()).toBe(0); + }); + + it("stamps only stable returns and sorts them in their actual return order", () => { + const tracker = createInboxReturnTracker(); + const a = { ...thread("a", false), createdAt: "1970-01-01T00:00:00Z" }; + const b = { ...thread("b", false), createdAt: "1970-01-01T00:00:00Z" }; + tracker.observe([thread("a", true), thread("b", true)]); + tracker.observe([a, thread("b", true)]); + vi.advanceTimersByTime(500); + tracker.observe([a, b]); + vi.advanceTimersByTime(1_500); + expect(tracker.returnedAt(a)).toBe(12_000); + expect(tracker.returnedAt(b)).toBeUndefined(); + vi.advanceTimersByTime(500); + expect(sortInboxThreadsByReturn([a, b], tracker.returnedAt).map((entry) => entry.id)).toEqual([ + "b", + "a", + ]); + }); + + it("cancels pending timers for removed threads and disabled shelves", () => { + const tracker = createInboxReturnTracker(); + const notify = vi.fn(); + tracker.subscribe(notify); + tracker.observe([thread("a", true)]); + tracker.observe([thread("a", false)]); + tracker.observe([]); + expect(vi.getTimerCount()).toBe(0); + tracker.observe([thread("a", true)]); + tracker.observe([thread("a", false)]); + tracker.observe(null); + vi.advanceTimersByTime(2_000); + expect(notify).not.toHaveBeenCalled(); + expect(tracker.returnedAt(thread("a", false))).toBeUndefined(); + }); }); describe("sortWorkingThreadsBySend", () => { diff --git a/packages/client-runtime/src/state/threadInbox.ts b/packages/client-runtime/src/state/threadInbox.ts index 880e6ebf31a08..896095feb0aae 100644 --- a/packages/client-runtime/src/state/threadInbox.ts +++ b/packages/client-runtime/src/state/threadInbox.ts @@ -1,5 +1,7 @@ +// @effect-diagnostics globalTimers:off - Shelf placement timers run in web and native clients, outside an Effect runtime. import { threadRuntimeIsActive, type EnvironmentThreadShell } from "./models.ts"; import { toSortableTimestamp } from "./threadSort.ts"; +import { effectiveSnoozed } from "./threadSettled.ts"; type WorkingThreadInput = Pick< EnvironmentThreadShell, @@ -86,40 +88,135 @@ function sortNewestFirst + >; + +/** Keep at module scope so shelf placement and return order survive route changes. */ export function createInboxReturnTracker() { const keyOf = (thread: Pick) => `${thread.environmentId}:${thread.id}`; - let lastWorkingKeys: ReadonlySet | null = null; + const states = new Map< + string, + { working: boolean; deadline: number | null; placement: string } + >(); const returns = new Map(); + const listeners = new Set<() => void>(); + let timer: ReturnType | undefined; + let version = 0; + const now = () => { + // @effect-diagnostics-next-line globalDate:off + return Date.now(); + }; + const schedule = () => { + clearTimeout(timer); + timer = undefined; + let next = Infinity; + for (const state of states.values()) { + if (state.deadline !== null) next = Math.min(next, state.deadline); + } + if (next === Infinity) return; + timer = setTimeout( + () => { + const at = now(); + for (const [key, state] of states) { + if (state.deadline !== null && state.deadline <= at) { + state.working = false; + state.deadline = null; + returns.set(key, at); + } + } + schedule(); + version++; + for (const listener of listeners) listener(); + }, + Math.max(0, next - now()), + ); + }; return { - observe(threads: ReadonlyArray | null): void { + subscribe(listener: () => void) { + listeners.add(listener); + return () => { + listeners.delete(listener); + }; + }, + getSnapshot: () => version, + observe(threads: readonly TrackedInboxThread[] | null): void { if (threads === null) { - lastWorkingKeys = null; + clearTimeout(timer); + timer = undefined; + states.clear(); returns.clear(); return; } - const working = new Set(); const present = new Set(); + const at = now(); + // @effect-diagnostics-next-line globalDate:off + const observedAt = new Date(at).toISOString(); for (const thread of threads) { const key = keyOf(thread); present.add(key); - if (isThreadWorking(thread)) working.add(key); - } - for (const key of returns.keys()) { - if (!present.has(key)) returns.delete(key); + const placement = JSON.stringify([ + thread.archivedAt, + thread.deletedAt, + thread.pinnedAt, + thread.settledOverride, + thread.settledAt, + thread.snoozedAt, + thread.snoozedUntil, + thread.activeOrderKey, + ]); + const previous = states.get(key); + const actionable = + thread.hasPendingApprovals || + thread.hasPendingUserInput || + thread.pendingSubagentRequest != null || + (thread.interactionMode === "plan" && thread.hasActionableProposedPlan) || + thread.runtime?.lastError != null || + (thread.latestRun != null && + ["failed", "cancelled", "interrupted", "rolled_back"].includes( + thread.latestRun.status, + )); + const moved = previous !== undefined && previous.placement !== placement; + const excluded = + thread.archivedAt != null || + thread.deletedAt != null || + thread.pinnedAt != null || + thread.settledOverride === "settled" || + effectiveSnoozed(thread, { now: observedAt }); + const actual = isThreadWorking(thread); + // Only shelf placement is held: execution state and attention requests stay authoritative. + const deadline = + !actual && previous?.working && !actionable && !moved && !excluded + ? (previous.deadline ?? at + 2_000) + : null; + const working = actual || (deadline !== null && deadline > at); + if (previous?.working && !working) returns.set(key, at); + states.set(key, { working, deadline: working ? deadline : null, placement }); } - // @effect-diagnostics-next-line globalDate:off - const at = Date.now(); - for (const key of lastWorkingKeys ?? []) { - if (present.has(key) && !working.has(key)) returns.set(key, at); + for (const key of states.keys()) { + if (!present.has(key)) { + states.delete(key); + returns.delete(key); + } } - lastWorkingKeys = working; + schedule(); }, + isWorking: ( + thread: WorkingThreadInput & Pick, + ) => states.get(keyOf(thread))?.working ?? isThreadWorking(thread), returnedAt: (thread: Pick) => returns.get(keyOf(thread)), }; From fda1e76ac0c66bbaa242622cfa429abf728a9a99 Mon Sep 17 00:00:00 2001 From: Omar <76748259+omar9848492@users.noreply.github.com> Date: Sat, 24 Oct 2026 12:09:45 -0400 Subject: [PATCH 2/2] fix(clients): hold completion notifications through brief continuation gaps When an orchestrator run ends and its continuation starts a moment later, the notification coordinator sent "Thread completed" right away, often after every intermediate run. A completion now waits for the Working shelf's exit grace (WORKING_EXIT_GRACE_MS, now exported from the inbox tracker module) and fires only if the thread is still settled when it expires. Resuming, archiving, a stop request, removal from the snapshot, an environment going away or unmount cancels it. Attention notifications stay immediate and replace a pending completion for good. Settings, focus and the active route are read when the notification is presented, not when it is scheduled. A completion pending when the environment shell leaves live is dropped. React Compiler memoizes the web sidebar partition, so a thread whose grace expired could stay in Working until another input changed. The inbox tracker's expiry version is now part of the partition clockKey. Mobile passes the same version into the thread list layout build so its rebuild does not depend on the compiler leaving that block unmemoized. Refs #24477 Co-Authored-By: Claude Opus 5.5 --- apps/mobile/src/features/home/HomeScreen.tsx | 6 +- .../threads/ThreadNavigationSidebar.tsx | 6 +- .../src/features/threads/threadListV2.ts | 3 + apps/web/src/components/Sidebar.tsx | 12 +- ...ficationCoordinator.badge.browser.test.tsx | 26 +- ...hreadNotificationCoordinator.logic.test.ts | 193 ++++++++++++++ .../ThreadNotificationCoordinator.logic.ts | 78 ++++++ .../ThreadNotificationCoordinator.tsx | 241 ++++++++++-------- .../client-runtime/src/state/threadInbox.ts | 9 +- 9 files changed, 457 insertions(+), 117 deletions(-) create mode 100644 apps/web/src/components/ThreadNotificationCoordinator.logic.test.ts create mode 100644 apps/web/src/components/ThreadNotificationCoordinator.logic.ts diff --git a/apps/mobile/src/features/home/HomeScreen.tsx b/apps/mobile/src/features/home/HomeScreen.tsx index 457069d67499c..dbb467a128438 100644 --- a/apps/mobile/src/features/home/HomeScreen.tsx +++ b/apps/mobile/src/features/home/HomeScreen.tsx @@ -488,7 +488,10 @@ export function HomeScreen(props: HomeScreenProps) { ...(workingShelfEnabled ? [] : sectionAvailability("active")), ]); })(); - useSyncExternalStore(threadListInboxReturns.subscribe, threadListInboxReturns.getSnapshot); + const inboxReturnsVersion = useSyncExternalStore( + threadListInboxReturns.subscribe, + threadListInboxReturns.getSnapshot, + ); const threadListV2Layout = (() => { threadListInboxReturns.observe(workingShelfEnabled ? props.threads : null); return buildThreadListV2Items({ @@ -507,6 +510,7 @@ export function HomeScreen(props: HomeScreenProps) { workingShelfExpanded, inboxReturnAt: threadListInboxReturns.returnedAt, isWorking: threadListInboxReturns.isWorking, + inboxReturnsVersion, snoozedShelfExpanded, settledShelfExpanded, selectedThreadKey, diff --git a/apps/mobile/src/features/threads/ThreadNavigationSidebar.tsx b/apps/mobile/src/features/threads/ThreadNavigationSidebar.tsx index 69a436262acf8..ebefcf8e47974 100644 --- a/apps/mobile/src/features/threads/ThreadNavigationSidebar.tsx +++ b/apps/mobile/src/features/threads/ThreadNavigationSidebar.tsx @@ -320,7 +320,10 @@ function ThreadNavigationSidebarPane( ...(workingShelfEnabled ? [] : sectionAvailability("active")), ]); })(); - useSyncExternalStore(threadListInboxReturns.subscribe, threadListInboxReturns.getSnapshot); + const inboxReturnsVersion = useSyncExternalStore( + threadListInboxReturns.subscribe, + threadListInboxReturns.getSnapshot, + ); const threadListV2Layout = (() => { threadListInboxReturns.observe(workingShelfEnabled ? threads : null); return buildThreadListV2Items({ @@ -339,6 +342,7 @@ function ThreadNavigationSidebarPane( workingShelfExpanded, inboxReturnAt: threadListInboxReturns.returnedAt, isWorking: threadListInboxReturns.isWorking, + inboxReturnsVersion, snoozedShelfExpanded, settledShelfExpanded, selectedThreadKey: props.selectedThreadKey ?? null, diff --git a/apps/mobile/src/features/threads/threadListV2.ts b/apps/mobile/src/features/threads/threadListV2.ts index 52acd1799b165..10f2186494805 100644 --- a/apps/mobile/src/features/threads/threadListV2.ts +++ b/apps/mobile/src/features/threads/threadListV2.ts @@ -591,6 +591,9 @@ export function buildThreadListV2Items(input: { readonly workingShelfExpanded?: boolean; readonly isWorking?: (thread: EnvironmentThreadShell) => boolean; readonly inboxReturnAt?: (thread: EnvironmentThreadShell) => number | undefined; + // isWorking and inboxReturnAt change when the exit grace expires, so the + // tracker's version is an input: a new version rebuilds the layout. + readonly inboxReturnsVersion?: number; readonly snoozedShelfExpanded?: boolean; readonly settledShelfExpanded?: boolean; readonly selectedThreadKey?: string | null; diff --git a/apps/web/src/components/Sidebar.tsx b/apps/web/src/components/Sidebar.tsx index f059463d13597..a70d5d3e1ef6e 100644 --- a/apps/web/src/components/Sidebar.tsx +++ b/apps/web/src/components/Sidebar.tsx @@ -306,8 +306,9 @@ interface SidebarThreadPartitionInput { readonly serverConfigs: ReadonlyMap; readonly optimisticDrop: SidebarOptimisticDrop | null; readonly workingShelfEnabled: boolean; - // Snooze wakes and the queued-turn grace depend on the clock, so the minute - // tick and the precise snooze-wake tick are inputs: a new key repartitions. + // Snooze wakes, the queued-turn grace and the Working shelf's exit grace depend + // on the clock, so the minute tick, the precise snooze-wake tick and the inbox + // tracker's expiry version are inputs: a new key repartitions. readonly clockKey: string; } @@ -386,7 +387,10 @@ function useSidebarThreads(input: { ), ); const workingShelfEnabled = useClientSettings((s) => s.sidebarWorkingShelfEnabled); - useSyncExternalStore(inboxReturns.subscribe, inboxReturns.getSnapshot); + const inboxReturnsVersion = useSyncExternalStore( + inboxReturns.subscribe, + inboxReturns.getSnapshot, + ); const nowMinute = useNowMinute(); // Snooze wake times are second-precise, so classifying with the quantized // minute would hold a woken thread on the shelf for up to a minute. @@ -428,7 +432,7 @@ function useSidebarThreads(input: { serverConfigs, optimisticDrop, workingShelfEnabled, - clockKey: `${nowMinute}:${snoozeWakeTick}`, + clockKey: `${nowMinute}:${snoozeWakeTick}:${inboxReturnsVersion}`, }); const partitionNextWakeAt = partition.snoozedThreads[0]?.snoozedUntil ?? null; if (partitionNextWakeAt !== nextSnoozeWakeAt) setNextSnoozeWakeAt(partitionNextWakeAt); diff --git a/apps/web/src/components/ThreadNotificationCoordinator.badge.browser.test.tsx b/apps/web/src/components/ThreadNotificationCoordinator.badge.browser.test.tsx index dd537f2a8d0b4..8b0804ce8037d 100644 --- a/apps/web/src/components/ThreadNotificationCoordinator.badge.browser.test.tsx +++ b/apps/web/src/components/ThreadNotificationCoordinator.badge.browser.test.tsx @@ -1,11 +1,12 @@ import { RegistryContext } from "@effect/atom-react"; +import { WORKING_EXIT_GRACE_MS } from "@t3tools/client-runtime/state/thread-inbox"; import { EnvironmentId } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; import * as Option from "effect/Option"; import { Atom, AtomRegistry } from "effect/reactivity"; import { expect, it, vi } from "vite-plus/test"; -import { render, stubGlobal, type Screen } from "../../test/render"; +import { fakeTimers, render, stubGlobal, type Screen } from "../../test/render"; const state = vi.hoisted(() => ({ mode: "notifications", @@ -215,3 +216,26 @@ it.each(["hasPendingApprovals", "hasPendingUserInput"] as const)( expect(notification.close).toHaveBeenCalled(); }, ); + +it("holds a completion through a continuation gap and notifies once it settles", async () => { + await using app = setup(); + await app.render(); + using _timers = fakeTimers(); + const finished = (turnId: string, completedAt: string) => ({ + latestTurn: { turnId, state: "completed", completedAt }, + }); + state.shells.set("one", shell(finished("turn-1", "2026-09-13T07:00:01.000Z"))); + await app.render(); + state.shells.set( + "one", + shell({ latestTurn: { turnId: "turn-2", state: "running", completedAt: null } }), + ); + await app.render(); + state.shells.set("one", shell(finished("turn-2", "2026-09-13T07:00:02.000Z"))); + await app.render(); + expect(TestNotification.sent).toEqual([]); + await vi.advanceTimersByTimeAsync(WORKING_EXIT_GRACE_MS - 1); + expect(TestNotification.sent).toEqual([]); + await vi.advanceTimersByTimeAsync(1); + expect(TestNotification.sent.map((sent) => sent.title)).toEqual(["Thread completed"]); +}); diff --git a/apps/web/src/components/ThreadNotificationCoordinator.logic.test.ts b/apps/web/src/components/ThreadNotificationCoordinator.logic.test.ts new file mode 100644 index 0000000000000..3c293497fde46 --- /dev/null +++ b/apps/web/src/components/ThreadNotificationCoordinator.logic.test.ts @@ -0,0 +1,193 @@ +import { WORKING_EXIT_GRACE_MS } from "@t3tools/client-runtime/state/thread-inbox"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vite-plus/test"; + +import { + createThreadNotificationGate, + type ThreadNotificationObservation, +} from "./ThreadNotificationCoordinator.logic"; + +const working: ThreadNotificationObservation = { + attention: null, + settled: false, + completedAt: null, +}; +const completed = (at: number): ThreadNotificationObservation => ({ + attention: null, + settled: true, + completedAt: at, +}); +const visible = { archived: false, stopRequested: false }; + +function harness() { + const notify = vi.fn(); + const gate = createThreadNotificationGate(notify); + let prior: ThreadNotificationObservation | undefined; + const observe = (next: ThreadNotificationObservation, options = visible) => { + gate.observe("thread", prior, next, options); + prior = next; + }; + return { gate, notify, observe }; +} + +describe("createThreadNotificationGate", () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + afterEach(() => { + vi.useRealTimers(); + }); + + it("drops a completion when the thread resumes within the grace", () => { + const { notify, observe } = harness(); + observe(working); + observe(completed(1)); + vi.advanceTimersByTime(WORKING_EXIT_GRACE_MS - 1); + observe(working); + vi.advanceTimersByTime(WORKING_EXIT_GRACE_MS * 2); + expect(notify).not.toHaveBeenCalled(); + }); + + it("notifies a real completion once after the grace", () => { + const { notify, observe } = harness(); + observe(working); + observe(completed(1)); + vi.advanceTimersByTime(WORKING_EXIT_GRACE_MS - 1); + expect(notify).not.toHaveBeenCalled(); + vi.advanceTimersByTime(1); + expect(notify.mock.calls).toEqual([["thread", "completion"]]); + observe({ ...completed(1) }); + vi.advanceTimersByTime(WORKING_EXIT_GRACE_MS * 2); + expect(notify).toHaveBeenCalledTimes(1); + }); + + it("notifies attention at once and replaces a pending completion", () => { + const { notify, observe } = harness(); + observe(working); + observe(completed(1)); + vi.advanceTimersByTime(500); + observe({ attention: "run-1:approval", settled: false, completedAt: null }); + expect(notify.mock.calls).toEqual([["thread", "input"]]); + vi.advanceTimersByTime(60_000); + // The request resolves and the same run reads as completed again. + observe(completed(1)); + vi.advanceTimersByTime(WORKING_EXIT_GRACE_MS * 2); + expect(notify.mock.calls).toEqual([["thread", "input"]]); + }); + + it("notifies one completion across continuation gaps", () => { + const { notify, observe } = harness(); + observe(working); + observe(completed(1)); + vi.advanceTimersByTime(300); + observe(working); + vi.advanceTimersByTime(5_000); + observe(completed(2)); + vi.advanceTimersByTime(300); + observe(working); + vi.advanceTimersByTime(5_000); + observe(completed(3)); + vi.advanceTimersByTime(WORKING_EXIT_GRACE_MS); + expect(notify.mock.calls).toEqual([["thread", "completion"]]); + }); + + it("keeps a completion interrupted by attention after a working gap from firing", () => { + const { notify, observe } = harness(); + observe(working); + observe(completed(1)); + vi.advanceTimersByTime(500); + observe(working); + observe({ attention: "run-1:approval", settled: false, completedAt: null }); + vi.advanceTimersByTime(60_000); + observe(completed(1)); + vi.advanceTimersByTime(WORKING_EXIT_GRACE_MS * 2); + expect(notify.mock.calls).toEqual([["thread", "input"]]); + }); + + it("adds nothing when attention follows a completion that already fired", () => { + const { notify, observe } = harness(); + observe(working); + observe(completed(1)); + vi.advanceTimersByTime(WORKING_EXIT_GRACE_MS); + observe({ attention: "run-1:limited", settled: false, completedAt: null }); + observe(completed(1)); + vi.advanceTimersByTime(WORKING_EXIT_GRACE_MS * 2); + expect(notify.mock.calls).toEqual([ + ["thread", "completion"], + ["thread", "input"], + ]); + }); + + it("still notifies a newer completion after attention", () => { + const { notify, observe } = harness(); + observe(working); + observe(completed(1)); + vi.advanceTimersByTime(500); + observe({ attention: "run-2:approval", settled: false, completedAt: null }); + observe(working); + observe(completed(2)); + vi.advanceTimersByTime(WORKING_EXIT_GRACE_MS); + expect(notify.mock.calls).toEqual([ + ["thread", "input"], + ["thread", "completion"], + ]); + }); + + it("ignores unrelated updates while a completion waits", () => { + const { notify, observe } = harness(); + observe(working); + observe(completed(1)); + vi.advanceTimersByTime(1_500); + observe({ ...completed(1) }); + vi.advanceTimersByTime(500); + expect(notify).toHaveBeenCalledTimes(1); + }); + + it("notifies a run that settles again after a brief runtime blip", () => { + const { notify, observe } = harness(); + observe(working); + observe(completed(1)); + vi.advanceTimersByTime(500); + observe(working); + observe(completed(1)); + vi.advanceTimersByTime(WORKING_EXIT_GRACE_MS); + expect(notify.mock.calls).toEqual([["thread", "completion"]]); + }); + + it("restarts the grace when a newer completion replaces a pending one", () => { + const { notify, observe } = harness(); + observe(working); + observe(completed(1)); + vi.advanceTimersByTime(1_500); + observe(completed(2)); + vi.advanceTimersByTime(1_500); + expect(notify).not.toHaveBeenCalled(); + vi.advanceTimersByTime(500); + expect(notify).toHaveBeenCalledTimes(1); + }); + + it("drops a pending completion on archive, stop request, removal or clear", () => { + for (const end of [ + (h: ReturnType) => h.observe(completed(1), { ...visible, archived: true }), + (h: ReturnType) => + h.observe({ ...completed(1) }, { ...visible, stopRequested: true }), + (h: ReturnType) => h.gate.retain(new Set()), + (h: ReturnType) => h.gate.clear(), + ]) { + const h = harness(); + h.observe(working); + h.observe(completed(1)); + end(h); + vi.advanceTimersByTime(WORKING_EXIT_GRACE_MS * 2); + expect(h.notify).not.toHaveBeenCalled(); + } + }); + + it("never notifies a stopped run or the first observation", () => { + const { notify, observe } = harness(); + observe(completed(1)); + observe(working); + observe(completed(2), { ...visible, stopRequested: true }); + vi.advanceTimersByTime(WORKING_EXIT_GRACE_MS * 2); + expect(notify).not.toHaveBeenCalled(); + }); +}); diff --git a/apps/web/src/components/ThreadNotificationCoordinator.logic.ts b/apps/web/src/components/ThreadNotificationCoordinator.logic.ts new file mode 100644 index 0000000000000..e0e7f270a54a6 --- /dev/null +++ b/apps/web/src/components/ThreadNotificationCoordinator.logic.ts @@ -0,0 +1,78 @@ +import { WORKING_EXIT_GRACE_MS } from "@t3tools/client-runtime/state/thread-inbox"; + +export type ThreadNotificationKind = "input" | "completion"; + +export interface ThreadNotificationObservation { + /** Run and status key while the thread needs attention, otherwise null. */ + readonly attention: string | null; + /** The thread is idle and its latest run completed. */ + readonly settled: boolean; + /** When the latest run completed, if it did. */ + readonly completedAt: number | null; +} + +/** + * Attention notifies at once. Completion waits out the Working shelf's exit grace, + * because an orchestrator run can end moments before its continuation starts. Any + * observation that is not settled during the wait cancels it. + */ +export function createThreadNotificationGate( + notify: (key: Key, kind: ThreadNotificationKind) => void, +) { + const pending = new Map }>(); + // The newest completion already notified, or deliberately skipped. + const handled = new Map(); + // The newest completion seen settled, kept while the thread works or waits. + const lastSettled = new Map(); + const cancel = (key: Key) => { + clearTimeout(pending.get(key)?.timer); + pending.delete(key); + }; + const skip = (key: Key, at: number | null) => { + if (at !== null && at > (handled.get(key) ?? -Infinity)) handled.set(key, at); + }; + return { + observe( + key: Key, + prior: ThreadNotificationObservation | undefined, + next: ThreadNotificationObservation, + options: { readonly archived: boolean; readonly stopRequested: boolean }, + ): void { + if (next.settled && next.completedAt !== null) lastSettled.set(key, next.completedAt); + if (!next.settled || options.archived || options.stopRequested) cancel(key); + // The first sighting, archived threads and stopped runs never notify completion. + if (!prior || options.archived || options.stopRequested) skip(key, next.completedAt); + if (!prior || options.archived) return; + if (next.attention && next.attention !== prior.attention) { + // Attention replaces the completion before it, so that run must not notify later. + skip(key, lastSettled.get(key) ?? null); + cancel(key); + notify(key, "input"); + return; + } + const at = next.completedAt; + if (!next.settled || at === null || at <= (handled.get(key) ?? -Infinity)) return; + if (pending.get(key)?.at === at) return; + cancel(key); + pending.set(key, { + at, + timer: setTimeout(() => { + pending.delete(key); + handled.set(key, at); + notify(key, "completion"); + }, WORKING_EXIT_GRACE_MS), + }); + }, + /** Forgets threads that left the snapshot. */ + retain(keys: ReadonlySet): void { + for (const key of pending.keys()) if (!keys.has(key)) cancel(key); + 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); + }, + clear(): void { + for (const key of pending.keys()) cancel(key); + handled.clear(); + lastSettled.clear(); + }, + }; +} diff --git a/apps/web/src/components/ThreadNotificationCoordinator.tsx b/apps/web/src/components/ThreadNotificationCoordinator.tsx index c9355fc7f8206..ea3bcdc78479c 100644 --- a/apps/web/src/components/ThreadNotificationCoordinator.tsx +++ b/apps/web/src/components/ThreadNotificationCoordinator.tsx @@ -1,4 +1,5 @@ import { + type EnvironmentThreadShell, type PendingSubagentRequest, presentThreadShell, } from "@t3tools/client-runtime/state/models"; @@ -13,7 +14,7 @@ import { MessageCircleQuestionIcon, ShieldQuestionIcon, } from "lucide-react"; -import { useEffect, useRef } from "react"; +import { useEffect, useEffectEvent, useRef } from "react"; import { getClientSettings, useClientSettings } from "../hooks/useSettings"; import { useEnvironmentIds } from "../state/environments"; @@ -26,7 +27,12 @@ import { threadNotificationTag, unlockNotificationAudio, } from "../threadNotifications"; -import { resolveSidebarThreadStatus } from "./Sidebar.logic"; +import { resolveSidebarThreadStatus, type SidebarThreadStatus } from "./Sidebar.logic"; +import { + createThreadNotificationGate, + type ThreadNotificationKind, + type ThreadNotificationObservation, +} from "./ThreadNotificationCoordinator.logic"; import { toastManager } from "./ui/toast"; export function ThreadNotificationCoordinator() { @@ -93,11 +99,11 @@ export function ThreadNotificationCoordinator() { )); } -interface NotificationState { +interface NotificationState extends ThreadNotificationObservation { readonly raw: OrchestrationV2ThreadShell; readonly subagentRequest: PendingSubagentRequest | null; - readonly attention: string | null; - readonly completion: number | null; + readonly thread: EnvironmentThreadShell; + readonly status: SidebarThreadStatus; } function EnvironmentNotifications({ @@ -120,13 +126,114 @@ function EnvironmentNotifications({ }); const previous = useRef(new Map()); + // Completions fire from a timer, so read settings, focus and route when presenting. + const present = useEffectEvent((threadId: ThreadId, kind: ThreadNotificationKind) => { + const entry = previous.current.get(threadId); + if (!entry) return; + const { thread, status } = entry; + const title = + kind === "completion" + ? "Thread completed" + : status === "approval" + ? "Approval needed" + : status === "limited" + ? "Usage limit reached" + : status === "failed" + ? "Thread failed" + : "Input needed"; + if (hasNotificationSound(mode)) { + void playNotificationSound(kind, () => + hasNotificationSound(getClientSettings().notificationMode), + ); + } + if ( + inAppNotificationsEnabled && + document.visibilityState === "visible" && + document.hasFocus() && + (activeEnvironmentId !== environmentId || activeThreadId !== thread.id) + ) { + const toastId = toastManager.add({ + type: kind === "completion" ? "success" : status === "failed" ? "error" : "warning", + title, + description: thread.title, + data: { + hideCopyButton: true, + leadingIcon: + kind === "completion" ? ( + + ) : status === "approval" ? ( + + ) : status === "failed" ? ( + + ) : ( + + ), + }, + actionProps: { + children: "Open thread", + onClick: () => { + toastManager.close(toastId); + void navigate({ + to: "/$environmentId/$threadId", + params: { environmentId, threadId: thread.id }, + }); + }, + }, + }); + return; + } + if ( + !hasDesktopNotifications(mode) || + (document.visibilityState === "visible" && document.hasFocus()) || + typeof Notification === "undefined" || + Notification.permission !== "granted" + ) + return; + try { + const notification = new Notification(title, { + body: thread.title, + tag: threadNotificationTag(environmentId, thread.id), + silent: true, + }); + onNotification(environmentId, notification); + notification.addEventListener("click", () => { + notification.close(); + window.focus(); + void window.desktopBridge?.revealWindow?.().catch(() => undefined); + void navigate({ + to: "/$environmentId/$threadId", + params: { environmentId, threadId: thread.id }, + }); + }); + } catch { + // Some browsers expose Notification but reject desktop presentation. + } + }); + const gate = useRef>>(null); useEffect(() => { + const created = createThreadNotificationGate((threadId, kind) => + present(threadId, kind), + ); + gate.current = created; + return () => { + created.clear(); + gate.current = null; + // A new gate has no pending completions, so forget the states they compared against. + previous.current.clear(); + }; + }, []); + + useEffect(() => { + const pendingCompletions = gate.current; + if (!pendingCompletions) return; if (threads === null) { + pendingCompletions.clear(); previous.current.clear(); return; } const next = new Map(); const subagentRequests = pendingSubagentRequestsByThread(threads); + const changed: Array<[NotificationState | undefined, NotificationState]> = []; for (const rawThread of threads) { if (rawThread.lineage.relationshipToParent === "subagent") continue; const prior = previous.current.get(rawThread.id); @@ -143,112 +250,28 @@ function EnvironmentNotifications({ ? `${thread.latestRun?.runId ?? ""}:${status}` : null; const completedAt = Date.parse(thread.latestRun?.completedAt ?? ""); - const completion = - status === "ready" && - thread.latestRun?.status === "completed" && - Number.isFinite(completedAt) - ? completedAt - : (prior?.completion ?? null); - next.set(thread.id, { raw: rawThread, subagentRequest, attention, completion }); - if (!prior || thread.archivedAt !== null) continue; - const kind = - attention && attention !== prior.attention - ? "input" - : !rawThread.latestRunStopRequested && - completion !== null && - (prior.completion === null || completion > prior.completion) - ? "completion" - : null; - if (!kind) continue; - const title = - kind === "completion" - ? "Thread completed" - : status === "approval" - ? "Approval needed" - : status === "limited" - ? "Usage limit reached" - : status === "failed" - ? "Thread failed" - : "Input needed"; - if (hasNotificationSound(mode)) { - void playNotificationSound(kind, () => - hasNotificationSound(getClientSettings().notificationMode), - ); - } - if ( - inAppNotificationsEnabled && - document.visibilityState === "visible" && - document.hasFocus() && - (activeEnvironmentId !== environmentId || activeThreadId !== thread.id) - ) { - const toastId = toastManager.add({ - type: kind === "completion" ? "success" : status === "failed" ? "error" : "warning", - title, - description: thread.title, - data: { - hideCopyButton: true, - leadingIcon: - kind === "completion" ? ( - - ) : status === "approval" ? ( - - ) : status === "failed" ? ( - - ) : ( - - ), - }, - actionProps: { - children: "Open thread", - onClick: () => { - toastManager.close(toastId); - void navigate({ - to: "/$environmentId/$threadId", - params: { environmentId, threadId: thread.id }, - }); - }, - }, - }); - continue; - } - if ( - !hasDesktopNotifications(mode) || - (document.visibilityState === "visible" && document.hasFocus()) || - typeof Notification === "undefined" || - Notification.permission !== "granted" - ) - continue; - try { - const notification = new Notification(title, { - body: thread.title, - tag: threadNotificationTag(environmentId, thread.id), - silent: true, - }); - onNotification(environmentId, notification); - notification.addEventListener("click", () => { - notification.close(); - window.focus(); - void window.desktopBridge?.revealWindow?.().catch(() => undefined); - void navigate({ - to: "/$environmentId/$threadId", - params: { environmentId, threadId: thread.id }, - }); - }); - } catch { - // Some browsers expose Notification but reject desktop presentation. - } + const settled = status === "ready" && thread.latestRun?.status === "completed"; + const state = { + raw: rawThread, + subagentRequest, + thread, + status, + attention, + settled, + completedAt: settled && Number.isFinite(completedAt) ? completedAt : null, + }; + next.set(thread.id, state); + changed.push([prior, state]); } previous.current = next; - }, [ - activeEnvironmentId, - activeThreadId, - environmentId, - inAppNotificationsEnabled, - mode, - navigate, - onNotification, - threads, - ]); + pendingCompletions.retain(new Set(next.keys())); + for (const [prior, state] of changed) { + pendingCompletions.observe(state.thread.id, prior, state, { + archived: state.thread.archivedAt !== null, + stopRequested: state.raw.latestRunStopRequested === true, + }); + } + }, [environmentId, threads]); return null; } diff --git a/packages/client-runtime/src/state/threadInbox.ts b/packages/client-runtime/src/state/threadInbox.ts index 896095feb0aae..4ca7baad1bb1b 100644 --- a/packages/client-runtime/src/state/threadInbox.ts +++ b/packages/client-runtime/src/state/threadInbox.ts @@ -104,6 +104,13 @@ type TrackedInboxThread = WorkingThreadInput & > >; +/** + * How long a thread that stopped working keeps reading as working. An orchestrator + * run can end moments before its continuation starts, so clients wait this long + * before treating the thread as done. + */ +export const WORKING_EXIT_GRACE_MS = 2_000; + /** Keep at module scope so shelf placement and return order survive route changes. */ export function createInboxReturnTracker() { const keyOf = (thread: Pick) => @@ -200,7 +207,7 @@ export function createInboxReturnTracker() { // Only shelf placement is held: execution state and attention requests stay authoritative. const deadline = !actual && previous?.working && !actionable && !moved && !excluded - ? (previous.deadline ?? at + 2_000) + ? (previous.deadline ?? at + WORKING_EXIT_GRACE_MS) : null; const working = actual || (deadline !== null && deadline > at); if (previous?.working && !working) returns.set(key, at);