diff --git a/apps/mobile/src/state/threads.ts b/apps/mobile/src/state/threads.ts index d6e784377cca6..d99538e833a54 100644 --- a/apps/mobile/src/state/threads.ts +++ b/apps/mobile/src/state/threads.ts @@ -21,6 +21,7 @@ import { pendingThreadCreationOutcomesAtom, selectThreadCreationDetail, } from "./pending-thread-creation"; +import { orchestrationEnvironment } from "./orchestration"; import { serverEnvironment } from "./server"; import { environmentShell } from "./shell"; import { threadOutboxManager } from "./thread-outbox"; @@ -45,6 +46,9 @@ export const environmentThreadShells = createEnvironmentThreadShellAtoms({ catalogValueAtom: environmentCatalog.catalogValueAtom, snapshotAtom: threadEnvironment.snapshotAtom, hiddenThreadTagsAtom: serverEnvironment.hiddenThreadTagsAtom, + archivedSnapshotAtom: (environmentId) => + orchestrationEnvironment.archivedShellSnapshot({ environmentId, input: {} }), + shellStateAtom: environmentShell.stateAtom, }); export const childThreadApprovalsAtom = createChildThreadApprovalsAtom({ childThreadInputsAtom: environmentThreadShells.childThreadInputsAtom, diff --git a/apps/mobile/src/state/use-composer-drafts.test.ts b/apps/mobile/src/state/use-composer-drafts.test.ts index f8a14977acca9..a9250969bf8ae 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( catalogValueAtom, snapshotAtom: () => snapshotAtom, hiddenThreadTagsAtom: () => Atom.make(null), + archivedSnapshotAtom: () => Atom.make(AsyncResult.initial()), + shellStateAtom: () => shellResultAtom, }); const selectedShell = shells.threadShellAtom({ environmentId, threadId: selectedThreadId }); onTestFinished(registry.mount(selectedShell)); diff --git a/apps/web/src/composerDraftStore.test.ts b/apps/web/src/composerDraftStore.test.ts index 547d21d5eaf07..728035618b6d6 100644 --- a/apps/web/src/composerDraftStore.test.ts +++ b/apps/web/src/composerDraftStore.test.ts @@ -1353,6 +1353,8 @@ function modelReconciliationHarness( catalogValueAtom, snapshotAtom: () => visibleSnapshotAtom, hiddenThreadTagsAtom: () => Atom.make(null), + archivedSnapshotAtom: () => Atom.make(AsyncResult.initial()), + shellStateAtom: () => shellResultAtom, }); const selectedShell = shells.threadShellAtom({ environmentId, threadId: selectedThreadId }); onTestFinished(registry.mount(selectedShell)); diff --git a/apps/web/src/state/threads.ts b/apps/web/src/state/threads.ts index 6d2ce2a07c90b..228e892c40ca0 100644 --- a/apps/web/src/state/threads.ts +++ b/apps/web/src/state/threads.ts @@ -18,6 +18,7 @@ import { AsyncResult, Atom } from "effect/reactivity"; import { environmentCatalog } from "../connection/catalog"; import { connectionAtomRuntime } from "../connection/runtime"; +import { orchestrationEnvironment } from "./orchestration"; import { serverEnvironment } from "./server"; import { environmentShell } from "./shell"; @@ -41,6 +42,9 @@ export const environmentThreadShells = createEnvironmentThreadShellAtoms({ catalogValueAtom: environmentCatalog.catalogValueAtom, snapshotAtom: threadEnvironment.snapshotAtom, hiddenThreadTagsAtom: serverEnvironment.hiddenThreadTagsAtom, + archivedSnapshotAtom: (environmentId) => + orchestrationEnvironment.archivedShellSnapshot({ environmentId, input: {} }), + shellStateAtom: environmentShell.stateAtom, }); export const childThreadApprovalsAtom = createChildThreadApprovalsAtom({ childThreadInputsAtom: environmentThreadShells.childThreadInputsAtom, diff --git a/packages/client-runtime/src/state/threadShell.test.ts b/packages/client-runtime/src/state/threadShell.test.ts index 39607575627dd..8e0e114fe056d 100644 --- a/packages/client-runtime/src/state/threadShell.test.ts +++ b/packages/client-runtime/src/state/threadShell.test.ts @@ -4,28 +4,53 @@ import { RunId, RuntimeRequestId, ThreadId, + type OrchestrationV2ArchivedShellSnapshot, type OrchestrationV2ShellSnapshot, type OrchestrationV2ThreadShell, } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; import * as Option from "effect/Option"; -import { Atom, AtomRegistry } from "effect/reactivity"; +import { AsyncResult, Atom, AtomRegistry } from "effect/reactivity"; 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 { EMPTY_SHELL_STATE, type EnvironmentShellState } from "./shell.ts"; import { isThreadWorking } from "./threadInbox.ts"; import { childThreadInputsTitle, createEnvironmentThreadShellAtoms } from "./threadShell.ts"; const environmentId = EnvironmentId.make("environment-v2"); const remoteEnvironmentId = EnvironmentId.make("remote-environment-v2"); const otherProjectId = ProjectId.make("other-project"); +const emptyArchive: OrchestrationV2ArchivedShellSnapshot = { + schemaVersion: 1, + snapshotSequence: 0, + projects: [], + threads: [], +}; function makeHarness(environmentIds: ReadonlyArray = [environmentId]) { const snapshotAtom = Atom.family((_environmentId: EnvironmentId) => Atom.make(v2ShellSnapshot), ); + type ArchiveResult = AsyncResult.AsyncResult; + const archiveRefreshes: Array = []; + const archivedSnapshotAtom = Atom.family((archiveEnvironmentId: EnvironmentId) => + Atom.writable( + (get): ArchiveResult => + Option.getOrElse(get.self(), () => AsyncResult.success(emptyArchive)), + (ctx, value: ArchiveResult) => ctx.setSelf(value), + () => { + archiveRefreshes.push(archiveEnvironmentId); + }, + ), + ); + const shellStateAtom = Atom.family((_environmentId: EnvironmentId) => + Atom.make>( + AsyncResult.success(EMPTY_SHELL_STATE), + ), + ); const catalogValueAtom = Atom.make({ isReady: true, entries: new Map( @@ -47,11 +72,16 @@ function makeHarness(environmentIds: ReadonlyArray = [environment return { registry: AtomRegistry.make(), snapshotAtom, + archivedSnapshotAtom, + archiveRefreshes, + shellStateAtom, catalogValueAtom, threads: createEnvironmentThreadShellAtoms({ catalogValueAtom, snapshotAtom, hiddenThreadTagsAtom: () => Atom.make(null), + archivedSnapshotAtom, + shellStateAtom, }), }; } @@ -314,10 +344,15 @@ describe("v2 thread shell lists", () => { }); it("lists a task a linked environment runs for a listed thread under that thread", () => { - const { registry, threads, snapshotAtom, catalogValueAtom } = makeHarness([ - environmentId, - remoteEnvironmentId, - ]); + const { + registry, + threads, + snapshotAtom, + archivedSnapshotAtom, + archiveRefreshes, + shellStateAtom, + catalogValueAtom, + } = makeHarness([environmentId, remoteEnvironmentId]); const parent = v2ThreadShell; const task = { ...v2ThreadShell, @@ -380,16 +415,65 @@ describe("v2 thread shell lists", () => { expect(registry.get(threads.threadShellAtom(parentRef))?.pendingSubagentRequest).toBeNull(); expect(registry.get(inputs).map((thread) => thread.id)).toEqual([helper.id]); + const archive = ( + snapshotSequence: number, + archived: ReadonlyArray, + ) => + registry.set( + archivedSnapshotAtom(environmentId), + AsyncResult.success({ ...emptyArchive, snapshotSequence, threads: archived }), + ); // A task whose parent is gone there is listed on its own, as is one whose - // parent's environment is switched off. + // parent's environment is switched off. An archive request made before the parent went + // missing may predate its archive, so the archive is read again first. + registry.set(archivedSnapshotAtom(environmentId), AsyncResult.initial(true)); registry.set(snapshotAtom(environmentId), { ...v2ShellSnapshot, threads: [] }); + expect(listed()).toEqual([]); + expect(archiveRefreshes).toEqual([environmentId]); + archive(0, []); + expect(listed()).toEqual([`${remoteEnvironmentId}/${task.id}`]); + // Archiving the parent removes it from the live snapshot. Its task stays under it, as a + // local subagent does, while the archive is older than the removal or still loading. + archive(10, []); + registry.set( + snapshotAtom(environmentId), + applyShellStreamEvent( + { ...v2ShellSnapshot, snapshotSequence: 10, threads: [parent] }, + { kind: "thread.removed", location: "active", sequence: 11, threadId: parent.id }, + ), + ); + registry.set( + shellStateAtom(environmentId), + AsyncResult.success({ ...EMPTY_SHELL_STATE, removedAt: new Map([[parent.id, 11]]) }), + ); + expect(listed()).toEqual([]); + expect(archiveRefreshes).toHaveLength(2); + registry.set(archivedSnapshotAtom(environmentId), AsyncResult.initial(true)); + expect(listed()).toEqual([]); + archive(12, [{ ...parent, archivedAt: v2Now }]); + expect(listed()).toEqual([]); + // Another client archived the parent after this archive was read, and a reconnect's snapshot + // then cleared the removal. The task stays under the parent while the archive is read again. + archive(10, []); + registry.set( + shellStateAtom(environmentId), + AsyncResult.success({ ...EMPTY_SHELL_STATE, generation: 1 }), + ); + expect(listed()).toEqual([]); + expect(archiveRefreshes).toHaveLength(3); + archive(13, [{ ...parent, archivedAt: v2Now }]); + expect(listed()).toEqual([]); + // A current archive without the parent means it was deleted, and an unreadable archive + // cannot vouch for it, so the task is listed either way. + archive(12, []); + expect(listed()).toEqual([`${remoteEnvironmentId}/${task.id}`]); + registry.set(archivedSnapshotAtom(environmentId), AsyncResult.fail("unavailable")); expect(listed()).toEqual([`${remoteEnvironmentId}/${task.id}`]); - // Archiving the parent here leaves the task there unarchived, so it is listed on its own. registry.set(snapshotAtom(environmentId), { ...v2ShellSnapshot, - threads: [{ ...parent, archivedAt: v2Now }], + threads: [{ ...parent, settledOverride: "settled" as const, settledAt: v2Now }], }); - expect(listed()).toEqual([`${remoteEnvironmentId}/${task.id}`]); + expect(listed()).toEqual([`${environmentId}/${parent.id}`]); registry.set(snapshotAtom(environmentId), { ...v2ShellSnapshot, threads: [parent] }); expect(listed()).toEqual([`${environmentId}/${parent.id}`]); const catalog = registry.get(catalogValueAtom); diff --git a/packages/client-runtime/src/state/threadShell.ts b/packages/client-runtime/src/state/threadShell.ts index f5ced761c5665..875f172d2b243 100644 --- a/packages/client-runtime/src/state/threadShell.ts +++ b/packages/client-runtime/src/state/threadShell.ts @@ -1,5 +1,6 @@ import type { EnvironmentId, + OrchestrationV2ArchivedShellSnapshot, OrchestrationV2ShellSnapshot, ProjectId, ScopedProjectRef, @@ -12,10 +13,11 @@ import { OrchestrationV2ThreadShell, } from "@t3tools/contracts"; import * as Schema from "effect/Schema"; -import { Atom } from "effect/reactivity"; +import { AsyncResult, Atom, type AtomRegistry } from "effect/reactivity"; import type { EnvironmentThreadShell, PendingSubagentRequest } from "./models.ts"; import { presentThreadShell, threadRuntimeIsActive } from "./models.ts"; +import type { EnvironmentShellState } from "./shell.ts"; import { hiddenThreadKeys, NO_HIDDEN_THREADS } from "./threadTagFilter.ts"; import { type EnvironmentCatalogState, enabledEnvironmentIds } from "./connections.ts"; import { @@ -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 or has archived (`delegatedUnderParent`, by thread key). A task whose parent + * is gone stays in the lists, so its work is never out of reach. */ export function isShownUnderParent( ref: ScopedThreadRef, @@ -209,6 +211,14 @@ export function createEnvironmentThreadShellAtoms(input: { readonly hiddenThreadTagsAtom: ( environmentId: EnvironmentId, ) => Atom.Atom | null>; + /** Archived threads, which the live snapshot leaves out. */ + readonly archivedSnapshotAtom: ( + environmentId: EnvironmentId, + ) => Atom.Atom>; + /** The shell state, whose removals tell a stale archive from one that lacks a thread. */ + readonly shellStateAtom: ( + environmentId: EnvironmentId, + ) => Atom.Atom>; }) { // Point reads and aggregate lists share values without keeping an atom alive // for every listed thread. Replaced source objects can be collected. @@ -663,19 +673,74 @@ export function createEnvironmentThreadShellAtoms(input: { return next; }).pipe(Atom.withLabel("environment-quiet-thread-shell-list")); + // The archive has no live updates, and an answer cached or requested before this atom first read + // it may predate the parent's archive. The first read, a removal, and a new authoritative shell + // snapshot each reread it, and the answer from before counts as loading. Kept outside the family + // so a swept node still sees a change it missed. + const archiveReads = new Map< + EnvironmentId, + WeakMap< + AtomRegistry.AtomRegistry, + { + readonly generation: number; + readonly removedAt: ReadonlyMap; + readonly stale: unknown; + } + > + >(); + // Null while the archived page has not arrived. A failed read counts as empty and current. + const archivedThreadIdsAtom = Atom.family((environmentId: EnvironmentId) => + Atom.make((get): { readonly ids: ReadonlySet; readonly sequence: number } | null => { + const archiveAtom = input.archivedSnapshotAtom(environmentId); + const result = get(archiveAtom); + const shell = AsyncResult.getOrElse(get(input.shellStateAtom(environmentId)), () => null); + const reads = archiveReads.get(environmentId) ?? new WeakMap(); + archiveReads.set(environmentId, reads); + const previous = reads.get(get.registry); + const generation = shell?.generation ?? previous?.generation ?? 0; + const removedAt = shell?.removedAt ?? previous?.removedAt ?? new Map(); + if ( + previous === undefined || + previous.generation !== generation || + previous.removedAt !== removedAt + ) { + reads.set(get.registry, { generation, removedAt, stale: result }); + get.refresh(archiveAtom); + return null; + } + const stale = previous?.stale === result; + reads.set(get.registry, { generation, removedAt, stale: stale ? result : undefined }); + if (stale) return null; + if (result.waiting || AsyncResult.isInitial(result)) return null; + if (!AsyncResult.isSuccess(result)) return { ids: new Set(), sequence: Infinity }; + const { threads, snapshotSequence } = result.value; + return { ids: new Set(threads.map((t) => t.id)), sequence: snapshotSequence }; + }).pipe(Atom.withLabel(`environment-archived-thread-ids:${environmentId}`)), + ); + let previousDelegatedUnderParent: ReadonlySet = new Set(); const delegatedUnderParentAtom = Atom.make((get) => { const environmentIds = new Set(enabledEnvironmentIds(get(input.catalogValueAtom))); const next = new Set(); + // An archived parent keeps its task, as it keeps a local subagent. The live snapshot drops + // archived threads, so a missing parent waits on an archived page read after its removal. + const parentKeepsTask = (parent: ScopedThreadRef) => { + if (get(environmentThreadIndexAtom(parent.environmentId)).has(parent.threadId)) return true; + const archived = get(archivedThreadIdsAtom(parent.environmentId)); + if (archived === null || archived.ids.has(parent.threadId)) return true; + const shell = AsyncResult.getOrElse( + get(input.shellStateAtom(parent.environmentId)), + () => null, + ); + return (shell?.removedAt.get(parent.threadId) ?? -1) > archived.sequence; + }; for (const environmentId of environmentIds) { 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 + parentKeepsTask(parent) ) { next.add(threadKey({ environmentId, threadId: thread.id })); }