fix(server,web,mobile): follow-ups sent during setup steer once the turn starts

apps/mobile/src/state/use-thread-composer-state.ts

@@ -370,8 +370,13 @@ export function useThreadComposerState() {
           item.status === "cancelled"),
     );
   })();
+  // The server holds a steer sent during setup until the turn starts.
+  const setupRunning = [selectedThreadActivityRun?.status, selectedThreadRuntime?.status].some(
+    (status) => status === "preparing" || status === "starting",
+  );
   // A compacting turn cannot take steering input, so follow-ups queue meanwhile.
-  const canSteerActiveTurn = queueWorkflow?.canPromoteToSteer === true && !isCompacting;
+  const canSteerActiveTurn =
+    (queueWorkflow?.canPromoteToSteer === true || setupRunning) && !isCompacting;
   // A stop or a restart holds the queue, completion wakes included, until it
   // is resumed; the queue sheet lists only the user's own messages.
   const queueHeld = queueWorkflow?.isHeld === true;
@@ -650,20 +655,13 @@ export function useThreadComposerState() {
     // An approval redirect must steer the waiting turn instead of queueing behind it.
     const activeTurnDefault = beforeDispatch !== undefined ? "steer" : followUpBehavior;
     const followUpAction = resolveComposerDispatchMode({
-      running: activeThreadBusy && canSteerActiveTurn,
+      running: setupRunning || (activeThreadBusy && canSteerActiveTurn),
       alternateModifier: followUpOverride !== undefined && followUpOverride !== activeTurnDefault,
       activeTurnDefault,
+      canSteer: !isCompacting,
     });
-    const setupRunning = [selectedThreadActivityRun?.status, selectedThreadRuntime?.status].some(
-      (status) => status === "preparing" || status === "starting",
-    );
-    const followUpDispatchMode = setupRunning
-      ? "queue"
-      : followUpAction === "auto"
-        ? null
-        : followUpAction === "queue"
-          ? "queue"
-          : "auto";
+    const followUpDispatchMode =
+      followUpAction === "auto" ? null : followUpAction === "queue" ? "queue" : "auto";
 
     const metadata = makeQueuedMessageMetadata();
     const messageId = MessageId.make(metadata.messageId);

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

@@ -141,19 +141,17 @@ it.each([
     "belongs to a previous attempt",
     [{ runAttemptId: RunAttemptId.make("previous-attempt"), status: "running" }],
   ],
-])("queues only automatic delivery when the active run's provider turn %s", (_, providerTurns) => {
+] as const)("queues follow-ups when the active run's provider turn %s", (_, providerTurns) => {
   const projection = {
     ...dispatchProjection(baseCapabilities),
     providerTurns,
   } as unknown as OrchestrationV2ThreadProjection;
-  assert.deepEqual(
-    CommandPolicy.resolveMessageDispatchIntent(projection, { type: "start_immediately" }, "auto"),
-    { type: "queue_after_active" },
-  );
-  assert.deepEqual(
-    CommandPolicy.resolveMessageDispatchIntent(projection, { type: "start_immediately" }, "steer"),
-    { type: "steer_active", targetRunId: activeRunId },
-  );
+  for (const intent of ["auto", "steer"] as const) {
+    assert.deepEqual(
+      CommandPolicy.resolveMessageDispatchIntent(projection, { type: "start_immediately" }, intent),
+      { type: "queue_after_active" },
+    );
+  }
   assert.deepEqual(
     CommandPolicy.resolveMessageDispatchIntent(
       projection,

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

@@ -150,8 +150,13 @@ export function goalCommandStopsActiveRun(
   }
 }
 
+type DeliveryIntentProjection = Pick<
+  OrchestrationV2ThreadProjection,
+  "runs" | "messages" | "providerTurns" | "providerThreads" | "providerSessions"
+>;
+
 export function resolveMessageDispatchIntent(
-  projection: OrchestrationV2ThreadProjection,
+  projection: DeliveryIntentProjection,
   requestedMode: MessageDispatchMode,
   deliveryIntent?: "auto" | "steer" | "restart",
 ): MessageDispatchMode {
@@ -165,8 +170,21 @@ export function resolveMessageDispatchIntent(
     : mode;
 }
 
+export function runLacksRunningProviderTurn(
+  projection: Pick<OrchestrationV2ThreadProjection, "providerTurns">,
+  run: Pick<OrchestrationV2Run, "status" | "activeAttemptId">,
+): boolean {
+  return (
+    run.status === "preparing" ||
+    run.status === "starting" ||
+    !projection.providerTurns.some(
+      (turn) => turn.runAttemptId === run.activeAttemptId && turn.status === "running",
+    )
+  );
+}
+
 function resolveDeliveryIntent(
-  projection: OrchestrationV2ThreadProjection,
+  projection: DeliveryIntentProjection,
   requestedMode: MessageDispatchMode,
   deliveryIntent?: "auto" | "steer" | "restart",
 ): MessageDispatchMode {
@@ -180,22 +198,12 @@ function resolveDeliveryIntent(
       run.status === "waiting",
   );
   if (activeRun === undefined) return { type: "start_immediately" };
-  if (deliveryIntent === "steer") {
-    return { type: "steer_active", targetRunId: activeRun.id };
-  }
   if (deliveryIntent === "restart") {
     return { type: "restart_active", targetRunId: activeRun.id };
   }
-  // A run can be active before its provider turn starts or after it finishes.
-  // Queue automatic input in those gaps so steering cannot reject the message.
-  if (
-    activeRun.status === "preparing" ||
-    activeRun.status === "starting" ||
-    !projection.providerTurns.some(
-      (turn) => turn.runAttemptId === activeRun.activeAttemptId && turn.status === "running",
-    )
-  ) {
-    return { type: "queue_after_active" };
+  if (runLacksRunningProviderTurn(projection, activeRun)) return { type: "queue_after_active" };
+  if (deliveryIntent === "steer") {
+    return { type: "steer_active", targetRunId: activeRun.id };
   }
 
   const providerThread = projection.providerThreads.find(

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

@@ -97,6 +97,7 @@ import {
   goalCommandStopsActiveRun,
   isNativeMaintenanceCommand,
   resolveMessageDispatchIntent,
+  runLacksRunningProviderTurn,
 } from "./CommandPolicy.ts";
 import { CommandReceiptStoreV2 } from "./CommandReceiptStore.ts";
 import { ContextHandoffServiceV2 } from "./ContextHandoffService.ts";
@@ -5496,6 +5497,20 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
         payload: { ...held, queueHeld: false },
       });
     }
+    const activeRunMessage = projection.messages.find(
+      (message) => message.id === activeRun?.userMessageId,
+    );
+    const startupDeliveryIntent =
+      activeRun !== undefined &&
+      (command.deliveryIntent === "auto" || command.deliveryIntent === "steer") &&
+      goalCommand === null &&
+      delegatedCompletion === undefined &&
+      command.notification === undefined &&
+      !isNativeMaintenanceCommand(command) &&
+      (activeRunMessage === undefined || !isNativeMaintenanceCommand(activeRunMessage)) &&
+      runLacksRunningProviderTurn(projection, activeRun)
+        ? command.deliveryIntent
+        : undefined;
     if (shouldQueue) {
       if (pendingMergeBackTransfers.length > 0) {
         return yield* new OrchestratorDispatchError({
@@ -5630,6 +5645,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
           ? {}
           : { restartContinuationOfRunId: command.restartContinuationOfRunId }),
         ...(continuedTaskId === undefined ? {} : { delegatedTaskId: continuedTaskId }),
+        ...(startupDeliveryIntent === undefined ? {} : { deliveryIntent: startupDeliveryIntent }),
       };
       const attempt: OrchestrationV2RunAttempt = {
         id: attemptId,
@@ -12401,6 +12417,98 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
       Effect.forkDetach,
     );
 
+  const deliverStartupQueuedMessages = Effect.fn("Orchestrator.deliverStartupQueuedMessages")(
+    function* (stored: OrchestrationV2StoredEvent) {
+      if (stored.event.type !== "provider-turn.updated") return;
+      const threadId = stored.event.threadId;
+      // Under the lock so a send still committing its queued run is seen. Each promotion can
+      // restart the target, so the next one decides from fresh records.
+      yield* threadDispatch.withLock(
+        threadId,
+        Effect.gen(function* () {
+          const attempted = new Set<RunId>();
+          while (true) {
+            const projection = yield* projectionStore.getThreadRecords(
+              threadId,
+              ["runs", "messages", "providerTurns", "providerThreads", "providerSessions"],
+              { turnItemTypes: [], messageRoles: ["user"] },
+            );
+            const next = queuedRunsInDeliveryOrder(projection)
+              .filter((queued) => {
+                if (queued.deliveryIntent === undefined || queued.queueHeld === true) return false;
+                if (attempted.has(queued.id)) return false;
+                // The message may have been edited into a command that must run as its own turn.
+                const message = projection.messages.find((m) => m.id === queued.userMessageId);
+                return (
+                  message !== undefined &&
+                  parseGoalCommand(message) === null &&
+                  !isNativeMaintenanceCommand(message)
+                );
+              })
+              .map((queued) => ({
+                queued,
+                mode: resolveMessageDispatchIntent(
+                  projection,
+                  { type: "queue_after_active" },
+                  queued.deliveryIntent,
+                ),
+              }))
+              .find(
+                ({ queued, mode }) =>
+                  mode.type === "steer_active" &&
+                  // A message for another model starts its own turn. Promotion steers with the
+                  // thread's selection, so that must match too or the steer would restart the turn.
+                  modelSelectionsEqual(queued.modelSelection, projection.thread.modelSelection) &&
+                  projection.runs.some(
+                    (run) =>
+                      run.id === mode.targetRunId &&
+                      modelSelectionsEqual(queued.modelSelection, run.modelSelection),
+                  ),
+              );
+            if (next?.mode.type !== "steer_active") return;
+            attempted.add(next.queued.id);
+            yield* dispatchWithReceiptEffect({
+              type: "queued-message.promote-to-steer",
+              commandId: CommandId.make(`command:system:startup-steer:${next.queued.id}`),
+              threadId,
+              queuedRunId: next.queued.id,
+              targetRunId: next.mode.targetRunId,
+            }).pipe(
+              Effect.catch((cause) =>
+                Effect.logWarning("Failed to steer a message queued before the turn started", {
+                  threadId,
+                  queuedRunId: next.queued.id,
+                  cause,
+                }),
+              ),
+            );
+          }
+        }),
+      );
+    },
+    (effect, stored) =>
+      effect.pipe(
+        Effect.catchCause((cause) =>
+          Effect.logWarning("Failed to steer messages queued before the turn started", {
+            threadId: stored.event.threadId,
+            sequence: stored.sequence,
+            cause,
+          }),
+        ),
+      ),
+  );
+  yield* eventSink
+    .stream({ afterSequence: terminalEventsAfterSequence, eventType: "provider-turn.updated" })
+    .pipe(
+      Stream.filter(
+        (stored) =>
+          stored.event.type === "provider-turn.updated" &&
+          stored.event.payload.status === "running",
+      ),
+      Stream.runForEach(deliverStartupQueuedMessages),
+      Effect.forkDetach,
+    );
+
   // A child whose run ends while its background tasks or its own subagents still run has no
   // result yet. When that work settles there is no run update to finalize it, so these events
   // do it. A background roster counts once it empties after it was seen non-empty.