Skip to content
5 changes: 5 additions & 0 deletions apps/server/src/orchestration-v2/ProjectionStore.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -6664,6 +6664,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
]) {
assert.equal(shell?.pendingRuntimeRequest?.id, "secret:b");
assert.isTrue(shell?.pendingRuntimeRequest?.blocking);
assert.isTrue(shell?.pendingRuntimeRequest?.userOnly);
}

const requestId = RuntimeRequestId.make(`request:${suffix}:async`);
Expand Down Expand Up @@ -6755,6 +6756,10 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
[thread?.pendingRuntimeRequest?.blocking, memoryShell.pendingRuntimeRequest?.blocking],
terminal ? [false, false] : [true, true],
);
assert.deepEqual(
[thread?.pendingRuntimeRequest?.userOnly, memoryShell.pendingRuntimeRequest?.userOnly],
terminal ? [undefined, undefined] : [true, true],
);
}
}),
);
Expand Down
1 change: 1 addition & 0 deletions apps/server/src/orchestration-v2/ProjectionStore.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1515,6 +1515,7 @@ function secretRequestAsPendingInput(
kind: "user_input",
createdAt: item.startedAt ?? item.updatedAt,
blocking: true,
userOnly: true,
};
}

Expand Down
20 changes: 17 additions & 3 deletions apps/server/src/orchestration-v2/ShellStream.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -38,8 +38,8 @@ const emptyShellSnapshot = {
} as OrchestrationV2ShellSnapshot;

describe("subagent shell events", () => {
const stored = (threadId: string, type: string) =>
({ event: { threadId, type } }) as ApplicationStoredEvent;
const stored = (threadId: string, type: string, itemType = "command_execution") =>
({ event: { threadId, type, payload: { type: itemType } } }) as ApplicationStoredEvent;

it("keeps regular threads and subagent state changes while omitting subagent transcript updates", () => {
const subagents = new Set(["child"]);
Expand All @@ -49,6 +49,14 @@ describe("subagent shell events", () => {
expect(
shouldPublishShellApplicationEvent(stored("child", "turn-item.updated"), subagents),
).toBe(false);
for (const itemType of ["secret_request", "link_request"]) {
expect(
shouldPublishShellApplicationEvent(
stored("child", "turn-item.updated", itemType),
subagents,
),
).toBe(true);
}
expect(shouldPublishShellApplicationEvent(stored("child", "node.updated"), subagents)).toBe(
false,
);
Expand Down Expand Up @@ -198,7 +206,13 @@ describe("omitIdleSubagentShells", () => {
expect(omitted.threads.map((thread) => thread.id)).toEqual(["root"]);
expect(
shouldPublishShellApplicationEvent(
{ event: { threadId: "idle", type: "turn-item.updated" } } as ApplicationStoredEvent,
{
event: {
threadId: "idle",
type: "turn-item.updated",
payload: { type: "command_execution" },
},
} as ApplicationStoredEvent,
omitted.subagentThreadIds,
),
).toBe(false);
Expand Down
6 changes: 6 additions & 0 deletions apps/server/src/orchestration-v2/ShellStream.ts
Original file line number Diff line number Diff line change
Expand Up @@ -428,6 +428,12 @@ export function shouldPublishShellApplicationEvent(
// Replay and the live tail overlap; keep known IDs so late events stay filtered.
return true;
}
if (
event.type === "turn-item.updated" &&
(event.payload.type === "secret_request" || event.payload.type === "link_request")
) {
return true;
}
return !(
subagentThreadIds.has(event.threadId) &&
(event.type === "node.updated" ||
Expand Down
60 changes: 60 additions & 0 deletions apps/server/src/relay/AgentAwarenessRelay.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -375,6 +375,66 @@ describe("AgentAwarenessRelay", () => {
}),
);

it.effect("publishes the parent as needing input when its subagent raises a secret card", () =>
Effect.gen(function* () {
const now = DateTime.makeUnsafe(NOW);
const child = waitingChild({
pendingRuntimeRequest: {
id: RuntimeRequestId.make("child-secret-card"),
kind: "user_input",
createdAt: now,
blocking: true,
userOnly: true,
},
});
const parent = shell({ status: "completed" });
const events = yield* Queue.unbounded<OrchestrationV2DomainEvent>();
const pulls = yield* Queue.unbounded<void>();
const { relay, publications } = yield* makeTestRelay({
readShell: (threadId) => Effect.succeed(threadId === child.id ? child : parent),
domainEvents: Stream.fromEffectRepeat(
Queue.offer(pulls, undefined).pipe(Effect.andThen(Queue.take(events))),
),
});
yield* relay.start();
yield* Queue.take(pulls);
const runId = RunId.make("run:child-secret-card");
// Recording a secret card emits only this event for the child.
yield* Queue.offer(events, {
id: EventId.make("event:child-secret-card"),
type: "turn-item.updated",
threadId: CHILD_THREAD_ID,
runId,
occurredAt: now,
payload: {
id: TurnItemId.make("child-secret-card"),
threadId: CHILD_THREAD_ID,
runId,
nodeId: NodeId.make("node:child-secret-card"),
providerThreadId: null,
providerTurnId: null,
nativeItemRef: null,
parentItemId: null,
ordinal: 1,
status: "waiting",
title: "SMS code",
startedAt: now,
completedAt: null,
updatedAt: now,
type: "secret_request",
label: "SMS code",
reason: "The agent needs the code.",
secretStatus: "pending",
},
});
yield* Queue.take(pulls);
yield* relay.drain;
assert.equal(publications.length, 1);
assert.isTrue(publications[0]?.url.includes(`/threads/${THREAD_ID}/`) ?? false);
assert.equal(publications[0]?.state?.phase, "waiting_for_input");
}).pipe(Effect.scoped),
);

it.effect("republishes a completed parent for its waiting child after a relink", () =>
Effect.gen(function* () {
const parent = shell({ status: "completed" });
Expand Down
20 changes: 15 additions & 5 deletions apps/server/src/relay/AgentAwarenessRelay.ts
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@ import {
import {
type SubagentRequestAwareness,
projectThreadAwarenessV2,
subagentAwaitsApproval,
subagentAwaitsUser,
} from "@t3tools/shared/agentAwareness";
import { turnItemUpdateCanEndBackgroundWork } from "@t3tools/shared/orchestrationV2PendingBackgroundWork";
import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker";
Expand Down Expand Up @@ -128,7 +128,13 @@ function shouldPublishAgentAwarenessEvent(
case "provider-session.detached":
case "provider-turn.updated":
case "turn-item.updated":
return isTurnItemPayload(event.payload) && turnItemUpdateCanEndBackgroundWork(event.payload);
return (
isTurnItemPayload(event.payload) &&
(turnItemUpdateCanEndBackgroundWork(event.payload) ||
// A subagent's secret or link card waits on the user through its parent.
event.payload.type === "secret_request" ||
event.payload.type === "link_request")
);
case "message.updated":
case "plan.updated":
case "checkpoint-scope.created":
Expand Down Expand Up @@ -308,7 +314,7 @@ interface SubagentAlert {
}

const waitingApprovalId = (thread: AgentAwarenessThread | null) =>
thread !== null && thread.archivedAt === null && subagentAwaitsApproval(thread)
thread !== null && thread.archivedAt === null && subagentAwaitsUser(thread)
? thread.pendingRuntimeRequest?.id
: undefined;

Expand Down Expand Up @@ -377,7 +383,7 @@ export function resolveAgentAwarenessRelayActiveThreadIds(input: {
return input.threads
.filter((thread) => {
// Publishing a waiting subagent republishes its top-level thread.
if (subagentAwaitsApproval(thread)) return true;
if (subagentAwaitsUser(thread)) return true;
const project = projectById.get(thread.projectId);
if (!project) {
return false;
Expand Down Expand Up @@ -758,7 +764,11 @@ export const make = Effect.gen(function* () {
if (requestId !== entry.requestId) subagentRequests.set(childId, { ...entry, requestId });
// The newest wait is the one to announce.
if (pending === null || DateTime.isGreaterThan(child.updatedAt, pending.updatedAt)) {
pending = { requestId, updatedAt: child.updatedAt };
pending = {
requestId,
updatedAt: child.updatedAt,
waitsForInput: child.pendingRuntimeRequest?.kind === "user_input",
};
}
}
return pending;
Expand Down
45 changes: 45 additions & 0 deletions packages/client-runtime/src/state/threadShell.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -160,6 +160,36 @@ describe("v2 thread shell lists", () => {
registry.dispose();
});

it("flags the parent while a live subagent waits on a secret card", () => {
const { registry, threads, snapshotAtom } = makeHarness();
const root = v2ThreadShell;
const child = {
...root,
id: ThreadId.make("child"),
creationSource: "mcp" as const,
status: "running" as const,
activityRunStatus: "running" as const,
lineage: {
...root.lineage,
parentThreadId: root.id,
relationshipToParent: "subagent" as const,
},
pendingRuntimeRequest: {
id: RuntimeRequestId.make("secret-card"),
kind: "user_input" as const,
createdAt: v2Now,
blocking: true,
userOnly: true,
},
};
registry.set(snapshotAtom(environmentId), { ...v2ShellSnapshot, threads: [root, child] });
expect(
registry.get(threads.threadShellAtom({ environmentId, threadId: root.id }))
?.pendingSubagentRequest,
).toBe("user_input");
registry.dispose();
});

it("rolls a side chat's questions and approvals up to its parent row", () => {
const { registry, threads, snapshotAtom } = makeHarness();
const root = v2ThreadShell;
Expand Down Expand Up @@ -380,6 +410,21 @@ describe("v2 thread shell lists", () => {
expect(registry.get(threads.threadShellAtom(parentRef))?.pendingSubagentRequest).toBeNull();
expect(registry.get(inputs).map((thread) => thread.id)).toEqual([helper.id]);

// A secret card needs the user, so it still reaches the parent here.
registry.set(snapshotAtom(remoteEnvironmentId), {
...v2ShellSnapshot,
threads: [
task,
{
...helper,
pendingRuntimeRequest: { ...approval, kind: "user_input" as const, userOnly: true },
},
],
});
expect(registry.get(threads.threadShellAtom(parentRef))?.pendingSubagentRequest).toBe(
"user_input",
);

// A task whose parent is gone there is listed on its own, as is one whose
// parent's environment is switched off.
registry.set(snapshotAtom(environmentId), { ...v2ShellSnapshot, threads: [] });
Expand Down
47 changes: 29 additions & 18 deletions packages/client-runtime/src/state/threadShell.ts
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@ import {
const EMPTY_THREADS: ReadonlyArray<OrchestrationV2ThreadShell> = Object.freeze([]);
const EMPTY_SCOPED_THREAD_REFS: ReadonlyArray<ScopedThreadRef> = Object.freeze([]);
const EMPTY_THREAD_INDEX: ReadonlyMap<ThreadId, OrchestrationV2ThreadShell> = new Map();
const EMPTY_SUBAGENT_REQUESTS: ReadonlyMap<ThreadId, PendingSubagentRequest> = new Map();
const EMPTY_SUBAGENT_REQUESTS: ReadonlyMap<ThreadId, never> = new Map<ThreadId, never>();
const EMPTY_SIDE_CHATS_BY_PARENT: ReadonlyMap<
ThreadId,
ReadonlyArray<OrchestrationV2ThreadShell>
Expand Down Expand Up @@ -85,6 +85,17 @@ function subagentRequest(thread: OrchestrationV2ThreadShell): PendingSubagentReq
return kind === "user_input" ? "user_input" : "approval";
}

/** A user-only card keeps its mark through roll-ups; clients see it as user_input. */
type RolledUpRequest = PendingSubagentRequest | "user_only_input";

const rolledUpRequest = (
thread: OrchestrationV2ThreadShell,
request: PendingSubagentRequest | null,
): RolledUpRequest | null =>
request === "user_input" && thread.pendingRuntimeRequest?.userOnly === true
? "user_only_input"
: request;

/** A side chat asks its user directly, so questions roll up to the parent as well as approvals. */
function sideChatRequest(thread: OrchestrationV2ThreadShell): PendingSubagentRequest | null {
const kind = thread.pendingRuntimeRequest?.kind;
Expand All @@ -104,17 +115,13 @@ function pendingSubagentRequestsByThread(
threads: ReadonlyArray<OrchestrationV2ThreadShell>,
delegated: ReadonlyArray<{
readonly parentThreadId: ThreadId;
readonly request: PendingSubagentRequest;
readonly request: RolledUpRequest;
}> = [],
): ReadonlyMap<ThreadId, PendingSubagentRequest> {
const requests = new Map<ThreadId, PendingSubagentRequest>();
): ReadonlyMap<ThreadId, RolledUpRequest> {
const requests = new Map<ThreadId, RolledUpRequest>();
let byId: ReadonlyMap<ThreadId, OrchestrationV2ThreadShell> | undefined;
const flag = (
firstParentId: ThreadId | null,
request: PendingSubagentRequest,
seen: Set<ThreadId>,
) => {
// The coordinator answers child questions; only approvals need user attention.
const flag = (firstParentId: ThreadId | null, request: RolledUpRequest, seen: Set<ThreadId>) => {
// The coordinator answers child questions; approvals and user-only cards need the user.
if (request === "user_input") return;
byId ??= new Map(threads.map((thread) => [thread.id, thread] as const));
let parentId = firstParentId;
Expand All @@ -138,16 +145,16 @@ function pendingSubagentRequestsByThread(
continue;
}
if (child.lineage.relationshipToParent !== "subagent") continue;
const request = subagentRequest(child);
const request = rolledUpRequest(child, subagentRequest(child));
if (request !== null) flag(parentId, request, new Set([child.id]));
}
for (const task of delegated) flag(task.parentThreadId, task.request, new Set());
return requests.size === 0 ? EMPTY_SUBAGENT_REQUESTS : requests;
}

const sameRequests = (
left: ReadonlyMap<ThreadId, PendingSubagentRequest>,
right: ReadonlyMap<ThreadId, PendingSubagentRequest>,
left: ReadonlyMap<ThreadId, RolledUpRequest>,
right: ReadonlyMap<ThreadId, RolledUpRequest>,
) =>
left.size === right.size &&
[...right].every(([threadId, request]) => left.get(threadId) === request);
Expand Down Expand Up @@ -274,7 +281,7 @@ export function createEnvironmentThreadShellAtoms(input: {
// environments run for it, so two environments delegating to each other
// never read each other's totals.
const localSubagentRequestsAtom = Atom.family((environmentId: EnvironmentId) => {
let previous = EMPTY_SUBAGENT_REQUESTS;
let previous: ReadonlyMap<ThreadId, RolledUpRequest> = EMPTY_SUBAGENT_REQUESTS;
return Atom.make((get) => {
const next = pendingSubagentRequestsByThread(get(environmentThreadsAtom(environmentId)));
if (sameRequests(previous, next)) return previous;
Expand Down Expand Up @@ -305,15 +312,15 @@ export function createEnvironmentThreadShellAtoms(input: {
else tasks.push(task);
}
}
let flags = new Map<EnvironmentId, ReadonlyMap<ThreadId, PendingSubagentRequest>>();
let flags = new Map<EnvironmentId, ReadonlyMap<ThreadId, RolledUpRequest>>();
const requestsOf = (environmentId: EnvironmentId) =>
flags.get(environmentId) ?? get(localSubagentRequestsAtom(environmentId));
for (let changed = tasksByDelegator.size > 0; changed;) {
changed = false;
const next = new Map(flags);
for (const [environmentId, tasks] of tasksByDelegator) {
const delegated = tasks.flatMap(({ runsIn, thread, parentThreadId }) => {
const own = subagentRequest(thread);
const own = rolledUpRequest(thread, subagentRequest(thread));
const nested = requestsOf(runsIn).get(thread.id);
const request = own === "approval" ? own : (nested ?? own);
return request === null ? [] : [{ parentThreadId, request }];
Expand All @@ -333,11 +340,15 @@ export function createEnvironmentThreadShellAtoms(input: {
}).pipe(Atom.withLabel("environment-delegated-subagent-requests"));

const pendingSubagentRequestsAtom = Atom.family((environmentId: EnvironmentId) => {
let previous = EMPTY_SUBAGENT_REQUESTS;
let previous: ReadonlyMap<ThreadId, PendingSubagentRequest> = EMPTY_SUBAGENT_REQUESTS;
return Atom.make((get) => {
const next =
const rolledUp =
get(delegatedSubagentRequestsAtom).get(environmentId) ??
get(localSubagentRequestsAtom(environmentId));
const next = new Map<ThreadId, PendingSubagentRequest>();
for (const [threadId, request] of rolledUp) {
next.set(threadId, request === "user_only_input" ? "user_input" : request);
}
if (sameRequests(previous, next)) return previous;
previous = next;
return next;
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 @@ -2080,6 +2080,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),
/** Secret and link cards: only the user can answer them, never a coordinating agent. */
userOnly: Schema.optional(Schema.Boolean),
});
export type OrchestrationV2PendingRuntimeRequestSummary =
typeof OrchestrationV2PendingRuntimeRequestSummary.Type;
Expand Down
Loading
Loading