From ba2e2fbc8c1115da9e1dc7772a867bc1bdfc55dd Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 29 Aug 2026 16:11:15 -0700 Subject: [PATCH 1/6] feat(mcp): configure existing conversations --- ...onversationConfigurationMcpService.test.ts | 445 +++++++++++ .../ConversationConfigurationMcpService.ts | 702 ++++++++++++++++++ apps/server/src/mcp/McpHttpServer.ts | 2 + ...OrchestratorMcpToolkit.integration.test.ts | 550 +++++++++++++- .../src/mcp/toolkits/orchestrator/handlers.ts | 13 + .../mcp/toolkits/orchestrator/tools.test.ts | 26 + .../src/mcp/toolkits/orchestrator/tools.ts | 37 + .../toolkits/worktree/registration.test.ts | 7 + .../Adapters/ClaudeAdapterV2.ts | 1 + apps/server/src/orchestration-v2/EventSink.ts | 9 +- .../src/orchestration-v2/Orchestrator.ts | 130 +++- .../src/orchestration-v2/runtimeLayer.ts | 2 + .../testkit/ProviderReplayHarness.ts | 39 +- .../orchestrator-mcp-server.md | 45 +- docs/user/agent-conversation-controls.md | 15 + .../src/conversationControlMcp.test.ts | 30 + .../contracts/src/conversationControlMcp.ts | 175 +++++ packages/contracts/src/index.ts | 1 + packages/contracts/src/orchestrationV2.ts | 15 + packages/shared/src/t3McpToolPresentation.ts | 2 + 20 files changed, 2219 insertions(+), 27 deletions(-) create mode 100644 apps/server/src/mcp/ConversationConfigurationMcpService.test.ts create mode 100644 apps/server/src/mcp/ConversationConfigurationMcpService.ts create mode 100644 docs/user/agent-conversation-controls.md create mode 100644 packages/contracts/src/conversationControlMcp.test.ts create mode 100644 packages/contracts/src/conversationControlMcp.ts diff --git a/apps/server/src/mcp/ConversationConfigurationMcpService.test.ts b/apps/server/src/mcp/ConversationConfigurationMcpService.test.ts new file mode 100644 index 000000000000..221cb5841207 --- /dev/null +++ b/apps/server/src/mcp/ConversationConfigurationMcpService.test.ts @@ -0,0 +1,445 @@ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, describe, it } from "@effect/vitest"; +import { + EnvironmentId, + ProviderDriverKind, + ProviderInstanceId, + type OrchestrationV2ThreadProjection, + type ServerProvider, + ThreadId, +} from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Ref from "effect/Ref"; + +import { CommandReceiptStoreV2 } from "../orchestration-v2/CommandReceiptStore.ts"; +import { OrchestratorCommandPreviouslyRejectedError } from "../orchestration-v2/Orchestrator.ts"; +import { ProviderSwitchServiceV2 } from "../orchestration-v2/ProviderSwitchService.ts"; +import { ThreadManagementService } from "../orchestration-v2/ThreadManagementService.ts"; +import { ProviderRegistry } from "../provider/Services/ProviderRegistry.ts"; +import type { McpInvocationScope } from "./McpInvocationContext.ts"; +import * as ConversationConfiguration from "./ConversationConfigurationMcpService.ts"; + +const projectId = "project:conversation-configuration" as never; +const parentThreadId = ThreadId.make("thread:conversation-configuration-parent"); +const targetThreadId = ThreadId.make("thread:conversation-configuration-target"); +const providerInstanceId = ProviderInstanceId.make("codex"); + +const provider = { + instanceId: providerInstanceId, + driver: ProviderDriverKind.make("codex"), + enabled: true, + installed: true, + version: "test", + status: "ready", + auth: { status: "authenticated" }, + checkedAt: "2026-08-29T00:00:00.000Z", + models: [ + { + slug: "gpt-5.6-sol", + name: "GPT-5.6 Sol", + isCustom: false, + capabilities: { + optionDescriptors: [ + { + id: "reasoning", + label: "Reasoning effort", + type: "select", + options: [ + { id: "low", label: "Low" }, + { id: "high", label: "High" }, + ], + }, + ], + }, + }, + ], + slashCommands: [], + skills: [], +} satisfies ServerProvider; + +function projection(input: { + readonly threadId: ThreadId; + readonly runtimeMode: "approval-required" | "full-access"; + readonly interactionMode: "plan" | "default"; +}): OrchestrationV2ThreadProjection { + return { + thread: { + id: input.threadId, + projectId, + modelSelection: { instanceId: providerInstanceId, model: "gpt-5.6-sol" }, + runtimeMode: input.runtimeMode, + interactionMode: input.interactionMode, + }, + providerSessions: [], + runs: [], + } as unknown as OrchestrationV2ThreadProjection; +} + +const scope = (capabilities: ReadonlyArray<"orchestration"> = ["orchestration"]) => + ({ + environmentId: EnvironmentId.make("environment:conversation-configuration"), + threadId: parentThreadId, + providerSessionId: "provider-session:conversation-configuration", + providerInstanceId, + capabilities: new Set(capabilities), + issuedAt: 1, + }) satisfies McpInvocationScope; + +function testLayer(input: { + readonly parent: OrchestrationV2ThreadProjection; + readonly target: OrchestrationV2ThreadProjection; + readonly dispatch: ThreadManagementService["Service"]["dispatch"]; + readonly getThreadProjection?: ThreadManagementService["Service"]["getThreadProjection"]; + readonly getReceipt?: CommandReceiptStoreV2["Service"]["getByCommandId"]; + readonly providers?: ReadonlyArray; +}) { + return ConversationConfiguration.layer.pipe( + Layer.provide( + Layer.mergeAll( + NodeServices.layer, + Layer.mock(ThreadManagementService)({ + getThreadProjection: + input.getThreadProjection ?? + ((threadId) => + Effect.succeed(threadId === parentThreadId ? input.parent : input.target)), + getProjectThread: ({ threadId }) => + threadId === targetThreadId || threadId === parentThreadId + ? Effect.succeed(threadId === parentThreadId ? input.parent : input.target) + : Effect.die("unexpected thread"), + dispatch: input.dispatch, + }), + Layer.mock(ProviderRegistry)({ + getProviders: Effect.succeed(input.providers ?? [provider]), + }), + Layer.mock(CommandReceiptStoreV2)({ + getByCommandId: input.getReceipt ?? (() => Effect.succeed(Option.none())), + }), + Layer.mock(ProviderSwitchServiceV2)({ + plan: () => + Effect.succeed({ + instanceChanged: false, + modelChanged: false, + targetProviderThreadId: null, + releaseProviderSessionIds: [], + transition: { type: "switch_model_in_session" }, + }), + }), + ), + ), + ); +} + +describe("ConversationConfigurationMcpService", () => { + it.effect("replays an accepted selection before provider availability planning", () => + Effect.gen(function* () { + const parent = projection({ + threadId: parentThreadId, + runtimeMode: "full-access", + interactionMode: "default", + }); + const target = projection({ + threadId: targetThreadId, + runtimeMode: "full-access", + interactionMode: "default", + }); + const layer = testLayer({ + parent, + target, + getReceipt: () => + Effect.succeed( + Option.some({ + status: "accepted", + threadId: targetThreadId, + commandType: "provider.switch", + } as never), + ), + dispatch: () => + Effect.succeed({ + sequence: 17, + storedEvents: [ + { event: { id: "event:accepted-provider-switch", type: "thread.provider-switched" } }, + ] as never, + }), + }); + + const result = yield* Effect.gen(function* () { + const service = yield* ConversationConfiguration.ConversationConfigurationMcpService; + return yield* service.configure(scope(), { + threadId: targetThreadId, + providerInstanceId: ProviderInstanceId.make("provider-now-unavailable"), + model: "unavailable-model", + options: [], + clientRequestId: "accepted-provider-switch", + }); + }).pipe(Effect.provide(layer)); + + assert.equal(result.changes[0]?.receipt?.commandType, "provider.switch"); + assert.equal(result.changes[0]?.receipt?.sequence, 17); + assert.equal(result.selection.providerInstanceId, providerInstanceId); + }), + ); + + it.effect("marks a failed refresh and does not infer replayed inputs as current", () => + Effect.gen(function* () { + const parent = projection({ + threadId: parentThreadId, + runtimeMode: "full-access", + interactionMode: "default", + }); + const target = projection({ + threadId: targetThreadId, + runtimeMode: "approval-required", + interactionMode: "default", + }); + const layer = testLayer({ + parent, + target, + getThreadProjection: (threadId) => + threadId === parentThreadId + ? Effect.succeed(parent) + : Effect.fail(new Error("post-dispatch projection unavailable") as never), + getReceipt: () => Effect.succeed(Option.none()), + dispatch: () => + Effect.succeed({ + sequence: 23, + storedEvents: [ + { + event: { + id: "event:accepted-runtime-replay", + type: "thread.runtime-mode-updated", + }, + }, + ] as never, + }), + }); + + const result = yield* Effect.gen(function* () { + const service = yield* ConversationConfiguration.ConversationConfigurationMcpService; + return yield* service.configure(scope(), { + threadId: targetThreadId, + runtimeMode: "full-access", + clientRequestId: "accepted-runtime-before-another-writer", + }); + }).pipe(Effect.provide(layer)); + + assert.equal(result.observation, "pre_dispatch_fallback"); + assert.equal(result.runtimeMode, "approval-required"); + assert.equal(result.changes[0]?.receipt?.sequence, 23); + }), + ); + + it.effect("validates every field before dispatch and enforces the caller ceiling", () => + Effect.gen(function* () { + const dispatched = yield* Ref.make>([]); + const parent = projection({ + threadId: parentThreadId, + runtimeMode: "approval-required", + interactionMode: "plan", + }); + const target = projection({ + threadId: targetThreadId, + runtimeMode: "approval-required", + interactionMode: "plan", + }); + const layer = testLayer({ + parent, + target, + dispatch: (command) => + Ref.update(dispatched, (commands) => [...commands, command]).pipe( + Effect.as({ sequence: 1, storedEvents: [] }), + ), + }); + + yield* Effect.gen(function* () { + const service = yield* ConversationConfiguration.ConversationConfigurationMcpService; + const invalidOptions = yield* service + .configure(scope(), { + threadId: targetThreadId, + options: [{ id: "reasoning", value: "ultra" }], + runtimeMode: "approval-required", + }) + .pipe(Effect.flip); + assert.equal(invalidOptions.code, "invalid_request"); + + const escalation = yield* service + .configure(scope(), { threadId: targetThreadId, interactionMode: "default" }) + .pipe(Effect.flip); + assert.equal(escalation.code, "interaction_mode_escalation_denied"); + + const unchanged = yield* service.configure(scope(), { + threadId: targetThreadId, + runtimeMode: "approval-required", + clientRequestId: "fresh-unchanged-runtime-mode", + }); + assert.equal(unchanged.outcome, "unchanged"); + assert.deepEqual(unchanged.changes, [ + { + setting: "runtime_mode", + behavior: "unchanged", + requestedEffects: [], + receipt: { + commandId: unchanged.changes[0]!.receipt!.commandId, + commandType: "thread.runtime-mode.set", + sequence: 1, + eventIds: [], + }, + }, + ]); + assert.deepEqual( + (yield* Ref.get(dispatched)).map((command) => (command as { type: string }).type), + ["thread.runtime-mode.set"], + ); + }).pipe(Effect.provide(layer)); + }), + ); + + it.effect("preserves accepted and rejected legs until a new request id is used", () => + Effect.gen(function* () { + const dispatched = yield* Ref.make>([]); + const receipts = new Map< + string, + { + readonly status: "accepted" | "rejected"; + readonly commandType: string; + readonly result?: { + readonly sequence: number; + readonly storedEvents: ReadonlyArray<{ readonly event: { readonly id: string } }>; + }; + } + >(); + let runtimeFailureResolved = false; + const parent = projection({ + threadId: parentThreadId, + runtimeMode: "full-access", + interactionMode: "default", + }); + const target = projection({ + threadId: targetThreadId, + runtimeMode: "full-access", + interactionMode: "default", + }); + const layer = testLayer({ + parent, + target, + getReceipt: (commandId) => { + const existing = receipts.get(commandId); + return Effect.succeed( + existing === undefined + ? Option.none() + : Option.some({ + commandId, + threadId: targetThreadId, + commandType: existing.commandType, + status: existing.status, + } as never), + ); + }, + dispatch: (command) => + Effect.gen(function* () { + yield* Ref.update(dispatched, (commands) => [...commands, command.type]); + const existing = receipts.get(command.commandId); + if (existing?.status === "accepted") { + return existing.result as never; + } + if (existing?.status === "rejected") { + return yield* new OrchestratorCommandPreviouslyRejectedError({ + commandId: command.commandId, + commandType: command.type, + detail: "runtime receipt failed", + }); + } + if (command.type === "thread.runtime-mode.set" && !runtimeFailureResolved) { + receipts.set(command.commandId, { status: "rejected", commandType: command.type }); + return yield* Effect.fail(new Error("runtime receipt failed") as never); + } + const result = { + sequence: receipts.size + 12, + storedEvents: [{ event: { id: `event:${command.commandId}` } }], + }; + receipts.set(command.commandId, { + status: "accepted", + commandType: command.type, + result, + }); + return result as never; + }), + }); + + yield* Effect.gen(function* () { + const service = yield* ConversationConfiguration.ConversationConfigurationMcpService; + const result = yield* service.configure(scope(), { + threadId: targetThreadId, + options: [{ id: "reasoning", value: "high" }], + runtimeMode: "approval-required", + clientRequestId: "partial-configuration", + }); + + assert.equal(result.outcome, "partially_applied"); + assert.equal(result.changes[0]?.receipt?.sequence, 12); + assert.deepEqual(result.errors, [ + { setting: "runtime_mode", message: "runtime receipt failed" }, + ]); + assert.deepEqual(yield* Ref.get(dispatched), [ + "thread.model-selection.set", + "thread.runtime-mode.set", + ]); + + const sameKey = yield* service.configure(scope(), { + threadId: targetThreadId, + options: [{ id: "reasoning", value: "high" }], + runtimeMode: "approval-required", + clientRequestId: "partial-configuration", + }); + assert.equal(sameKey.outcome, "partially_applied"); + assert.equal(sameKey.changes[0]?.receipt?.commandId, result.changes[0]?.receipt?.commandId); + assert.match(sameKey.errors[0]?.message ?? "", /previously rejected/); + assert.deepEqual(sameKey.retryGuidance, { + sameClientRequestId: "replays_durable_decisions", + afterRejectedLeg: "use_new_client_request_id", + }); + + runtimeFailureResolved = true; + const stillRejected = yield* service.configure(scope(), { + threadId: targetThreadId, + options: [{ id: "reasoning", value: "high" }], + runtimeMode: "approval-required", + clientRequestId: "partial-configuration", + }); + assert.equal(stillRejected.outcome, "partially_applied"); + assert.match(stillRejected.errors[0]?.message ?? "", /previously rejected/); + + const newAttempt = yield* service.configure(scope(), { + threadId: targetThreadId, + options: [{ id: "reasoning", value: "high" }], + runtimeMode: "approval-required", + clientRequestId: "partial-configuration-after-fix", + }); + assert.equal(newAttempt.outcome, "applied"); + assert.deepEqual(newAttempt.errors, []); + }).pipe(Effect.provide(layer)); + }), + ); + + it.effect("denies credentials without orchestration capability", () => + Effect.gen(function* () { + const parent = projection({ + threadId: parentThreadId, + runtimeMode: "full-access", + interactionMode: "default", + }); + const layer = testLayer({ + parent, + target: parent, + dispatch: () => Effect.die("dispatch should not run"), + }); + + const error = yield* Effect.gen(function* () { + const service = yield* ConversationConfiguration.ConversationConfigurationMcpService; + return yield* service.read(scope([]), {}).pipe(Effect.flip); + }).pipe(Effect.provide(layer)); + assert.equal(error.code, "capability_denied"); + }), + ); +}); diff --git a/apps/server/src/mcp/ConversationConfigurationMcpService.ts b/apps/server/src/mcp/ConversationConfigurationMcpService.ts new file mode 100644 index 000000000000..ac719d1bd05f --- /dev/null +++ b/apps/server/src/mcp/ConversationConfigurationMcpService.ts @@ -0,0 +1,702 @@ +import { + CommandId, + type ConversationConfigurationBehavior, + type ConversationConfigurationChange, + type ConversationConfigurationInput, + type ConversationConfigurationProvider, + type ConversationConfigurationReceipt, + type ConversationConfigurationResult, + type ConversationConfigurationSelection, + type ConversationConfigureInput, + type ConversationConfigureResult, + isProviderAvailable, + type ModelSelection, + OrchestratorMcpFailure, + type OrchestrationV2Command, + type OrchestrationV2PolicyCeiling, + type OrchestrationV2ThreadProjection, + type ProviderInteractionMode, + type ProviderOptionDescriptor, + type ProviderOptionSelection, + type RuntimeMode, + type ServerProvider, + ThreadId, +} from "@t3tools/contracts"; +import { modelSelectionsEqual } from "@t3tools/shared/model"; +import * as Context from "effect/Context"; +import * as Crypto from "effect/Crypto"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Result from "effect/Result"; +import * as Schema from "effect/Schema"; + +import { isBuiltInProviderAdapterDriverV2 } from "../orchestration-v2/builtInProviderAdapterDrivers.ts"; +import { CommandReceiptStoreV2 } from "../orchestration-v2/CommandReceiptStore.ts"; +import type { OrchestratorV2DispatchResult } from "../orchestration-v2/Orchestrator.ts"; +import { ProviderSwitchServiceV2 } from "../orchestration-v2/ProviderSwitchService.ts"; +import { + isActiveRun, + ThreadManagementService, + ThreadManagementThreadNotFoundError, +} from "../orchestration-v2/ThreadManagementService.ts"; +import { ProviderRegistry } from "../provider/Services/ProviderRegistry.ts"; +import type { McpInvocationScope } from "./McpInvocationContext.ts"; + +type ConfigurationSetting = ConversationConfigurationChange["setting"]; +type ConfigurationCommand = Extract< + OrchestrationV2Command, + { + readonly type: + | "thread.model-selection.set" + | "provider.switch" + | "thread.runtime-mode.set" + | "thread.interaction-mode.set"; + } +>; +type ConfigurationCommandType = ConfigurationCommand["type"]; +type SelectionCommandType = Extract< + ConfigurationCommandType, + "thread.model-selection.set" | "provider.switch" +>; + +export class ConversationConfigurationMcpService extends Context.Service< + ConversationConfigurationMcpService, + { + readonly read: ( + scope: McpInvocationScope, + input: ConversationConfigurationInput, + ) => Effect.Effect; + readonly configure: ( + scope: McpInvocationScope, + input: ConversationConfigureInput, + ) => Effect.Effect; + } +>()("t3/mcp/ConversationConfigurationMcpService") {} + +const isThreadNotFound = Schema.is(ThreadManagementThreadNotFoundError); + +function failure(code: OrchestratorMcpFailure["code"], message: string): OrchestratorMcpFailure { + return new OrchestratorMcpFailure({ code, message }); +} + +function errorMessage(error: unknown): string { + return error instanceof Error ? error.message : String(error); +} + +function runtimeModeRank(mode: RuntimeMode): number { + switch (mode) { + case "approval-required": + return 0; + case "auto-accept-edits": + return 1; + case "auto": + return 2; + case "full-access": + return 3; + } +} + +function interactionModeRank(mode: ProviderInteractionMode): number { + return mode === "plan" ? 0 : 1; +} + +const runtimeModes: ReadonlyArray = [ + "approval-required", + "auto-accept-edits", + "auto", + "full-access", +]; +const interactionModes: ReadonlyArray = ["plan", "default"]; + +function providerConstraints(provider: ServerProvider): ReadonlyArray { + const constraints: Array = []; + if (!isBuiltInProviderAdapterDriverV2(provider.driver)) { + constraints.push("No V2 provider adapter is registered."); + } + if (!provider.enabled) constraints.push("Provider instance is disabled."); + if (!provider.installed) constraints.push("Provider executable is not installed."); + if (!isProviderAvailable(provider)) { + constraints.push(provider.unavailableReason ?? "Provider driver is unavailable."); + } + if (provider.status === "error" || provider.status === "disabled") { + constraints.push(provider.message ?? `Provider status is ${provider.status}.`); + } + if (provider.auth.status === "unauthenticated") { + constraints.push("Provider is not authenticated."); + } + return constraints; +} + +function providerSummary(provider: ServerProvider): ConversationConfigurationProvider { + const constraints = providerConstraints(provider); + return { + providerInstanceId: provider.instanceId, + driverKind: provider.driver, + displayName: provider.displayName ?? null, + selectable: constraints.length === 0, + constraints: [...constraints], + models: provider.models.map((model) => ({ + id: model.slug, + label: model.name ?? null, + options: [...(model.capabilities?.optionDescriptors ?? [])], + })), + }; +} + +function invalidOptionSelections( + selections: ReadonlyArray, + descriptors: ReadonlyArray | undefined, +): ReadonlyArray { + if (selections.length > 0 && descriptors === undefined) { + return ["The selected model does not advertise configurable options."]; + } + const problems: Array = []; + const seen = new Set(); + for (const selection of selections) { + if (seen.has(selection.id)) { + problems.push(`Option ${selection.id} was specified more than once.`); + continue; + } + seen.add(selection.id); + if (descriptors === undefined) continue; + const descriptor = descriptors.find((candidate) => candidate.id === selection.id); + if (descriptor === undefined) { + const known = descriptors.map((candidate) => candidate.id).join(", "); + problems.push(`Unknown option ${selection.id}; supported options: ${known || "none"}.`); + continue; + } + if (descriptor.type === "boolean" && typeof selection.value !== "boolean") { + problems.push(`Option ${selection.id} expects a boolean value.`); + continue; + } + if ( + descriptor.type === "select" && + !descriptor.options.some((choice) => choice.id === selection.value) + ) { + const choices = descriptor.options.map((choice) => choice.id).join(", "); + problems.push(`Option ${selection.id} must be one of: ${choices}.`); + } + } + return problems; +} + +function selectionSummary( + projection: OrchestrationV2ThreadProjection, + providers: ReadonlyArray, +): ConversationConfigurationSelection { + const selection = projection.thread.modelSelection; + const provider = providers.find((candidate) => candidate.instanceId === selection.instanceId); + const session = projection.providerSessions.find( + (candidate) => candidate.providerInstanceId === selection.instanceId, + ); + return { + providerInstanceId: selection.instanceId, + driverKind: provider?.driver ?? session?.driver ?? null, + model: selection.model, + options: [...(selection.options ?? [])], + }; +} + +function stablePart(value: string): string { + return encodeURIComponent(value); +} + +function stableCommandId(input: { + readonly scope: McpInvocationScope; + readonly threadId: ThreadId; + readonly requestKey: string; + readonly operation: string; +}): CommandId { + return CommandId.make( + [ + "command", + "mcp", + stablePart(input.scope.providerSessionId), + stablePart(input.threadId), + stablePart(input.operation), + stablePart(input.requestKey), + ].join(":"), + ); +} + +function receipt( + command: ConfigurationCommand, + acceptedCommandType: ConfigurationCommandType, + result: OrchestratorV2DispatchResult, +): ConversationConfigurationReceipt { + return { + commandId: command.commandId, + commandType: acceptedCommandType, + sequence: result.sequence, + eventIds: result.storedEvents.map((stored) => stored.event.id), + }; +} + +function requestedEffects( + result: OrchestratorV2DispatchResult, +): ConversationConfigurationChange["requestedEffects"] { + return result.storedEvents.some((stored) => stored.event.type === "provider-session.detached") + ? ["provider_session_detach"] + : []; +} + +function committedBehavior( + acceptedCommandType: ConfigurationCommandType, + result: OrchestratorV2DispatchResult, +): ConversationConfigurationBehavior { + if (result.storedEvents.length === 0) return "unchanged"; + if (acceptedCommandType === "provider.switch") return "handoff_required_next_turn"; + if (requestedEffects(result).length > 0) return "session_detach_requested"; + return "next_turn"; +} + +export const make = Effect.gen(function* () { + const crypto = yield* Crypto.Crypto; + const threads = yield* ThreadManagementService; + const providerRegistry = yield* ProviderRegistry; + const providerSwitch = yield* ProviderSwitchServiceV2; + const commandReceipts = yield* CommandReceiptStoreV2; + + const requireCapability = (scope: McpInvocationScope) => + scope.capabilities.has("orchestration") + ? Effect.void + : Effect.fail(failure("capability_denied", "This MCP credential cannot control threads.")); + + const loadScoped = Effect.fn("ConversationConfigurationMcpService.loadScoped")(function* ( + scope: McpInvocationScope, + requestedThreadId: ThreadId | undefined, + ) { + yield* requireCapability(scope); + const parent = yield* threads + .getThreadProjection(scope.threadId) + .pipe( + Effect.mapError((error) => + failure( + "orchestration_error", + `Unable to load the calling thread: ${errorMessage(error)}`, + ), + ), + ); + const threadId = requestedThreadId ?? scope.threadId; + const target = yield* threads + .getProjectThread({ projectId: parent.thread.projectId, threadId }) + .pipe( + Effect.mapError((error) => + isThreadNotFound(error) + ? failure( + "thread_not_found", + `Thread ${threadId} was not found in the calling project.`, + ) + : failure( + "orchestration_error", + `Unable to load thread ${threadId}: ${errorMessage(error)}`, + ), + ), + ); + return { parent, target }; + }); + + const configurationResult = ( + parent: OrchestrationV2ThreadProjection, + target: OrchestrationV2ThreadProjection, + providers: ReadonlyArray, + ): ConversationConfigurationResult => ({ + threadId: target.thread.id, + selection: selectionSummary(target, providers), + runtimeMode: target.thread.runtimeMode, + interactionMode: target.thread.interactionMode, + allowedRuntimeModes: runtimeModes.filter( + (mode) => runtimeModeRank(mode) <= runtimeModeRank(parent.thread.runtimeMode), + ), + allowedInteractionModes: interactionModes.filter( + (mode) => interactionModeRank(mode) <= interactionModeRank(parent.thread.interactionMode), + ), + providers: providers.map(providerSummary), + }); + + const resolveSelection = Effect.fn("ConversationConfigurationMcpService.resolveSelection")( + function* ( + current: ModelSelection, + input: ConversationConfigureInput, + providers: ReadonlyArray, + ) { + const instanceId = input.providerInstanceId ?? current.instanceId; + const provider = providers.find((candidate) => candidate.instanceId === instanceId); + if (provider === undefined) { + return yield* failure( + "provider_unavailable", + `Provider instance ${instanceId} is unavailable.`, + ); + } + const constraints = providerConstraints(provider); + if (constraints.length > 0) { + return yield* failure( + "provider_unavailable", + `Provider instance ${instanceId} is unavailable: ${constraints.join(" ")}`, + ); + } + const sameProvider = instanceId === current.instanceId; + const model = + input.model ?? + (sameProvider + ? current.model + : (provider.models.find((candidate) => candidate.isDefault)?.slug ?? + provider.models[0]?.slug)); + if (model === undefined) { + return yield* failure( + "model_unavailable", + `Provider instance ${instanceId} did not advertise a default model.`, + ); + } + const advertisedModel = provider.models.find((candidate) => candidate.slug === model); + if (provider.models.length > 0 && advertisedModel === undefined) { + return yield* failure( + "model_unavailable", + `Model ${model} is not advertised by provider instance ${instanceId}.`, + ); + } + const sameModel = sameProvider && model === current.model; + const selectedOptions = + input.options === undefined ? (sameModel ? current.options : undefined) : input.options; + const optionProblems = + input.options === undefined && sameModel + ? [] + : invalidOptionSelections( + selectedOptions ?? [], + advertisedModel?.capabilities?.optionDescriptors, + ); + if (optionProblems.length > 0) { + return yield* failure("invalid_request", optionProblems.join(" ")); + } + return selectedOptions === undefined + ? ({ instanceId, model } satisfies ModelSelection) + : ({ instanceId, model, options: selectedOptions } satisfies ModelSelection); + }, + ); + + return ConversationConfigurationMcpService.of({ + read: (scope, input) => + Effect.gen(function* () { + const { parent, target } = yield* loadScoped(scope, input.threadId); + const providers = yield* providerRegistry.getProviders; + return configurationResult(parent, target, providers); + }), + configure: (scope, input) => + Effect.gen(function* () { + const { parent, target } = yield* loadScoped(scope, input.threadId); + const policyCeiling = { + callerThreadId: parent.thread.id, + runtimeMode: parent.thread.runtimeMode, + interactionMode: parent.thread.interactionMode, + } satisfies OrchestrationV2PolicyCeiling; + const selectionRequested = + input.providerInstanceId !== undefined || + input.model !== undefined || + input.options !== undefined; + const requestKey = input.clientRequestId ?? (yield* crypto.randomUUIDv4.pipe(Effect.orDie)); + const replayableRequest = input.clientRequestId !== undefined; + const selectionCommandId = selectionRequested + ? stableCommandId({ + scope, + threadId: target.thread.id, + requestKey, + operation: "thread-configure-selection", + }) + : undefined; + const runtimeCommandId = + input.runtimeMode === undefined + ? undefined + : stableCommandId({ + scope, + threadId: target.thread.id, + requestKey, + operation: "thread-configure-runtime-mode", + }); + const interactionCommandId = + input.interactionMode === undefined + ? undefined + : stableCommandId({ + scope, + threadId: target.thread.id, + requestKey, + operation: "thread-configure-interaction-mode", + }); + const acceptedReceipt = (commandId: CommandId | undefined, commandTypes: Set) => + commandId !== undefined && replayableRequest + ? commandReceipts.getByCommandId(commandId).pipe( + Effect.map( + Option.filter( + (existing) => + existing.status === "accepted" && + existing.threadId === target.thread.id && + commandTypes.has(existing.commandType), + ), + ), + Effect.mapError((error) => + failure( + "orchestration_error", + `Unable to inspect the configuration retry receipt: ${errorMessage(error)}`, + ), + ), + ) + : Effect.succeed(Option.none()); + const priorSelectionReceipt = yield* acceptedReceipt( + selectionCommandId, + new Set(["thread.model-selection.set", "provider.switch"]), + ); + const priorRuntimeReceipt = yield* acceptedReceipt( + runtimeCommandId, + new Set(["thread.runtime-mode.set"]), + ); + const priorInteractionReceipt = yield* acceptedReceipt( + interactionCommandId, + new Set(["thread.interaction-mode.set"]), + ); + const hasFreshSelection = selectionRequested && Option.isNone(priorSelectionReceipt); + const hasFreshRuntime = + input.runtimeMode !== undefined && Option.isNone(priorRuntimeReceipt); + const hasFreshInteraction = + input.interactionMode !== undefined && Option.isNone(priorInteractionReceipt); + if (hasFreshSelection || hasFreshRuntime || hasFreshInteraction) { + const prospectiveRuntimeMode = hasFreshRuntime + ? input.runtimeMode + : target.thread.runtimeMode; + const prospectiveInteractionMode = hasFreshInteraction + ? input.interactionMode + : target.thread.interactionMode; + if ( + prospectiveRuntimeMode !== undefined && + runtimeModeRank(prospectiveRuntimeMode) > runtimeModeRank(parent.thread.runtimeMode) + ) { + return yield* failure( + "runtime_mode_escalation_denied", + `Runtime mode ${prospectiveRuntimeMode} is broader than the calling thread's ${parent.thread.runtimeMode} ceiling.`, + ); + } + if ( + prospectiveInteractionMode !== undefined && + interactionModeRank(prospectiveInteractionMode) > + interactionModeRank(parent.thread.interactionMode) + ) { + return yield* failure( + "interaction_mode_escalation_denied", + `Interaction mode ${prospectiveInteractionMode} is broader than the calling thread's ${parent.thread.interactionMode} ceiling.`, + ); + } + } + const providers = yield* providerRegistry.getProviders; + const targetSelection = + selectionRequested && Option.isNone(priorSelectionReceipt) + ? yield* resolveSelection(target.thread.modelSelection, input, providers) + : target.thread.modelSelection; + const selectionChanged = !modelSelectionsEqual( + target.thread.modelSelection, + targetSelection, + ); + if (selectionChanged && Option.isNone(priorSelectionReceipt)) { + yield* providerSwitch + .plan({ projection: target, targetModelSelection: targetSelection }) + .pipe( + Effect.mapError((error) => + failure( + "provider_unavailable", + `The requested provider selection cannot be applied: ${errorMessage(error.cause ?? error)}`, + ), + ), + ); + } + + const pending: Array<{ + readonly setting: ConfigurationSetting; + readonly acceptedCommandType: ConfigurationCommandType; + readonly command: ConfigurationCommand; + }> = []; + const changes: Array = []; + if (selectionRequested) { + if (selectionCommandId === undefined) { + return yield* Effect.die(new Error("Selection command id was not initialized")); + } + const commandId = selectionCommandId; + const acceptedCommandType: SelectionCommandType = Option.match(priorSelectionReceipt, { + onNone: () => + targetSelection.instanceId === target.thread.modelSelection.instanceId + ? ("thread.model-selection.set" as const) + : ("provider.switch" as const), + onSome: (existing) => existing.commandType as SelectionCommandType, + }); + const command = { + type: acceptedCommandType, + commandId, + threadId: target.thread.id, + modelSelection: targetSelection, + policyCeiling, + ...(Option.isNone(priorSelectionReceipt) && + (input.providerInstanceId === undefined || + input.model === undefined || + input.options === undefined) + ? { expectedModelSelection: target.thread.modelSelection } + : {}), + } satisfies ConfigurationCommand; + if (!selectionChanged && !replayableRequest) { + changes.push({ + setting: "selection", + behavior: "unchanged", + requestedEffects: [], + receipt: null, + }); + } else { + pending.push({ + setting: "selection", + acceptedCommandType, + command, + }); + } + } + if (input.runtimeMode !== undefined) { + if (runtimeCommandId === undefined) { + return yield* Effect.die(new Error("Runtime command id was not initialized")); + } + const command = { + type: "thread.runtime-mode.set" as const, + commandId: runtimeCommandId, + threadId: target.thread.id, + runtimeMode: input.runtimeMode, + policyCeiling, + }; + const preflightUnchanged = + Option.isNone(priorRuntimeReceipt) && input.runtimeMode === target.thread.runtimeMode; + if (preflightUnchanged && !replayableRequest) { + changes.push({ + setting: "runtime_mode", + behavior: "unchanged", + requestedEffects: [], + receipt: null, + }); + } else { + pending.push({ + setting: "runtime_mode", + acceptedCommandType: "thread.runtime-mode.set", + command, + }); + } + } + if (input.interactionMode !== undefined) { + if (interactionCommandId === undefined) { + return yield* Effect.die(new Error("Interaction command id was not initialized")); + } + const command = { + type: "thread.interaction-mode.set" as const, + commandId: interactionCommandId, + threadId: target.thread.id, + interactionMode: input.interactionMode, + policyCeiling, + }; + const preflightUnchanged = + Option.isNone(priorInteractionReceipt) && + input.interactionMode === target.thread.interactionMode; + if (preflightUnchanged && !replayableRequest) { + changes.push({ + setting: "interaction_mode", + behavior: "unchanged", + requestedEffects: [], + receipt: null, + }); + } else { + pending.push({ + setting: "interaction_mode", + acceptedCommandType: "thread.interaction-mode.set", + command, + }); + } + } + + const targetAboveCallerCeiling = + runtimeModeRank(target.thread.runtimeMode) > runtimeModeRank(parent.thread.runtimeMode) || + interactionModeRank(target.thread.interactionMode) > + interactionModeRank(parent.thread.interactionMode); + const dispatchPriority = (change: (typeof pending)[number]) => { + if (!targetAboveCallerCeiling) return 0; + if ( + change.command.type === "thread.runtime-mode.set" && + runtimeModeRank(change.command.runtimeMode) < runtimeModeRank(target.thread.runtimeMode) + ) { + return -1; + } + if ( + change.command.type === "thread.interaction-mode.set" && + interactionModeRank(change.command.interactionMode) < + interactionModeRank(target.thread.interactionMode) + ) { + return -1; + } + return 0; + }; + const dispatchOrder = [...pending].sort( + (left, right) => dispatchPriority(left) - dispatchPriority(right), + ); + const errors: Array<{ setting: ConfigurationSetting; message: string }> = []; + for (const change of dispatchOrder) { + const dispatched = yield* Effect.result(threads.dispatch(change.command)); + if (Result.isFailure(dispatched)) { + errors.push({ setting: change.setting, message: errorMessage(dispatched.failure) }); + break; + } + const effects = requestedEffects(dispatched.success); + changes.push({ + setting: change.setting, + behavior: committedBehavior(change.acceptedCommandType, dispatched.success), + requestedEffects: effects, + receipt: receipt(change.command, change.acceptedCommandType, dispatched.success), + }); + } + const settingOrder: ReadonlyArray = [ + "selection", + "runtime_mode", + "interaction_mode", + ]; + const presentedChanges = [...changes].sort( + (left, right) => settingOrder.indexOf(left.setting) - settingOrder.indexOf(right.setting), + ); + const appliedCount = presentedChanges.filter((change) => change.receipt !== null).length; + const changedCount = presentedChanges.filter( + (change) => change.receipt !== null && change.behavior !== "unchanged", + ).length; + if (errors.length > 0 && appliedCount === 0) { + return yield* failure("orchestration_error", errors[0]!.message); + } + + const refreshed = yield* Effect.option(threads.getThreadProjection(target.thread.id)); + const projected = refreshed._tag === "Some" ? refreshed.value : target; + return { + threadId: target.thread.id, + outcome: + errors.length > 0 ? "partially_applied" : changedCount === 0 ? "unchanged" : "applied", + observation: refreshed._tag === "Some" ? "post_dispatch" : "pre_dispatch_fallback", + selection: selectionSummary(projected, providers), + runtimeMode: projected.thread.runtimeMode, + interactionMode: projected.thread.interactionMode, + activeRunIds: projected.runs.filter(isActiveRun).map((run) => run.id), + queuedRunIds: projected.runs + .filter((run) => run.status === "queued") + .map((run) => run.id), + changes: presentedChanges, + errors, + retryGuidance: { + sameClientRequestId: "replays_durable_decisions", + afterRejectedLeg: "use_new_client_request_id", + }, + } satisfies ConversationConfigureResult; + }), + }); +}); + +export const layer: Layer.Layer< + ConversationConfigurationMcpService, + never, + | Crypto.Crypto + | ThreadManagementService + | ProviderRegistry + | ProviderSwitchServiceV2 + | CommandReceiptStoreV2 +> = Layer.effect(ConversationConfigurationMcpService, make); diff --git a/apps/server/src/mcp/McpHttpServer.ts b/apps/server/src/mcp/McpHttpServer.ts index aeb47996e690..21923e152ec2 100644 --- a/apps/server/src/mcp/McpHttpServer.ts +++ b/apps/server/src/mcp/McpHttpServer.ts @@ -12,6 +12,7 @@ import { HttpRouter, HttpServerRequest, HttpServerResponse } from "effect/unstab import packageJson from "../../package.json" with { type: "json" }; import * as McpInvocationContext from "./McpInvocationContext.ts"; import * as OrchestratorMcpService from "./OrchestratorMcpService.ts"; +import * as ConversationConfigurationMcpService from "./ConversationConfigurationMcpService.ts"; import * as ThreadMetadataMcpService from "./ThreadMetadataMcpService.ts"; import * as McpSessionRegistry from "./McpSessionRegistry.ts"; import * as PreviewAutomationBroker from "./PreviewAutomationBroker.ts"; @@ -224,6 +225,7 @@ export const PreviewToolkitRegistrationLive = Layer.mergeAll( export const OrchestratorToolkitRegistrationLive = McpServer.toolkit(OrchestratorToolkit).pipe( Layer.provide(OrchestratorToolkitHandlersLive), + Layer.provide(ConversationConfigurationMcpService.layer), Layer.provide(OrchestratorMcpService.layer), Layer.provide(ThreadMetadataMcpService.layer), ); diff --git a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts index d9e06a5bacb6..afe5a933d827 100644 --- a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts @@ -2,6 +2,8 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; import { describe, expect, it } from "@effect/vitest"; import { CommandId, + ConversationConfigurationResult, + ConversationConfigureResult, EnvironmentId, IsoDateTime, MessageId, @@ -37,6 +39,7 @@ import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; +import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as PubSub from "effect/PubSub"; @@ -49,7 +52,10 @@ import { ClaudeProviderCapabilitiesV2 } from "../orchestration-v2/Adapters/Claud import { CodexProviderCapabilitiesV2 } from "../orchestration-v2/Adapters/CodexAdapterV2.ts"; import { CodexOrchestratorReplayHarness } from "../orchestration-v2/Adapters/CodexAdapterV2.testkit.ts"; import { OrchestratorV2, type OrchestratorV2Shape } from "../orchestration-v2/Orchestrator.ts"; -import { layer as threadManagementServiceLayer } from "../orchestration-v2/ThreadManagementService.ts"; +import { + layer as threadManagementServiceLayer, + ThreadManagementService, +} from "../orchestration-v2/ThreadManagementService.ts"; import { type ProviderAdapterV2Event, ProviderAdapterProtocolError, @@ -57,6 +63,7 @@ import { type ProviderAdapterV2TurnInput, } from "../orchestration-v2/ProviderAdapter.ts"; import { makeLayer as makeProviderAdapterRegistryLayer } from "../orchestration-v2/ProviderAdapterRegistry.ts"; +import { layer as providerSwitchServiceLayer } from "../orchestration-v2/ProviderSwitchService.ts"; import { type ProviderContinuationRequest, ProviderContinuationRequests, @@ -91,6 +98,8 @@ const queuedFollowupPrompt = "Complete the queued follow-up and return the final const queuedFollowupResult = "Queued delegated follow-up completed."; const decodeCreateThreadsResult = Schema.decodeUnknownEffect(OrchestratorMcpCreateThreadsResult); +const decodeConfigurationResult = Schema.decodeUnknownEffect(ConversationConfigurationResult); +const decodeConfigureResult = Schema.decodeUnknownEffect(ConversationConfigureResult); const decodeCreatedThread = Schema.decodeUnknownEffect(OrchestratorMcpCreatedThread); const decodeDelegateTaskResult = Schema.decodeUnknownEffect(OrchestratorMcpDelegateTaskResult); const decodeTaskCancelResult = Schema.decodeUnknownEffect(OrchestratorMcpTaskCancelResult); @@ -177,13 +186,17 @@ function makeDeterministicAdapter(input: { readonly capturedTurns: Ref.Ref>; readonly shouldComplete: (turn: ProviderAdapterV2TurnInput) => boolean; readonly terminalGate?: (turn: ProviderAdapterV2TurnInput) => Deferred.Deferred | undefined; + readonly selectionPlanGate?: Effect.Effect; readonly response: (turn: ProviderAdapterV2TurnInput) => string; }): ProviderAdapterV2Shape { return { instanceId: input.instanceId, driver: input.driver, getCapabilities: () => Effect.succeed(input.capabilities), - planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }), + planSelectionTransition: () => + input.selectionPlanGate === undefined + ? Effect.succeed({ type: "apply_on_next_turn" }) + : input.selectionPlanGate.pipe(Effect.as({ type: "apply_on_next_turn" as const })), openSession: (sessionInput) => Effect.gen(function* () { const events = yield* PubSub.unbounded(); @@ -473,6 +486,13 @@ describe("orchestrator MCP toolkit", () => { Effect.gen(function* () { const cwd = yield* checkpointWorkspace("orchestrator-mcp-toolkit"); const capturedTurns = yield* Ref.make>([]); + const selectionPlanArmed = yield* Ref.make(false); + const selectionPlanEntered = yield* Deferred.make(); + const selectionPlanRelease = yield* Deferred.make(); + const runtimeDispatchEntered = yield* Deferred.make(); + const runtimeDispatchRelease = yield* Deferred.make(); + const callerCeilingDispatchEntered = yield* Deferred.make(); + const callerCeilingDispatchRelease = yield* Deferred.make(); const parentTerminalGates = new Map>(); const deliveryTerminalGates = new Map>(); const registryLayer = makeProviderAdapterRegistryLayer([ @@ -481,6 +501,11 @@ describe("orchestrator MCP toolkit", () => { driver: ProviderDriverKind.make("codex"), capabilities: CodexProviderCapabilitiesV2, capturedTurns, + selectionPlanGate: Effect.gen(function* () { + if (!(yield* Ref.getAndSet(selectionPlanArmed, false))) return; + yield* Deferred.succeed(selectionPlanEntered, undefined); + yield* Deferred.await(selectionPlanRelease); + }), shouldComplete: (turn) => turn.threadId !== parentThreadId && turn.message.text !== cancellationPrompt, terminalGate: (turn) => @@ -554,10 +579,33 @@ describe("orchestrator MCP toolkit", () => { }, registryLayer, ).pipe(Layer.provide(continuationProbeLayer)); - const orchestrationLayer = Layer.merge( - orchestratorLayer, - threadManagementServiceLayer.pipe(Layer.provide(orchestratorLayer)), + const baseThreadManagementLayer = threadManagementServiceLayer.pipe( + Layer.provide(orchestratorLayer), ); + const gatedThreadManagementLayer = Layer.effect( + ThreadManagementService, + Effect.gen(function* () { + const service = yield* ThreadManagementService; + return ThreadManagementService.of({ + ...service, + dispatch: (command) => + command.type === "thread.runtime-mode.set" && + command.commandId.includes("configuration-preflight-equal-race") + ? Deferred.succeed(runtimeDispatchEntered, undefined).pipe( + Effect.andThen(Deferred.await(runtimeDispatchRelease)), + Effect.andThen(service.dispatch(command)), + ) + : command.type === "thread.interaction-mode.set" && + command.commandId.includes("configuration-caller-ceiling-race") + ? Deferred.succeed(callerCeilingDispatchEntered, undefined).pipe( + Effect.andThen(Deferred.await(callerCeilingDispatchRelease)), + Effect.andThen(service.dispatch(command)), + ) + : service.dispatch(command), + }); + }), + ).pipe(Layer.provide(baseThreadManagementLayer)); + const orchestrationLayer = Layer.merge(orchestratorLayer, gatedThreadManagementLayer); const providerRegistryLayer = makeProviderRegistryLayer([ makeProviderSnapshot({ instanceId: codexInstanceId, @@ -616,6 +664,7 @@ describe("orchestrator MCP toolkit", () => { const testLayer = McpHttpServer.OrchestratorToolkitRegistrationLive.pipe( Layer.provideMerge(McpServer.McpServer.layer), Layer.provideMerge(orchestrationLayer), + Layer.provide(providerSwitchServiceLayer.pipe(Layer.provide(registryLayer))), Layer.provide(providerRegistryLayer), Layer.provide(scheduledTaskStubLayer), Layer.provide(NodeServices.layer), @@ -1223,6 +1272,28 @@ describe("orchestrator MCP toolkit", () => { ({ tool }) => tool.name === "t3_thread_interrupt", ); expect(threadInterruptTool?.tool.annotations?.destructiveHint).toBe(true); + const threadConfigurationTool = server.tools.find( + ({ tool }) => tool.name === "t3_thread_configuration", + ); + expect(threadConfigurationTool?.tool.annotations?.readOnlyHint).toBe(true); + expect(threadConfigurationTool?.tool.inputSchema).toMatchObject({ + type: "object", + properties: { threadId: expect.any(Object) }, + }); + const threadConfigureTool = server.tools.find( + ({ tool }) => tool.name === "t3_thread_configure", + ); + expect(threadConfigureTool?.tool.annotations?.destructiveHint).toBe(true); + expect(threadConfigureTool?.tool.inputSchema).toMatchObject({ + type: "object", + properties: { + providerInstanceId: expect.any(Object), + model: expect.any(Object), + options: expect.any(Object), + runtimeMode: expect.any(Object), + interactionMode: expect.any(Object), + }, + }); const capabilities = yield* invoke("orchestrator_capabilities", {}); expect(capabilities.isError).toBe(false); @@ -2876,6 +2947,475 @@ describe("orchestrator MCP toolkit", () => { removedDelivery.subagents.find((task) => task.id === removeTask.id), ).toMatchObject({ result: expect.any(String), status: "completed" }); yield* expectOffersToStay(0); + + const configurationBefore = yield* invoke("t3_thread_configuration", { + threadId: lateParentThreadId, + }); + expect(configurationBefore.isError).toBe(false); + const decodedConfigurationBefore = yield* decodeConfigurationResult( + configurationBefore.structuredContent, + ); + expect(decodedConfigurationBefore.selection).toMatchObject({ + providerInstanceId: codexInstanceId, + model: codexModel, + options: [], + }); + expect(decodedConfigurationBefore.allowedRuntimeModes).toContain("full-access"); + expect(decodedConfigurationBefore.providers[0]?.models[0]?.options).toEqual( + expect.arrayContaining([expect.objectContaining({ id: "reasoning" })]), + ); + + const invalidConfiguration = yield* invoke("t3_thread_configure", { + threadId: lateParentThreadId, + model: "missing-model", + runtimeMode: "approval-required", + clientRequestId: "configuration-invalid-before-dispatch", + }); + expect(invalidConfiguration.structuredContent).toMatchObject({ + _tag: "OrchestratorMcpFailure", + code: "model_unavailable", + }); + expect( + (yield* orchestrator.getThreadProjection(lateParentThreadId)).thread, + ).toMatchObject({ runtimeMode: "full-access", interactionMode: "default" }); + + const configureInput = { + threadId: lateParentThreadId, + options: [{ id: "reasoning", value: "high" }], + runtimeMode: "approval-required", + interactionMode: "plan", + clientRequestId: "configuration-restart-retry", + } as const; + const configuredCall = yield* invoke("t3_thread_configure", configureInput); + expect(configuredCall.isError).toBe(false); + const configured = yield* decodeConfigureResult(configuredCall.structuredContent); + expect(configured.outcome).toBe("applied"); + expect(configured.selection.options).toEqual([{ id: "reasoning", value: "high" }]); + expect(configured.runtimeMode).toBe("approval-required"); + expect(configured.interactionMode).toBe("plan"); + expect(configured.activeRunIds).toContain(removeParentRun.id); + expect(configured.changes).toEqual( + expect.arrayContaining([ + expect.objectContaining({ setting: "selection", behavior: "next_turn" }), + expect.objectContaining({ + setting: "runtime_mode", + behavior: "next_turn", + }), + expect.objectContaining({ setting: "interaction_mode", behavior: "next_turn" }), + ]), + ); + expect(configured.changes.every((change) => change.receipt !== null)).toBe(true); + const committedConfiguration = + yield* orchestrator.getThreadProjection(lateParentThreadId); + expect(committedConfiguration.thread.modelSelection.options).toEqual([ + { id: "reasoning", value: "high" }, + ]); + expect(committedConfiguration.thread.runtimeMode).toBe("approval-required"); + expect(committedConfiguration.thread.interactionMode).toBe("plan"); + const repeatedConfiguredCall = yield* invoke("t3_thread_configure", configureInput); + expect(repeatedConfiguredCall.isError).toBe(false); + const repeatedConfigured = yield* decodeConfigureResult( + repeatedConfiguredCall.structuredContent, + ); + expect( + repeatedConfigured.changes.map((change) => change.receipt?.commandId ?? null), + ).toEqual(configured.changes.map((change) => change.receipt?.commandId ?? null)); + expect( + repeatedConfigured.changes.map((change) => change.receipt?.eventIds ?? []), + ).toEqual(configured.changes.map((change) => change.receipt?.eventIds ?? [])); + + const reversedCall = yield* invoke("t3_thread_configure", { + threadId: lateParentThreadId, + options: [], + runtimeMode: "full-access", + interactionMode: "default", + clientRequestId: "configuration-reverse", + }); + expect(reversedCall.isError).toBe(false); + const reversed = yield* decodeConfigureResult(reversedCall.structuredContent); + expect(reversed.selection.options).toEqual([]); + expect(reversed.runtimeMode).toBe("full-access"); + expect(reversed.interactionMode).toBe("default"); + + const claudeChildBeforeNoop = yield* orchestrator.getThreadProjection( + delegated.childThreadId, + ); + expect( + claudeChildBeforeNoop.providerSessions.some( + (session) => + session.status !== "stopped" && + session.status !== "error" && + !session.capabilities.sessions.supportsRuntimeModeSwitchInSession, + ), + ).toBe(true); + const unchangedRuntimeCall = yield* invoke("t3_thread_configure", { + threadId: delegated.childThreadId, + runtimeMode: claudeChildBeforeNoop.thread.runtimeMode, + clientRequestId: "configuration-fresh-unchanged-claude-runtime", + }); + const unchangedRuntime = yield* decodeConfigureResult( + unchangedRuntimeCall.structuredContent, + ); + expect(unchangedRuntime).toMatchObject({ + outcome: "unchanged", + changes: [ + { + setting: "runtime_mode", + behavior: "unchanged", + requestedEffects: [], + receipt: expect.objectContaining({ commandType: "thread.runtime-mode.set" }), + }, + ], + }); + const claudeChildAfterNoop = yield* orchestrator.getThreadProjection( + delegated.childThreadId, + ); + expect(claudeChildAfterNoop.providerSessions).toEqual( + claudeChildBeforeNoop.providerSessions, + ); + const directRuntimeNoop = yield* orchestrator.dispatch({ + type: "thread.runtime-mode.set", + commandId: CommandId.make("command:mcp-configuration:direct-runtime-noop"), + threadId: delegated.childThreadId, + runtimeMode: claudeChildAfterNoop.thread.runtimeMode, + }); + expect(directRuntimeNoop.storedEvents).toEqual([]); + expect( + (yield* orchestrator.getThreadProjection(delegated.childThreadId)).providerSessions, + ).toEqual(claudeChildAfterNoop.providerSessions); + + yield* invoke("t3_thread_configure", { + threadId: delegated.childThreadId, + runtimeMode: "approval-required", + clientRequestId: "configuration-change-after-durable-noop", + }); + const reopenedAfterNoopChangeCall = yield* invoke("t3_thread_send", { + threadId: delegated.childThreadId, + message: "Reopen after changing the field covered by a durable no-op.", + clientRequestId: "configuration-reopen-after-noop-change", + }); + const reopenedAfterNoopChange = yield* decodeThreadSendResult( + reopenedAfterNoopChangeCall.structuredContent, + ); + yield* invoke("t3_thread_wait", { + threadId: delegated.childThreadId, + runId: reopenedAfterNoopChange.runId, + timeoutMs: 10_000, + }); + const claudeBeforeNoopRetry = yield* orchestrator.getThreadProjection( + delegated.childThreadId, + ); + const repeatedNoopRuntime = yield* decodeConfigureResult( + (yield* invoke("t3_thread_configure", { + threadId: delegated.childThreadId, + runtimeMode: claudeChildBeforeNoop.thread.runtimeMode, + clientRequestId: "configuration-fresh-unchanged-claude-runtime", + })).structuredContent, + ); + expect(repeatedNoopRuntime.runtimeMode).toBe("approval-required"); + expect(repeatedNoopRuntime).toMatchObject({ + outcome: "unchanged", + changes: [ + { + setting: "runtime_mode", + behavior: "unchanged", + requestedEffects: [], + }, + ], + }); + expect(repeatedNoopRuntime.changes[0]?.receipt).toEqual( + unchangedRuntime.changes[0]?.receipt, + ); + expect( + (yield* orchestrator.getThreadProjection(delegated.childThreadId)).providerSessions, + ).toEqual(claudeBeforeNoopRetry.providerSessions); + + const preflightEqualRaceFiber = yield* Effect.forkChild( + invoke("t3_thread_configure", { + threadId: delegated.childThreadId, + runtimeMode: "approval-required", + clientRequestId: "configuration-preflight-equal-race", + }), + ); + yield* Deferred.await(runtimeDispatchEntered); + yield* orchestrator.dispatch({ + type: "thread.runtime-mode.set", + commandId: CommandId.make("command:mcp-configuration:concurrent-runtime-writer"), + threadId: delegated.childThreadId, + runtimeMode: "full-access", + }); + const reopenedDuringRaceCall = yield* invoke("t3_thread_send", { + threadId: delegated.childThreadId, + message: "Reopen before the preflight-equal configuration command commits.", + clientRequestId: "configuration-reopen-during-preflight-race", + }); + const reopenedDuringRace = yield* decodeThreadSendResult( + reopenedDuringRaceCall.structuredContent, + ); + yield* invoke("t3_thread_wait", { + threadId: delegated.childThreadId, + runId: reopenedDuringRace.runId, + timeoutMs: 10_000, + }); + yield* Deferred.succeed(runtimeDispatchRelease, undefined); + const preflightEqualRace = yield* decodeConfigureResult( + (yield* Fiber.join(preflightEqualRaceFiber)).structuredContent, + ); + expect(preflightEqualRace).toMatchObject({ + outcome: "applied", + runtimeMode: "approval-required", + changes: [ + { + setting: "runtime_mode", + behavior: "session_detach_requested", + requestedEffects: ["provider_session_detach"], + }, + ], + }); + + const reopenedAfterRaceCall = yield* invoke("t3_thread_send", { + threadId: delegated.childThreadId, + message: "Reopen before the next accepted runtime policy change.", + clientRequestId: "configuration-reopen-after-preflight-race", + }); + const reopenedAfterRace = yield* decodeThreadSendResult( + reopenedAfterRaceCall.structuredContent, + ); + yield* invoke("t3_thread_wait", { + threadId: delegated.childThreadId, + runId: reopenedAfterRace.runId, + timeoutMs: 10_000, + }); + + const acceptedClaudeRuntimeInput = { + threadId: delegated.childThreadId, + runtimeMode: "full-access", + clientRequestId: "configuration-accepted-claude-runtime", + } as const; + const acceptedClaudeRuntime = yield* decodeConfigureResult( + (yield* invoke("t3_thread_configure", acceptedClaudeRuntimeInput)).structuredContent, + ); + expect(acceptedClaudeRuntime.changes).toEqual([ + expect.objectContaining({ + setting: "runtime_mode", + behavior: "session_detach_requested", + requestedEffects: ["provider_session_detach"], + }), + ]); + const reopenedClaudeCall = yield* invoke("t3_thread_send", { + threadId: delegated.childThreadId, + message: "Reopen the provider session after the runtime policy change.", + clientRequestId: "configuration-reopen-claude-session", + }); + const reopenedClaude = yield* decodeThreadSendResult( + reopenedClaudeCall.structuredContent, + ); + yield* invoke("t3_thread_wait", { + threadId: delegated.childThreadId, + runId: reopenedClaude.runId, + timeoutMs: 10_000, + }); + yield* invoke("t3_thread_configure", { + threadId: delegated.childThreadId, + interactionMode: "plan", + clientRequestId: "configuration-claude-interaction-after-runtime", + }); + const claudeBeforeAcceptedRetry = yield* orchestrator.getThreadProjection( + delegated.childThreadId, + ); + expect( + claudeBeforeAcceptedRetry.providerSessions.some( + (session) => session.status !== "stopped" && session.status !== "error", + ), + ).toBe(true); + const repeatedClaudeRuntime = yield* decodeConfigureResult( + (yield* invoke("t3_thread_configure", acceptedClaudeRuntimeInput)).structuredContent, + ); + expect(repeatedClaudeRuntime.changes).toEqual(acceptedClaudeRuntime.changes); + expect( + (yield* orchestrator.getThreadProjection(delegated.childThreadId)).providerSessions, + ).toEqual(claudeBeforeAcceptedRetry.providerSessions); + + const callerCeilingRaceFiber = yield* Effect.forkChild( + invoke("t3_thread_configure", { + threadId: delegated.childThreadId, + interactionMode: "default", + clientRequestId: "configuration-caller-ceiling-race", + }), + ); + yield* Deferred.await(callerCeilingDispatchEntered); + yield* orchestrator.dispatch({ + type: "thread.runtime-mode.set", + commandId: CommandId.make("command:mcp-configuration:caller-downgrade"), + threadId: parentThreadId, + runtimeMode: "approval-required", + }); + const acceptedRetryAfterCallerDowngrade = yield* decodeConfigureResult( + (yield* invoke("t3_thread_configure", acceptedClaudeRuntimeInput)).structuredContent, + ); + expect(acceptedRetryAfterCallerDowngrade.changes).toEqual( + acceptedClaudeRuntime.changes, + ); + yield* Deferred.succeed(callerCeilingDispatchRelease, undefined); + const callerCeilingRace = yield* Fiber.join(callerCeilingRaceFiber); + expect(callerCeilingRace.structuredContent).toMatchObject({ + _tag: "OrchestratorMcpFailure", + code: "orchestration_error", + }); + expect( + (yield* orchestrator.getThreadProjection(delegated.childThreadId)).thread, + ).toMatchObject({ runtimeMode: "full-access", interactionMode: "plan" }); + yield* orchestrator.dispatch({ + type: "thread.runtime-mode.set", + commandId: CommandId.make("command:mcp-configuration:caller-restore"), + threadId: parentThreadId, + runtimeMode: "full-access", + }); + + yield* orchestrator.dispatch({ + type: "thread.runtime-mode.set", + commandId: CommandId.make("command:mcp-configuration:caller-restrict-runtime"), + threadId: parentThreadId, + runtimeMode: "approval-required", + }); + yield* orchestrator.dispatch({ + type: "thread.interaction-mode.set", + commandId: CommandId.make("command:mcp-configuration:caller-restrict-interaction"), + threadId: parentThreadId, + interactionMode: "plan", + }); + const restrictiveCombinedInput = { + threadId: lateParentThreadId, + options: [{ id: "reasoning", value: "high" }], + runtimeMode: "approval-required", + interactionMode: "plan", + clientRequestId: "configuration-restrict-both-before-selection", + } as const; + const restrictiveCombined = yield* decodeConfigureResult( + (yield* invoke("t3_thread_configure", restrictiveCombinedInput)).structuredContent, + ); + expect(restrictiveCombined.outcome).toBe("applied"); + expect(restrictiveCombined.changes.map((change) => change.setting)).toEqual([ + "selection", + "runtime_mode", + "interaction_mode", + ]); + const restrictiveReceiptSequence = (setting: string) => + restrictiveCombined.changes.find((change) => change.setting === setting)?.receipt + ?.sequence ?? Number.MAX_SAFE_INTEGER; + expect(restrictiveReceiptSequence("runtime_mode")).toBeLessThan( + restrictiveReceiptSequence("interaction_mode"), + ); + expect(restrictiveReceiptSequence("interaction_mode")).toBeLessThan( + restrictiveReceiptSequence("selection"), + ); + expect( + (yield* orchestrator.getThreadProjection(lateParentThreadId)).thread, + ).toMatchObject({ + runtimeMode: "approval-required", + interactionMode: "plan", + modelSelection: expect.objectContaining({ + options: [{ id: "reasoning", value: "high" }], + }), + }); + const restrictiveCombinedRetry = yield* decodeConfigureResult( + (yield* invoke("t3_thread_configure", restrictiveCombinedInput)).structuredContent, + ); + expect( + restrictiveCombinedRetry.changes.map((change) => change.receipt?.commandId), + ).toEqual(restrictiveCombined.changes.map((change) => change.receipt?.commandId)); + yield* orchestrator.dispatch({ + type: "thread.runtime-mode.set", + commandId: CommandId.make( + "command:mcp-configuration:caller-restore-after-restrictions", + ), + threadId: parentThreadId, + runtimeMode: "full-access", + }); + yield* orchestrator.dispatch({ + type: "thread.interaction-mode.set", + commandId: CommandId.make( + "command:mcp-configuration:caller-interaction-restore-after-restrictions", + ), + threadId: parentThreadId, + interactionMode: "default", + }); + yield* invoke("t3_thread_configure", { + threadId: lateParentThreadId, + options: [], + runtimeMode: "full-access", + interactionMode: "default", + clientRequestId: "configuration-target-restore-after-restrictions", + }); + + yield* Ref.set(selectionPlanArmed, true); + const stalePartialFiber = yield* Effect.forkChild( + invoke("t3_thread_configure", { + threadId: lateParentThreadId, + options: [{ id: "reasoning", value: "high" }], + clientRequestId: "configuration-stale-partial-selection", + }), + ); + yield* Deferred.await(selectionPlanEntered); + const concurrentSelection = { + instanceId: codexInstanceId, + model: "gpt-5.5-concurrent", + } satisfies ModelSelection; + yield* orchestrator.dispatch({ + type: "thread.model-selection.set", + commandId: CommandId.make("command:mcp-configuration:concurrent-selection"), + threadId: lateParentThreadId, + modelSelection: concurrentSelection, + }); + yield* Deferred.succeed(selectionPlanRelease, undefined); + const stalePartialCall = yield* Fiber.join(stalePartialFiber); + expect(stalePartialCall.structuredContent).toMatchObject({ + _tag: "OrchestratorMcpFailure", + code: "orchestration_error", + }); + expect( + (yield* orchestrator.getThreadProjection(lateParentThreadId)).thread.modelSelection, + ).toEqual(concurrentSelection); + + const crossProviderInput = { + threadId: lateParentThreadId, + providerInstanceId: claudeInstanceId, + model: claudeModel, + options: [], + clientRequestId: "configuration-cross-provider-retry", + } as const; + const crossProviderCall = yield* invoke("t3_thread_configure", crossProviderInput); + const crossProvider = yield* decodeConfigureResult(crossProviderCall.structuredContent); + expect(crossProvider.selection).toMatchObject({ + providerInstanceId: claudeInstanceId, + model: claudeModel, + options: [], + }); + expect(crossProvider.changes).toEqual([ + expect.objectContaining({ + setting: "selection", + behavior: "handoff_required_next_turn", + requestedEffects: ["provider_session_detach"], + receipt: expect.objectContaining({ commandType: "provider.switch" }), + }), + ]); + yield* orchestrator.dispatch({ + type: "provider.switch", + commandId: CommandId.make( + "command:mcp-configuration:selection-after-accepted-switch", + ), + threadId: lateParentThreadId, + modelSelection: codexSelection, + }); + const repeatedCrossProviderCall = yield* invoke( + "t3_thread_configure", + crossProviderInput, + ); + const repeatedCrossProvider = yield* decodeConfigureResult( + repeatedCrossProviderCall.structuredContent, + ); + expect(repeatedCrossProvider.changes).toEqual(crossProvider.changes); + expect(repeatedCrossProvider.selection.providerInstanceId).toBe(codexInstanceId); }).pipe(Effect.provide(testLayer)); }), ), diff --git a/apps/server/src/mcp/toolkits/orchestrator/handlers.ts b/apps/server/src/mcp/toolkits/orchestrator/handlers.ts index c72ba2da07a2..4d865a5fac96 100644 --- a/apps/server/src/mcp/toolkits/orchestrator/handlers.ts +++ b/apps/server/src/mcp/toolkits/orchestrator/handlers.ts @@ -3,6 +3,7 @@ import * as Effect from "effect/Effect"; import { McpInvocationContext } from "../../McpInvocationContext.ts"; import { OrchestratorMcpService } from "../../OrchestratorMcpService.ts"; +import { ConversationConfigurationMcpService } from "../../ConversationConfigurationMcpService.ts"; import { ThreadMetadataMcpService } from "../../ThreadMetadataMcpService.ts"; const handlers = { @@ -116,6 +117,18 @@ const handlers = { const service = yield* OrchestratorMcpService; return yield* service.interruptThread(scope, input); }), + t3_thread_configuration: (input) => + Effect.gen(function* () { + const scope = yield* McpInvocationContext; + const service = yield* ConversationConfigurationMcpService; + return yield* service.read(scope, input); + }), + t3_thread_configure: (input) => + Effect.gen(function* () { + const scope = yield* McpInvocationContext; + const service = yield* ConversationConfigurationMcpService; + return yield* service.configure(scope, input); + }), } satisfies Parameters[0]; export const OrchestratorToolkitHandlersLive = OrchestratorToolkit.toLayer(handlers); diff --git a/apps/server/src/mcp/toolkits/orchestrator/tools.test.ts b/apps/server/src/mcp/toolkits/orchestrator/tools.test.ts index de85b6f72e0c..d960ed1184cf 100644 --- a/apps/server/src/mcp/toolkits/orchestrator/tools.test.ts +++ b/apps/server/src/mcp/toolkits/orchestrator/tools.test.ts @@ -5,6 +5,8 @@ import { CreateThreadsTool, DelegateTaskTool, ScheduleTaskTool, + ThreadConfigurationTool, + ThreadConfigureTool, ThreadUpdateTool, } from "./tools.ts"; @@ -77,4 +79,28 @@ describe("orchestrator MCP tool guidance", () => { ]); assert.include(ThreadUpdateTool.description ?? "", "Workspace and branch changes"); }); + + it("publishes discoverable root objects for conversation configuration", () => { + const readSchema = Tool.getJsonSchema(ThreadConfigurationTool) as { + readonly type?: unknown; + readonly properties?: Readonly>; + }; + const configureSchema = Tool.getJsonSchema(ThreadConfigureTool) as { + readonly type?: unknown; + readonly properties?: Readonly>; + }; + + assert.equal(readSchema.type, "object"); + assert.hasAllKeys(readSchema.properties ?? {}, ["threadId"]); + assert.equal(configureSchema.type, "object"); + assert.hasAllKeys(configureSchema.properties ?? {}, [ + "threadId", + "providerInstanceId", + "model", + "options", + "runtimeMode", + "interactionMode", + "clientRequestId", + ]); + }); }); diff --git a/apps/server/src/mcp/toolkits/orchestrator/tools.ts b/apps/server/src/mcp/toolkits/orchestrator/tools.ts index 2a884c6de109..20133a74e435 100644 --- a/apps/server/src/mcp/toolkits/orchestrator/tools.ts +++ b/apps/server/src/mcp/toolkits/orchestrator/tools.ts @@ -1,4 +1,8 @@ import { + ConversationConfigurationInput, + ConversationConfigurationResult, + ConversationConfigureInput, + ConversationConfigureResult, OrchestratorMcpCapabilitiesResult, OrchestratorMcpCreatedThread, OrchestratorMcpCreateThreadsInput, @@ -32,6 +36,7 @@ import { import { Tool, Toolkit } from "effect/unstable/ai"; import * as McpInvocationContext from "../../McpInvocationContext.ts"; +import { ConversationConfigurationMcpService } from "../../ConversationConfigurationMcpService.ts"; import { OrchestratorMcpService } from "../../OrchestratorMcpService.ts"; import { ThreadMetadataMcpService } from "../../ThreadMetadataMcpService.ts"; @@ -40,6 +45,10 @@ const threadMetadataDependencies = [ McpInvocationContext.McpInvocationContext, ThreadMetadataMcpService, ]; +const configurationDependencies = [ + McpInvocationContext.McpInvocationContext, + ConversationConfigurationMcpService, +]; export const OrchestratorCapabilitiesTool = Tool.make("orchestrator_capabilities", { description: @@ -249,6 +258,32 @@ export const ThreadInterruptTool = Tool.make("t3_thread_interrupt", { .annotate(Tool.Title, "Interrupt a T3 thread") .annotate(Tool.Destructive, true); +export const ThreadConfigurationTool = Tool.make("t3_thread_configuration", { + description: + "Read the current provider instance, model, model options, runtime mode, and interaction mode for a T3 thread in the calling project. Omit threadId for this thread. The result lists provider models and option descriptors plus the runtime and interaction modes allowed by this caller's permission ceiling.", + parameters: ConversationConfigurationInput, + success: ConversationConfigurationResult, + failure: OrchestratorMcpFailure, + failureMode: "return", + dependencies: configurationDependencies, +}) + .annotate(Tool.Title, "Get thread configuration") + .annotate(Tool.Readonly, true) + .annotate(Tool.Destructive, false) + .annotate(Tool.Idempotent, true); + +export const ThreadConfigureTool = Tool.make("t3_thread_configure", { + description: + "Change an existing T3 thread's provider, model, full model-option selection, runtime mode, or interaction mode through V2 orchestration. Omit threadId for this thread. Read t3_thread_configuration first to choose advertised values. The calling thread's runtime and interaction modes are hard ceilings. The result separates committed settings from requested provider-session detaches and next-turn context-handoff requirements; a detach can interrupt active provider work. It also reports current active and queued runs, durable command receipts, and any partial failure. Reusing clientRequestId replays accepted or rejected decisions without reapplying accepted legs. After resolving the cause of a rejected leg, use a new clientRequestId for a new attempt.", + parameters: ConversationConfigureInput, + success: ConversationConfigureResult, + failure: OrchestratorMcpFailure, + failureMode: "return", + dependencies: configurationDependencies, +}) + .annotate(Tool.Title, "Configure a T3 thread") + .annotate(Tool.Destructive, true); + export const OrchestratorToolkit = Toolkit.make( OrchestratorCapabilitiesTool, DelegateTaskTool, @@ -266,4 +301,6 @@ export const OrchestratorToolkit = Toolkit.make( ThreadSendTool, ThreadWaitTool, ThreadInterruptTool, + ThreadConfigurationTool, + ThreadConfigureTool, ); diff --git a/apps/server/src/mcp/toolkits/worktree/registration.test.ts b/apps/server/src/mcp/toolkits/worktree/registration.test.ts index 3300a869fe67..4a436530afcc 100644 --- a/apps/server/src/mcp/toolkits/worktree/registration.test.ts +++ b/apps/server/src/mcp/toolkits/worktree/registration.test.ts @@ -4,12 +4,15 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; import { ProviderInstanceId, ThreadId } from "@t3tools/contracts"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; import * as Schema from "effect/Schema"; import { HttpBody, HttpClient, HttpRouter } from "effect/unstable/http"; import * as ServerEnvironment from "../../../environment/ServerEnvironment.ts"; import * as GitWorkflowService from "../../../git/GitWorkflowService.ts"; import { ThreadManagementService } from "../../../orchestration-v2/ThreadManagementService.ts"; +import { CommandReceiptStoreV2 } from "../../../orchestration-v2/CommandReceiptStore.ts"; +import { ProviderSwitchServiceV2 } from "../../../orchestration-v2/ProviderSwitchService.ts"; import * as ProjectService from "../../../project/ProjectService.ts"; import * as ProjectSetupScriptRunner from "../../../project/ProjectSetupScriptRunner.ts"; import { ProviderRegistry } from "../../../provider/Services/ProviderRegistry.ts"; @@ -22,6 +25,10 @@ import * as PreviewAutomationBroker from "../../PreviewAutomationBroker.ts"; const StubServicesLive = Layer.mergeAll( Layer.mock(ThreadManagementService)({}), + Layer.mock(CommandReceiptStoreV2)({ + getByCommandId: () => Effect.succeed(Option.none()), + }), + Layer.mock(ProviderSwitchServiceV2)({}), Layer.mock(ProviderRegistry)({}), Layer.mock(ScheduledTaskService)({}), Layer.mock(ProjectService.ProjectService)({}), diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index c0684ca2357c..ba3d733d685a 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -808,6 +808,7 @@ export const CLAUDE_READ_ONLY_T3_MCP_ALLOWED_TOOLS: ReadonlyArray = [ "mcp__t3-code__list_scheduled_tasks", "mcp__t3-code__t3_thread_list", "mcp__t3-code__t3_thread_wait", + "mcp__t3-code__t3_thread_configuration", ]; // The SDK's `allowedTools` only pre-approves tool calls; availability is the diff --git a/apps/server/src/orchestration-v2/EventSink.ts b/apps/server/src/orchestration-v2/EventSink.ts index bfa86d769bc8..c6b7f7f1535e 100644 --- a/apps/server/src/orchestration-v2/EventSink.ts +++ b/apps/server/src/orchestration-v2/EventSink.ts @@ -483,12 +483,9 @@ const baseLayer: Layer.Layer< commandId: input.commandId, events: normalized, }); - const sequence = storedEvents.at(-1)?.sequence; - if (sequence === undefined) { - return yield* Effect.die( - new Error(`Command ${input.commandId} produced no orchestration events.`), - ); - } + const sequence = + storedEvents.at(-1)?.sequence ?? + (yield* eventStore.latestSequence({ threadId: input.threadId })); yield* applyStoredEvents(storedEvents); yield* effectOutbox.enqueue(input.effects); const receipt: CommandReceiptV2 = { diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 88fe7598325f..f57b42f77c07 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -25,8 +25,10 @@ import { type OrchestrationV2ThreadProjection, type OrchestrationV2TurnItem, ProviderInstanceId, + type ProviderInteractionMode, type ProviderSessionId, RunId, + type RuntimeMode, ThreadId, } from "@t3tools/contracts"; import { modelSelectionsEqual } from "@t3tools/shared/model"; @@ -288,6 +290,42 @@ function commandThreadId(command: OrchestrationV2Command): ThreadId { } } +function runtimeModeRank(mode: RuntimeMode): number { + switch (mode) { + case "approval-required": + return 0; + case "auto-accept-edits": + return 1; + case "auto": + return 2; + case "full-access": + return 3; + } +} + +function interactionModeRank(mode: ProviderInteractionMode): number { + return mode === "plan" ? 0 : 1; +} + +function commandPolicyCeiling(command: OrchestrationV2Command) { + switch (command.type) { + case "thread.runtime-mode.set": + case "thread.interaction-mode.set": + case "thread.model-selection.set": + case "provider.switch": + return command.policyCeiling; + default: + return undefined; + } +} + +function dispatchLockKeys(command: OrchestrationV2Command): ReadonlyArray { + const policyCeiling = commandPolicyCeiling(command); + return [...new Set([commandThreadId(command), policyCeiling?.callerThreadId])] + .filter((threadId): threadId is ThreadId => threadId !== undefined) + .toSorted((left, right) => (left < right ? -1 : left > right ? 1 : 0)); +} + function pendingThreadTitleGenerationEffect( commandId: CommandId, threadId: ThreadId, @@ -1476,6 +1514,65 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio cause: `Thread ${command.threadId} is deleted.`, }); } + const policyCeiling = commandPolicyCeiling(command); + if (policyCeiling !== undefined) { + const callerProjection = + policyCeiling.callerThreadId === command.threadId + ? projection + : yield* projectionStore.getThreadProjection(policyCeiling.callerThreadId).pipe( + Effect.mapError( + (cause) => + new OrchestratorProjectionError({ + threadId: policyCeiling.callerThreadId, + cause, + }), + ), + ); + if (callerProjection.thread.deletedAt !== null) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: `Calling thread ${policyCeiling.callerThreadId} is deleted.`, + }); + } + const prospectiveRuntimeMode = + command.type === "thread.runtime-mode.set" ? command.runtimeMode : thread.runtimeMode; + const prospectiveInteractionMode = + command.type === "thread.interaction-mode.set" + ? command.interactionMode + : thread.interactionMode; + const isStrictlyRestrictiveModeChange = + (command.type === "thread.runtime-mode.set" && + runtimeModeRank(command.runtimeMode) < runtimeModeRank(thread.runtimeMode)) || + (command.type === "thread.interaction-mode.set" && + interactionModeRank(command.interactionMode) < + interactionModeRank(thread.interactionMode)); + if ( + !isStrictlyRestrictiveModeChange && + (runtimeModeRank(prospectiveRuntimeMode) > runtimeModeRank(policyCeiling.runtimeMode) || + runtimeModeRank(prospectiveRuntimeMode) > + runtimeModeRank(callerProjection.thread.runtimeMode)) + ) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: `Target runtime mode ${prospectiveRuntimeMode} exceeds the calling thread ceiling.`, + }); + } + if ( + !isStrictlyRestrictiveModeChange && + (interactionModeRank(prospectiveInteractionMode) > + interactionModeRank(policyCeiling.interactionMode) || + interactionModeRank(prospectiveInteractionMode) > + interactionModeRank(callerProjection.thread.interactionMode)) + ) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: `Target interaction mode ${prospectiveInteractionMode} exceeds the calling thread ceiling.`, + }); + } + } if ( command.type === "thread.metadata.update" && command.expectedWorktreePath !== undefined && @@ -1487,6 +1584,26 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio cause: `Thread ${command.threadId} worktree changed before the metadata update could be applied.`, }); } + if ( + (command.type === "thread.model-selection.set" || command.type === "provider.switch") && + command.expectedModelSelection !== undefined && + !modelSelectionsEqual(command.expectedModelSelection, thread.modelSelection) + ) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: `Thread ${command.threadId} model selection changed before the partial selection update could be applied.`, + }); + } + if ( + (command.type === "thread.runtime-mode.set" && command.runtimeMode === thread.runtimeMode) || + (command.type === "thread.interaction-mode.set" && + command.interactionMode === thread.interactionMode) || + ((command.type === "thread.model-selection.set" || command.type === "provider.switch") && + modelSelectionsEqual(command.modelSelection, thread.modelSelection)) + ) { + return; + } if (command.type === "thread.archive" && thread.archivedAt !== null) { return yield* new OrchestratorDispatchError({ commandId: command.commandId, @@ -1873,7 +1990,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio command.worktreePath !== undefined && command.worktreePath !== thread.worktreePath ? projection.providerSessions.map((session) => session.id) - : command.type === "thread.runtime-mode.set" + : command.type === "thread.runtime-mode.set" && command.runtimeMode !== thread.runtimeMode ? projection.providerSessions .filter( (session) => !session.capabilities.sessions.supportsRuntimeModeSwitchInSession, @@ -7106,7 +7223,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const plan = yield* dispatchOnce(command).pipe( Effect.flatMap((planned) => - planned.events.length > 0 + planned.events.length > 0 || + command.type === "thread.runtime-mode.set" || + command.type === "thread.interaction-mode.set" || + command.type === "thread.model-selection.set" || + command.type === "provider.switch" ? Effect.succeed(planned) : Effect.fail( new OrchestratorDispatchError({ @@ -7184,7 +7305,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }); const dispatchWithReceipt = (command: OrchestrationV2Command) => - threadDispatch.withLock(commandThreadId(command), dispatchWithReceiptEffect(command)); + dispatchLockKeys(command).reduceRight( + (effect, threadId) => threadDispatch.withLock(threadId, effect), + dispatchWithReceiptEffect(command), + ); const handleTerminalRun = (stored: OrchestrationV2StoredEvent) => Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/runtimeLayer.ts b/apps/server/src/orchestration-v2/runtimeLayer.ts index 637916e7f9bf..f6840520e3b6 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.ts @@ -262,7 +262,9 @@ const providerRuntimeRecoveryProvided = providerRuntimeRecoveryLayer.pipe( export const OrchestrationV2LayerLive = Layer.mergeAll( orchestratorProvided, + commandReceiptStoreProvided, threadManagementProvided, + providerSwitchServiceProvided, effectWorkerProvided, providerSessionManagerProvided, providerAuthServiceProvided, diff --git a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts index 73750176fd39..5b6279edc4ba 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts @@ -21,7 +21,10 @@ import { layer as checkpointCaptureServiceLayer } from "../CheckpointCaptureServ import { layer as checkpointServiceLayer } from "../CheckpointService.ts"; import { layer as checkpointRollbackServiceLayer } from "../CheckpointRollbackService.ts"; import { layer as commandPolicyLayer } from "../CommandPolicy.ts"; -import { layer as commandReceiptStoreLayer } from "../CommandReceiptStore.ts"; +import { + CommandReceiptStoreV2, + layer as commandReceiptStoreLayer, +} from "../CommandReceiptStore.ts"; import { layer as contextHandoffServiceLayer } from "../ContextHandoffService.ts"; import { layer as effectOutboxLayer } from "../EffectOutbox.ts"; import { @@ -221,7 +224,10 @@ export function makeOrchestratorV2ProviderReplayLayer< readonly runEffectWorker?: boolean; readonly replayGate?: ProviderReplayGate; } = {}, -): Layer.Layer { +): Layer.Layer< + OrchestratorV2 | CommandReceiptStoreV2, + Error | MigrationError | PlatformError.PlatformError | SqlError +> { const registryLayer = harness.makeProviderAdapterRegistryLayer( scenario.transcript, options.replayGate === undefined ? {} : { replayGate: options.replayGate }, @@ -240,7 +246,10 @@ export function makeOrchestratorV2ReplayLayerWithRegistry( readonly enableLegacyTokenStreaming?: boolean; readonly runEffectWorker?: boolean; } = {}, -): Layer.Layer { +): Layer.Layer< + OrchestratorV2 | CommandReceiptStoreV2, + Error | MigrationError | PlatformError.PlatformError | SqlError +> { const serverConfigLayer = Layer.effect( ServerConfig, makeReplayServerConfig(scenario.name).pipe(Effect.orDie), @@ -405,14 +414,20 @@ export function makeOrchestratorV2ReplayLayerWithRegistry( // orchestrator. Keeping this acquisition in the replay layer makes the // outbox lifecycle explicit and prevents test-only command-side draining. if (options.runEffectWorker === false) { - return orchestratorProvided.pipe(Layer.provide(NodeServices.layer)); + return Layer.merge( + orchestratorProvided.pipe(Layer.provide(NodeServices.layer)), + commandReceiptStoreProvided, + ); } - return Layer.effect( - OrchestratorV2, - Effect.gen(function* () { - const orchestrator = yield* OrchestratorV2; - yield* runEffectWorkerDaemon.pipe(Effect.forkScoped); - return orchestrator; - }), - ).pipe(Layer.provide(replayRuntime)); + return Layer.merge( + Layer.effect( + OrchestratorV2, + Effect.gen(function* () { + const orchestrator = yield* OrchestratorV2; + yield* runEffectWorkerDaemon.pipe(Effect.forkScoped); + return orchestrator; + }), + ).pipe(Layer.provide(replayRuntime)), + commandReceiptStoreProvided, + ); } diff --git a/docs/orchestration-v2/orchestrator-mcp-server.md b/docs/orchestration-v2/orchestrator-mcp-server.md index 215a350a2f04..4904b9b8986c 100644 --- a/docs/orchestration-v2/orchestrator-mcp-server.md +++ b/docs/orchestration-v2/orchestrator-mcp-server.md @@ -141,7 +141,7 @@ selection model-visible without allowing a request that cannot run. ## Tool Surface -The server exposes eleven orchestration tools. +The server exposes orchestration tools for delegation, thread management, scheduling, and conversation configuration. ### `orchestrator_capabilities` @@ -327,6 +327,49 @@ Without `runId`, it selects the newest interruptible run. A terminal run is returned unchanged, and a thread with no active provider turn returns `no_active_run`. +### `t3_thread_configuration` + +Reads one current-project thread's complete provider selection, including its +provider instance, model, and selected model options. It also returns runtime +and interaction modes, the caller-bounded allowed mode lists, and configured +providers with their advertised model option descriptors. Omitting `threadId` +targets the calling thread. + +### `t3_thread_configure` + +Validates a requested provider, model, full option replacement, runtime mode, +and interaction mode before dispatching the first command. Omitted options are +preserved when the provider and model stay the same. An explicit empty array +clears them, while changing provider or model starts without options unless the +request supplies a compatible replacement. + +Selection changes use `thread.model-selection.set` within a provider instance +and `provider.switch` across instances. Runtime and interaction changes use +their existing thread commands. The service asks `ProviderSwitchServiceV2` for +the exact selection transition before dispatch. The response normally reports +one post-dispatch configuration snapshot, active and queued run IDs, and a +receipt for every V2 command. If that optional read fails after a command is +accepted, `observation: pre_dispatch_fallback` marks the settings and run IDs +as the last pre-dispatch observation instead of inferring current state from +the request. A stored provider-session detach event is reported as a +requested effect because its worker can interrupt active shared-session work. +Cross-provider handoff is reported as required next-turn planning; the command +does not claim a context transfer was already queued or completed. + +Queued runs retain the model selection captured at creation. Their runtime and +interaction policies are resolved from the projected thread when execution +starts. + +All fields are validated before the first command. If a later command fails +after an earlier command committed, the tool returns `partially_applied` with +the committed receipts and the failed setting. Reusing `clientRequestId` +replays the same durable accepted or rejected decisions and never reapplies an +accepted leg. A rejected leg stays rejected under that key; after resolving its +cause, the caller must use a new `clientRequestId` to request another attempt. +The result repeats this policy in `retryGuidance`. Keyed requests record a +receipt even when the requested setting already matches, preventing a later +retry from turning that accepted no-op into a new mutation. + ## Delegated Task Lifecycle The MCP server is a command ingress into V2. It does not call provider adapters diff --git a/docs/user/agent-conversation-controls.md b/docs/user/agent-conversation-controls.md new file mode 100644 index 000000000000..9904bd02de96 --- /dev/null +++ b/docs/user/agent-conversation-controls.md @@ -0,0 +1,15 @@ +# Agent conversation controls + +Agents connected through T3 Code's built-in MCP server can inspect and change an existing conversation's provider settings. These controls use the same durable thread state as the web, desktop, and mobile clients. + +`t3_thread_configuration` returns the selected provider instance, model, model options, runtime mode, and interaction mode. It also lists the models and option values currently advertised by configured providers. When `threadId` is omitted, the tool reads the agent's current conversation. + +`t3_thread_configure` changes one or more of those settings. Omitted fields keep their current values, except that changing the provider while omitting `model` selects that provider's default or first advertised model. Passing an empty `options` array clears model options. A provider change does not copy options from the previous provider or model. + +Queued runs keep the model selection captured when they were created, but resolve runtime and interaction modes from the thread settings when they execute. A provider, model, or runtime change can request immediate provider-session detachment when the adapter cannot apply it in session; detaching a shared session can interrupt active provider work. The result lists the active and queued run IDs observed after the commands commit. + +Some providers can use the new selection on the next turn without detaching. Others require a provider-session detach, or a cross-provider context handoff that is planned when the next turn starts. The tool reports the committed command separately from requested detach effects and required next-turn handoff work; it does not claim that a replacement session or handoff already completed. + +An agent cannot use these tools to grant itself broader access. The calling conversation's runtime and interaction modes set the ceiling for every target conversation. A plan-mode caller cannot switch a target to default interaction mode, and an approval-required caller cannot select a broader runtime mode. + +Targets must belong to the calling conversation's project. Use `clientRequestId` for a mutation that may be retried. T3 Code derives stable V2 command IDs from it and records a receipt even when the requested setting is already selected, so a later retry cannot turn that accepted no-op into a new change. Results normally identify their settings and run IDs as a post-dispatch observation. If the final read is unavailable, the result explicitly labels them as a pre-dispatch fallback instead of presenting requested values as current state. diff --git a/packages/contracts/src/conversationControlMcp.test.ts b/packages/contracts/src/conversationControlMcp.test.ts new file mode 100644 index 000000000000..4417049dfb47 --- /dev/null +++ b/packages/contracts/src/conversationControlMcp.test.ts @@ -0,0 +1,30 @@ +import { expect, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Result from "effect/Result"; +import * as Schema from "effect/Schema"; + +import { ConversationConfigureInput } from "./conversationControlMcp.ts"; + +const decodeConfigureInput = Schema.decodeUnknownEffect(ConversationConfigureInput); + +it.effect("keeps omitted and explicitly empty model options distinct", () => + Effect.gen(function* () { + const preserved = yield* decodeConfigureInput({ model: "gpt-5.6-sol" }); + const cleared = yield* decodeConfigureInput({ options: [] }); + + expect(preserved.options).toBeUndefined(); + expect(cleared.options).toEqual([]); + }), +); + +it.effect("rejects empty updates and malformed Unicode request keys", () => + Effect.gen(function* () { + const empty = yield* Effect.result(decodeConfigureInput({})); + const malformed = yield* Effect.result( + decodeConfigureInput({ runtimeMode: "approval-required", clientRequestId: "retry-\ud800" }), + ); + + expect(Result.isFailure(empty)).toBe(true); + expect(Result.isFailure(malformed)).toBe(true); + }), +); diff --git a/packages/contracts/src/conversationControlMcp.ts b/packages/contracts/src/conversationControlMcp.ts new file mode 100644 index 000000000000..ae6fa7bc2f93 --- /dev/null +++ b/packages/contracts/src/conversationControlMcp.ts @@ -0,0 +1,175 @@ +import * as Schema from "effect/Schema"; + +import { + CommandId, + NonNegativeInt, + RunId, + ThreadId, + TrimmedNonEmptyString, +} from "./baseSchemas.ts"; +import { ProviderOptionDescriptor, ProviderOptionSelection } from "./model.ts"; +import { ProviderInteractionMode, RuntimeMode } from "./providerPolicy.ts"; +import { ProviderDriverKind, ProviderInstanceId } from "./providerInstance.ts"; + +function hasWellFormedUnicode(value: string): boolean { + for (let index = 0; index < value.length; index += 1) { + const code = value.charCodeAt(index); + if (code >= 0xd800 && code <= 0xdbff) { + if (index + 1 >= value.length) return false; + const next = value.charCodeAt(index + 1); + if (next < 0xdc00 || next > 0xdfff) return false; + index += 1; + } else if (code >= 0xdc00 && code <= 0xdfff) { + return false; + } + } + return true; +} + +const WellFormedRequestKey = Schema.String.check( + Schema.makeFilter( + (value: string) => + (value.length > 0 && value === value.trim()) || + "The request key must be non-empty and must not have surrounding whitespace.", + ), + Schema.isMaxLength(256), + Schema.makeFilter( + (value: string) => + hasWellFormedUnicode(value) || + "The request key must not contain unpaired Unicode surrogates.", + ), +).annotate({ description: "Stable idempotency key to reuse when retrying this mutation." }); + +export const ConversationConfigurationInput = Schema.Struct({ + threadId: Schema.optional( + ThreadId.annotate({ + description: "Target thread in the calling project; defaults to this thread.", + }), + ), +}); +export type ConversationConfigurationInput = typeof ConversationConfigurationInput.Type; + +export const ConversationConfigurationSelection = Schema.Struct({ + providerInstanceId: ProviderInstanceId, + driverKind: Schema.NullOr(ProviderDriverKind), + model: Schema.String, + options: Schema.Array(ProviderOptionSelection), +}); +export type ConversationConfigurationSelection = typeof ConversationConfigurationSelection.Type; + +export const ConversationConfigurationProvider = Schema.Struct({ + providerInstanceId: ProviderInstanceId, + driverKind: ProviderDriverKind, + displayName: Schema.NullOr(Schema.String), + selectable: Schema.Boolean, + constraints: Schema.Array(Schema.String), + models: Schema.Array( + Schema.Struct({ + id: Schema.String, + label: Schema.NullOr(Schema.String), + options: Schema.Array(ProviderOptionDescriptor), + }), + ), +}); +export type ConversationConfigurationProvider = typeof ConversationConfigurationProvider.Type; + +export const ConversationConfigurationResult = Schema.Struct({ + threadId: ThreadId, + selection: ConversationConfigurationSelection, + runtimeMode: RuntimeMode, + interactionMode: ProviderInteractionMode, + allowedRuntimeModes: Schema.Array(RuntimeMode), + allowedInteractionModes: Schema.Array(ProviderInteractionMode), + providers: Schema.Array(ConversationConfigurationProvider), +}); +export type ConversationConfigurationResult = typeof ConversationConfigurationResult.Type; + +export const ConversationConfigureInput = Schema.Struct({ + threadId: Schema.optional( + ThreadId.annotate({ + description: "Target thread in the calling project; defaults to this thread.", + }), + ), + providerInstanceId: Schema.optional( + ProviderInstanceId.annotate({ + description: "Configured provider instance id returned by t3_thread_configuration.", + }), + ), + model: Schema.optional( + TrimmedNonEmptyString.annotate({ + description: "Model id advertised for the selected provider instance.", + }), + ), + options: Schema.optional( + Schema.Array(ProviderOptionSelection).annotate({ + description: + "Full replacement option list for the selected model. Pass [] to clear all options.", + }), + ), + runtimeMode: Schema.optional(RuntimeMode), + interactionMode: Schema.optional(ProviderInteractionMode), + clientRequestId: Schema.optional(WellFormedRequestKey), +}).check( + Schema.makeFilter( + (input) => + input.providerInstanceId !== undefined || + input.model !== undefined || + input.options !== undefined || + input.runtimeMode !== undefined || + input.interactionMode !== undefined || + "Provide at least one selection, runtime mode, or interaction mode change.", + ), +); +export type ConversationConfigureInput = typeof ConversationConfigureInput.Type; + +export const ConversationConfigurationBehavior = Schema.Literals([ + "unchanged", + "next_turn", + "session_detach_requested", + "handoff_required_next_turn", +]); +export type ConversationConfigurationBehavior = typeof ConversationConfigurationBehavior.Type; + +export const ConversationConfigurationReceipt = Schema.Struct({ + commandId: CommandId, + commandType: Schema.Literals([ + "thread.model-selection.set", + "provider.switch", + "thread.runtime-mode.set", + "thread.interaction-mode.set", + ]), + sequence: NonNegativeInt, + eventIds: Schema.Array(Schema.String), +}); +export type ConversationConfigurationReceipt = typeof ConversationConfigurationReceipt.Type; + +export const ConversationConfigurationChange = Schema.Struct({ + setting: Schema.Literals(["selection", "runtime_mode", "interaction_mode"]), + behavior: ConversationConfigurationBehavior, + requestedEffects: Schema.Array(Schema.Literal("provider_session_detach")), + receipt: Schema.NullOr(ConversationConfigurationReceipt), +}); +export type ConversationConfigurationChange = typeof ConversationConfigurationChange.Type; + +export const ConversationConfigureResult = Schema.Struct({ + threadId: ThreadId, + outcome: Schema.Literals(["unchanged", "applied", "partially_applied"]), + observation: Schema.Literals(["post_dispatch", "pre_dispatch_fallback"]), + selection: ConversationConfigurationSelection, + runtimeMode: RuntimeMode, + interactionMode: ProviderInteractionMode, + activeRunIds: Schema.Array(RunId), + queuedRunIds: Schema.Array(RunId), + changes: Schema.Array(ConversationConfigurationChange), + errors: Schema.Array( + Schema.Struct({ + setting: Schema.Literals(["selection", "runtime_mode", "interaction_mode"]), + message: Schema.String, + }), + ), + retryGuidance: Schema.Struct({ + sameClientRequestId: Schema.Literal("replays_durable_decisions"), + afterRejectedLeg: Schema.Literal("use_new_client_request_id"), + }), +}); +export type ConversationConfigureResult = typeof ConversationConfigureResult.Type; diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index 092de0f48f26..7d779c857975 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -33,6 +33,7 @@ export * from "./orchestrationProject.ts"; export * from "./orchestrationV2.ts"; export * from "./applicationEvent.ts"; export * from "./orchestratorMcp.ts"; +export * from "./conversationControlMcp.ts"; export * from "./threadMetadataMcp.ts"; export * from "./orchestration.ts"; export * from "./t3ProjectFile.ts"; diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 541e65767b0b..c5de792795c8 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -73,6 +73,13 @@ const OrchestrationV2CreationFields = { creationSource: OrchestrationV2CreationSource, } as const; +export const OrchestrationV2PolicyCeiling = Schema.Struct({ + callerThreadId: ThreadId, + runtimeMode: RuntimeMode, + interactionMode: ProviderInteractionMode, +}); +export type OrchestrationV2PolicyCeiling = typeof OrchestrationV2PolicyCeiling.Type; + export const OrchestrationV2NativeRefStrength = Schema.Literals(["strong", "weak", "none"]); export type OrchestrationV2NativeRefStrength = typeof OrchestrationV2NativeRefStrength.Type; @@ -2155,18 +2162,23 @@ export const OrchestrationV2Command = Schema.Union([ commandId: CommandId, threadId: ThreadId, runtimeMode: RuntimeMode, + policyCeiling: Schema.optional(OrchestrationV2PolicyCeiling), }), Schema.Struct({ type: Schema.Literal("thread.interaction-mode.set"), commandId: CommandId, threadId: ThreadId, interactionMode: ProviderInteractionMode, + policyCeiling: Schema.optional(OrchestrationV2PolicyCeiling), }), Schema.Struct({ type: Schema.Literal("thread.model-selection.set"), commandId: CommandId, threadId: ThreadId, modelSelection: ModelSelection, + /** Reject this partial-selection write if the thread changed since it was resolved. */ + expectedModelSelection: Schema.optional(ModelSelection), + policyCeiling: Schema.optional(OrchestrationV2PolicyCeiling), }), Schema.Struct({ type: Schema.Literal("provider-session.detach"), @@ -2345,6 +2357,9 @@ export const OrchestrationV2Command = Schema.Union([ commandId: CommandId, threadId: ThreadId, modelSelection: ModelSelection, + /** Reject this partial-selection write if the thread changed since it was resolved. */ + expectedModelSelection: Schema.optional(ModelSelection), + policyCeiling: Schema.optional(OrchestrationV2PolicyCeiling), }), ]); export type OrchestrationV2Command = typeof OrchestrationV2Command.Type; diff --git a/packages/shared/src/t3McpToolPresentation.ts b/packages/shared/src/t3McpToolPresentation.ts index a9dcd685cc0c..c4f1b2d4e9c2 100644 --- a/packages/shared/src/t3McpToolPresentation.ts +++ b/packages/shared/src/t3McpToolPresentation.ts @@ -52,6 +52,8 @@ const T3_MCP_TOOLS: Record< t3_thread_send: { displayName: "Send to a T3 thread", summaryAction: "thread-send" }, t3_thread_wait: { displayName: "Wait for a T3 thread", summaryAction: "thread-wait" }, t3_thread_interrupt: { displayName: "Interrupt a T3 thread", summaryAction: "thread-interrupt" }, + t3_thread_configuration: { displayName: "Get thread configuration" }, + t3_thread_configure: { displayName: "Configure a T3 thread" }, t3_worktree_handoff: { displayName: "Hand off thread to a git worktree" }, t3_worktree_status: { displayName: "Get thread worktree status" }, preview_status: { displayName: "Get preview browser status" }, From 94bd5f99f29ed483ddcbd96402f20d54c38020bf Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 30 Aug 2026 10:37:39 -0700 Subject: [PATCH 2/6] fix(mcp): replay rejected configuration receipts --- ...onversationConfigurationMcpService.test.ts | 50 +++++++++++++++++++ .../ConversationConfigurationMcpService.ts | 9 ++-- ...OrchestratorMcpToolkit.integration.test.ts | 20 ++++++-- 3 files changed, 69 insertions(+), 10 deletions(-) diff --git a/apps/server/src/mcp/ConversationConfigurationMcpService.test.ts b/apps/server/src/mcp/ConversationConfigurationMcpService.test.ts index 221cb5841207..1954a2675ccd 100644 --- a/apps/server/src/mcp/ConversationConfigurationMcpService.test.ts +++ b/apps/server/src/mcp/ConversationConfigurationMcpService.test.ts @@ -181,6 +181,56 @@ describe("ConversationConfigurationMcpService", () => { }), ); + it.effect("replays a rejected selection before provider availability planning", () => + Effect.gen(function* () { + const parent = projection({ + threadId: parentThreadId, + runtimeMode: "full-access", + interactionMode: "default", + }); + const target = projection({ + threadId: targetThreadId, + runtimeMode: "full-access", + interactionMode: "default", + }); + const layer = testLayer({ + parent, + target, + providers: [], + getReceipt: () => + Effect.succeed( + Option.some({ + status: "rejected", + threadId: targetThreadId, + commandType: "provider.switch", + } as never), + ), + dispatch: (command) => + new OrchestratorCommandPreviouslyRejectedError({ + commandId: command.commandId, + commandType: command.type, + detail: "provider switch was rejected", + }), + }); + + const error = yield* Effect.gen(function* () { + const service = yield* ConversationConfiguration.ConversationConfigurationMcpService; + return yield* service + .configure(scope(), { + threadId: targetThreadId, + providerInstanceId: ProviderInstanceId.make("provider-now-unavailable"), + model: "unavailable-model", + options: [], + clientRequestId: "rejected-provider-switch", + }) + .pipe(Effect.flip); + }).pipe(Effect.provide(layer)); + + assert.equal(error.code, "orchestration_error"); + assert.match(error.message, /previously rejected/); + }), + ); + it.effect("marks a failed refresh and does not infer replayed inputs as current", () => Effect.gen(function* () { const parent = projection({ diff --git a/apps/server/src/mcp/ConversationConfigurationMcpService.ts b/apps/server/src/mcp/ConversationConfigurationMcpService.ts index ac719d1bd05f..599f634daa1e 100644 --- a/apps/server/src/mcp/ConversationConfigurationMcpService.ts +++ b/apps/server/src/mcp/ConversationConfigurationMcpService.ts @@ -422,13 +422,12 @@ export const make = Effect.gen(function* () { requestKey, operation: "thread-configure-interaction-mode", }); - const acceptedReceipt = (commandId: CommandId | undefined, commandTypes: Set) => + const durableReceipt = (commandId: CommandId | undefined, commandTypes: Set) => commandId !== undefined && replayableRequest ? commandReceipts.getByCommandId(commandId).pipe( Effect.map( Option.filter( (existing) => - existing.status === "accepted" && existing.threadId === target.thread.id && commandTypes.has(existing.commandType), ), @@ -441,15 +440,15 @@ export const make = Effect.gen(function* () { ), ) : Effect.succeed(Option.none()); - const priorSelectionReceipt = yield* acceptedReceipt( + const priorSelectionReceipt = yield* durableReceipt( selectionCommandId, new Set(["thread.model-selection.set", "provider.switch"]), ); - const priorRuntimeReceipt = yield* acceptedReceipt( + const priorRuntimeReceipt = yield* durableReceipt( runtimeCommandId, new Set(["thread.runtime-mode.set"]), ); - const priorInteractionReceipt = yield* acceptedReceipt( + const priorInteractionReceipt = yield* durableReceipt( interactionCommandId, new Set(["thread.interaction-mode.set"]), ); diff --git a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts index afe5a933d827..bcac8941f8d0 100644 --- a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts @@ -3236,12 +3236,13 @@ describe("orchestrator MCP toolkit", () => { (yield* orchestrator.getThreadProjection(delegated.childThreadId)).providerSessions, ).toEqual(claudeBeforeAcceptedRetry.providerSessions); + const callerCeilingRaceInput = { + threadId: delegated.childThreadId, + interactionMode: "default", + clientRequestId: "configuration-caller-ceiling-race", + } as const; const callerCeilingRaceFiber = yield* Effect.forkChild( - invoke("t3_thread_configure", { - threadId: delegated.childThreadId, - interactionMode: "default", - clientRequestId: "configuration-caller-ceiling-race", - }), + invoke("t3_thread_configure", callerCeilingRaceInput), ); yield* Deferred.await(callerCeilingDispatchEntered); yield* orchestrator.dispatch({ @@ -3262,6 +3263,15 @@ describe("orchestrator MCP toolkit", () => { _tag: "OrchestratorMcpFailure", code: "orchestration_error", }); + const rejectedCallerCeilingRetry = yield* invoke( + "t3_thread_configure", + callerCeilingRaceInput, + ); + expect(rejectedCallerCeilingRetry.structuredContent).toMatchObject({ + _tag: "OrchestratorMcpFailure", + code: "orchestration_error", + message: expect.stringContaining("previously rejected"), + }); expect( (yield* orchestrator.getThreadProjection(delegated.childThreadId)).thread, ).toMatchObject({ runtimeMode: "full-access", interactionMode: "plan" }); From 41dc41053257fa4bb6a71e2bc919f6b8b0ddcc64 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 30 Aug 2026 10:42:23 -0700 Subject: [PATCH 3/6] fix(mcp): prioritize rejected configuration replay --- ...onversationConfigurationMcpService.test.ts | 87 +++++++++++++++++-- .../ConversationConfigurationMcpService.ts | 60 ++++++++++--- 2 files changed, 125 insertions(+), 22 deletions(-) diff --git a/apps/server/src/mcp/ConversationConfigurationMcpService.test.ts b/apps/server/src/mcp/ConversationConfigurationMcpService.test.ts index 1954a2675ccd..36f01b2599c6 100644 --- a/apps/server/src/mcp/ConversationConfigurationMcpService.test.ts +++ b/apps/server/src/mcp/ConversationConfigurationMcpService.test.ts @@ -1,6 +1,7 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, describe, it } from "@effect/vitest"; import { + CommandId, EnvironmentId, ProviderDriverKind, ProviderInstanceId, @@ -8,12 +9,16 @@ import { type ServerProvider, ThreadId, } from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Ref from "effect/Ref"; -import { CommandReceiptStoreV2 } from "../orchestration-v2/CommandReceiptStore.ts"; +import { + type CommandReceiptV2, + CommandReceiptStoreV2, +} from "../orchestration-v2/CommandReceiptStore.ts"; import { OrchestratorCommandPreviouslyRejectedError } from "../orchestration-v2/Orchestrator.ts"; import { ProviderSwitchServiceV2 } from "../orchestration-v2/ProviderSwitchService.ts"; import { ThreadManagementService } from "../orchestration-v2/ThreadManagementService.ts"; @@ -87,6 +92,22 @@ const scope = (capabilities: ReadonlyArray<"orchestration"> = ["orchestration"]) issuedAt: 1, }) satisfies McpInvocationScope; +function rejectedReceipt( + commandId: CommandId, + commandType: "provider.switch" | "thread.runtime-mode.set" | "thread.interaction-mode.set", + resultSequence = 0, +): CommandReceiptV2 { + return { + commandId, + threadId: targetThreadId, + commandType, + acceptedAt: DateTime.makeUnsafe("2026-08-30T00:00:00.000Z"), + resultSequence, + status: "rejected", + error: `${commandType} was rejected`, + }; +} + function testLayer(input: { readonly parent: OrchestrationV2ThreadProjection; readonly target: OrchestrationV2ThreadProjection; @@ -197,14 +218,8 @@ describe("ConversationConfigurationMcpService", () => { parent, target, providers: [], - getReceipt: () => - Effect.succeed( - Option.some({ - status: "rejected", - threadId: targetThreadId, - commandType: "provider.switch", - } as never), - ), + getReceipt: (commandId) => + Effect.succeed(Option.some(rejectedReceipt(commandId, "provider.switch"))), dispatch: (command) => new OrchestratorCommandPreviouslyRejectedError({ commandId: command.commandId, @@ -231,6 +246,60 @@ describe("ConversationConfigurationMcpService", () => { }), ); + it.effect("replays a rejected leg before planning any unattempted later legs", () => + Effect.gen(function* () { + const dispatched = yield* Ref.make>([]); + const parent = projection({ + threadId: parentThreadId, + runtimeMode: "approval-required", + interactionMode: "plan", + }); + const target = projection({ + threadId: targetThreadId, + runtimeMode: "full-access", + interactionMode: "default", + }); + const layer = testLayer({ + parent, + target, + providers: [], + getReceipt: (commandId) => + String(commandId).includes("thread-configure-selection") + ? Effect.succeed(Option.some(rejectedReceipt(commandId, "provider.switch", 41))) + : Effect.succeed(Option.none()), + dispatch: (command) => + Ref.update(dispatched, (types) => [...types, command.type]).pipe( + Effect.andThen( + new OrchestratorCommandPreviouslyRejectedError({ + commandId: command.commandId, + commandType: command.type, + detail: "provider switch was rejected", + }), + ), + ), + }); + + const error = yield* Effect.gen(function* () { + const service = yield* ConversationConfiguration.ConversationConfigurationMcpService; + return yield* service + .configure(scope(), { + threadId: targetThreadId, + providerInstanceId: ProviderInstanceId.make("provider-now-unavailable"), + model: "unavailable-model", + options: [], + runtimeMode: "full-access", + interactionMode: "default", + clientRequestId: "rejected-selection-with-unattempted-legs", + }) + .pipe(Effect.flip); + }).pipe(Effect.provide(layer)); + + assert.equal(error.code, "orchestration_error"); + assert.match(error.message, /previously rejected/); + assert.deepEqual(yield* Ref.get(dispatched), ["provider.switch"]); + }), + ); + it.effect("marks a failed refresh and does not infer replayed inputs as current", () => Effect.gen(function* () { const parent = projection({ diff --git a/apps/server/src/mcp/ConversationConfigurationMcpService.ts b/apps/server/src/mcp/ConversationConfigurationMcpService.ts index 599f634daa1e..67d716bcdfeb 100644 --- a/apps/server/src/mcp/ConversationConfigurationMcpService.ts +++ b/apps/server/src/mcp/ConversationConfigurationMcpService.ts @@ -452,11 +452,21 @@ export const make = Effect.gen(function* () { interactionCommandId, new Set(["thread.interaction-mode.set"]), ); - const hasFreshSelection = selectionRequested && Option.isNone(priorSelectionReceipt); + const hasRejectedReceipt = [ + priorSelectionReceipt, + priorRuntimeReceipt, + priorInteractionReceipt, + ].some(Option.exists((existing) => existing.status === "rejected")); + const hasFreshSelection = + !hasRejectedReceipt && selectionRequested && Option.isNone(priorSelectionReceipt); const hasFreshRuntime = - input.runtimeMode !== undefined && Option.isNone(priorRuntimeReceipt); + !hasRejectedReceipt && + input.runtimeMode !== undefined && + Option.isNone(priorRuntimeReceipt); const hasFreshInteraction = - input.interactionMode !== undefined && Option.isNone(priorInteractionReceipt); + !hasRejectedReceipt && + input.interactionMode !== undefined && + Option.isNone(priorInteractionReceipt); if (hasFreshSelection || hasFreshRuntime || hasFreshInteraction) { const prospectiveRuntimeMode = hasFreshRuntime ? input.runtimeMode @@ -485,15 +495,14 @@ export const make = Effect.gen(function* () { } } const providers = yield* providerRegistry.getProviders; - const targetSelection = - selectionRequested && Option.isNone(priorSelectionReceipt) - ? yield* resolveSelection(target.thread.modelSelection, input, providers) - : target.thread.modelSelection; + const targetSelection = hasFreshSelection + ? yield* resolveSelection(target.thread.modelSelection, input, providers) + : target.thread.modelSelection; const selectionChanged = !modelSelectionsEqual( target.thread.modelSelection, targetSelection, ); - if (selectionChanged && Option.isNone(priorSelectionReceipt)) { + if (selectionChanged && hasFreshSelection) { yield* providerSwitch .plan({ projection: target, targetModelSelection: targetSelection }) .pipe( @@ -512,7 +521,7 @@ export const make = Effect.gen(function* () { readonly command: ConfigurationCommand; }> = []; const changes: Array = []; - if (selectionRequested) { + if (selectionRequested && (!hasRejectedReceipt || Option.isSome(priorSelectionReceipt))) { if (selectionCommandId === undefined) { return yield* Effect.die(new Error("Selection command id was not initialized")); } @@ -552,7 +561,10 @@ export const make = Effect.gen(function* () { }); } } - if (input.runtimeMode !== undefined) { + if ( + input.runtimeMode !== undefined && + (!hasRejectedReceipt || Option.isSome(priorRuntimeReceipt)) + ) { if (runtimeCommandId === undefined) { return yield* Effect.die(new Error("Runtime command id was not initialized")); } @@ -580,7 +592,10 @@ export const make = Effect.gen(function* () { }); } } - if (input.interactionMode !== undefined) { + if ( + input.interactionMode !== undefined && + (!hasRejectedReceipt || Option.isSome(priorInteractionReceipt)) + ) { if (interactionCommandId === undefined) { return yield* Effect.die(new Error("Interaction command id was not initialized")); } @@ -631,8 +646,27 @@ export const make = Effect.gen(function* () { } return 0; }; - const dispatchOrder = [...pending].sort( - (left, right) => dispatchPriority(left) - dispatchPriority(right), + const priorReceiptForSetting = (setting: ConfigurationSetting) => + setting === "selection" + ? priorSelectionReceipt + : setting === "runtime_mode" + ? priorRuntimeReceipt + : priorInteractionReceipt; + const dispatchOrder = [...pending].sort((left, right) => + hasRejectedReceipt + ? Option.getOrElse( + Option.map(priorReceiptForSetting(left.setting), (receipt) => + Number(receipt.resultSequence), + ), + () => Number.MAX_SAFE_INTEGER, + ) - + Option.getOrElse( + Option.map(priorReceiptForSetting(right.setting), (receipt) => + Number(receipt.resultSequence), + ), + () => Number.MAX_SAFE_INTEGER, + ) + : dispatchPriority(left) - dispatchPriority(right), ); const errors: Array<{ setting: ConfigurationSetting; message: string }> = []; for (const change of dispatchOrder) { From dbacc3c0ec0c62809403fe033dd6a5f741f51938 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 30 Aug 2026 10:56:45 -0700 Subject: [PATCH 4/6] fix(mcp): preserve tied configuration replays --- .../ConversationConfigurationMcpService.ts | 26 +++-- ...OrchestratorMcpToolkit.integration.test.ts | 108 +++++++++++++++--- 2 files changed, 109 insertions(+), 25 deletions(-) diff --git a/apps/server/src/mcp/ConversationConfigurationMcpService.ts b/apps/server/src/mcp/ConversationConfigurationMcpService.ts index 67d716bcdfeb..a4f88e5fc06d 100644 --- a/apps/server/src/mcp/ConversationConfigurationMcpService.ts +++ b/apps/server/src/mcp/ConversationConfigurationMcpService.ts @@ -652,20 +652,22 @@ export const make = Effect.gen(function* () { : setting === "runtime_mode" ? priorRuntimeReceipt : priorInteractionReceipt; + const durableReplayOrder = ( + left: (typeof pending)[number], + right: (typeof pending)[number], + ) => { + const leftReceipt = Option.getOrUndefined(priorReceiptForSetting(left.setting)); + const rightReceipt = Option.getOrUndefined(priorReceiptForSetting(right.setting)); + const sequenceDifference = + Number(leftReceipt?.resultSequence ?? Number.MAX_SAFE_INTEGER) - + Number(rightReceipt?.resultSequence ?? Number.MAX_SAFE_INTEGER); + if (sequenceDifference !== 0) return sequenceDifference; + if (leftReceipt?.status === rightReceipt?.status) return 0; + return leftReceipt?.status === "accepted" ? -1 : 1; + }; const dispatchOrder = [...pending].sort((left, right) => hasRejectedReceipt - ? Option.getOrElse( - Option.map(priorReceiptForSetting(left.setting), (receipt) => - Number(receipt.resultSequence), - ), - () => Number.MAX_SAFE_INTEGER, - ) - - Option.getOrElse( - Option.map(priorReceiptForSetting(right.setting), (receipt) => - Number(receipt.resultSequence), - ), - () => Number.MAX_SAFE_INTEGER, - ) + ? durableReplayOrder(left, right) : dispatchPriority(left) - dispatchPriority(right), ); const errors: Array<{ setting: ConfigurationSetting; message: string }> = []; diff --git a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts index bcac8941f8d0..469fc33fb77e 100644 --- a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts @@ -51,6 +51,7 @@ import { McpSchema, McpServer } from "effect/unstable/ai"; import { ClaudeProviderCapabilitiesV2 } from "../orchestration-v2/Adapters/ClaudeAdapterV2.ts"; import { CodexProviderCapabilitiesV2 } from "../orchestration-v2/Adapters/CodexAdapterV2.ts"; import { CodexOrchestratorReplayHarness } from "../orchestration-v2/Adapters/CodexAdapterV2.testkit.ts"; +import { CommandReceiptStoreV2 } from "../orchestration-v2/CommandReceiptStore.ts"; import { OrchestratorV2, type OrchestratorV2Shape } from "../orchestration-v2/Orchestrator.ts"; import { layer as threadManagementServiceLayer, @@ -672,6 +673,7 @@ describe("orchestrator MCP toolkit", () => { yield* Effect.gen(function* () { const orchestrator = yield* OrchestratorV2; + const commandReceipts = yield* CommandReceiptStoreV2; const server = yield* McpServer.McpServer; yield* orchestrator.dispatch({ type: "thread.create", @@ -3358,18 +3360,28 @@ describe("orchestrator MCP toolkit", () => { clientRequestId: "configuration-target-restore-after-restrictions", }); + yield* orchestrator.dispatch({ + type: "thread.runtime-mode.set", + commandId: CommandId.make("command:mcp-configuration:caller-restrict-for-replay"), + threadId: parentThreadId, + runtimeMode: "approval-required", + }); + const stalePartialInput = { + threadId: lateParentThreadId, + options: [{ id: "reasoning", value: "high" }], + runtimeMode: "approval-required", + interactionMode: "plan", + clientRequestId: "configuration-stale-partial-selection", + } as const; yield* Ref.set(selectionPlanArmed, true); const stalePartialFiber = yield* Effect.forkChild( - invoke("t3_thread_configure", { - threadId: lateParentThreadId, - options: [{ id: "reasoning", value: "high" }], - clientRequestId: "configuration-stale-partial-selection", - }), + invoke("t3_thread_configure", stalePartialInput), ); yield* Deferred.await(selectionPlanEntered); const concurrentSelection = { instanceId: codexInstanceId, - model: "gpt-5.5-concurrent", + model: codexModel, + options: [{ id: "reasoning", value: "low" }], } satisfies ModelSelection; yield* orchestrator.dispatch({ type: "thread.model-selection.set", @@ -3378,14 +3390,84 @@ describe("orchestrator MCP toolkit", () => { modelSelection: concurrentSelection, }); yield* Deferred.succeed(selectionPlanRelease, undefined); - const stalePartialCall = yield* Fiber.join(stalePartialFiber); - expect(stalePartialCall.structuredContent).toMatchObject({ - _tag: "OrchestratorMcpFailure", - code: "orchestration_error", + const stalePartial = yield* decodeConfigureResult( + (yield* Fiber.join(stalePartialFiber)).structuredContent, + ); + expect(stalePartial).toMatchObject({ + outcome: "partially_applied", + changes: [ + { setting: "runtime_mode", receipt: expect.any(Object) }, + { setting: "interaction_mode", receipt: expect.any(Object) }, + ], + errors: [{ setting: "selection", message: expect.any(String) }], + }); + const selectionCommandId = CommandId.make( + [ + "command", + "mcp", + encodeURIComponent(invocation.providerSessionId), + encodeURIComponent(lateParentThreadId), + "thread-configure-selection", + encodeURIComponent(stalePartialInput.clientRequestId), + ].join(":"), + ); + const selectionReceipt = yield* commandReceipts.getByCommandId(selectionCommandId); + const interactionReceiptId = stalePartial.changes.find( + (change) => change.setting === "interaction_mode", + )?.receipt?.commandId; + if (interactionReceiptId === undefined || Option.isNone(selectionReceipt)) { + return yield* Effect.die( + new Error("Expected durable interaction and rejected selection receipts."), + ); + } + const interactionReceipt = yield* commandReceipts.getByCommandId(interactionReceiptId); + expect(selectionReceipt.value.status).toBe("rejected"); + expect(Option.getOrThrow(interactionReceipt).status).toBe("accepted"); + expect(Option.getOrThrow(interactionReceipt).resultSequence).toBe( + selectionReceipt.value.resultSequence, + ); + yield* orchestrator.dispatch({ + type: "thread.interaction-mode.set", + commandId: CommandId.make( + "command:mcp-configuration:caller-interaction-restrict-for-replay", + ), + threadId: parentThreadId, + interactionMode: "plan", + }); + const beforeStalePartialRetry = + yield* orchestrator.getThreadProjection(lateParentThreadId); + const repeatedStalePartial = yield* decodeConfigureResult( + (yield* invoke("t3_thread_configure", stalePartialInput)).structuredContent, + ); + expect(repeatedStalePartial.changes).toEqual(stalePartial.changes); + expect(repeatedStalePartial.errors[0]?.message).toContain("previously rejected"); + const afterStalePartialRetry = + yield* orchestrator.getThreadProjection(lateParentThreadId); + expect(afterStalePartialRetry.thread).toEqual(beforeStalePartialRetry.thread); + expect(afterStalePartialRetry.providerSessions).toEqual( + beforeStalePartialRetry.providerSessions, + ); + yield* orchestrator.dispatch({ + type: "thread.runtime-mode.set", + commandId: CommandId.make("command:mcp-configuration:caller-restore-after-replay"), + threadId: parentThreadId, + runtimeMode: "full-access", + }); + yield* orchestrator.dispatch({ + type: "thread.interaction-mode.set", + commandId: CommandId.make( + "command:mcp-configuration:caller-interaction-restore-after-replay", + ), + threadId: parentThreadId, + interactionMode: "default", + }); + yield* invoke("t3_thread_configure", { + threadId: lateParentThreadId, + options: [], + runtimeMode: "full-access", + interactionMode: "default", + clientRequestId: "configuration-target-restore-after-replay", }); - expect( - (yield* orchestrator.getThreadProjection(lateParentThreadId)).thread.modelSelection, - ).toEqual(concurrentSelection); const crossProviderInput = { threadId: lateParentThreadId, From 8310b79036688081623f38f0369345353596777d Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 4 Sep 2026 23:01:01 -0700 Subject: [PATCH 5/6] test(mcp): provide provider switch in replay fixture --- apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts | 3 +++ 1 file changed, 3 insertions(+) diff --git a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts index 469fc33fb77e..6774e0148468 100644 --- a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts @@ -3544,6 +3544,9 @@ describe("orchestrator MCP toolkit", () => { const testLayer = McpHttpServer.OrchestratorToolkitRegistrationLive.pipe( Layer.provideMerge(McpServer.McpServer.layer), Layer.provideMerge(orchestrationLayer), + Layer.provide( + providerSwitchServiceLayer.pipe(Layer.provide(makeProviderAdapterRegistryLayer([]))), + ), Layer.provide(providerRegistryLayer), Layer.provide(unusedScheduledTaskStubLayer), Layer.provide(NodeServices.layer), From 26d25cba6e28d34bd0139f25e29cb9d48ebdd1ba Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 7 Sep 2026 11:30:36 -0700 Subject: [PATCH 6/6] fix(mcp): reject unadvertised conversation models --- ...onversationConfigurationMcpService.test.ts | 61 ++++++++++++++++--- .../ConversationConfigurationMcpService.ts | 2 +- 2 files changed, 54 insertions(+), 9 deletions(-) diff --git a/apps/server/src/mcp/ConversationConfigurationMcpService.test.ts b/apps/server/src/mcp/ConversationConfigurationMcpService.test.ts index 36f01b2599c6..66aaf8bf4e00 100644 --- a/apps/server/src/mcp/ConversationConfigurationMcpService.test.ts +++ b/apps/server/src/mcp/ConversationConfigurationMcpService.test.ts @@ -115,6 +115,7 @@ function testLayer(input: { readonly getThreadProjection?: ThreadManagementService["Service"]["getThreadProjection"]; readonly getReceipt?: CommandReceiptStoreV2["Service"]["getByCommandId"]; readonly providers?: ReadonlyArray; + readonly plan?: ProviderSwitchServiceV2["Service"]["plan"]; }) { return ConversationConfiguration.layer.pipe( Layer.provide( @@ -138,14 +139,16 @@ function testLayer(input: { getByCommandId: input.getReceipt ?? (() => Effect.succeed(Option.none())), }), Layer.mock(ProviderSwitchServiceV2)({ - plan: () => - Effect.succeed({ - instanceChanged: false, - modelChanged: false, - targetProviderThreadId: null, - releaseProviderSessionIds: [], - transition: { type: "switch_model_in_session" }, - }), + plan: + input.plan ?? + (() => + Effect.succeed({ + instanceChanged: false, + modelChanged: false, + targetProviderThreadId: null, + releaseProviderSessionIds: [], + transition: { type: "switch_model_in_session" }, + })), }), ), ), @@ -153,6 +156,48 @@ function testLayer(input: { } describe("ConversationConfigurationMcpService", () => { + it.effect("rejects an explicit model when the provider advertises no models", () => + Effect.gen(function* () { + const planned = yield* Ref.make(0); + const dispatched = yield* Ref.make(0); + const parent = projection({ + threadId: parentThreadId, + runtimeMode: "full-access", + interactionMode: "default", + }); + const target = projection({ + threadId: targetThreadId, + runtimeMode: "full-access", + interactionMode: "default", + }); + const layer = testLayer({ + parent, + target, + providers: [{ ...provider, models: [] }], + plan: () => + Ref.update(planned, (count) => count + 1).pipe(Effect.andThen(Effect.die("plan"))), + dispatch: () => + Ref.update(dispatched, (count) => count + 1).pipe(Effect.andThen(Effect.die("dispatch"))), + }); + + const error = yield* Effect.gen(function* () { + const service = yield* ConversationConfiguration.ConversationConfigurationMcpService; + return yield* service + .configure(scope(), { + threadId: targetThreadId, + model: "unadvertised-model", + clientRequestId: "empty-model-catalog", + }) + .pipe(Effect.flip); + }).pipe(Effect.provide(layer)); + + assert.equal(error.code, "model_unavailable"); + assert.match(error.message, /not advertised/); + assert.equal(yield* Ref.get(planned), 0); + assert.equal(yield* Ref.get(dispatched), 0); + }), + ); + it.effect("replays an accepted selection before provider availability planning", () => Effect.gen(function* () { const parent = projection({ diff --git a/apps/server/src/mcp/ConversationConfigurationMcpService.ts b/apps/server/src/mcp/ConversationConfigurationMcpService.ts index a4f88e5fc06d..eebf0916572c 100644 --- a/apps/server/src/mcp/ConversationConfigurationMcpService.ts +++ b/apps/server/src/mcp/ConversationConfigurationMcpService.ts @@ -350,7 +350,7 @@ export const make = Effect.gen(function* () { ); } const advertisedModel = provider.models.find((candidate) => candidate.slug === model); - if (provider.models.length > 0 && advertisedModel === undefined) { + if (advertisedModel === undefined) { return yield* failure( "model_unavailable", `Model ${model} is not advertised by provider instance ${instanceId}.`,