diff --git a/apps/server/src/mcp/toolkits/project/handlers.test.ts b/apps/server/src/mcp/toolkits/project/handlers.test.ts index 7232a3ebdaa1b..e11a9c7ceb477 100644 --- a/apps/server/src/mcp/toolkits/project/handlers.test.ts +++ b/apps/server/src/mcp/toolkits/project/handlers.test.ts @@ -95,6 +95,7 @@ const clientLaunchHarness = (input: { }, }), Layer.mock(ThreadLaunch.ThreadLaunchService)({ + getAcceptedLaunch: () => Effect.succeedNone, checkWorktreeBase: input.checkWorktreeBase ?? (() => Effect.void), launch: (launch) => { input.launched.push(launch); diff --git a/apps/server/src/mcp/toolkits/project/handlers.ts b/apps/server/src/mcp/toolkits/project/handlers.ts index af97cd3694f11..1e2ee32d865b5 100644 --- a/apps/server/src/mcp/toolkits/project/handlers.ts +++ b/apps/server/src/mcp/toolkits/project/handlers.ts @@ -116,6 +116,19 @@ export const layer = McpToolAccess.toLayer(ProjectToolkit, { ), )).projectId : yield* resolveProjectId(context, input.projectId); + // Accepted creates skip mutable base/default-model preflights. + // Message intake still handles their replay and preparation. + const acceptedLaunch = + input.clientRequestId === undefined + ? undefined + : Option.getOrUndefined( + yield* ThreadLaunch.ThreadLaunchService.pipe( + Effect.flatMap((launches) => + launches.getAcceptedLaunch({ commandId, threadId, projectId }), + ), + Effect.mapError(unavailable), + ), + ); const readProject = Project.ProjectService.pipe( Effect.flatMap((projects) => projects.getById(projectId)), Effect.mapError(unavailable), @@ -131,7 +144,9 @@ export const layer = McpToolAccess.toLayer(ProjectToolkit, { const launchedModel = launched === null || launched.deletedAt !== null ? undefined : launched.modelSelection; const replaying = launchedModel !== undefined; - if (!replaying && input.workspaceStrategy?.type === "existing_worktree") { + // A retry may bind a workspace to an idle accepted launch, so membership + // is checked on every launch, unlike the checks below. + if (input.workspaceStrategy?.type === "existing_worktree") { const project = yield* readProject; if (project === undefined) return yield* new OrchestratorMcpFailure({ @@ -142,7 +157,7 @@ export const layer = McpToolAccess.toLayer(ProjectToolkit, { } // Reject unresolved base refs before accepting a launch; worktree provisioning happens later. const strategy = input.workspaceStrategy; - if (!replaying && strategy?.type === "worktree") + if (acceptedLaunch === undefined && !replaying && strategy?.type === "worktree") yield* ThreadLaunch.ThreadLaunchService.pipe( Effect.flatMap((launches) => launches.checkWorktreeBase({ projectId, workspaceStrategy: strategy }), @@ -176,6 +191,7 @@ export const layer = McpToolAccess.toLayer(ProjectToolkit, { ) : undefined; const modelSelection = + acceptedLaunch?.modelSelection ?? input.modelSelection ?? launchedModel ?? caller?.modelSelection ?? diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index e27f32ef84420..b1cffb8f8f816 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -12193,6 +12193,33 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }), ), ); + if ( + command.type === "message.dispatch" && + command.dispatchMode?.type === "defer_start" && + (yield* DispatchModeLimit) !== undefined + ) { + const runCreated = storedEvents.find( + (stored) => stored.event.type === "run.created", + )?.event; + if (runCreated?.type === "run.created") { + const current = yield* projectionStore + .getThreadRecords(command.threadId, ["runs"]) + .pipe( + Effect.mapError( + (cause) => new OrchestratorProjectionError({ threadId: command.threadId, cause }), + ), + ); + // Replaying this receipt can resume workspace preparation, so it must + // respect the caller's current ceiling just like a fresh dispatch. + if ( + current.runs.some( + (run) => run.id === runCreated.payload.id && run.status === "preparing", + ) + ) { + yield* refuseAboveDispatchModeLimit(command, command.threadId, current.thread); + } + } + } if (command.type === "queue.resume" || command.type === "queued-run.start") { yield* mapDispatchError(command)( startNextQueuedRun( diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts index f3a7ede997239..bdba7becc3da3 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts @@ -15,6 +15,7 @@ import { ComposerContextId, type ChatAttachment, CommandId, + EnvironmentId, DEFAULT_SERVER_SETTINGS, GitCommandError, MessageId, @@ -59,6 +60,10 @@ import * as EffectOutbox from "./EffectOutbox.ts"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; import * as ThreadLaunch from "./ThreadLaunchService.ts"; +import * as McpInvocationContext from "../mcp/McpInvocationContext.ts"; +import * as McpToolAccess from "../mcp/McpToolAccess.ts"; +import * as ProjectHandlers from "../mcp/toolkits/project/handlers.ts"; +import { ProjectToolkit } from "../mcp/toolkits/project/tools.ts"; import * as GitVcsDriver from "../vcs/GitVcsDriver.ts"; import * as ThreadManagement from "./ThreadManagementService.ts"; import * as ThreadTitleRegeneration from "./ThreadTitleRegenerationService.ts"; @@ -114,6 +119,7 @@ const claudeAdapter = { } as ProviderAdapter.ProviderAdapterV2["Service"]; interface HarnessOptions { + readonly getProject?: ProjectService.ProjectService["Service"]["getById"]; readonly closeTerminal?: TerminalManager.TerminalManager["Service"]["close"]; readonly adapters?: ReadonlyArray; readonly managedFolders?: Layer.Layer; @@ -179,14 +185,16 @@ function makeHarness(options: HarnessOptions = {}) { bootstrap: () => Effect.die("unused"), update: () => Effect.die("unused"), delete: () => Effect.die("unused"), - getById: (id) => - Effect.succeed( - id === projectId - ? Option.some(project) - : id === otherProjectId - ? Option.some(otherProject) - : Option.none(), - ), + getById: + options.getProject ?? + ((id) => + Effect.succeed( + id === projectId + ? Option.some(project) + : id === otherProjectId + ? Option.some(otherProject) + : Option.none(), + )), getByWorkspaceRoot: () => Effect.succeedSome(project), snapshot: Effect.die("unused"), getShell: () => Effect.die("unused"), @@ -1386,6 +1394,180 @@ it.effect("replays a server-allocated launch", () => }), ); +it.effect("MCP replays accepted launches and checks worktree membership", () => + Effect.gen(function* () { + const setupEntered = yield* Deferred.make(); + const allowSetup = yield* Deferred.make(); + let changed = false; + const harness = makeHarness({ + getProject: (id) => + Effect.sync(() => + id === projectId + ? Option.some({ ...project, defaultModelSelection: changed ? null : modelSelection }) + : id === otherProjectId + ? Option.some(otherProject) + : Option.none(), + ), + hasCommit: ({ refName }) => Effect.sync(() => refName === "HEAD" || !changed), + git: { hasRefNamed: () => Effect.succeed(false) }, + runSetup: () => + Deferred.succeed(setupEntered, undefined).pipe( + Effect.andThen(Deferred.await(allowSetup)), + Effect.as({ status: "no-script" as const }), + ), + }); + const context = McpInvocationContext.McpInvocationContext.of({ + environmentId: EnvironmentId.make("launch-retry"), + requestNamespace: "client:launch-retry", + thread: undefined, + client: { sessionId: "launch-retry", label: "Test", access: "full-access" }, + issuedAt: 0, + capabilities: new Set(["orchestration" as const]), + }); + const dependencies = Layer.mergeAll( + harness.layer, + NodeCrypto.layer, + DocumentText.layer.pipe(Layer.provideMerge(NodeServices.layer)), + Layer.succeed(McpInvocationContext.McpInvocationContext, context), + Layer.mock(GitVcsDriver.GitVcsDriver)({ + listWorktreePaths: () => Effect.succeed([project.workspaceRoot, "/repo-worktrees/feature"]), + }), + ServerConfig.layerTest(process.cwd(), { prefix: "t3-launch-retry-" }).pipe( + Layer.provide(NodeServices.layer), + ), + ); + yield* Effect.gen(function* () { + const toolkit = yield* ProjectToolkit.pipe( + Effect.provide(McpToolAccess.HandlersLayer.layer(ProjectHandlers.layer)), + ); + const input = { + title: "Only once", + projectId, + clientRequestId: "lost-response", + runtimeMode: "approval-required" as const, + message: "Do the work", + workspaceStrategy: { type: "worktree" as const, baseRef: "temporary-base" }, + }; + const launch = (params: Parameters>[1]) => + toolkit.handle("t3_thread_launch", params).pipe( + Stream.unwrap, + Stream.runCollect, + Effect.map((results) => results.at(-1)?.result), + ); + const fs = yield* FileSystem.FileSystem; + const outsidePath = yield* fs.makeTempDirectoryScoped({ prefix: "launch-unregistered-" }); + const outsideStrategy = { type: "existing_worktree" as const, worktreePath: outsidePath }; + const refusal = { + code: "invalid_request", + message: + "worktreePath must be one of the project's git worktrees. t3_worktree_list shows them.", + }; + const idleInput = { + title: "Idle accepted launch", + projectId, + clientRequestId: "idle-retry", + workspaceStrategy: { type: "root" as const }, + }; + const idleLaunch = yield* launch(idleInput); + assert.include(idleLaunch, { runId: null, status: null }); + assert(idleLaunch !== undefined && "threadId" in idleLaunch); + assert.include( + yield* launch({ + ...idleInput, + message: "Run outside the project", + workspaceStrategy: outsideStrategy, + }), + refusal, + ); + const threads = yield* ThreadManagement.ThreadManagementService; + const idle = yield* threads.getThreadProjection(idleLaunch.threadId); + assert.equal(idle.messages.length, 0); + assert.equal(idle.runs.length, 0); + assert.equal(idle.thread.worktreePath, null); + assert.equal(harness.createWorktree.mock.calls.length, 0); + assert.equal(harness.runSetup.mock.calls.length, 0); + + const first = yield* launch(input); + assert(first !== undefined && "threadId" in first); + assert.notProperty(first, "code"); + yield* Deferred.await(setupEntered); + changed = true; + + assert.include(yield* launch({ ...input, workspaceStrategy: outsideStrategy }), refusal); + + assert.deepEqual(yield* launch(input), first); + assert.deepEqual(yield* launch({ ...input, modelSelection: claudeModelSelection }), first); + assert.include(yield* launch({ ...input, projectId: otherProjectId }), { + code: "orchestration_error", + }); + assert.include( + yield* launch(input).pipe( + Effect.provideService(McpInvocationContext.McpInvocationContext, { + ...context, + client: { ...context.client!, access: "read-only" }, + }), + ), + { code: "capability_denied" }, + ); + assert.include(yield* launch({ ...input, clientRequestId: "fresh" }), { + code: "invalid_request", + }); + assert.include( + yield* launch({ + ...input, + clientRequestId: "fresh-root", + workspaceStrategy: { type: "root" }, + }), + { code: "invalid_request" }, + ); + const threadId = first.threadId; + assert.equal((yield* threads.getThreadShell(threadId))?.runtimeMode, "approval-required"); + yield* threads.dispatch({ + type: "thread.runtime-mode.set", + commandId: CommandId.make("owner:raise-launch-modes"), + threadId, + runtimeMode: "full-access", + }); + assert.include( + yield* launch(input).pipe( + Effect.provideService(McpInvocationContext.McpInvocationContext, { + ...context, + client: { ...context.client!, access: "approval-required" }, + }), + ), + { code: "orchestration_error" }, + ); + const projection = yield* threads.getThreadProjection(threadId); + assert.equal(projection.thread.runtimeMode, "full-access"); + assert.equal(projection.messages.length, 1); + assert.equal(projection.runs.length, 1); + assert.equal(projection.runs[0]?.status, "preparing"); + assert.deepEqual(projection.runs[0]?.workspacePreparation, input.workspaceStrategy); + assert.equal(projection.thread.worktreePath, "/repo-worktrees/feature"); + assert.equal(harness.createWorktree.mock.calls.length, 1); + assert.equal(harness.runSetup.mock.calls.length, 1); + yield* Deferred.succeed(allowSetup, undefined); + yield* threads.streamStoredEventsFrom({ threadId }).pipe( + Stream.filter( + (stored) => + stored.commandId === CommandId.make(`${threadId}:release`) && + stored.event.type === "run.updated", + ), + Stream.runHead, + ); + assert.include( + yield* launch(input).pipe( + Effect.provideService(McpInvocationContext.McpInvocationContext, { + ...context, + client: { ...context.client!, access: "approval-required" }, + }), + ), + { threadId, status: "starting" }, + ); + }).pipe(Effect.provide(dependencies)); + }), +); + it.effect("replays an accepted launch after the project disables the instance", () => { const harness = makeHarness(); return Effect.gen(function* () { @@ -1742,19 +1924,38 @@ it.effect("schedules an accepted preparing message exactly once across concurren text: "Resume preparation", attachments: [], modelSelection: input.modelSelection, - dispatchMode: { type: "defer_start" }, + dispatchMode: { type: "defer_start", workspaceStrategy: input.workspaceStrategy }, createdBy: input.createdBy, creationSource: input.creationSource, }); const preparing = yield* threads.getThreadProjection(input.threadId); assert.equal(preparing.runs[0]?.status, "preparing"); - const [first, second] = yield* Effect.all([launches.launch(input), launches.launch(input)], { + const retry = { ...input, workspaceStrategy: { type: "root" as const } }; + const [first, second] = yield* Effect.all([launches.launch(retry), launches.launch(retry)], { concurrency: "unbounded", }); - yield* Deferred.await(setupEntered); + const prepared = yield* Effect.raceFirst( + Deferred.await(setupEntered).pipe(Effect.as(true)), + threads.streamStoredEventsFrom({ threadId: input.threadId }).pipe( + Stream.filter( + (stored) => + stored.commandId === CommandId.make(input.commandId + ":release") && + stored.event.type === "run.updated", + ), + Stream.runHead, + Effect.as(false), + ), + ); + assert.isTrue(prepared, "the accepted worktree setup must run before releasing the run"); assert.isTrue(first.resumed); assert.isTrue(second.resumed); + assert.equal(harness.createWorktree.mock.calls.length, 1); + assert.equal(harness.runSetup.mock.calls[0]?.[0].worktreePath, "/repo-worktrees/feature"); + assert.deepEqual( + (yield* threads.getThreadProjection(input.threadId)).runs[0]?.workspacePreparation, + input.workspaceStrategy, + ); assert.equal(harness.runSetup.mock.calls.length, 1); yield* Deferred.succeed(allowSetup, undefined); }).pipe(Effect.provide(harness.layer)); @@ -1867,6 +2068,7 @@ it.effect("shared intake preserves durable attachment bytes after a lost launch }), ), ), + getAcceptedLaunch: launches.getAcceptedLaunch, checkWorktreeBase: launches.checkWorktreeBase, retryPreparation: launches.retryPreparation, prepareRun: launches.prepareRun, diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.ts index 763ef25b43626..aad0540f2e5ab 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.ts @@ -11,6 +11,7 @@ import { type OrchestrationV2CreationSource, type OrchestrationV2ProviderThreadNativeMetadata, type OrchestrationV2ThreadProjection, + type OrchestrationV2ThreadShell, type ProviderDriverKind, type ProviderInteractionMode, ProjectId, @@ -168,6 +169,9 @@ export class ThreadLaunchService extends Context.Service< ThreadLaunchResult, ThreadLaunchError | ThreadManagement.ThreadManagementProviderInstanceDisabledError >; + readonly getAcceptedLaunch: ( + input: Pick, + ) => Effect.Effect, ThreadLaunchError>; /** Check before creation for early errors; launch otherwise accepts and reports base failures inside the thread. */ readonly checkWorktreeBase: ( input: Pick, @@ -213,7 +217,11 @@ const make = Effect.gen(function* () { yield* Effect.addFinalizer(() => Scope.close(preparationScope, Exit.void)); const mapError = - (input: PreparationInput, operation: ThreadLaunchError["operation"], threadId?: ThreadId) => + ( + input: Pick, + operation: ThreadLaunchError["operation"], + threadId?: ThreadId, + ) => (cause: unknown) => new ThreadLaunchError({ operation, @@ -223,11 +231,39 @@ const make = Effect.gen(function* () { cause, }); - const readReceipt = (input: ThreadLaunchInput, commandId: CommandId) => + const readReceipt = ( + input: Pick, + commandId: CommandId, + ) => receipts .getByCommandId(commandId) .pipe(Effect.mapError(mapError(input, "read-receipt", input.threadId))); + const getAcceptedLaunch: ThreadLaunchService["Service"]["getAcceptedLaunch"] = Effect.fn( + "ThreadLaunchService.getAcceptedLaunch", + )(function* (input) { + const receipt = yield* readReceipt(input, input.commandId); + if (Option.isNone(receipt)) return Option.none(); + if ( + receipt.value.status !== "accepted" || + receipt.value.commandType !== "thread.create" || + (input.threadId !== undefined && input.threadId !== receipt.value.threadId) + ) { + return yield* mapError(input, "read-receipt")("This command cannot be replayed as a launch."); + } + const shell = yield* threads + .getThreadShell(receipt.value.threadId) + .pipe(Effect.mapError(mapError(input, "create-thread", receipt.value.threadId))); + if (shell === null || shell.projectId !== input.projectId) { + return yield* mapError( + input, + "resolve-project", + receipt.value.threadId, + )("The accepted launch does not belong to this project."); + } + return Option.some(shell); + }); + const validateReusableThread = Effect.fn("ThreadLaunchService.validateReusableThread")(function* ( input: ThreadLaunchInput, threadId: ThreadId, @@ -993,9 +1029,8 @@ const make = Effect.gen(function* () { const projection = yield* threads .getThreadProjection(threadId) .pipe(Effect.mapError(mapError(input, "create-thread", threadId))); - const runIsPreparing = - runId !== null && - projection.runs.some((run) => run.id === runId && run.status === "preparing"); + const acceptedRun = projection.runs.find((run) => run.id === runId); + const runIsPreparing = acceptedRun?.status === "preparing"; const shouldSchedule = runId === null ? Option.isNone(launchReceipt) : runIsPreparing; const boundWorktreePath = projection.thread.worktreePath; const preparationStrategy: ThreadLaunchWorkspaceStrategy = @@ -1022,11 +1057,20 @@ const make = Effect.gen(function* () { Effect.mapError(mapError(input, "update-thread", threadId)), ); if (preparationStillRequired) { - yield* schedulePreparation( - { ...input, workspaceStrategy: preparationStrategy }, - threadId, - runId, - ); + if (acceptedRun?.workspacePreparation !== undefined) { + yield* scheduleRunPreparation( + { commandId: input.commandId, threadId, runId: acceptedRun.id }, + projection, + acceptedRun, + acceptedRun.workspacePreparation, + ); + } else { + yield* schedulePreparation( + { ...input, workspaceStrategy: preparationStrategy }, + threadId, + runId, + ); + } } else { yield* releasePreparation(input.commandId); } @@ -1120,7 +1164,13 @@ const make = Effect.gen(function* () { ); }; - return ThreadLaunchService.of({ launch, checkWorktreeBase, retryPreparation, prepareRun }); + return ThreadLaunchService.of({ + launch, + getAcceptedLaunch, + checkWorktreeBase, + retryPreparation, + prepareRun, + }); }); export const layer = Layer.effect(ThreadLaunchService, make); diff --git a/docs/orchestration-v2/orchestrator-mcp-server.md b/docs/orchestration-v2/orchestrator-mcp-server.md index e41f42c8248a3..8cb92cf4d5cd3 100644 --- a/docs/orchestration-v2/orchestrator-mcp-server.md +++ b/docs/orchestration-v2/orchestrator-mcp-server.md @@ -364,7 +364,9 @@ Pass the task in `message`. Project, model, and modes inherit when omitted; workspace does not. `scratch: true` launches without a project, in a folder of its own under the environment's Scratch project. For stacked PRs, use the parent branch as `baseRef` with `startFromOrigin: false`. The new thread may not run with broader runtime or -interaction modes than the caller. Launch has no retry key, so inspect existing threads after a failed or lost response before +interaction modes than the caller. Reusing a `clientRequestId` within the same +MCP session returns the launch it first accepted instead of launching again. +Without a key, inspect existing threads after a failed or lost response before launching again. `create_threads` remains the batch option for a shared checkout. ### `t3_thread_list`