From 199cc1c18452c1e626ae50ddf4935135ccc7642c Mon Sep 17 00:00:00 2001 From: Yash Singh Date: Wed, 11 Nov 2026 15:57:29 -0600 Subject: [PATCH 1/3] fix: deliver steering after worktree setup --- .../src/orchestration-v2/Orchestrator.ts | 107 ++++++++++++++++++ .../SteeringCompletion.integration.test.ts | 102 +++++++++++++++++ .../web/src/components/ChatView.logic.test.ts | 6 + apps/web/src/components/ChatView.logic.ts | 4 +- apps/web/src/components/ChatView.tsx | 2 +- apps/web/src/components/chat/ChatComposer.tsx | 12 +- .../src/components/chat/composerCommandKey.ts | 4 +- .../src/state/threadWorkflows.test.ts | 4 +- .../src/state/threadWorkflows.ts | 4 + packages/contracts/src/orchestrationV2.ts | 1 + 10 files changed, 234 insertions(+), 12 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 7af71e48d67ea..b916732c803c5 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -6523,6 +6523,34 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio payload: { ...sourcePlan, status: "completed" }, }); + let steerWhenReadyRunId: RunId | undefined; + if (dispatchMode.type === "steer_active") { + const targetRunId = dispatchMode.targetRunId; + const target = projection.runs.find((run) => run.id === targetRunId); + if ( + target?.status === "preparing" || + target?.status === "starting" || + (target?.status === "running" && + !projection.providerTurns.some( + (turn) => turn.runAttemptId === target.activeAttemptId && turn.status === "running", + )) + ) { + const adapter = yield* providerAdapters + .get(target.providerInstanceId) + .pipe(mapDispatchError(command)); + const capabilities = yield* adapter.getCapabilities().pipe(mapDispatchError(command)); + yield* commandPolicy + .decideSteeringExecution({ + commandId: command.commandId, + threadId: command.threadId, + providerInstanceId: target.providerInstanceId, + capabilities, + }) + .pipe(mapDispatchError(command)); + steerWhenReadyRunId = target.id; + dispatchMode = { type: "queue_after_active" }; + } + } if (dispatchMode.type === "steer_active" || dispatchMode.type === "restart_active") { yield* ensureGoalCommandSupported(command, command, modelSelection.instanceId); yield* dispatchSteerIntoRun({ @@ -6737,6 +6765,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio rootNodeId, activeAttemptId: attemptId, status: "queued", + ...(steerWhenReadyRunId === undefined ? {} : { steerWhenReadyRunId }), ...(worktreeMissingRun !== undefined || (heldRunsToRelease.length === 0 && queueIsHeld && @@ -10373,6 +10402,37 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio now, }); } + if ( + target?.status === "preparing" || + target?.status === "starting" || + (target?.status === "running" && + !projection.providerTurns.some( + (turn) => turn.runAttemptId === target.activeAttemptId && turn.status === "running", + )) + ) { + const adapter = yield* providerAdapters + .get(target.providerInstanceId) + .pipe(mapDispatchError(command)); + const capabilities = yield* adapter.getCapabilities().pipe(mapDispatchError(command)); + yield* commandPolicy + .decideSteeringExecution({ + commandId: command.commandId, + threadId: command.threadId, + providerInstanceId: target.providerInstanceId, + capabilities, + }) + .pipe(mapDispatchError(command)); + yield* emitEvent({ + type: "run.updated", + threadId: command.threadId, + runId: queuedRun.id, + nodeId: queuedRootNode.id, + providerInstanceId: queuedRun.providerInstanceId, + occurredAt: now, + payload: { ...queuedRun, steerWhenReadyRunId: target.id }, + }); + return; + } yield* emitEvent({ type: "run.updated", threadId: command.threadId, @@ -14573,6 +14633,53 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio Effect.forkScoped, ); + 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( + Effect.fn("Orchestrator.deliverDeferredSteers")( + function* (stored) { + if (stored.event.type !== "provider-turn.updated") return; + const threadId = stored.event.threadId; + const turn = stored.event.payload; + const projection = yield* projectionStore.getThreadRecords(threadId, ["runs"]); + const run = projection.runs.find( + (candidate) => + candidate.activeAttemptId === turn.runAttemptId && candidate.status === "running", + ); + if (run === undefined) return; + const pending = projection.runs + .filter( + (queued) => queued.status === "queued" && queued.steerWhenReadyRunId === run.id, + ) + .toSorted( + (left, right) => + (left.queuePosition ?? left.ordinal) - (right.queuePosition ?? right.ordinal), + ); + for (const queued of pending) { + yield* dispatchWithReceiptEffect({ + type: "queued-message.promote-to-steer", + commandId: CommandId.make(`command:deferred-steer:${queued.id}:${run.id}`), + threadId, + queuedRunId: queued.id, + targetRunId: run.id, + }); + } + }, + (effect, stored) => threadDispatch.withLock(stored.event.threadId, effect), + Effect.catchCause((cause) => + Effect.logWarning("Failed to deliver deferred steering", { cause }), + ), + ), + ), + Effect.forkScoped, + ); + // 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/SteeringCompletion.integration.test.ts b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts index 7d66d8d101996..1baa1a43dc856 100644 --- a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts +++ b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts @@ -847,6 +847,108 @@ it.effect("sends a selected held message once and preserves the remaining queue ), ); +it.effect.each( + (["preparing", "starting", "running"] as const).flatMap((status) => + [false, true].map((direct) => ({ status, direct })), + ), +)("delivers a deferred steer for a $status run (direct: $direct)", ({ status, direct }) => + Effect.scoped( + Effect.gen(function* () { + const { started, steered, layer, startFirstTurn, events } = yield* nextTurnSelectionHarness( + `deferred-steer-${status}-${direct}`, + ); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const threadId = yield* startFirstTurn; + const before = yield* orchestrator.getThreadProjection(threadId); + const run = before.runs[0]!; + const turn = before.providerTurns[0]!; + const now = yield* DateTime.now; + yield* (yield* EventSink.EventSinkV2).write({ + events: [ + { + id: EventId.make("setup-run"), + type: "run.updated", + threadId, + occurredAt: now, + payload: { ...run, status }, + }, + { + id: EventId.make("setup-turn"), + type: "provider-turn.updated", + threadId, + occurredAt: now, + payload: { ...turn, status: "pending" }, + }, + ], + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("queue"), + threadId, + messageId: MessageId.make("message:steer-after-setup"), + text: "steer after setup", + attachments: [], + dispatchMode: { type: "queue_after_active" }, + ...(direct ? { deliveryIntent: "steer" as const } : {}), + createdBy: "user", + creationSource: "web", + }); + const queued = (yield* orchestrator.getThreadProjection(threadId)).runs.find( + (candidate) => candidate.status === "queued", + )!; + const command = { + type: "queued-message.promote-to-steer" as const, + commandId: CommandId.make("promote"), + threadId, + queuedRunId: queued.id, + targetRunId: run.id, + }; + if (!direct) { + yield* orchestrator.dispatch(command); + yield* orchestrator.dispatch(command); + } + yield* worker.drain(); + assert.deepEqual(steered, []); + const promoted = yield* orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => + event.type === "run.updated" && + event.payload.id === queued.id && + event.payload.status === "cancelled", + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + yield* (yield* EventSink.EventSinkV2).write({ + events: [ + { + id: EventId.make("ready-run"), + type: "run.updated", + threadId, + occurredAt: now, + payload: run, + }, + ], + }); + yield* Queue.offer(events, { type: "provider_turn.updated", driver, providerTurn: turn }); + yield* Fiber.join(promoted); + yield* worker.drain(); + assert.deepEqual(steered, ["steer after setup"]); + assert.lengthOf(started, 1); + const after = yield* orchestrator.getThreadProjection(threadId); + assert.equal( + after.messages.find((message) => message.id === queued.userMessageId)?.runId, + run.id, + ); + assert.equal(after.runs.find((candidate) => candidate.id === run.id)?.status, "running"); + }).pipe(Effect.provide(layer)); + }), + ), +); + it.effect("force sends multiple queued messages and can force send a requeued message again", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/web/src/components/ChatView.logic.test.ts b/apps/web/src/components/ChatView.logic.test.ts index c1f68878e9644..e25b6175b7ca6 100644 --- a/apps/web/src/components/ChatView.logic.test.ts +++ b/apps/web/src/components/ChatView.logic.test.ts @@ -1127,6 +1127,12 @@ describe("planComposerTurn", () => { shouldQueueBehindActiveRun: false, baseBranchForWorktree: null, }); + expect( + planComposerTurn({ ...running, sendQueuesBehindSetup: true, dispatchMode: "steer" }), + ).toMatchObject({ + turnDispatchMode: "steer", + shouldQueueBehindActiveRun: true, + }); // A fresh session cannot join the running turn, and a stale one is not compacted first. expect( planComposerTurn({ diff --git a/apps/web/src/components/ChatView.logic.ts b/apps/web/src/components/ChatView.logic.ts index d35b6ce28c433..fa2762385cc14 100644 --- a/apps/web/src/components/ChatView.logic.ts +++ b/apps/web/src/components/ChatView.logic.ts @@ -1321,7 +1321,9 @@ export function planComposerTurn(input: { baseBranchForWorktree: shouldCreateWorktree ? input.activeThreadBranch : null, compactBeforeSend, turnDispatchMode: - compactBeforeSend || input.sendQueuesBehindSetup ? ("queue" as const) : dispatchMode, + compactBeforeSend || (input.sendQueuesBehindSetup && dispatchMode !== "steer") + ? ("queue" as const) + : dispatchMode, shouldQueueBehindActiveRun: compactBeforeSend || input.sendQueuesBehindSetup || diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index a14d24c947b35..36d4e7b63b548 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -2675,7 +2675,7 @@ export default function ChatView(props: ChatViewProps) { document.activeElement?.getAttribute("data-testid") === "composer-editor" && isInSideChatPane(document.activeElement) === embedded, draftThreadRoute: routeKind === "draft", - turnRunning: phase === "running", + turnRunning: phase === "running" || phase === "connecting", composerRequiresModifier: composerRequiresModifier(settings.sendShortcut, promptRef.current), isWeb: !isElectron, isDesktop: isElectron, diff --git a/apps/web/src/components/chat/ChatComposer.tsx b/apps/web/src/components/chat/ChatComposer.tsx index 6903994b62f40..9e4493b5f75ae 100644 --- a/apps/web/src/components/chat/ChatComposer.tsx +++ b/apps/web/src/components/chat/ChatComposer.tsx @@ -2142,7 +2142,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) sendEvent, dispatchMode ?? resolveComposerDispatchMode({ - running: phase === "running", + running: phase === "running" || phase === "connecting", alternateModifier: false, activeTurnDefault: followUpBehavior, canSteer, @@ -2174,7 +2174,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) keybindings, isSoftwareKeyboardOpen, isDraftThread: routeKind === "draft", - isRunning: phase === "running", + isRunning: phase === "running" || phase === "connecting", sendShortcut: settings.sendShortcut, prompt: promptRef.current, }); @@ -2184,7 +2184,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) submitComposer( undefined, resolveComposerDispatchMode({ - running: phase === "running", + running: phase === "running" || phase === "connecting", alternateModifier: intent === "alternate", canSteer, activeTurnDefault: followUpBehavior, @@ -2198,7 +2198,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) submitComposer( event, resolveComposerDispatchMode({ - running: phase === "running", + running: phase === "running" || phase === "connecting", alternateModifier: event.metaKey || event.ctrlKey, activeTurnDefault: followUpBehavior, canSteer, @@ -2210,7 +2210,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) return submitComposer( undefined, resolveComposerDispatchMode({ - running: phase === "running", + running: phase === "running" || phase === "connecting", alternateModifier: false, activeTurnDefault: followUpBehavior, canSteer, @@ -3594,7 +3594,7 @@ export const ChatComposer = memo(function ChatComposer(props: ChatComposerProps) compact={isComposerPrimaryActionsCompact} canOperateThread={canOperateThread} pendingAction={pendingPrimaryAction} - isRunning={phase === "running"} + isRunning={phase === "running" || phase === "connecting"} canInterrupt={canInterrupt} runId={activeThread?.runtime?.activeRunId ?? null} followUpBehavior={followUpBehavior} diff --git a/apps/web/src/components/chat/composerCommandKey.ts b/apps/web/src/components/chat/composerCommandKey.ts index beaab007028ae..40650fe6c8952 100644 --- a/apps/web/src/components/chat/composerCommandKey.ts +++ b/apps/web/src/components/chat/composerCommandKey.ts @@ -205,7 +205,7 @@ export function composerCommandKeyHandler(input: { keybindings, isSoftwareKeyboardOpen, isDraftThread: routeKind === "draft", - isRunning: phase === "running", + isRunning: phase === "running" || phase === "connecting", sendShortcut, prompt: promptRef.current, }); @@ -264,7 +264,7 @@ export function composerCommandKeyHandler(input: { submitComposer( undefined, resolveComposerDispatchMode({ - running: phase === "running", + running: phase === "running" || phase === "connecting", alternateModifier: submissionIntent === "alternate", activeTurnDefault: followUpBehavior, canSteer, diff --git a/packages/client-runtime/src/state/threadWorkflows.test.ts b/packages/client-runtime/src/state/threadWorkflows.test.ts index 51f4d5a9e46ca..c47b5ed28a738 100644 --- a/packages/client-runtime/src/state/threadWorkflows.test.ts +++ b/packages/client-runtime/src/state/threadWorkflows.test.ts @@ -305,7 +305,7 @@ describe("thread workflows", () => { }); it.each(["preparing", "starting", "waiting"] as const)( - "does not promote queued work into a %s run", + "defers steering during setup and blocks it while waiting (%s)", (status) => { const state = deriveThreadQueueWorkflowState({ thread: { id: "thread", activeProviderThreadId: "provider-thread" }, @@ -343,7 +343,7 @@ describe("thread workflows", () => { ], } as never); - expect(state.canPromoteToSteer).toBe(false); + expect(state.canPromoteToSteer).toBe(status !== "waiting"); }, ); diff --git a/packages/client-runtime/src/state/threadWorkflows.ts b/packages/client-runtime/src/state/threadWorkflows.ts index 0aae5835eff03..c96090427c4fd 100644 --- a/packages/client-runtime/src/state/threadWorkflows.ts +++ b/packages/client-runtime/src/state/threadWorkflows.ts @@ -179,6 +179,10 @@ export function deriveThreadQueueWorkflowState(projection: Projection): ThreadQu projection.thread.deletedAt == null, canPromoteToSteer: promoteSkipsCompaction || + ((activeRun?.status === "preparing" || activeRun?.status === "starting") && + (capabilities === undefined || + capabilities.supportsActiveSteering || + capabilities.supportsSteeringByInterruptRestart)) || (hasSteerableProviderTurn && (capabilities?.supportsActiveSteering === true || capabilities?.supportsSteeringByInterruptRestart === true)), diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 915bb37a001a9..3189bb3eb7e65 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -696,6 +696,7 @@ export const OrchestrationV2Run = Schema.Struct({ status: OrchestrationV2RunStatus, queuePosition: Schema.optional(Schema.NullOr(PositiveInt)), queueHeld: Schema.optional(Schema.Boolean), + steerWhenReadyRunId: Schema.optional(RunId), requestedAt: Schema.DateTimeUtc, startedAt: Schema.NullOr(Schema.DateTimeUtc), completedAt: Schema.NullOr(Schema.DateTimeUtc), From b56fc5ac8b51a6098b0a3f2fbafee5c7c0003865 Mon Sep 17 00:00:00 2001 From: Yash Singh Date: Wed, 11 Nov 2026 16:08:13 -0600 Subject: [PATCH 2/3] fix(server): preserve deferred steering across restarts --- .../src/orchestration-v2/Orchestrator.ts | 53 +++++---- .../SteeringCompletion.integration.test.ts | 104 +++++++++++++++++- 2 files changed, 135 insertions(+), 22 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index b916732c803c5..175e2de528125 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -5275,6 +5275,33 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }); }); + const ensureSteeringMessage = Effect.fn("Orchestrator.ensureSteeringMessage")(function* ( + command: Extract< + OrchestrationV2Command, + { readonly type: "message.dispatch" | "queued-message.promote-to-steer" } + >, + message: Pick, + ) { + if (isNativeMaintenanceCommand(message)) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: + message.text.trim().toLowerCase() === "/compact" + ? "Context compaction must run as a separate turn. Queue it or wait for the active turn to finish." + : "Signing out must run as a separate turn. Queue it or wait for the active turn to finish.", + }); + } + if (parseGoalCommand(message) !== null) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: + "Goal commands must run as a separate turn. Queue it or wait for the active turn to finish.", + }); + } + }); + const dispatchSteerIntoRun = Effect.fn("Orchestrator.dispatchSteerIntoRun")(function* (input: { readonly command: Extract< OrchestrationV2Command, @@ -5318,24 +5345,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio cause: `Target run ${input.targetRunId} was not found.`, }); } - if (isNativeMaintenanceCommand(input)) { - return yield* new OrchestratorDispatchError({ - commandId: input.command.commandId, - commandType: input.command.type, - cause: - input.text.trim().toLowerCase() === "/compact" - ? "Context compaction must run as a separate turn. Queue it or wait for the active turn to finish." - : "Signing out must run as a separate turn. Queue it or wait for the active turn to finish.", - }); - } - if (parseGoalCommand(input) !== null) { - return yield* new OrchestratorDispatchError({ - commandId: input.command.commandId, - commandType: input.command.type, - cause: - "Goal commands must run as a separate turn. Queue it or wait for the active turn to finish.", - }); - } + yield* ensureSteeringMessage(input.command, input); const targetMessage = input.projection.messages.find( (message) => message.id === targetRun.userMessageId, ); @@ -6547,6 +6557,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio capabilities, }) .pipe(mapDispatchError(command)); + yield* ensureSteeringMessage(command, command); steerWhenReadyRunId = target.id; dispatchMode = { type: "queue_after_active" }; } @@ -6611,6 +6622,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const incomingMessage = { ...command, delegatedCompletion }; if ( shouldQueue && + steerWhenReadyRunId === undefined && command.modelSelection === undefined && queuedRunStop === undefined && !followsCurrentConversation(incomingMessage) && @@ -10410,6 +10422,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio (turn) => turn.runAttemptId === target.activeAttemptId && turn.status === "running", )) ) { + yield* ensureSteeringMessage(command, queuedMessage); const adapter = yield* providerAdapters .get(target.providerInstanceId) .pipe(mapDispatchError(command)); @@ -14664,7 +14677,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio for (const queued of pending) { yield* dispatchWithReceiptEffect({ type: "queued-message.promote-to-steer", - commandId: CommandId.make(`command:deferred-steer:${queued.id}:${run.id}`), + commandId: CommandId.make( + `command:deferred-steer:${queued.id}:${run.id}:${turn.runAttemptId}`, + ), threadId, queuedRunId: queued.id, targetRunId: run.id, diff --git a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts index 1baa1a43dc856..fcd9acaf53e83 100644 --- a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts +++ b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts @@ -854,8 +854,11 @@ it.effect.each( )("delivers a deferred steer for a $status run (direct: $direct)", ({ status, direct }) => Effect.scoped( Effect.gen(function* () { + const restarting = status === "starting" && !direct; const { started, steered, layer, startFirstTurn, events } = yield* nextTurnSelectionHarness( `deferred-steer-${status}-${direct}`, + restarting, + !restarting, ); yield* Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; @@ -883,6 +886,20 @@ it.effect.each( }, ], }); + if (direct) { + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("earlier-queue"), + threadId, + messageId: MessageId.make("message:earlier-queue"), + text: "earlier queued selection", + modelSelection: composerSelection, + attachments: [], + dispatchMode: { type: "queue_after_active" }, + createdBy: "user", + creationSource: "web", + }); + } yield* orchestrator.dispatch({ type: "message.dispatch", commandId: CommandId.make("queue"), @@ -896,8 +913,35 @@ it.effect.each( creationSource: "web", }); const queued = (yield* orchestrator.getThreadProjection(threadId)).runs.find( - (candidate) => candidate.status === "queued", + (candidate) => candidate.userMessageId === MessageId.make("message:steer-after-setup"), )!; + assert.deepEqual(queued.modelSelection, runSelection); + for (const text of ["/compact", "/logout"]) { + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(`queue-${text}`), + threadId, + messageId: MessageId.make(`message:${text}`), + text, + attachments: [], + dispatchMode: { type: "queue_after_active" }, + createdBy: "user", + creationSource: "web", + }); + const maintenance = (yield* orchestrator.getThreadProjection(threadId)).runs.find( + (candidate) => candidate.userMessageId === MessageId.make(`message:${text}`), + )!; + const error = yield* orchestrator + .dispatch({ + type: "queued-message.promote-to-steer", + commandId: CommandId.make(`promote-${text}`), + threadId, + queuedRunId: maintenance.id, + targetRunId: run.id, + }) + .pipe(Effect.flip); + assert.equal(error._tag, "OrchestratorDispatchError"); + } const command = { type: "queued-message.promote-to-steer" as const, commandId: CommandId.make("promote"), @@ -909,6 +953,50 @@ it.effect.each( yield* orchestrator.dispatch(command); yield* orchestrator.dispatch(command); } + let secondQueuedId: typeof queued.id | undefined; + if (restarting) { + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("second"), + threadId, + messageId: MessageId.make("message:second"), + text: "second steer", + attachments: [], + dispatchMode: { type: "queue_after_active" }, + createdBy: "user", + creationSource: "web", + }); + secondQueuedId = (yield* orchestrator.getThreadProjection(threadId)).runs.find( + (candidate) => candidate.userMessageId === MessageId.make("message:second"), + )!.id; + yield* orchestrator.dispatch({ + ...command, + commandId: CommandId.make("promote-second"), + queuedRunId: secondQueuedId, + }); + } + const secondReDeferred = yield* orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => + event.type === "run.updated" && + event.payload.id === secondQueuedId && + event.payload.status === "queued", + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + const secondPromoted = yield* orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => + event.type === "run.updated" && + event.payload.id === secondQueuedId && + event.payload.status === "cancelled", + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); yield* worker.drain(); assert.deepEqual(steered, []); const promoted = yield* orchestrator.streamDomainEvents.pipe( @@ -935,9 +1023,19 @@ it.effect.each( }); yield* Queue.offer(events, { type: "provider_turn.updated", driver, providerTurn: turn }); yield* Fiber.join(promoted); + if (restarting) yield* Fiber.join(secondReDeferred); yield* worker.drain(); - assert.deepEqual(steered, ["steer after setup"]); - assert.lengthOf(started, 1); + if (restarting) { + yield* Fiber.join(secondPromoted); + yield* worker.drain(); + assert.deepEqual( + started.map((turn) => turn.message.text), + ["first", "steer after setup", "second steer"], + ); + } else { + assert.deepEqual(steered, ["steer after setup"]); + assert.lengthOf(started, 1); + } const after = yield* orchestrator.getThreadProjection(threadId); assert.equal( after.messages.find((message) => message.id === queued.userMessageId)?.runId, From 685dfcb17cb68fc0fc282a62a8e79395727093dc Mon Sep 17 00:00:00 2001 From: Yash Singh Date: Wed, 11 Nov 2026 16:15:22 -0600 Subject: [PATCH 3/3] fix(server): isolate deferred steering failures and honor Stop --- .../src/orchestration-v2/Orchestrator.ts | 36 +++-- .../SteeringCompletion.integration.test.ts | 149 ++++++++++++------ 2 files changed, 124 insertions(+), 61 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 175e2de528125..fd7dbceed8e97 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -6538,12 +6538,13 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const targetRunId = dispatchMode.targetRunId; const target = projection.runs.find((run) => run.id === targetRunId); if ( - target?.status === "preparing" || - target?.status === "starting" || - (target?.status === "running" && - !projection.providerTurns.some( - (turn) => turn.runAttemptId === target.activeAttemptId && turn.status === "running", - )) + (target?.status === "preparing" || + target?.status === "starting" || + (target?.status === "running" && + !projection.providerTurns.some( + (turn) => turn.runAttemptId === target.activeAttemptId && turn.status === "running", + ))) && + !(yield* stopReachedRun(command, command.threadId, target.id)) ) { const adapter = yield* providerAdapters .get(target.providerInstanceId) @@ -10415,12 +10416,13 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }); } if ( - target?.status === "preparing" || - target?.status === "starting" || - (target?.status === "running" && - !projection.providerTurns.some( - (turn) => turn.runAttemptId === target.activeAttemptId && turn.status === "running", - )) + (target?.status === "preparing" || + target?.status === "starting" || + (target?.status === "running" && + !projection.providerTurns.some( + (turn) => turn.runAttemptId === target.activeAttemptId && turn.status === "running", + ))) && + !(yield* stopReachedRun(command, command.threadId, target.id)) ) { yield* ensureSteeringMessage(command, queuedMessage); const adapter = yield* providerAdapters @@ -14683,7 +14685,15 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio threadId, queuedRunId: queued.id, targetRunId: run.id, - }); + }).pipe( + Effect.catchCause((cause) => + Effect.logWarning("Failed to deliver a deferred steer", { + threadId, + runId: queued.id, + cause, + }), + ), + ); } }, (effect, stored) => threadDispatch.withLock(stored.event.threadId, effect), diff --git a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts index fcd9acaf53e83..5401c6cc738c7 100644 --- a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts +++ b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts @@ -886,6 +886,30 @@ it.effect.each( }, ], }); + if (status === "preparing" && !direct) { + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("edited-steer"), + threadId, + messageId: MessageId.make("message:edited-steer"), + text: "edit this steer", + attachments: [], + deliveryIntent: "steer", + dispatchMode: { type: "queue_after_active" }, + createdBy: "user", + creationSource: "web", + }); + const edited = (yield* orchestrator.getThreadProjection(threadId)).runs.find( + (candidate) => candidate.userMessageId === MessageId.make("message:edited-steer"), + )!; + yield* orchestrator.dispatch({ + type: "queued-run.edit", + commandId: CommandId.make("edit-steer-to-compact"), + threadId, + runId: edited.id, + text: "/compact", + }); + } if (direct) { yield* orchestrator.dispatch({ type: "message.dispatch", @@ -1981,61 +2005,90 @@ it.effect("keeps a queued message queued when promoted into a run that Stop reac ); // regression: #32891 a restart that Stop already reached is lost -it.effect("refuses a steer but starts a restart sent into a run that Stop reached", () => - Effect.scoped( - Effect.gen(function* () { - const { started, steered, layer, startFirstTurn } = - yield* nextTurnSelectionHarness("restart-after-stop"); - yield* Effect.gen(function* () { - const orchestrator = yield* Orchestrator.OrchestratorV2; - const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; - const threadId = yield* startFirstTurn; - const first = started[0]!; - // The worker has not delivered the interrupt yet, so the run still reads as running. - yield* orchestrator.dispatch({ - type: "run.interrupt", - commandId: CommandId.make("stop"), - threadId, - runId: first.runId, - }); - const steerError = yield* orchestrator - .dispatch({ +it.effect.each([false, true])( + "refuses a steer after Stop (provider turn ready: %s)", + (providerTurnReady) => + Effect.scoped( + Effect.gen(function* () { + const { started, steered, layer, startFirstTurn } = + yield* nextTurnSelectionHarness("restart-after-stop"); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const threadId = yield* startFirstTurn; + const first = started[0]!; + // The worker has not delivered the interrupt yet, so the run still reads as running. + yield* orchestrator.dispatch({ + type: "run.interrupt", + commandId: CommandId.make("stop"), + threadId, + runId: first.runId, + }); + const turn = (yield* orchestrator.getThreadProjection(threadId)).providerTurns[0]!; + if (!providerTurnReady) { + yield* (yield* EventSink.EventSinkV2).write({ + events: [ + { + id: EventId.make("turn-not-ready"), + type: "provider-turn.updated", + threadId, + occurredAt: yield* DateTime.now, + payload: { ...turn, status: "pending" }, + }, + ], + }); + } + const steerError = yield* orchestrator + .dispatch({ + type: "message.dispatch", + commandId: CommandId.make("steer"), + threadId, + messageId: MessageId.make("message:steer"), + text: "steer", + attachments: [], + modelSelection: runSelection, + dispatchMode: { type: "steer_active", targetRunId: first.runId }, + createdBy: "user", + creationSource: "web", + }) + .pipe(Effect.flip); + assert.equal(steerError._tag, "OrchestratorDispatchError"); + assert.include(String(steerError.cause), "is stopping and cannot be steered"); + if (!providerTurnReady) { + yield* (yield* EventSink.EventSinkV2).write({ + events: [ + { + id: EventId.make("restore-ready-turn"), + type: "provider-turn.updated", + threadId, + occurredAt: yield* DateTime.now, + payload: turn, + }, + ], + }); + } + yield* orchestrator.dispatch({ type: "message.dispatch", - commandId: CommandId.make("steer"), + commandId: CommandId.make("restart"), threadId, - messageId: MessageId.make("message:steer"), - text: "steer", + messageId: MessageId.make("message:restart"), + text: "restart", attachments: [], modelSelection: runSelection, - dispatchMode: { type: "steer_active", targetRunId: first.runId }, + dispatchMode: { type: "restart_active", targetRunId: first.runId }, createdBy: "user", creationSource: "web", - }) - .pipe(Effect.flip); - assert.equal(steerError._tag, "OrchestratorDispatchError"); - assert.include(String(steerError.cause), "is stopping and cannot be steered"); - yield* orchestrator.dispatch({ - type: "message.dispatch", - commandId: CommandId.make("restart"), - threadId, - messageId: MessageId.make("message:restart"), - text: "restart", - attachments: [], - modelSelection: runSelection, - dispatchMode: { type: "restart_active", targetRunId: first.runId }, - createdBy: "user", - creationSource: "web", - }); - yield* worker.drain(); + }); + yield* worker.drain(); - assert.deepEqual(steered, []); - assert.deepEqual( - started.map((turn) => turn.message.text), - ["first", "restart"], - ); - }).pipe(Effect.provide(layer)); - }), - ), + assert.deepEqual(steered, []); + assert.deepEqual( + started.map((turn) => turn.message.text), + ["first", "restart"], + ); + }).pipe(Effect.provide(layer)); + }), + ), ); // regression: #32891 a restart that arrives after the turn ended is lost