From 25dd0a7c77cf16f9ecbfea7d6d834062c371c0ed Mon Sep 17 00:00:00 2001 From: flamboh Date: Sun, 1 Nov 2026 22:41:48 +0000 Subject: [PATCH 1/2] fix(server,web,mobile): follow-ups sent during setup steer once the turn starts --- .../src/state/use-thread-composer-state.ts | 22 ++- .../orchestration-v2/CommandPolicy.test.ts | 16 +- .../src/orchestration-v2/CommandPolicy.ts | 43 ++--- .../src/orchestration-v2/Orchestrator.ts | 106 +++++++++++++ .../src/orchestration-v2/runtimeLayer.test.ts | 149 ++++++++++++++++++ .../web/src/components/ChatView.logic.test.ts | 12 ++ apps/web/src/components/ChatView.logic.ts | 4 +- apps/web/src/components/ChatView.tsx | 1 + apps/web/src/components/chat/ChatComposer.tsx | 13 +- .../src/components/chat/composerCommandKey.ts | 8 +- packages/contracts/src/orchestrationV2.ts | 2 + 11 files changed, 326 insertions(+), 50 deletions(-) diff --git a/apps/mobile/src/state/use-thread-composer-state.ts b/apps/mobile/src/state/use-thread-composer-state.ts index 288d37d9f169e..e386abb9bd16a 100644 --- a/apps/mobile/src/state/use-thread-composer-state.ts +++ b/apps/mobile/src/state/use-thread-composer-state.ts @@ -357,8 +357,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; @@ -644,20 +649,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); diff --git a/apps/server/src/orchestration-v2/CommandPolicy.test.ts b/apps/server/src/orchestration-v2/CommandPolicy.test.ts index 8ed4de3e9dea0..a4ec0be8c9d02 100644 --- a/apps/server/src/orchestration-v2/CommandPolicy.test.ts +++ b/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, diff --git a/apps/server/src/orchestration-v2/CommandPolicy.ts b/apps/server/src/orchestration-v2/CommandPolicy.ts index 02f329e536feb..f864027b37e82 100644 --- a/apps/server/src/orchestration-v2/CommandPolicy.ts +++ b/apps/server/src/orchestration-v2/CommandPolicy.ts @@ -89,8 +89,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 { @@ -125,10 +130,7 @@ export function isActiveRun(run: OrchestrationV2Run): boolean { ); } -function providerTurnCapabilities( - projection: OrchestrationV2ThreadProjection, - run: OrchestrationV2Run, -) { +function providerTurnCapabilities(projection: DeliveryIntentProjection, run: OrchestrationV2Run) { const providerThread = projection.providerThreads.find( (candidate) => candidate.id === run.providerThreadId, ); @@ -141,8 +143,21 @@ function providerTurnCapabilities( return providerSession?.capabilities.turns; } +export function runLacksRunningProviderTurn( + projection: Pick, + run: Pick, +): 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 { @@ -150,22 +165,12 @@ function resolveDeliveryIntent( const activeRun = projection.runs.findLast(isActiveRun); 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 capabilities = providerTurnCapabilities(projection, activeRun); diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 1fed13e08ad0b..2b3d69449afd3 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -106,6 +106,7 @@ import { isActiveRun, isNativeMaintenanceCommand, resolveMessageDispatchIntent, + runLacksRunningProviderTurn, } from "./CommandPolicy.ts"; import { CommandReceiptStoreV2 } from "./CommandReceiptStore.ts"; import { @@ -5853,6 +5854,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({ @@ -5996,6 +6011,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, @@ -13712,6 +13728,96 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio Effect.forkScoped, ); + 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(); + 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 instead of restarting this one. + 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 stops holding after it was seen holding. diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index dbbdd5c66bafa..240262d3b2f21 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -1573,6 +1573,155 @@ it.layer(layerTest)("OrchestrationV2LayerLive", (it) => { }), ); + it.effect("steers a message sent before the active provider turn starts once it does", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const eventSink = yield* EventSink.EventSinkV2; + const sessions = yield* ProviderSessionManager.ProviderSessionManagerV2; + const threadId = ThreadId.make("runtime-startup-steer"); + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("runtime-startup-steer-create"), + threadId, + projectId: ProjectId.make("runtime-startup-steer-project"), + title: "Startup steer", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: process.cwd(), + }); + const dispatchFollowUp = (id: string, text: string, model = modelSelection.model) => + orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`runtime-startup-steer-${id}`), + threadId, + messageId: MessageId.make(`runtime-startup-steer-${id}`), + text, + attachments: [], + modelSelection: { ...modelSelection, model }, + dispatchMode: { type: "start_immediately" }, + deliveryIntent: "steer", + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("runtime-startup-steer-first"), + threadId, + messageId: MessageId.make("runtime-startup-steer-first"), + text: "Start work.", + attachments: [], + dispatchMode: { type: "start_immediately" }, + }); + yield* dispatchFollowUp("other-model", "Use another model for this.", "other-model"); + yield* dispatchFollowUp("rejected", "This promotion was already rejected."); + const rejectedRun = (yield* orchestrator.getThreadProjection(threadId)).runs[2]!; + yield* orchestrator + .dispatch({ + type: "queued-message.promote-to-steer", + commandId: CommandId.make(`command:system:startup-steer:${rejectedRun.id}`), + threadId, + queuedRunId: rejectedRun.id, + targetRunId: (yield* orchestrator.getThreadProjection(threadId)).runs[0]!.id, + }) + .pipe(Effect.flip); + yield* dispatchFollowUp("goal", "Check the goal."); + yield* orchestrator.dispatch({ + type: "queued-run.edit", + commandId: CommandId.make("runtime-startup-steer-goal-edit"), + threadId, + runId: (yield* orchestrator.getThreadProjection(threadId)).runs[3]!.id, + text: "/goal finish the work", + }); + yield* dispatchFollowUp("steer", "Also do this."); + const steerMessageId = MessageId.make("runtime-startup-steer-steer"); + + const starting = yield* orchestrator.getThreadProjection(threadId); + const run = starting.runs[0]!; + const providerThread = starting.providerThreads[0]!; + const now = yield* DateTime.now; + const providerSession = { + id: providerThread.providerSessionId!, + driver, + providerInstanceId: modelSelection.instanceId, + status: "running" as const, + cwd: process.cwd(), + model: modelSelection.model, + capabilities: CodexProviderCapabilitiesV2, + createdAt: now, + updatedAt: now, + lastError: null, + }; + const sessionSpy = vi.spyOn(sessions, "get").mockReturnValue( + Effect.succeedSome({ + providerSession, + } as ProviderAdapter.ProviderAdapterV2SessionRuntime), + ); + yield* Effect.addFinalizer(() => Effect.sync(() => sessionSpy.mockRestore())); + const afterSequence = yield* orchestrator.getThreadEventSequence(threadId); + yield* eventSink.write({ + commandId: CommandId.make("runtime-startup-steer-running"), + events: [ + { + id: EventId.make("runtime-startup-steer-run-event"), + type: "run.updated", + threadId, + runId: run.id, + occurredAt: now, + payload: { ...run, status: "running", startedAt: now }, + }, + { + id: EventId.make("runtime-startup-steer-session-event"), + type: "provider-session.attached", + threadId, + occurredAt: now, + payload: providerSession, + }, + { + id: EventId.make("runtime-startup-steer-turn-event"), + type: "provider-turn.updated", + threadId, + runId: run.id, + occurredAt: now, + payload: { + id: ProviderTurnId.make("runtime-startup-steer-turn"), + providerThreadId: providerThread.id, + nodeId: run.rootNodeId!, + runAttemptId: run.activeAttemptId, + nativeTurnRef: null, + ordinal: 1, + status: "running", + startedAt: now, + completedAt: null, + }, + }, + ], + }); + + yield* eventSink.stream({ threadId, afterSequence }).pipe( + Stream.filter( + (stored) => + stored.event.type === "turn-item.updated" && + stored.event.payload.type === "user_message" && + stored.event.payload.messageId === steerMessageId, + ), + Stream.take(1), + Stream.runDrain, + ); + assert.deepEqual( + (yield* orchestrator.getThreadProjection(threadId)).runs.map( + (candidate) => candidate.status, + ), + ["running", "queued", "queued", "queued", "cancelled"], + ); + }), + ); + it.effect("answers an async question after its provider exits and commits the answer once", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; diff --git a/apps/web/src/components/ChatView.logic.test.ts b/apps/web/src/components/ChatView.logic.test.ts index 87354e0d257f9..ebea53f431cc0 100644 --- a/apps/web/src/components/ChatView.logic.test.ts +++ b/apps/web/src/components/ChatView.logic.test.ts @@ -1128,6 +1128,18 @@ describe("planComposerTurn", () => { baseBranchForWorktree: null, }); }); + + it("waits for setup but keeps the chosen follow-up so the server can steer once it starts", () => { + const setup = { ...base, sendQueuesBehindSetup: true }; + expect(planComposerTurn({ ...setup, dispatchMode: "steer" })).toMatchObject({ + turnDispatchMode: "steer", + shouldQueueBehindActiveRun: true, + }); + expect(planComposerTurn({ ...setup, dispatchMode: "restart" })).toMatchObject({ + turnDispatchMode: "queue", + shouldQueueBehindActiveRun: true, + }); + }); }); describe("restorePlanFollowUpComposer", () => { diff --git a/apps/web/src/components/ChatView.logic.ts b/apps/web/src/components/ChatView.logic.ts index 93468b9e04abf..8a70d1dc78060 100644 --- a/apps/web/src/components/ChatView.logic.ts +++ b/apps/web/src/components/ChatView.logic.ts @@ -1433,7 +1433,9 @@ export function planComposerTurn(input: { baseBranchForWorktree: shouldCreateWorktree ? input.activeThreadBranch : null, compactBeforeSend, turnDispatchMode: - compactBeforeSend || input.sendQueuesBehindSetup ? ("queue" as const) : input.dispatchMode, + compactBeforeSend || (input.sendQueuesBehindSetup && input.dispatchMode === "restart") + ? ("queue" as const) + : input.dispatchMode, shouldQueueBehindActiveRun: compactBeforeSend || input.sendQueuesBehindSetup || diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 6cd1c7775d96b..014f98b141571 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -3593,6 +3593,7 @@ export default function ChatView(props: ChatViewProps) { isLocalDraftThread && activeProject === null } phase={phase} + sendQueuesBehindSetup={sendQueuesBehindSetup} isCompacting={isCompacting} canInterrupt={canInterruptRunningThread} isSendBusy={ diff --git a/apps/web/src/components/chat/ChatComposer.tsx b/apps/web/src/components/chat/ChatComposer.tsx index aa2c52443a5cb..4dc8235ad1fcc 100644 --- a/apps/web/src/components/chat/ChatComposer.tsx +++ b/apps/web/src/components/chat/ChatComposer.tsx @@ -371,6 +371,7 @@ export interface ChatComposerProps { projectSelectionRequired: boolean; phase: SessionPhase; + sendQueuesBehindSetup: boolean; canInterrupt: boolean; isSendBusy: boolean; canResume: boolean; @@ -525,6 +526,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) forceExpandedOnMobile, projectSelectionRequired, phase, + sendQueuesBehindSetup, canInterrupt, isSendBusy, canResume, @@ -1253,6 +1255,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) isComplete: Boolean(activePendingResolvedAnswers), } : null; + const turnActive = phase === "running" || sendQueuesBehindSetup; const showResumeAction = canResume && !composerDraftHasUserContent(composerDraft) && !isEditingQueuedMessage; const collapsedComposerPrimaryActionDisabled = @@ -2111,7 +2114,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) sendEvent, dispatchMode ?? resolveComposerDispatchMode({ - running: phase === "running", + running: turnActive, alternateModifier: false, activeTurnDefault: followUpBehavior, canSteer: !isCompacting, @@ -2134,7 +2137,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) submitComposer( event, resolveComposerDispatchMode({ - running: phase === "running", + running: turnActive, alternateModifier: event.metaKey || event.ctrlKey, activeTurnDefault: followUpBehavior, canSteer: !isCompacting, @@ -2146,7 +2149,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) submitComposer( undefined, resolveComposerDispatchMode({ - running: phase === "running", + running: turnActive, alternateModifier: false, activeTurnDefault: followUpBehavior, canSteer: !isCompacting, @@ -2207,7 +2210,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) keybindings, sendShortcut: settings.sendShortcut, routeKind, - phase, + turnActive, isCompacting, followUpBehavior, isSoftwareKeyboardOpen, @@ -3508,7 +3511,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) reserveContextWindowMeter={reserveContextWindowMeter} activeThreadModelDisplayName={activeThreadModelDisplayName} pendingAction={pendingPrimaryAction} - isRunning={phase === "running"} + isRunning={turnActive} canInterrupt={canInterrupt} followUpBehavior={followUpBehavior} canSteer={!isCompacting} diff --git a/apps/web/src/components/chat/composerCommandKey.ts b/apps/web/src/components/chat/composerCommandKey.ts index 11c7710035b5c..7561f59b81424 100644 --- a/apps/web/src/components/chat/composerCommandKey.ts +++ b/apps/web/src/components/chat/composerCommandKey.ts @@ -37,7 +37,7 @@ export function composerCommandKeyHandler(input: { readonly keybindings: ChatComposerProps["keybindings"]; readonly sendShortcut: ChatComposerProps["settings"]["sendShortcut"]; readonly routeKind: ChatComposerProps["routeKind"]; - readonly phase: ChatComposerProps["phase"]; + readonly turnActive: boolean; readonly isCompacting: boolean; readonly followUpBehavior: ChatComposerProps["settings"]["followUpBehavior"]; readonly isSoftwareKeyboardOpen: boolean; @@ -89,7 +89,7 @@ export function composerCommandKeyHandler(input: { keybindings, sendShortcut, routeKind, - phase, + turnActive, isCompacting, followUpBehavior, isSoftwareKeyboardOpen, @@ -204,7 +204,7 @@ export function composerCommandKeyHandler(input: { keybindings, isSoftwareKeyboardOpen, isDraftThread: routeKind === "draft", - isRunning: phase === "running", + isRunning: turnActive, sendShortcut, prompt: promptRef.current, }); @@ -246,7 +246,7 @@ export function composerCommandKeyHandler(input: { submitComposer( undefined, resolveComposerDispatchMode({ - running: phase === "running", + running: turnActive, alternateModifier: submissionIntent === "alternate", activeTurnDefault: followUpBehavior, canSteer: !isCompacting, diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 1c80128f8bbc5..b03455a077a8b 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -700,6 +700,8 @@ export const OrchestrationV2Run = Schema.Struct({ ), delegatedCompletion: Schema.optional(OrchestrationV2DelegatedCompletionCohort), workspacePreparation: Schema.optional(OrchestrationV2ThreadLaunchWorkspaceStrategy), + /** A message queued only because the active turn had not started yet; delivered so once it does. */ + deliveryIntent: Schema.optional(Schema.Literals(["auto", "steer"])), }); export type OrchestrationV2Run = typeof OrchestrationV2Run.Type; From 59e1a7336c18c1318fd48b420a99e2b66034eaab Mon Sep 17 00:00:00 2001 From: flamboh Date: Sun, 1 Nov 2026 22:59:39 +0000 Subject: [PATCH 2/2] fix(server): stop the startup-steer consumer with the orchestrator --- apps/server/src/orchestration-v2/Orchestrator.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 2b3d69449afd3..719acd296d352 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -13815,7 +13815,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio stored.event.payload.status === "running", ), Stream.runForEach(deliverStartupQueuedMessages), - Effect.forkDetach, + Effect.forkScoped, ); // A child whose run ends while its background tasks or its own subagents still run has no