From cc3e3d0ef13df7d7888ba89e6255485fc5622c4d Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 21 Oct 2026 08:31:03 +0000 Subject: [PATCH 1/5] test(server): takeover tests for active provider instance switches --- .../Orchestrator.control-reads.test.ts | 180 ++++++++++++++++++ 1 file changed, 180 insertions(+) diff --git a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts index eda76857a9efe..1dcad82577fac 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts @@ -57,6 +57,186 @@ const layerTest = Layer.mergeAll( ), ); +it.effect.each( + (["provider.switch", "thread.model-selection.set"] as const).flatMap((type) => + [true, false].map((active) => ({ type, active })), + ), +)("$type instance switch (active: $active)", ({ type, active }) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const outbox = yield* EffectOutbox.EffectOutboxV2; + const threadId = ThreadId.make("thread:instance-switch"); + const now = yield* DateTime.now; + const sessionId = ProviderSessionId.make("session:instance-switch"); + const providerThreadId = ProviderThreadId.make("provider-thread:instance-switch"); + const runId = RunId.make("run:instance-switch"); + const nodeId = NodeId.make("node:instance-switch"); + const attemptId = RunAttemptId.make("attempt:instance-switch"); + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make("create-instance-switch"), + threadId, + projectId: ProjectId.make("project:instance-switch"), + title: "Instance switch", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + createdBy: "user", + creationSource: "web", + }); + yield* projections.apply({ + id: EventId.make("instance-switch:session"), + type: "provider-session.attached", + threadId, + occurredAt: now, + payload: { + id: sessionId, + driver: adapter.driver, + providerInstanceId: instanceId, + status: "ready", + cwd: "/repo", + model: modelSelection.model, + capabilities: CodexProviderCapabilitiesV2, + createdAt: now, + updatedAt: now, + lastError: null, + }, + }); + yield* projections.apply({ + id: EventId.make("instance-switch:provider-thread"), + type: "provider-thread.updated", + threadId, + occurredAt: now, + payload: { + id: providerThreadId, + driver: adapter.driver, + providerInstanceId: instanceId, + providerSessionId: sessionId, + appThreadId: threadId, + ownerNodeId: null, + nativeThreadRef: { + driver: adapter.driver, + nativeId: "native:instance-switch", + strength: "strong", + }, + nativeConversationHeadRef: null, + status: "idle", + firstRunOrdinal: 1, + lastRunOrdinal: 1, + handoffIds: [], + forkedFrom: null, + createdAt: now, + updatedAt: now, + }, + }); + if (active) { + yield* projections.apply({ + id: EventId.make("instance-switch:run"), + type: "run.created", + threadId, + occurredAt: now, + payload: { + id: runId, + threadId, + ordinal: 1, + providerInstanceId: instanceId, + modelSelection, + providerThreadId, + userMessageId: MessageId.make("message:instance-switch"), + rootNodeId: nodeId, + activeAttemptId: attemptId, + status: "running", + requestedAt: now, + startedAt: now, + completedAt: null, + checkpointId: null, + contextHandoffId: null, + }, + }); + yield* projections.apply({ + id: EventId.make("instance-switch:turn"), + type: "provider-turn.updated", + threadId, + runId, + occurredAt: now, + payload: { + id: ProviderTurnId.make("provider-turn:instance-switch"), + providerThreadId, + nodeId, + runAttemptId: attemptId, + nativeTurnRef: { driver: adapter.driver, nativeId: "native:turn", strength: "strong" }, + ordinal: 1, + status: "running", + startedAt: now, + completedAt: null, + }, + }); + } + const before = yield* projections.getThreadProjection(threadId); + const sequence = yield* orchestrator.getThreadEventSequence(threadId); + const commandId = CommandId.make("switch-instance"); + const target = { + instanceId: ProviderInstanceId.make("codex-other"), + model: modelSelection.model, + }; + const outcome = yield* orchestrator + .dispatch({ + type, + commandId, + threadId, + modelSelection: target, + }) + .pipe(Effect.result); + const after = yield* projections.getThreadProjection(threadId); + if (active) { + assert.deepEqual( + after, + before, + "a refused switch must preserve the turn, session and selection", + ); + assert.deepEqual( + yield* outbox.listByCommandId(commandId), + [], + "a refused switch must not release the provider session", + ); + assert.equal(yield* orchestrator.getThreadEventSequence(threadId), sequence); + assert.ok( + Result.isFailure(outcome), + "retry the instance switch after the active turn finishes", + ); + assert.instanceOf(outcome.failure, Orchestrator.OrchestratorDispatchError); + const reason = String(outcome.failure.cause); + assert.match(reason, /(?:active|running).*turn|turn.*(?:active|running)/i); + assert.match(reason, /(?:retry|try again|wait|finish|complete)/i); + } else { + assert.ok(Result.isSuccess(outcome)); + assert.deepEqual(after.thread.modelSelection, target); + assert.equal(after.thread.providerInstanceId, target.instanceId); + assert.deepEqual(after.runs, before.runs); + assert.deepEqual(after.providerTurns, before.providerTurns); + } + }).pipe( + Effect.provide( + Layer.mergeAll( + layerDatabase, + ProjectionStore.layer.pipe(Layer.provide(layerDatabase)), + EffectOutbox.layer.pipe(Layer.provide(layerDatabase)), + ProviderReplayHarness.layerWithRegistry( + { name: "instance-switch" }, + ProviderAdapterRegistry.layerFromAdapters([ + adapter, + { ...adapter, instanceId: ProviderInstanceId.make("codex-other") }, + ]), + { databaseLayer: layerDatabase, runEffectWorker: false }, + ), + ), + ), + ), +); + it.effect("marks a never-visited thread's first completion unread", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; From 131a5532725e24baa0439c967770506557de77be Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 21 Oct 2026 08:35:11 +0000 Subject: [PATCH 2/5] fix(server): refuse provider switches during active turns --- .../Orchestrator.control-reads.test.ts | 26 ++++++++++++++----- .../src/orchestration-v2/Orchestrator.ts | 16 +++++++++++- 2 files changed, 35 insertions(+), 7 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts index 1dcad82577fac..1e6d9b201cc1c 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts @@ -59,9 +59,14 @@ const layerTest = Layer.mergeAll( it.effect.each( (["provider.switch", "thread.model-selection.set"] as const).flatMap((type) => - [true, false].map((active) => ({ type, active })), + [ + { active: true, selectionChange: "instance" }, + { active: false, selectionChange: "instance" }, + { active: true, selectionChange: "next-turn" }, + { active: true, selectionChange: "restart" }, + ].map((scenario) => ({ type, ...scenario })), ), -)("$type instance switch (active: $active)", ({ type, active }) => +)("$type $selectionChange ($active)", ({ type, active, selectionChange }) => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; const projections = yield* ProjectionStore.ProjectionStoreV2; @@ -179,8 +184,9 @@ it.effect.each( const sequence = yield* orchestrator.getThreadEventSequence(threadId); const commandId = CommandId.make("switch-instance"); const target = { - instanceId: ProviderInstanceId.make("codex-other"), - model: modelSelection.model, + instanceId: + selectionChange === "instance" ? ProviderInstanceId.make("codex-other") : instanceId, + model: selectionChange === "instance" ? modelSelection.model : "gpt-6", }; const outcome = yield* orchestrator .dispatch({ @@ -191,7 +197,7 @@ it.effect.each( }) .pipe(Effect.result); const after = yield* projections.getThreadProjection(threadId); - if (active) { + if (active && selectionChange !== "next-turn") { assert.deepEqual( after, before, @@ -227,7 +233,15 @@ it.effect.each( ProviderReplayHarness.layerWithRegistry( { name: "instance-switch" }, ProviderAdapterRegistry.layerFromAdapters([ - adapter, + { + ...adapter, + planSelectionTransition: () => + Effect.succeed( + selectionChange === "restart" + ? { type: "restart_session" as const } + : { type: "apply_on_next_turn" as const }, + ), + }, { ...adapter, instanceId: ProviderInstanceId.make("codex-other") }, ]), { databaseLayer: layerDatabase, runEffectWorker: false }, diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 213a1a31ab74c..105dd77eb41d7 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -2931,12 +2931,26 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }), ), ); - return yield* providerSwitchService + const plan = yield* providerSwitchService .plan({ projection: providerContext!, targetModelSelection: command.modelSelection, }) .pipe(mapDispatchError(command)); + if (plan.instanceChanged || plan.releaseProviderSessionIds.length > 0) { + const projection = yield* loadProjectionForCommand(command, ["runs"], { + turnItemTypes: [], + }); + if (projection.runs.some(isBlockingRun)) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: + "Wait for the active turn to finish, then retry the provider selection change.", + }); + } + } + return plan; }) : null; From 9362d8b7167b35496ece07ec89b56236079d3c36 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 22 Oct 2026 02:40:53 +0000 Subject: [PATCH 3/5] test(server): hold the queue before a mid-queue provider switch The active-run guard refuses an instance switch while a turn is running, and completing the active run starts the queue at once. Complete the run with the queue held, switch, then resume so the queued monitor update is still checked against the new selection. Co-Authored-By: Claude Opus 5.5 --- .../src/orchestration-v2/runtimeLayer.test.ts | 31 ++++++++++++++----- 1 file changed, 24 insertions(+), 7 deletions(-) diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 9db0564cced01..477f344c9adf1 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -4261,12 +4261,7 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { id: ProviderThreadId.make(`${threadId}:current`), updatedAt: now, }; - yield* orchestrator.dispatch({ - type: "thread.model-selection.set", - commandId: CommandId.make(`${threadId}:select-current`), - threadId, - modelSelection: currentSelection, - }); + // Provider selection only changes between turns, so hold the queue past the active run. yield* eventSink.write({ events: [ { @@ -4277,6 +4272,24 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { occurredAt: now, payload: { ...active, status: "completed", completedAt: now }, }, + { + id: EventId.make(`${threadId}:held`), + type: "run.updated", + threadId, + runId: queued.id, + occurredAt: now, + payload: { ...queued, queueHeld: true }, + }, + ], + }); + yield* orchestrator.dispatch({ + type: "thread.model-selection.set", + commandId: CommandId.make(`${threadId}:select-current`), + threadId, + modelSelection: currentSelection, + }); + yield* eventSink.write({ + events: [ { id: EventId.make(`${threadId}:conversation`), type: "provider-thread.updated", @@ -4288,7 +4301,11 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { }); const expectedThreadId = creationSource === "provider" ? queued.providerThreadId : current.id; - yield* orchestrator.resumeQueuedRuns; + yield* orchestrator.dispatch({ + type: "queue.resume", + commandId: CommandId.make(`${threadId}:resume`), + threadId, + }); const after = yield* orchestrator.getThreadProjection(threadId); const delivered = after.runs.find((run) => run.id === queued.id)!; assert.equal(delivered.status, "starting"); From 6b3bdce4a6278924d6236d892f8179b434a7df80 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 22 Oct 2026 08:31:23 +0000 Subject: [PATCH 4/5] fix(server): a busy provider switch fails with its own tagged error The active-turn guard stored its reason as a synthetic cause on a generic dispatch error. It now raises OrchestratorProviderSelectionBusyError, whose message comes from its attributes, and intake treats it as not accepted. Co-Authored-By: Claude Opus 5.5 --- .../Orchestrator.control-reads.test.ts | 7 +++---- apps/server/src/orchestration-v2/Orchestrator.ts | 16 ++++++++++++---- .../src/orchestration-v2/ThreadMessageIntake.ts | 1 + 3 files changed, 16 insertions(+), 8 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts index 1e6d9b201cc1c..7aaed6c3d466d 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts @@ -213,10 +213,9 @@ it.effect.each( Result.isFailure(outcome), "retry the instance switch after the active turn finishes", ); - assert.instanceOf(outcome.failure, Orchestrator.OrchestratorDispatchError); - const reason = String(outcome.failure.cause); - assert.match(reason, /(?:active|running).*turn|turn.*(?:active|running)/i); - assert.match(reason, /(?:retry|try again|wait|finish|complete)/i); + assert.instanceOf(outcome.failure, Orchestrator.OrchestratorProviderSelectionBusyError); + assert.match(outcome.failure.message, /active turn/i); + assert.match(outcome.failure.message, /retry/i); } else { assert.ok(Result.isSuccess(outcome)); assert.deepEqual(after.thread.modelSelection, target); diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 1b9c6c56805a4..875a0fe2fa13c 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -240,6 +240,15 @@ export class OrchestratorThreadAboveModeLimitError extends Schema.TaggedError()( + "OrchestratorProviderSelectionBusyError", + { commandId: CommandId, threadId: ThreadId }, +) { + override get message(): string { + return `Thread ${this.threadId} has an active turn. Wait for it to finish, then retry the provider selection change.`; + } +} + export class OrchestratorCommandPreviouslyRejectedError extends Schema.TaggedError()( "OrchestratorCommandPreviouslyRejectedError", { @@ -288,6 +297,7 @@ export const OrchestratorV2Error = Schema.Union([ OrchestratorCommandIdConflictError, OrchestratorSubagentThreadReadOnlyError, OrchestratorThreadAboveModeLimitError, + OrchestratorProviderSelectionBusyError, ]); export type OrchestratorV2Error = typeof OrchestratorV2Error.Type; @@ -2958,11 +2968,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio turnItemTypes: [], }); if (projection.runs.some(isBlockingRun)) { - return yield* new OrchestratorDispatchError({ + return yield* new OrchestratorProviderSelectionBusyError({ commandId: command.commandId, - commandType: command.type, - cause: - "Wait for the active turn to finish, then retry the provider selection change.", + threadId: command.threadId, }); } } diff --git a/apps/server/src/orchestration-v2/ThreadMessageIntake.ts b/apps/server/src/orchestration-v2/ThreadMessageIntake.ts index fa8f068849da1..f71d30c3b3c75 100644 --- a/apps/server/src/orchestration-v2/ThreadMessageIntake.ts +++ b/apps/server/src/orchestration-v2/ThreadMessageIntake.ts @@ -25,6 +25,7 @@ function dispatchWasNotAccepted( case "OrchestratorCommandIdConflictError": case "OrchestratorSubagentThreadReadOnlyError": case "OrchestratorThreadAboveModeLimitError": + case "OrchestratorProviderSelectionBusyError": case "ThreadManagementProviderInstanceDisabledError": return true; default: From dd9772c252c558b852e7a7ffbc232bd87f6e82d0 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 11 Nov 2026 06:01:15 +0000 Subject: [PATCH 5/5] fix(server): MCP callers see why a mid-turn provider switch was refused t3_thread_configure now returns the busy error's retry message as an invalid_request instead of the redacted generic failure. Main's queued-delivery tests switched provider instance during an active run, which this change refuses; their same-instance rows still cover the queued-selection behavior, and the refusal is covered in control-reads. Drops copy-text assertions from the guard test. AI-assisted; verified with the tests and tools listed above. --- apps/server/src/mcp/threadAccess.ts | 3 +- apps/server/src/mcp/toolkits/core.test.ts | 6 ++++ .../Orchestrator.control-reads.test.ts | 7 +--- .../src/orchestration-v2/runtimeLayer.test.ts | 33 ++++--------------- 4 files changed, 16 insertions(+), 33 deletions(-) diff --git a/apps/server/src/mcp/threadAccess.ts b/apps/server/src/mcp/threadAccess.ts index 4bf90f0182fc4..48d972c5be2fd 100644 --- a/apps/server/src/mcp/threadAccess.ts +++ b/apps/server/src/mcp/threadAccess.ts @@ -48,7 +48,8 @@ export const dispatchFailure = ( error: OrchestratorV2Error | ThreadManagement.ThreadManagementProviderInstanceDisabledError, ) => error._tag === "ThreadManagementProviderInstanceDisabledError" || - error._tag === "OrchestratorRuntimeModeAboveCeilingError" + error._tag === "OrchestratorRuntimeModeAboveCeilingError" || + error._tag === "OrchestratorProviderSelectionBusyError" ? new OrchestratorMcpFailure({ code: "invalid_request", message: error.message }) : (error._tag === "OrchestratorDispatchError" || error._tag === "OrchestratorCommandRejectedError") && diff --git a/apps/server/src/mcp/toolkits/core.test.ts b/apps/server/src/mcp/toolkits/core.test.ts index fd07107b08bc6..711160533c568 100644 --- a/apps/server/src/mcp/toolkits/core.test.ts +++ b/apps/server/src/mcp/toolkits/core.test.ts @@ -35,6 +35,7 @@ import { OrchestratorCommandRejectedError, OrchestratorDispatchError, OrchestratorProjectionError, + OrchestratorProviderSelectionBusyError, } from "../../orchestration-v2/Orchestrator.ts"; import * as ServerSecretStore from "../../auth/ServerSecretStore.ts"; @@ -227,6 +228,11 @@ it("bounds public command rejections and redacts internal dispatch causes", () = new OrchestratorCommandRejectedError({ ...command, cause: "Run is not queued." }), ).message, ).toBe("Run is not queued."); + const busy = new OrchestratorProviderSelectionBusyError({ + commandId: command.commandId, + threadId, + }); + expect(dispatchFailure(busy)).toMatchObject({ code: "invalid_request", message: busy.message }); for (const cause of [ undefined, "", diff --git a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts index 1b665bec06b8f..d79d1a2d9ca97 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts @@ -214,13 +214,8 @@ it.effect.each( "a refused switch must not release the provider session", ); assert.equal(yield* orchestrator.getThreadEventSequence(threadId), sequence); - assert.ok( - Result.isFailure(outcome), - "retry the instance switch after the active turn finishes", - ); + assert.ok(Result.isFailure(outcome)); assert.instanceOf(outcome.failure, Orchestrator.OrchestratorProviderSelectionBusyError); - assert.match(outcome.failure.message, /active turn/i); - assert.match(outcome.failure.message, /retry/i); } else { assert.ok(Result.isSuccess(outcome)); assert.deepEqual(after.thread.modelSelection, target); diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index ec9869b7af0f5..3495e940702be 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -5954,7 +5954,6 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { it.effect.each([ ["server", modelSelection], ["provider", modelSelection], - ["provider", { ...modelSelection, instanceId: alternateInstanceId }], ] as const)( "delivers a queued monitor update from %s after conversation selection %j", ([creationSource, currentSelection]) => @@ -6010,7 +6009,12 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { id: ProviderThreadId.make(`${threadId}:current`), updatedAt: now, }; - // Provider selection only changes between turns, so hold the queue past the active run. + yield* orchestrator.dispatch({ + type: "thread.model-selection.set", + commandId: CommandId.make(`${threadId}:select-current`), + threadId, + modelSelection: currentSelection, + }); yield* eventSink.write({ events: [ { @@ -6021,24 +6025,6 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { occurredAt: now, payload: { ...active, status: "completed", completedAt: now }, }, - { - id: EventId.make(`${threadId}:held`), - type: "run.updated", - threadId, - runId: queued.id, - occurredAt: now, - payload: { ...queued, queueHeld: true }, - }, - ], - }); - yield* orchestrator.dispatch({ - type: "thread.model-selection.set", - commandId: CommandId.make(`${threadId}:select-current`), - threadId, - modelSelection: currentSelection, - }); - yield* eventSink.write({ - events: [ { id: EventId.make(`${threadId}:conversation`), type: "provider-thread.updated", @@ -6050,11 +6036,7 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { }); const expectedThreadId = creationSource === "provider" ? queued.providerThreadId : current.id; - yield* orchestrator.dispatch({ - type: "queue.resume", - commandId: CommandId.make(`${threadId}:resume`), - threadId, - }); + yield* orchestrator.resumeQueuedRuns; const after = yield* orchestrator.getThreadProjection(threadId); const delivered = after.runs.find((run) => run.id === queued.id)!; assert.equal(delivered.status, "starting"); @@ -6084,7 +6066,6 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { it.effect.each([ { ...modelSelection, model: "gpt-5.5", options: [{ id: "reasoningEffort", value: "high" }] }, - { instanceId: alternateInstanceId, model: "gpt-5.5" }, ])( "starts a queued delegated completion on a selection set while it waited: %j", (newerSelection) =>