Skip to content

Commit 2605816

Browse files
committed
feat(agents): message a Codex subagent straight from its transcript
A spawned Codex agent could only be read. Steering it meant asking the parent to relay, and the parent does not always listen. The agent drill-in in the Agents tab now ends in a one-line composer while the agent is live. Enter sends; the reply lands in the transcript above like any other turn. A new server.sendSubagentInput RPC routes to the provider adapter; the Codex adapter authorizes the thread against the session's root (the same ancestry walk the transcript read uses, now shared), refuses threads the app-server marks as not accepting direct input, and starts a turn on the child thread. Claude has no host-side way to reach a subagent, so the composer stays hidden there.
1 parent 37d248a commit 2605816

18 files changed

Lines changed: 414 additions & 37 deletions

apps/server/src/orchestration/Layers/CheckpointReactor.test.ts

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -152,6 +152,7 @@ function createProviderServiceHarness(
152152
}),
153153
rollbackConversation,
154154
readSubagentTranscript: () => unsupported(),
155+
sendSubagentInput: () => unsupported(),
155156
resolveSubagentWorktree: () => Effect.succeed(null),
156157
deleteThread: () => unsupported(),
157158
get streamEvents() {

apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -444,6 +444,7 @@ describe("ProviderCommandReactor", () => {
444444
},
445445
rollbackConversation: () => unsupported(),
446446
readSubagentTranscript: () => unsupported(),
447+
sendSubagentInput: () => unsupported(),
447448
resolveSubagentWorktree: () => Effect.succeed(null),
448449
deleteThread: () => unsupported(),
449450
get streamEvents() {

apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -141,6 +141,7 @@ function createProviderServiceHarness() {
141141
},
142142
rollbackConversation: () => unsupported(),
143143
readSubagentTranscript: () => unsupported(),
144+
sendSubagentInput: () => unsupported(),
144145
resolveSubagentWorktree: ({ toolUseId }) =>
145146
Effect.succeed(subagentWorktreesByToolUseId.get(toolUseId) ?? null),
146147
deleteThread: () => unsupported(),

apps/server/src/provider/Layers/CodexAdapter.test.ts

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -346,6 +346,11 @@ class FakeCodexRuntime implements CodexSessionRuntimeShape {
346346
): Promise<EffectCodexSchema.V2ThreadItemsListResponse> => Promise.resolve({ data: [] }),
347347
);
348348

349+
public readonly startStoredThreadTurnImpl = vi.fn(
350+
(_providerThreadId: string, _text: string): Promise<EffectCodexSchema.V2TurnStartResponse> =>
351+
Promise.resolve({ turn: { id: "turn-direct-1", items: [], status: "inProgress" } }),
352+
);
353+
349354
public readonly rollbackThreadImpl = vi.fn((_numTurns: number): Promise<CodexThreadSnapshot> =>
350355
Promise.resolve({
351356
threadId: "provider-thread-1",
@@ -431,6 +436,10 @@ class FakeCodexRuntime implements CodexSessionRuntimeShape {
431436
return Effect.promise(() => this.readStoredThreadItemsImpl(input));
432437
}
433438

439+
startStoredThreadTurn(providerThreadId: string, text: string) {
440+
return Effect.promise(() => this.startStoredThreadTurnImpl(providerThreadId, text));
441+
}
442+
434443
rollbackThread(numTurns: number) {
435444
return Effect.promise(() => this.rollbackThreadImpl(numTurns));
436445
}
@@ -875,6 +884,49 @@ transcriptLayer("CodexAdapterLive subagent transcripts", (it) => {
875884
assert.match(result.failure.detail, /not a subagent of this conversation/);
876885
}),
877886
);
887+
888+
it.effect("starts a turn on an authorized subagent thread for direct input", () =>
889+
Effect.gen(function* () {
890+
const adapter = yield* CodexAdapter;
891+
const threadId = asThreadId("thread-direct-input");
892+
yield* adapter.startSession({
893+
provider: ProviderDriverKind.make("codex"),
894+
threadId,
895+
runtimeMode: "full-access",
896+
});
897+
const runtime = transcriptRuntimeFactory.lastRuntime;
898+
assert.ok(runtime);
899+
900+
const sendSubagentInput = adapter.sendSubagentInput;
901+
assert.ok(sendSubagentInput);
902+
const result = yield* sendSubagentInput(threadId, {
903+
threadId,
904+
agentId: "child-provider-thread",
905+
text: "Focus on the router first.",
906+
});
907+
assert.deepStrictEqual(result, { turnId: "turn-direct-1" });
908+
assert.deepStrictEqual(runtime.startStoredThreadTurnImpl.mock.calls, [
909+
["child-provider-thread", "Focus on the router first."],
910+
]);
911+
912+
// A child the app-server says cannot take input is refused before any turn starts.
913+
runtime.readStoredThreadMetadataImpl.mockImplementation((providerThreadId: string) =>
914+
Promise.resolve({
915+
...makeStoredThread({ id: providerThreadId, parentThreadId: "provider-thread-1" }),
916+
canAcceptDirectInput: false,
917+
}),
918+
);
919+
const refused = yield* sendSubagentInput(threadId, {
920+
threadId,
921+
agentId: "child-provider-thread",
922+
text: "Anyone there?",
923+
}).pipe(Effect.result);
924+
assert.ok(refused._tag === "Failure");
925+
assert.ok(refused.failure instanceof ProviderAdapterRequestError);
926+
assert.match(refused.failure.detail, /does not accept direct input/);
927+
assert.equal(runtime.startStoredThreadTurnImpl.mock.calls.length, 1);
928+
}),
929+
);
878930
});
879931

880932
const sessionRuntimeFactory = makeRuntimeFactory();

apps/server/src/provider/Layers/CodexAdapter.ts

Lines changed: 75 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -3228,6 +3228,76 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* (
32283228
})),
32293229
);
32303230

3231+
/** Resolves a stored provider thread the session may act on: one whose
3232+
* ancestry reaches this session's root within the supported nesting depth.
3233+
* Returns the thread's metadata; anything else is a request error. */
3234+
const authorizeSubagentThread = Effect.fn("authorizeSubagentThread")(function* (
3235+
context: CodexAdapterSessionContext,
3236+
threadId: ThreadId,
3237+
agentId: string,
3238+
options: {
3239+
readonly requestError: (detail: string) => ProviderAdapterRequestError;
3240+
readonly parentThreadDetail: string;
3241+
},
3242+
) {
3243+
const { requestError } = options;
3244+
const rootThreadId = yield* context.runtime.readProviderThreadId.pipe(
3245+
Effect.mapError((cause) => mapCodexRuntimeError(threadId, "thread/read", cause)),
3246+
);
3247+
if (agentId === rootThreadId) {
3248+
return yield* requestError(options.parentThreadDetail);
3249+
}
3250+
3251+
const readStoredThreadMetadata = (providerThreadId: string) =>
3252+
context.runtime
3253+
.readStoredThreadMetadata(providerThreadId)
3254+
.pipe(Effect.mapError((cause) => mapCodexRuntimeError(threadId, "thread/read", cause)));
3255+
const candidate = yield* readStoredThreadMetadata(agentId);
3256+
const visited = new Set<string>([candidate.id]);
3257+
let current = candidate;
3258+
3259+
for (let depth = 0; depth < CODEX_SUBAGENT_MAX_ANCESTRY_DEPTH; depth += 1) {
3260+
const parentThreadId = readCodexSubagentParentThreadId(current);
3261+
if (parentThreadId === rootThreadId) {
3262+
return candidate;
3263+
}
3264+
if (!parentThreadId || visited.has(parentThreadId)) {
3265+
return yield* requestError(
3266+
`Codex thread '${agentId}' is not a subagent of this conversation.`,
3267+
);
3268+
}
3269+
visited.add(parentThreadId);
3270+
current = yield* readStoredThreadMetadata(parentThreadId);
3271+
}
3272+
3273+
return yield* requestError(
3274+
`Codex thread '${agentId}' exceeded the supported subagent nesting depth.`,
3275+
);
3276+
});
3277+
3278+
const sendSubagentInput: NonNullable<CodexAdapterShape["sendSubagentInput"]> = Effect.fn(
3279+
"sendSubagentInput",
3280+
)(function* (threadId, input) {
3281+
const requestError = (detail: string) =>
3282+
new ProviderAdapterRequestError({
3283+
provider: PROVIDER,
3284+
method: "sendSubagentInput",
3285+
detail,
3286+
});
3287+
const context = yield* requireSession(threadId);
3288+
const candidate = yield* authorizeSubagentThread(context, threadId, input.agentId, {
3289+
requestError,
3290+
parentThreadDetail: "Send to the parent thread through the composer instead.",
3291+
});
3292+
if (candidate.canAcceptDirectInput === false) {
3293+
return yield* requestError("This agent does not accept direct input right now.");
3294+
}
3295+
const response = yield* context.runtime
3296+
.startStoredThreadTurn(candidate.id, input.text)
3297+
.pipe(Effect.mapError((cause) => mapCodexRuntimeError(threadId, "turn/start", cause)));
3298+
return { turnId: response.turn.id };
3299+
});
3300+
32313301
const readSubagentTranscript: NonNullable<CodexAdapterShape["readSubagentTranscript"]> =
32323302
Effect.fn("readSubagentTranscript")(function* (threadId, input) {
32333303
const requestError = (detail: string) =>
@@ -3237,42 +3307,10 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* (
32373307
detail,
32383308
});
32393309
const context = yield* requireSession(threadId);
3240-
const rootThreadId = yield* context.runtime.readProviderThreadId.pipe(
3241-
Effect.mapError((cause) => mapCodexRuntimeError(threadId, "thread/read", cause)),
3242-
);
3243-
if (input.agentId === rootThreadId) {
3244-
return yield* requestError("The requested transcript belongs to the parent thread.");
3245-
}
3246-
3247-
const readStoredThreadMetadata = (providerThreadId: string) =>
3248-
context.runtime
3249-
.readStoredThreadMetadata(providerThreadId)
3250-
.pipe(Effect.mapError((cause) => mapCodexRuntimeError(threadId, "thread/read", cause)));
3251-
const candidate = yield* readStoredThreadMetadata(input.agentId);
3252-
const visited = new Set<string>([candidate.id]);
3253-
let current = candidate;
3254-
let authorized = false;
3255-
3256-
for (let depth = 0; depth < CODEX_SUBAGENT_MAX_ANCESTRY_DEPTH; depth += 1) {
3257-
const parentThreadId = readCodexSubagentParentThreadId(current);
3258-
if (parentThreadId === rootThreadId) {
3259-
authorized = true;
3260-
break;
3261-
}
3262-
if (!parentThreadId || visited.has(parentThreadId)) {
3263-
return yield* requestError(
3264-
`Codex thread '${input.agentId}' is not a subagent of this conversation.`,
3265-
);
3266-
}
3267-
visited.add(parentThreadId);
3268-
current = yield* readStoredThreadMetadata(parentThreadId);
3269-
}
3270-
3271-
if (!authorized) {
3272-
return yield* requestError(
3273-
`Codex thread '${input.agentId}' exceeded the supported subagent nesting depth.`,
3274-
);
3275-
}
3310+
const candidate = yield* authorizeSubagentThread(context, threadId, input.agentId, {
3311+
requestError,
3312+
parentThreadDetail: "The requested transcript belongs to the parent thread.",
3313+
});
32763314

32773315
// Legacy threads expose their stored turns through `thread/read`. The
32783316
// cursor API only exists for Codex's explicit paginated history mode;
@@ -3638,6 +3676,7 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* (
36383676
clearThreadGoal,
36393677
readThread,
36403678
readSubagentTranscript,
3679+
sendSubagentInput,
36413680
rollbackThread,
36423681
deleteThread,
36433682
respondToRequest,

apps/server/src/provider/Layers/CodexSessionRuntime.ts

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -284,6 +284,13 @@ export interface CodexSessionRuntimeShape {
284284
readonly readStoredThreadItems: (
285285
input: EffectCodexSchema.V2ThreadItemsListParams,
286286
) => Effect.Effect<EffectCodexSchema.V2ThreadItemsListResponse, CodexSessionRuntimeError>;
287+
/** Start a turn on another loaded provider thread (a spawned subagent) with
288+
* a plain text message. Callers must authorize the thread first; the
289+
* app-server rejects threads that do not accept direct input. */
290+
readonly startStoredThreadTurn: (
291+
providerThreadId: string,
292+
text: string,
293+
) => Effect.Effect<EffectCodexSchema.V2TurnStartResponse, CodexSessionRuntimeError>;
287294
readonly rollbackThread: (
288295
numTurns: number,
289296
) => Effect.Effect<CodexThreadSnapshot, CodexSessionRuntimeError>;
@@ -2428,6 +2435,11 @@ export const makeCodexSessionRuntime = (
24282435
})
24292436
.pipe(Effect.map((response) => response.thread)),
24302437
readStoredThreadItems: (input) => client.request("thread/items/list", input),
2438+
startStoredThreadTurn: (providerThreadId, text) =>
2439+
client.request("turn/start", {
2440+
threadId: providerThreadId,
2441+
input: [{ type: "text", text }],
2442+
}),
24312443
rollbackThread: (numTurns) =>
24322444
Effect.gen(function* () {
24332445
const providerThreadId = yield* readProviderThreadId;

apps/server/src/provider/Layers/ProviderService.ts

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@ import {
2626
ProviderSendTurnInput,
2727
ProviderSessionStartInput,
2828
ProviderStartReviewInput,
29+
ProviderSubagentInputRequest,
2930
ProviderSubagentTranscriptInput,
3031
ProviderSteerTurnInput,
3132
ProviderStopSessionInput,
@@ -1876,6 +1877,34 @@ const makeProviderService = Effect.fn("makeProviderService")(function* (
18761877
return yield* readTranscript(routed.threadId, input);
18771878
});
18781879

1880+
const sendSubagentInput: ProviderServiceShape["sendSubagentInput"] = Effect.fn(
1881+
"sendSubagentInput",
1882+
)(function* (rawInput) {
1883+
const input = yield* decodeInputOrValidationError({
1884+
operation: "ProviderService.sendSubagentInput",
1885+
schema: ProviderSubagentInputRequest,
1886+
payload: rawInput,
1887+
});
1888+
const routed = yield* resolveRoutableSession({
1889+
threadId: input.threadId,
1890+
operation: "ProviderService.sendSubagentInput",
1891+
allowRecovery: false,
1892+
});
1893+
yield* Effect.annotateCurrentSpan({
1894+
"provider.operation": "send-subagent-input",
1895+
"provider.kind": routed.adapter.provider,
1896+
"provider.thread_id": input.threadId,
1897+
});
1898+
const send = routed.adapter.sendSubagentInput;
1899+
if (send === undefined) {
1900+
return yield* toValidationError(
1901+
"ProviderService.sendSubagentInput",
1902+
`Provider '${routed.adapter.provider}' does not accept direct input to subagents.`,
1903+
);
1904+
}
1905+
return yield* send(routed.threadId, input);
1906+
});
1907+
18791908
const resolveSubagentWorktree: ProviderServiceShape["resolveSubagentWorktree"] = Effect.fn(
18801909
"resolveSubagentWorktree",
18811910
)(function* (input) {
@@ -2019,6 +2048,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* (
20192048
getInstanceInfo,
20202049
rollbackConversation,
20212050
readSubagentTranscript,
2051+
sendSubagentInput,
20222052
resolveSubagentWorktree,
20232053
deleteThread,
20242054
// Each access creates a fresh PubSub subscription so that multiple

apps/server/src/provider/Layers/ProviderSessionReaper.test.ts

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -233,6 +233,7 @@ describe("ProviderSessionReaper", () => {
233233
},
234234
rollbackConversation: () => unsupported(),
235235
readSubagentTranscript: () => unsupported(),
236+
sendSubagentInput: () => unsupported(),
236237
resolveSubagentWorktree: () => Effect.succeed(null),
237238
deleteThread: () => unsupported(),
238239
streamEvents: Stream.empty,

apps/server/src/provider/Services/ProviderAdapter.ts

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,8 @@ import type {
2424
ProviderRealtimeAppendAudioInput,
2525
ProviderRealtimeListVoicesResult,
2626
ProviderRealtimeOutputModality,
27+
ProviderSubagentInputRequest,
28+
ProviderSubagentInputResult,
2729
ProviderSubagentTranscriptInput,
2830
ProviderSubagentTranscriptResult,
2931
ProviderSteerTurnInput,
@@ -283,6 +285,16 @@ export interface ProviderAdapterShape<TError> {
283285
input: ProviderSubagentTranscriptInput,
284286
) => Effect.Effect<ProviderSubagentTranscriptResult, TError>;
285287

288+
/**
289+
* Send a user message straight to a spawned subagent, starting a turn on
290+
* its own thread. Optional: only providers whose runtime accepts direct
291+
* input to a child (Codex) implement it.
292+
*/
293+
readonly sendSubagentInput?: (
294+
threadId: ThreadId,
295+
input: ProviderSubagentInputRequest,
296+
) => Effect.Effect<ProviderSubagentInputResult, TError>;
297+
286298
/**
287299
* Where a spawned subagent is working, when it was given its own checkout.
288300
*

apps/server/src/provider/Services/ProviderService.ts

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,8 @@ import type {
3636
RuntimeThreadGoalSnapshot,
3737
ThreadGoalStatus,
3838
ThreadId,
39+
ProviderSubagentInputRequest,
40+
ProviderSubagentInputResult,
3941
ProviderSubagentTranscriptInput,
4042
ProviderSubagentTranscriptResult,
4143
ProviderTurnStartResult,
@@ -202,6 +204,12 @@ export interface ProviderServiceShape {
202204
input: ProviderSubagentTranscriptInput,
203205
) => Effect.Effect<ProviderSubagentTranscriptResult, ProviderServiceError>;
204206

207+
/** Send a user message straight to a spawned subagent (see the adapter's
208+
* `sendSubagentInput`). */
209+
readonly sendSubagentInput: (
210+
input: ProviderSubagentInputRequest,
211+
) => Effect.Effect<ProviderSubagentInputResult, ProviderServiceError>;
212+
205213
/**
206214
* Where an isolated subagent is working, or null when that is unknown —
207215
* because the provider does not record it, the session is not routable, or

0 commit comments

Comments
 (0)