From 4b033a3b7a1dfa995c312269872397287eca9a3b Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 17 Sep 2026 02:11:43 -0700 Subject: [PATCH 01/12] fix(orchestrator): defer queued provider handoff until delivery --- .../src/state/use-thread-outbox-drain.ts | 5 +- .../src/orchestration-v2/Orchestrator.ts | 293 +++++++++++++++++- .../src/orchestration-v2/ProjectionStore.ts | 8 +- .../ProviderSwitch.integration.test.ts | 187 +++++++++++ apps/web/src/components/ChatView.tsx | 4 +- 5 files changed, 481 insertions(+), 16 deletions(-) diff --git a/apps/mobile/src/state/use-thread-outbox-drain.ts b/apps/mobile/src/state/use-thread-outbox-drain.ts index d40608ff041f..007077fc6c92 100644 --- a/apps/mobile/src/state/use-thread-outbox-drain.ts +++ b/apps/mobile/src/state/use-thread-outbox-drain.ts @@ -705,7 +705,10 @@ export function useThreadOutboxDrain(): void { } const { reportFailure } = makeDeliveryHelpers(queuedMessage); - if (!modelSelectionsEqual(settings.modelSelection, thread.modelSelection)) { + if ( + queuedMessage.dispatchMode !== "queue" && + !modelSelectionsEqual(settings.modelSelection, thread.modelSelection) + ) { const updateResult = await updateThreadMetadata({ environmentId: queuedMessage.environmentId, input: { diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 8cfd4d097c8f..124b3677f156 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -932,6 +932,142 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const commandId = CommandId.make(`command:system:start-queued:${queuedRun.id}`); const now = yield* DateTime.now; + const selectionChanged = !modelSelectionsEqual( + projection.thread.modelSelection, + queuedRun.modelSelection, + ); + const switchPlan = selectionChanged + ? yield* providerSwitchService + .plan({ projection, targetModelSelection: queuedRun.modelSelection }) + .pipe( + Effect.mapError( + (cause) => + new OrchestratorDispatchError({ + commandId, + commandType: "message.dispatch", + cause, + }), + ), + ) + : null; + const targetAdapter = yield* providerAdapters.get(queuedRun.providerInstanceId).pipe( + Effect.mapError( + (cause) => + new OrchestratorDispatchError({ + commandId, + commandType: "message.dispatch", + cause, + }), + ), + ); + const targetCapabilities = yield* targetAdapter.getCapabilities().pipe( + Effect.mapError( + (cause) => + new OrchestratorDispatchError({ + commandId, + commandType: "message.dispatch", + cause, + }), + ), + ); + const latestCompletedRun = projection.runs.findLast((run) => run.status === "completed"); + const coveredRuns = + latestCompletedRun === undefined || + latestCompletedRun.providerInstanceId === queuedRun.providerInstanceId + ? [] + : projection.runs.filter( + (run) => + run.status === "completed" && + run.ordinal > (queuedProviderThread.lastRunOrdinal ?? 0) && + run.ordinal <= latestCompletedRun.ordinal, + ); + const needsFullContext = queuedProviderThread.nativeThreadRef === null; + const handoffStrategy = needsFullContext + ? ("full_thread_summary" as const) + : ("delta_since_target_last_seen" as const); + const transferId = + coveredRuns.length === 0 + ? null + : yield* idAllocator.allocate + .contextTransfer({ + sourceThreadId: threadId, + targetThreadId: threadId, + type: "provider_handoff", + }) + .pipe( + Effect.mapError( + (cause) => + new OrchestratorDispatchError({ + commandId, + commandType: "message.dispatch", + cause, + }), + ), + ); + if (transferId !== null) { + yield* commandPolicy + .ensureContextHandoff({ + commandId, + threadId, + providerInstanceId: queuedRun.providerInstanceId, + capabilities: targetCapabilities, + strategy: needsFullContext ? "full_thread_summary" : "delta_context", + }) + .pipe( + Effect.mapError( + (cause) => + new OrchestratorDispatchError({ + commandId, + commandType: "message.dispatch", + cause, + }), + ), + ); + } + const handoff = + transferId === null || latestCompletedRun === undefined + ? null + : yield* contextHandoffService + .prepareProviderHandoff({ + threadId, + targetRunId: queuedRun.id, + transferId, + fromProviderThreadIds: Array.from( + new Set( + coveredRuns.flatMap((run) => + run.providerThreadId === null ? [] : [run.providerThreadId], + ), + ), + ), + toProviderThreadId: queuedProviderThread.id, + fromProviderInstanceId: latestCompletedRun.providerInstanceId, + toProviderInstanceId: queuedRun.providerInstanceId, + coveredRunOrdinals: { + from: coveredRuns[0]!.ordinal, + to: coveredRuns.at(-1)!.ordinal, + }, + strategy: handoffStrategy, + items: [ + ...(needsFullContext && projection.thread.historyOrigin === "v1_import" + ? projection.turnItems.filter((item) => item.runId === null) + : []), + ...projection.turnItems.filter( + (item) => + item.runId !== null && coveredRuns.some((run) => run.id === item.runId), + ), + ], + createdAt: now, + }) + .pipe( + Effect.mapError( + (cause) => + new OrchestratorDispatchError({ + commandId, + commandType: "message.dispatch", + cause, + }), + ), + ); const checkpointScope = storedCheckpointScope ?? (yield* runtimePolicy @@ -957,7 +1093,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ), )); const providerSessionId = - queuedProviderThread.providerSessionId ?? + (queuedProviderThread.providerSessionId !== null && + !switchPlan?.releaseProviderSessionIds.includes(queuedProviderThread.providerSessionId) + ? queuedProviderThread.providerSessionId + : null) ?? (yield* providerAdapters.get(queuedRun.providerInstanceId).pipe( Effect.flatMap((adapter) => providerSessionIdFor({ @@ -981,6 +1120,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio status: "not_loaded", firstRunOrdinal: queuedProviderThread.firstRunOrdinal ?? queuedRun.ordinal, lastRunOrdinal: queuedRun.ordinal, + handoffIds: appendContextHandoffId(queuedProviderThread.handoffIds, handoff?.id ?? null), updatedAt: now, }; const startingRun: OrchestrationV2Run = { @@ -988,6 +1128,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio status: "starting", queuePosition: null, startedAt: null, + contextHandoffId: handoff?.id ?? null, }; const userTurnItem: OrchestrationV2TurnItem = { ...(legacyQueuedTurnItem ?? { @@ -1041,8 +1182,97 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }, ] : []; + const sessionsToDetach = projection.providerSessions.filter( + (session) => + switchPlan?.releaseProviderSessionIds.includes(session.id) && + session.status !== "stopped" && + session.status !== "error", + ); yield* writeSystemEvents( [ + ...(selectionChanged + ? [ + { + type: + queuedRun.providerInstanceId === projection.thread.providerInstanceId + ? ("thread.model-selection-updated" as const) + : ("thread.provider-switched" as const), + threadId, + providerInstanceId: queuedRun.providerInstanceId, + occurredAt: now, + payload: { + ...projection.thread, + providerInstanceId: queuedRun.providerInstanceId, + modelSelection: queuedRun.modelSelection, + updatedAt: now, + }, + }, + ] + : []), + ...(handoff === null || transferId === null || latestCompletedRun === undefined + ? [] + : [ + { + type: "context-transfer.created" as const, + threadId, + runId: queuedRun.id, + providerInstanceId: queuedRun.providerInstanceId, + occurredAt: now, + payload: { + id: transferId, + type: "provider_handoff" as const, + sourceThreadId: threadId, + targetThreadId: threadId, + sourcePoint: contextSourcePointForRun(projection, latestCompletedRun), + basePoint: + needsFullContext || queuedProviderThread.lastRunOrdinal === null + ? null + : (() => { + const baseRun = projection.runs.find( + (run) => run.ordinal === queuedProviderThread.lastRunOrdinal, + ); + return baseRun === undefined + ? null + : contextSourcePointForRun(projection, baseRun); + })(), + sourceProviderInstanceId: latestCompletedRun.providerInstanceId, + targetProviderInstanceId: queuedRun.providerInstanceId, + targetRunId: queuedRun.id, + status: "consumed" as const, + resolution: { + strategy: needsFullContext + ? ("portable_context" as const) + : ("delta_context" as const), + contextHandoffId: handoff.id, + }, + createdBy: queuedMessage.createdBy, + error: null, + createdAt: now, + updatedAt: now, + consumedAt: now, + }, + }, + { + type: "context-handoff.updated" as const, + threadId, + runId: queuedRun.id, + providerInstanceId: queuedRun.providerInstanceId, + occurredAt: now, + payload: handoff, + }, + ]), + ...sessionsToDetach.map((session) => ({ + type: "provider-session.detached" as const, + threadId, + driver: session.driver, + providerInstanceId: session.providerInstanceId, + occurredAt: now, + payload: { + providerSessionId: session.id, + detachedAt: now, + reason: "Provider or model selection changed.", + }, + })), ...checkpointEvents, { type: "provider-thread.updated", @@ -1071,6 +1301,16 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }, ], [ + ...sessionsToDetach.map((session) => ({ + id: `effect:${commandId}:provider-session.detach:${session.id}`, + commandId, + threadId, + request: { + type: "provider-session.detach" as const, + providerSessionId: session.id, + detail: "Provider or model selection changed.", + }, + })), { id: `effect:${commandId}:provider-turn.start:${queuedRun.id}`, commandId, @@ -3591,13 +3831,6 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio cause: `Active run ${activeRun.id} has no provider thread for queued dispatch.`, }); } - if (modelSelection.instanceId !== queueProviderThread.providerInstanceId) { - return yield* new OrchestratorDispatchError({ - commandId: command.commandId, - commandType: command.type, - cause: `Queued dispatch for provider instance ${modelSelection.instanceId} cannot run behind active provider instance ${queueProviderThread.providerInstanceId}.`, - }); - } const existingProviderSession = queueProviderThread.providerSessionId === null ? undefined @@ -3618,6 +3851,33 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const now = yield* DateTime.now; const ordinal = nextRunOrdinal(projection); const runId = idAllocator.derive.run({ threadId: command.threadId, ordinal }); + const targetProviderThread = + modelSelection.instanceId === queueProviderThread.providerInstanceId + ? queueProviderThread + : rootProviderThreadsForProvider(projection, modelSelection.instanceId)[0]; + const queuedAdapter = yield* providerAdapters + .get(modelSelection.instanceId) + .pipe(mapDispatchError(command)); + const queuedProviderThread: OrchestrationV2ProviderThread = targetProviderThread ?? { + id: idAllocator.derive.providerThread({ + driver: queuedAdapter.driver, + nativeThreadId: `pending:${runId}`, + }), + driver: queuedAdapter.driver, + providerInstanceId: modelSelection.instanceId, + providerSessionId: null, + appThreadId: command.threadId, + ownerNodeId: null, + nativeThreadRef: null, + nativeConversationHeadRef: null, + status: "not_loaded", + firstRunOrdinal: null, + lastRunOrdinal: null, + handoffIds: [], + forkedFrom: null, + createdAt: now, + updatedAt: now, + }; const attemptId = idAllocator.derive.runAttempt({ runId, attemptOrdinal: 1 }); const rootNodeId = idAllocator.derive.rootNode({ runId }); const checkpointScope = @@ -3629,7 +3889,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio threadId: command.threadId, runId, rootNodeId, - providerThreadId: queueProviderThread.id, + providerThreadId: queuedProviderThread.id, cwd: resolvedRuntimePolicy.cwd ?? existingProviderSession?.cwd ?? @@ -3653,7 +3913,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ordinal, providerInstanceId: modelSelection.instanceId, modelSelection, - providerThreadId: queueProviderThread.id, + providerThreadId: queuedProviderThread.id, userMessageId: command.messageId, rootNodeId, activeAttemptId: attemptId, @@ -3681,7 +3941,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio attemptOrdinal: 1, rootNodeId, providerInstanceId: modelSelection.instanceId, - providerThreadId: queueProviderThread.id, + providerThreadId: queuedProviderThread.id, providerTurnId: null, reason: "initial", status: "pending", @@ -3697,7 +3957,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio kind: "root_turn", status: "pending", countsForRun: true, - providerThreadId: queueProviderThread.id, + providerThreadId: queuedProviderThread.id, providerTurnId: null, nativeItemRef: null, runtimeRequestId: null, @@ -3726,6 +3986,15 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ...(command.notification === undefined ? {} : { notification: command.notification }), }; const emitEvent = emit(events, command); + if (targetProviderThread === undefined) { + yield* emitEvent({ + type: "provider-thread.updated", + threadId: command.threadId, + providerInstanceId: modelSelection.instanceId, + occurredAt: now, + payload: queuedProviderThread, + }); + } yield* emitEvent({ type: "run.created", threadId: command.threadId, diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 09a79aed4ca8..ac427ee87c4c 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -546,7 +546,8 @@ export function applyToProjection( return { ...base, thread: - event.payload.appThreadId === base.thread.id + event.payload.appThreadId === base.thread.id && + !(event.payload.status === "not_loaded" && event.payload.firstRunOrdinal === null) ? { ...base.thread, activeProviderThreadId: event.payload.id, @@ -1830,7 +1831,10 @@ export const layer: Layer.Layer = updated_at = excluded.updated_at, payload_json = excluded.payload_json `; - if (event.payload.appThreadId !== null) { + if ( + event.payload.appThreadId !== null && + !(event.payload.status === "not_loaded" && event.payload.firstRunOrdinal === null) + ) { const threadRows = yield* sql` SELECT payload_json FROM orchestration_v2_projection_threads diff --git a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts index 9b009720d3a0..268ec69a7fd3 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -1,6 +1,7 @@ import { assert, describe, it } from "@effect/vitest"; import { CommandId, + EventId, MessageId, type ModelSelection, type OrchestrationV2Command, @@ -16,6 +17,7 @@ import { ProviderDriverKind, } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as PubSub from "effect/PubSub"; @@ -28,12 +30,14 @@ import { ClaudeProviderCapabilitiesV2 } from "../Adapters/ClaudeAdapterV2.ts"; import { CodexProviderCapabilitiesV2 } from "../Adapters/CodexAdapterV2.ts"; import { CursorProviderCapabilitiesV2 } from "../Adapters/CursorAdapterV2.ts"; import { layer as eventSinkLayer } from "../EventSink.ts"; +import { EventSinkV2 } from "../EventSink.ts"; import { layer as eventStoreLayer } from "../EventStore.ts"; import { LegacyV1ThreadImporter, layer as legacyV1ThreadImporterLayer, } from "../LegacyV1ThreadImporter.ts"; import { OrchestratorV2 } from "../Orchestrator.ts"; +import { OrchestrationEffectWorkerV2 } from "../EffectWorker.ts"; import { ProjectionMaintenanceV2, layer as projectionMaintenanceLayer, @@ -84,6 +88,7 @@ function makeTestAdapter(input: { readonly capturedTurns: Ref.Ref>; readonly failResume?: boolean; readonly failedRunOrdinals?: ReadonlySet; + readonly holdFirstTurn?: Deferred.Deferred; }): ProviderAdapterV2Shape { return { instanceId: input.instanceId, @@ -155,6 +160,10 @@ function makeTestAdapter(input: { text: turnInput.message.text, }, ]); + if (turnInput.runOrdinal === 1 && input.holdFirstTurn !== undefined) { + yield* Deferred.succeed(input.holdFirstTurn, undefined); + return; + } const eventTime = yield* DateTime.now; const providerTurnId = ProviderTurnId.make( `provider-turn:${input.driver}:${turnInput.threadId}:${turnInput.runOrdinal}`, @@ -296,6 +305,184 @@ const waitForIdle = Effect.fn("ProviderSwitchTest.waitForIdle")(function* ( }); describe("orchestration v2 provider switching", () => { + it.live("finishes earlier queued Codex turns before handing context to queued Claude", () => + Effect.scoped( + Effect.gen(function* () { + const cwd = yield* checkpointWorkspace("queued-provider-switch"); + const capturedTurns = yield* Ref.make>([]); + const started = yield* Deferred.make(); + const registryLayer = makeProviderAdapterRegistryLayer([ + makeTestAdapter({ + instanceId: ProviderInstanceId.make("codex"), + driver: CODEX_DRIVER, + capabilities: CodexProviderCapabilitiesV2, + modelSelection: CODEX_MODEL_SELECTION, + responseByRunOrdinal: { + 1: "Codex current turn complete", + 2: "Codex first queued turn complete", + 3: "Codex second queued turn complete", + }, + capturedTurns, + holdFirstTurn: started, + }), + makeTestAdapter({ + instanceId: ProviderInstanceId.make("claudeAgent"), + driver: CLAUDE_DRIVER, + capabilities: ClaudeProviderCapabilitiesV2, + modelSelection: CLAUDE_MODEL_SELECTION, + responseByRunOrdinal: { 4: "Claude turn complete" }, + capturedTurns, + }), + ]); + const queuedThreadId = ThreadId.make("thread:queued-provider-switch"); + const projection = yield* Effect.gen(function* () { + const orchestrator = yield* OrchestratorV2; + const worker = yield* OrchestrationEffectWorkerV2; + const eventSink = yield* EventSinkV2; + const dispatch = (ordinal: number, modelSelection: ModelSelection) => + orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`command:queued-provider-switch:${ordinal}`), + threadId: queuedThreadId, + messageId: MessageId.make(`message:queued-provider-switch:${ordinal}`), + text: `Prompt ${ordinal}`, + attachments: [], + modelSelection, + dispatchMode: { + type: ordinal === 1 ? "start_immediately" : "queue_after_active", + }, + }); + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("command:queued-provider-switch:create"), + threadId: queuedThreadId, + projectId: ProjectId.make("project:queued-provider-switch"), + title: "Queued provider switch", + modelSelection: CODEX_MODEL_SELECTION, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: cwd, + }); + yield* dispatch(1, CODEX_MODEL_SELECTION); + yield* Deferred.await(started); + yield* dispatch(2, CODEX_MODEL_SELECTION); + yield* dispatch(3, CODEX_MODEL_SELECTION); + yield* dispatch(4, CLAUDE_MODEL_SELECTION); + const queued = yield* orchestrator.getThreadProjection(queuedThreadId); + assert.deepEqual( + queued.runs.map((run) => run.status), + ["running", "queued", "queued", "queued"], + ); + assert.equal(queued.thread.activeProviderThreadId, queued.runs[0]?.providerThreadId); + assert.equal(queued.thread.providerInstanceId, CODEX_MODEL_SELECTION.instanceId); + assert.lengthOf(queued.contextHandoffs, 0); + const now = yield* DateTime.now; + yield* eventSink.write({ + events: [ + { + id: EventId.make("event:queued-provider-switch:first-response"), + type: "turn-item.updated", + threadId: queuedThreadId, + runId: queued.runs[0]!.id, + nodeId: queued.runs[0]!.rootNodeId!, + providerInstanceId: CODEX_MODEL_SELECTION.instanceId, + occurredAt: now, + payload: { + id: TurnItemId.make("turn-item:queued-provider-switch:first-response"), + threadId: queuedThreadId, + runId: queued.runs[0]!.id, + nodeId: queued.runs[0]!.rootNodeId!, + providerThreadId: queued.runs[0]!.providerThreadId, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 101, + status: "completed", + title: null, + startedAt: now, + completedAt: now, + updatedAt: now, + type: "assistant_message", + messageId: MessageId.make("message:queued-provider-switch:first-response"), + text: "Codex current turn complete", + streaming: false, + }, + }, + { + id: EventId.make("event:queued-provider-switch:first-complete"), + type: "run.updated", + threadId: queuedThreadId, + runId: queued.runs[0]!.id, + providerInstanceId: CODEX_MODEL_SELECTION.instanceId, + occurredAt: now, + payload: { ...queued.runs[0]!, status: "completed", completedAt: now }, + }, + ], + }); + yield* orchestrator.resumeQueuedRuns; + yield* orchestrator.streamStoredEvents.pipe( + Stream.filter( + (event) => + event.event.type === "run.updated" && + event.event.runId === queued.runs[3]?.id && + event.event.payload.status === "completed", + ), + Stream.runHead, + ); + yield* worker.drain(); + return yield* orchestrator.getThreadProjection(queuedThreadId); + }).pipe( + Effect.provide( + makeOrchestratorV2ReplayLayerWithRegistry( + { + name: "queued-provider-switch", + runtimePolicyOverride: { + cwd, + approvalPolicy: "never", + sandboxPolicy: { + type: "readOnly", + access: { type: "fullAccess" }, + networkAccess: false, + }, + }, + }, + registryLayer, + ), + ), + ); + const turns = yield* Ref.get(capturedTurns); + assert.deepEqual( + projection.runs.map((run) => [run.providerInstanceId, run.status]), + [ + ["codex", "completed"], + ["codex", "completed"], + ["codex", "completed"], + ["claudeAgent", "completed"], + ], + ); + assert.deepEqual( + turns.map((turn) => [turn.driver, turn.text.includes("Prompt 4")]), + [ + ["codex", false], + ["codex", false], + ["codex", false], + ["claudeAgent", true], + ], + ); + assert.lengthOf(projection.contextHandoffs, 1); + assert.equal(projection.contextHandoffs[0]?.targetRunId, projection.runs[3]?.id); + assert.include(turns[3]?.text ?? "", "Codex current turn complete"); + assert.include(turns[3]?.text ?? "", "Codex first queued turn complete"); + assert.include(turns[3]?.text ?? "", "Codex second queued turn complete"); + }), + ), + ); + it.live("reissues imported v1 context when switching after the first provider fails", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index a075c7ce9f72..8f38287af5ea 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -8817,7 +8817,9 @@ export default function ChatView(props: ChatViewProps) { const settingsResult = await persistThreadSettingsForNextTurn({ threadId: threadIdForSend, createdAt: messageCreatedAt, - ...(ctxSelectedModel ? { modelSelection: ctxSelectedModelSelection } : {}), + ...(ctxSelectedModel && !shouldQueueBehindActiveRun + ? { modelSelection: ctxSelectedModelSelection } + : {}), ...(localCheckoutBranchMismatch ? { branch: localCheckoutBranchMismatch.currentBranch } : {}), From d1c2161725e262f0b6edda733019e80abfe01856 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 17 Sep 2026 02:16:10 -0700 Subject: [PATCH 02/12] fix(orchestrator): preserve imported native provider mapping --- .../src/orchestration-v2/ProjectionStore.ts | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index ac427ee87c4c..cbef23b92134 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -6,6 +6,7 @@ import type { OrchestrationV2ConversationMessage, OrchestrationV2DomainEvent, OrchestrationV2ProjectedTurnItem, + OrchestrationV2ProviderThread, OrchestrationV2ProviderTurn, OrchestrationV2Run, OrchestrationV2Subagent, @@ -451,6 +452,16 @@ export function emptyProjection( }; } +// A future queued provider has a reserved thread record but is not active until delivery. +function isQueuedProviderThreadPlaceholder(providerThread: OrchestrationV2ProviderThread): boolean { + return ( + providerThread.status === "not_loaded" && + providerThread.firstRunOrdinal === null && + providerThread.nativeThreadRef === null && + providerThread.providerSessionId === null + ); +} + export function applyToProjection( projection: OrchestrationV2ThreadProjection, event: OrchestrationV2DomainEvent, @@ -547,7 +558,7 @@ export function applyToProjection( ...base, thread: event.payload.appThreadId === base.thread.id && - !(event.payload.status === "not_loaded" && event.payload.firstRunOrdinal === null) + !isQueuedProviderThreadPlaceholder(event.payload) ? { ...base.thread, activeProviderThreadId: event.payload.id, @@ -1833,7 +1844,7 @@ export const layer: Layer.Layer = `; if ( event.payload.appThreadId !== null && - !(event.payload.status === "not_loaded" && event.payload.firstRunOrdinal === null) + !isQueuedProviderThreadPlaceholder(event.payload) ) { const threadRows = yield* sql` SELECT payload_json From 81961410fbd8cf63e1b9a712ebeeb042db99a0af Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 17 Sep 2026 02:20:22 -0700 Subject: [PATCH 03/12] fix(queue): retain queued delivery mode and native continuation --- apps/mobile/src/state/thread-outbox-model.ts | 1 + apps/mobile/src/state/thread-outbox.test.ts | 9 + .../src/orchestration-v2/Orchestrator.ts | 25 ++- .../ProviderSwitch.integration.test.ts | 183 ++++++++++++++++++ 4 files changed, 215 insertions(+), 3 deletions(-) diff --git a/apps/mobile/src/state/thread-outbox-model.ts b/apps/mobile/src/state/thread-outbox-model.ts index 97b367a07b6a..99a53ae9a4ba 100644 --- a/apps/mobile/src/state/thread-outbox-model.ts +++ b/apps/mobile/src/state/thread-outbox-model.ts @@ -55,6 +55,7 @@ export const QueuedThreadMessageSchema = Schema.Struct({ context: Schema.optional(OrchestrationMessageContext), attachments: Schema.Array(DraftComposerAttachmentSchema), modelSelection: Schema.optional(ModelSelection), + dispatchMode: Schema.optional(Schema.Literals(["auto", "queue", "steer", "restart"])), runtimeMode: Schema.optional(RuntimeMode), interactionMode: Schema.optional(ProviderInteractionMode), // Present when the queued item creates a brand-new thread (pending task) diff --git a/apps/mobile/src/state/thread-outbox.test.ts b/apps/mobile/src/state/thread-outbox.test.ts index 5a8f2d90f606..734caeab226a 100644 --- a/apps/mobile/src/state/thread-outbox.test.ts +++ b/apps/mobile/src/state/thread-outbox.test.ts @@ -140,6 +140,15 @@ describe("thread outbox", () => { decodeQueuedThreadMessage(JSON.parse(JSON.stringify(encodeQueuedThreadMessage(message)))), ).toEqual(message); }); + it("retains queue mode when a queued provider switch reloads from storage", () => { + const message: QueuedThreadMessage = { + ...queuedMessage({ messageId: "queued-switch", createdAt: "2026-09-17T09:00:00.000Z" }), + dispatchMode: "queue", + }; + expect( + decodeQueuedThreadMessage(JSON.parse(JSON.stringify(encodeQueuedThreadMessage(message)))), + ).toEqual(message); + }); it.each(["read", "json", "schema"] as const)( "recovers usable messages without permitting cleanup after a record %s failure", async (failure) => { diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 124b3677f156..6ef763b8608b 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -950,6 +950,23 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ), ) : null; + const activeProviderThread = projection.providerThreads.find( + (candidate) => candidate.id === projection.thread.activeProviderThreadId, + ); + const canResumeAcrossInstances = + switchPlan?.instanceChanged === true && + switchPlan.transition.type === "restart_and_resume" && + activeProviderThread !== undefined && + activeProviderThread.nativeThreadRef !== null; + const deliveryProviderThread = + canResumeAcrossInstances && activeProviderThread !== undefined + ? { + ...queuedProviderThread, + nativeThreadRef: activeProviderThread.nativeThreadRef, + nativeConversationHeadRef: activeProviderThread.nativeConversationHeadRef, + nativeMetadata: activeProviderThread.nativeMetadata, + } + : queuedProviderThread; const targetAdapter = yield* providerAdapters.get(queuedRun.providerInstanceId).pipe( Effect.mapError( (cause) => @@ -972,6 +989,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ); const latestCompletedRun = projection.runs.findLast((run) => run.status === "completed"); const coveredRuns = + canResumeAcrossInstances || latestCompletedRun === undefined || latestCompletedRun.providerInstanceId === queuedRun.providerInstanceId ? [] @@ -981,7 +999,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio run.ordinal > (queuedProviderThread.lastRunOrdinal ?? 0) && run.ordinal <= latestCompletedRun.ordinal, ); - const needsFullContext = queuedProviderThread.nativeThreadRef === null; + const needsFullContext = deliveryProviderThread.nativeThreadRef === null; const handoffStrategy = needsFullContext ? ("full_thread_summary" as const) : ("delta_since_target_last_seen" as const); @@ -1093,7 +1111,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ), )); const providerSessionId = - (queuedProviderThread.providerSessionId !== null && + (!canResumeAcrossInstances && + queuedProviderThread.providerSessionId !== null && !switchPlan?.releaseProviderSessionIds.includes(queuedProviderThread.providerSessionId) ? queuedProviderThread.providerSessionId : null) ?? @@ -1115,7 +1134,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ), )); const providerThread: OrchestrationV2ProviderThread = { - ...queuedProviderThread, + ...deliveryProviderThread, providerSessionId, status: "not_loaded", firstRunOrdinal: queuedProviderThread.firstRunOrdinal ?? queuedRun.ordinal, diff --git a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts index 268ec69a7fd3..faa225550985 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -49,6 +49,10 @@ import { type ProviderAdapterV2Shape, } from "../ProviderAdapter.ts"; import { makeLayer as makeProviderAdapterRegistryLayer } from "../ProviderAdapterRegistry.ts"; +import { + ProviderAdapterRegistryLookupError, + ProviderAdapterRegistryV2, +} from "../ProviderAdapterRegistry.ts"; import { makeProviderFailure } from "../ProviderFailure.ts"; import { CLAUDE_MODEL_SELECTION, @@ -305,6 +309,185 @@ const waitForIdle = Effect.fn("ProviderSwitchTest.waitForIdle")(function* ( }); describe("orchestration v2 provider switching", () => { + it.live("resumes a queued account switch without requiring a portable handoff", () => + Effect.scoped( + Effect.gen(function* () { + const cwd = yield* checkpointWorkspace("queued-account-switch"); + const capturedTurns = yield* Ref.make>([]); + const started = yield* Deferred.make(); + const alternateSelection: ModelSelection = { + ...CODEX_MODEL_SELECTION, + instanceId: ProviderInstanceId.make("codex-alternate"), + }; + const alternateCapabilities = { + ...CodexProviderCapabilitiesV2, + canConsumeHandoffSummaries: false, + }; + const adapters = [ + makeTestAdapter({ + instanceId: CODEX_MODEL_SELECTION.instanceId, + driver: CODEX_DRIVER, + capabilities: CodexProviderCapabilitiesV2, + modelSelection: CODEX_MODEL_SELECTION, + responseByRunOrdinal: {}, + capturedTurns, + holdFirstTurn: started, + }), + makeTestAdapter({ + instanceId: alternateSelection.instanceId, + driver: CODEX_DRIVER, + capabilities: alternateCapabilities, + modelSelection: alternateSelection, + responseByRunOrdinal: { 2: "Alternate account complete" }, + capturedTurns, + }), + ]; + const registryLayer = Layer.succeed( + ProviderAdapterRegistryV2, + ProviderAdapterRegistryV2.of({ + get: (instanceId) => { + const adapter = adapters.find((candidate) => candidate.instanceId === instanceId); + return adapter === undefined + ? Effect.fail(new ProviderAdapterRegistryLookupError({ instanceId })) + : Effect.succeed(adapter); + }, + list: () => Effect.succeed(adapters.map((adapter) => adapter.instanceId)), + getMetadata: (instanceId) => { + const adapter = adapters.find((candidate) => candidate.instanceId === instanceId); + return adapter === undefined + ? Effect.fail(new ProviderAdapterRegistryLookupError({ instanceId })) + : Effect.succeed({ + driver: CODEX_DRIVER, + continuationKey: "codex:shared-native-account-history", + enabled: true, + capabilities: + instanceId === alternateSelection.instanceId + ? alternateCapabilities + : CodexProviderCapabilitiesV2, + }); + }, + }), + ); + const queuedThreadId = ThreadId.make("thread:queued-account-switch"); + const projection = yield* Effect.gen(function* () { + const orchestrator = yield* OrchestratorV2; + const eventSink = yield* EventSinkV2; + const worker = yield* OrchestrationEffectWorkerV2; + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("command:queued-account-switch:create"), + threadId: queuedThreadId, + projectId: ProjectId.make("project:queued-account-switch"), + title: "Queued account switch", + modelSelection: CODEX_MODEL_SELECTION, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: cwd, + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("command:queued-account-switch:first"), + threadId: queuedThreadId, + messageId: MessageId.make("message:queued-account-switch:first"), + text: "First account turn", + attachments: [], + modelSelection: CODEX_MODEL_SELECTION, + dispatchMode: { type: "start_immediately" }, + }); + yield* Deferred.await(started); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("command:queued-account-switch:second"), + threadId: queuedThreadId, + messageId: MessageId.make("message:queued-account-switch:second"), + text: "Alternate account turn", + attachments: [], + modelSelection: alternateSelection, + dispatchMode: { type: "queue_after_active" }, + }); + const queued = yield* orchestrator.getThreadProjection(queuedThreadId); + assert.deepEqual( + queued.runs.map((run) => run.status), + ["running", "queued"], + ); + assert.equal(queued.thread.providerInstanceId, CODEX_MODEL_SELECTION.instanceId); + const sourceNativeRef = queued.providerThreads.find( + (providerThread) => providerThread.id === queued.runs[0]?.providerThreadId, + )?.nativeThreadRef; + assert.isNotNull(sourceNativeRef); + const now = yield* DateTime.now; + yield* eventSink.write({ + events: [ + { + id: EventId.make("event:queued-account-switch:first-complete"), + type: "run.updated", + threadId: queuedThreadId, + runId: queued.runs[0]!.id, + providerInstanceId: CODEX_MODEL_SELECTION.instanceId, + occurredAt: now, + payload: { ...queued.runs[0]!, status: "completed", completedAt: now }, + }, + ], + }); + yield* orchestrator.resumeQueuedRuns; + yield* orchestrator.streamStoredEvents.pipe( + Stream.filter( + (event) => + event.event.type === "run.updated" && + event.event.runId === queued.runs[1]?.id && + event.event.payload.status === "completed", + ), + Stream.runHead, + ); + yield* worker.drain(); + const delivered = yield* orchestrator.getThreadProjection(queuedThreadId); + const targetNativeRef = delivered.providerThreads.find( + (providerThread) => providerThread.id === delivered.runs[1]?.providerThreadId, + )?.nativeThreadRef; + assert.deepEqual(targetNativeRef, sourceNativeRef); + return delivered; + }).pipe( + Effect.provide( + makeOrchestratorV2ReplayLayerWithRegistry( + { + name: "queued-account-switch", + runtimePolicyOverride: { + cwd, + approvalPolicy: "never", + sandboxPolicy: { + type: "readOnly", + access: { type: "fullAccess" }, + networkAccess: false, + }, + }, + }, + registryLayer, + ), + ), + ); + assert.deepEqual( + projection.runs.map((run) => [run.providerInstanceId, run.status]), + [ + [CODEX_MODEL_SELECTION.instanceId, "completed"], + [alternateSelection.instanceId, "completed"], + ], + ); + assert.lengthOf(projection.contextHandoffs, 0); + assert.deepEqual( + (yield* Ref.get(capturedTurns)).map((turn) => turn.text), + ["First account turn", "Alternate account turn"], + ); + }), + ), + ); + it.live("finishes earlier queued Codex turns before handing context to queued Claude", () => Effect.scoped( Effect.gen(function* () { From 3c74ce1c3a5bcde6b4c389395a782e633f2c2647 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 17 Sep 2026 03:25:26 -0700 Subject: [PATCH 04/12] fix(queue): carry steered provider context into queued turns --- .../src/state/use-thread-outbox-drain.ts | 23 -- .../src/orchestration-v2/Orchestrator.ts | 45 ++-- .../ProviderSwitch.integration.test.ts | 215 ++++++++++++++++++ apps/web/src/components/ChatView.tsx | 2 +- 4 files changed, 239 insertions(+), 46 deletions(-) diff --git a/apps/mobile/src/state/use-thread-outbox-drain.ts b/apps/mobile/src/state/use-thread-outbox-drain.ts index 007077fc6c92..cbafb40918e1 100644 --- a/apps/mobile/src/state/use-thread-outbox-drain.ts +++ b/apps/mobile/src/state/use-thread-outbox-drain.ts @@ -47,7 +47,6 @@ import { import { removeThreadOutboxMessage } from "./thread-outbox-removal"; import { isQueuedThreadCreationSendable, - modelSelectionsEqual, resolveThreadOutboxDeliveryAction, resolveThreadOutboxDispatchStep, resolveThreadOutboxFailureAction, @@ -546,9 +545,6 @@ async function preserveUploadedAttachmentsForEditor( export function useThreadOutboxDrain(): void { const startTurn = useAtomCommand(threadEnvironment.startTurn, { reportFailure: false }); - const updateThreadMetadata = useAtomCommand(threadEnvironment.updateMetadata, { - reportFailure: false, - }); const setThreadRuntimeMode = useAtomCommand(threadEnvironment.setRuntimeMode, { reportFailure: false, }); @@ -705,24 +701,6 @@ export function useThreadOutboxDrain(): void { } const { reportFailure } = makeDeliveryHelpers(queuedMessage); - if ( - queuedMessage.dispatchMode !== "queue" && - !modelSelectionsEqual(settings.modelSelection, thread.modelSelection) - ) { - const updateResult = await updateThreadMetadata({ - environmentId: queuedMessage.environmentId, - input: { - commandId: settingsCommandId(queuedMessage, "model-selection"), - threadId: queuedMessage.threadId, - modelSelection: settings.modelSelection, - }, - }); - if (AsyncResult.isFailure(updateResult)) { - reportFailure(updateResult, "settings-sync"); - return false; - } - } - if (settings.runtimeMode !== thread.runtimeMode) { const runtimeResult = await setThreadRuntimeMode({ environmentId: queuedMessage.environmentId, @@ -858,7 +836,6 @@ export function useThreadOutboxDrain(): void { setThreadInteractionMode, setThreadRuntimeMode, startTurn, - updateThreadMetadata, restoreQueuedMessage, ], ); diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 6ef763b8608b..7171cfe8e7c3 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -575,6 +575,15 @@ function rootProviderThreadsForProvider( ); } +function lastCompletedRunForProviderThread( + projection: OrchestrationV2ThreadProjection, + providerThreadId: OrchestrationV2ProviderThread["id"], +): OrchestrationV2Run | undefined { + return projection.runs.findLast( + (run) => run.status === "completed" && run.providerThreadId === providerThreadId, + ); +} + const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(function* () { const checkpointService = yield* CheckpointServiceV2; const commandPolicy = yield* CommandPolicyV2; @@ -988,6 +997,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ), ); const latestCompletedRun = projection.runs.findLast((run) => run.status === "completed"); + const targetLastCompletedRun = lastCompletedRunForProviderThread( + projection, + queuedProviderThread.id, + ); const coveredRuns = canResumeAcrossInstances || latestCompletedRun === undefined || @@ -996,7 +1009,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio : projection.runs.filter( (run) => run.status === "completed" && - run.ordinal > (queuedProviderThread.lastRunOrdinal ?? 0) && + run.ordinal > (targetLastCompletedRun?.ordinal ?? 0) && run.ordinal <= latestCompletedRun.ordinal, ); const needsFullContext = deliveryProviderThread.nativeThreadRef === null; @@ -1244,16 +1257,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio targetThreadId: threadId, sourcePoint: contextSourcePointForRun(projection, latestCompletedRun), basePoint: - needsFullContext || queuedProviderThread.lastRunOrdinal === null + needsFullContext || targetLastCompletedRun === undefined ? null - : (() => { - const baseRun = projection.runs.find( - (run) => run.ordinal === queuedProviderThread.lastRunOrdinal, - ); - return baseRun === undefined - ? null - : contextSourcePointForRun(projection, baseRun); - })(), + : contextSourcePointForRun(projection, targetLastCompletedRun), sourceProviderInstanceId: latestCompletedRun.providerInstanceId, targetProviderInstanceId: queuedRun.providerInstanceId, targetRunId: queuedRun.id, @@ -4676,6 +4682,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ); const requiresFullProviderSwitchContext = isProviderSwitch && pendingMergeBackTransfer !== undefined; + const targetLastCompletedRun = + targetProviderThread === undefined + ? undefined + : lastCompletedRunForProviderThread(projection, targetProviderThread.id); const providerSwitchCoveredRuns = !isProviderSwitch || canResumeAcrossInstances || latestCompletedRun === undefined ? [] @@ -4685,7 +4695,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio run.ordinal > (requiresFullProviderSwitchContext ? 0 - : (targetProviderThread?.lastRunOrdinal ?? 0)) && + : (targetLastCompletedRun?.ordinal ?? 0)) && run.ordinal <= latestCompletedRun.ordinal, ); const providerSwitchItems = @@ -5174,18 +5184,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio targetThreadId: command.threadId, sourcePoint: contextSourcePointForRun(projection, latestCompletedRun), basePoint: - requiresFullProviderSwitchContext || - targetProviderThread?.lastRunOrdinal === null || - targetProviderThread?.lastRunOrdinal === undefined + requiresFullProviderSwitchContext || targetLastCompletedRun === undefined ? null - : (() => { - const baseRun = projection.runs.find( - (run) => run.ordinal === targetProviderThread.lastRunOrdinal, - ); - return baseRun === undefined - ? null - : contextSourcePointForRun(projection, baseRun); - })(), + : contextSourcePointForRun(projection, targetLastCompletedRun), sourceProviderInstanceId: latestCompletedRun.providerInstanceId, targetProviderInstanceId: modelSelection.instanceId, targetRunId: runId, diff --git a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts index faa225550985..943299982b54 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -19,6 +19,7 @@ import { import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as PubSub from "effect/PubSub"; import * as Ref from "effect/Ref"; @@ -28,6 +29,7 @@ import * as SqlClient from "effect/unstable/sql/SqlClient"; import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts"; import { ClaudeProviderCapabilitiesV2 } from "../Adapters/ClaudeAdapterV2.ts"; import { CodexProviderCapabilitiesV2 } from "../Adapters/CodexAdapterV2.ts"; +import { AcpProviderCapabilitiesV2 } from "../Adapters/AcpAdapterV2.ts"; import { CursorProviderCapabilitiesV2 } from "../Adapters/CursorAdapterV2.ts"; import { layer as eventSinkLayer } from "../EventSink.ts"; import { EventSinkV2 } from "../EventSink.ts"; @@ -58,6 +60,7 @@ import { CLAUDE_MODEL_SELECTION, CODEX_MODEL_SELECTION, CURSOR_MODEL_SELECTION, + GROK_MODEL_SELECTION, } from "./fixtures/shared.ts"; import { makeOrchestratorV2ReplayLayerWithRegistry } from "./ProviderReplayHarness.ts"; import { checkpointWorkspace } from "./ReplayFixtureWorkspace.ts"; @@ -70,6 +73,7 @@ const returnPrompt = "Respond with exactly: codex after return"; const CODEX_DRIVER = ProviderDriverKind.make("codex"); const CLAUDE_DRIVER = ProviderDriverKind.make("claudeAgent"); const CURSOR_DRIVER = ProviderDriverKind.make("cursor"); +const GROK_DRIVER = ProviderDriverKind.make("acp"); interface CapturedTurn { readonly driver: ProviderDriverKind; @@ -309,6 +313,217 @@ const waitForIdle = Effect.fn("ProviderSwitchTest.waitForIdle")(function* ( }); describe("orchestration v2 provider switching", () => { + it.live("hands completed Grok steering context to earlier queued Codex and later Claude", () => + Effect.scoped( + Effect.gen(function* () { + const cwd = yield* checkpointWorkspace("queued-steer-provider-switch"); + const capturedTurns = yield* Ref.make>([]); + const started = yield* Deferred.make(); + const registryLayer = makeProviderAdapterRegistryLayer([ + makeTestAdapter({ + instanceId: CODEX_MODEL_SELECTION.instanceId, + driver: CODEX_DRIVER, + capabilities: CodexProviderCapabilitiesV2, + modelSelection: CODEX_MODEL_SELECTION, + responseByRunOrdinal: { 2: "Codex queued response" }, + capturedTurns, + holdFirstTurn: started, + }), + makeTestAdapter({ + instanceId: CLAUDE_MODEL_SELECTION.instanceId, + driver: CLAUDE_DRIVER, + capabilities: ClaudeProviderCapabilitiesV2, + modelSelection: CLAUDE_MODEL_SELECTION, + responseByRunOrdinal: { 3: "Claude queued response" }, + capturedTurns, + }), + makeTestAdapter({ + instanceId: GROK_MODEL_SELECTION.instanceId, + driver: GROK_DRIVER, + capabilities: AcpProviderCapabilitiesV2, + modelSelection: GROK_MODEL_SELECTION, + responseByRunOrdinal: { 1: "Grok steered response" }, + capturedTurns, + }), + ]); + const queuedThreadId = ThreadId.make("thread:queued-steer-provider-switch"); + const projection = yield* Effect.gen(function* () { + const orchestrator = yield* OrchestratorV2; + const worker = yield* OrchestrationEffectWorkerV2; + const eventSink = yield* EventSinkV2; + const dispatch = ( + key: string, + modelSelection: ModelSelection, + dispatchMode: Extract< + OrchestrationV2Command, + { readonly type: "message.dispatch" } + >["dispatchMode"], + ) => + orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`command:queued-steer-provider-switch:${key}`), + threadId: queuedThreadId, + messageId: MessageId.make(`message:queued-steer-provider-switch:${key}`), + text: `Prompt ${key}`, + attachments: [], + modelSelection, + dispatchMode, + }); + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("command:queued-steer-provider-switch:create"), + threadId: queuedThreadId, + projectId: ProjectId.make("project:queued-steer-provider-switch"), + title: "Queued steer provider switch", + modelSelection: CODEX_MODEL_SELECTION, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: cwd, + }); + yield* dispatch("first", CODEX_MODEL_SELECTION, { type: "start_immediately" }); + yield* Deferred.await(started); + yield* worker.drain(); + yield* dispatch("codex-queued", CODEX_MODEL_SELECTION, { type: "queue_after_active" }); + yield* dispatch("claude-queued", CLAUDE_MODEL_SELECTION, { type: "queue_after_active" }); + const beforeSteer = yield* orchestrator.getThreadProjection(queuedThreadId); + assert.deepEqual( + beforeSteer.runs.map((run) => run.status), + ["running", "queued", "queued"], + ); + assert.equal(beforeSteer.thread.providerInstanceId, CODEX_MODEL_SELECTION.instanceId); + assert.lengthOf(beforeSteer.contextHandoffs, 0); + const now = yield* DateTime.now; + yield* eventSink.write({ + events: [ + { + id: EventId.make("event:queued-steer-provider-switch:first-turn-running"), + type: "provider-turn.updated", + threadId: queuedThreadId, + runId: beforeSteer.runs[0]!.id, + nodeId: beforeSteer.runs[0]!.rootNodeId!, + driver: CODEX_DRIVER, + providerInstanceId: CODEX_MODEL_SELECTION.instanceId, + occurredAt: now, + payload: { + id: ProviderTurnId.make("provider-turn:queued-steer-provider-switch:first"), + providerThreadId: beforeSteer.runs[0]!.providerThreadId!, + nodeId: beforeSteer.runs[0]!.rootNodeId!, + runAttemptId: beforeSteer.runs[0]!.activeAttemptId!, + nativeTurnRef: null, + ordinal: 1, + status: "running", + startedAt: now, + completedAt: null, + }, + }, + ], + }); + const queuedClaudeCompleted = yield* orchestrator.streamStoredEvents.pipe( + Stream.filter( + (event) => + event.event.type === "run.updated" && + event.event.runId === beforeSteer.runs[2]?.id && + event.event.payload.status === "completed", + ), + Stream.runHead, + Effect.forkScoped, + ); + yield* dispatch("grok-steer", GROK_MODEL_SELECTION, { + type: "steer_active", + targetRunId: beforeSteer.runs[0]!.id, + }); + const afterSteer = yield* orchestrator.getThreadProjection(queuedThreadId); + const interruptedTurn = afterSteer.providerTurns.find( + (turn) => turn.runAttemptId === beforeSteer.runs[0]?.activeAttemptId, + )!; + const interruptedAttempt = afterSteer.attempts.find( + (attempt) => attempt.id === beforeSteer.runs[0]?.activeAttemptId, + )!; + const interruptedAt = yield* DateTime.now; + yield* eventSink.write({ + events: [ + { + id: EventId.make("event:queued-steer-provider-switch:first-turn-interrupted"), + type: "provider-turn.updated", + threadId: queuedThreadId, + runId: beforeSteer.runs[0]!.id, + nodeId: beforeSteer.runs[0]!.rootNodeId!, + driver: CODEX_DRIVER, + providerInstanceId: CODEX_MODEL_SELECTION.instanceId, + occurredAt: interruptedAt, + payload: { + ...interruptedTurn, + status: "interrupted", + completedAt: interruptedAt, + }, + }, + { + id: EventId.make("event:queued-steer-provider-switch:first-attempt-interrupted"), + type: "run-attempt.updated", + threadId: queuedThreadId, + runId: beforeSteer.runs[0]!.id, + nodeId: beforeSteer.runs[0]!.rootNodeId!, + providerInstanceId: CODEX_MODEL_SELECTION.instanceId, + occurredAt: interruptedAt, + payload: { + ...interruptedAttempt, + status: "interrupted", + completedAt: interruptedAt, + }, + }, + ], + }); + yield* worker.drain(); + yield* Fiber.join(queuedClaudeCompleted); + return yield* orchestrator.getThreadProjection(queuedThreadId); + }).pipe( + Effect.provide( + makeOrchestratorV2ReplayLayerWithRegistry( + { + name: "queued-steer-provider-switch", + runtimePolicyOverride: { + cwd, + approvalPolicy: "never", + sandboxPolicy: { + type: "readOnly", + access: { type: "fullAccess" }, + networkAccess: false, + }, + }, + }, + registryLayer, + ), + ), + ); + assert.deepEqual( + projection.runs.map((run) => [run.providerInstanceId, run.status]), + [ + [GROK_MODEL_SELECTION.instanceId, "completed"], + [CODEX_MODEL_SELECTION.instanceId, "completed"], + [CLAUDE_MODEL_SELECTION.instanceId, "completed"], + ], + ); + const turns = yield* Ref.get(capturedTurns); + assert.deepEqual( + turns.map((turn) => turn.driver), + [CODEX_DRIVER, GROK_DRIVER, CODEX_DRIVER, CLAUDE_DRIVER], + ); + assert.include(turns[2]?.text ?? "", "Grok steered response"); + assert.include(turns[3]?.text ?? "", "Grok steered response"); + assert.include(turns[3]?.text ?? "", "Codex queued response"); + assert.deepEqual( + projection.contextHandoffs.map((handoff) => handoff.targetRunId), + [projection.runs[0]?.id, projection.runs[1]?.id, projection.runs[2]?.id], + ); + }), + ), + ); + it.live("resumes a queued account switch without requiring a portable handoff", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 8f38287af5ea..60de6e61e5f2 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -8817,7 +8817,7 @@ export default function ChatView(props: ChatViewProps) { const settingsResult = await persistThreadSettingsForNextTurn({ threadId: threadIdForSend, createdAt: messageCreatedAt, - ...(ctxSelectedModel && !shouldQueueBehindActiveRun + ...(ctxSelectedModel && phase !== "running" ? { modelSelection: ctxSelectedModelSelection } : {}), ...(localCheckoutBranchMismatch From 305ed6a024da924b4f3ed6b606071a26331f6cf6 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 17 Sep 2026 03:27:26 -0700 Subject: [PATCH 05/12] fix(orchestrator): record selected provider during steering restart --- apps/server/src/orchestration-v2/Orchestrator.ts | 16 ++++++++++++++++ .../testkit/ProviderSwitch.integration.test.ts | 1 + 2 files changed, 17 insertions(+) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 7171cfe8e7c3..630b31729fd6 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -3420,6 +3420,22 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio payload: { ...currentRootNode, status: "interrupted", completedAt: now }, }); } + if (selectionChanged) { + yield* emitEvent({ + type: providerInstanceChanged + ? "thread.provider-switched" + : "thread.model-selection-updated", + threadId: input.command.threadId, + providerInstanceId: input.modelSelection.instanceId, + occurredAt: now, + payload: { + ...input.projection.thread, + providerInstanceId: input.modelSelection.instanceId, + modelSelection: input.modelSelection, + updatedAt: now, + }, + }); + } if (requiresProviderThreadHandoff || requiresProviderSessionRestart) { yield* emitEvent({ type: "provider-thread.updated", diff --git a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts index 943299982b54..7f1038ee80e4 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -438,6 +438,7 @@ describe("orchestration v2 provider switching", () => { targetRunId: beforeSteer.runs[0]!.id, }); const afterSteer = yield* orchestrator.getThreadProjection(queuedThreadId); + assert.deepEqual(afterSteer.thread.modelSelection, GROK_MODEL_SELECTION); const interruptedTurn = afterSteer.providerTurns.find( (turn) => turn.runAttemptId === beforeSteer.runs[0]?.activeAttemptId, )!; From 5b53c10e641e980853890e46d05579a4154e55f1 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 17 Sep 2026 03:33:33 -0700 Subject: [PATCH 06/12] fix(orchestrator): record model selection when dispatching --- .../src/orchestration-v2/Orchestrator.ts | 23 +++++++++++++++++++ .../ProviderSwitch.integration.test.ts | 8 +++++++ apps/web/src/components/ChatView.tsx | 7 ------ 3 files changed, 31 insertions(+), 7 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 630b31729fd6..eddc2b352ae5 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -4133,6 +4133,29 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio (transfer) => transfer.id !== pendingMergeBackTransfer?.id, ); const now = yield* DateTime.now; + if ( + !modelSelectionsEqual(projection.thread.modelSelection, modelSelection) || + projection.thread.providerInstanceId !== modelSelection.instanceId + ) { + yield* emit( + events, + command, + )({ + type: + projection.thread.providerInstanceId === modelSelection.instanceId + ? "thread.model-selection-updated" + : "thread.provider-switched", + threadId: command.threadId, + providerInstanceId: modelSelection.instanceId, + occurredAt: now, + payload: { + ...projection.thread, + providerInstanceId: modelSelection.instanceId, + modelSelection, + updatedAt: now, + }, + }); + } const ordinal = nextRunOrdinal(projection); const runId = idAllocator.derive.run({ threadId: command.threadId, ordinal }); const latestCompletedRun = projection.runs.findLast((run) => run.status === "completed"); diff --git a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts index 7f1038ee80e4..636f8cb83ca1 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -1193,8 +1193,16 @@ describe("orchestration v2 provider switching", () => { yield* orchestrator.dispatch(commands[1]!); yield* waitForIdle(threadId); yield* orchestrator.dispatch(commands[2]!); + assert.deepEqual( + (yield* orchestrator.getThreadProjection(threadId)).thread.modelSelection, + CLAUDE_MODEL_SELECTION, + ); yield* waitForIdle(threadId); yield* orchestrator.dispatch(commands[3]!); + assert.deepEqual( + (yield* orchestrator.getThreadProjection(threadId)).thread.modelSelection, + CODEX_MODEL_SELECTION, + ); return yield* waitForIdle(threadId); }).pipe( Effect.provide( diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 60de6e61e5f2..17c712ed3d04 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -5659,7 +5659,6 @@ export default function ChatView(props: ChatViewProps) { async (input: { threadId: ThreadId; createdAt: string; - modelSelection?: ModelSelection; branch?: string; runtimeMode: RuntimeMode; interactionMode: ProviderInteractionMode; @@ -5671,7 +5670,6 @@ export default function ChatView(props: ChatViewProps) { let result: AtomCommandResult = AsyncResult.success(undefined); const metadataUpdate = resolveThreadMetadataUpdateForNextTurn({ currentModelSelection: serverThread.modelSelection, - ...(input.modelSelection ? { nextModelSelection: input.modelSelection } : {}), currentBranch: serverThread.branch, ...(input.branch ? { nextBranch: input.branch } : {}), }); @@ -7735,7 +7733,6 @@ export default function ChatView(props: ChatViewProps) { const settingsResult = await persistThreadSettingsForNextTurn({ threadId, createdAt, - modelSelection: context.selectedModelSelection, ...(localCheckoutBranchMismatch ? { branch: localCheckoutBranchMismatch.currentBranch } : {}), @@ -8817,9 +8814,6 @@ export default function ChatView(props: ChatViewProps) { const settingsResult = await persistThreadSettingsForNextTurn({ threadId: threadIdForSend, createdAt: messageCreatedAt, - ...(ctxSelectedModel && phase !== "running" - ? { modelSelection: ctxSelectedModelSelection } - : {}), ...(localCheckoutBranchMismatch ? { branch: localCheckoutBranchMismatch.currentBranch } : {}), @@ -9307,7 +9301,6 @@ export default function ChatView(props: ChatViewProps) { const settingsResult = await persistThreadSettingsForNextTurn({ threadId: threadIdForSend, createdAt: messageCreatedAt, - modelSelection: ctxSelectedModelSelection, ...(localCheckoutBranchMismatch ? { branch: localCheckoutBranchMismatch.currentBranch } : {}), runtimeMode, interactionMode: nextInteractionMode, From 2645797fbe42b6f73b4dadddcf2cd7ba44b8b36e Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 17 Sep 2026 03:36:22 -0700 Subject: [PATCH 07/12] fix(queue): validate the selected provider capability --- .../src/orchestration-v2/Orchestrator.ts | 37 ++--- .../ProviderSwitch.integration.test.ts | 127 +++++++++++++++++- 2 files changed, 145 insertions(+), 19 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index eddc2b352ae5..5e854c662627 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -3872,23 +3872,6 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio cause: `Active run ${activeRun.id} has no provider thread for queued dispatch.`, }); } - const existingProviderSession = - queueProviderThread.providerSessionId === null - ? undefined - : projection.providerSessions.find( - (candidate) => candidate.id === queueProviderThread.providerSessionId, - ); - if (existingProviderSession !== undefined) { - yield* enforceCommandPolicy(command)( - commandPolicy.ensureQueuedMessages({ - commandId: command.commandId, - threadId: command.threadId, - providerInstanceId: modelSelection.instanceId, - capabilities: existingProviderSession.capabilities, - }), - ); - } - const now = yield* DateTime.now; const ordinal = nextRunOrdinal(projection); const runId = idAllocator.derive.run({ threadId: command.threadId, ordinal }); @@ -3899,6 +3882,24 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const queuedAdapter = yield* providerAdapters .get(modelSelection.instanceId) .pipe(mapDispatchError(command)); + const selectedProviderSession = + targetProviderThread?.providerSessionId === null || + targetProviderThread?.providerSessionId === undefined + ? undefined + : projection.providerSessions.find( + (candidate) => candidate.id === targetProviderThread.providerSessionId, + ); + const queuedCapabilities = + selectedProviderSession?.capabilities ?? + (yield* queuedAdapter.getCapabilities().pipe(mapDispatchError(command))); + yield* enforceCommandPolicy(command)( + commandPolicy.ensureQueuedMessages({ + commandId: command.commandId, + threadId: command.threadId, + providerInstanceId: modelSelection.instanceId, + capabilities: queuedCapabilities, + }), + ); const queuedProviderThread: OrchestrationV2ProviderThread = targetProviderThread ?? { id: idAllocator.derive.providerThread({ driver: queuedAdapter.driver, @@ -3933,7 +3934,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio providerThreadId: queuedProviderThread.id, cwd: resolvedRuntimePolicy.cwd ?? - existingProviderSession?.cwd ?? + selectedProviderSession?.cwd ?? projection.thread.worktreePath ?? process.cwd(), createdAt: now, diff --git a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts index 636f8cb83ca1..d6f2f336efb3 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -27,6 +27,7 @@ import * as Stream from "effect/Stream"; import * as SqlClient from "effect/unstable/sql/SqlClient"; import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts"; +import { CommandPolicyCapabilityUnsupportedError } from "../CommandPolicy.ts"; import { ClaudeProviderCapabilitiesV2 } from "../Adapters/ClaudeAdapterV2.ts"; import { CodexProviderCapabilitiesV2 } from "../Adapters/CodexAdapterV2.ts"; import { AcpProviderCapabilitiesV2 } from "../Adapters/AcpAdapterV2.ts"; @@ -38,7 +39,7 @@ import { LegacyV1ThreadImporter, layer as legacyV1ThreadImporterLayer, } from "../LegacyV1ThreadImporter.ts"; -import { OrchestratorV2 } from "../Orchestrator.ts"; +import { OrchestratorDispatchError, OrchestratorV2 } from "../Orchestrator.ts"; import { OrchestrationEffectWorkerV2 } from "../EffectWorker.ts"; import { ProjectionMaintenanceV2, @@ -313,6 +314,130 @@ const waitForIdle = Effect.fn("ProviderSwitchTest.waitForIdle")(function* ( }); describe("orchestration v2 provider switching", () => { + it.live("checks the queued provider's capability while the current provider stays running", () => + Effect.scoped( + Effect.gen(function* () { + for (const scenario of [ + { activeSupportsQueue: false, selectedSupportsQueue: true }, + { activeSupportsQueue: true, selectedSupportsQueue: false }, + ]) { + const key = `active-${scenario.activeSupportsQueue}-selected-${scenario.selectedSupportsQueue}`; + const cwd = yield* checkpointWorkspace(`queued-capability-${key}`); + const capturedTurns = yield* Ref.make>([]); + const started = yield* Deferred.make(); + const scenarioThreadId = ThreadId.make(`thread:queued-capability:${key}`); + const registryLayer = makeProviderAdapterRegistryLayer([ + makeTestAdapter({ + instanceId: CODEX_MODEL_SELECTION.instanceId, + driver: CODEX_DRIVER, + capabilities: { + ...CodexProviderCapabilitiesV2, + turns: { + ...CodexProviderCapabilitiesV2.turns, + supportsQueuedMessages: scenario.activeSupportsQueue, + }, + }, + modelSelection: CODEX_MODEL_SELECTION, + responseByRunOrdinal: {}, + capturedTurns, + holdFirstTurn: started, + }), + makeTestAdapter({ + instanceId: CLAUDE_MODEL_SELECTION.instanceId, + driver: CLAUDE_DRIVER, + capabilities: { + ...ClaudeProviderCapabilitiesV2, + turns: { + ...ClaudeProviderCapabilitiesV2.turns, + supportsQueuedMessages: scenario.selectedSupportsQueue, + }, + }, + modelSelection: CLAUDE_MODEL_SELECTION, + responseByRunOrdinal: {}, + capturedTurns, + }), + ]); + yield* Effect.gen(function* () { + const orchestrator = yield* OrchestratorV2; + const worker = yield* OrchestrationEffectWorkerV2; + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`command:queued-capability:create:${key}`), + threadId: scenarioThreadId, + projectId: ProjectId.make(`project:queued-capability:${key}`), + title: "Queued capability", + modelSelection: CODEX_MODEL_SELECTION, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: cwd, + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`command:queued-capability:first:${key}`), + threadId: scenarioThreadId, + messageId: MessageId.make(`message:queued-capability:first:${key}`), + text: "Current Codex turn", + attachments: [], + modelSelection: CODEX_MODEL_SELECTION, + dispatchMode: { type: "start_immediately" }, + }); + yield* Deferred.await(started); + yield* worker.drain(); + const queue = orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`command:queued-capability:claude:${key}`), + threadId: scenarioThreadId, + messageId: MessageId.make(`message:queued-capability:claude:${key}`), + text: "Queued Claude turn", + attachments: [], + modelSelection: CLAUDE_MODEL_SELECTION, + dispatchMode: { type: "queue_after_active" }, + }); + if (scenario.selectedSupportsQueue) { + yield* queue; + } else { + const error = yield* queue.pipe(Effect.flip); + assert.instanceOf(error, OrchestratorDispatchError); + assert.instanceOf(error.cause, CommandPolicyCapabilityUnsupportedError); + assert.equal(error.cause.capability, "queued_messages"); + } + const projection = yield* orchestrator.getThreadProjection(scenarioThreadId); + assert.deepEqual( + projection.runs.map((run) => run.status), + scenario.selectedSupportsQueue ? ["running", "queued"] : ["running"], + ); + assert.deepEqual(projection.thread.modelSelection, CODEX_MODEL_SELECTION); + }).pipe( + Effect.provide( + makeOrchestratorV2ReplayLayerWithRegistry( + { + name: `queued-capability-${key}`, + runtimePolicyOverride: { + cwd, + approvalPolicy: "never", + sandboxPolicy: { + type: "readOnly", + access: { type: "fullAccess" }, + networkAccess: false, + }, + }, + }, + registryLayer, + ), + ), + ); + } + }), + ), + ); + it.live("hands completed Grok steering context to earlier queued Codex and later Claude", () => Effect.scoped( Effect.gen(function* () { From ff8e024daa73bb6c7aacf20d42bf9da3fc995086 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 17 Sep 2026 11:18:22 -0700 Subject: [PATCH 08/12] fix(server): restore imported history for queued provider takeover --- .../src/orchestration-v2/Orchestrator.ts | 50 +++- .../ProviderSwitch.integration.test.ts | 228 +++++++++++------- 2 files changed, 180 insertions(+), 98 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 5e854c662627..7205b0d3bccc 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -1013,6 +1013,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio run.ordinal <= latestCompletedRun.ordinal, ); const needsFullContext = deliveryProviderThread.nativeThreadRef === null; + const legacyImportItems = + projection.thread.historyOrigin === "v1_import" + ? projection.turnItems.filter((item) => item.runId === null) + : []; const handoffStrategy = needsFullContext ? ("full_thread_summary" as const) : ("delta_since_target_last_seen" as const); @@ -1079,9 +1083,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }, strategy: handoffStrategy, items: [ - ...(needsFullContext && projection.thread.historyOrigin === "v1_import" - ? projection.turnItems.filter((item) => item.runId === null) - : []), + ...(needsFullContext ? legacyImportItems : []), ...projection.turnItems.filter( (item) => item.runId !== null && coveredRuns.some((run) => run.id === item.runId), @@ -1099,6 +1101,29 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }), ), ); + const legacyImportRecoveryHandoff = + latestCompletedRun === undefined && needsFullContext && legacyImportItems.length > 0 + ? yield* contextHandoffService + .prepareLegacyImport({ + threadId, + targetRunId: queuedRun.id, + toProviderThreadId: queuedProviderThread.id, + toProviderInstanceId: queuedRun.providerInstanceId, + items: legacyImportItems, + createdAt: now, + }) + .pipe( + Effect.mapError( + (cause) => + new OrchestratorDispatchError({ + commandId, + commandType: "message.dispatch", + cause, + }), + ), + ) + : null; + const activeHandoff = handoff ?? legacyImportRecoveryHandoff; const checkpointScope = storedCheckpointScope ?? (yield* runtimePolicy @@ -1152,7 +1177,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio status: "not_loaded", firstRunOrdinal: queuedProviderThread.firstRunOrdinal ?? queuedRun.ordinal, lastRunOrdinal: queuedRun.ordinal, - handoffIds: appendContextHandoffId(queuedProviderThread.handoffIds, handoff?.id ?? null), + handoffIds: appendContextHandoffId( + queuedProviderThread.handoffIds, + activeHandoff?.id ?? null, + ), updatedAt: now, }; const startingRun: OrchestrationV2Run = { @@ -1160,7 +1188,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio status: "starting", queuePosition: null, startedAt: null, - contextHandoffId: handoff?.id ?? null, + contextHandoffId: activeHandoff?.id ?? null, }; const userTurnItem: OrchestrationV2TurnItem = { ...(legacyQueuedTurnItem ?? { @@ -1286,6 +1314,18 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio payload: handoff, }, ]), + ...(legacyImportRecoveryHandoff === null + ? [] + : [ + { + type: "context-handoff.updated" as const, + threadId, + runId: queuedRun.id, + providerInstanceId: queuedRun.providerInstanceId, + occurredAt: now, + payload: legacyImportRecoveryHandoff, + }, + ]), ...sessionsToDetach.map((session) => ({ type: "provider-session.detached" as const, threadId, diff --git a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts index d6f2f336efb3..a81778fdff07 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -98,6 +98,7 @@ function makeTestAdapter(input: { readonly failResume?: boolean; readonly failedRunOrdinals?: ReadonlySet; readonly holdFirstTurn?: Deferred.Deferred; + readonly releaseFirstTurn?: Deferred.Deferred; }): ProviderAdapterV2Shape { return { instanceId: input.instanceId, @@ -171,7 +172,8 @@ function makeTestAdapter(input: { ]); if (turnInput.runOrdinal === 1 && input.holdFirstTurn !== undefined) { yield* Deferred.succeed(input.holdFirstTurn, undefined); - return; + if (input.releaseFirstTurn === undefined) return; + yield* Deferred.await(input.releaseFirstTurn); } const eventTime = yield* DateTime.now; const providerTurnId = ProviderTurnId.make( @@ -1007,7 +1009,7 @@ describe("orchestration v2 provider switching", () => { ), ); - it.live("reissues imported v1 context when switching after the first provider fails", () => + const importedFailureScenario = (queueBeforeFailure: boolean) => Effect.scoped( Effect.gen(function* () { const importedThreadId = ThreadId.make("thread:provider-switch:legacy-import"); @@ -1016,6 +1018,8 @@ describe("orchestration v2 provider switching", () => { const recoveryPrompt = "What was the imported release marker?"; const cwd = yield* checkpointWorkspace("provider-switch-legacy-import"); const capturedTurns = yield* Ref.make>([]); + const firstTurnStarted = yield* Deferred.make(); + const releaseFirstTurn = yield* Deferred.make(); const registryLayer = makeProviderAdapterRegistryLayer([ makeTestAdapter({ instanceId: ProviderInstanceId.make("codex"), @@ -1025,6 +1029,7 @@ describe("orchestration v2 provider switching", () => { responseByRunOrdinal: {}, capturedTurns, failedRunOrdinals: new Set([1]), + ...(queueBeforeFailure ? { holdFirstTurn: firstTurnStarted, releaseFirstTurn } : {}), }), makeTestAdapter({ instanceId: ProviderInstanceId.make("claudeAgent"), @@ -1078,98 +1083,99 @@ describe("orchestration v2 provider switching", () => { const importer = yield* LegacyV1ThreadImporter; const maintenance = yield* ProjectionMaintenanceV2; const orchestrator = yield* OrchestratorV2; + const worker = yield* OrchestrationEffectWorkerV2; yield* sql` - INSERT INTO projection_projects ( - project_id, - title, - workspace_root, - default_model_selection_json, - scripts_json, - created_at, - updated_at, - deleted_at - ) VALUES ( - ${importedProjectId}, - 'Imported provider switch project', - ${cwd}, - '{"instanceId":"codex","model":"gpt-5.4"}', - '[]', - '2026-01-01T00:00:00.000Z', - '2026-01-01T00:00:00.000Z', - NULL - ) - `; + INSERT INTO projection_projects ( + project_id, + title, + workspace_root, + default_model_selection_json, + scripts_json, + created_at, + updated_at, + deleted_at + ) VALUES ( + ${importedProjectId}, + 'Imported provider switch project', + ${cwd}, + '{"instanceId":"codex","model":"gpt-5.4"}', + '[]', + '2026-01-01T00:00:00.000Z', + '2026-01-01T00:00:00.000Z', + NULL + ) + `; yield* sql` - INSERT INTO projection_threads ( - thread_id, - project_id, - title, - model_selection_json, - runtime_mode, - interaction_mode, - branch, - worktree_path, - latest_turn_id, - created_at, - updated_at, - archived_at, - settled_override, - settled_at, - deleted_at - ) VALUES ( - ${importedThreadId}, - ${importedProjectId}, - 'Imported provider switch thread', - '{"instanceId":"codex","model":"gpt-5.4"}', - 'full-access', - 'default', - 'main', - ${cwd}, - NULL, - '2026-01-01T00:00:00.000Z', - '2026-01-01T00:00:00.000Z', - NULL, - NULL, - NULL, - NULL - ) - `; + INSERT INTO projection_threads ( + thread_id, + project_id, + title, + model_selection_json, + runtime_mode, + interaction_mode, + branch, + worktree_path, + latest_turn_id, + created_at, + updated_at, + archived_at, + settled_override, + settled_at, + deleted_at + ) VALUES ( + ${importedThreadId}, + ${importedProjectId}, + 'Imported provider switch thread', + '{"instanceId":"codex","model":"gpt-5.4"}', + 'full-access', + 'default', + 'main', + ${cwd}, + NULL, + '2026-01-01T00:00:00.000Z', + '2026-01-01T00:00:00.000Z', + NULL, + NULL, + NULL, + NULL + ) + `; yield* sql` - INSERT INTO projection_thread_messages ( - message_id, - thread_id, - turn_id, - role, - text, - attachments_json, - is_streaming, - created_at, - updated_at - ) VALUES - ( - 'message:provider-switch:legacy-import:user', - ${importedThreadId}, - NULL, - 'user', - 'Remember that the imported release marker is violet.', - '[]', - 0, - '2026-01-01T01:00:00.000Z', - '2026-01-01T01:00:00.000Z' - ), - ( - 'message:provider-switch:legacy-import:assistant', - ${importedThreadId}, - NULL, - 'assistant', - 'I will remember violet.', - '[]', - 0, - '2026-01-01T01:01:00.000Z', - '2026-01-01T01:01:00.000Z' - ) - `; + INSERT INTO projection_thread_messages ( + message_id, + thread_id, + turn_id, + role, + text, + attachments_json, + is_streaming, + created_at, + updated_at + ) VALUES + ( + 'message:provider-switch:legacy-import:user', + ${importedThreadId}, + NULL, + 'user', + 'Remember that the imported release marker is violet.', + '[]', + 0, + '2026-01-01T01:00:00.000Z', + '2026-01-01T01:00:00.000Z' + ), + ( + 'message:provider-switch:legacy-import:assistant', + ${importedThreadId}, + NULL, + 'assistant', + 'I will remember violet.', + '[]', + 0, + '2026-01-01T01:01:00.000Z', + '2026-01-01T01:01:00.000Z' + ) + `; yield* importer.reconcileShells; yield* maintenance.rebuild; @@ -1187,7 +1193,11 @@ describe("orchestration v2 provider switching", () => { modelSelection: CODEX_MODEL_SELECTION, dispatchMode: { type: "start_immediately" }, }); - yield* waitForIdle(importedThreadId); + if (queueBeforeFailure) { + yield* Deferred.await(firstTurnStarted); + } else { + yield* waitForIdle(importedThreadId); + } yield* orchestrator.dispatch({ type: "message.dispatch", createdBy: "user", @@ -1198,8 +1208,33 @@ describe("orchestration v2 provider switching", () => { text: recoveryPrompt, attachments: [], modelSelection: CLAUDE_MODEL_SELECTION, - dispatchMode: { type: "start_immediately" }, + dispatchMode: { + type: queueBeforeFailure ? "queue_after_active" : "start_immediately", + }, }); + if (queueBeforeFailure) { + const queued = yield* orchestrator.getThreadProjection(importedThreadId); + assert.deepEqual( + queued.runs.map((run) => run.status), + ["running", "queued"], + ); + assert.deepEqual( + queued.contextHandoffs.map((handoff) => handoff.targetRunId), + [queued.runs[0]?.id], + ); + yield* Deferred.succeed(releaseFirstTurn, undefined); + yield* orchestrator.streamStoredEvents.pipe( + Stream.filter( + (event) => + event.event.type === "run.updated" && + event.event.runId === queued.runs[1]?.id && + event.event.payload.status === "completed", + ), + Stream.runHead, + ); + yield* worker.drain(); + return yield* orchestrator.getThreadProjection(importedThreadId); + } return yield* waitForIdle(importedThreadId); }).pipe(Effect.provide(testLayer)); @@ -1229,7 +1264,14 @@ describe("orchestration v2 provider switching", () => { assert.include(turns[1]?.text ?? "", recoveryPrompt); assert.notInclude(turns[1]?.text ?? "", failedPrompt); }), - ), + ); + + it.live("reissues imported v1 context when switching after the first provider fails", () => + importedFailureScenario(false), + ); + it.live( + "reissues imported v1 context when a queued provider starts after the first provider fails", + () => importedFailureScenario(true), ); it.live("uses portable fallback when native resume fails after a provider switch", () => From d8ed6ca6288678accbb1e3574b0b911f0822d5d0 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 17 Sep 2026 11:29:29 -0700 Subject: [PATCH 09/12] fix(queue): advance after queued provider handoff rejection --- .../src/orchestration-v2/Orchestrator.ts | 83 ++++++- .../ProviderSwitch.integration.test.ts | 203 ++++++++++++++++++ 2 files changed, 276 insertions(+), 10 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 7205b0d3bccc..91f7b170b2a1 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -77,6 +77,7 @@ import { import type { ProviderAdapterV2Shape } from "./ProviderAdapter.ts"; import { ProviderAdapterRegistryV2 } from "./ProviderAdapterRegistry.ts"; import { ProviderContinuationRequests } from "./ProviderContinuationRequests.ts"; +import { makeProviderFailure } from "./ProviderFailure.ts"; import { ProviderSessionManagerV2 } from "./ProviderSessionManager.ts"; import { ProviderSwitchServiceV2 } from "./ProviderSwitchService.ts"; import { isAutomaticCompletionRun, queuedRunsInDeliveryOrder } from "./QueuedRunOrder.ts"; @@ -1040,7 +1041,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ), ); if (transferId !== null) { - yield* commandPolicy + const handoffRejection = yield* commandPolicy .ensureContextHandoff({ commandId, threadId, @@ -1049,15 +1050,77 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio strategy: needsFullContext ? "full_thread_summary" : "delta_context", }) .pipe( - Effect.mapError( - (cause) => - new OrchestratorDispatchError({ - commandId, - commandType: "message.dispatch", - cause, - }), - ), + Effect.match({ + onFailure: (cause) => cause, + onSuccess: () => null, + }), ); + if (handoffRejection !== null) { + const failure = makeProviderFailure({ + message: handoffRejection.message, + code: "context_handoff_unsupported", + class: "validation_error", + }); + yield* writeSystemEvents([ + { + type: "run-attempt.updated", + threadId, + runId: queuedRun.id, + nodeId: rootNodeId, + providerInstanceId: queuedRun.providerInstanceId, + occurredAt: now, + payload: { ...attempt, status: "failed", completedAt: now }, + }, + { + type: "node.updated", + threadId, + runId: queuedRun.id, + nodeId: rootNodeId, + providerInstanceId: queuedRun.providerInstanceId, + occurredAt: now, + payload: { ...rootNode, status: "failed", completedAt: now }, + }, + { + type: "turn-item.updated", + threadId, + runId: queuedRun.id, + nodeId: rootNodeId, + providerInstanceId: queuedRun.providerInstanceId, + occurredAt: now, + payload: { + id: idAllocator.derive.turnItemFromProviderItem({ + driver: queuedProviderThread.driver, + nativeItemId: `queued-handoff-failure:${queuedRun.id}`, + }), + threadId, + runId: queuedRun.id, + nodeId: rootNodeId, + providerThreadId: queuedProviderThread.id, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: nextTurnItemOrdinal(projection), + status: "failed", + title: "Queued provider handoff failed", + startedAt: now, + completedAt: now, + updatedAt: now, + type: "error", + failure, + }, + }, + { + type: "run.updated", + threadId, + runId: queuedRun.id, + nodeId: rootNodeId, + providerInstanceId: queuedRun.providerInstanceId, + occurredAt: now, + payload: { ...queuedRun, status: "failed", queuePosition: null, completedAt: now }, + }, + ]); + return; + } } const handoff = transferId === null || latestCompletedRun === undefined @@ -4149,7 +4212,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ...existingItem, runId, nodeId: rootNodeId, - providerThreadId: queueProviderThread.id, + providerThreadId: queuedProviderThread.id, providerTurnId: null, inputIntent: "queued_turn", updatedAt: now, diff --git a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts index a81778fdff07..625848d6e58b 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -1009,6 +1009,209 @@ describe("orchestration v2 provider switching", () => { ), ); + it.live("fails an unsupported queued handoff and advances to the next queued provider", () => + Effect.scoped( + Effect.gen(function* () { + const cwd = yield* checkpointWorkspace("queued-handoff-rejection"); + const capturedTurns = yield* Ref.make>([]); + const started = yield* Deferred.make(); + const rejectedThreadId = ThreadId.make("thread:queued-handoff-rejection"); + const rejectedMessageId = MessageId.make("message:queued-handoff-rejection:claude"); + const unsupportedClaudeCapabilities = { + ...ClaudeProviderCapabilitiesV2, + context: { + ...ClaudeProviderCapabilitiesV2.context, + canConsumeHandoffSummaries: false, + }, + }; + const registryLayer = makeProviderAdapterRegistryLayer([ + makeTestAdapter({ + instanceId: CODEX_MODEL_SELECTION.instanceId, + driver: CODEX_DRIVER, + capabilities: CodexProviderCapabilitiesV2, + modelSelection: CODEX_MODEL_SELECTION, + responseByRunOrdinal: { 3: "Later Codex queued turn complete" }, + capturedTurns, + holdFirstTurn: started, + }), + makeTestAdapter({ + instanceId: CLAUDE_MODEL_SELECTION.instanceId, + driver: CLAUDE_DRIVER, + capabilities: unsupportedClaudeCapabilities, + modelSelection: CLAUDE_MODEL_SELECTION, + responseByRunOrdinal: {}, + capturedTurns, + }), + ]); + const projection = yield* Effect.gen(function* () { + const orchestrator = yield* OrchestratorV2; + const eventSink = yield* EventSinkV2; + const worker = yield* OrchestrationEffectWorkerV2; + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("command:queued-handoff-rejection:create"), + threadId: rejectedThreadId, + projectId: ProjectId.make("project:queued-handoff-rejection"), + title: "Queued handoff rejection", + modelSelection: CODEX_MODEL_SELECTION, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: cwd, + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("command:queued-handoff-rejection:first"), + threadId: rejectedThreadId, + messageId: MessageId.make("message:queued-handoff-rejection:first"), + text: "First Codex turn", + attachments: [], + modelSelection: CODEX_MODEL_SELECTION, + dispatchMode: { type: "start_immediately" }, + }); + yield* Deferred.await(started); + const active = yield* orchestrator.getThreadProjection(rejectedThreadId); + const now = yield* DateTime.now; + yield* eventSink.write({ + events: [ + { + id: EventId.make("event:queued-handoff-rejection:existing-item"), + type: "turn-item.updated", + threadId: rejectedThreadId, + runId: active.runs[0]!.id, + nodeId: active.runs[0]!.rootNodeId!, + providerInstanceId: CODEX_MODEL_SELECTION.instanceId, + occurredAt: now, + payload: { + id: TurnItemId.make("turn-item:queued-handoff-rejection:existing-item"), + threadId: rejectedThreadId, + runId: active.runs[0]!.id, + nodeId: active.runs[0]!.rootNodeId!, + providerThreadId: active.runs[0]!.providerThreadId, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 150, + status: "completed", + title: null, + inputIntent: "turn_start", + startedAt: now, + completedAt: now, + updatedAt: now, + type: "user_message", + messageId: rejectedMessageId, + text: "Unsupported Claude turn", + attachments: [], + createdBy: "user", + creationSource: "web", + }, + }, + ], + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("command:queued-handoff-rejection:claude"), + threadId: rejectedThreadId, + messageId: rejectedMessageId, + text: "Unsupported Claude turn", + attachments: [], + modelSelection: CLAUDE_MODEL_SELECTION, + dispatchMode: { type: "queue_after_active" }, + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("command:queued-handoff-rejection:later"), + threadId: rejectedThreadId, + messageId: MessageId.make("message:queued-handoff-rejection:later"), + text: "Later Codex turn", + attachments: [], + modelSelection: CODEX_MODEL_SELECTION, + dispatchMode: { type: "queue_after_active" }, + }); + const queued = yield* orchestrator.getThreadProjection(rejectedThreadId); + assert.deepEqual( + queued.runs.map((run) => run.status), + ["running", "queued", "queued"], + ); + const queuedItem = queued.turnItems.find( + (item) => item.type === "user_message" && item.messageId === rejectedMessageId, + ); + assert.equal(queuedItem?.runId, queued.runs[1]?.id); + assert.equal(queuedItem?.providerThreadId, queued.runs[1]?.providerThreadId); + yield* eventSink.write({ + events: [ + { + id: EventId.make("event:queued-handoff-rejection:first-complete"), + type: "run.updated", + threadId: rejectedThreadId, + runId: queued.runs[0]!.id, + providerInstanceId: CODEX_MODEL_SELECTION.instanceId, + occurredAt: now, + payload: { ...queued.runs[0]!, status: "completed", completedAt: now }, + }, + ], + }); + yield* orchestrator.resumeQueuedRuns; + yield* orchestrator.streamStoredEvents.pipe( + Stream.filter( + (event) => + event.event.type === "run.updated" && + event.event.runId === queued.runs[2]?.id && + event.event.payload.status === "completed", + ), + Stream.runHead, + ); + yield* worker.drain(); + return yield* orchestrator.getThreadProjection(rejectedThreadId); + }).pipe( + Effect.provide( + makeOrchestratorV2ReplayLayerWithRegistry( + { + name: "queued-handoff-rejection", + runtimePolicyOverride: { + cwd, + approvalPolicy: "never", + sandboxPolicy: { + type: "readOnly", + access: { type: "fullAccess" }, + networkAccess: false, + }, + }, + }, + registryLayer, + ), + ), + ); + assert.deepEqual( + projection.runs.map((run) => run.status), + ["completed", "failed", "completed"], + ); + assert.equal(projection.runs[1]?.queuePosition, null); + assert.equal(projection.thread.providerInstanceId, CODEX_MODEL_SELECTION.instanceId); + assert.lengthOf(projection.contextHandoffs, 0); + const failureItem = projection.turnItems.find( + (item) => item.type === "error" && item.runId === projection.runs[1]?.id, + ); + assert.equal( + failureItem?.type === "error" ? failureItem.failure.code : null, + "context_handoff_unsupported", + ); + assert.deepEqual( + (yield* Ref.get(capturedTurns)).map((turn) => turn.driver), + [CODEX_DRIVER, CODEX_DRIVER], + ); + }), + ), + ); + const importedFailureScenario = (queueBeforeFailure: boolean) => Effect.scoped( Effect.gen(function* () { From 0b45a610064404f5015ee28759eb7e19d5499a46 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 17 Sep 2026 11:36:08 -0700 Subject: [PATCH 10/12] fix(queue): fail queued turns when startup cannot proceed --- .../src/orchestration-v2/Orchestrator.ts | 190 ++++++++++-------- 1 file changed, 109 insertions(+), 81 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 91f7b170b2a1..0a819ea48fc1 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -881,6 +881,107 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio } }); + const failQueuedRunStart = (threadId: ThreadId, cause: unknown) => + Effect.gen(function* () { + const projection = yield* projectionStore.getThreadProjection(threadId); + const queuedRun = nextQueuedRun(projection); + if (queuedRun === undefined) return; + const now = yield* DateTime.now; + const rootNode = projection.nodes.find((node) => node.id === queuedRun.rootNodeId); + const attempt = projection.attempts.find((entry) => entry.id === queuedRun.activeAttemptId); + const providerThread = projection.providerThreads.find( + (entry) => entry.id === queuedRun.providerThreadId, + ); + const handoffUnsupported = + typeof cause === "object" && + cause !== null && + "_tag" in cause && + cause._tag === "CommandPolicyCapabilityUnsupportedError"; + const failureCause = + typeof cause === "object" && + cause !== null && + "_tag" in cause && + cause._tag === "OrchestratorDispatchError" && + "cause" in cause + ? cause.cause + : cause; + const failure = makeProviderFailure({ + cause: failureCause, + code: handoffUnsupported ? "context_handoff_unsupported" : "queued_start_failed", + class: handoffUnsupported ? "validation_error" : "unknown", + }); + yield* writeSystemEvents([ + ...(attempt !== undefined && rootNode !== undefined + ? [ + { + type: "run-attempt.updated" as const, + threadId, + runId: queuedRun.id, + nodeId: rootNode.id, + providerInstanceId: queuedRun.providerInstanceId, + occurredAt: now, + payload: { ...attempt, status: "failed" as const, completedAt: now }, + }, + ] + : []), + ...(rootNode === undefined + ? [] + : [ + { + type: "node.updated" as const, + threadId, + runId: queuedRun.id, + nodeId: rootNode.id, + providerInstanceId: queuedRun.providerInstanceId, + occurredAt: now, + payload: { ...rootNode, status: "failed" as const, completedAt: now }, + }, + ]), + ...(rootNode === undefined || providerThread === undefined + ? [] + : [ + { + type: "turn-item.updated" as const, + threadId, + runId: queuedRun.id, + nodeId: rootNode.id, + providerInstanceId: queuedRun.providerInstanceId, + occurredAt: now, + payload: { + id: idAllocator.derive.turnItemFromProviderItem({ + driver: providerThread.driver, + nativeItemId: `queued-start-failure:${queuedRun.id}`, + }), + threadId, + runId: queuedRun.id, + nodeId: rootNode.id, + providerThreadId: providerThread.id, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: nextTurnItemOrdinal(projection), + status: "failed" as const, + title: "Queued provider could not start", + startedAt: now, + completedAt: now, + updatedAt: now, + type: "error" as const, + failure, + }, + }, + ]), + { + type: "run.updated", + threadId, + runId: queuedRun.id, + ...(rootNode === undefined ? {} : { nodeId: rootNode.id }), + providerInstanceId: queuedRun.providerInstanceId, + occurredAt: now, + payload: { ...queuedRun, status: "failed", queuePosition: null, completedAt: now }, + }, + ]); + }); + const startNextQueuedRun = (threadId: ThreadId) => Effect.gen(function* () { const projection = yield* projectionStore.getThreadProjection(threadId); @@ -1041,86 +1142,13 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ), ); if (transferId !== null) { - const handoffRejection = yield* commandPolicy - .ensureContextHandoff({ - commandId, - threadId, - providerInstanceId: queuedRun.providerInstanceId, - capabilities: targetCapabilities, - strategy: needsFullContext ? "full_thread_summary" : "delta_context", - }) - .pipe( - Effect.match({ - onFailure: (cause) => cause, - onSuccess: () => null, - }), - ); - if (handoffRejection !== null) { - const failure = makeProviderFailure({ - message: handoffRejection.message, - code: "context_handoff_unsupported", - class: "validation_error", - }); - yield* writeSystemEvents([ - { - type: "run-attempt.updated", - threadId, - runId: queuedRun.id, - nodeId: rootNodeId, - providerInstanceId: queuedRun.providerInstanceId, - occurredAt: now, - payload: { ...attempt, status: "failed", completedAt: now }, - }, - { - type: "node.updated", - threadId, - runId: queuedRun.id, - nodeId: rootNodeId, - providerInstanceId: queuedRun.providerInstanceId, - occurredAt: now, - payload: { ...rootNode, status: "failed", completedAt: now }, - }, - { - type: "turn-item.updated", - threadId, - runId: queuedRun.id, - nodeId: rootNodeId, - providerInstanceId: queuedRun.providerInstanceId, - occurredAt: now, - payload: { - id: idAllocator.derive.turnItemFromProviderItem({ - driver: queuedProviderThread.driver, - nativeItemId: `queued-handoff-failure:${queuedRun.id}`, - }), - threadId, - runId: queuedRun.id, - nodeId: rootNodeId, - providerThreadId: queuedProviderThread.id, - providerTurnId: null, - nativeItemRef: null, - parentItemId: null, - ordinal: nextTurnItemOrdinal(projection), - status: "failed", - title: "Queued provider handoff failed", - startedAt: now, - completedAt: now, - updatedAt: now, - type: "error", - failure, - }, - }, - { - type: "run.updated", - threadId, - runId: queuedRun.id, - nodeId: rootNodeId, - providerInstanceId: queuedRun.providerInstanceId, - occurredAt: now, - payload: { ...queuedRun, status: "failed", queuePosition: null, completedAt: now }, - }, - ]); - return; - } + yield* commandPolicy.ensureContextHandoff({ + commandId, + threadId, + providerInstanceId: queuedRun.providerInstanceId, + capabilities: targetCapabilities, + strategy: needsFullContext ? "full_thread_summary" : "delta_context", + }); } const handoff = transferId === null || latestCompletedRun === undefined @@ -1447,7 +1475,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }, ], ); - }); + }).pipe(Effect.catch((cause) => failQueuedRunStart(threadId, cause))); const resumeQueuedRuns = Effect.gen(function* () { const threadIds = yield* projectionStore.getRecoveryThreadIds("queued-runs"); From c9f0de21321e8042b3c859d8484ffc6c054129d3 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 17 Sep 2026 11:44:27 -0700 Subject: [PATCH 11/12] fix(queue): show queued handoff in conversation timeline --- .../src/orchestration-v2/Orchestrator.ts | 54 +++++++++++++++++++ .../ProviderSwitch.integration.test.ts | 36 +++++++++++++ 2 files changed, 90 insertions(+) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 0a819ea48fc1..3362e2465848 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -1310,6 +1310,47 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio completedAt: now, updatedAt: now, }; + const handoffTurnItem: OrchestrationV2TurnItem | null = + activeHandoff === null + ? null + : { + id: idAllocator.derive.runSignalTurnItem({ + runId: queuedRun.id, + signal: `context-handoff:${activeHandoff.id}`, + }), + threadId, + runId: queuedRun.id, + nodeId: rootNodeId, + providerThreadId: queuedProviderThread.id, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: queuedRun.ordinal * 100 - 1, + status: "completed", + title: handoff === null ? "Imported context" : "Provider handoff", + startedAt: now, + completedAt: now, + updatedAt: now, + type: "handoff", + contextHandoffId: activeHandoff.id, + fromProviderThreadIds: activeHandoff.fromProviderThreadIds, + toProviderThreadId: activeHandoff.toProviderThreadId, + fromProviderInstanceIds: Array.from( + new Set(coveredRuns.map((run) => run.providerInstanceId)), + ), + toProviderInstanceId: queuedRun.providerInstanceId, + fromModelSelections: Array.from( + new Map( + coveredRuns.map((run) => [ + `${run.modelSelection.instanceId}\0${run.modelSelection.model}`, + run.modelSelection, + ]), + ).values(), + ), + toModel: queuedRun.modelSelection.model, + strategy: activeHandoff.strategy, + summary: activeHandoff.summaryText, + }; const checkpointEvents: ReadonlyArray> = storedCheckpointScope === undefined ? [ @@ -1417,6 +1458,19 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio payload: legacyImportRecoveryHandoff, }, ]), + ...(handoffTurnItem === null + ? [] + : [ + { + type: "turn-item.updated" as const, + threadId, + runId: queuedRun.id, + nodeId: rootNodeId, + providerInstanceId: queuedRun.providerInstanceId, + occurredAt: now, + payload: handoffTurnItem, + }, + ]), ...sessionsToDetach.map((session) => ({ type: "provider-session.detached" as const, threadId, diff --git a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts index 625848d6e58b..5ea424186852 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -1002,6 +1002,21 @@ describe("orchestration v2 provider switching", () => { ); assert.lengthOf(projection.contextHandoffs, 1); assert.equal(projection.contextHandoffs[0]?.targetRunId, projection.runs[3]?.id); + const handoffItem = projection.turnItems.find( + (item) => item.type === "handoff" && item.runId === projection.runs[3]?.id, + ); + assert.equal( + handoffItem?.type === "handoff" ? handoffItem.contextHandoffId : null, + projection.contextHandoffs[0]?.id, + ); + const queuedUserItem = projection.turnItems.find( + (item) => item.type === "user_message" && item.runId === projection.runs[3]?.id, + ); + assert.isBelow(handoffItem?.ordinal ?? Infinity, queuedUserItem?.ordinal ?? -Infinity); + assert.include( + handoffItem?.type === "handoff" ? handoffItem.summary : "", + "Codex second queued turn complete", + ); assert.include(turns[3]?.text ?? "", "Codex current turn complete"); assert.include(turns[3]?.text ?? "", "Codex first queued turn complete"); assert.include(turns[3]?.text ?? "", "Codex second queued turn complete"); @@ -1195,6 +1210,14 @@ describe("orchestration v2 provider switching", () => { ["completed", "failed", "completed"], ); assert.equal(projection.runs[1]?.queuePosition, null); + assert.equal( + projection.attempts.find((attempt) => attempt.runId === projection.runs[1]?.id)?.status, + "failed", + ); + assert.equal( + projection.nodes.find((node) => node.runId === projection.runs[1]?.id)?.status, + "failed", + ); assert.equal(projection.thread.providerInstanceId, CODEX_MODEL_SELECTION.instanceId); assert.lengthOf(projection.contextHandoffs, 0); const failureItem = projection.turnItems.find( @@ -1461,6 +1484,19 @@ describe("orchestration v2 provider switching", () => { ], ); assert.equal(projection.runs[1]?.contextHandoffId, projection.contextHandoffs[1]?.id); + if (queueBeforeFailure) { + const handoffItem = projection.turnItems.find( + (item) => item.type === "handoff" && item.runId === projection.runs[1]?.id, + ); + assert.equal( + handoffItem?.type === "handoff" ? handoffItem.contextHandoffId : null, + projection.contextHandoffs[1]?.id, + ); + assert.include( + handoffItem?.type === "handoff" ? handoffItem.summary : "", + "imported release marker is violet", + ); + } assert.include(turns[1]?.text ?? "", "Context handoff (manual_context):"); assert.include(turns[1]?.text ?? "", "imported release marker is violet"); assert.include(turns[1]?.text ?? "", "I will remember violet."); From cbbd1ce40a10af2c756430c254435d27bfce7b68 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 17 Sep 2026 11:51:36 -0700 Subject: [PATCH 12/12] test(queue): cover queued provider session detachment --- .../ProviderSwitch.integration.test.ts | 74 ++++++++++++++++--- 1 file changed, 62 insertions(+), 12 deletions(-) diff --git a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts index 5ea424186852..4d2e5a2a5cb3 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -10,6 +10,7 @@ import { type OrchestrationV2ProviderThread, ProjectId, ProviderInstanceId, + ProviderSessionId, ProviderThreadId, ProviderTurnId, ThreadId, @@ -41,6 +42,7 @@ import { } from "../LegacyV1ThreadImporter.ts"; import { OrchestratorDispatchError, OrchestratorV2 } from "../Orchestrator.ts"; import { OrchestrationEffectWorkerV2 } from "../EffectWorker.ts"; +import { EffectOutboxV2, layer as effectOutboxLayer } from "../EffectOutbox.ts"; import { ProjectionMaintenanceV2, layer as projectionMaintenanceLayer, @@ -861,10 +863,13 @@ describe("orchestration v2 provider switching", () => { }), ]); const queuedThreadId = ThreadId.make("thread:queued-provider-switch"); + const databaseLayer = SqlitePersistenceMemory; + const outboxProvided = effectOutboxLayer.pipe(Layer.provide(databaseLayer)); const projection = yield* Effect.gen(function* () { const orchestrator = yield* OrchestratorV2; const worker = yield* OrchestrationEffectWorkerV2; const eventSink = yield* EventSinkV2; + const effectOutbox = yield* EffectOutboxV2; const dispatch = (ordinal: number, modelSelection: ModelSelection) => orchestrator.dispatch({ type: "message.dispatch", @@ -907,9 +912,25 @@ describe("orchestration v2 provider switching", () => { assert.equal(queued.thread.activeProviderThreadId, queued.runs[0]?.providerThreadId); assert.equal(queued.thread.providerInstanceId, CODEX_MODEL_SELECTION.instanceId); assert.lengthOf(queued.contextHandoffs, 0); + const activeSession = queued.providerSessions.find( + (session) => session.providerInstanceId === CODEX_MODEL_SELECTION.instanceId, + ); + assert.isDefined(activeSession); const now = yield* DateTime.now; yield* eventSink.write({ events: [ + ...(["stopped", "error"] as const).map((status) => ({ + id: EventId.make(`event:queued-provider-switch:dead-session:${status}`), + type: "provider-session.updated" as const, + threadId: queuedThreadId, + providerInstanceId: CODEX_MODEL_SELECTION.instanceId, + occurredAt: now, + payload: { + ...activeSession!, + id: ProviderSessionId.make(`provider-session:queued-provider-switch:${status}`), + status, + }, + })), { id: EventId.make("event:queued-provider-switch:first-response"), type: "turn-item.updated", @@ -961,23 +982,52 @@ describe("orchestration v2 provider switching", () => { Stream.runHead, ); yield* worker.drain(); + const detachEvents = yield* eventSink.stream({ threadId: queuedThreadId }).pipe( + Stream.filter((stored) => stored.event.type === "provider-session.detached"), + Stream.take(1), + Stream.runCollect, + ); + assert.equal( + detachEvents[0]?.event.type === "provider-session.detached" + ? detachEvents[0].event.payload.providerSessionId + : null, + activeSession?.id, + ); + const startCommandId = CommandId.make( + `command:system:start-queued:${queued.runs[3]!.id}`, + ); + const detachEffects = (yield* effectOutbox.listByCommandId(startCommandId)).filter( + (effect) => effect.request.type === "provider-session.detach", + ); + assert.deepEqual( + detachEffects.map((effect) => + effect.request.type === "provider-session.detach" + ? effect.request.providerSessionId + : null, + ), + [activeSession?.id], + ); return yield* orchestrator.getThreadProjection(queuedThreadId); }).pipe( Effect.provide( - makeOrchestratorV2ReplayLayerWithRegistry( - { - name: "queued-provider-switch", - runtimePolicyOverride: { - cwd, - approvalPolicy: "never", - sandboxPolicy: { - type: "readOnly", - access: { type: "fullAccess" }, - networkAccess: false, + Layer.merge( + makeOrchestratorV2ReplayLayerWithRegistry( + { + name: "queued-provider-switch", + runtimePolicyOverride: { + cwd, + approvalPolicy: "never", + sandboxPolicy: { + type: "readOnly", + access: { type: "fullAccess" }, + networkAccess: false, + }, }, }, - }, - registryLayer, + registryLayer, + { databaseLayer }, + ), + outboxProvided, ), ), );