From b1ffff3fc2bd4661053554664f75f488644b4bdc Mon Sep 17 00:00:00 2001 From: psick22 Date: Tue, 3 Nov 2026 17:30:53 +0900 Subject: [PATCH 1/6] fix(clients): archiving a thread keeps its linked-environment tasks under it A task that a linked environment runs for a thread shows under that thread. When the parent was archived, the task left its parent and appeared in the sidebar and mobile thread lists on its own. A local subagent stays under an archived parent, and the archive already lists archived tasks under their parent. The task now stays under an archived parent too. A task whose parent the client does not have is still listed on its own. Co-Authored-By: Claude Opus 5.5 --- packages/client-runtime/src/state/threadShell.test.ts | 9 +++++++-- packages/client-runtime/src/state/threadShell.ts | 9 ++++----- 2 files changed, 11 insertions(+), 7 deletions(-) diff --git a/packages/client-runtime/src/state/threadShell.test.ts b/packages/client-runtime/src/state/threadShell.test.ts index 39607575627dd..35936501fda3b 100644 --- a/packages/client-runtime/src/state/threadShell.test.ts +++ b/packages/client-runtime/src/state/threadShell.test.ts @@ -384,12 +384,17 @@ describe("v2 thread shell lists", () => { // parent's environment is switched off. registry.set(snapshotAtom(environmentId), { ...v2ShellSnapshot, threads: [] }); expect(listed()).toEqual([`${remoteEnvironmentId}/${task.id}`]); - // Archiving the parent here leaves the task there unarchived, so it is listed on its own. + // An archived parent is still in the archive, so its task stays under it, as a local subagent does. registry.set(snapshotAtom(environmentId), { ...v2ShellSnapshot, threads: [{ ...parent, archivedAt: v2Now }], }); - expect(listed()).toEqual([`${remoteEnvironmentId}/${task.id}`]); + expect(listed()).toEqual([]); + registry.set(snapshotAtom(environmentId), { + ...v2ShellSnapshot, + threads: [{ ...parent, settledOverride: "settled" as const, settledAt: v2Now }], + }); + 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..d8be0bd45908e 100644 --- a/packages/client-runtime/src/state/threadShell.ts +++ b/packages/client-runtime/src/state/threadShell.ts @@ -58,8 +58,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 has, archived or not (`delegatedUnderParent`, by thread key). A task whose parent + * the client does not have stays in the lists, so its work is never out of reach. */ export function isShownUnderParent( ref: ScopedThreadRef, @@ -673,9 +673,8 @@ export function createEnvironmentThreadShellAtoms(input: { 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 + // An archived parent keeps its task, as it keeps a local subagent. + get(environmentThreadIndexAtom(parent.environmentId)).has(parent.threadId) ) { next.add(threadKey({ environmentId, threadId: thread.id })); } From c660be075c17db1498ba28be22d7cf75fedf36f6 Mon Sep 17 00:00:00 2001 From: psick22 Date: Tue, 3 Nov 2026 19:19:38 +0900 Subject: [PATCH 2/6] fix(clients): a delegated task whose parent was archived stays under it The live shell snapshot drops archived threads, so after an archive the parent is gone from the client index and its tasks on linked environments read as orphans and show up in the sidebar on their own. A local subagent under the same parent stays hidden. When a task's parent is missing from the live snapshot, the thread shell now reads the parent environment's archived snapshot. An archived parent keeps the task under it. A parent the archive lacks, or an archive that fails to load, still lists the task so its work stays reachable. While the archive loads, the task waits under the parent instead of flashing in. Co-Authored-By: Claude Opus 5.5 --- apps/mobile/src/state/threads.ts | 3 ++ .../src/state/use-composer-drafts.test.ts | 1 + apps/web/src/composerDraftStore.test.ts | 1 + apps/web/src/state/threads.ts | 3 ++ .../src/state/threadShell.test.ts | 47 +++++++++++++++---- .../client-runtime/src/state/threadShell.ts | 30 ++++++++++-- 6 files changed, 70 insertions(+), 15 deletions(-) diff --git a/apps/mobile/src/state/threads.ts b/apps/mobile/src/state/threads.ts index d6e784377cca6..5e3efa31221e0 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,8 @@ export const environmentThreadShells = createEnvironmentThreadShellAtoms({ catalogValueAtom: environmentCatalog.catalogValueAtom, snapshotAtom: threadEnvironment.snapshotAtom, hiddenThreadTagsAtom: serverEnvironment.hiddenThreadTagsAtom, + archivedSnapshotAtom: (environmentId) => + orchestrationEnvironment.archivedShellSnapshot({ environmentId, input: {} }), }); 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 1f9345da8284e..4a1cee078ddb6 100644 --- a/apps/mobile/src/state/use-composer-drafts.test.ts +++ b/apps/mobile/src/state/use-composer-drafts.test.ts @@ -314,6 +314,7 @@ function modelReconciliationHarness( catalogValueAtom, snapshotAtom: () => snapshotAtom, hiddenThreadTagsAtom: () => Atom.make(null), + archivedSnapshotAtom: () => Atom.make(AsyncResult.initial()), }); 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 cd0e74d231aac..61adab2ddb727 100644 --- a/apps/web/src/composerDraftStore.test.ts +++ b/apps/web/src/composerDraftStore.test.ts @@ -1336,6 +1336,7 @@ function modelReconciliationHarness( catalogValueAtom, snapshotAtom: () => snapshotAtom, hiddenThreadTagsAtom: () => Atom.make(null), + archivedSnapshotAtom: () => Atom.make(AsyncResult.initial()), }); 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..71b1cb2154e27 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,8 @@ export const environmentThreadShells = createEnvironmentThreadShellAtoms({ catalogValueAtom: environmentCatalog.catalogValueAtom, snapshotAtom: threadEnvironment.snapshotAtom, hiddenThreadTagsAtom: serverEnvironment.hiddenThreadTagsAtom, + archivedSnapshotAtom: (environmentId) => + orchestrationEnvironment.archivedShellSnapshot({ environmentId, input: {} }), }); 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 35936501fda3b..da5522921ed98 100644 --- a/packages/client-runtime/src/state/threadShell.test.ts +++ b/packages/client-runtime/src/state/threadShell.test.ts @@ -4,12 +4,13 @@ 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"; @@ -21,11 +22,22 @@ import { childThreadInputsTitle, createEnvironmentThreadShellAtoms } from "./thr 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), ); + const archivedSnapshotAtom = Atom.family((_environmentId: EnvironmentId) => + Atom.make>( + AsyncResult.success(emptyArchive), + ), + ); const catalogValueAtom = Atom.make({ isReady: true, entries: new Map( @@ -47,11 +59,13 @@ function makeHarness(environmentIds: ReadonlyArray = [environment return { registry: AtomRegistry.make(), snapshotAtom, + archivedSnapshotAtom, catalogValueAtom, threads: createEnvironmentThreadShellAtoms({ catalogValueAtom, snapshotAtom, hiddenThreadTagsAtom: () => Atom.make(null), + archivedSnapshotAtom, }), }; } @@ -314,10 +328,9 @@ 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, catalogValueAtom } = makeHarness( + [environmentId, remoteEnvironmentId], + ); const parent = v2ThreadShell; const task = { ...v2ThreadShell, @@ -384,12 +397,26 @@ describe("v2 thread shell lists", () => { // parent's environment is switched off. registry.set(snapshotAtom(environmentId), { ...v2ShellSnapshot, threads: [] }); expect(listed()).toEqual([`${remoteEnvironmentId}/${task.id}`]); - // An archived parent is still in the archive, so its task stays under it, as a local subagent does. - registry.set(snapshotAtom(environmentId), { - ...v2ShellSnapshot, - threads: [{ ...parent, archivedAt: v2Now }], - }); + // Archiving the parent moves it out of the live snapshot. Its task stays under it, as a + // local subagent does, while the archive loads and once the archive has it. + const archivedParent = { ...parent, archivedAt: v2Now }; + registry.set(archivedSnapshotAtom(environmentId), AsyncResult.initial(true)); + registry.set( + snapshotAtom(environmentId), + applyShellStreamEvent( + { ...v2ShellSnapshot, threads: [parent] }, + { kind: "thread.updated", location: "archive", sequence: 1, thread: archivedParent }, + ), + ); expect(listed()).toEqual([]); + registry.set( + archivedSnapshotAtom(environmentId), + AsyncResult.success({ ...emptyArchive, threads: [archivedParent] }), + ); + expect(listed()).toEqual([]); + // An archive that cannot be read cannot vouch for the parent, so the task is listed. + registry.set(archivedSnapshotAtom(environmentId), AsyncResult.fail("unavailable")); + expect(listed()).toEqual([`${remoteEnvironmentId}/${task.id}`]); registry.set(snapshotAtom(environmentId), { ...v2ShellSnapshot, threads: [{ ...parent, settledOverride: "settled" as const, settledAt: v2Now }], diff --git a/packages/client-runtime/src/state/threadShell.ts b/packages/client-runtime/src/state/threadShell.ts index d8be0bd45908e..f4714ded67606 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,7 +13,7 @@ import { OrchestrationV2ThreadShell, } from "@t3tools/contracts"; import * as Schema from "effect/Schema"; -import { Atom } from "effect/reactivity"; +import { AsyncResult, Atom } from "effect/reactivity"; import type { EnvironmentThreadShell, PendingSubagentRequest } from "./models.ts"; import { presentThreadShell, threadRuntimeIsActive } from "./models.ts"; @@ -58,8 +59,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 has, archived or not (`delegatedUnderParent`, by thread key). A task whose parent - * the client does not have 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 +210,10 @@ 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>; }) { // Point reads and aggregate lists share values without keeping an atom alive // for every listed thread. Replaced source objects can be collected. @@ -663,18 +668,33 @@ export function createEnvironmentThreadShellAtoms(input: { return next; }).pipe(Atom.withLabel("environment-quiet-thread-shell-list")); + // Null while the archived page has not arrived. A failed read counts as empty. + const archivedThreadIdsAtom = Atom.family((environmentId: EnvironmentId) => + Atom.make((get): ReadonlySet | null => { + const result = get(input.archivedSnapshotAtom(environmentId)); + if (result.waiting || AsyncResult.isInitial(result)) return null; + return new Set(AsyncResult.isSuccess(result) ? result.value.threads.map((t) => t.id) : []); + }).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 the archived page before its task lists. + const parentKeepsTask = (parent: ScopedThreadRef) => { + if (get(environmentThreadIndexAtom(parent.environmentId)).has(parent.threadId)) return true; + const archived = get(archivedThreadIdsAtom(parent.environmentId)); + return archived === null || archived.has(parent.threadId); + }; 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 keeps its task, as it keeps a local subagent. - get(environmentThreadIndexAtom(parent.environmentId)).has(parent.threadId) + parentKeepsTask(parent) ) { next.add(threadKey({ environmentId, threadId: thread.id })); } From fdbee7d51ebe71760b242f1e6abc7863dd9c4acf Mon Sep 17 00:00:00 2001 From: psick22 Date: Tue, 3 Nov 2026 19:35:42 +0900 Subject: [PATCH 3/6] fix(clients): a cached archive no longer lists a just-archived parent's tasks If the archived snapshot was already loaded, archiving another parent left a short window before the refresh where that read still lacked the parent, so its tasks showed in the sidebar. A missing parent now counts as gone only when the archive was read at or after the shell removed it, the same rule thread families use with removedAt. The test applies the thread.removed event current servers send on archive. Co-Authored-By: Claude Opus 5.5 --- apps/mobile/src/state/threads.ts | 1 + .../src/state/use-composer-drafts.test.ts | 1 + apps/web/src/composerDraftStore.test.ts | 1 + apps/web/src/state/threads.ts | 1 + .../src/state/threadShell.test.ts | 52 ++++++++++++++----- .../client-runtime/src/state/threadShell.ts | 22 ++++++-- 6 files changed, 60 insertions(+), 18 deletions(-) diff --git a/apps/mobile/src/state/threads.ts b/apps/mobile/src/state/threads.ts index 5e3efa31221e0..d99538e833a54 100644 --- a/apps/mobile/src/state/threads.ts +++ b/apps/mobile/src/state/threads.ts @@ -48,6 +48,7 @@ export const environmentThreadShells = createEnvironmentThreadShellAtoms({ 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 4a1cee078ddb6..0742f78e85b19 100644 --- a/apps/mobile/src/state/use-composer-drafts.test.ts +++ b/apps/mobile/src/state/use-composer-drafts.test.ts @@ -315,6 +315,7 @@ function modelReconciliationHarness( 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 61adab2ddb727..45a150115890e 100644 --- a/apps/web/src/composerDraftStore.test.ts +++ b/apps/web/src/composerDraftStore.test.ts @@ -1337,6 +1337,7 @@ function modelReconciliationHarness( 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/state/threads.ts b/apps/web/src/state/threads.ts index 71b1cb2154e27..228e892c40ca0 100644 --- a/apps/web/src/state/threads.ts +++ b/apps/web/src/state/threads.ts @@ -44,6 +44,7 @@ export const environmentThreadShells = createEnvironmentThreadShellAtoms({ 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 da5522921ed98..5053dac091149 100644 --- a/packages/client-runtime/src/state/threadShell.test.ts +++ b/packages/client-runtime/src/state/threadShell.test.ts @@ -16,6 +16,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 { EMPTY_SHELL_STATE, type EnvironmentShellState } from "./shell.ts"; import { isThreadWorking } from "./threadInbox.ts"; import { childThreadInputsTitle, createEnvironmentThreadShellAtoms } from "./threadShell.ts"; @@ -38,6 +39,11 @@ function makeHarness(environmentIds: ReadonlyArray = [environment AsyncResult.success(emptyArchive), ), ); + const shellStateAtom = Atom.family((_environmentId: EnvironmentId) => + Atom.make>( + AsyncResult.success(EMPTY_SHELL_STATE), + ), + ); const catalogValueAtom = Atom.make({ isReady: true, entries: new Map( @@ -60,12 +66,14 @@ function makeHarness(environmentIds: ReadonlyArray = [environment registry: AtomRegistry.make(), snapshotAtom, archivedSnapshotAtom, + shellStateAtom, catalogValueAtom, threads: createEnvironmentThreadShellAtoms({ catalogValueAtom, snapshotAtom, hiddenThreadTagsAtom: () => Atom.make(null), archivedSnapshotAtom, + shellStateAtom, }), }; } @@ -328,9 +336,14 @@ describe("v2 thread shell lists", () => { }); it("lists a task a linked environment runs for a listed thread under that thread", () => { - const { registry, threads, snapshotAtom, archivedSnapshotAtom, catalogValueAtom } = makeHarness( - [environmentId, remoteEnvironmentId], - ); + const { + registry, + threads, + snapshotAtom, + archivedSnapshotAtom, + shellStateAtom, + catalogValueAtom, + } = makeHarness([environmentId, remoteEnvironmentId]); const parent = v2ThreadShell; const task = { ...v2ThreadShell, @@ -397,24 +410,37 @@ describe("v2 thread shell lists", () => { // parent's environment is switched off. registry.set(snapshotAtom(environmentId), { ...v2ShellSnapshot, threads: [] }); expect(listed()).toEqual([`${remoteEnvironmentId}/${task.id}`]); - // Archiving the parent moves it out of the live snapshot. Its task stays under it, as a - // local subagent does, while the archive loads and once the archive has it. - const archivedParent = { ...parent, archivedAt: v2Now }; - registry.set(archivedSnapshotAtom(environmentId), AsyncResult.initial(true)); + // 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. + const archive = ( + snapshotSequence: number, + archived: ReadonlyArray, + ) => + registry.set( + archivedSnapshotAtom(environmentId), + AsyncResult.success({ ...emptyArchive, snapshotSequence, threads: archived }), + ); + archive(10, []); registry.set( snapshotAtom(environmentId), applyShellStreamEvent( - { ...v2ShellSnapshot, threads: [parent] }, - { kind: "thread.updated", location: "archive", sequence: 1, thread: archivedParent }, + { ...v2ShellSnapshot, snapshotSequence: 10, threads: [parent] }, + { kind: "thread.removed", location: "active", sequence: 11, threadId: parent.id }, ), ); - expect(listed()).toEqual([]); registry.set( - archivedSnapshotAtom(environmentId), - AsyncResult.success({ ...emptyArchive, threads: [archivedParent] }), + shellStateAtom(environmentId), + AsyncResult.success({ ...EMPTY_SHELL_STATE, removedAt: new Map([[parent.id, 11]]) }), ); expect(listed()).toEqual([]); - // An archive that cannot be read cannot vouch for the parent, so the task is listed. + registry.set(archivedSnapshotAtom(environmentId), AsyncResult.initial(true)); + expect(listed()).toEqual([]); + archive(12, [{ ...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}`]); registry.set(snapshotAtom(environmentId), { diff --git a/packages/client-runtime/src/state/threadShell.ts b/packages/client-runtime/src/state/threadShell.ts index f4714ded67606..9413cbe18c155 100644 --- a/packages/client-runtime/src/state/threadShell.ts +++ b/packages/client-runtime/src/state/threadShell.ts @@ -17,6 +17,7 @@ import { AsyncResult, Atom } 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 { @@ -214,6 +215,10 @@ export function createEnvironmentThreadShellAtoms(input: { 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. @@ -668,12 +673,14 @@ export function createEnvironmentThreadShellAtoms(input: { return next; }).pipe(Atom.withLabel("environment-quiet-thread-shell-list")); - // Null while the archived page has not arrived. A failed read counts as empty. + // 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): ReadonlySet | null => { + Atom.make((get): { readonly ids: ReadonlySet; readonly sequence: number } | null => { const result = get(input.archivedSnapshotAtom(environmentId)); if (result.waiting || AsyncResult.isInitial(result)) return null; - return new Set(AsyncResult.isSuccess(result) ? result.value.threads.map((t) => t.id) : []); + 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}`)), ); @@ -682,11 +689,16 @@ export function createEnvironmentThreadShellAtoms(input: { 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 the archived page before its task lists. + // 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)); - return archived === null || archived.has(parent.threadId); + 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))) { From 5e4cce6a884aa7893ed038d2ec98dfba9630d9f0 Mon Sep 17 00:00:00 2001 From: psick22 Date: Wed, 4 Nov 2026 11:48:49 +0900 Subject: [PATCH 4/6] fix(clients): a stale archive is read again before a delegated task is listed The archived snapshot has no live updates. When another client archived the parent, this client kept the task hidden only while the shell held the removal. A reconnect's authoritative snapshot clears removals, and a cached archive read before the archive still lacked the parent, so the task showed up in the sidebar on its own. A removal or a new authoritative shell snapshot now rereads the archive, and the answer from before counts as loading until the new one lands. The archive is still read only while a delegated task's parent is missing. Co-Authored-By: Claude Opus 5.5 --- .../src/state/threadShell.test.ts | 27 ++++++++++++-- .../client-runtime/src/state/threadShell.ts | 36 +++++++++++++++++-- 2 files changed, 58 insertions(+), 5 deletions(-) diff --git a/packages/client-runtime/src/state/threadShell.test.ts b/packages/client-runtime/src/state/threadShell.test.ts index 5053dac091149..ffba6ed2367b9 100644 --- a/packages/client-runtime/src/state/threadShell.test.ts +++ b/packages/client-runtime/src/state/threadShell.test.ts @@ -34,9 +34,16 @@ function makeHarness(environmentIds: ReadonlyArray = [environment const snapshotAtom = Atom.family((_environmentId: EnvironmentId) => Atom.make(v2ShellSnapshot), ); - const archivedSnapshotAtom = Atom.family((_environmentId: EnvironmentId) => - Atom.make>( - AsyncResult.success(emptyArchive), + 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) => @@ -66,6 +73,7 @@ function makeHarness(environmentIds: ReadonlyArray = [environment registry: AtomRegistry.make(), snapshotAtom, archivedSnapshotAtom, + archiveRefreshes, shellStateAtom, catalogValueAtom, threads: createEnvironmentThreadShellAtoms({ @@ -341,6 +349,7 @@ describe("v2 thread shell lists", () => { threads, snapshotAtom, archivedSnapshotAtom, + archiveRefreshes, shellStateAtom, catalogValueAtom, } = makeHarness([environmentId, remoteEnvironmentId]); @@ -433,10 +442,22 @@ describe("v2 thread shell lists", () => { AsyncResult.success({ ...EMPTY_SHELL_STATE, removedAt: new Map([[parent.id, 11]]) }), ); expect(listed()).toEqual([]); + expect(archiveRefreshes).toEqual([environmentId]); 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).toEqual([environmentId, environmentId]); + 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, []); diff --git a/packages/client-runtime/src/state/threadShell.ts b/packages/client-runtime/src/state/threadShell.ts index 9413cbe18c155..a8923196c65e3 100644 --- a/packages/client-runtime/src/state/threadShell.ts +++ b/packages/client-runtime/src/state/threadShell.ts @@ -13,7 +13,7 @@ import { OrchestrationV2ThreadShell, } from "@t3tools/contracts"; import * as Schema from "effect/Schema"; -import { AsyncResult, 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"; @@ -673,10 +673,42 @@ export function createEnvironmentThreadShellAtoms(input: { return next; }).pipe(Atom.withLabel("environment-quiet-thread-shell-list")); + // The archive has no live updates, so a removal or a new authoritative shell snapshot rereads 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 result = get(input.archivedSnapshotAtom(environmentId)); + 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; From a52ea9be4187289c7a97005e317204515c7621c1 Mon Sep 17 00:00:00 2001 From: psick22 Date: Wed, 4 Nov 2026 11:55:51 +0900 Subject: [PATCH 5/6] fix(clients): an archive cached before a parent went missing is read again The archived snapshot is read only once a delegated task's parent is missing, so its first read trusted any answer already in the cache, such as one the Archive settings page loaded. If another client had archived the parent and a new authoritative snapshot arrived without it, the task was listed on its own. A settled answer at the first read now counts as loading and is read again. Co-Authored-By: Claude Opus 5.5 --- .../src/state/threadShell.test.ts | 20 +++++++++++-------- .../client-runtime/src/state/threadShell.ts | 12 ++++++----- 2 files changed, 19 insertions(+), 13 deletions(-) diff --git a/packages/client-runtime/src/state/threadShell.test.ts b/packages/client-runtime/src/state/threadShell.test.ts index ffba6ed2367b9..ecee582e3d293 100644 --- a/packages/client-runtime/src/state/threadShell.test.ts +++ b/packages/client-runtime/src/state/threadShell.test.ts @@ -415,12 +415,6 @@ describe("v2 thread shell lists", () => { expect(registry.get(threads.threadShellAtom(parentRef))?.pendingSubagentRequest).toBeNull(); expect(registry.get(inputs).map((thread) => thread.id)).toEqual([helper.id]); - // A task whose parent is gone there is listed on its own, as is one whose - // parent's environment is switched off. - registry.set(snapshotAtom(environmentId), { ...v2ShellSnapshot, threads: [] }); - 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. const archive = ( snapshotSequence: number, archived: ReadonlyArray, @@ -429,6 +423,16 @@ describe("v2 thread shell lists", () => { 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. An archive answer cached before the parent went + // missing is read again first. + 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), @@ -442,7 +446,7 @@ describe("v2 thread shell lists", () => { AsyncResult.success({ ...EMPTY_SHELL_STATE, removedAt: new Map([[parent.id, 11]]) }), ); expect(listed()).toEqual([]); - expect(archiveRefreshes).toEqual([environmentId]); + expect(archiveRefreshes).toHaveLength(2); registry.set(archivedSnapshotAtom(environmentId), AsyncResult.initial(true)); expect(listed()).toEqual([]); archive(12, [{ ...parent, archivedAt: v2Now }]); @@ -455,7 +459,7 @@ describe("v2 thread shell lists", () => { AsyncResult.success({ ...EMPTY_SHELL_STATE, generation: 1 }), ); expect(listed()).toEqual([]); - expect(archiveRefreshes).toEqual([environmentId, environmentId]); + 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 diff --git a/packages/client-runtime/src/state/threadShell.ts b/packages/client-runtime/src/state/threadShell.ts index a8923196c65e3..dbf459c757599 100644 --- a/packages/client-runtime/src/state/threadShell.ts +++ b/packages/client-runtime/src/state/threadShell.ts @@ -673,9 +673,9 @@ export function createEnvironmentThreadShellAtoms(input: { return next; }).pipe(Atom.withLabel("environment-quiet-thread-shell-list")); - // The archive has no live updates, so a removal or a new authoritative shell snapshot rereads it, - // and the answer from before counts as loading. Kept outside the family so a swept node still - // sees a change it missed. + // The archive has no live updates. An answer cached before this atom first read it, a removal, or + // a new authoritative shell snapshot rereads 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< @@ -698,9 +698,11 @@ export function createEnvironmentThreadShellAtoms(input: { const previous = reads.get(get.registry); const generation = shell?.generation ?? previous?.generation ?? 0; const removedAt = shell?.removedAt ?? previous?.removedAt ?? new Map(); + const settled = !result.waiting && !AsyncResult.isInitial(result); if ( - previous !== undefined && - (previous.generation !== generation || previous.removedAt !== removedAt) + previous === undefined + ? settled + : previous.generation !== generation || previous.removedAt !== removedAt ) { reads.set(get.registry, { generation, removedAt, stale: result }); get.refresh(archiveAtom); From 49e60447d1e15178964b18a7c7d6ff875771bc28 Mon Sep 17 00:00:00 2001 From: psick22 Date: Wed, 4 Nov 2026 12:04:29 +0900 Subject: [PATCH 6/6] fix(clients): an archive request made before a parent went missing is read again The previous commit reread only an archive answer that had already settled at the first read. A request the Archive settings page started earlier could still be in flight, and its answer could predate another client's archive of the parent. The first read now rereads the archive whatever its state, which costs at most one extra request per environment. Co-Authored-By: Claude Opus 5.5 --- .../client-runtime/src/state/threadShell.test.ts | 5 +++-- packages/client-runtime/src/state/threadShell.ts | 14 +++++++------- 2 files changed, 10 insertions(+), 9 deletions(-) diff --git a/packages/client-runtime/src/state/threadShell.test.ts b/packages/client-runtime/src/state/threadShell.test.ts index ecee582e3d293..8e0e114fe056d 100644 --- a/packages/client-runtime/src/state/threadShell.test.ts +++ b/packages/client-runtime/src/state/threadShell.test.ts @@ -424,8 +424,9 @@ describe("v2 thread shell lists", () => { 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. An archive answer cached before the parent went - // missing is read again first. + // 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]); diff --git a/packages/client-runtime/src/state/threadShell.ts b/packages/client-runtime/src/state/threadShell.ts index dbf459c757599..875f172d2b243 100644 --- a/packages/client-runtime/src/state/threadShell.ts +++ b/packages/client-runtime/src/state/threadShell.ts @@ -673,9 +673,10 @@ export function createEnvironmentThreadShellAtoms(input: { return next; }).pipe(Atom.withLabel("environment-quiet-thread-shell-list")); - // The archive has no live updates. An answer cached before this atom first read it, a removal, or - // a new authoritative shell snapshot rereads it, and the answer from before counts as loading. - // Kept outside the family so a swept node still sees a change it missed. + // 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< @@ -698,11 +699,10 @@ export function createEnvironmentThreadShellAtoms(input: { const previous = reads.get(get.registry); const generation = shell?.generation ?? previous?.generation ?? 0; const removedAt = shell?.removedAt ?? previous?.removedAt ?? new Map(); - const settled = !result.waiting && !AsyncResult.isInitial(result); if ( - previous === undefined - ? settled - : previous.generation !== generation || previous.removedAt !== removedAt + previous === undefined || + previous.generation !== generation || + previous.removedAt !== removedAt ) { reads.set(get.registry, { generation, removedAt, stale: result }); get.refresh(archiveAtom);