diff --git a/apps/mobile/src/features/home/HomeScreen.tsx b/apps/mobile/src/features/home/HomeScreen.tsx index f20a8fee636ad..dbb467a128438 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,10 @@ export function HomeScreen(props: HomeScreenProps) { ...(workingShelfEnabled ? [] : sectionAvailability("active")), ]); })(); + const inboxReturnsVersion = useSyncExternalStore( + threadListInboxReturns.subscribe, + threadListInboxReturns.getSnapshot, + ); const threadListV2Layout = (() => { threadListInboxReturns.observe(workingShelfEnabled ? props.threads : null); return buildThreadListV2Items({ @@ -505,6 +509,8 @@ export function HomeScreen(props: HomeScreenProps) { workingShelfEnabled, 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 13ce478929d25..ebefcf8e47974 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,10 @@ function ThreadNavigationSidebarPane( ...(workingShelfEnabled ? [] : sectionAvailability("active")), ]); })(); + const inboxReturnsVersion = useSyncExternalStore( + threadListInboxReturns.subscribe, + threadListInboxReturns.getSnapshot, + ); const threadListV2Layout = (() => { threadListInboxReturns.observe(workingShelfEnabled ? threads : null); return buildThreadListV2Items({ @@ -337,6 +341,8 @@ function ThreadNavigationSidebarPane( workingShelfEnabled, workingShelfExpanded, inboxReturnAt: threadListInboxReturns.returnedAt, + isWorking: threadListInboxReturns.isWorking, + inboxReturnsVersion, 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..10f2186494805 100644 --- a/apps/mobile/src/features/threads/threadListV2.ts +++ b/apps/mobile/src/features/threads/threadListV2.ts @@ -589,7 +589,11 @@ export function buildThreadListV2Items(input: { readonly now: string; readonly workingShelfEnabled?: boolean; 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; @@ -671,7 +675,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..a70d5d3e1ef6e 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, @@ -303,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; } @@ -383,6 +387,10 @@ function useSidebarThreads(input: { ), ); const workingShelfEnabled = useClientSettings((s) => s.sidebarWorkingShelfEnabled); + 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. @@ -424,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); @@ -520,8 +528,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 +557,7 @@ function partitionSidebarThreads(input: SidebarThreadPartitionInput) { ? pinned : optimisticDrop.section === "settled" ? settled - : inbox(projected) + : inbox(projected, true) ).push( optimisticDrop.clearsSnooze ? projected 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.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..4ca7baad1bb1b 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,142 @@ function sortNewestFirst + >; + /** - * Keep one at module scope so the inbox order survives routes that unmount the - * list. Call `observe` with every thread shell on each list rebuild, or with - * null to reset while the beta is off. + * 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) => `${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); + 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 + WORKING_EXIT_GRACE_MS) + : 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 }); } - for (const key of returns.keys()) { - if (!present.has(key)) returns.delete(key); - } - // @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)), };