fix(client): show Input when a hidden subagent needs an answer

packages/client-runtime/src/state/threadShell.test.ts

@@ -1,6 +1,8 @@
 import {
   EnvironmentId,
   ProjectId,
+  RunId,
+  RuntimeRequestId,
   ThreadId,
   type OrchestrationV2ShellSnapshot,
 } from "@t3tools/contracts";
@@ -9,7 +11,8 @@ import { Atom, AtomRegistry } from "effect/reactivity";
 import { describe, expect, it } from "vite-plus/test";
 
 import { PrimaryConnectionTarget } from "../connection/model.ts";
-import { v2ShellSnapshot, v2ThreadShell } from "./orchestrationV2TestFixtures.ts";
+import { isThreadWorking } from "./threadInbox.ts";
+import { v2Now, v2ShellSnapshot, v2ThreadShell } from "./orchestrationV2TestFixtures.ts";
 import { applyShellStreamEvent } from "./shellReducer.ts";
 import { createEnvironmentThreadShellAtoms } from "./threadShell.ts";
 
@@ -103,6 +106,133 @@ describe("v2 thread shell lists", () => {
     registry.dispose();
   });
 
+  it("returns a running parent to Input until its last hidden descendant question is answered", () => {
+    const { registry, threads, snapshotAtom } = makeHarness();
+    const root = {
+      ...v2ThreadShell,
+      status: "running" as const,
+      latestRunId: RunId.make("running-root"),
+      activeRunId: RunId.make("running-root"),
+    };
+    const child = {
+      ...v2ThreadShell,
+      id: ThreadId.make("child"),
+      creationSource: "mcp" as const,
+      lineage: {
+        ...root.lineage,
+        parentThreadId: root.id,
+        relationshipToParent: "subagent" as const,
+      },
+    };
+    const question = {
+      ...child,
+      id: ThreadId.make("nested-question"),
+      lineage: { ...child.lineage, parentThreadId: child.id },
+      pendingRuntimeRequest: {
+        id: RuntimeRequestId.make("question"),
+        kind: "user_input" as const,
+        createdAt: v2Now,
+      },
+    };
+    const secondQuestion = { ...question, id: ThreadId.make("second-question") };
+    const snapshot = { ...v2ShellSnapshot, threads: [root, child, question, secondQuestion] };
+    registry.set(snapshotAtom(environmentId), snapshot);
+    const dispose = registry.mount(threads.threadShellsAtom);
+    const disposeNavigation = registry.mount(threads.navigationThreadShellsAtom);
+    const ref = { environmentId, threadId: root.id };
+    const parent = () => registry.get(threads.threadShellsAtom)[0]!;
+    try {
+      expect(parent().hasPendingUserInput).toBe(true);
+      expect(isThreadWorking(parent())).toBe(false);
+      expect(registry.get(threads.navigationThreadShellsAtom)[0]).toBe(parent());
+      expect(
+        registry.get(
+          threads.threadShellsForProjectRefsAtom([{ environmentId, projectId: root.projectId }]),
+        )[0],
+      ).toBe(parent());
+      expect(registry.get(threads.threadShellAtom(ref))?.hasPendingUserInput).toBe(false);
+      expect(registry.get(threads.childThreadInputsAtom(ref)).map((t) => t.id)).toEqual([
+        question.id,
+        secondQuestion.id,
+      ]);
+      const before = parent();
+      registry.set(snapshotAtom(environmentId), {
+        ...snapshot,
+        threads: [root, { ...child, title: "Streaming" }, question, secondQuestion],
+      });
+      expect(parent()).toBe(before);
+      registry.set(snapshotAtom(environmentId), {
+        ...snapshot,
+        threads: [root, child, { ...question, pendingRuntimeRequest: null }, secondQuestion],
+      });
+      expect(parent().hasPendingUserInput).toBe(true);
+      registry.set(snapshotAtom(environmentId), {
+        ...snapshot,
+        threads: [
+          root,
+          child,
+          { ...question, pendingRuntimeRequest: null },
+          { ...secondQuestion, pendingRuntimeRequest: null },
+        ],
+      });
+      expect(parent().hasPendingUserInput).toBe(false);
+      expect(isThreadWorking(parent())).toBe(true);
+      expect(registry.get(threads.childThreadInputsAtom(ref))).toEqual([]);
+    } finally {
+      disposeNavigation();
+      dispose();
+      registry.dispose();
+    }
+  });
+
+  it("keeps fork, archived, provider-native and other-environment questions out of parent attention", () => {
+    const { registry, threads, snapshotAtom } = makeHarness([environmentId, remoteEnvironmentId]);
+    const root = v2ThreadShell;
+    const question = {
+      ...root,
+      id: ThreadId.make("question"),
+      creationSource: "mcp" as const,
+      lineage: {
+        ...root.lineage,
+        parentThreadId: root.id,
+        relationshipToParent: "subagent" as const,
+      },
+      pendingRuntimeRequest: {
+        id: RuntimeRequestId.make("question"),
+        kind: "user_input" as const,
+        createdAt: v2Now,
+      },
+    };
+    registry.set(snapshotAtom(environmentId), {
+      ...v2ShellSnapshot,
+      threads: [
+        root,
+        {
+          ...question,
+          id: ThreadId.make("fork"),
+          lineage: { ...question.lineage, relationshipToParent: "fork" },
+        },
+        { ...question, id: ThreadId.make("archived"), archivedAt: v2Now },
+        { ...question, id: ThreadId.make("native"), creationSource: "provider" },
+      ],
+    });
+    registry.set(snapshotAtom(remoteEnvironmentId), {
+      ...v2ShellSnapshot,
+      threads: [root, question],
+    });
+    const dispose = registry.mount(threads.threadShellsAtom);
+    try {
+      const roots = registry.get(threads.threadShellsAtom).filter((t) => t.id === root.id);
+      expect(roots.map((t) => [t.environmentId, t.hasPendingUserInput])).toEqual([
+        [environmentId, false],
+        [remoteEnvironmentId, true],
+      ]);
+    } finally {
+      dispose();
+      registry.dispose();
+    }
+  });
+
   it("leaves switched-off environments out of navigation", () => {
     const { registry, threads, catalogValueAtom } = makeHarness([
       environmentId,

packages/client-runtime/src/state/threadShell.ts

@@ -73,6 +73,51 @@ export function createEnvironmentThreadShellAtoms(input: {
     }).pipe(Atom.withLabel(`environment-thread-index:${environmentId}`)),
   );
 
+  const environmentChildInputAncestorsAtom = Atom.family((environmentId: EnvironmentId) =>
+    Atom.make((get): ReadonlySet<ThreadId> => {
+      const threads = get(environmentThreadIndexAtom(environmentId));
+      const ancestors = new Set<ThreadId>();
+      for (const child of threads.values()) {
+        if (
+          child.archivedAt !== null ||
+          child.pendingRuntimeRequest?.kind !== "user_input" ||
+          isProviderNativeSubagentThread(child)
+        )
+          continue;
+        let current = child;
+        const seen = new Set<ThreadId>([child.id]);
+        while (current.lineage.relationshipToParent === "subagent") {
+          const parentId = current.lineage.parentThreadId;
+          if (parentId === null || seen.has(parentId)) break;
+          seen.add(parentId);
+          const parent = threads.get(parentId);
+          if (parent === undefined || parent.archivedAt !== null) break;
+          ancestors.add(parentId);
+          current = parent;
+        }
+      }
+      return ancestors;
+    }).pipe(Atom.withLabel(`environment-child-input-ancestors:${environmentId}`)),
+  );
+
+  const inputListShells = new WeakMap<EnvironmentThreadShell, EnvironmentThreadShell>();
+  // List attention includes hidden descendants. Point reads keep the parent's own
+  // request state so its composer still offers the child question's navigation.
+  const listThread = (
+    environmentId: EnvironmentId,
+    source: OrchestrationV2ThreadShell,
+    childInputAncestors: ReadonlySet<ThreadId>,
+  ) => {
+    const thread = scopedThread(environmentId, source);
+    if (thread.hasPendingUserInput || !childInputAncestors.has(source.id)) return thread;
+    let withChildInput = inputListShells.get(thread);
+    if (withChildInput === undefined) {
+      withChildInput = { ...thread, hasPendingUserInput: true };
+      inputListShells.set(thread, withChildInput);
+    }
+    return withChildInput;
+  };
+
   const environmentThreadRefsAtom = Atom.family((environmentId: EnvironmentId) => {
     let previous: ReadonlyArray<ScopedThreadRef> = [];
     return Atom.make((get) => {
@@ -204,7 +249,13 @@ export function createEnvironmentThreadShellAtoms(input: {
           seen.add(key);
           const thread = threads.get(ref.threadId);
           if (thread !== undefined) {
-            next.push(scopedThread(ref.environmentId, thread));
+            next.push(
+              listThread(
+                ref.environmentId,
+                thread,
+                get(environmentChildInputAncestorsAtom(ref.environmentId)),
+              ),
+            );
           }
         }
       }
@@ -233,8 +284,9 @@ export function createEnvironmentThreadShellAtoms(input: {
   const threadShellsAtom = Atom.make((get) => {
     const next: EnvironmentThreadShell[] = [];
     for (const environmentId of enabledEnvironmentIds(get(input.catalogValueAtom))) {
+      const childInputAncestors = get(environmentChildInputAncestorsAtom(environmentId));
       for (const thread of get(environmentThreadsAtom(environmentId))) {
-        next.push(scopedThread(environmentId, thread));
+        next.push(listThread(environmentId, thread, childInputAncestors));
       }
     }
     if (arrayElementsEqual(previousThreadShells, next)) {
@@ -248,10 +300,11 @@ export function createEnvironmentThreadShellAtoms(input: {
   const navigationThreadShellsAtom = Atom.make((get) => {
     const next: EnvironmentThreadShell[] = [];
     for (const environmentId of enabledEnvironmentIds(get(input.catalogValueAtom))) {
+      const childInputAncestors = get(environmentChildInputAncestorsAtom(environmentId));
       for (const thread of get(environmentThreadsAtom(environmentId))) {
         if (thread.archivedAt !== null || thread.lineage.relationshipToParent === "subagent")
           continue;
-        next.push(scopedThread(environmentId, thread));
+        next.push(listThread(environmentId, thread, childInputAncestors));
       }
     }
     if (arrayElementsEqual(previousNavigationShells, next)) return previousNavigationShells;