diff --git a/apps/server/src/mcp/toolkits/project/handlers.test.ts b/apps/server/src/mcp/toolkits/project/handlers.test.ts index 4212fe5b16df..3df436ca99d9 100644 --- a/apps/server/src/mcp/toolkits/project/handlers.test.ts +++ b/apps/server/src/mcp/toolkits/project/handlers.test.ts @@ -6,6 +6,7 @@ import { type Project as ProjectRecord, type ModelSelection, type RuntimeMode, + type OrchestrationV2ThreadShell, } from "@t3tools/contracts"; import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; import * as NodeServices from "@effect/platform-node/NodeServices"; @@ -44,6 +45,10 @@ const clientLaunchHarness = (input: { instanceId: ProviderInstanceId.make("claudeAgent"), model: "claude-opus", }; + // The project's own default model, which a test can clear between launches. + const projectDefaults: { model: ModelSelection | null } = { + model: input.overrideModel === undefined ? modelSelection : null, + }; const layerDependencies = Layer.mergeAll( ServerSettings.layerTest({ ...(input.defaultRuntimeMode === undefined @@ -71,7 +76,22 @@ const clientLaunchHarness = (input: { issuedAt: 0, capabilities: new Set(["orchestration" as const]), }), - Layer.mock(ThreadManagement.ThreadManagementService)({}), + Layer.mock(ThreadManagement.ThreadManagementService)({ + // A thread exists once a launch started it. + getThreadShell: (threadId) => { + const launch = input.launched.find((candidate) => candidate.threadId === threadId); + return Effect.succeed( + launch === undefined + ? null + : ({ + id: launch.threadId, + projectId: launch.projectId, + modelSelection: launch.modelSelection, + deletedAt: null, + } as unknown as OrchestrationV2ThreadShell), + ); + }, + }), Layer.mock(ThreadLaunch.ThreadLaunchService)({ checkWorktreeBase: input.checkWorktreeBase ?? (() => Effect.void), launch: (launch) => { @@ -98,7 +118,7 @@ const clientLaunchHarness = (input: { ? Option.some({ id, workspaceRoot: input.workspaceRoot ?? "/projects/client-target", - defaultModelSelection: input.overrideModel === undefined ? modelSelection : null, + defaultModelSelection: projectDefaults.model, } as unknown as ProjectRecord) : Option.none(), ), @@ -115,7 +135,7 @@ const clientLaunchHarness = (input: { ), Layer.provideMerge(NodeServices.layer), ); - return { projectId, modelSelection, dependencies: layerDependencies }; + return { projectId, modelSelection, projectDefaults, dependencies: layerDependencies }; }; it.effect( @@ -200,6 +220,71 @@ it.effect( }), ); +it.effect("a launch retried with the same clientRequestId replays the first one", () => + Effect.gen(function* () { + const launched: Array = []; + const { projectId, dependencies } = clientLaunchHarness({ + runtimeModeCeiling: "auto", + launched, + }); + const toolkit = yield* ProjectToolkit.pipe( + Effect.provide( + McpToolAccess.HandlersLayer.layer(ProjectHandlers.layer).pipe(Layer.provide(dependencies)), + ), + ); + const launch = (clientRequestId: string | undefined) => + toolkit + .handle("t3_thread_launch", { + title: "Fix", + projectId, + message: "Fix the bug", + ...(clientRequestId === undefined ? {} : { clientRequestId }), + }) + .pipe(Stream.unwrap, Stream.runCollect, Effect.provide(dependencies)); + + yield* launch("fix-1"); + yield* launch("fix-1"); + yield* launch("fix-2"); + yield* launch(undefined); + yield* launch(undefined); + const ids = launched.map((input) => [input.commandId, input.threadId]); + // The same key sends the same command and thread ids, which the launch + // service replays instead of starting a second thread. + expect(ids[0]).toEqual(ids[1]); + expect(ids[0]?.[0]).toContain("client%3Asession-1"); + expect(ids[2]).not.toEqual(ids[0]); + // Without a key every launch is new. + expect(ids[3]).not.toEqual(ids[4]); + }), +); + +it.effect("a retried launch replays after the project loses its default model", () => + Effect.gen(function* () { + const launched: Array = []; + const { projectId, modelSelection, projectDefaults, dependencies } = clientLaunchHarness({ + runtimeModeCeiling: "auto", + launched, + }); + const toolkit = yield* ProjectToolkit.pipe( + Effect.provide( + McpToolAccess.HandlersLayer.layer(ProjectHandlers.layer).pipe(Layer.provide(dependencies)), + ), + ); + const launch = toolkit + .handle("t3_thread_launch", { title: "Fix", projectId, clientRequestId: "fix-1" }) + .pipe(Stream.unwrap, Stream.runCollect, Effect.provide(dependencies)); + + yield* launch; + projectDefaults.model = null; + const retried = (yield* launch).at(-1)?.result; + expect(retried).toMatchObject({ projectId, modelSelection }); + expect(launched.map((input) => input.threadId)).toEqual([ + launched[0]?.threadId, + launched[0]?.threadId, + ]); + }), +); + it.effect("a launch binds only an existing checkout that is one of the project's worktrees", () => Effect.gen(function* () { const fileSystem = yield* FileSystem.FileSystem; @@ -302,3 +387,46 @@ it.effect("a launch into a new worktree creates nothing when its base ref has no expect(launched).toHaveLength(1); }), ); + +it.effect("a retried worktree launch replays after its base ref is gone", () => + Effect.gen(function* () { + const launched: Array = []; + const base = { exists: true }; + const { projectId, dependencies } = clientLaunchHarness({ + runtimeModeCeiling: "auto", + launched, + checkWorktreeBase: ({ projectId, workspaceStrategy }) => + base.exists || workspaceStrategy.type !== "worktree" + ? Effect.void + : Effect.fail( + new ThreadLaunch.ThreadLaunchBaseRefError({ + projectId, + projectTitle: "Client target", + workspaceRoot: "/projects/client-target", + baseRef: workspaceStrategy.baseRef, + }), + ), + }); + const toolkit = yield* ProjectToolkit.pipe( + Effect.provide( + McpToolAccess.HandlersLayer.layer(ProjectHandlers.layer).pipe(Layer.provide(dependencies)), + ), + ); + const launch = toolkit + .handle("t3_thread_launch", { + title: "Fix", + projectId, + clientRequestId: "fix-worktree", + workspaceStrategy: { type: "worktree", baseRef: "feature", startFromOrigin: false }, + }) + .pipe(Stream.unwrap, Stream.runCollect, Effect.provide(dependencies)); + + yield* launch; + base.exists = false; + expect((yield* launch).at(-1)?.result).toMatchObject({ projectId }); + expect(launched.map((input) => input.threadId)).toEqual([ + launched[0]?.threadId, + launched[0]?.threadId, + ]); + }), +); diff --git a/apps/server/src/mcp/toolkits/project/handlers.ts b/apps/server/src/mcp/toolkits/project/handlers.ts index aaf37468f8cd..c5f9e65254b0 100644 --- a/apps/server/src/mcp/toolkits/project/handlers.ts +++ b/apps/server/src/mcp/toolkits/project/handlers.ts @@ -1,4 +1,10 @@ -import { MessageId, ThreadId, OrchestratorMcpFailure, ProjectId } from "@t3tools/contracts"; +import { + CommandId, + MessageId, + ThreadId, + OrchestratorMcpFailure, + ProjectId, +} from "@t3tools/contracts"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; import * as Option from "effect/Option"; @@ -58,8 +64,16 @@ export const layer = McpToolAccess.toLayer(ProjectToolkit, { (input, { runtimeMode, interactionMode }) => Effect.gen(function* () { const context = yield* readCaller(); - const { caller } = context; - const commandId = yield* newCommandId(); + const { caller, scope } = context; + // A retry with the same key replays the first launch: the ids derive + // from the caller and the key, and the launch service replays a + // command it already accepted. + const commandId = + input.clientRequestId === undefined + ? yield* newCommandId() + : CommandId.make( + `mcp:launch:${encodeURIComponent(scope.requestNamespace)}:${encodeURIComponent(input.clientRequestId)}`, + ); const threadId = ThreadId.make(commandId); const messageId = MessageId.make(commandId); const attachments = input.attachments ?? []; @@ -95,7 +109,17 @@ export const layer = McpToolAccess.toLayer(ProjectToolkit, { Effect.mapError(unavailable), Effect.map(Option.getOrUndefined), ); - if (input.workspaceStrategy?.type === "existing_worktree") { + // A retry of an accepted launch replays it: it keeps the model it started + // with and skips the checks only a new launch needs, which the project + // may no longer pass, such as a base branch deleted since. + const launched = + input.clientRequestId === undefined + ? null + : yield* context.threads.getThreadShell(threadId).pipe(Effect.mapError(unavailable)); + const launchedModel = + launched === null || launched.deletedAt !== null ? undefined : launched.modelSelection; + const replaying = launchedModel !== undefined; + if (!replaying && input.workspaceStrategy?.type === "existing_worktree") { const project = yield* readProject; if (project === undefined) return yield* new OrchestratorMcpFailure({ @@ -106,7 +130,7 @@ export const layer = McpToolAccess.toLayer(ProjectToolkit, { } // Reject unresolved base refs before accepting a launch; worktree provisioning happens later. const strategy = input.workspaceStrategy; - if (strategy?.type === "worktree") + if (!replaying && strategy?.type === "worktree") yield* ThreadLaunch.ThreadLaunchService.pipe( Effect.flatMap((launches) => launches.checkWorktreeBase({ projectId, workspaceStrategy: strategy }), @@ -127,7 +151,7 @@ export const layer = McpToolAccess.toLayer(ProjectToolkit, { // project's effective defaults, as the composer does. Its mode stays // within the ceiling it was approved with. const projectDefaults = - caller === undefined + caller === undefined && !replaying ? yield* ServerSettings.ServerSettingsService.pipe( Effect.flatMap((settings) => settings.getSettings), Effect.mapError(unavailable), @@ -141,6 +165,7 @@ export const layer = McpToolAccess.toLayer(ProjectToolkit, { : undefined; const modelSelection = input.modelSelection ?? + launchedModel ?? caller?.modelSelection ?? projectDefaults?.defaultModelSelection ?? undefined; diff --git a/apps/server/src/mcp/toolkits/project/tools.ts b/apps/server/src/mcp/toolkits/project/tools.ts index 6acd7b1a4a27..563a1c1c0bde 100644 --- a/apps/server/src/mcp/toolkits/project/tools.ts +++ b/apps/server/src/mcp/toolkits/project/tools.ts @@ -13,6 +13,7 @@ import { ProjectCreatePayload, ProjectUpdatePayload, ProjectId, + OrchestratorMcpClientRequestId, OrchestratorMcpFailure, SourceControlCloneRepositoryInput, SourceControlCloneRepositoryResult, @@ -101,7 +102,7 @@ const ProjectCloneTool = Tool.make("t3_project_clone", { const ThreadLaunchTool = Tool.make("t3_thread_launch", { ...shared, description: - "Create one new top-level thread, bound to its workspace before the agent starts. Task in `message`. Omit `projectId`, model and modes to inherit yours (outside a thread, pass `projectId` or `scratch`); modes cannot exceed yours. No retry key: after an error or lost reply, check `t3_thread_list` before retrying. Follow it with `t3_thread_read` or `t3_thread_wait`. Mention it with its `link`.", + "Create one new top-level thread, bound to its workspace before the agent starts. Task in `message`. Omit `projectId`, model and modes to inherit yours (outside a thread, pass `projectId` or `scratch`); modes cannot exceed yours. Pass `clientRequestId` and reuse it when retrying: the same key returns the first call's thread. Follow it with `t3_thread_read` or `t3_thread_wait`. Mention it with its `link`.", parameters: Schema.Struct({ projectId: Schema.optional(ProjectId), scratch: Schema.optional( @@ -126,6 +127,7 @@ const ThreadLaunchTool = Tool.make("t3_thread_launch", { }), ), attachments: Schema.optional(Schema.Array(McpAttachmentInput).check(Schema.isMaxLength(8))), + clientRequestId: Schema.optional(OrchestratorMcpClientRequestId), }), success: Schema.Struct({ threadId: ThreadId, diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts index a091d96feda6..f3a7ede99723 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts @@ -1423,6 +1423,40 @@ it.effect("replays an accepted launch after the project disables the instance", }).pipe(Effect.provide(harness.layer)); }); +it.effect("an agent's launch never starts work above the modes it asks for", () => { + const harness = makeHarness(); + return Effect.gen(function* () { + const launches = yield* ThreadLaunch.ThreadLaunchService; + const threads = yield* ThreadManagement.ThreadManagementService; + // The first launch, at full access, creates the thread without a message. + const created = yield* launches.launch( + launchInput({ command: "command:launch:narrowed-retry", thread: "thread:narrowed" }), + ); + // The same key comes back from a narrower agent, now carrying a message, + // as a retry or a concurrent launch that lost the create would. + const failed = yield* launches + .launch({ + ...launchInput({ + command: "command:launch:narrowed-retry", + thread: "thread:narrowed", + message: "Run this", + }), + runtimeMode: "approval-required", + interactionMode: "plan", + createdBy: "agent", + creationSource: "mcp", + }) + .pipe(Effect.flip); + if (failed._tag !== "ThreadLaunchError") { + assert.fail(`expected ThreadLaunchError, got ${failed._tag}`); + } + assert.equal(failed.operation, "dispatch-message"); + const projection = yield* threads.getThreadProjection(created.threadId); + assert.equal(projection.runs.length, 0); + assert.equal(projection.messages.length, 0); + }).pipe(Effect.provide(harness.layer)); +}); + it.effect("rejects a server-allocated launch replay with a mismatching thread id", () => { const harness = makeHarness(); return Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.ts index ff032f757bbe..763ef25b4362 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.ts @@ -49,6 +49,7 @@ import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import type * as Orchestrator from "./Orchestrator.ts"; import { makeProviderFailure } from "@t3tools/provider-core/server/failure"; import { randomUuidV4 } from "@t3tools/provider-core/server/randomUuid"; +import { DispatchModeLimit } from "./DispatchModeLimit.ts"; import * as ThreadManagement from "./ThreadManagementService.ts"; export type ThreadLaunchWorkspaceStrategy = @@ -929,7 +930,6 @@ const make = Effect.gen(function* () { if (project.id !== input.projectId) { return yield* mapError(input, "resolve-project", threadId)("Project identity changed."); } - let runId: RunId | null = null; let messageWasAlreadyAccepted = false; if (input.initialMessage !== undefined) { @@ -962,7 +962,19 @@ const make = Effect.gen(function* () { createdBy: input.createdBy, creationSource: input.creationSource, }) - .pipe(Effect.mapError(mapError(input, "dispatch-message", threadId))); + .pipe( + // An agent's launch can land on a thread that a replay, or a concurrent + // launch with the same key, created with broader modes. The orchestrator + // refuses the message under the thread's lock; the user's own clients + // launch into their own drafts, whatever modes those hold. + input.createdBy === "agent" + ? Effect.provideService(DispatchModeLimit, { + runtimeMode: input.runtimeMode, + interactionMode: input.interactionMode, + }) + : (effect) => effect, + Effect.mapError(mapError(input, "dispatch-message", threadId)), + ); // A concurrent retry can read no receipt and still replay the dispatch. messageWasAlreadyAccepted ||= dispatched.replayed === true; const runCreated = dispatched.storedEvents.find( diff --git a/packages/contracts/src/orchestratorMcp.ts b/packages/contracts/src/orchestratorMcp.ts index df65ee6ca320..f114d1529ab7 100644 --- a/packages/contracts/src/orchestratorMcp.ts +++ b/packages/contracts/src/orchestratorMcp.ts @@ -42,7 +42,7 @@ import { ProviderDriverKind, ProviderInstanceId } from "./providerInstance.ts"; const OrchestratorMcpPrompt = TrimmedNonEmptyString.check(Schema.isMaxLength(120_000)); const OrchestratorMcpTitle = TrimmedNonEmptyString.check(Schema.isMaxLength(512)); -const OrchestratorMcpClientRequestId = TrimmedNonEmptyString.check( +export const OrchestratorMcpClientRequestId = TrimmedNonEmptyString.check( Schema.isMaxLength(256), ).annotate({ description: "Reuse it to retry the same call. A new call gets a new key." });