fix(clients): archiving a thread keeps its linked-environment tasks under it

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,

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));

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));

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,

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> = [environmentId]) {
   const snapshotAtom = Atom.family((_environmentId: EnvironmentId) =>
     Atom.make<OrchestrationV2ShellSnapshot | null>(v2ShellSnapshot),
   );
+  type ArchiveResult = AsyncResult.AsyncResult<OrchestrationV2ArchivedShellSnapshot, string>;
+  const archiveRefreshes: Array<EnvironmentId> = [];
+  const archivedSnapshotAtom = Atom.family((archiveEnvironmentId: EnvironmentId) =>
+    Atom.writable(
+      (get): ArchiveResult =>
+        Option.getOrElse(get.self<ArchiveResult>(), () => AsyncResult.success(emptyArchive)),
+      (ctx, value: ArchiveResult) => ctx.setSelf(value),
+      () => {
+        archiveRefreshes.push(archiveEnvironmentId);
+      },
+    ),
+  );
+  const shellStateAtom = Atom.family((_environmentId: EnvironmentId) =>
+    Atom.make<AsyncResult.AsyncResult<EnvironmentShellState, string>>(
+      AsyncResult.success(EMPTY_SHELL_STATE),
+    ),
+  );
   const catalogValueAtom = Atom.make({
     isReady: true,
     entries: new Map(
@@ -47,11 +72,16 @@ function makeHarness(environmentIds: ReadonlyArray<EnvironmentId> = [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<OrchestrationV2ThreadShell>,
+    ) =>
+      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);

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<ReadonlyArray<string> | null>;
+  /** Archived threads, which the live snapshot leaves out. */
+  readonly archivedSnapshotAtom: (
+    environmentId: EnvironmentId,
+  ) => Atom.Atom<AsyncResult.AsyncResult<OrchestrationV2ArchivedShellSnapshot, unknown>>;
+  /** The shell state, whose removals tell a stale archive from one that lacks a thread. */
+  readonly shellStateAtom: (
+    environmentId: EnvironmentId,
+  ) => Atom.Atom<AsyncResult.AsyncResult<EnvironmentShellState, unknown>>;
 }) {
   // 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<ThreadId, number>;
+        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<ThreadId>; 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<string> = new Set();
   const delegatedUnderParentAtom = Atom.make((get) => {
     const environmentIds = new Set(enabledEnvironmentIds(get(input.catalogValueAtom)));
     const next = new Set<string>();
+    // 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 }));
         }