Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions apps/server/src/orchestration-v2/ProjectionStore.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -3356,6 +3356,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
kind: "user_input",
createdAt: now,
blocking,
secret: false,
});
}
}
Expand Down Expand Up @@ -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`);
Expand Down
2 changes: 2 additions & 0 deletions apps/server/src/orchestration-v2/ProjectionStore.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1529,6 +1529,7 @@ function pendingRuntimeRequestSummary(
kind: request.kind,
createdAt: request.createdAt,
blocking: request.responseCapability.type !== "message",
secret: false,
};
}

Expand All @@ -1541,6 +1542,7 @@ function secretRequestAsPendingInput(
kind: "user_input",
createdAt: item.startedAt ?? item.updatedAt,
blocking: true,
secret: true,
};
}

Expand Down
88 changes: 88 additions & 0 deletions packages/client-runtime/src/state/threadShell.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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 = {
Expand Down
42 changes: 34 additions & 8 deletions packages/client-runtime/src/state/threadShell.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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<OrchestrationV2ThreadShell, "activityRunStatus" | "status">,
) => {
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<ThreadId, OrchestrationV2ThreadShell>,
): 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(
Expand All @@ -67,9 +91,10 @@ export function pendingSubagentRequestsByThread(
const requests = new Map<ThreadId, PendingSubagentRequest>();
let byId: ReadonlyMap<ThreadId, OrchestrationV2ThreadShell> | 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<ThreadId>([child.id]);
let current: OrchestrationV2ThreadShell | undefined = child;
while (current?.lineage.relationshipToParent === "subagent") {
Expand Down Expand Up @@ -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<ThreadId>([ref.threadId]);
const pending = [ref.threadId];
const next: EnvironmentThreadShell[] = [];
Expand All @@ -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));
}
}
Expand Down
2 changes: 2 additions & 0 deletions packages/contracts/src/orchestrationV2.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
Loading