fix(server): steering no longer cancels a pending question or approval

apps/server/src/orchestration-v2/CommandPolicy.test.ts

@@ -60,6 +60,7 @@ function dispatchProjection(
       sessionCapabilities === undefined
         ? []
         : [{ runAttemptId: activeAttemptId, status: "running" }],
+    runtimeRequests: [],
   } as unknown as OrchestrationV2ThreadProjection;
 }
 
@@ -244,6 +245,58 @@ it.each(["/compact", " /LOGOUT "])("queues a steer sent while %s runs", (text) =
   );
 });
 
+it.each(["user_input", "command"] as const)(
+  "queues every steer while the active turn waits on a pending %s request",
+  (kind) => {
+    const projection = dispatchProjection(baseCapabilities);
+    const waiting = {
+      ...projection,
+      runtimeRequests: [{ kind, status: "pending", responseCapability: { type: "live" } }],
+    } as unknown as OrchestrationV2ThreadProjection;
+    for (const deliveryIntent of ["auto", "steer"] as const) {
+      assert.deepEqual(
+        CommandPolicy.resolveMessageDispatchIntent(
+          waiting,
+          { type: "start_immediately" },
+          deliveryIntent,
+        ),
+        { type: "queue_after_active" },
+      );
+    }
+    assert.deepEqual(
+      CommandPolicy.resolveMessageDispatchIntent(waiting, {
+        type: "steer_active",
+        targetRunId: activeRunId,
+      }),
+      { type: "queue_after_active" },
+    );
+    assert.deepEqual(
+      CommandPolicy.resolveMessageDispatchIntent(waiting, { type: "start_immediately" }, "restart"),
+      { type: "restart_active", targetRunId: activeRunId },
+    );
+    // An answered request, or one answered by a later message, has no callback to abort.
+    for (const request of [
+      { kind, status: "resolved", responseCapability: { type: "live" } },
+      { kind, status: "pending", responseCapability: { type: "message" } },
+    ]) {
+      const nonBlocking = {
+        ...projection,
+        runtimeRequests: [request],
+      } as unknown as OrchestrationV2ThreadProjection;
+      for (const deliveryIntent of ["auto", "steer"] as const) {
+        assert.deepEqual(
+          CommandPolicy.resolveMessageDispatchIntent(
+            nonBlocking,
+            { type: "start_immediately" },
+            deliveryIntent,
+          ),
+          { type: "steer_active", targetRunId: activeRunId },
+        );
+      }
+    }
+  },
+);
+
 const layer = it.layer(CommandPolicy.layer);
 
 layer("CommandPolicyV2", (it) => {

apps/server/src/orchestration-v2/CommandPolicy.ts

@@ -90,13 +90,27 @@ export function goalCommandStopsActiveRun(
   }
 }
 
+/**
+ * A steer aborts the provider's pending question or approval callback before the person
+ * answers it, so while one is pending, any steer waits behind the turn instead. A request
+ * answered by a message has no callback and does not hold steers.
+ */
+export function waitsOnPendingRequest(
+  projection: Pick<OrchestrationV2ThreadProjection, "runtimeRequests">,
+): boolean {
+  return projection.runtimeRequests.some(
+    (request) => request.status === "pending" && request.responseCapability.type !== "message",
+  );
+}
+
 export function resolveMessageDispatchIntent(
   projection: OrchestrationV2ThreadProjection,
   requestedMode: MessageDispatchMode,
   deliveryIntent?: "auto" | "steer" | "restart",
 ): MessageDispatchMode {
   const mode = resolveDeliveryIntent(projection, requestedMode, deliveryIntent);
   if (mode.type !== "steer_active") return mode;
+  if (waitsOnPendingRequest(projection)) return { type: "queue_after_active" };
   // Compaction and sign-out cannot accept steers; queue follow-ups.
   const target = projection.runs.find((run) => run.id === mode.targetRunId);
   const targetMessage = projection.messages.find((message) => message.id === target?.userMessageId);

apps/server/src/orchestration-v2/Orchestrator.ts

@@ -114,6 +114,7 @@ import {
   isActiveRun,
   isNativeMaintenanceCommand,
   resolveMessageDispatchIntent,
+  waitsOnPendingRequest,
 } from "./CommandPolicy.ts";
 import { CommandReceiptStoreV2, type CommandReceiptV2 } from "./CommandReceiptStore.ts";
 import {
@@ -6737,6 +6738,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
           (message) => message.id === target.userMessageId,
         );
         if (
+          (resolvedMode.type === "restart_active" || !waitsOnPendingRequest(projection)) &&
           isLiveRun(target) &&
           goalCommandStopsActiveRun(goalCommand, goal?.status ?? null) &&
           (targetMessage === undefined || !isNativeMaintenanceCommand(targetMessage)) &&
@@ -10869,6 +10871,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
         payload: { ...queuedRun, queuePosition: 1 },
       });
       const stops =
+        !waitsOnPendingRequest(projection) &&
         target !== undefined &&
         isLiveRun(target) &&
         goalCommandStopsActiveRun(goalCommand, goal?.status ?? null) &&
@@ -10885,6 +10888,28 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
         now,
       });
     }
+    if (waitsOnPendingRequest(projection)) {
+      // It goes first in the queue instead and runs after the answer.
+      yield* emitEvent({
+        type: "run.updated",
+        threadId: command.threadId,
+        runId: queuedRun.id,
+        nodeId: queuedRootNode.id,
+        providerInstanceId: queuedRun.providerInstanceId,
+        occurredAt: now,
+        payload: { ...queuedRun, queuePosition: 1 },
+      });
+      return yield* stopRunForQueuedCommand({
+        command,
+        events,
+        effects,
+        projection,
+        queuedRunId: queuedRun.id,
+        targetRunId: undefined,
+        reason: "Queued behind a pending request",
+        now,
+      });
+    }
     if (
       (target?.status === "preparing" ||
         target?.status === "starting" ||

apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts

@@ -9,8 +9,10 @@ import {
   ProjectId,
   ProviderDriverKind,
   ProviderInstanceId,
+  ProviderSessionId,
   ProviderThreadId,
   ProviderTurnId,
+  RuntimeRequestId,
   ThreadId,
   TurnItemId,
   type ModelSelection,
@@ -2270,6 +2272,126 @@ it.effect("keeps a queued message queued when promoted into a run that Stop reac
   ),
 );
 
+it.effect.each(["ordinary", "goal-steer", "goal-promotion", "goal-restart"] as const)(
+  "queues a steer and puts a promoted message first while the turn waits on the person: %s",
+  (route) =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const { started, steered, layer, startFirstTurn } = yield* nextTurnSelectionHarness(
+          `steering-pending-request-${route}`,
+        );
+        yield* Effect.gen(function* () {
+          const orchestrator = yield* Orchestrator.OrchestratorV2;
+          const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
+          const sink = yield* EventSink.EventSinkV2;
+          const threadId = yield* startFirstTurn;
+          const first = started[0]!;
+          const messageText = (id: string) =>
+            ((route === "goal-steer" || route === "goal-restart") && id === "steered") ||
+            (route === "goal-promotion" && id === "promoted")
+              ? "/goal ship it"
+              : id;
+          const send = (id: string, delivery: "queue" | "steer") =>
+            orchestrator.dispatch({
+              type: "message.dispatch",
+              commandId: CommandId.make(id),
+              threadId,
+              messageId: MessageId.make(`message:${id}`),
+              text: messageText(id),
+              attachments: [],
+              modelSelection: runSelection,
+              ...(delivery === "queue"
+                ? { dispatchMode: { type: "queue_after_active" } }
+                : {
+                    dispatchMode: { type: "start_immediately" },
+                    deliveryIntent:
+                      route === "goal-restart" && id === "steered" ? "restart" : "steer",
+                  }),
+              createdBy: "user",
+              creationSource: "web",
+            });
+          yield* send("queued", "queue");
+          yield* send("promoted", "queue");
+          const promotedRun = (yield* orchestrator.getThreadProjection(threadId)).runs.find(
+            (run) => run.userMessageId === MessageId.make("message:promoted"),
+          )!;
+          const now = yield* DateTime.now;
+          yield* sink.write({
+            events: [
+              {
+                id: EventId.make("request:question"),
+                type: "runtime-request.updated",
+                threadId,
+                runId: first.runId,
+                nodeId: first.rootNodeId,
+                occurredAt: now,
+                payload: {
+                  id: RuntimeRequestId.make("request:question"),
+                  nodeId: first.rootNodeId,
+                  providerTurnId: null,
+                  nativeRequestRef: null,
+                  kind: "user_input",
+                  status: "pending",
+                  responseCapability: {
+                    type: "live",
+                    providerSessionId: ProviderSessionId.make("session:waiting"),
+                  },
+                  createdAt: now,
+                  resolvedAt: null,
+                },
+              },
+            ],
+          });
+
+          yield* orchestrator.dispatch({
+            type: "queued-message.promote-to-steer",
+            commandId: CommandId.make("promote"),
+            threadId,
+            queuedRunId: promotedRun.id,
+            targetRunId: first.runId,
+          });
+          yield* send("steered", "steer");
+          yield* worker.drain();
+
+          const projection = yield* orchestrator.getThreadProjection(threadId);
+          if (route === "goal-restart") {
+            assert.isTrue(
+              projection.turnItems.some((item) => item.type === "run_interrupt_request"),
+            );
+            assert.deepEqual(
+              started.map((turn) => turn.message.text),
+              ["first", "/goal ship it"],
+            );
+            assert.deepEqual(
+              projection.runtimeRequests.map((request) => request.status),
+              ["cancelled"],
+            );
+            return;
+          }
+          assert.isFalse(
+            projection.turnItems.some((item) => item.type === "run_interrupt_request"),
+          );
+          assert.deepEqual(steered, []);
+          assert.equal(started.length, 1);
+          assert.deepEqual(
+            projection.runs
+              .filter((run) => run.status === "queued")
+              .sort((left, right) => (left.queuePosition ?? 0) - (right.queuePosition ?? 0))
+              .map(
+                (run) =>
+                  projection.messages.find((message) => message.id === run.userMessageId)?.text,
+              ),
+            ["promoted", "queued", "steered"].map(messageText),
+          );
+          assert.deepEqual(
+            projection.runtimeRequests.map((request) => request.status),
+            ["pending"],
+          );
+        }).pipe(Effect.provide(layer));
+      }),
+    ),
+);
+
 // regression: #32891 a restart that Stop already reached is lost
 it.effect.each([false, true])(
   "refuses a steer after Stop (provider turn ready: %s)",

docs/user/composer.md

@@ -69,6 +69,9 @@ are saved on the server and can be edited or removed before they send.
 When Enter sends, `Cmd+Enter` on macOS or `Ctrl+Enter` on Windows and Linux uses
 the opposite action: it steers when your default is Queue and queues when your
 default is Steer.
+While the agent waits on your answer to a question or an approval, a steer queues
+instead, and a queued message you steer moves to the front of the queue, so the
+question stays until you answer it.
 If your **Send shortcut** requires `Cmd+Enter` or `Ctrl+Enter`, that shortcut uses
 your default follow-up behavior; add **Shift** to use the opposite action. This
 applies only to multiline messages when you choose the multiline send option.