diff --git a/packages/client-runtime/src/state/threadShell.test.ts b/packages/client-runtime/src/state/threadShell.test.ts index d3653d30035d7..58eb1acc62af0 100644 --- a/packages/client-runtime/src/state/threadShell.test.ts +++ b/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, diff --git a/packages/client-runtime/src/state/threadShell.ts b/packages/client-runtime/src/state/threadShell.ts index 3cb54087ac14d..7e4d02af466d0 100644 --- a/packages/client-runtime/src/state/threadShell.ts +++ b/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 => { + const threads = get(environmentThreadIndexAtom(environmentId)); + const ancestors = new Set(); + 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([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(); + // 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, + ) => { + 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 = []; 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;