diff --git a/apps/server/src/mcp/OrchestratorMcpService.test.ts b/apps/server/src/mcp/OrchestratorMcpService.test.ts index b524921f02a6..1ea5e1ce414c 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.test.ts @@ -3,6 +3,7 @@ import { assert, describe, it } from "@effect/vitest"; import { CommandId, EnvironmentId, + MessageId, NodeId, type OrchestrationV2ThreadShell, type ScheduledTask, @@ -15,11 +16,18 @@ import { type OrchestrationV2ThreadProjection, type ServerProvider, } from "@t3tools/contracts"; +import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Ref from "effect/Ref"; +import { CodexProviderCapabilitiesV2 } from "../orchestration-v2/Adapters/CodexAdapterV2.ts"; +import * as CommandReceiptStore from "../orchestration-v2/CommandReceiptStore.ts"; +import * as LegacyV1ThreadImporter from "../orchestration-v2/legacy/LegacyV1ThreadImporter.ts"; +import * as Orchestrator from "../orchestration-v2/Orchestrator.ts"; +import * as ProviderReplayHarness from "../orchestration-v2/testkit/ProviderReplayHarness.ts"; +import * as SqlitePersistence from "../persistence/Sqlite.ts"; import { DispatchModeLimit, type DispatchModeRefusal, @@ -1677,6 +1685,168 @@ describe("OrchestratorMcpService provider resolution", () => { }), ); + it.live.each([ + { name: "parent instance", parentEnabled: true, model: "gpt-5.4" }, + { name: "alternate instance", parentEnabled: false, model: "gpt-5.5" }, + ])( + "replays accepted creation on the $name without authorizing new disabled work", + ({ parentEnabled, model }) => + Effect.scoped( + Effect.gen(function* () { + const altInstanceId = ProviderInstanceId.make("codex-alt"); + const driver = ProviderDriverKind.make("codex"); + const cwd = yield* checkpointWorkspace("mcp-create-retry"); + const layerRegistry = ProviderAdapterRegistry.layerFromAdapters( + [codexInstanceId, altInstanceId].map((instanceId) => ({ + instanceId, + driver, + getCapabilities: () => Effect.succeed(CodexProviderCapabilitiesV2), + planSelectionTransition: () => + Effect.succeed({ type: "apply_on_next_turn" as const }), + openSession: () => + Effect.die("These tests exercise committed commands before provider execution."), + })), + ); + const layerDatabase = SqlitePersistence.layerMemory; + const layerOrchestrator = ProviderReplayHarness.layerWithRegistry( + { name: "mcp-create-retry", runtimePolicyOverride: { cwd } }, + layerRegistry, + { databaseLayer: layerDatabase, runEffectWorker: false }, + ); + const layerSettings = ServerSettings.layerTest({ + providerInstances: { + [codexInstanceId]: { driver: "codex", enabled: true }, + [altInstanceId]: { driver: "codex", enabled: true }, + }, + projectSettingsOverrides: { + [projectId]: { providerInstances: { [codexInstanceId]: { enabled: parentEnabled } } }, + }, + }); + const layerThreads = ThreadManagementService.layer.pipe( + Layer.provideMerge(layerOrchestrator), + Layer.provide( + LegacyV1ThreadImporter.layer.pipe( + Layer.provide(layerOrchestrator), + Layer.provide(layerDatabase), + ), + ), + Layer.provide(CommandReceiptStore.layer.pipe(Layer.provide(layerDatabase))), + Layer.provideMerge(layerSettings), + ); + const layerTest = layerOrchestratorService.pipe( + Layer.provideMerge(layerThreads), + Layer.provide( + Layer.mergeAll( + NodeServices.layer, + layerRegistry, + layerSettings, + providerRegistryLayer([ + providerSnapshot({ instanceId: codexInstanceId, driver, model: "gpt-5.4" }), + providerSnapshot({ instanceId: altInstanceId, driver, model: "gpt-5.5" }), + ]), + Layer.mock(ProjectService.ProjectService)({}), + Layer.mock(SecretRequests.SecretRequests)({}), + Layer.mock(ScheduledTaskService.ScheduledTaskService)({}), + ), + ), + ); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const service = yield* OrchestratorMcpService.OrchestratorMcpService; + const settings = yield* ServerSettings.ServerSettingsService; + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("parent:create"), + threadId: parentThreadId, + projectId, + title: "MCP parent", + modelSelection: { instanceId: codexInstanceId, model: "gpt-5.4" }, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: cwd, + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("parent:start"), + threadId: parentThreadId, + messageId: MessageId.make("parent:message"), + text: "Create a thread.", + attachments: [], + dispatchMode: { type: "start_immediately" }, + }); + const request = { + threads: [{ target: { driverKind: driver, model } }], + clientRequestId: "create-retry", + }; + const first = yield* service.createThreads(scope, request); + const child = first.threads[0]!; + const selection = parentEnabled + ? { instanceId: codexInstanceId, model: "gpt-5.4" } + : { instanceId: altInstanceId, model: "gpt-5.5" }; + assert.equal(first.threads.length, 1); + assert.equal(child.providerInstanceId, selection.instanceId); + assert.equal(child.model, selection.model); + assert.deepEqual(yield* service.createThreads(scope, request), first); + + // The creation receipt exists, but this prompt has not been accepted yet. + const withPrompt = { + ...request, + threads: [{ ...request.threads[0]!, prompt: "Start work." }], + }; + const started = yield* service.createThreads(scope, withPrompt); + assert.equal(started.threads[0]?.threadId, child.threadId); + const before = yield* orchestrator.getThreadProjection(child.threadId); + assert.deepEqual(before.runs[0]?.modelSelection, selection); + assert.equal(before.messages.length, 1); + const empty = yield* service.createThreads(scope, { + ...request, + clientRequestId: "empty", + }); + + yield* settings.updateSettings({ + projectSettingsOverrides: { + [projectId]: { + providerInstances: { + [codexInstanceId]: { enabled: false }, + [altInstanceId]: { enabled: false }, + }, + }, + }, + }); + assert.deepEqual(yield* service.createThreads(scope, withPrompt), started); + const after = yield* orchestrator.getThreadProjection(child.threadId); + assert.deepEqual(after.thread.modelSelection, selection); + assert.deepEqual(after.runs, before.runs); + assert.deepEqual(after.messages, before.messages); + + const freshError = yield* service + .createThreads(scope, { + ...request, + clientRequestId: "fresh", + }) + .pipe(Effect.flip); + assert.equal(freshError.code, "provider_unavailable"); + const promptError = yield* service + .createThreads(scope, { + ...withPrompt, + clientRequestId: "empty", + }) + .pipe(Effect.flip); + assert.equal(promptError.code, "invalid_request"); + assert.include(promptError.message, "disabled"); + const untouched = yield* orchestrator.getThreadProjection(empty.threads[0]!.threadId); + assert.equal(untouched.messages.length, 0); + assert.equal(untouched.runs.length, 0); + }).pipe(Effect.provide(layerTest)); + }).pipe(Effect.provide(NodeServices.layer)), + ), + ); + it.effect("re-probes an unavailable target once before refusing it", () => Effect.gen(function* () { const claudeInstanceId = ProviderInstanceId.make("claudeAgent"); diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 86b54b238a16..b31c12d9ca27 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -2336,10 +2336,34 @@ const make = Effect.gen(function* () { input.threads, (request, index) => Effect.gen(function* () { + const commandId = stableCommandId({ + scope, + requestKey: key, + operation: "create-thread", + index, + }); + const threadId = stableThreadId({ + scope, + requestKey: key, + index, + }); + const isRetryOfAcceptedCreation = yield* threadManagement.hasReceipt(commandId); + const committedThread = isRetryOfAcceptedCreation + ? yield* threadManagement + .getThreadShell(threadId) + .pipe(Effect.mapError(threadManagementFailure)) + : null; const target = yield* resolveTargetRechecking({ parent, - target: request.target, + target: + committedThread === null + ? request.target + : { + ...request.target, + providerInstanceId: committedThread.modelSelection.instanceId, + }, providers, + skipProjectEnablementFilter: isRetryOfAcceptedCreation, }); const runtimeMode = yield* resolveRuntimeMode( parent.thread.runtimeMode, @@ -2350,16 +2374,7 @@ const make = Effect.gen(function* () { parent.thread.interactionMode, request.interactionMode, ); - const threadId = stableThreadId({ - scope, - requestKey: key, - index, - }); - yield* assertTargetEnabledForProject( - parent, - stableCommandId({ scope, requestKey: key, operation: "create-thread", index }), - target, - ); + yield* assertTargetEnabledForProject(parent, commandId, target); const title = threadTitle({ parentTitle: parent.thread.title, prompt: request.prompt, @@ -2371,12 +2386,7 @@ const make = Effect.gen(function* () { type: "thread.create", createdBy: "agent", creationSource: "mcp", - commandId: stableCommandId({ - scope, - requestKey: key, - operation: "create-thread", - index, - }), + commandId, threadId, projectId: parent.thread.projectId, title, @@ -2394,6 +2404,10 @@ const make = Effect.gen(function* () { ), ), ); + // A receipt can appear after target resolution; use the committed thread's selection. + const resolvedModelSelection = (yield* threadManagement + .getThreadRecords(threadId, []) + .pipe(Effect.mapError(threadManagementFailure))).thread.modelSelection; if (request.prompt !== undefined) { yield* threadManagement .dispatch({ @@ -2415,15 +2429,17 @@ const make = Effect.gen(function* () { }), text: request.prompt, attachments: [], - modelSelection: target.modelSelection, + modelSelection: resolvedModelSelection, dispatchMode: { type: "start_immediately" }, }) .pipe( Effect.mapError((error) => - failure( - "orchestration_error", - `Unable to start thread ${index + 1}: ${errorMessage(error)}`, - ), + error._tag === "ThreadManagementProviderInstanceDisabledError" + ? threadManagementFailure(error) + : failure( + "orchestration_error", + `Unable to start thread ${index + 1}: ${errorMessage(error)}`, + ), ), ); } @@ -2459,8 +2475,8 @@ const make = Effect.gen(function* () { title: projection.thread.title, createdBy: projection.thread.createdBy, creationSource: projection.thread.creationSource, - providerInstanceId: target.modelSelection.instanceId, - model: target.modelSelection.model, + providerInstanceId: resolvedModelSelection.instanceId, + model: resolvedModelSelection.model, } satisfies OrchestratorMcpCreatedThread; }), { concurrency: 1 },