feat(agents): message a Codex subagent straight from its transcript

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

@@ -152,6 +152,7 @@ function createProviderServiceHarness(
       }),
     rollbackConversation,
     readSubagentTranscript: () => unsupported(),
+    sendSubagentInput: () => unsupported(),
     resolveSubagentWorktree: () => Effect.succeed(null),
     deleteThread: () => unsupported(),
     get streamEvents() {

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

@@ -444,6 +444,7 @@ describe("ProviderCommandReactor", () => {
       },
       rollbackConversation: () => unsupported(),
       readSubagentTranscript: () => unsupported(),
+      sendSubagentInput: () => unsupported(),
       resolveSubagentWorktree: () => Effect.succeed(null),
       deleteThread: () => unsupported(),
       get streamEvents() {

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

@@ -141,6 +141,7 @@ function createProviderServiceHarness() {
     },
     rollbackConversation: () => unsupported(),
     readSubagentTranscript: () => unsupported(),
+    sendSubagentInput: () => unsupported(),
     resolveSubagentWorktree: ({ toolUseId }) =>
       Effect.succeed(subagentWorktreesByToolUseId.get(toolUseId) ?? null),
     deleteThread: () => unsupported(),

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

@@ -346,6 +346,11 @@ class FakeCodexRuntime implements CodexSessionRuntimeShape {
     ): Promise<EffectCodexSchema.V2ThreadItemsListResponse> => Promise.resolve({ data: [] }),
   );
 
+  public readonly startStoredThreadTurnImpl = vi.fn(
+    (_providerThreadId: string, _text: string): Promise<EffectCodexSchema.V2TurnStartResponse> =>
+      Promise.resolve({ turn: { id: "turn-direct-1", items: [], status: "inProgress" } }),
+  );
+
   public readonly rollbackThreadImpl = vi.fn((_numTurns: number): Promise<CodexThreadSnapshot> =>
     Promise.resolve({
       threadId: "provider-thread-1",
@@ -431,6 +436,10 @@ class FakeCodexRuntime implements CodexSessionRuntimeShape {
     return Effect.promise(() => this.readStoredThreadItemsImpl(input));
   }
 
+  startStoredThreadTurn(providerThreadId: string, text: string) {
+    return Effect.promise(() => this.startStoredThreadTurnImpl(providerThreadId, text));
+  }
+
   rollbackThread(numTurns: number) {
     return Effect.promise(() => this.rollbackThreadImpl(numTurns));
   }
@@ -875,6 +884,49 @@ transcriptLayer("CodexAdapterLive subagent transcripts", (it) => {
       assert.match(result.failure.detail, /not a subagent of this conversation/);
     }),
   );
+
+  it.effect("starts a turn on an authorized subagent thread for direct input", () =>
+    Effect.gen(function* () {
+      const adapter = yield* CodexAdapter;
+      const threadId = asThreadId("thread-direct-input");
+      yield* adapter.startSession({
+        provider: ProviderDriverKind.make("codex"),
+        threadId,
+        runtimeMode: "full-access",
+      });
+      const runtime = transcriptRuntimeFactory.lastRuntime;
+      assert.ok(runtime);
+
+      const sendSubagentInput = adapter.sendSubagentInput;
+      assert.ok(sendSubagentInput);
+      const result = yield* sendSubagentInput(threadId, {
+        threadId,
+        agentId: "child-provider-thread",
+        text: "Focus on the router first.",
+      });
+      assert.deepStrictEqual(result, { turnId: "turn-direct-1" });
+      assert.deepStrictEqual(runtime.startStoredThreadTurnImpl.mock.calls, [
+        ["child-provider-thread", "Focus on the router first."],
+      ]);
+
+      // A child the app-server says cannot take input is refused before any turn starts.
+      runtime.readStoredThreadMetadataImpl.mockImplementation((providerThreadId: string) =>
+        Promise.resolve({
+          ...makeStoredThread({ id: providerThreadId, parentThreadId: "provider-thread-1" }),
+          canAcceptDirectInput: false,
+        }),
+      );
+      const refused = yield* sendSubagentInput(threadId, {
+        threadId,
+        agentId: "child-provider-thread",
+        text: "Anyone there?",
+      }).pipe(Effect.result);
+      assert.ok(refused._tag === "Failure");
+      assert.ok(refused.failure instanceof ProviderAdapterRequestError);
+      assert.match(refused.failure.detail, /does not accept direct input/);
+      assert.equal(runtime.startStoredThreadTurnImpl.mock.calls.length, 1);
+    }),
+  );
 });
 
 const sessionRuntimeFactory = makeRuntimeFactory();

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

@@ -3228,6 +3228,76 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* (
       })),
     );
 
+  /** Resolves a stored provider thread the session may act on: one whose
+   *  ancestry reaches this session's root within the supported nesting depth.
+   *  Returns the thread's metadata; anything else is a request error. */
+  const authorizeSubagentThread = Effect.fn("authorizeSubagentThread")(function* (
+    context: CodexAdapterSessionContext,
+    threadId: ThreadId,
+    agentId: string,
+    options: {
+      readonly requestError: (detail: string) => ProviderAdapterRequestError;
+      readonly parentThreadDetail: string;
+    },
+  ) {
+    const { requestError } = options;
+    const rootThreadId = yield* context.runtime.readProviderThreadId.pipe(
+      Effect.mapError((cause) => mapCodexRuntimeError(threadId, "thread/read", cause)),
+    );
+    if (agentId === rootThreadId) {
+      return yield* requestError(options.parentThreadDetail);
+    }
+
+    const readStoredThreadMetadata = (providerThreadId: string) =>
+      context.runtime
+        .readStoredThreadMetadata(providerThreadId)
+        .pipe(Effect.mapError((cause) => mapCodexRuntimeError(threadId, "thread/read", cause)));
+    const candidate = yield* readStoredThreadMetadata(agentId);
+    const visited = new Set<string>([candidate.id]);
+    let current = candidate;
+
+    for (let depth = 0; depth < CODEX_SUBAGENT_MAX_ANCESTRY_DEPTH; depth += 1) {
+      const parentThreadId = readCodexSubagentParentThreadId(current);
+      if (parentThreadId === rootThreadId) {
+        return candidate;
+      }
+      if (!parentThreadId || visited.has(parentThreadId)) {
+        return yield* requestError(
+          `Codex thread '${agentId}' is not a subagent of this conversation.`,
+        );
+      }
+      visited.add(parentThreadId);
+      current = yield* readStoredThreadMetadata(parentThreadId);
+    }
+
+    return yield* requestError(
+      `Codex thread '${agentId}' exceeded the supported subagent nesting depth.`,
+    );
+  });
+
+  const sendSubagentInput: NonNullable<CodexAdapterShape["sendSubagentInput"]> = Effect.fn(
+    "sendSubagentInput",
+  )(function* (threadId, input) {
+    const requestError = (detail: string) =>
+      new ProviderAdapterRequestError({
+        provider: PROVIDER,
+        method: "sendSubagentInput",
+        detail,
+      });
+    const context = yield* requireSession(threadId);
+    const candidate = yield* authorizeSubagentThread(context, threadId, input.agentId, {
+      requestError,
+      parentThreadDetail: "Send to the parent thread through the composer instead.",
+    });
+    if (candidate.canAcceptDirectInput === false) {
+      return yield* requestError("This agent does not accept direct input right now.");
+    }
+    const response = yield* context.runtime
+      .startStoredThreadTurn(candidate.id, input.text)
+      .pipe(Effect.mapError((cause) => mapCodexRuntimeError(threadId, "turn/start", cause)));
+    return { turnId: response.turn.id };
+  });
+
   const readSubagentTranscript: NonNullable<CodexAdapterShape["readSubagentTranscript"]> =
     Effect.fn("readSubagentTranscript")(function* (threadId, input) {
       const requestError = (detail: string) =>
@@ -3237,42 +3307,10 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* (
           detail,
         });
       const context = yield* requireSession(threadId);
-      const rootThreadId = yield* context.runtime.readProviderThreadId.pipe(
-        Effect.mapError((cause) => mapCodexRuntimeError(threadId, "thread/read", cause)),
-      );
-      if (input.agentId === rootThreadId) {
-        return yield* requestError("The requested transcript belongs to the parent thread.");
-      }
-
-      const readStoredThreadMetadata = (providerThreadId: string) =>
-        context.runtime
-          .readStoredThreadMetadata(providerThreadId)
-          .pipe(Effect.mapError((cause) => mapCodexRuntimeError(threadId, "thread/read", cause)));
-      const candidate = yield* readStoredThreadMetadata(input.agentId);
-      const visited = new Set<string>([candidate.id]);
-      let current = candidate;
-      let authorized = false;
-
-      for (let depth = 0; depth < CODEX_SUBAGENT_MAX_ANCESTRY_DEPTH; depth += 1) {
-        const parentThreadId = readCodexSubagentParentThreadId(current);
-        if (parentThreadId === rootThreadId) {
-          authorized = true;
-          break;
-        }
-        if (!parentThreadId || visited.has(parentThreadId)) {
-          return yield* requestError(
-            `Codex thread '${input.agentId}' is not a subagent of this conversation.`,
-          );
-        }
-        visited.add(parentThreadId);
-        current = yield* readStoredThreadMetadata(parentThreadId);
-      }
-
-      if (!authorized) {
-        return yield* requestError(
-          `Codex thread '${input.agentId}' exceeded the supported subagent nesting depth.`,
-        );
-      }
+      const candidate = yield* authorizeSubagentThread(context, threadId, input.agentId, {
+        requestError,
+        parentThreadDetail: "The requested transcript belongs to the parent thread.",
+      });
 
       // Legacy threads expose their stored turns through `thread/read`. The
       // cursor API only exists for Codex's explicit paginated history mode;
@@ -3638,6 +3676,7 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* (
     clearThreadGoal,
     readThread,
     readSubagentTranscript,
+    sendSubagentInput,
     rollbackThread,
     deleteThread,
     respondToRequest,

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

@@ -284,6 +284,13 @@ export interface CodexSessionRuntimeShape {
   readonly readStoredThreadItems: (
     input: EffectCodexSchema.V2ThreadItemsListParams,
   ) => Effect.Effect<EffectCodexSchema.V2ThreadItemsListResponse, CodexSessionRuntimeError>;
+  /** Start a turn on another loaded provider thread (a spawned subagent) with
+   *  a plain text message. Callers must authorize the thread first; the
+   *  app-server rejects threads that do not accept direct input. */
+  readonly startStoredThreadTurn: (
+    providerThreadId: string,
+    text: string,
+  ) => Effect.Effect<EffectCodexSchema.V2TurnStartResponse, CodexSessionRuntimeError>;
   readonly rollbackThread: (
     numTurns: number,
   ) => Effect.Effect<CodexThreadSnapshot, CodexSessionRuntimeError>;
@@ -2428,6 +2435,11 @@ export const makeCodexSessionRuntime = (
           })
           .pipe(Effect.map((response) => response.thread)),
       readStoredThreadItems: (input) => client.request("thread/items/list", input),
+      startStoredThreadTurn: (providerThreadId, text) =>
+        client.request("turn/start", {
+          threadId: providerThreadId,
+          input: [{ type: "text", text }],
+        }),
       rollbackThread: (numTurns) =>
         Effect.gen(function* () {
           const providerThreadId = yield* readProviderThreadId;

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

@@ -26,6 +26,7 @@ import {
   ProviderSendTurnInput,
   ProviderSessionStartInput,
   ProviderStartReviewInput,
+  ProviderSubagentInputRequest,
   ProviderSubagentTranscriptInput,
   ProviderSteerTurnInput,
   ProviderStopSessionInput,
@@ -1876,6 +1877,34 @@ const makeProviderService = Effect.fn("makeProviderService")(function* (
     return yield* readTranscript(routed.threadId, input);
   });
 
+  const sendSubagentInput: ProviderServiceShape["sendSubagentInput"] = Effect.fn(
+    "sendSubagentInput",
+  )(function* (rawInput) {
+    const input = yield* decodeInputOrValidationError({
+      operation: "ProviderService.sendSubagentInput",
+      schema: ProviderSubagentInputRequest,
+      payload: rawInput,
+    });
+    const routed = yield* resolveRoutableSession({
+      threadId: input.threadId,
+      operation: "ProviderService.sendSubagentInput",
+      allowRecovery: false,
+    });
+    yield* Effect.annotateCurrentSpan({
+      "provider.operation": "send-subagent-input",
+      "provider.kind": routed.adapter.provider,
+      "provider.thread_id": input.threadId,
+    });
+    const send = routed.adapter.sendSubagentInput;
+    if (send === undefined) {
+      return yield* toValidationError(
+        "ProviderService.sendSubagentInput",
+        `Provider '${routed.adapter.provider}' does not accept direct input to subagents.`,
+      );
+    }
+    return yield* send(routed.threadId, input);
+  });
+
   const resolveSubagentWorktree: ProviderServiceShape["resolveSubagentWorktree"] = Effect.fn(
     "resolveSubagentWorktree",
   )(function* (input) {
@@ -2019,6 +2048,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* (
     getInstanceInfo,
     rollbackConversation,
     readSubagentTranscript,
+    sendSubagentInput,
     resolveSubagentWorktree,
     deleteThread,
     // Each access creates a fresh PubSub subscription so that multiple

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

@@ -233,6 +233,7 @@ describe("ProviderSessionReaper", () => {
       },
       rollbackConversation: () => unsupported(),
       readSubagentTranscript: () => unsupported(),
+      sendSubagentInput: () => unsupported(),
       resolveSubagentWorktree: () => Effect.succeed(null),
       deleteThread: () => unsupported(),
       streamEvents: Stream.empty,

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

@@ -24,6 +24,8 @@ import type {
   ProviderRealtimeAppendAudioInput,
   ProviderRealtimeListVoicesResult,
   ProviderRealtimeOutputModality,
+  ProviderSubagentInputRequest,
+  ProviderSubagentInputResult,
   ProviderSubagentTranscriptInput,
   ProviderSubagentTranscriptResult,
   ProviderSteerTurnInput,
@@ -283,6 +285,16 @@ export interface ProviderAdapterShape<TError> {
     input: ProviderSubagentTranscriptInput,
   ) => Effect.Effect<ProviderSubagentTranscriptResult, TError>;
 
+  /**
+   * Send a user message straight to a spawned subagent, starting a turn on
+   * its own thread. Optional: only providers whose runtime accepts direct
+   * input to a child (Codex) implement it.
+   */
+  readonly sendSubagentInput?: (
+    threadId: ThreadId,
+    input: ProviderSubagentInputRequest,
+  ) => Effect.Effect<ProviderSubagentInputResult, TError>;
+
   /**
    * Where a spawned subagent is working, when it was given its own checkout.
    *

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

@@ -36,6 +36,8 @@ import type {
   RuntimeThreadGoalSnapshot,
   ThreadGoalStatus,
   ThreadId,
+  ProviderSubagentInputRequest,
+  ProviderSubagentInputResult,
   ProviderSubagentTranscriptInput,
   ProviderSubagentTranscriptResult,
   ProviderTurnStartResult,
@@ -202,6 +204,12 @@ export interface ProviderServiceShape {
     input: ProviderSubagentTranscriptInput,
   ) => Effect.Effect<ProviderSubagentTranscriptResult, ProviderServiceError>;
 
+  /** Send a user message straight to a spawned subagent (see the adapter's
+   *  `sendSubagentInput`). */
+  readonly sendSubagentInput: (
+    input: ProviderSubagentInputRequest,
+  ) => Effect.Effect<ProviderSubagentInputResult, ProviderServiceError>;
+
   /**
    * Where an isolated subagent is working, or null when that is unknown —
    * because the provider does not record it, the session is not routable, or

apps/server/src/ws.ts

@@ -45,6 +45,7 @@ import {
   ProviderExtensionsError,
   ProviderExternalThreadError,
   ProviderRealtimeError,
+  ProviderSubagentInputError,
   ProviderSubagentTranscriptError,
   ThreadId,
   type TerminalEvent,
@@ -1112,6 +1113,22 @@ const makeWsRpcLayer = (currentSessionId: AuthSessionId) =>
             ),
             { "rpc.aggregate": "server" },
           ),
+        [WS_METHODS.serverSendSubagentInput]: (input) =>
+          observeRpcEffect(
+            WS_METHODS.serverSendSubagentInput,
+            providerService.sendSubagentInput(input).pipe(
+              Effect.mapError(
+                (error) =>
+                  new ProviderSubagentInputError({
+                    message:
+                      error.message.trim().length > 0
+                        ? error.message
+                        : "Failed to send the message to the agent.",
+                  }),
+              ),
+            ),
+            { "rpc.aggregate": "server" },
+          ),
         [WS_METHODS.serverListExternalProviderThreads]: (input) =>
           observeRpcEffect(
             WS_METHODS.serverListExternalProviderThreads,