diff --git a/apps/mobile/src/features/threads/threadListV2.test.ts b/apps/mobile/src/features/threads/threadListV2.test.ts index 58d06bd5e696d..cb23296d41c80 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"; @@ -1562,6 +1563,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 2486d33fee89f..424bd792c40d2 100644 --- a/apps/mobile/src/features/threads/threadListV2.ts +++ b/apps/mobile/src/features/threads/threadListV2.ts @@ -577,6 +577,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; @@ -659,7 +660,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/mobile/src/features/threads/useThreadListV2.ts b/apps/mobile/src/features/threads/useThreadListV2.ts index 12d861d193a7d..2a2ab54033ea8 100644 --- a/apps/mobile/src/features/threads/useThreadListV2.ts +++ b/apps/mobile/src/features/threads/useThreadListV2.ts @@ -1,5 +1,5 @@ import type { EnvironmentId } from "@t3tools/contracts"; -import { useMemo } from "react"; +import { useMemo, useSyncExternalStore } from "react"; import { scopedProjectKey } from "../../lib/scopedEntities"; import type { PendingNewTask } from "../../state/use-pending-new-tasks"; @@ -54,10 +54,15 @@ export function useThreadListV2(input: { readonly selectedThreadKey: string | null; }) { const { threads, workingShelfEnabled } = input; - // Observed once per change to the threads: that is when a thread can stop working. - const inboxReturnAt = useMemo( + // A thread can stop working when the threads change, or when its exit grace expires. + const inboxVersion = useSyncExternalStore( + threadListInboxReturns.subscribe, + threadListInboxReturns.getSnapshot, + ); + const inbox = useMemo( () => threadListInboxReturns.observe(workingShelfEnabled ? threads : null), - [threads, workingShelfEnabled], + // oxlint-disable-next-line react/memo-dependencies -- the version signals an expired grace + [threads, workingShelfEnabled, inboxVersion], ); // Up/down menu availability for every card, computed once per section per // rebuild (see computeThreadMoveAvailability): per-thread planner calls made @@ -97,7 +102,8 @@ export function useThreadListV2(input: { now: input.now, workingShelfEnabled, workingShelfExpanded: input.workingShelfExpanded, - inboxReturnAt, + inboxReturnAt: inbox.returnedAt, + isWorking: inbox.isWorking, snoozedShelfExpanded: input.snoozedShelfExpanded, settledShelfExpanded: input.settledShelfExpanded, selectedThreadKey: input.selectedThreadKey, diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts index 184a0295b653b..b978f211589b1 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts @@ -1340,7 +1340,7 @@ describe("PullRequestSyncReactor", () => { "gh pr merge 7 --squash 2>&1 | tail -3", "cafe pr merge 7 --squash", "az repos pr update --id 7 --status completed", - ])("reads open links fresh only when a run that ran `%s` ends", (mergeCommand) => + ])("reads open links fresh only for a thread that ran `%s`", (mergeCommand) => Effect.scoped( Effect.gen(function* () { yield* TestClock.setTime(Date.parse(NOW)); @@ -1392,4 +1392,92 @@ describe("PullRequestSyncReactor", () => { }), ), ); + + it.effect("reads open links again when the run ends, after `merge --auto` left them open", () => + Effect.scoped( + Effect.gen(function* () { + yield* TestClock.setTime(Date.parse(NOW)); + const state = yield* Ref.make("open"); + const fixture = yield* makeHarness({ + snapshot: makeSnapshot([ + makeThread("agent", { pullRequests: [makeLink(7, { state: "open" })] }), + ]), + summary: (input) => + Ref.get(state).pipe( + Effect.map((current) => + makeSummary(input, current === "merged" ? { state: current, mergedAt: NOW } : {}), + ), + ), + }); + + yield* Effect.gen(function* () { + const reactor = yield* startAndSweep(fixture); + yield* Queue.offer( + fixture.domainEvents, + commandRan(ThreadId.make("agent"), "gh pr merge 7 --auto"), + ); + yield* Queue.take(fixture.snapshotReads); + yield* Queue.take(fixture.snapshotReads); + yield* reactor.drain; + yield* Ref.set(state, "merged"); + yield* Ref.set(fixture.syncCommands, []); + + yield* Queue.offer(fixture.domainEvents, runUpdated(ThreadId.make("agent"), "completed")); + yield* Queue.take(fixture.snapshotReads); + yield* Queue.take(fixture.snapshotReads); + yield* reactor.drain; + + assert.deepStrictEqual( + (yield* Ref.get(fixture.syncCommands)).map((command) => [ + command.number, + command.snapshot.state, + ]), + [[7, "merged"]], + ); + }).pipe(Effect.provide(fixture.layer)); + }), + ), + ); + + it.effect("reads open links fresh as soon as a merge command ends, before its run ends", () => + Effect.scoped( + Effect.gen(function* () { + yield* TestClock.setTime(Date.parse(NOW)); + const state = yield* Ref.make("open"); + const fixture = yield* makeHarness({ + snapshot: makeSnapshot([ + makeThread("agent", { pullRequests: [makeLink(7, { state: "open" })] }), + ]), + summary: (input) => + Ref.get(state).pipe( + Effect.map((current) => + makeSummary(input, current === "merged" ? { state: current, mergedAt: NOW } : {}), + ), + ), + }); + + yield* Effect.gen(function* () { + const reactor = yield* startAndSweep(fixture); + yield* Ref.set(state, "merged"); + yield* Ref.set(fixture.summaryCalls, []); + + yield* Queue.offer( + fixture.domainEvents, + commandRan(ThreadId.make("agent"), "gh pr merge 7"), + ); + yield* Queue.take(fixture.snapshotReads); + yield* Queue.take(fixture.snapshotReads); + yield* reactor.drain; + + assert.deepStrictEqual( + (yield* Ref.get(fixture.syncCommands)).map((command) => [ + command.number, + command.snapshot.state, + ]), + [[7, "merged"]], + ); + }).pipe(Effect.provide(fixture.layer)); + }), + ), + ); }); diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts index e2be84c5663db..fb9fa25c6e1bf 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts @@ -1,5 +1,6 @@ import { CommandId, + isOrchestrationV2WorkActive, type PullRequestSummary, type ThreadId, type ThreadPullRequestKey, @@ -481,7 +482,8 @@ export const make = Effect.gen(function* () { }).pipe(Effect.catchCause(logSkipped("pull request sync sweep failed", {}))), ); - const closeCommandThreads = new Set(); + // Threads that ran a close command this run, with the commands already read for. + const closeCommands = new Map>(); const refreshOpenLinks = (threadId: ThreadId) => projections.getThreadsWithPullRequests(threadId).pipe( Effect.flatMap((threads) => @@ -525,20 +527,22 @@ export const make = Effect.gen(function* () { { discard: true }, ); // An agent can merge or close its pull request from a shell with its host's CLI, which - // sends no merge notification. When a run that ran such a command ends, read the - // thread's open links fresh, so settlement does not wait for the next sweep and the - // cached summary. Other runs add no host reads. - case "turn-item.updated": - if ( - event.payload.type === "command_execution" && - isChangeRequestCloseCommand(event.payload.input) - ) { - closeCommandThreads.add(event.threadId); + // sends no merge notification. Read the thread's open links fresh once that command + // ends, while the agent usually still replies, so settlement can land before the run + // ends. Read again when the run ends, since `merge --auto` returns before it merges. + case "turn-item.updated": { + const item = event.payload; + if (item.type !== "command_execution" || !isChangeRequestCloseCommand(item.input)) { + return Effect.void; } - return Effect.void; + const ended = closeCommands.get(event.threadId) ?? new Set(); + closeCommands.set(event.threadId, ended); + if (isOrchestrationV2WorkActive(item.status) || ended.has(item.id)) return Effect.void; + ended.add(item.id); + return refreshOpenLinks(event.threadId); + } case "run.updated": - return isTerminalRunStatus(event.payload.status) && - closeCommandThreads.delete(event.threadId) + return isTerminalRunStatus(event.payload.status) && closeCommands.delete(event.threadId) ? refreshOpenLinks(event.threadId) : Effect.void; default: diff --git a/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts b/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts index c16a012945c5d..50ac83b814626 100644 --- a/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts @@ -6,6 +6,7 @@ import { ProjectId, EventId, ProviderInstanceId, + RunId, ThreadId, type OrchestrationProjectShell, type OrchestrationV2AppThread, @@ -892,6 +893,164 @@ describe("ThreadSettlementServiceV2 worker", () => { ), ); + it.effect("settles a merged thread when its background work ends, during a slow full sweep", () => + Effect.scoped( + Effect.gen(function* () { + yield* TestClock.setTime(Date.parse(NOW)); + const periodicLookupStarted = yield* Deferred.make(); + const releasePeriodicLookup = yield* Deferred.make(); + const settled = yield* Deferred.make(); + const branchLookupCount = yield* Ref.make(0); + const mergedLink: NonNullable[number] = { + ...persistedLink("open", DateTime.makeUnsafe("2026-08-20T00:00:00.000Z")), + snapshot: { + state: "merged", + title: "Pull request", + headBranch: "feature", + baseBranch: "main", + isDraft: false, + updatedAt: NOW, + syncedAt: NOW, + mergedAt: NOW, + }, + }; + const waiting = makeThread("waiting-on-subagent", { + pullRequests: [mergedLink], + pendingBackgroundTasks: [{ taskId: "child", kind: "subagent", description: "Review" }], + }); + const slow = makeThread("slow-periodic-lookup", { + branch: "another-feature", + latestUserMessageAt: DateTime.makeUnsafe("2026-08-27T00:00:00.000Z"), + }); + const fixture = yield* makeHarness({ + snapshot: makeSnapshot([waiting, slow]), + branchPullRequest: () => + Ref.updateAndGet(branchLookupCount, (count) => count + 1).pipe( + Effect.flatMap((count) => + count === 1 + ? Effect.succeed(makeBranchPullRequest("open")) + : Deferred.succeed(periodicLookupStarted, undefined).pipe( + Effect.andThen(Deferred.await(releasePeriodicLookup)), + Effect.as(makeBranchPullRequest("open")), + ), + ), + ), + onDispatch: () => Deferred.succeed(settled, undefined), + }); + + yield* Effect.gen(function* () { + const reactor = yield* ThreadSettlementService.ThreadSettlementServiceV2; + yield* startHarness(reactor, fixture.activation, fixture.snapshotReads); + assert.deepStrictEqual(yield* Ref.get(fixture.commands), []); + yield* fixture.updateSettings({ sidebarAutoSettleAfterDays: 4 }); + yield* Deferred.await(periodicLookupStarted); + + yield* Ref.set( + fixture.snapshots, + makeSnapshot([{ ...waiting, pendingBackgroundTasks: [] }, slow]), + ); + yield* fixture.publishEvent({ + type: "run.background-work-cancelled", + id: EventId.make("event:background-ended"), + threadId: waiting.id, + occurredAt: DateTime.makeUnsafe(NOW), + payload: { runId: RunId.make("run:waiting"), restartCancelledBackgroundWork: [] }, + }); + yield* Deferred.await(settled); + + assert.deepStrictEqual( + (yield* Ref.get(fixture.commands)).map((command) => command.threadId), + [waiting.id], + ); + yield* Deferred.succeed(releasePeriodicLookup, undefined); + yield* reactor.drain; + }).pipe(Effect.provide(fixture.layer)); + }), + ), + ); + + it.effect("settles a stopped thread while another thread's lookup is slow", () => + Effect.scoped( + Effect.gen(function* () { + yield* TestClock.setTime(Date.parse(NOW)); + const slowLookupStarted = yield* Deferred.make(); + const releaseSlowLookup = yield* Deferred.make(); + const settled = yield* Deferred.make(); + const branchLookupCount = yield* Ref.make(0); + const recent = DateTime.makeUnsafe("2026-08-27T00:00:00.000Z"); + const slow = makeThread("slow-lookup", { branch: "feature", latestUserMessageAt: recent }); + const merged = makeThread("merged", { + latestUserMessageAt: recent, + pullRequests: [ + { + ...persistedLink("open", recent), + snapshot: { + state: "merged", + title: "Pull request", + headBranch: "feature", + baseBranch: "main", + isDraft: false, + updatedAt: NOW, + syncedAt: NOW, + mergedAt: NOW, + }, + }, + ], + pendingBackgroundTasks: [{ taskId: "child", kind: "subagent", description: "Review" }], + }); + const fixture = yield* makeHarness({ + snapshot: makeSnapshot([slow, merged]), + branchPullRequest: () => + Ref.updateAndGet(branchLookupCount, (count) => count + 1).pipe( + Effect.flatMap((count) => + count === 1 + ? Effect.succeed(makeBranchPullRequest("open")) + : Deferred.succeed(slowLookupStarted, undefined).pipe( + Effect.andThen(Deferred.await(releaseSlowLookup)), + Effect.as(makeBranchPullRequest("open")), + ), + ), + ), + onDispatch: () => Deferred.succeed(settled, undefined), + }); + const backgroundEnded = (threadId: ThreadId): OrchestrationV2DomainEvent => ({ + type: "run.background-work-cancelled", + id: EventId.make(`event:background-ended:${threadId}`), + threadId, + occurredAt: DateTime.makeUnsafe(NOW), + payload: { runId: RunId.make(`run:${threadId}`), restartCancelledBackgroundWork: [] }, + }); + + yield* Effect.gen(function* () { + const reactor = yield* ThreadSettlementService.ThreadSettlementServiceV2; + yield* startHarness(reactor, fixture.activation, fixture.snapshotReads); + yield* fixture.publishEvent(backgroundEnded(slow.id)); + yield* Deferred.await(slowLookupStarted); + // More events for the slow thread coalesce into one more pass, not more sweeps. + for (let index = 0; index < 9; index++) { + yield* fixture.publishEvent(backgroundEnded(slow.id)); + } + + yield* Ref.set( + fixture.snapshots, + makeSnapshot([slow, { ...merged, pendingBackgroundTasks: [] }]), + ); + yield* fixture.publishEvent(backgroundEnded(merged.id)); + yield* Deferred.await(settled); + + assert.deepStrictEqual( + (yield* Ref.get(fixture.commands)).map((command) => command.threadId), + [merged.id], + ); + yield* Deferred.succeed(releaseSlowLookup, undefined); + yield* reactor.drain; + // The startup sweep, the first per-thread sweep, and one coalesced rerun. + assert.strictEqual(yield* Ref.get(branchLookupCount), 3); + }).pipe(Effect.provide(fixture.layer)); + }), + ), + ); + it.effect( "settles branch threads on a pull request merge without waiting for the next sweep", () => diff --git a/apps/server/src/orchestration-v2/ThreadSettlementService.ts b/apps/server/src/orchestration-v2/ThreadSettlementService.ts index bd5b7d346fb12..a14601ae84054 100644 --- a/apps/server/src/orchestration-v2/ThreadSettlementService.ts +++ b/apps/server/src/orchestration-v2/ThreadSettlementService.ts @@ -1,8 +1,12 @@ -import { backgroundWorkHoldsCompletion } from "@t3tools/shared/orchestrationV2PendingBackgroundWork"; +import { + backgroundWorkHoldsCompletion, + turnItemUpdateCanEndBackgroundWork, +} from "@t3tools/shared/orchestrationV2PendingBackgroundWork"; import { resolveProjectSettings } from "@t3tools/shared/projectSettings"; import { visibleThreadPullRequests } from "@t3tools/shared/threadPullRequests"; import { CommandId, + isOrchestrationV2WorkActive, type ThreadId, type OrchestrationV2DomainEvent, type OrchestrationV2ThreadShell, @@ -13,9 +17,11 @@ import * as Context from "effect/Context"; import * as Crypto from "effect/Crypto"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; +import * as FiberSet from "effect/FiberSet"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Schedule from "effect/Schedule"; +import * as Semaphore from "effect/Semaphore"; import type * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; @@ -523,9 +529,35 @@ export const make = Effect.gen(function* () { }), ), ); - const worker = yield* makeDrainableWorker((threadId: ThreadId | undefined) => - runSweep(null, threadId), - ); + // A thread that just stopped must not wait behind a full sweep or another thread's host + // reads: clients hold it in Working only briefly, so a late settle makes it appear and then vanish. + const fullSweeps = yield* makeDrainableWorker(() => runSweep(null)); + const threadSweeps = yield* FiberSet.make(); + const threadSweepSlots = yield* Semaphore.make(8); + // One sweep per thread at a time; events during a running sweep coalesce into one more pass. + const threadSweepStates = new Map(); + const sweepThread = ( + threadId: ThreadId, + ): Effect.Effect>> => + Effect.sync(() => threadSweepStates.set(threadId, "running")).pipe( + Effect.andThen(runSweep(null, threadId)), + threadSweepSlots.withPermits(1), + Effect.andThen( + Effect.suspend(() => { + if (threadSweepStates.get(threadId) === "rerun") return sweepThread(threadId); + threadSweepStates.delete(threadId); + return Effect.void; + }), + ), + ); + const enqueueThread = (threadId: ThreadId) => + Effect.suspend(() => { + const state = threadSweepStates.get(threadId); + if (state === "running") threadSweepStates.set(threadId, "rerun"); + if (state !== undefined) return Effect.void; + threadSweepStates.set(threadId, "queued"); + return sweepThread(threadId).pipe(FiberSet.run(threadSweeps), Effect.asVoid); + }); // Skip settle scripts in shared checkouts where other threads may still work. const cleanUpSettledThread = Effect.fn("ThreadSettlementServiceV2.cleanUpSettledThread")( @@ -572,16 +604,30 @@ export const make = Effect.gen(function* () { case "thread.pull-request-link-updated": case "thread.pull-request-synced": case "provider-session.detached": - return worker.enqueue(event.threadId); + case "run.background-work-cancelled": + return enqueueThread(event.threadId); + // Sweep only when the work can have ended: progress updates cannot settle anything. + case "subagent.updated": + return isOrchestrationV2WorkActive(event.payload.status) + ? Effect.void + : enqueueThread(event.threadId); + case "provider-thread.updated": + return backgroundWorkHoldsCompletion(event.payload.pendingBackgroundTasks ?? []) + ? Effect.void + : enqueueThread(event.threadId); + case "turn-item.updated": + return turnItemUpdateCanEndBackgroundWork(event.payload) + ? enqueueThread(event.threadId) + : Effect.void; case "provider-session.updated": return event.payload.status !== "starting" && event.payload.status !== "running" - ? worker.enqueue(event.threadId) + ? enqueueThread(event.threadId) : Effect.void; case "run.updated": return ["completed", "interrupted", "failed", "cancelled", "rolled_back"].includes( event.payload.status, ) - ? worker.enqueue(event.threadId) + ? enqueueThread(event.threadId) : Effect.void; default: return Effect.void; @@ -598,8 +644,8 @@ export const make = Effect.gen(function* () { let lastSettlementSettings = autoSettlementSettingsKey(initialSettings); yield* forkParked( Effect.gen(function* () { - yield* worker.enqueue(undefined); - yield* worker.drain; + yield* fullSweeps.enqueue(undefined); + yield* fullSweeps.drain; }).pipe(Effect.repeat(Schedule.spaced("1 minute")), Effect.asVoid), ); yield* forkParked( @@ -609,7 +655,7 @@ export const make = Effect.gen(function* () { return Effect.void; } lastSettlementSettings = key; - return worker.enqueue(undefined); + return fullSweeps.enqueue(undefined); }), ); yield* forkParked(Stream.runForEach(mergedPullRequests, (event) => runSweep(event))); @@ -622,7 +668,10 @@ export const make = Effect.gen(function* () { ); }); - return { start, drain: worker.drain } satisfies ThreadSettlementServiceV2["Service"]; + return { + start, + drain: Effect.andThen(fullSweeps.drain, FiberSet.awaitEmpty(threadSweeps)), + } satisfies ThreadSettlementServiceV2["Service"]; }); export const layer = Layer.effect(ThreadSettlementServiceV2, make); diff --git a/apps/web/src/components/Sidebar.tsx b/apps/web/src/components/Sidebar.tsx index 8d40d5f01a48d..2101d2ddae557 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, @@ -154,7 +158,6 @@ import { filterSidebarProjectScopeItems, firstValidTimestampMs, isSidebarNestedLinkClick, - isSidebarThreadWorking, isTrailingDoubleClick, orderItemsByPreferredIds, planSidebarThreadDrop, @@ -302,6 +305,9 @@ interface SidebarThreadPartitionInput { readonly serverConfigs: ReadonlyMap; readonly optimisticDrop: SidebarOptimisticDrop | null; readonly workingShelfEnabled: boolean; + // 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; } @@ -375,6 +381,10 @@ function useSidebarThreads(input: { ), ); const workingShelfEnabled = useClientSettings((s) => s.sidebarWorkingShelfEnabled); + const inboxReturnsVersion = useSyncExternalStore( + inboxReturns.subscribe, + inboxReturns.getSnapshot, + ); const nowMinute = useNowMinute(); // Minute-quantized clocks would delay second-precise snooze wakes by up to a minute. const [snoozeWakeTick, bumpSnoozeWakeTick] = useState(0); @@ -414,7 +424,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); @@ -506,12 +516,15 @@ function partitionSidebarThreads(input: SidebarThreadPartitionInput) { const { threads, scopedProjectKeys, serverConfigs, optimisticDrop, workingShelfEnabled } = input; const preciseNow = new Date().toISOString(); const visible = filterSidebarV2VisibleThreads(threads, scopedProjectKeys); - inboxReturns.observe(workingShelfEnabled ? threads : null); + const inboxObservation = inboxReturns.observe(workingShelfEnabled ? threads : null); 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) : inboxObservation.isWorking(thread)) + ? working + : active; const snoozed: EnvironmentThreadShell[] = []; const settled: EnvironmentThreadShell[] = []; const draggable = new Set(); @@ -536,7 +549,7 @@ function partitionSidebarThreads(input: SidebarThreadPartitionInput) { ? pinned : optimisticDrop.section === "settled" ? settled - : inbox(projected) + : inbox(projected, true) ).push( optimisticDrop.clearsSnooze ? projected @@ -561,7 +574,7 @@ function partitionSidebarThreads(input: SidebarThreadPartitionInput) { // Capability gates dragging only; mixed server versions must preserve the same pinned order across clients. const sortedPinned = sortPinnedThreadsForSidebar(pinned); const sortedActive = workingShelfEnabled - ? sortInboxThreadsByReturn(active, inboxReturns.returnedAt) + ? sortInboxThreadsByReturn(active, inboxObservation.returnedAt) : sortThreadsForSidebar(active); return { pinnedThreads: diff --git a/apps/web/src/components/ThreadNotificationCoordinator.badge.browser.test.tsx b/apps/web/src/components/ThreadNotificationCoordinator.badge.browser.test.tsx index 145d6f740f02c..fe0e21190d02b 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", @@ -235,6 +236,31 @@ it.each([ latestTurn: { turnId: "turn", state: status, completedAt: "2026-09-13T07:01:00.000Z" }, }), ); + using _timers = fakeTimers(); await screen.rerender(app.ui()); + await vi.advanceTimersByTimeAsync(WORKING_EXIT_GRACE_MS); expect(TestNotification.sent.map((notification) => notification.title)).toEqual(titles); }); + +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..2bba09f54dcb1 --- /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, quiet: 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, quiet: 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, quiet: 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..77edccfbe7741 --- /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 quiet: boolean }, + ): void { + if (next.settled && next.completedAt !== null) lastSettled.set(key, next.completedAt); + if (!next.settled || options.archived || options.quiet) cancel(key); + // The first sighting, archived threads, and stopped or silent runs never notify completion. + if (!prior || options.archived || options.quiet) 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 9769fbce75e62..1d6f37ed351d8 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,116 @@ 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" + : status === "request" + ? "Guest request" + : "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); @@ -147,115 +256,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 && - !rawThread.latestRunSilent && - 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" - : status === "request" - ? "Guest request" - : "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, + quiet: state.raw.latestRunStopRequested === true || state.raw.latestRunSilent === 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 f5675d3a72eef..35c7fa8355e39 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,18 +35,185 @@ 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(); const first = tracker.observe([thread("a", true), thread("b", false)]); - expect(first(thread("a", true))).toBeUndefined(); - expect(first(thread("b", false))).toBeUndefined(); + expect(first.returnedAt(thread("a", true))).toBeUndefined(); + expect(first.returnedAt(thread("b", false))).toBeUndefined(); + const held = tracker.observe([thread("a", false), thread("b", false)]); + vi.advanceTimersByTime(2_000); const second = tracker.observe([thread("a", false), thread("b", false)]); - expect(second(thread("a", false))).toBeDefined(); - expect(second(thread("b", false))).toBeUndefined(); - expect(tracker.returnedAt(thread("a", false))).toBe(second(thread("a", false))); + expect(second.returnedAt(thread("a", false))).toBeDefined(); + expect(second.returnedAt(thread("b", false))).toBeUndefined(); + expect(tracker.returnedAt(thread("a", false))).toBe(second.returnedAt(thread("a", false))); // An earlier observation keeps its answer, so lists built from it stay valid. - expect(first(thread("a", false))).toBeUndefined(); + expect(held.isWorking(thread("a", false))).toBe(true); + expect(held.returnedAt(thread("a", 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 }, + { pendingGuestMessageCount: 1 }, + { 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.each(["interrupted", "cancelled", "rolled_back"] as const)( + "holds a %s run, because the server can still settle it", + (status) => { + const tracker = createInboxReturnTracker(); + const stopped = { + ...thread("a", false), + latestRun: { + runId: RunId.make("stopped"), + status, + requestedAt: null, + startedAt: null, + completedAt: null, + assistantMessageId: null, + }, + }; + tracker.observe([thread("a", true)]); + tracker.observe([stopped]); + expect(tracker.isWorking(stopped)).toBe(true); + tracker.observe([{ ...stopped, settledOverride: "settled" as const }]); + expect(tracker.isWorking(stopped)).toBe(false); + }, + ); + + 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(); }); }); diff --git a/packages/client-runtime/src/state/threadInbox.ts b/packages/client-runtime/src/state/threadInbox.ts index 8083dd045385b..de9cd822398d0 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,46 +88,154 @@ function sortNewestFirst + >; + +type ThreadKeyInput = Pick; + +/** + * 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 one at module scope so the inbox order survives routes that unmount the - * list. Call `observe` with every thread shell whenever the shells change, or - * with null to reset while the beta is off. It returns the return times as of - * that observation, which stay fixed so a list built from them can be cached. + * Keep one at module scope so placement survives routes that unmount the list. Call + * `observe` when the shells or the snapshot version change, or with null while the beta + * is off. Its result stays fixed, so a list built from it can be cached. */ export function createInboxReturnTracker() { - const keyOf = (thread: Pick) => - `${thread.environmentId}:${thread.id}`; - let lastWorkingKeys: ReadonlySet | null = null; + const keyOf = (thread: ThreadKeyInput) => `${thread.environmentId}:${thread.id}`; + 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 - The grace deadline is wall-clock time shared by web and native clients. + 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, - ): (thread: Pick) => number | undefined { + subscribe(listener: () => void) { + listeners.add(listener); + return () => { + listeners.delete(listener); + }; + }, + getSnapshot: () => version, + observe(threads: readonly TrackedInboxThread[] | null): { + readonly isWorking: (thread: WorkingThreadInput & ThreadKeyInput) => boolean; + readonly returnedAt: (thread: ThreadKeyInput) => number | undefined; + } { if (threads === null) { - lastWorkingKeys = null; + clearTimeout(timer); + timer = undefined; + states.clear(); returns.clear(); - return () => undefined; + return { isWorking: isThreadWorking, returnedAt: () => undefined }; } - const working = new Set(); const present = new Set(); + const at = now(); + // @effect-diagnostics-next-line globalDate:off - Snooze checks compare wall-clock ISO times. + 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.pendingGuestMessageCount ?? 0) > 0 || + (thread.interactionMode === "plan" && thread.hasActionableProposedPlan) || + thread.runtime?.lastError != null || + thread.latestRun?.status === "failed"; + 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 }); } - // @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; - const observed = new Map(returns); - return (thread) => observed.get(keyOf(thread)); + schedule(); + const observedWorking = new Map([...states].map(([key, state]) => [key, state.working])); + const observedReturns = new Map(returns); + return { + isWorking: (thread) => observedWorking.get(keyOf(thread)) ?? isThreadWorking(thread), + returnedAt: (thread) => observedReturns.get(keyOf(thread)), + }; }, - returnedAt: (thread: Pick) => - returns.get(keyOf(thread)), + isWorking: (thread: WorkingThreadInput & ThreadKeyInput) => + states.get(keyOf(thread))?.working ?? isThreadWorking(thread), + returnedAt: (thread: ThreadKeyInput) => returns.get(keyOf(thread)), }; }