diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 251716acb446f..8e5129e59c657 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -3356,6 +3356,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { kind: "user_input", createdAt: now, blocking, + secret: false, }); } } @@ -4418,6 +4419,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { ]) { assert.equal(shell?.pendingRuntimeRequest?.id, "secret:b"); assert.isTrue(shell?.pendingRuntimeRequest?.blocking); + assert.isTrue(shell?.pendingRuntimeRequest?.secret); } const requestId = RuntimeRequestId.make(`request:${suffix}:async`); diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 60b5de0edb5d7..e578b72af1629 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -1529,6 +1529,7 @@ function pendingRuntimeRequestSummary( kind: request.kind, createdAt: request.createdAt, blocking: request.responseCapability.type !== "message", + secret: false, }; } @@ -1541,6 +1542,7 @@ function secretRequestAsPendingInput( kind: "user_input", createdAt: item.startedAt ?? item.updatedAt, blocking: true, + secret: true, }; } diff --git a/packages/client-runtime/src/state/threadShell.test.ts b/packages/client-runtime/src/state/threadShell.test.ts index 2dcc3233db3a9..9583d60d1cf12 100644 --- a/packages/client-runtime/src/state/threadShell.test.ts +++ b/packages/client-runtime/src/state/threadShell.test.ts @@ -152,6 +152,94 @@ describe("v2 thread shell lists", () => { registry.dispose(); }); + it.each([{ blocking: true }, { blocking: false }])( + "leaves a subagent question (blocking: $blocking) to its running or queued parent", + ({ blocking }) => { + const { registry, threads, snapshotAtom } = makeHarness(); + const parentRunId = RunId.make("parent-run"); + const runningParent: OrchestrationV2ThreadShell = { + ...v2ThreadShell, + latestRunId: parentRunId, + activeRunId: parentRunId, + status: "running", + activityRunStatus: "running", + }; + const child: OrchestrationV2ThreadShell = { + ...v2ThreadShell, + id: ThreadId.make("delegated"), + creationSource: "mcp", + status: "running", + activityRunStatus: "running", + lineage: { + ...v2ThreadShell.lineage, + parentThreadId: v2ThreadShell.id, + relationshipToParent: "subagent", + }, + pendingRuntimeRequest: { + id: RuntimeRequestId.make("question"), + kind: "user_input", + createdAt: v2Now, + blocking, + secret: false, + }, + }; + const parent = threads.threadShellAtom({ environmentId, threadId: v2ThreadShell.id }); + const inputs = threads.childThreadInputsAtom({ environmentId, threadId: v2ThreadShell.id }); + const disposeParent = registry.mount(parent); + const disposeInputs = registry.mount(inputs); + const setThreads = (...shells: OrchestrationV2ThreadShell[]) => + registry.set(snapshotAtom(environmentId), { ...v2ShellSnapshot, threads: shells }); + const expectAsking = (asking: boolean) => { + expect(registry.get(parent)?.pendingSubagentRequest).toBe(asking ? "user_input" : null); + expect(registry.get(inputs).map((thread) => thread.id)).toEqual(asking ? [child.id] : []); + }; + + setThreads(runningParent, child); + expectAsking(false); + setThreads( + { ...runningParent, activeRunId: null, status: "queued", activityRunStatus: null }, + child, + ); + expectAsking(false); + + // Only a blocked child needs a person once no parent turn can answer. A + // waiting parent has ended its turn and is only draining checkpoints. + setThreads({ ...runningParent, status: "waiting", activityRunStatus: "waiting" }, child); + expectAsking(blocking); + setThreads( + { ...runningParent, activeRunId: null, status: "completed", activityRunStatus: null }, + child, + ); + expectAsking(blocking); + + // The parent cannot grant approvals, so they reach a person even mid-run. + setThreads(runningParent, { + ...child, + pendingRuntimeRequest: { + id: RuntimeRequestId.make("approval"), + kind: "command", + createdAt: v2Now, + blocking: true, + }, + }); + expect(registry.get(parent)?.pendingSubagentRequest).toBe("approval"); + expect(registry.get(inputs).map((thread) => thread.id)).toEqual([child.id]); + + // Only a person can enter a secret, and older servers do not mark one. + for (const secret of [true, undefined]) { + const { secret: _, ...question } = child.pendingRuntimeRequest!; + setThreads(runningParent, { + ...child, + pendingRuntimeRequest: secret === undefined ? question : { ...question, secret }, + }); + expectAsking(blocking); + } + disposeInputs(); + disposeParent(); + registry.dispose(); + }, + ); + it("brings a parent back from Working while a nested subagent waits for approval", () => { const { registry, threads, snapshotAtom } = makeHarness(); const child = { diff --git a/packages/client-runtime/src/state/threadShell.ts b/packages/client-runtime/src/state/threadShell.ts index 8d61ff6ac5677..f97060f5eea39 100644 --- a/packages/client-runtime/src/state/threadShell.ts +++ b/packages/client-runtime/src/state/threadShell.ts @@ -45,20 +45,44 @@ const isChildRunLive = ( thread.status === "running" || thread.status === "waiting"; +// Waiting runs have ended their provider turn and are only draining effects. +const isAgentTurnWorking = ( + thread: Pick, +) => { + const status = thread.activityRunStatus ?? thread.status; + return status === "preparing" || status === "starting" || status === "running"; +}; + /** Provider-native subagents ask on their parent, so they are skipped. */ -function subagentRequest(thread: OrchestrationV2ThreadShell): PendingSubagentRequest | null { - const kind = thread.pendingRuntimeRequest?.kind; +function subagentRequest( + thread: OrchestrationV2ThreadShell, + byId: ReadonlyMap, +): PendingSubagentRequest | null { + const request = thread.pendingRuntimeRequest; if ( - kind === undefined || - kind === "auth_refresh" || + request === null || + request.kind === "auth_refresh" || thread.lineage.relationshipToParent !== "subagent" || isProviderNativeSubagentThread(thread) || !isChildRunLive(thread) ) { return null; } - return kind === "user_input" ? "user_input" : "approval"; + if (request.kind !== "user_input") return "approval"; + if (request.blocking === false && isAgentTurnWorking(thread)) return null; + // A running or queued parent turn reads and answers the question itself. + // Only a person can enter a secret, and older servers do not mark one. + const parentId = thread.lineage.parentThreadId; + const parent = parentId === null ? undefined : byId.get(parentId); + if ( + request.secret === false && + parent !== undefined && + (parent.status === "queued" || isAgentTurnWorking(parent)) + ) { + return null; + } + return "user_input"; } export function pendingSubagentRequestsByThread( @@ -67,9 +91,10 @@ export function pendingSubagentRequestsByThread( const requests = new Map(); let byId: ReadonlyMap | undefined; for (const child of threads) { - const request = subagentRequest(child); - if (request === null) continue; + if (child.pendingRuntimeRequest === null) continue; byId ??= new Map(threads.map((thread) => [thread.id, thread] as const)); + const request = subagentRequest(child, byId); + if (request === null) continue; const seen = new Set([child.id]); let current: OrchestrationV2ThreadShell | undefined = child; while (current?.lineage.relationshipToParent === "subagent") { @@ -294,6 +319,7 @@ export function createEnvironmentThreadShellAtoms(input: { else children.set(parent, [thread]); } const subagentRequests = get(pendingSubagentRequestsAtom(ref.environmentId)); + const threadsById = get(environmentThreadIndexAtom(ref.environmentId)); const seen = new Set([ref.threadId]); const pending = [ref.threadId]; const next: EnvironmentThreadShell[] = []; @@ -302,7 +328,7 @@ export function createEnvironmentThreadShellAtoms(input: { if (seen.has(child.id)) continue; seen.add(child.id); pending.push(child.id); - if (subagentRequest(child) !== null) { + if (subagentRequest(child, threadsById) !== null) { next.push(scopedThread(ref.environmentId, child, subagentRequests)); } } diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 6aa4ea7bbe66a..6b0a852bdace8 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -1868,6 +1868,8 @@ export const OrchestrationV2PendingRuntimeRequestSummary = Schema.Struct({ createdAt: Schema.DateTimeUtc, /** False for async questions the run keeps working past. Absent from older servers. */ blocking: Schema.optional(Schema.Boolean), + /** True for a secret request summarized as user input. Absent from older servers. */ + secret: Schema.optional(Schema.Boolean), }); export type OrchestrationV2PendingRuntimeRequestSummary = typeof OrchestrationV2PendingRuntimeRequestSummary.Type;