diff --git a/apps/mobile/src/state/threads.ts b/apps/mobile/src/state/threads.ts index bf2013aef5398..0199523f27d76 100644 --- a/apps/mobile/src/state/threads.ts +++ b/apps/mobile/src/state/threads.ts @@ -45,6 +45,8 @@ export const environmentThreadDetails = createEnvironmentThreadDetailAtoms( export const environmentThreadShells = createEnvironmentThreadShellAtoms({ catalogValueAtom: environmentCatalog.catalogValueAtom, snapshotAtom: threadEnvironment.snapshotAtom, + shellStatusAtom: environmentShell.statusAtom, + removedAtAtom: environmentShell.removedAtAtom, hiddenThreadTagsAtom: serverEnvironment.hiddenThreadTagsAtom, }); const steadyShells = createSteadyShellAtoms({ @@ -62,6 +64,8 @@ export const steadyShellsHoldingAtom = steadyShells.holdingAtom; export const steadyNavigationThreadShellsAtom = createEnvironmentThreadShellAtoms({ catalogValueAtom: environmentCatalog.catalogValueAtom, snapshotAtom: steadyShells.snapshotAtom, + shellStatusAtom: environmentShell.statusAtom, + removedAtAtom: environmentShell.removedAtAtom, hiddenThreadTagsAtom: serverEnvironment.hiddenThreadTagsAtom, }).navigationThreadShellsAtom; export const childThreadApprovalsAtom = createChildThreadApprovalsAtom({ diff --git a/apps/mobile/src/state/use-composer-drafts.test.ts b/apps/mobile/src/state/use-composer-drafts.test.ts index f25037f40f0b4..1290907b95caf 100644 --- a/apps/mobile/src/state/use-composer-drafts.test.ts +++ b/apps/mobile/src/state/use-composer-drafts.test.ts @@ -315,6 +315,8 @@ function modelReconciliationHarness( const shells = createEnvironmentThreadShellAtoms({ catalogValueAtom, snapshotAtom: () => snapshotAtom, + shellStatusAtom: () => Atom.make("live" as const), + removedAtAtom: () => Atom.make(new Map()), hiddenThreadTagsAtom: () => Atom.make(null), }); const selectedShell = shells.threadShellAtom({ environmentId, threadId: selectedThreadId }); diff --git a/apps/web/src/components/ThreadNotificationCoordinator.badge.browser.test.ts b/apps/web/src/components/ThreadNotificationCoordinator.badge.browser.test.ts index 6c6986edf451e..9242c2500c8de 100644 --- a/apps/web/src/components/ThreadNotificationCoordinator.badge.browser.test.ts +++ b/apps/web/src/components/ThreadNotificationCoordinator.badge.browser.test.ts @@ -60,7 +60,7 @@ vi.mock(import("../state/threads"), async (original) => { environmentThreadShells: { ...actual.environmentThreadShells, pendingSubagentRequestsAtom: (() => noRequestsAtom) as never, - get delegatedUnderParentAtom() { + get delegatedUnderListedParentAtom() { return noDelegatedAtom as never; }, }, diff --git a/apps/web/src/components/ThreadNotificationCoordinator.tsx b/apps/web/src/components/ThreadNotificationCoordinator.tsx index 6e6b2b37b6830..67d81e7c3b407 100644 --- a/apps/web/src/components/ThreadNotificationCoordinator.tsx +++ b/apps/web/src/components/ThreadNotificationCoordinator.tsx @@ -240,7 +240,9 @@ export function startThreadNotifications( const subagentRequests = registry.get( environmentThreadShells.pendingSubagentRequestsAtom(environmentId), ); - const delegatedUnderParent = registry.get(environmentThreadShells.delegatedUnderParentAtom); + const delegatedUnderParent = registry.get( + environmentThreadShells.delegatedUnderListedParentAtom, + ); const changed: Array<[NotificationState | undefined, NotificationState]> = []; for (const rawThread of threads) { if ( @@ -343,7 +345,7 @@ export function startThreadNotifications( reobserve, ); const unsubscribeDelegated = registry.subscribe( - environmentThreadShells.delegatedUnderParentAtom, + environmentThreadShells.delegatedUnderListedParentAtom, reobserve, ); entry.unsubscribe = () => { diff --git a/apps/web/src/composerDraftStore.test.ts b/apps/web/src/composerDraftStore.test.ts index c698f96e92e39..ec2c8eb89e4e0 100644 --- a/apps/web/src/composerDraftStore.test.ts +++ b/apps/web/src/composerDraftStore.test.ts @@ -1386,6 +1386,8 @@ function modelReconciliationHarness( const shells = createEnvironmentThreadShellAtoms({ catalogValueAtom, snapshotAtom: () => visibleSnapshotAtom, + shellStatusAtom: () => Atom.make("live" as const), + removedAtAtom: () => Atom.make(new Map()), hiddenThreadTagsAtom: () => Atom.make(null), }); const selectedShell = shells.threadShellAtom({ environmentId, threadId: selectedThreadId }); diff --git a/apps/web/src/state/threads.ts b/apps/web/src/state/threads.ts index 6d2ce2a07c90b..34170cb979dac 100644 --- a/apps/web/src/state/threads.ts +++ b/apps/web/src/state/threads.ts @@ -40,6 +40,8 @@ export const environmentThreadDetails = createEnvironmentThreadDetailAtoms( export const environmentThreadShells = createEnvironmentThreadShellAtoms({ catalogValueAtom: environmentCatalog.catalogValueAtom, snapshotAtom: threadEnvironment.snapshotAtom, + shellStatusAtom: environmentShell.statusAtom, + removedAtAtom: environmentShell.removedAtAtom, hiddenThreadTagsAtom: serverEnvironment.hiddenThreadTagsAtom, }); export const childThreadApprovalsAtom = createChildThreadApprovalsAtom({ diff --git a/packages/client-runtime/src/state/shell-sync.test.ts b/packages/client-runtime/src/state/shell-sync.test.ts index 96a5b59c5d743..1f6560ce23c09 100644 --- a/packages/client-runtime/src/state/shell-sync.test.ts +++ b/packages/client-runtime/src/state/shell-sync.test.ts @@ -242,7 +242,7 @@ describe("environment shell synchronization", () => { const removedObserved = yield* SubscriptionRef.changes(shellState).pipe( Stream.filter( - (state) => Option.isSome(state.snapshot) && state.snapshot.value.snapshotSequence === 52, + (state) => Option.isSome(state.snapshot) && state.snapshot.value.snapshotSequence === 53, ), Stream.runHead, Effect.forkScoped({ startImmediately: true }), @@ -261,10 +261,23 @@ describe("environment shell synchronization", () => { location: "active", threadId: ThreadId.make("thread-1"), }, + { + kind: "thread.updated", + sequence: 53, + location: "archive", + thread: { + ...v2ThreadShell, + id: ThreadId.make("thread-1"), + archivedAt: v2ThreadShell.updatedAt, + }, + }, ]); const removedState = Option.getOrThrow(yield* Fiber.join(removedObserved)); expect(removedState.generation).toBe(states.at(-1)!.generation); - expect([...removedState.removedAt]).toEqual([[ThreadId.make("thread-0"), 52]]); + expect([...removedState.removedAt]).toEqual([ + [ThreadId.make("thread-0"), 52], + [ThreadId.make("thread-1"), 53], + ]); // An authoritative snapshot resets even a higher live cursor in the same batch. const resetRow = { ...v2ThreadShell, id: ThreadId.make("reset-row") }; diff --git a/packages/client-runtime/src/state/shell.ts b/packages/client-runtime/src/state/shell.ts index 86584c4a03faa..8ca9f1e6be8a9 100644 --- a/packages/client-runtime/src/state/shell.ts +++ b/packages/client-runtime/src/state/shell.ts @@ -328,6 +328,11 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") removedAt.set(delta.threadId, delta.sequence); break; case "thread.updated": + if (delta.location === "archive") { + removedAt ??= new Map(next.removedAt); + removedAt.set(delta.thread.id, delta.sequence); + } + break; case "project.updated": case "project.removed": break; @@ -838,6 +843,11 @@ function createEnvironmentShellProgressAtoms( Atom.withLabel(`environment-shell-status:${environmentId}`), ), ), + removedAtAtom: Atom.family((environmentId: EnvironmentId) => + Atom.make((get) => get(stateValueAtom(environmentId)).removedAt).pipe( + Atom.withLabel(`environment-shell-removed-at:${environmentId}`), + ), + ), hasSnapshotAtom: Atom.family((environmentId: EnvironmentId) => Atom.make((get) => Option.isSome(get(stateValueAtom(environmentId)).snapshot)).pipe( Atom.withLabel(`environment-shell-has-snapshot:${environmentId}`), @@ -867,12 +877,14 @@ export function createEnvironmentShellAtoms( ).pipe(Atom.withLabel(`environment-shell-state-value:${environmentId}`)), ); - const { statusAtom, hasSnapshotAtom } = createEnvironmentShellProgressAtoms(stateValueAtom); + const { statusAtom, removedAtAtom, hasSnapshotAtom } = + createEnvironmentShellProgressAtoms(stateValueAtom); return { stateAtom, stateValueAtom, statusAtom, + removedAtAtom, hasSnapshotAtom, }; } diff --git a/packages/client-runtime/src/state/threadShell.test.ts b/packages/client-runtime/src/state/threadShell.test.ts index d18a3e3adae1c..b79fa67cf3730 100644 --- a/packages/client-runtime/src/state/threadShell.test.ts +++ b/packages/client-runtime/src/state/threadShell.test.ts @@ -15,6 +15,7 @@ import { describe, expect, it } from "vite-plus/test"; import { PrimaryConnectionTarget } from "../connection/model.ts"; import { v2Now, v2ShellSnapshot, v2ThreadShell } from "./orchestrationV2TestFixtures.ts"; import { applyShellStreamEvent } from "./shellReducer.ts"; +import type { EnvironmentShellStatus } from "./shell.ts"; import { isThreadWorking } from "./threadInbox.ts"; import { childThreadInputsTitle, createEnvironmentThreadShellAtoms } from "./threadShell.ts"; @@ -44,13 +45,23 @@ function makeHarness(environmentIds: ReadonlyArray = [environment ]), ), }); + const shellStatusAtom = Atom.family((_environmentId: EnvironmentId) => + Atom.make("live"), + ); + const removedAtAtom = Atom.family((_environmentId: EnvironmentId) => + Atom.make>(new Map()), + ); return { registry: AtomRegistry.make(), snapshotAtom, catalogValueAtom, + shellStatusAtom, + removedAtAtom, threads: createEnvironmentThreadShellAtoms({ catalogValueAtom, snapshotAtom, + shellStatusAtom, + removedAtAtom, hiddenThreadTagsAtom: () => Atom.make(null), }), }; @@ -448,6 +459,146 @@ describe("v2 thread shell lists", () => { registry.dispose(); }); + it("keeps quiet completed tasks grouped while their missing parent's shell is unavailable", () => { + const { registry, threads, snapshotAtom, catalogValueAtom, shellStatusAtom, removedAtAtom } = + makeHarness([environmentId, remoteEnvironmentId]); + const parent = { ...v2ThreadShell, settledOverride: "settled" as const, settledAt: v2Now }; + const task: OrchestrationV2ThreadShell = { + ...v2ThreadShell, + id: ThreadId.make("completed-remote-task"), + status: "completed", + activityRunStatus: null, + latestRunId: RunId.make("completed-run"), + delegatedFrom: { environmentId, threadId: parent.id, title: parent.title }, + }; + registry.set(snapshotAtom(remoteEnvironmentId), { ...v2ShellSnapshot, threads: [task] }); + const unmount = registry.mount(threads.navigationThreadShellsAtom); + const listed = () => + registry.get(threads.navigationThreadShellsAtom).map((thread) => thread.id); + + for (const status of ["empty", "cached", "synchronizing"] as const) { + registry.set(shellStatusAtom(environmentId), status); + registry.set( + snapshotAtom(environmentId), + status === "empty" ? null : { ...v2ShellSnapshot, threads: [] }, + ); + expect(listed()).toEqual([]); + } + // No listed parent speaks for the task, so its completion still notifies. + expect(registry.get(threads.delegatedUnderListedParentAtom).size).toBe(0); + expect( + registry.get( + threads.threadShellAtom({ environmentId: remoteEnvironmentId, threadId: task.id }), + )?.settledOverride, + ).toBeNull(); + + const approval = { + id: RuntimeRequestId.make("approval"), + kind: "command" as const, + createdAt: v2Now, + }; + for (const change of [ + { status: "running" as const }, + { status: "failed" as const }, + { activeRunId: RunId.make("continuation") }, + { activityRunStatus: "waiting" as const }, + { pendingRuntimeRequest: approval }, + { pendingRuntimeRequest: { ...approval, kind: "user_input" as const } }, + { pendingGuestMessageCount: 1 }, + { pendingBackgroundTasks: [{ taskId: "continuation", kind: "monitor" as const }] }, + { lastError: "Continuation failed" }, + { hasActionableProposedPlan: true }, + { pinnedAt: v2Now }, + { settledOverride: "active" as const }, + ]) { + registry.set(snapshotAtom(remoteEnvironmentId), { + ...v2ShellSnapshot, + threads: [{ ...task, ...change }], + }); + expect(listed()).toEqual([task.id]); + } + const helper = { + ...task, + id: ThreadId.make("asking-helper"), + status: "running" as const, + activityRunStatus: "running" as const, + delegatedFrom: undefined, + lineage: { + ...task.lineage, + parentThreadId: task.id, + relationshipToParent: "subagent" as const, + }, + pendingRuntimeRequest: approval, + }; + registry.set(snapshotAtom(remoteEnvironmentId), { + ...v2ShellSnapshot, + threads: [task, helper], + }); + expect(listed()).toEqual([task.id]); + for (const descendant of [ + { + ...helper, + status: "completed" as const, + activityRunStatus: null, + pendingRuntimeRequest: null, + pendingBackgroundTasks: [{ taskId: "awaited-command", kind: "monitor" as const }], + }, + { + ...task, + id: ThreadId.make("nested-task"), + delegatedFrom: { environmentId: remoteEnvironmentId, threadId: task.id, title: task.title }, + }, + ]) { + registry.set(snapshotAtom(remoteEnvironmentId), { + ...v2ShellSnapshot, + threads: [task, descendant], + }); + expect(listed()).toEqual([task.id]); + } + registry.set(snapshotAtom(remoteEnvironmentId), { ...v2ShellSnapshot, threads: [task] }); + expect(listed()).toEqual([]); + registry.set(snapshotAtom(remoteEnvironmentId), { + ...v2ShellSnapshot, + threads: [{ ...task, activityRunStatus: undefined }], + }); + expect(listed()).toEqual([]); + + registry.set(shellStatusAtom(environmentId), "live"); + expect(listed()).toEqual([task.id]); + registry.set(snapshotAtom(environmentId), { ...v2ShellSnapshot, threads: [parent] }); + expect(listed()).toEqual([parent.id]); + registry.set(shellStatusAtom(environmentId), "cached"); + expect(listed()).toEqual([parent.id]); + for (const event of [ + { + kind: "thread.updated", + location: "archive", + sequence: 1, + thread: { ...parent, archivedAt: v2Now }, + }, + { kind: "thread.removed", location: "active", sequence: 1, threadId: parent.id }, + ] as const) { + registry.set(shellStatusAtom(environmentId), "live"); + registry.set( + snapshotAtom(environmentId), + applyShellStreamEvent({ ...v2ShellSnapshot, threads: [parent] }, event), + ); + registry.set(removedAtAtom(environmentId), new Map([[parent.id, 1]])); + registry.set(shellStatusAtom(environmentId), "cached"); + expect(listed()).toEqual([task.id]); + } + registry.set(removedAtAtom(environmentId), new Map()); + registry.set(snapshotAtom(environmentId), null); + expect(listed()).toEqual([]); + const catalog = registry.get(catalogValueAtom); + const entries = new Map(catalog.entries); + entries.set(environmentId, { ...entries.get(environmentId)!, enabled: false }); + registry.set(catalogValueAtom, { ...catalog, entries }); + expect(listed()).toEqual([task.id]); + unmount(); + registry.dispose(); + }); + it("brings an approval up a chain of tasks across three environments to its root", () => { const thirdEnvironmentId = EnvironmentId.make("third-environment-v2"); const { registry, threads, snapshotAtom } = makeHarness([ diff --git a/packages/client-runtime/src/state/threadShell.ts b/packages/client-runtime/src/state/threadShell.ts index 3aaaab30720ad..c2321edaab2d2 100644 --- a/packages/client-runtime/src/state/threadShell.ts +++ b/packages/client-runtime/src/state/threadShell.ts @@ -8,6 +8,7 @@ import type { } from "@t3tools/contracts"; import { isListedThreadLineage, + isLiveSubagentShell, isProviderNativeSubagentThread, OrchestrationV2ThreadShell, } from "@t3tools/contracts"; @@ -18,6 +19,7 @@ import type { EnvironmentThreadShell, PendingSubagentRequest } from "./models.ts import { presentThreadShell, threadRuntimeIsActive } from "./models.ts"; import { hiddenThreadKeys, NO_HIDDEN_THREADS } from "./threadTagFilter.ts"; import { type EnvironmentCatalogState, enabledEnvironmentIds } from "./connections.ts"; +import type { EnvironmentShellStatus } from "./shell.ts"; import { arrayElementsEqual, parseProjectRefCollectionKey, @@ -58,8 +60,8 @@ const isChildRunLive = ( /** * Whether a thread shows under its parent rather than in thread lists: a * subagent, a side chat, or a task a linked environment delegated here whose parent the - * client lists (`delegatedUnderParent`, by thread key). A task whose parent - * the client cannot list stays in the lists, so its work is never out of reach. + * client lists (`delegatedUnderParent`, by thread key). Quiet completed tasks also wait + * under an unavailable parent; other tasks stay listed so their work remains reachable. */ export function isShownUnderParent( ref: ScopedThreadRef, @@ -212,6 +214,10 @@ export function createEnvironmentThreadShellAtoms(input: { readonly snapshotAtom: ( environmentId: EnvironmentId, ) => Atom.Atom; + readonly shellStatusAtom: (environmentId: EnvironmentId) => Atom.Atom; + readonly removedAtAtom: ( + environmentId: EnvironmentId, + ) => Atom.Atom>; /** Tags whose threads display lists leave out; null while the filter is off. */ readonly hiddenThreadTagsAtom: ( environmentId: EnvironmentId, @@ -674,33 +680,82 @@ export function createEnvironmentThreadShellAtoms(input: { return next; }).pipe(Atom.withLabel("environment-quiet-thread-shell-list")); - let previousDelegatedUnderParent: ReadonlySet = new Set(); - const delegatedUnderParentAtom = Atom.make((get) => { + const sameKeys = (a: ReadonlySet, b: ReadonlySet) => + a.size === b.size && [...a].every((key) => b.has(key)); + let previousDelegated = { + underListedParent: new Set() as ReadonlySet, + underParent: new Set() as ReadonlySet, + }; + const delegatedTasksAtom = Atom.make((get) => { const environmentIds = new Set(enabledEnvironmentIds(get(input.catalogValueAtom))); - const next = new Set(); + const delegators = new Set(); for (const environmentId of environmentIds) { + for (const thread of get(environmentThreadsAtom(environmentId))) { + if (thread.delegatedFrom !== undefined) delegators.add(threadKey(thread.delegatedFrom)); + } + } + const underListedParent = new Set(); + const underParent = new Set(); + for (const environmentId of environmentIds) { + let busyRoots: Set | undefined; + const isBusyLineage = (thread: OrchestrationV2ThreadShell) => { + if (busyRoots === undefined) { + busyRoots = new Set(); + for (const member of get(environmentThreadsAtom(environmentId))) { + if (member.activeRunId !== null || isLiveSubagentShell(member)) { + busyRoots.add(member.lineage.rootThreadId); + } + } + } + return busyRoots.has(thread.lineage.rootThreadId); + }; for (const thread of get(environmentThreadsAtom(environmentId))) { const parent = thread.delegatedFrom; - if ( - parent !== undefined && - environmentIds.has(parent.environmentId) && - // An archived parent leaves navigation; its task there stays listed on its own. - get(environmentThreadIndexAtom(parent.environmentId)).get(parent.threadId)?.archivedAt === - null - ) { - next.add(threadKey({ environmentId, threadId: thread.id })); + if (parent === undefined || !environmentIds.has(parent.environmentId)) continue; + const key = threadKey({ environmentId, threadId: thread.id }); + const parentThread = get(environmentThreadIndexAtom(parent.environmentId)).get( + parent.threadId, + ); + if (parentThread?.archivedAt === null) { + underListedParent.add(key); + underParent.add(key); + continue; } + // A missing parent in a cached shell does not prove its completed task is an orphan. + const awaitingParent = + parentThread === undefined && + get(input.shellStatusAtom(parent.environmentId)) !== "live" && + !get(input.removedAtAtom(parent.environmentId)).has(parent.threadId) && + thread.status === "completed" && + (thread.pendingGuestMessageCount ?? 0) === 0 && + thread.lastError == null && + !thread.hasActionableProposedPlan && + thread.pinnedAt == null && + thread.settledOverride !== "active" && + !get(pendingSubagentRequestsAtom(environmentId)).has(thread.id) && + !delegators.has(key) && + !isBusyLineage(thread); + if (awaitingParent) underParent.add(key); } } - if ( - next.size === previousDelegatedUnderParent.size && - [...next].every((key) => previousDelegatedUnderParent.has(key)) - ) { - return previousDelegatedUnderParent; - } - previousDelegatedUnderParent = next; - return next; - }).pipe(Atom.withLabel("environment-delegated-under-parent")); + previousDelegated = { + underListedParent: sameKeys(underListedParent, previousDelegated.underListedParent) + ? previousDelegated.underListedParent + : underListedParent, + underParent: sameKeys(underParent, previousDelegated.underParent) + ? previousDelegated.underParent + : underParent, + }; + return previousDelegated; + }).pipe(Atom.withLabel("environment-delegated-tasks")); + /** Tasks whose parent is listed; that parent speaks for them, so notifications skip them. */ + const delegatedUnderListedParentAtom = Atom.make( + (get) => get(delegatedTasksAtom).underListedParent, + ).pipe(Atom.withLabel("environment-delegated-under-listed-parent")); + /** Tasks lists show under their parent, including quiet ones whose parent is unavailable. */ + const delegatedUnderParentAtom = Atom.make((get) => get(delegatedTasksAtom).underParent).pipe( + Atom.withLabel("environment-delegated-under-parent"), + ); let previousNavigationShells: ReadonlyArray = []; const navigationThreadShellsAtom = Atom.make((get) => { @@ -738,6 +793,7 @@ export function createEnvironmentThreadShellAtoms(input: { quietThreadShellsAtom, navigationThreadShellsAtom, delegatedUnderParentAtom, + delegatedUnderListedParentAtom, pendingSubagentRequestsAtom, threadShellsForProjectRefsAtom: (refs: ReadonlyArray) => threadShellsForProjectRefsAtomFamily(projectRefCollectionKey(refs)),