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/mobile/src/state/use-thread-outbox-drain.ts b/apps/mobile/src/state/use-thread-outbox-drain.ts index d40608ff041f..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,21 +701,6 @@ export function useThreadOutboxDrain(): void { } const { reportFailure } = makeDeliveryHelpers(queuedMessage); - if (!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, @@ -855,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 8cfd4d097c8f..3362e2465848 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"; @@ -575,6 +576,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; @@ -871,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); @@ -932,6 +1043,178 @@ 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 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) => + 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 targetLastCompletedRun = lastCompletedRunForProviderThread( + projection, + queuedProviderThread.id, + ); + const coveredRuns = + canResumeAcrossInstances || + latestCompletedRun === undefined || + latestCompletedRun.providerInstanceId === queuedRun.providerInstanceId + ? [] + : projection.runs.filter( + (run) => + run.status === "completed" && + run.ordinal > (targetLastCompletedRun?.ordinal ?? 0) && + 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); + 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", + }); + } + 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 ? legacyImportItems : []), + ...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 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 @@ -957,7 +1240,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ), )); const providerSessionId = - queuedProviderThread.providerSessionId ?? + (!canResumeAcrossInstances && + queuedProviderThread.providerSessionId !== null && + !switchPlan?.releaseProviderSessionIds.includes(queuedProviderThread.providerSessionId) + ? queuedProviderThread.providerSessionId + : null) ?? (yield* providerAdapters.get(queuedRun.providerInstanceId).pipe( Effect.flatMap((adapter) => providerSessionIdFor({ @@ -976,11 +1263,15 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ), )); const providerThread: OrchestrationV2ProviderThread = { - ...queuedProviderThread, + ...deliveryProviderThread, providerSessionId, status: "not_loaded", firstRunOrdinal: queuedProviderThread.firstRunOrdinal ?? queuedRun.ordinal, lastRunOrdinal: queuedRun.ordinal, + handoffIds: appendContextHandoffId( + queuedProviderThread.handoffIds, + activeHandoff?.id ?? null, + ), updatedAt: now, }; const startingRun: OrchestrationV2Run = { @@ -988,6 +1279,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio status: "starting", queuePosition: null, startedAt: null, + contextHandoffId: activeHandoff?.id ?? null, }; const userTurnItem: OrchestrationV2TurnItem = { ...(legacyQueuedTurnItem ?? { @@ -1018,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 ? [ @@ -1041,8 +1374,115 @@ 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 || targetLastCompletedRun === undefined + ? null + : contextSourcePointForRun(projection, targetLastCompletedRun), + 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, + }, + ]), + ...(legacyImportRecoveryHandoff === null + ? [] + : [ + { + type: "context-handoff.updated" as const, + threadId, + runId: queuedRun.id, + providerInstanceId: queuedRun.providerInstanceId, + occurredAt: now, + 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, + 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 +1511,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, @@ -1079,7 +1529,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"); @@ -3155,6 +3605,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", @@ -3591,33 +4057,54 @@ 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 - : 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 }); + const targetProviderThread = + modelSelection.instanceId === queueProviderThread.providerInstanceId + ? queueProviderThread + : rootProviderThreadsForProvider(projection, modelSelection.instanceId)[0]; + 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, + 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,10 +4116,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio threadId: command.threadId, runId, rootNodeId, - providerThreadId: queueProviderThread.id, + providerThreadId: queuedProviderThread.id, cwd: resolvedRuntimePolicy.cwd ?? - existingProviderSession?.cwd ?? + selectedProviderSession?.cwd ?? projection.thread.worktreePath ?? process.cwd(), createdAt: now, @@ -3653,7 +4140,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 +4168,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 +4184,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 +4213,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, @@ -3798,7 +4294,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, @@ -3823,6 +4319,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"); @@ -4388,6 +4907,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 ? [] @@ -4397,7 +4920,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 = @@ -4886,18 +5409,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/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 09a79aed4ca8..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, @@ -546,7 +557,8 @@ export function applyToProjection( return { ...base, thread: - event.payload.appThreadId === base.thread.id + event.payload.appThreadId === base.thread.id && + !isQueuedProviderThreadPlaceholder(event.payload) ? { ...base.thread, activeProviderThreadId: event.payload.id, @@ -1830,7 +1842,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 && + !isQueuedProviderThreadPlaceholder(event.payload) + ) { 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..4d2e5a2a5cb3 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, @@ -9,6 +10,7 @@ import { type OrchestrationV2ProviderThread, ProjectId, ProviderInstanceId, + ProviderSessionId, ProviderThreadId, ProviderTurnId, ThreadId, @@ -16,7 +18,9 @@ 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 Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as PubSub from "effect/PubSub"; import * as Ref from "effect/Ref"; @@ -24,16 +28,21 @@ 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"; 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 { OrchestratorDispatchError, OrchestratorV2 } from "../Orchestrator.ts"; +import { OrchestrationEffectWorkerV2 } from "../EffectWorker.ts"; +import { EffectOutboxV2, layer as effectOutboxLayer } from "../EffectOutbox.ts"; import { ProjectionMaintenanceV2, layer as projectionMaintenanceLayer, @@ -45,11 +54,16 @@ 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, CODEX_MODEL_SELECTION, CURSOR_MODEL_SELECTION, + GROK_MODEL_SELECTION, } from "./fixtures/shared.ts"; import { makeOrchestratorV2ReplayLayerWithRegistry } from "./ProviderReplayHarness.ts"; import { checkpointWorkspace } from "./ReplayFixtureWorkspace.ts"; @@ -62,6 +76,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; @@ -84,6 +99,8 @@ function makeTestAdapter(input: { readonly capturedTurns: Ref.Ref>; readonly failResume?: boolean; readonly failedRunOrdinals?: ReadonlySet; + readonly holdFirstTurn?: Deferred.Deferred; + readonly releaseFirstTurn?: Deferred.Deferred; }): ProviderAdapterV2Shape { return { instanceId: input.instanceId, @@ -155,6 +172,11 @@ function makeTestAdapter(input: { text: turnInput.message.text, }, ]); + if (turnInput.runOrdinal === 1 && input.holdFirstTurn !== undefined) { + yield* Deferred.succeed(input.holdFirstTurn, undefined); + if (input.releaseFirstTurn === undefined) return; + yield* Deferred.await(input.releaseFirstTurn); + } const eventTime = yield* DateTime.now; const providerTurnId = ProviderTurnId.make( `provider-turn:${input.driver}:${turnInput.threadId}:${turnInput.runOrdinal}`, @@ -296,7 +318,974 @@ const waitForIdle = Effect.fn("ProviderSwitchTest.waitForIdle")(function* ( }); describe("orchestration v2 provider switching", () => { - it.live("reissues imported v1 context when switching after the first provider fails", () => + 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* () { + 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); + assert.deepEqual(afterSteer.thread.modelSelection, GROK_MODEL_SELECTION); + 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* () { + 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* () { + 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 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", + 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 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", + 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(); + 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( + Layer.merge( + makeOrchestratorV2ReplayLayerWithRegistry( + { + name: "queued-provider-switch", + runtimePolicyOverride: { + cwd, + approvalPolicy: "never", + sandboxPolicy: { + type: "readOnly", + access: { type: "fullAccess" }, + networkAccess: false, + }, + }, + }, + registryLayer, + { databaseLayer }, + ), + outboxProvided, + ), + ), + ); + 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); + 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"); + }), + ), + ); + + 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.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( + (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* () { const importedThreadId = ThreadId.make("thread:provider-switch:legacy-import"); @@ -305,6 +1294,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"), @@ -314,6 +1305,7 @@ describe("orchestration v2 provider switching", () => { responseByRunOrdinal: {}, capturedTurns, failedRunOrdinals: new Set([1]), + ...(queueBeforeFailure ? { holdFirstTurn: firstTurnStarted, releaseFirstTurn } : {}), }), makeTestAdapter({ instanceId: ProviderInstanceId.make("claudeAgent"), @@ -367,98 +1359,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; @@ -476,7 +1469,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", @@ -487,8 +1484,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)); @@ -512,13 +1534,33 @@ 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."); 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", () => @@ -607,8 +1649,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 a075c7ce9f72..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,7 +8814,6 @@ export default function ChatView(props: ChatViewProps) { const settingsResult = await persistThreadSettingsForNextTurn({ threadId: threadIdForSend, createdAt: messageCreatedAt, - ...(ctxSelectedModel ? { modelSelection: ctxSelectedModelSelection } : {}), ...(localCheckoutBranchMismatch ? { branch: localCheckoutBranchMismatch.currentBranch } : {}), @@ -9305,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,