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 6544fafdacbff..d79d1a2d9ca97 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts @@ -62,6 +62,194 @@ const layerTest = Layer.mergeAll( ), ); +it.effect.each( + (["provider.switch", "thread.model-selection.set"] as const).flatMap((type) => + [ + { active: true, selectionChange: "instance" }, + { active: false, selectionChange: "instance" }, + { active: true, selectionChange: "next-turn" }, + { active: true, selectionChange: "restart" }, + ].map((scenario) => ({ type, ...scenario })), + ), +)("$type $selectionChange ($active)", ({ type, active, selectionChange }) => + 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: + selectionChange === "instance" ? ProviderInstanceId.make("codex-other") : instanceId, + model: selectionChange === "instance" ? modelSelection.model : "gpt-6", + }; + const outcome = yield* orchestrator + .dispatch({ + type, + commandId, + threadId, + modelSelection: target, + }) + .pipe(Effect.result); + const after = yield* projections.getThreadProjection(threadId); + if (active && selectionChange !== "next-turn") { + 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)); + assert.instanceOf(outcome.failure, Orchestrator.OrchestratorProviderSelectionBusyError); + } 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, + 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 }, + ), + ), + ), + ), +); + it.effect( "records approval feedback and targets only its runtime request without dispatching a parent turn", () => diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 07c1a544798be..8073b17ac07f2 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -291,6 +291,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.`; + } +} + const isOrchestratorDispatchError = Schema.is(OrchestratorDispatchError); /** The ceiling refusal under however many dispatch errors wrapped it on the way out. */ @@ -355,6 +364,7 @@ export const OrchestratorV2Error = Schema.Union([ OrchestratorCommandIdConflictError, OrchestratorSubagentThreadReadOnlyError, OrchestratorThreadAboveModeLimitError, + OrchestratorProviderSelectionBusyError, OrchestratorRuntimeModeAboveCeilingError, ]); export type OrchestratorV2Error = typeof OrchestratorV2Error.Type; @@ -3791,12 +3801,24 @@ 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(isActiveRun)) { + return yield* new OrchestratorProviderSelectionBusyError({ + commandId: command.commandId, + threadId: command.threadId, + }); + } + } + return plan; }) : null; diff --git a/apps/server/src/orchestration-v2/ThreadMessageIntake.ts b/apps/server/src/orchestration-v2/ThreadMessageIntake.ts index 3e8922c3ba706..a48cc3ca7cd90 100644 --- a/apps/server/src/orchestration-v2/ThreadMessageIntake.ts +++ b/apps/server/src/orchestration-v2/ThreadMessageIntake.ts @@ -24,6 +24,7 @@ function dispatchWasNotAccepted( case "OrchestratorCommandIdConflictError": case "OrchestratorSubagentThreadReadOnlyError": case "OrchestratorThreadAboveModeLimitError": + case "OrchestratorProviderSelectionBusyError": case "OrchestratorRuntimeModeAboveCeilingError": case "ThreadManagementProviderInstanceDisabledError": return true; diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index a49e323770cc1..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]) => @@ -6067,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) =>