fix(client): completed remote tasks stay grouped while their parent is offline

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({

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<ThreadId, number>()),
     hiddenThreadTagsAtom: () => Atom.make(null),
   });
   const selectedShell = shells.threadShellAtom({ environmentId, threadId: selectedThreadId });

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;
       },
     },

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 = () => {

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<ThreadId, number>()),
     hiddenThreadTagsAtom: () => Atom.make(null),
   });
   const selectedShell = shells.threadShellAtom({ environmentId, threadId: selectedThreadId });

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({

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

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<R, E>(
     ).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,
   };
 }

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<EnvironmentId> = [environment
       ]),
     ),
   });
+  const shellStatusAtom = Atom.family((_environmentId: EnvironmentId) =>
+    Atom.make<EnvironmentShellStatus>("live"),
+  );
+  const removedAtAtom = Atom.family((_environmentId: EnvironmentId) =>
+    Atom.make<ReadonlyMap<ThreadId, number>>(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([