fix(server): parents handle steering while waiting for child work

apps/server/src/mcp/OrchestratorMcpService.test.ts

@@ -1273,6 +1273,7 @@ describe("OrchestratorMcpService provider resolution", () => {
 
   it.effect.each([
     "parent completion",
+    "caller steering",
     "nested child completion",
     "child run completion",
     "child question",
@@ -1285,15 +1286,19 @@ describe("OrchestratorMcpService provider resolution", () => {
       const nestedId = NodeId.make("node:mcp-wait-nested");
       const childRunId = RunId.make("run:mcp-wait-child");
       const childRootId = NodeId.make("node:mcp-wait-child-root");
-      const eventThreadId = wake === "parent completion" ? parentThreadId : childThreadId;
+      const eventThreadId =
+        wake === "parent completion" || wake === "caller steering" ? parentThreadId : childThreadId;
       const eventType =
-        wake === "child run completion"
-          ? "run.updated"
-          : wake === "child question" || wake === "child approval"
-            ? "runtime-request.updated"
-            : "subagent.updated";
+        wake === "caller steering"
+          ? "turn-item.updated"
+          : wake === "child run completion"
+            ? "run.updated"
+            : wake === "child question" || wake === "child approval"
+              ? "runtime-request.updated"
+              : "subagent.updated";
       const terminal = wake.endsWith("completion");
       let ready = false;
+      const commands: Array<OrchestrationV2ServerCommand> = [];
       const task = () => ({
         id: taskId,
         threadId: parentThreadId,
@@ -1326,7 +1331,7 @@ describe("OrchestratorMcpService provider resolution", () => {
               : [],
           nodes: [{ id: childRootId, runId: childRunId }],
           runtimeRequests:
-            ready && !terminal
+            ready && !terminal && wake !== "caller steering"
               ? [
                   {
                     id: "request:mcp-wait-answer",
@@ -1357,11 +1362,13 @@ describe("OrchestratorMcpService provider resolution", () => {
                   Stream.concat(Stream.fromQueue(events)),
                 )
               : Stream.never,
-          dispatch: () =>
-            Effect.succeed({
+          dispatch: (command) => {
+            commands.push(command);
+            return Effect.succeed({
               sequence: 1,
               storedEvents: [{ event: { type: "subagent.updated", payload: task() } }],
-            } as never),
+            } as never);
+          },
         }),
         ServerSettings.layerTest(),
         providerRegistryLayer([
@@ -1417,7 +1424,15 @@ describe("OrchestratorMcpService provider resolution", () => {
           event: {
             type: eventType,
             threadId: eventThreadId,
-            payload: { id: wake === "parent completion" ? taskId : nestedId },
+            payload:
+              wake === "caller steering"
+                ? {
+                    type: "user_message",
+                    runId: parentRunId,
+                    messageId: MessageId.make("message:mcp-wait-steer"),
+                    inputIntent: "steer",
+                  }
+                : { id: wake === "parent completion" ? taskId : nestedId },
           },
         } as unknown as OrchestrationV2StoredEvent);
         const result = yield* Fiber.join(waiting);
@@ -1427,7 +1442,17 @@ describe("OrchestratorMcpService provider resolution", () => {
           if (wake === "parent completion") assert.equal(result.summary, "done");
         } else {
           assert.equal(result.status, "running");
-          if (wake === "child question") assert.isTrue(result.hasPendingQuestion);
+          if (wake === "caller steering") {
+            assert.equal(result.interruptedByMessageId, "message:mcp-wait-steer");
+            assert.notProperty(result, "next");
+            assert.isTrue(
+              commands.some(
+                (command) =>
+                  command.type === "delegated_task.wake-policy" &&
+                  command.completionWake === "always",
+              ),
+            );
+          } else if (wake === "child question") assert.isTrue(result.hasPendingQuestion);
           else {
             assert.deepEqual(result.pendingApproval, {
               requestId: "request:mcp-wait-answer",

apps/server/src/mcp/OrchestratorMcpService.ts

@@ -136,6 +136,7 @@ const DEFAULT_TASK_WAIT_TIMEOUT_MS = 4 * 60 * 1_000;
 const MAX_WAIT_TIMEOUT_MS = 60 * 60 * 1_000;
 const TASK_WAKE_EVENTS = [
   { thread: "parent", eventType: "subagent.updated" },
+  { thread: "parent", eventType: "turn-item.updated" },
   { thread: "child", eventType: "run.updated" },
   { thread: "child", eventType: "subagent.updated" },
   { thread: "child", eventType: "provider-thread.updated" },
@@ -1768,7 +1769,7 @@ const make = Effect.gen(function* () {
         Exit.isSuccess(exit) && delivered(exit.value) ? Effect.void : upgradeWake,
       ),
     );
-    if (Option.isSome(waited) && delivered(waited)) return waited.value;
+    if (Option.isSome(waited)) return waited.value;
     return yield* readTask(scope, taskId, Option.isNone(waited), true);
   });
 
@@ -1892,18 +1893,30 @@ const make = Effect.gen(function* () {
             })
             .pipe(
               Stream.filter(
-                ({ event }) =>
+                (stored) =>
                   thread === "child" ||
-                  event.type !== "subagent.updated" ||
-                  event.payload.id === taskId,
+                  (stored.event.type === "turn-item.updated"
+                    ? ThreadManagementService.steeringInputMessage(stored) !== undefined
+                    : stored.event.type !== "subagent.updated" ||
+                      stored.event.payload.id === taskId),
               ),
             ),
         ),
         { concurrency: "unbounded" },
       ).pipe(
         Stream.mapError(streamError),
-        Stream.mapEffect(() => readTask(scope, taskId, false, true)),
-        Stream.filter(endsTaskWait),
+        Stream.mapEffect((stored) =>
+          Effect.gen(function* () {
+            const result = yield* readTask(scope, taskId, false, true);
+            const messageId = ThreadManagementService.steeringInputMessage(stored);
+            if (messageId === undefined) return result;
+            const { next: _next, ...current } = result;
+            return { ...current, interruptedByMessageId: messageId };
+          }),
+        ),
+        Stream.filter(
+          (result) => result.interruptedByMessageId !== undefined || endsTaskWait(result),
+        ),
         Stream.runHead,
       );
     },
@@ -3220,7 +3233,15 @@ const make = Effect.gen(function* () {
       } satisfies OrchestratorMcpThreadSendResult;
     }),
     waitForThread: Effect.fn("OrchestratorMcpService.waitForThread")(function* (scope, input) {
-      const { parent, target } = yield* loadScopedThread(scope, input.threadId, []);
+      const { parent, target } = yield* loadScopedThread(scope, input.threadId, ["runs"]);
+      const callerRun =
+        parent === undefined
+          ? undefined
+          : ThreadManagementService.latestActiveRun({
+              runs: parent.runs.filter(
+                (run) => run.providerInstanceId === scope.thread?.providerInstanceId,
+              ),
+            });
       const result = yield* threadManagement
         .waitForThread({
           projectId: target.thread.projectId,
@@ -3231,6 +3252,11 @@ const make = Effect.gen(function* () {
             Math.max(1, input.timeoutMs ?? DEFAULT_WAIT_TIMEOUT_MS),
           ),
           ...(input.untilDelegatedWorkSettled === true ? { untilDelegatedWorkSettled: true } : {}),
+          ...(callerRun === undefined || parent === undefined
+            ? {}
+            : {
+                caller: { threadId: parent.thread.id, runId: callerRun.id },
+              }),
         })
         .pipe(Effect.mapError(threadManagementFailure));
       const status = result.run === null ? "idle" : runOutcome(result.run);
@@ -3261,6 +3287,11 @@ const make = Effect.gen(function* () {
         runId: result.run?.id ?? null,
         status,
         timedOut: result.timedOut,
+        ...(result.interruptedByMessageId === undefined
+          ? {}
+          : {
+              interruptedByMessageId: result.interruptedByMessageId,
+            }),
         ...(wakesCaller ? { next: ORCHESTRATOR_MCP_RUNNING_TASK_NEXT } : {}),
         ...(result.resultRun === undefined
           ? {}

apps/server/src/mcp/toolkits/orchestrator/tools.ts

@@ -290,7 +290,7 @@ const ThreadSendTool = Tool.make("t3_thread_send", {
 
 const ThreadWaitTool = Tool.make("t3_thread_wait", {
   description:
-    "Wait for a thread's run to finish. No `runId`: latest. Idle: at once. On timeout, with `next` end your turn; without, call again.",
+    "Wait (latest run by default). Idle returns. interruptedByMessageId: handle input; target runs. Timeout: next ends turn; else retry.",
   parameters: OrchestratorMcpThreadWaitInput,
   success: OrchestratorMcpThreadWaitResult,
   failure: OrchestratorMcpFailure,

apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts

@@ -333,6 +333,7 @@ const resumed = (
     readonly resolveExtractedTextPath?: (path: string) => Effect.Effect<string | undefined>;
     readonly fileSystem?: FileSystem.FileSystem;
     readonly onPrompt?: (input: unknown) => void;
+    readonly onRequest?: (operation: string) => void;
     readonly borrowers?: { current: number };
   },
 ) =>
@@ -371,6 +372,7 @@ const resumed = (
         ...(options?.borrowers === undefined ? {} : { borrowers: options.borrowers }),
         ...(options?.fileSystem === undefined ? {} : { fileSystem: options.fileSystem }),
         ...(options?.onPrompt === undefined ? {} : { onPrompt: options.onPrompt }),
+        ...(options?.onRequest === undefined ? {} : { onRequest: options.onRequest }),
         ...(options?.external === undefined ? {} : { external: options.external }),
         ...(options?.replayGate === undefined ? {} : { replayGate: options.replayGate }),
         ...(options?.attachmentsDir === undefined
@@ -1648,6 +1650,132 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => {
     event("session.execution.started", { sessionID: CHILD }),
   ];
 
+  it.effect.each([false, true])(
+    "answers steering while foreground subagents keep running, active goal=%s",
+    (goal) =>
+      Effect.gen(function* () {
+        const gate = makeProviderReplayGate(["release-child"]);
+        const requests: Array<string> = [];
+        const steerId = `msg_t3_steer_${SESSION}:message:opencode2-adapter:steer`;
+        const foreground = backgroundLaunch(CHILD).map((entry) => {
+          if (entry.type !== "emit_inbound" || !entry.frame || typeof entry.frame !== "object")
+            return entry;
+          const frame = entry.frame as { event?: { type: string; data: Record<string, unknown> } };
+          if (frame.event?.type !== "session.tool.called") return entry;
+          return event("session.tool.called", {
+            ...frame.event.data,
+            input: { description: "Sleep", prompt: "sleep", background: false },
+          });
+        });
+        const { runtime, thread } = yield* resumed(
+          [
+            ...foreground,
+            ...(goal
+              ? [
+                  event("rpc.opencode.goal.changed", {
+                    sessionID: SESSION,
+                    goal: nativeGoal("active"),
+                  }),
+                ]
+              : []),
+            event("session.execution.started", { sessionID: CHILD }),
+            out("session.prompt", { sessionID: SESSION, id: steerId, delivery: "steer" }),
+            replyData("session.prompt", {
+              id: steerId,
+              sessionID: SESSION,
+              time: { created: 1790656601500 },
+              type: "user",
+              payload: { text: "Also say STEERED." },
+              delivery: "steer",
+            }),
+            out("session.background", { sessionID: SESSION }),
+            reply("session.background", null),
+            returnedRunning("call-background"),
+            event("session.inbox.delivered", { sessionID: SESSION, inboxID: steerId }),
+            event("session.text.ended", {
+              sessionID: SESSION,
+              assistantMessageID: "msg_steer_reply",
+              ordinal: 0,
+              text: "STEERED",
+            }),
+            event("session.execution.succeeded", { sessionID: SESSION }),
+            event("session.execution.succeeded", { sessionID: CHILD }, "release-child"),
+            childReport("msg_child_report", "CHILD_DONE"),
+            ...(goal
+              ? [
+                  event("rpc.opencode.goal.changed", {
+                    sessionID: SESSION,
+                    goal: nativeGoal("complete"),
+                  }),
+                ]
+              : []),
+          ],
+          {
+            version: "2.2.5",
+            replayGate: gate,
+            onRequest: (operation) => requests.push(operation),
+          },
+        );
+        const childStarted = yield* Deferred.make<void>();
+        const childDone = yield* Deferred.make<void>();
+        const answered = yield* Deferred.make<void>();
+        const parentEnded = yield* Deferred.make<void>();
+        const terminals: Array<ProviderAdapter.ProviderAdapterV2Event> = [];
+        let childStatus: string | undefined;
+        let callStatus: string | undefined;
+        let result: string | null = null;
+        yield* runtime.events.pipe(
+          Stream.runForEach((e) => {
+            if (e.type === "turn.terminal") {
+              terminals.push(e);
+              return Deferred.succeed(parentEnded, undefined);
+            }
+            if (
+              e.type === "provider_turn.updated" &&
+              e.providerTurn.nativeTurnRef?.nativeId?.startsWith(`${CHILD}:turn:`)
+            ) {
+              childStatus = e.providerTurn.status;
+              if (childStatus === "running") return Deferred.succeed(childStarted, undefined);
+            }
+            if (e.type === "subagent.updated" && e.subagent.title === "Sleep") {
+              callStatus = e.subagent.status;
+              result = e.subagent.result;
+              if (callStatus === "completed") return Deferred.succeed(childDone, undefined);
+            }
+            return e.type === "message.updated" && e.message.text === "STEERED"
+              ? Deferred.succeed(answered, undefined)
+              : Effect.void;
+          }),
+          Effect.forkScoped,
+        );
+        yield* runtime.startTurn(withLineage(thread));
+        yield* Deferred.await(childStarted);
+        yield* runtime.steerTurn({
+          threadId,
+          runId: RunId.make("run:opencode2-adapter"),
+          providerThread: thread,
+          providerTurnId: yield* providerTurnId,
+          message: {
+            ...turnInput(thread).message,
+            messageId: MessageId.make("message:opencode2-adapter:steer"),
+            text: "Also say STEERED.",
+          },
+        });
+        assert.include(requests, "session.background");
+        assert.isTrue(yield* runtime.hasPendingBackgroundWorkForThread!(thread));
+        yield* Deferred.await(answered);
+        yield* Effect.promise(() => gate.waitForReached("release-child"));
+        if (!goal) yield* Deferred.await(parentEnded);
+        assert.equal(childStatus, "running");
+        assert.equal(callStatus, "running");
+        assert.lengthOf(terminals, goal ? 0 : 1);
+        gate.release("release-child");
+        yield* Deferred.await(childDone);
+        assert.equal(result, "CHILD_DONE");
+        yield* Deferred.await(parentEnded);
+      }).pipe(Effect.scoped),
+  );
+
   it.effect.each([false, true])(
     "settles a failed background subagent with recovered child=%s",
     (recoveredChild) =>