diff --git a/apps/server/src/git/GitWorkflowService.ts b/apps/server/src/git/GitWorkflowService.ts index 100b9beadba..0bfa616a4fd 100644 --- a/apps/server/src/git/GitWorkflowService.ts +++ b/apps/server/src/git/GitWorkflowService.ts @@ -65,6 +65,7 @@ export class GitWorkflowService extends Context.Service< readonly createWorktree: ( input: VcsCreateWorktreeInput, ) => Effect.Effect; + readonly listLocalBranchNames: (cwd: string) => Effect.Effect; readonly fetchRemote: (input: { readonly cwd: string; readonly remoteName: string; @@ -299,6 +300,10 @@ export const make = Effect.gen(function* () { ensureGitCommand("GitWorkflowService.createWorktree", input.cwd).pipe( Effect.andThen(git.createWorktree(input)), ), + listLocalBranchNames: (cwd) => + ensureGitCommand("GitWorkflowService.listLocalBranchNames", cwd).pipe( + Effect.andThen(git.listLocalBranchNames(cwd)), + ), fetchRemote: (input) => ensureGitCommand("GitWorkflowService.fetchRemote", input.cwd).pipe( Effect.andThen(git.fetchRemote(input)), diff --git a/apps/server/src/mcp/McpHttpServer.ts b/apps/server/src/mcp/McpHttpServer.ts index 01f58d07012..3a236937232 100644 --- a/apps/server/src/mcp/McpHttpServer.ts +++ b/apps/server/src/mcp/McpHttpServer.ts @@ -25,6 +25,9 @@ import { PreviewSnapshotToolkit, PreviewStandardToolkit, } from "./toolkits/preview/tools.ts"; +import { WorktreeToolkitHandlersLive } from "./toolkits/worktree/handlers.ts"; +import { WorktreeToolkit } from "./toolkits/worktree/tools.ts"; +import * as WorktreeMcpService from "./WorktreeMcpService.ts"; const unauthorized = HttpServerResponse.jsonUnsafe( { @@ -216,6 +219,11 @@ export const OrchestratorToolkitRegistrationLive = McpServer.toolkit(Orchestrato Layer.provide(OrchestratorMcpService.layer), ); +export const WorktreeToolkitRegistrationLive = McpServer.toolkit(WorktreeToolkit).pipe( + Layer.provide(WorktreeToolkitHandlersLive), + Layer.provide(WorktreeMcpService.layer), +); + const McpTransportLive = McpServer.layerHttp({ name: "T3 Code", version: packageJson.version, @@ -225,4 +233,5 @@ const McpTransportLive = McpServer.layerHttp({ export const layer = Layer.mergeAll( PreviewToolkitRegistrationLive, OrchestratorToolkitRegistrationLive, + WorktreeToolkitRegistrationLive, ).pipe(Layer.provideMerge(McpTransportLive)); diff --git a/apps/server/src/mcp/McpInvocationContext.ts b/apps/server/src/mcp/McpInvocationContext.ts index 6e2c087d05e..4f28874f199 100644 --- a/apps/server/src/mcp/McpInvocationContext.ts +++ b/apps/server/src/mcp/McpInvocationContext.ts @@ -7,7 +7,8 @@ import { import * as Context from "effect/Context"; import * as Effect from "effect/Effect"; -export type McpCapability = "preview" | "orchestration"; +export const ALL_MCP_CAPABILITIES = ["preview", "orchestration", "worktree"] as const; +export type McpCapability = (typeof ALL_MCP_CAPABILITIES)[number]; export interface McpInvocationScope { readonly environmentId: EnvironmentId; diff --git a/apps/server/src/mcp/McpSessionRegistry.test.ts b/apps/server/src/mcp/McpSessionRegistry.test.ts index 4fb273889df..f8e275461a5 100644 --- a/apps/server/src/mcp/McpSessionRegistry.test.ts +++ b/apps/server/src/mcp/McpSessionRegistry.test.ts @@ -43,7 +43,7 @@ it.effect("stores only a token hash, resolves the bearer token, and revokes by t const resolved = yield* registry.resolve(token); expect(resolved?.threadId).toBe(threadId); - expect(resolved?.capabilities).toEqual(new Set(["preview", "orchestration"])); + expect(resolved?.capabilities).toEqual(new Set(["preview", "orchestration", "worktree"])); yield* registry.revokeThread(threadId); expect(yield* registry.resolve(token)).toBeUndefined(); diff --git a/apps/server/src/mcp/McpSessionRegistry.ts b/apps/server/src/mcp/McpSessionRegistry.ts index c79685f79c9..069cf608345 100644 --- a/apps/server/src/mcp/McpSessionRegistry.ts +++ b/apps/server/src/mcp/McpSessionRegistry.ts @@ -94,7 +94,7 @@ const makeWithOptions = Effect.fn("McpSessionRegistry.make")(function* ( threadId: ThreadId.make(request.threadId), providerSessionId, providerInstanceId: ProviderInstanceId.make(request.providerInstanceId), - capabilities: new Set(["preview", "orchestration"]), + capabilities: new Set(McpInvocationContext.ALL_MCP_CAPABILITIES), issuedAt, }; yield* SynchronizedRef.update(state, ({ records }) => { diff --git a/apps/server/src/mcp/WorktreeMcpService.test.ts b/apps/server/src/mcp/WorktreeMcpService.test.ts new file mode 100644 index 00000000000..1917d9f507a --- /dev/null +++ b/apps/server/src/mcp/WorktreeMcpService.test.ts @@ -0,0 +1,1032 @@ +import { describe, expect, it, vi } from "@effect/vitest"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { + CommandId, + EnvironmentId, + type OrchestrationV2ThreadProjection, + type Project, + ProjectId, + ProviderInstanceId, + ThreadId, + WorktreeMcpHandoffInput, +} from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as Fiber from "effect/Fiber"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Path from "effect/Path"; +import * as Schema from "effect/Schema"; + +import * as GitWorkflowService from "../git/GitWorkflowService.ts"; +import { + OrchestratorDispatchError, + OrchestratorProjectionError, +} from "../orchestration-v2/Orchestrator.ts"; +import { + ThreadManagementError, + ThreadManagementService, + type ThreadManagementSendResult, +} from "../orchestration-v2/ThreadManagementService.ts"; +import * as ProjectService from "../project/ProjectService.ts"; +import * as ProjectSetupScriptRunner from "../project/ProjectSetupScriptRunner.ts"; +import * as ServerSettings from "../serverSettings.ts"; +import { VcsStatusBroadcaster } from "../vcs/VcsStatusBroadcaster.ts"; +import type * as McpInvocationContext from "./McpInvocationContext.ts"; +import { layer as worktreeMcpServiceLayer, WorktreeMcpService } from "./WorktreeMcpService.ts"; + +const environmentId = EnvironmentId.make("environment-worktree-test"); +const threadId = ThreadId.make("thread-worktree-test"); +const projectId = ProjectId.make("project-worktree-test"); +const workspaceRoot = "/repo/project"; + +const makeScope = ( + capabilities: ReadonlySet, +): McpInvocationContext.McpInvocationScope => ({ + environmentId, + threadId, + providerSessionId: "provider-session-worktree-test", + providerInstanceId: ProviderInstanceId.make("claudeAgent"), + capabilities, + issuedAt: 1, +}); + +interface ThreadFixture { + readonly branch?: string | null; + readonly worktreePath?: string | null; + readonly archivedAt?: string | null; + readonly deletedAt?: string | null; +} + +const makeProjection = (overrides: ThreadFixture = {}): OrchestrationV2ThreadProjection => + ({ + thread: { + id: threadId, + projectId, + title: "Worktree test thread", + branch: null, + worktreePath: null, + archivedAt: null, + deletedAt: null, + ...overrides, + }, + }) as OrchestrationV2ThreadProjection; + +const project: Project = { + id: projectId, + title: "Worktree test project", + workspaceRoot, + defaultModelSelection: null, + scripts: [], + createdAt: "2026-01-01T00:00:00.000Z", + updatedAt: "2026-01-01T00:00:00.000Z", + deletedAt: null, +} as Project; + +interface HarnessOptions { + readonly thread?: ThreadFixture | null; + readonly threadReadError?: "projection" | "dispatch"; + readonly capabilities?: ReadonlySet; + readonly currentBranch?: string | null; + readonly notARepo?: boolean; + readonly newWorktreesStartFromOrigin?: boolean; + readonly setupScript?: "started" | "no-script" | "fails" | "dies"; + readonly dispatchFails?: boolean; + readonly dispatchDies?: boolean; + readonly dispatchInterrupts?: boolean; + readonly dispatchGate?: Effect.Effect; + readonly threadAttachedOnRecheck?: boolean; + readonly threadArchivedOnRecheck?: boolean; + readonly threadReadFailsOnRecheck?: boolean; + readonly continuation?: "queued" | "fails" | "dies"; + readonly projectMissing?: boolean; + readonly projectReadFails?: boolean; + readonly existingBranchWorktreePath?: string | null; + readonly pathSemantics?: "win32" | "posix"; + readonly createWorktreeFails?: boolean; + readonly fetchRemoteFails?: boolean; + readonly resolveRemoteFails?: boolean; + readonly removeWorktreeFails?: boolean; + readonly createWorktreeGate?: Effect.Effect; +} + +const makeHarness = (options: HarnessOptions = {}) => { + const thread = options.thread === undefined ? {} : options.thread; + const scope = makeScope(options.capabilities ?? new Set(["preview", "worktree"])); + const dispatch = vi.fn((_: unknown) => + (options.dispatchGate ?? Effect.void).pipe( + Effect.andThen( + options.dispatchInterrupts + ? (Effect.failCause(Cause.interrupt()) as never) + : options.dispatchDies + ? Effect.die(new Error("dispatch defect")) + : options.dispatchFails + ? (Effect.fail("simulated dispatch failure") as never) + : Effect.succeed({ sequence: 1, storedEvents: [] }), + ), + ), + ); + const listLocalBranchNames = vi.fn((_: string) => + Effect.succeed( + options.existingBranchWorktreePath === undefined + ? ["dev"] + : ["dev", "feature/taken", "feature/taken-idle"], + ), + ); + const getThreadProjection = vi.fn((id: ThreadId) => { + if (options.threadReadError === "dispatch") { + return Effect.fail( + new OrchestratorDispatchError({ + commandId: CommandId.make("command:test:read"), + commandType: "thread.metadata.update", + }), + ) as never; + } + if (options.threadReadFailsOnRecheck === true && getThreadProjection.mock.calls.length > 1) { + return Effect.fail( + new OrchestratorDispatchError({ + commandId: CommandId.make("command:test:recheck"), + commandType: "thread.metadata.update", + }), + ) as never; + } + if ( + options.threadAttachedOnRecheck === true && + getThreadProjection.mock.calls.length > 1 && + thread !== null + ) { + return Effect.succeed( + makeProjection({ ...thread, worktreePath: "/worktrees/project/raced" }), + ); + } + if ( + options.threadArchivedOnRecheck === true && + getThreadProjection.mock.calls.length > 1 && + thread !== null + ) { + return Effect.succeed(makeProjection({ ...thread, archivedAt: "2026-01-02T00:00:00.000Z" })); + } + return id === threadId && thread !== null + ? Effect.succeed(makeProjection(thread)) + : Effect.fail(new OrchestratorProjectionError({ threadId: id })); + }); + const sendToThread = vi.fn((_: unknown) => { + switch (options.continuation ?? "queued") { + case "fails": + return Effect.fail( + new ThreadManagementError({ + code: "thread_not_sendable", + message: "simulated send failure", + }), + ); + case "dies": + return Effect.die(new Error("send defect")); + default: + return Effect.succeed({ delivery: "queued" } as ThreadManagementSendResult); + } + }); + const getById = vi.fn((id: ProjectId) => + options.projectReadFails + ? (Effect.fail("simulated project read failure") as never) + : Effect.succeed( + id === projectId && options.projectMissing !== true + ? Option.some(project) + : Option.none(), + ), + ); + const removeWorktree = vi.fn((_: unknown) => + options.removeWorktreeFails + ? (Effect.fail("simulated worktree removal failure") as never) + : Effect.void, + ); + const fetchRemote = vi.fn((_: unknown) => + options.fetchRemoteFails ? (Effect.fail("simulated fetch failure") as never) : Effect.void, + ); + const resolveRemoteTrackingCommit = vi.fn((_: unknown) => + options.resolveRemoteFails + ? (Effect.fail("simulated remote resolve failure") as never) + : Effect.succeed({ commitSha: "abc123", remoteRefName: "origin/dev" }), + ); + const createWorktree = vi.fn( + (input: { readonly newRefName?: string | undefined; readonly path: string | null }) => + options.createWorktreeFails + ? (Effect.fail("simulated worktree creation failure") as never) + : (options.createWorktreeGate ?? Effect.void).pipe( + Effect.andThen( + Effect.succeed({ + worktree: { + path: input.path ?? `/worktrees/project/${input.newRefName}`, + refName: input.newRefName ?? "detached", + }, + }), + ), + ), + ); + const listRefs = vi.fn((input: { readonly query?: string | undefined }) => + Effect.succeed({ + refs: + options.existingBranchWorktreePath === undefined + ? [] + : [ + { + name: input.query ?? "", + current: false, + isDefault: false, + worktreePath: options.existingBranchWorktreePath, + }, + ], + isRepo: true, + hasPrimaryRemote: true, + nextCursor: null, + totalCount: options.existingBranchWorktreePath === undefined ? 0 : 1, + }), + ); + const localStatus = vi.fn((_: unknown) => + Effect.succeed({ + isRepo: options.notARepo !== true, + hasPrimaryRemote: true, + isDefaultRef: false, + refName: options.currentBranch === undefined ? "dev" : options.currentBranch, + hasWorkingTreeChanges: false, + workingTree: { files: [], insertions: 0, deletions: 0 }, + }), + ); + const refreshStatus = vi.fn((_: string) => Effect.die("refreshStatus stub")); + const runForThread = vi.fn((input: { readonly worktreePath: string }) => { + switch (options.setupScript ?? "started") { + case "no-script": + return Effect.succeed({ status: "no-script" } as const); + case "dies": + return Effect.die(new Error("setup runner defect")); + case "fails": + return Effect.fail( + new ProjectSetupScriptRunner.ProjectSetupScriptProjectNotFoundError({ + threadId, + worktreePath: input.worktreePath, + }), + ); + default: + return Effect.succeed({ + status: "started", + scriptId: "setup", + scriptName: "Setup", + terminalId: "setup-terminal", + cwd: input.worktreePath, + } as const); + } + }); + + // Optional deterministic Path semantics: providing this BEFORE the general + // mocks means the service resolves Path here rather than from NodeServices, + // so absolute-path validation is testable independently of the host OS. The + // service only calls isAbsolute; the minimal per-platform semantics are + // inlined so the test does not depend on the host's path module. + const win32IsAbsolute = (value: string) => /^(?:[a-zA-Z]:[\\/]|[\\/])/.test(value); + const posixIsAbsolute = (value: string) => value.startsWith("/"); + const serviceLayer = + options.pathSemantics === undefined + ? worktreeMcpServiceLayer + : worktreeMcpServiceLayer.pipe( + Layer.provide( + Layer.succeed(Path.Path, { + isAbsolute: options.pathSemantics === "win32" ? win32IsAbsolute : posixIsAbsolute, + } as unknown as Path.Path), + ), + ); + const layer = serviceLayer.pipe( + Layer.provide( + Layer.mergeAll( + Layer.mock(ThreadManagementService)({ + dispatch, + getThreadProjection, + sendToThread, + } satisfies Partial), + Layer.mock(ProjectService.ProjectService)({ + getById, + } satisfies Partial), + ServerSettings.layerTest({ + newWorktreesStartFromOrigin: options.newWorktreesStartFromOrigin ?? false, + }), + Layer.mock(GitWorkflowService.GitWorkflowService)({ + listRefs, + listLocalBranchNames, + localStatus, + fetchRemote, + resolveRemoteTrackingCommit, + createWorktree, + removeWorktree, + } satisfies Partial), + Layer.mock(ProjectSetupScriptRunner.ProjectSetupScriptRunner)({ + runForThread, + } satisfies Partial), + Layer.mock(VcsStatusBroadcaster)({ + refreshStatus, + } satisfies Partial), + NodeServices.layer, + ), + ), + ); + + return { + layer, + scope, + dispatch, + sendToThread, + fetchRemote, + resolveRemoteTrackingCommit, + createWorktree, + removeWorktree, + localStatus, + runForThread, + }; +}; + +const expectTypedFailure = (exit: Exit.Exit, expected: object): void => { + if (!Exit.isFailure(exit)) { + expect.fail(`Expected a failure exit, got: ${JSON.stringify(exit)}`); + } + const reason = exit.cause.reasons[0]; + if (reason?._tag !== "Fail") { + expect.fail(`Expected a typed Fail cause, got: ${reason?._tag ?? "no reason"}`); + } + expect(reason.error).toMatchObject(expected); +}; + +// Resolves the service once from the harness layer; used by tests that must +// make several calls against the SAME instance (the in-flight guard lives in +// the layer's closure, so a fresh layer per call would never see it). +const resolveService = (harness: ReturnType) => + Effect.gen(function* () { + return yield* WorktreeMcpService; + }).pipe(Effect.provide(harness.layer)); + +const runHandoff = ( + harness: ReturnType, + input: Parameters[1], +) => + Effect.gen(function* () { + const service = yield* WorktreeMcpService; + return yield* service.handoff(harness.scope, input); + }).pipe(Effect.provide(harness.layer)); + +const runStatus = (harness: ReturnType) => + Effect.gen(function* () { + const service = yield* WorktreeMcpService; + return yield* service.status(harness.scope); + }).pipe(Effect.provide(harness.layer)); + +describe("t3_worktree_handoff", () => { + it.effect("creates a worktree from the current branch and re-points the thread", () => { + const harness = makeHarness(); + return Effect.gen(function* () { + const result = yield* runHandoff(harness, { branch: "feature/handoff" }); + + expect(result.branch).toBe("feature/handoff"); + expect(result.baseRef).toBe("dev"); + expect(result.startedFromOrigin).toBe(false); + expect(result.worktreePath).toBe("/worktrees/project/feature/handoff"); + expect(result.setupScript).toMatchObject({ status: "started", scriptName: "Setup" }); + + expect(harness.fetchRemote).not.toHaveBeenCalled(); + expect(harness.createWorktree).toHaveBeenCalledWith({ + cwd: workspaceRoot, + refName: "dev", + newRefName: "feature/handoff", + baseRefName: "dev", + path: null, + }); + expect(harness.dispatch).toHaveBeenCalledWith( + expect.objectContaining({ + type: "thread.metadata.update", + threadId, + branch: "feature/handoff", + worktreePath: "/worktrees/project/feature/handoff", + }), + ); + expect(harness.runForThread).toHaveBeenCalledWith({ + threadId, + projectId, + projectCwd: workspaceRoot, + worktreePath: "/worktrees/project/feature/handoff", + project: { workspaceRoot, scripts: [] }, + }); + }); + }); + + it.effect("skips the continuation when no continuationPrompt is given", () => { + const harness = makeHarness(); + return Effect.gen(function* () { + const result = yield* runHandoff(harness, { branch: "feature/no-continuation" }); + expect(result.continuation).toEqual({ status: "skipped" }); + expect(harness.sendToThread).not.toHaveBeenCalled(); + }); + }); + + it.effect("queues the continuation prompt as the thread's next message", () => { + const harness = makeHarness(); + return Effect.gen(function* () { + const result = yield* runHandoff(harness, { + branch: "feature/continue", + continuationPrompt: "Keep fixing the login bug in the new worktree.", + }); + + expect(result.continuation).toEqual({ status: "scheduled", delivery: "queued" }); + expect(harness.sendToThread).toHaveBeenCalledWith( + expect.objectContaining({ + projectId, + threadId, + text: "Keep fixing the login bug in the new worktree.", + mode: "queue", + createdBy: "agent", + creationSource: "mcp", + }), + ); + // The continuation must be durably queued before anything slower runs. + expect(harness.sendToThread.mock.invocationCallOrder[0]).toBeLessThan( + harness.runForThread.mock.invocationCallOrder[0]!, + ); + }); + }); + + it.effect("reports a continuation failure without failing the handoff", () => { + const harness = makeHarness({ continuation: "fails" }); + return Effect.gen(function* () { + const result = yield* runHandoff(harness, { + branch: "feature/continue-fails", + continuationPrompt: "Keep going.", + }); + expect(result.continuation).toMatchObject({ status: "failed" }); + expect(result.worktreePath).toBe("/worktrees/project/feature/continue-fails"); + expect(harness.dispatch).toHaveBeenCalled(); + }); + }); + + it.effect("reports a continuation defect without failing the handoff", () => { + const harness = makeHarness({ continuation: "dies" }); + return Effect.gen(function* () { + const result = yield* runHandoff(harness, { + branch: "feature/continue-dies", + continuationPrompt: "Keep going.", + }); + expect(result.continuation).toEqual({ status: "failed", detail: "send defect" }); + expect(result.setupScript).toMatchObject({ status: "started" }); + }); + }); + + it.effect("does not queue a continuation when the thread update fails", () => { + const harness = makeHarness({ dispatchFails: true }); + return Effect.gen(function* () { + const exit = yield* Effect.exit( + runHandoff(harness, { + branch: "feature/dispatch-fails-continue", + continuationPrompt: "Keep going.", + }), + ); + expectTypedFailure(exit, { _tag: "WorktreeMcpFailure", code: "operation_failed" }); + expect(harness.sendToThread).not.toHaveBeenCalled(); + }); + }); + + it.effect("starts from origin and honors explicit baseRef and path", () => { + const harness = makeHarness(); + return Effect.gen(function* () { + const result = yield* runHandoff(harness, { + branch: "feature/from-origin", + baseRef: "dev", + startFromOrigin: true, + path: "/custom/worktree/location", + runSetupScript: false, + }); + + expect(harness.fetchRemote).toHaveBeenCalledWith({ + cwd: workspaceRoot, + remoteName: "origin", + }); + expect(harness.resolveRemoteTrackingCommit).toHaveBeenCalledWith({ + cwd: workspaceRoot, + refName: "dev", + fallbackRemoteName: "origin", + }); + expect(harness.createWorktree).toHaveBeenCalledWith({ + cwd: workspaceRoot, + refName: "abc123", + newRefName: "feature/from-origin", + baseRefName: "dev", + path: "/custom/worktree/location", + }); + // localStatus is always consulted now (repo pre-check), but its branch + // must not override the explicit baseRef. + expect(harness.localStatus).toHaveBeenCalled(); + expect(harness.runForThread).not.toHaveBeenCalled(); + expect(result.worktreePath).toBe("/custom/worktree/location"); + expect(result.startedFromOrigin).toBe(true); + expect(result.setupScript).toEqual({ status: "skipped" }); + }); + }); + + it.effect("uses the server setting for startFromOrigin when unspecified", () => { + const harness = makeHarness({ newWorktreesStartFromOrigin: true }); + return Effect.gen(function* () { + const result = yield* runHandoff(harness, { branch: "feature/settings-origin" }); + expect(result.startedFromOrigin).toBe(true); + expect(harness.fetchRemote).toHaveBeenCalled(); + }); + }); + + it.effect("fails when the thread is already attached to a worktree", () => { + const harness = makeHarness({ + thread: { branch: "feature/existing", worktreePath: "/worktrees/project/existing" }, + }); + return Effect.gen(function* () { + const exit = yield* Effect.exit(runHandoff(harness, { branch: "feature/second" })); + expectTypedFailure(exit, { + _tag: "WorktreeMcpFailure", + code: "already_in_worktree", + }); + expect(harness.createWorktree).not.toHaveBeenCalled(); + }); + }); + + it.effect("fails when the thread does not exist", () => { + const harness = makeHarness({ thread: null }); + return Effect.gen(function* () { + const exit = yield* Effect.exit(runHandoff(harness, { branch: "feature/missing" })); + expectTypedFailure(exit, { _tag: "WorktreeMcpFailure", code: "thread_not_found" }); + expect(harness.createWorktree).not.toHaveBeenCalled(); + }); + }); + + it.effect("treats a soft-deleted thread as not found", () => { + const harness = makeHarness({ thread: { deletedAt: "2026-01-02T00:00:00.000Z" } }); + return Effect.gen(function* () { + const exit = yield* Effect.exit(runHandoff(harness, { branch: "feature/deleted" })); + expectTypedFailure(exit, { _tag: "WorktreeMcpFailure", code: "thread_not_found" }); + expect(harness.createWorktree).not.toHaveBeenCalled(); + }); + }); + + it.effect("maps a non-projection orchestrator error to operation_failed", () => { + const harness = makeHarness({ threadReadError: "dispatch" }); + return Effect.gen(function* () { + const exit = yield* Effect.exit(runHandoff(harness, { branch: "feature/read-error" })); + expectTypedFailure(exit, { _tag: "WorktreeMcpFailure", code: "operation_failed" }); + expect(harness.createWorktree).not.toHaveBeenCalled(); + }); + }); + + it.effect("fails when the project does not exist", () => { + const harness = makeHarness({ projectMissing: true }); + return Effect.gen(function* () { + const exit = yield* Effect.exit(runHandoff(harness, { branch: "feature/no-project" })); + expectTypedFailure(exit, { _tag: "WorktreeMcpFailure", code: "project_not_found" }); + expect(harness.createWorktree).not.toHaveBeenCalled(); + }); + }); + + it.effect("maps a project read error to operation_failed", () => { + const harness = makeHarness({ projectReadFails: true }); + return Effect.gen(function* () { + const exit = yield* Effect.exit(runHandoff(harness, { branch: "feature/project-error" })); + expectTypedFailure(exit, { _tag: "WorktreeMcpFailure", code: "operation_failed" }); + expect(harness.createWorktree).not.toHaveBeenCalled(); + }); + }); + + it.effect("fails when the project workspace is not a git repository", () => { + const harness = makeHarness({ notARepo: true }); + return Effect.gen(function* () { + const exit = yield* Effect.exit(runHandoff(harness, { branch: "feature/no-repo" })); + expectTypedFailure(exit, { _tag: "WorktreeMcpFailure", code: "invalid_request" }); + expect(harness.createWorktree).not.toHaveBeenCalled(); + }); + }); + + it.effect("rejects an existing branch with an actionable error naming its checkout", () => { + const harness = makeHarness({ existingBranchWorktreePath: "/elsewhere/checkout" }); + return Effect.gen(function* () { + const exit = yield* Effect.exit(runHandoff(harness, { branch: "feature/taken" })); + if (!Exit.isFailure(exit)) { + return expect.fail("expected a failure exit"); + } + const reason = exit.cause.reasons[0]; + expect(reason?._tag).toBe("Fail"); + const error = (reason as { readonly error: { code: string; message: string } }).error; + expect(error.code).toBe("invalid_request"); + expect(error.message).toContain("feature/taken"); + expect(error.message).toContain("already exists"); + expect(error.message).toContain("/elsewhere/checkout"); + expect(harness.createWorktree).not.toHaveBeenCalled(); + expect(harness.fetchRemote).not.toHaveBeenCalled(); + }); + }); + + it.effect("rejects an existing branch that is not checked out anywhere", () => { + const harness = makeHarness({ existingBranchWorktreePath: null }); + return Effect.gen(function* () { + const exit = yield* Effect.exit(runHandoff(harness, { branch: "feature/taken-idle" })); + expectTypedFailure(exit, { _tag: "WorktreeMcpFailure", code: "invalid_request" }); + expect(harness.createWorktree).not.toHaveBeenCalled(); + }); + }); + + it.effect("does not trip the pre-flight on similarly named branches", () => { + const harness = makeHarness({ existingBranchWorktreePath: null }); + return Effect.gen(function* () { + // "feature/taken" exists in the mock branch list; "feature/take" does + // not, and a substring-based check would wrongly match it. + const result = yield* runHandoff(harness, { branch: "feature/take" }); + expect(result.branch).toBe("feature/take"); + expect(harness.createWorktree).toHaveBeenCalledTimes(1); + }); + }); + + it.effect("rejects a non-repository workspace even when baseRef is explicit", () => { + const harness = makeHarness({ notARepo: true }); + return Effect.gen(function* () { + const exit = yield* Effect.exit( + runHandoff(harness, { branch: "feature/no-repo-baseref", baseRef: "dev" }), + ); + expectTypedFailure(exit, { _tag: "WorktreeMcpFailure", code: "invalid_request" }); + expect(harness.createWorktree).not.toHaveBeenCalled(); + }); + }); + + it.effect("propagates an interrupted dispatch without rolling back the worktree", () => { + const harness = makeHarness({ dispatchInterrupts: true }); + return Effect.gen(function* () { + const exit = yield* Effect.exit(runHandoff(harness, { branch: "feature/dispatch-int" })); + if (!Exit.isFailure(exit)) { + return expect.fail("expected a failure exit"); + } + // Whether the binding committed is unknown on interruption, so the + // worktree must not be force-deleted and no typed failure is invented. + expect(Cause.hasInterruptsOnly(exit.cause)).toBe(true); + expect(harness.removeWorktree).not.toHaveBeenCalled(); + }); + }); + + it.effect("maps a worktree creation failure to operation_failed", () => { + const harness = makeHarness({ createWorktreeFails: true }); + return Effect.gen(function* () { + const exit = yield* Effect.exit(runHandoff(harness, { branch: "feature/create-fails" })); + expectTypedFailure(exit, { _tag: "WorktreeMcpFailure", code: "operation_failed" }); + expect(harness.dispatch).not.toHaveBeenCalled(); + expect(harness.removeWorktree).not.toHaveBeenCalled(); + }); + }); + + it.effect("maps an origin fetch failure to operation_failed", () => { + const harness = makeHarness({ fetchRemoteFails: true }); + return Effect.gen(function* () { + const exit = yield* Effect.exit( + runHandoff(harness, { branch: "feature/fetch-fails", startFromOrigin: true }), + ); + expectTypedFailure(exit, { _tag: "WorktreeMcpFailure", code: "operation_failed" }); + expect(harness.createWorktree).not.toHaveBeenCalled(); + }); + }); + + it.effect("maps a remote-tracking resolve failure to operation_failed", () => { + const harness = makeHarness({ resolveRemoteFails: true }); + return Effect.gen(function* () { + const exit = yield* Effect.exit( + runHandoff(harness, { branch: "feature/resolve-fails", startFromOrigin: true }), + ); + expectTypedFailure(exit, { _tag: "WorktreeMcpFailure", code: "operation_failed" }); + expect(harness.createWorktree).not.toHaveBeenCalled(); + }); + }); + + it.effect("removes the created worktree when the thread update dies with a defect", () => { + const harness = makeHarness({ dispatchDies: true }); + return Effect.gen(function* () { + const exit = yield* Effect.exit(runHandoff(harness, { branch: "feature/dispatch-defect" })); + expectTypedFailure(exit, { _tag: "WorktreeMcpFailure", code: "operation_failed" }); + expect(harness.removeWorktree).toHaveBeenCalledWith({ + cwd: workspaceRoot, + path: "/worktrees/project/feature/dispatch-defect", + force: true, + }); + }); + }); + + it.effect("re-checks attachment after creating the worktree and backs out on a race", () => { + const harness = makeHarness({ threadAttachedOnRecheck: true }); + return Effect.gen(function* () { + const exit = yield* Effect.exit(runHandoff(harness, { branch: "feature/raced" })); + expectTypedFailure(exit, { _tag: "WorktreeMcpFailure", code: "already_in_worktree" }); + expect(harness.createWorktree).toHaveBeenCalledTimes(1); + // The freshly created worktree must not be left orphaned. + expect(harness.removeWorktree).toHaveBeenCalledWith({ + cwd: workspaceRoot, + path: "/worktrees/project/feature/raced", + force: true, + }); + expect(harness.dispatch).not.toHaveBeenCalled(); + }); + }); + + it.effect("removes the created worktree when the recheck read fails", () => { + const harness = makeHarness({ threadReadFailsOnRecheck: true }); + return Effect.gen(function* () { + const exit = yield* Effect.exit(runHandoff(harness, { branch: "feature/recheck-fails" })); + expectTypedFailure(exit, { _tag: "WorktreeMcpFailure", code: "operation_failed" }); + expect(harness.removeWorktree).toHaveBeenCalledWith({ + cwd: workspaceRoot, + path: "/worktrees/project/feature/recheck-fails", + force: true, + }); + expect(harness.dispatch).not.toHaveBeenCalled(); + }); + }); + + it.effect("backs out when the thread is archived during worktree creation", () => { + const harness = makeHarness({ threadArchivedOnRecheck: true }); + return Effect.gen(function* () { + const exit = yield* Effect.exit(runHandoff(harness, { branch: "feature/archived-race" })); + expectTypedFailure(exit, { _tag: "WorktreeMcpFailure", code: "invalid_request" }); + expect(harness.dispatch).not.toHaveBeenCalled(); + // The created worktree must not be left orphaned. + expect(harness.removeWorktree).toHaveBeenCalledWith({ + cwd: workspaceRoot, + path: "/worktrees/project/feature/archived-race", + force: true, + }); + }); + }); + + it.effect("rejects a handoff for an archived thread", () => { + const harness = makeHarness({ thread: { archivedAt: "2026-01-02T00:00:00.000Z" } }); + return Effect.gen(function* () { + const exit = yield* Effect.exit(runHandoff(harness, { branch: "feature/archived" })); + expectTypedFailure(exit, { _tag: "WorktreeMcpFailure", code: "invalid_request" }); + expect(harness.createWorktree).not.toHaveBeenCalled(); + }); + }); + + it.effect("queues the continuation even when interrupted during the binding dispatch", () => + Effect.gen(function* () { + const gate = yield* Deferred.make(); + const harness = makeHarness({ dispatchGate: Deferred.await(gate) }); + + // Interrupt arrives while the metadata dispatch is in flight; the + // binding-plus-continuation section must run to completion anyway so the + // continuation is never lost between the commit and the queue. + const fiber = yield* Effect.forkChild( + runHandoff(harness, { + branch: "feature/interrupted", + continuationPrompt: "Keep going in the worktree.", + }), + ); + yield* Effect.yieldNow; + const interruption = yield* Effect.forkChild(Fiber.interrupt(fiber)); + yield* Effect.yieldNow; + yield* Deferred.succeed(gate, undefined); + yield* Fiber.await(fiber); + yield* Fiber.join(interruption); + + expect(harness.dispatch).toHaveBeenCalledTimes(1); + expect(harness.sendToThread).toHaveBeenCalledTimes(1); + // The setup script must also survive the pending interrupt; otherwise + // the continuation run starts in a worktree that was never set up. + expect(harness.runForThread).toHaveBeenCalledTimes(1); + }), + ); + + it.effect("still fails with a typed error when the rollback removal also fails", () => { + const harness = makeHarness({ dispatchFails: true, removeWorktreeFails: true }); + return Effect.gen(function* () { + const exit = yield* Effect.exit(runHandoff(harness, { branch: "feature/rollback-fails" })); + expectTypedFailure(exit, { _tag: "WorktreeMcpFailure", code: "operation_failed" }); + expect(harness.removeWorktree).toHaveBeenCalledTimes(1); + }); + }); + + it.effect("accepts a Windows drive path under win32 path semantics", () => { + const harness = makeHarness({ pathSemantics: "win32" }); + return Effect.gen(function* () { + const result = yield* runHandoff(harness, { + branch: "feature/windows-path", + path: "C:\\worktrees\\custom", + }); + expect(result.worktreePath).toBe("C:\\worktrees\\custom"); + expect(harness.createWorktree).toHaveBeenCalledWith( + expect.objectContaining({ path: "C:\\worktrees\\custom" }), + ); + }); + }); + + it.effect("rejects a Windows drive path under posix path semantics", () => { + const harness = makeHarness({ pathSemantics: "posix" }); + return Effect.gen(function* () { + // The schema-level pattern admits drive paths cross-platform; the + // runtime check reflects the host the worktree would be created on. + const exit = yield* Effect.exit( + runHandoff(harness, { branch: "feature/windows-path", path: "C:\\worktrees\\custom" }), + ); + expectTypedFailure(exit, { _tag: "WorktreeMcpFailure", code: "invalid_request" }); + expect(harness.createWorktree).not.toHaveBeenCalled(); + }); + }); + + it.effect("releases the per-thread guard after a failed handoff", () => { + const harness = makeHarness({ + thread: { worktreePath: "/worktrees/project/existing" }, + }); + return Effect.gen(function* () { + const service = yield* resolveService(harness); + + const first = yield* Effect.exit( + service.handoff(harness.scope, { branch: "feature/guard-1" }), + ); + expectTypedFailure(first, { _tag: "WorktreeMcpFailure", code: "already_in_worktree" }); + // A leaked guard would surface as handoff_in_progress here. + const second = yield* Effect.exit( + service.handoff(harness.scope, { branch: "feature/guard-2" }), + ); + expectTypedFailure(second, { _tag: "WorktreeMcpFailure", code: "already_in_worktree" }); + }); + }); + + it.effect("fails when the worktree capability is missing", () => { + const harness = makeHarness({ capabilities: new Set(["preview"]) }); + return Effect.gen(function* () { + const exit = yield* Effect.exit(runHandoff(harness, { branch: "feature/no-capability" })); + expectTypedFailure(exit, { _tag: "WorktreeMcpFailure", code: "capability_denied" }); + }); + }); + + it.effect("serializes concurrent handoffs for the same thread", () => + Effect.gen(function* () { + const gate = yield* Deferred.make(); + const harness = makeHarness({ createWorktreeGate: Deferred.await(gate) }); + + const service = yield* resolveService(harness); + + // First handoff acquires the per-thread guard and blocks on the gate. + const first = yield* Effect.forkChild( + Effect.exit(service.handoff(harness.scope, { branch: "feature/race-1" })), + ); + yield* Effect.yieldNow; + + // Second handoff for the same thread must be refused while the first + // is still in flight. + const second = yield* Effect.exit( + service.handoff(harness.scope, { branch: "feature/race-2" }), + ); + expectTypedFailure(second, { _tag: "WorktreeMcpFailure", code: "handoff_in_progress" }); + + yield* Deferred.succeed(gate, undefined); + const firstExit = yield* Fiber.join(first); + expect(Exit.isSuccess(firstExit)).toBe(true); + expect(harness.createWorktree).toHaveBeenCalledTimes(1); + }), + ); + + it.effect("removes the created worktree when the thread update fails", () => { + const harness = makeHarness({ dispatchFails: true }); + return Effect.gen(function* () { + const exit = yield* Effect.exit(runHandoff(harness, { branch: "feature/dispatch-fails" })); + expectTypedFailure(exit, { _tag: "WorktreeMcpFailure", code: "operation_failed" }); + expect(harness.createWorktree).toHaveBeenCalledTimes(1); + expect(harness.removeWorktree).toHaveBeenCalledWith({ + cwd: workspaceRoot, + path: "/worktrees/project/feature/dispatch-fails", + force: true, + }); + }); + }); + + it.effect("rejects a relative path", () => { + const harness = makeHarness(); + return Effect.gen(function* () { + const exit = yield* Effect.exit( + runHandoff(harness, { branch: "feature/relative-path", path: "worktrees/nested" }), + ); + expectTypedFailure(exit, { _tag: "WorktreeMcpFailure", code: "invalid_request" }); + expect(harness.createWorktree).not.toHaveBeenCalled(); + }); + }); + + it.effect("fails when baseRef is omitted and HEAD is detached", () => { + const harness = makeHarness({ currentBranch: null }); + return Effect.gen(function* () { + const exit = yield* Effect.exit(runHandoff(harness, { branch: "feature/detached" })); + expectTypedFailure(exit, { _tag: "WorktreeMcpFailure", code: "invalid_request" }); + }); + }); + + it.effect("reports setup script failure without failing the handoff", () => { + const harness = makeHarness({ setupScript: "fails" }); + return Effect.gen(function* () { + const result = yield* runHandoff(harness, { branch: "feature/setup-fails" }); + expect(result.setupScript.status).toBe("failed"); + expect(harness.dispatch).toHaveBeenCalled(); + }); + }); + + it.effect("reports a setup script defect without failing the handoff", () => { + const harness = makeHarness({ setupScript: "dies" }); + return Effect.gen(function* () { + const result = yield* runHandoff(harness, { branch: "feature/setup-dies" }); + expect(result.setupScript).toEqual({ status: "failed", detail: "setup runner defect" }); + expect(harness.dispatch).toHaveBeenCalled(); + }); + }); +}); + +describe("t3_worktree_status", () => { + it.effect("reports an unattached thread", () => { + const harness = makeHarness({ newWorktreesStartFromOrigin: true }); + return Effect.gen(function* () { + const result = yield* runStatus(harness); + expect(result).toEqual({ + attached: false, + worktreePath: null, + branch: null, + projectWorkspaceRoot: workspaceRoot, + defaultStartFromOrigin: true, + }); + }); + }); + + it.effect("reports an attached thread's worktree and branch", () => { + const harness = makeHarness({ + thread: { + worktreePath: "/worktrees/project/existing", + branch: "feature/existing", + }, + }); + return Effect.gen(function* () { + const result = yield* runStatus(harness); + expect(result).toMatchObject({ + attached: true, + worktreePath: "/worktrees/project/existing", + branch: "feature/existing", + defaultStartFromOrigin: false, + }); + }); + }); + + it.effect("fails when the worktree capability is missing", () => { + const harness = makeHarness({ capabilities: new Set(["preview"]) }); + return Effect.gen(function* () { + const exit = yield* Effect.exit(runStatus(harness)); + expectTypedFailure(exit, { _tag: "WorktreeMcpFailure", code: "capability_denied" }); + }); + }); + + it.effect("fails when the thread does not exist", () => { + const harness = makeHarness({ thread: null }); + return Effect.gen(function* () { + const exit = yield* Effect.exit(runStatus(harness)); + expectTypedFailure(exit, { _tag: "WorktreeMcpFailure", code: "thread_not_found" }); + }); + }); + + it.effect("fails when the project does not exist", () => { + const harness = makeHarness({ projectMissing: true }); + return Effect.gen(function* () { + const exit = yield* Effect.exit(runStatus(harness)); + expectTypedFailure(exit, { _tag: "WorktreeMcpFailure", code: "project_not_found" }); + }); + }); +}); + +describe("WorktreeMcpHandoffInput schema", () => { + const decode = Schema.decodeUnknownEffect(WorktreeMcpHandoffInput); + + it.effect("accepts POSIX, Windows drive, and UNC absolute paths", () => + Effect.gen(function* () { + for (const path of ["/abs/posix", "C:\\abs\\drive", "C:/abs/drive", "\\\\host\\share"]) { + const decoded = yield* decode({ branch: "feature/x", path }); + expect(decoded.path).toBe(path); + } + }), + ); + + it.effect("rejects relative paths", () => + Effect.gen(function* () { + for (const path of ["worktrees/nested", "./nested", "../sibling"]) { + const exit = yield* Effect.exit(decode({ branch: "feature/x", path })); + expect(Exit.isFailure(exit), `path '${path}' should be rejected`).toBe(true); + } + }), + ); + + it.effect("rejects a missing or blank branch", () => + Effect.gen(function* () { + expect(Exit.isFailure(yield* Effect.exit(decode({})))).toBe(true); + expect(Exit.isFailure(yield* Effect.exit(decode({ branch: " " })))).toBe(true); + }), + ); + + it.effect("rejects a blank continuationPrompt", () => + Effect.gen(function* () { + const exit = yield* Effect.exit(decode({ branch: "feature/x", continuationPrompt: " " })); + expect(Exit.isFailure(exit)).toBe(true); + }), + ); +}); diff --git a/apps/server/src/mcp/WorktreeMcpService.ts b/apps/server/src/mcp/WorktreeMcpService.ts new file mode 100644 index 00000000000..1053c9cdeb9 --- /dev/null +++ b/apps/server/src/mcp/WorktreeMcpService.ts @@ -0,0 +1,481 @@ +import { + CommandId, + MessageId, + type ProjectId, + WorktreeMcpFailure, + type WorktreeMcpContinuationStatus, + type WorktreeMcpHandoffInput, + type WorktreeMcpHandoffResult, + type WorktreeMcpSetupScriptStatus, + type WorktreeMcpStatusResult, +} from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; +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 Path from "effect/Path"; + +import * as GitWorkflowService from "../git/GitWorkflowService.ts"; +import { ThreadManagementService } from "../orchestration-v2/ThreadManagementService.ts"; +import * as ProjectService from "../project/ProjectService.ts"; +import * as ProjectSetupScriptRunner from "../project/ProjectSetupScriptRunner.ts"; +import * as ServerSettings from "../serverSettings.ts"; +import * as VcsStatusBroadcaster from "../vcs/VcsStatusBroadcaster.ts"; +import type { McpInvocationScope } from "./McpInvocationContext.ts"; + +export class WorktreeMcpService extends Context.Service< + WorktreeMcpService, + { + readonly handoff: ( + scope: McpInvocationScope, + input: WorktreeMcpHandoffInput, + ) => Effect.Effect; + readonly status: ( + scope: McpInvocationScope, + ) => Effect.Effect; + } +>()("t3/mcp/WorktreeMcpService") {} + +function failure(code: WorktreeMcpFailure["code"], message: string): WorktreeMcpFailure { + return new WorktreeMcpFailure({ code, message }); +} + +function errorMessage(error: unknown): string { + if (error instanceof Error) return error.message; + if (typeof error === "object" && error !== null && "message" in error) { + return String((error as { message: unknown }).message); + } + return String(error); +} + +const asOperationFailed = (prefix: string) => + Effect.mapError((error: unknown) => + failure("operation_failed", `${prefix}: ${errorMessage(error)}`), + ); + +const make = Effect.gen(function* () { + const crypto = yield* Crypto.Crypto; + const path = yield* Path.Path; + const threadManagement = yield* ThreadManagementService; + const projects = yield* ProjectService.ProjectService; + const serverSettings = yield* ServerSettings.ServerSettingsService; + const gitWorkflow = yield* GitWorkflowService.GitWorkflowService; + const setupScriptRunner = yield* ProjectSetupScriptRunner.ProjectSetupScriptRunner; + const vcsStatusBroadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + + // Serializes handoffs per thread: two concurrent calls could otherwise both + // pass the worktreePath === null check and each create a worktree, leaving + // one untracked on disk. + const handoffThreadsInFlight = new Set(); + + const requireCapability = (scope: McpInvocationScope) => + scope.capabilities.has("worktree") + ? Effect.void + : Effect.fail( + failure("capability_denied", "This MCP credential does not grant worktree capabilities."), + ); + + const loadThread = (scope: McpInvocationScope) => + threadManagement.getThreadProjection(scope.threadId).pipe( + Effect.mapError((error) => + error._tag === "OrchestratorProjectionError" + ? failure("thread_not_found", `Thread '${scope.threadId}' was not found.`) + : failure( + "operation_failed", + `Unable to read thread ${scope.threadId}: ${errorMessage(error)}`, + ), + ), + Effect.filterOrFail( + (projection) => projection.thread.deletedAt === null, + () => failure("thread_not_found", `Thread '${scope.threadId}' was not found.`), + ), + ); + + const loadProject = (scope: McpInvocationScope, projectId: ProjectId) => + projects.getById(projectId).pipe( + asOperationFailed(`Unable to read project ${projectId}`), + Effect.flatMap( + Option.match({ + onNone: () => + Effect.fail( + failure( + "project_not_found", + `Project '${projectId}' was not found for thread '${scope.threadId}'.`, + ), + ), + onSome: Effect.succeed, + }), + ), + ); + + const readDefaultStartFromOrigin = serverSettings.getSettings.pipe( + Effect.map((settings) => settings.newWorktreesStartFromOrigin), + asOperationFailed("Unable to read server settings"), + ); + + const handoffIds = (scope: McpInvocationScope) => + crypto.randomUUIDv4.pipe( + Effect.map((uuid) => { + const part = (kind: string, operation: string) => + [kind, "mcp", encodeURIComponent(scope.providerSessionId), operation, uuid].join(":"); + return { + commandId: CommandId.make(part("command", "worktree-handoff")), + continuationCommandId: CommandId.make(part("command", "worktree-continuation")), + continuationMessageId: MessageId.make(part("message", "worktree-continuation")), + }; + }), + Effect.orDie, + ); + + const performHandoff = Effect.fn("WorktreeMcpService.performHandoff")(function* ( + scope: McpInvocationScope, + input: WorktreeMcpHandoffInput, + ) { + const alreadyInWorktree = (worktreePath: string) => + failure( + "already_in_worktree", + `Thread '${scope.threadId}' is already attached to worktree '${worktreePath}'.`, + ); + + const projection = yield* loadThread(scope); + if (projection.thread.worktreePath !== null) { + return yield* alreadyInWorktree(projection.thread.worktreePath); + } + // An archived thread would accept the binding but refuse the continuation + // message (and any other follow-up), so reject the handoff outright. + if (projection.thread.archivedAt !== null) { + return yield* failure( + "invalid_request", + `Thread '${scope.threadId}' is archived and cannot be handed off to a worktree.`, + ); + } + + const project = yield* loadProject(scope, projection.thread.projectId); + const projectCwd = project.workspaceRoot; + + if (input.path !== undefined && !path.isAbsolute(input.path)) { + return yield* failure( + "invalid_request", + `path must be an absolute filesystem path, got '${input.path}'. A relative path would be created relative to the project workspace but stored verbatim as the thread's worktree binding.`, + ); + } + + // The repo check runs regardless of whether baseRef was supplied, so a + // non-repository workspace fails with an actionable error instead of an + // opaque git failure further down. + const localStatus = yield* gitWorkflow + .localStatus({ cwd: projectCwd }) + .pipe(asOperationFailed("Unable to read git status")); + if (!localStatus.isRepo) { + return yield* failure( + "invalid_request", + `Project workspace '${projectCwd}' is not a git repository.`, + ); + } + + // Fail fast with an actionable message when the branch already exists: + // the git driver deliberately keeps stderr out of its errors, so letting + // `git worktree add` fail would surface only an opaque failure. The + // existence check uses the complete local branch list (exact match); the + // paginated substring search only enriches the message with the checkout + // location when available. + const localBranchNames = yield* gitWorkflow + .listLocalBranchNames(projectCwd) + .pipe(asOperationFailed("Unable to list branches")); + if (localBranchNames.includes(input.branch)) { + const existingRef = yield* gitWorkflow + .listRefs({ cwd: projectCwd, query: input.branch, refKind: "local" }) + .pipe( + Effect.map((result) => + result.refs.find((ref) => ref.name === input.branch && ref.isRemote !== true), + ), + Effect.orElseSucceed(() => undefined), + ); + const checkoutPath = existingRef?.worktreePath ?? null; + return yield* failure( + "invalid_request", + `Branch '${input.branch}' already exists${ + checkoutPath === null ? "" : ` and is checked out at '${checkoutPath}'` + }. Choose a different branch name, or delete the existing branch${ + checkoutPath === null ? "" : " and its worktree" + } first.`, + ); + } + + let baseRef = input.baseRef; + if (baseRef === undefined) { + if (localStatus.refName === null) { + return yield* failure( + "invalid_request", + "Could not determine the current branch of the project workspace (detached HEAD?). Pass baseRef explicitly.", + ); + } + baseRef = localStatus.refName; + } + + const startFromOrigin = input.startFromOrigin ?? (yield* readDefaultStartFromOrigin); + + let worktreeBaseRef = baseRef; + if (startFromOrigin) { + yield* gitWorkflow + .fetchRemote({ cwd: projectCwd, remoteName: "origin" }) + .pipe(asOperationFailed("Unable to fetch origin")); + const resolvedRemoteBase = yield* gitWorkflow + .resolveRemoteTrackingCommit({ + cwd: projectCwd, + refName: baseRef, + fallbackRemoteName: "origin", + }) + .pipe(asOperationFailed(`Unable to resolve the remote-tracking commit of '${baseRef}'`)); + worktreeBaseRef = resolvedRemoteBase.commitSha; + } + + const ids = yield* handoffIds(scope); + + // uninterruptibleMask: only the potentially slow worktree creation itself + // stays interruptible (restore). From the moment it succeeds, through the + // binding, continuation queue, setup script launch, and result + // construction, there is no interruptible gap: a client cancel can + // therefore neither orphan the created worktree before the rollback is + // armed, nor skip setting up a worktree the thread was just bound to + // (once the binding commits, the scheduled session detach can sever this + // request's connection and interrupt the fiber). + return yield* Effect.uninterruptibleMask((restore) => + Effect.gen(function* () { + const worktree = yield* restore( + gitWorkflow + .createWorktree({ + cwd: projectCwd, + refName: worktreeBaseRef, + newRefName: input.branch, + baseRefName: baseRef, + path: input.path ?? null, + }) + .pipe(asOperationFailed("Unable to create the worktree")), + ); + const worktreePath = worktree.worktree.path; + + // Shared shape for "the handoff already succeeded, so report the failure + // in the result instead of failing the call" (continuation, setup script). + const reportFailed = (logMessage: string) => + Effect.catchCause((cause: Cause.Cause) => { + const detail = errorMessage(Cause.squash(cause)); + return Effect.logWarning(logMessage, { + threadId: scope.threadId, + worktreePath, + detail, + }).pipe(Effect.as({ status: "failed", detail } as const)); + }); + + // suspend: build the removal call only if cleanup actually runs. + const removeCreatedWorktree = Effect.suspend(() => + gitWorkflow.removeWorktree({ cwd: projectCwd, path: worktreePath, force: true }), + ).pipe(Effect.ignoreCause({ log: true })); + + const recheckAndBind = Effect.gen(function* () { + // The projection was read before the potentially slow git work + // above; a concurrent binding (for example from the UI) could have + // attached the thread in the meantime. Re-check before committing so + // the race cannot leave a second, untracked worktree. + const recheck = yield* loadThread(scope); + if (recheck.thread.worktreePath !== null) { + return yield* alreadyInWorktree(recheck.thread.worktreePath); + } + // Mirror the up-front archived check: the thread may have been + // archived during the slow git work, and an archived thread must + // not be bound to a fresh worktree it can never use. + if (recheck.thread.archivedAt !== null) { + return yield* failure( + "invalid_request", + `Thread '${scope.threadId}' was archived while the worktree was being created; the handoff was rolled back.`, + ); + } + yield* threadManagement + .dispatch({ + type: "thread.metadata.update", + commandId: ids.commandId, + threadId: scope.threadId, + branch: worktree.worktree.refName, + worktreePath, + }) + .pipe( + Effect.catchCause((cause) => + // Interrupt-only causes propagate unchanged: whether the + // dispatch committed is unknown, so neither a typed failure + // nor a rollback would be correct. Failures and defects + // (including mixed causes) map to a typed operation_failed. + Cause.hasInterruptsOnly(cause) + ? Effect.failCause(cause as Cause.Cause) + : Effect.fail( + failure( + "operation_failed", + `Unable to re-point the thread at the worktree: ${errorMessage(Cause.squash(cause))}`, + ), + ), + ), + ); + }).pipe( + // onError: the worktree was already created, so any failure between + // here and the committed binding (recheck read, recheck race, + // dispatch typed failure or defect) must remove it again so a failed + // handoff leaves nothing behind on disk. Interrupt-only causes skip + // the removal: the binding may have committed, and force-deleting a + // worktree the thread now points at would be worse than leaking one. + Effect.onError((cause) => + Cause.hasInterruptsOnly(cause) ? Effect.void : removeCreatedWorktree, + ), + ); + + // Queue the continuation right after the binding commits: the detach + // that the metadata update schedules will terminate the calling + // session, and a durably queued message is what guarantees the thread + // re-launches inside the worktree. When the dying run reaches a + // terminal state the orchestrator promotes the queued run, which + // derives its cwd from the updated projection. + // suspend: build the send effect only when the binding has succeeded, + // so a failed dispatch never even constructs the continuation call. + const queueContinuation: Effect.Effect = + Effect.suspend(() => + input.continuationPrompt === undefined + ? Effect.succeed({ status: "skipped" }) + : threadManagement + .sendToThread({ + projectId: projection.thread.projectId, + commandId: ids.continuationCommandId, + threadId: scope.threadId, + messageId: ids.continuationMessageId, + text: input.continuationPrompt, + attachments: [], + mode: "queue", + createdBy: "agent", + creationSource: "mcp", + }) + .pipe( + Effect.map( + (sendResult): WorktreeMcpContinuationStatus => ({ + status: "scheduled", + delivery: sendResult.delivery, + }), + ), + // catchCause via reportFailed: the binding is already recorded, + // so a failed continuation must be reported, not fail the handoff. + reportFailed("worktree handoff continuation failed to queue"), + ), + ); + + const continuation = yield* recheckAndBind.pipe(Effect.andThen(queueContinuation)); + + yield* vcsStatusBroadcaster + .refreshStatus(worktreePath) + .pipe(Effect.ignoreCause({ log: true }), Effect.forkDetach); + + let setupScript: WorktreeMcpSetupScriptStatus = { status: "skipped" }; + if (input.runSetupScript ?? true) { + setupScript = yield* setupScriptRunner + .runForThread({ + threadId: scope.threadId, + projectId: projection.thread.projectId, + projectCwd, + worktreePath, + project: { + workspaceRoot: project.workspaceRoot, + scripts: project.scripts, + }, + }) + .pipe( + Effect.map( + (result): WorktreeMcpSetupScriptStatus => + result.status === "started" + ? { + status: "started", + scriptName: result.scriptName, + terminalId: result.terminalId, + } + : { status: "no-script" }, + ), + // catchCause via reportFailed: the thread is already re-pointed at the + // worktree, so even a defect in the setup runner must not fail the handoff. + reportFailed("worktree handoff setup script failed"), + ); + } + + const result: WorktreeMcpHandoffResult = { + worktreePath, + branch: worktree.worktree.refName, + baseRef, + startedFromOrigin: startFromOrigin, + setupScript, + continuation, + note: + continuation.status === "scheduled" + ? "Handoff recorded. Changing the workspace detaches this provider session, so the current turn ends shortly after this call; the queued continuation prompt then starts the next turn inside the worktree with the conversation preserved. The worktree is not removed automatically when the thread is deleted." + : "Handoff recorded. Changing the workspace detaches this provider session, so the current turn ends shortly after this call; the conversation continues inside the worktree when the thread receives its next message. Pass continuationPrompt to resume automatically. The worktree is not removed automatically when the thread is deleted.", + }; + return result; + }), + ); + }); + + const handoff: WorktreeMcpService["Service"]["handoff"] = Effect.fn("WorktreeMcpService.handoff")( + function* (scope, input) { + yield* requireCapability(scope); + // uninterruptibleMask: the guard acquisition and the registration of the + // releasing finalizer happen with no interruptible gap in between. An + // interrupt landing between a bare add() and the start of an ensured + // effect would otherwise leak the guard entry and block every future + // handoff for this thread until restart. + return yield* Effect.uninterruptibleMask((restore) => + Effect.suspend(() => { + if (handoffThreadsInFlight.has(scope.threadId)) { + return Effect.fail( + failure( + "handoff_in_progress", + `A worktree handoff is already in progress for thread '${scope.threadId}'.`, + ), + ); + } + handoffThreadsInFlight.add(scope.threadId); + return restore(performHandoff(scope, input)).pipe( + Effect.ensuring(Effect.sync(() => handoffThreadsInFlight.delete(scope.threadId))), + ); + }), + ); + }, + ); + + const status: WorktreeMcpService["Service"]["status"] = Effect.fn("WorktreeMcpService.status")( + function* (scope) { + yield* requireCapability(scope); + const projection = yield* loadThread(scope); + const project = yield* loadProject(scope, projection.thread.projectId); + + const defaultStartFromOrigin = yield* readDefaultStartFromOrigin; + + const result: WorktreeMcpStatusResult = { + attached: projection.thread.worktreePath !== null, + worktreePath: projection.thread.worktreePath, + branch: projection.thread.branch, + projectWorkspaceRoot: project.workspaceRoot, + defaultStartFromOrigin, + }; + return result; + }, + ); + + return WorktreeMcpService.of({ handoff, status }); +}); + +export const layer: Layer.Layer< + WorktreeMcpService, + never, + | Crypto.Crypto + | Path.Path + | ThreadManagementService + | ProjectService.ProjectService + | ServerSettings.ServerSettingsService + | GitWorkflowService.GitWorkflowService + | ProjectSetupScriptRunner.ProjectSetupScriptRunner + | VcsStatusBroadcaster.VcsStatusBroadcaster +> = Layer.effect(WorktreeMcpService, make); diff --git a/apps/server/src/mcp/toolkits/worktree/handlers.ts b/apps/server/src/mcp/toolkits/worktree/handlers.ts new file mode 100644 index 00000000000..b75e0c5dfbe --- /dev/null +++ b/apps/server/src/mcp/toolkits/worktree/handlers.ts @@ -0,0 +1,22 @@ +import * as Effect from "effect/Effect"; + +import { McpInvocationContext } from "../../McpInvocationContext.ts"; +import { WorktreeMcpService } from "../../WorktreeMcpService.ts"; +import { WorktreeToolkit } from "./tools.ts"; + +const handlers = { + t3_worktree_handoff: (input) => + Effect.gen(function* () { + const scope = yield* McpInvocationContext; + const service = yield* WorktreeMcpService; + return yield* service.handoff(scope, input); + }), + t3_worktree_status: () => + Effect.gen(function* () { + const scope = yield* McpInvocationContext; + const service = yield* WorktreeMcpService; + return yield* service.status(scope); + }), +} satisfies Parameters[0]; + +export const WorktreeToolkitHandlersLive = WorktreeToolkit.toLayer(handlers); diff --git a/apps/server/src/mcp/toolkits/worktree/registration.test.ts b/apps/server/src/mcp/toolkits/worktree/registration.test.ts new file mode 100644 index 00000000000..773f4edc323 --- /dev/null +++ b/apps/server/src/mcp/toolkits/worktree/registration.test.ts @@ -0,0 +1,136 @@ +import { expect, it } from "@effect/vitest"; +import { NodeHttpServer } from "@effect/platform-node"; +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 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 * as ProjectService from "../../../project/ProjectService.ts"; +import * as ProjectSetupScriptRunner from "../../../project/ProjectSetupScriptRunner.ts"; +import { ProviderRegistry } from "../../../provider/Services/ProviderRegistry.ts"; +import { ScheduledTaskService } from "../../../scheduledTasks/ScheduledTaskService.ts"; +import * as ServerSettings from "../../../serverSettings.ts"; +import { VcsStatusBroadcaster } from "../../../vcs/VcsStatusBroadcaster.ts"; +import * as McpHttpServer from "../../McpHttpServer.ts"; +import * as McpSessionRegistry from "../../McpSessionRegistry.ts"; +import * as PreviewAutomationBroker from "../../PreviewAutomationBroker.ts"; + +const StubServicesLive = Layer.mergeAll( + Layer.mock(ThreadManagementService)({}), + Layer.mock(ProviderRegistry)({}), + Layer.mock(ScheduledTaskService)({}), + Layer.mock(ProjectService.ProjectService)({}), + ServerSettings.layerTest({}), + Layer.mock(GitWorkflowService.GitWorkflowService)({}), + Layer.mock(ProjectSetupScriptRunner.ProjectSetupScriptRunner)({}), + Layer.mock(VcsStatusBroadcaster)({}), +); + +it.effect("production mcp layer lists worktree tools over http", () => + Effect.scoped( + Effect.gen(function* () { + const routes = McpHttpServer.layer.pipe(Layer.provide(McpSessionRegistry.layer)); + yield* HttpRouter.serve(routes, { + disableListenLog: true, + disableLogger: true, + }).pipe( + Layer.provide( + Layer.mock(ServerEnvironment.ServerEnvironment)({ + getEnvironmentId: Effect.succeed("environment-scratch" as never), + }), + ), + Layer.provide(PreviewAutomationBroker.layer), + Layer.provide(StubServicesLive), + Layer.build, + ); + + const registry = McpSessionRegistry.issueActiveMcpCredential({ + threadId: ThreadId.make("thread-scratch"), + providerInstanceId: ProviderInstanceId.make("claudeAgent"), + }); + const credential = yield* registry; + expect(credential).toBeDefined(); + + const httpClient = yield* HttpClient.HttpClient; + const auth = credential!.config.authorizationHeader; + const initResponse = yield* httpClient.post("/mcp", { + headers: { + accept: "application/json, text/event-stream", + authorization: auth, + }, + body: HttpBody.text( + `{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"scratch","version":"1.0.0"}}}`, + "application/json", + ), + }); + expect(initResponse.status).toBe(200); + const sessionId = initResponse.headers["mcp-session-id"]; + + const listResponse = yield* httpClient.post("/mcp", { + headers: { + accept: "application/json, text/event-stream", + authorization: auth, + ...(sessionId ? { "mcp-session-id": sessionId } : {}), + }, + body: HttpBody.text( + `{"jsonrpc":"2.0","id":2,"method":"tools/list","params":{}}`, + "application/json", + ), + }); + const bodyText = yield* listResponse.text; + const ToolsListPayload = Schema.fromJsonString( + Schema.Struct({ + result: Schema.Struct({ + tools: Schema.Array( + Schema.Struct({ + name: Schema.String, + inputSchema: Schema.Struct({ type: Schema.optional(Schema.String) }), + annotations: Schema.optional( + Schema.Struct({ + readOnlyHint: Schema.optional(Schema.Boolean), + destructiveHint: Schema.optional(Schema.Boolean), + openWorldHint: Schema.optional(Schema.Boolean), + }), + ), + }), + ), + }), + }), + ); + const payload = yield* Schema.decodeUnknownEffect(ToolsListPayload)( + bodyText.match(/\{.*\}/s)![0], + ); + const tools = payload.result.tools; + const toolNames = tools.map((tool) => tool.name); + expect(toolNames).toContain("t3_worktree_handoff"); + expect(toolNames).toContain("t3_worktree_status"); + // The worktree registration merges alongside the other toolkits rather + // than replacing them. + expect(toolNames).toContain("preview_status"); + expect(toolNames).toContain("delegate_task"); + + // The handoff tool mutates thread state, reaches the network (origin + // fetch), and runs project setup scripts, so its MCP hints must not + // promise a read-only, closed-world, non-destructive tool. + const handoff = tools.find((tool) => tool.name === "t3_worktree_handoff"); + expect(handoff?.annotations?.readOnlyHint).toBe(false); + expect(handoff?.annotations?.destructiveHint).toBe(true); + expect(handoff?.annotations?.openWorldHint).toBe(true); + const status = tools.find((tool) => tool.name === "t3_worktree_status"); + expect(status?.annotations?.readOnlyHint).toBe(true); + expect(status?.annotations?.destructiveHint).toBe(false); + + // MCP requires every tool input schema to be a top-level object schema. + // A non-object schema (e.g. the anyOf produced by an empty + // Schema.Struct({})) makes clients reject the entire server. + for (const tool of tools) { + expect(tool.inputSchema.type, `inputSchema.type of ${tool.name}`).toBe("object"); + } + }), + ).pipe(Effect.provide(Layer.mergeAll(NodeHttpServer.layerTest, NodeServices.layer))), +); diff --git a/apps/server/src/mcp/toolkits/worktree/tools.ts b/apps/server/src/mcp/toolkits/worktree/tools.ts new file mode 100644 index 00000000000..69a33cb79d0 --- /dev/null +++ b/apps/server/src/mcp/toolkits/worktree/tools.ts @@ -0,0 +1,47 @@ +import { + WorktreeMcpFailure, + WorktreeMcpHandoffInput, + WorktreeMcpHandoffResult, + WorktreeMcpStatusResult, +} from "@t3tools/contracts"; +import { Tool, Toolkit } from "effect/unstable/ai"; + +import * as McpInvocationContext from "../../McpInvocationContext.ts"; +import { WorktreeMcpService } from "../../WorktreeMcpService.ts"; + +const dependencies = [McpInvocationContext.McpInvocationContext, WorktreeMcpService]; + +export const WorktreeHandoffTool = Tool.make("t3_worktree_handoff", { + description: + "Move this agent thread into a new git worktree. Creates the worktree branch (optionally from origin), re-points the thread at the worktree, and by default runs the project's setup script there. Changing the workspace detaches the live provider session, so the current turn ends shortly after the handoff is recorded; call this as the last action of the turn. To keep working after the handoff, pass continuationPrompt with the remaining work: it is queued as the thread's next message and starts a new turn inside the worktree with the conversation preserved. Without it the thread stays idle until the next message. The worktree is not removed automatically when the thread is deleted. Fails if the thread is already attached to a worktree.", + parameters: WorktreeMcpHandoffInput, + success: WorktreeMcpHandoffResult, + failure: WorktreeMcpFailure, + failureMode: "return", + dependencies, +}) + .annotate(Tool.Title, "Hand off thread to a git worktree") + .annotate(Tool.Readonly, false) + .annotate(Tool.Destructive, true) + .annotate(Tool.Idempotent, false) + .annotate(Tool.OpenWorld, true); + +export const WorktreeStatusTool = Tool.make("t3_worktree_status", { + description: + "Report this agent thread's worktree binding: whether it is attached to a git worktree, the worktree path and branch, the project's main workspace root, and the server default for t3_worktree_handoff's startFromOrigin. Call this before t3_worktree_handoff to check whether a handoff is possible or has already happened.", + // No `parameters`: Tool.make defaults to Tool.EmptyParams, which serializes + // to a top-level `type: "object"` JSON Schema. An explicit empty + // Schema.Struct({}) serializes to `anyOf: [object, array]`, which is not a + // valid MCP tool input schema and makes clients reject the whole server. + success: WorktreeMcpStatusResult, + failure: WorktreeMcpFailure, + failureMode: "return", + dependencies, +}) + .annotate(Tool.Title, "Get thread worktree status") + .annotate(Tool.Readonly, true) + .annotate(Tool.Destructive, false) + .annotate(Tool.Idempotent, true) + .annotate(Tool.OpenWorld, false); + +export const WorktreeToolkit = Toolkit.make(WorktreeHandoffTool, WorktreeStatusTool); diff --git a/apps/server/src/orchestration-v2/EffectOutbox.ts b/apps/server/src/orchestration-v2/EffectOutbox.ts index 75ead585167..f7f49826fd8 100644 --- a/apps/server/src/orchestration-v2/EffectOutbox.ts +++ b/apps/server/src/orchestration-v2/EffectOutbox.ts @@ -28,6 +28,8 @@ export const OrchestrationEffectRequestV2 = Schema.Union([ type: Schema.Literal("provider-session.detach"), providerSessionId: ProviderSessionId, detail: Schema.optional(Schema.String), + /** Set on terminal detaches (thread archive/delete): revoke the thread's MCP credentials. */ + revokeMcpCredential: Schema.optional(Schema.Boolean), }), Schema.Struct({ type: Schema.Literal("provider-turn.start"), diff --git a/apps/server/src/orchestration-v2/EffectWorker.ts b/apps/server/src/orchestration-v2/EffectWorker.ts index b543edeff24..85c915af3f1 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.ts @@ -88,6 +88,9 @@ export const executorLayer: Layer.Layer< providerSessionId: effect.request.providerSessionId, threadId: effect.threadId, ...(effect.request.detail === undefined ? {} : { detail: effect.request.detail }), + ...(effect.request.revokeMcpCredential === undefined + ? {} + : { revokeMcpCredential: effect.request.revokeMcpCredential }), }) .pipe( Effect.mapError( diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index e570c06439e..ae012dba5a5 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -1104,6 +1104,12 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio : command.type === "thread.runtime-mode.set" ? "Runtime mode changed." : "Provider or model selection changed.", + // Terminal detaches revoke the thread's MCP credentials; other + // detach reasons keep them so a re-attaching provider process + // stays authorized. + ...(command.type === "thread.archive" || command.type === "thread.delete" + ? { revokeMcpCredential: true } + : {}), }, } satisfies PendingOrchestrationEffectV2; yield* Ref.update(effects, (existing) => [...existing, pendingEffect]); diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index 41622130ead..b2fec0e3db2 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -764,7 +764,7 @@ it.effect( assert.isDefined(token); const resolved = yield* registry.resolve(token!); assert.equal(resolved?.threadId, threadId); - assert.deepEqual(resolved?.capabilities, new Set(["preview", "orchestration"])); + assert.deepEqual(resolved?.capabilities, new Set(["preview", "orchestration", "worktree"])); yield* manager.close(providerSessionId); assert.isUndefined(McpProviderSession.readMcpProviderSession(threadId)); @@ -905,6 +905,333 @@ it.effect("ProviderSessionManagerV2 duplicate detach preserves replacement MCP c }), ); +it.effect( + "ProviderSessionManagerV2 detach of a superseded live session preserves replacement MCP credentials", + () => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + const mcpConfigs = yield* Ref.make< + ReadonlyArray + >([]); + const effect = Effect.gen(function* () { + const eventSink = yield* EventSinkV2; + const idAllocator = yield* IdAllocatorV2; + const manager = yield* ProviderSessionManagerV2; + const registry = yield* McpSessionRegistry.McpSessionRegistry; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread-provider-session-manager-superseded-mcp"); + const oldSessionId = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId, + }); + const replacementSessionId = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId, + }); + + yield* eventSink.write({ + events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })], + }); + yield* manager.open({ + threadId, + providerSessionId: oldSessionId, + modelSelection, + runtimePolicy, + }); + // The replacement opens while the old session is still attached: this is + // the workspace-handoff sequence, where the queued continuation run can + // start its session before the outbox executes the old session's detach. + yield* manager.open({ + threadId, + providerSessionId: replacementSessionId, + modelSelection, + runtimePolicy, + }); + + const replacement = (yield* Ref.get(mcpConfigs)).at(-1); + assert.isDefined(replacement); + const replacementToken = replacement?.authorizationHeader.replace(/^Bearer\s+/, ""); + assert.isDefined(replacementToken); + assert.equal( + McpProviderSession.readMcpProviderSession(threadId)?.providerSessionId, + replacement?.providerSessionId, + ); + + // First (non-duplicate) detach of the superseded session must not revoke + // the replacement's credential or clear its config slot. + yield* manager.detach({ providerSessionId: oldSessionId, threadId }); + + assert.equal( + McpProviderSession.readMcpProviderSession(threadId)?.providerSessionId, + replacement?.providerSessionId, + ); + assert.equal((yield* registry.resolve(replacementToken!))?.threadId, threadId); + }); + + yield* effect.pipe( + Effect.provide( + makeTestLayer({ + state, + idleTimeoutMs: 1_000, + capabilities: ExclusiveCapabilities, + mcpConfigs, + }), + ), + ); + }), +); + +it.effect( + "ProviderSessionManagerV2 keeps a thread's MCP credential stable across detach and re-attach on a shared session", + () => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + const mcpConfigs = yield* Ref.make< + ReadonlyArray + >([]); + const effect = Effect.gen(function* () { + const eventSink = yield* EventSinkV2; + const idAllocator = yield* IdAllocatorV2; + const manager = yield* ProviderSessionManagerV2; + const registry = yield* McpSessionRegistry.McpSessionRegistry; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread-provider-session-manager-stable-mcp"); + const providerSessionId = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId, + }); + + yield* eventSink.write({ + events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })], + }); + yield* manager.open({ + threadId, + providerSessionId, + modelSelection, + runtimePolicy, + }); + + const original = (yield* Ref.get(mcpConfigs)).at(-1); + assert.isDefined(original); + const originalToken = original?.authorizationHeader.replace(/^Bearer\s+/, ""); + assert.isDefined(originalToken); + + // Workspace-change handoff on a shared multi-thread session (codex): + // the thread detaches while the provider process keeps running, and the + // process's MCP client keeps using the credential it was started with. + yield* manager.detach({ providerSessionId, threadId, detail: "Workspace changed." }); + assert.equal( + (yield* registry.resolve(originalToken!))?.threadId, + threadId, + "detach must not revoke the credential the live provider process still holds", + ); + + // The continuation run re-attaches the same thread to the same session; + // the credential must be reused, not rotated, so the provider process's + // long-lived MCP client stays authorized. + yield* manager.open({ + threadId, + providerSessionId, + modelSelection, + runtimePolicy, + }); + assert.equal( + McpProviderSession.readMcpProviderSession(threadId)?.providerSessionId, + original?.providerSessionId, + "re-attach must reuse the existing credential, not rotate it", + ); + assert.equal((yield* registry.resolve(originalToken!))?.threadId, threadId); + + // Releasing the session (provider process gone) still revokes. + yield* manager.close(providerSessionId); + assert.isUndefined(yield* registry.resolve(originalToken!)); + }); + + yield* effect.pipe( + Effect.provide( + makeTestLayer({ + state, + idleTimeoutMs: 1_000, + mcpConfigs, + }), + ), + ); + }), +); + +it.effect( + "ProviderSessionManagerV2 revokes a rotated credential despite a stale record on another live session", + () => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + const mcpConfigs = yield* Ref.make< + ReadonlyArray + >([]); + const effect = Effect.gen(function* () { + const eventSink = yield* EventSinkV2; + const idAllocator = yield* IdAllocatorV2; + const manager = yield* ProviderSessionManagerV2; + const registry = yield* McpSessionRegistry.McpSessionRegistry; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread-provider-session-manager-stale-record"); + const s1 = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId, + }); + const s2 = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId, + }); + + yield* eventSink.write({ + events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })], + }); + // S1 (shared session) records credential C1 for the thread, then the + // thread detaches; S1 stays alive with the stale record. + yield* manager.open({ threadId, providerSessionId: s1, modelSelection, runtimePolicy }); + yield* manager.detach({ providerSessionId: s1, threadId }); + + // The credential dies externally, so S2's attach must rotate to C2. + yield* registry.revokeThread(threadId); + yield* manager.open({ threadId, providerSessionId: s2, modelSelection, runtimePolicy }); + const rotated = McpProviderSession.readMcpProviderSession(threadId); + assert.isDefined(rotated); + const rotatedToken = rotated?.authorizationHeader.replace(/^Bearer\s+/, ""); + assert.isDefined(yield* registry.resolve(rotatedToken!)); + + // Releasing S2 must revoke C2 even though S1 still carries a stale + // record (of dead C1) for the same thread. + yield* manager.close(s2); + assert.isUndefined( + yield* registry.resolve(rotatedToken!), + "stale record on S1 must not veto revoking S2's rotated credential", + ); + yield* manager.close(s1); + }); + + yield* effect.pipe( + Effect.provide(makeTestLayer({ state, idleTimeoutMs: 1_000, mcpConfigs })), + ); + }), +); + +it.effect( + "ProviderSessionManagerV2 protects a reused credential from a predecessor release during open", + () => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + const mcpConfigs = yield* Ref.make< + ReadonlyArray + >([]); + const duringOpen = yield* Ref.make>(Effect.void); + const effect = Effect.gen(function* () { + const eventSink = yield* EventSinkV2; + const idAllocator = yield* IdAllocatorV2; + const manager = yield* ProviderSessionManagerV2; + const registry = yield* McpSessionRegistry.McpSessionRegistry; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread-provider-session-manager-open-race"); + const s1 = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId, + }); + const s2 = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId, + }); + + yield* eventSink.write({ + events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })], + }); + yield* manager.open({ threadId, providerSessionId: s1, modelSelection, runtimePolicy }); + const original = (yield* Ref.get(mcpConfigs)).at(-1); + const originalToken = original?.authorizationHeader.replace(/^Bearer\s+/, ""); + assert.isDefined(originalToken); + yield* manager.detach({ providerSessionId: s1, threadId }); + + // While S2's provider process is spawning (after prepare reused the + // credential, before the entry is visible), the predecessor session + // releases. Eager adapters (ACP, OpenCode) bake the credential into + // the process during openSession, so the release must not revoke it; + // rotating afterwards cannot repair those adapters. + yield* Ref.set(duringOpen, manager.close(s1).pipe(Effect.orDie)); + yield* manager.open({ threadId, providerSessionId: s2, modelSelection, runtimePolicy }); + + const slot = McpProviderSession.readMcpProviderSession(threadId); + assert.equal( + slot?.providerSessionId, + original?.providerSessionId, + "the credential the adapter was configured with must remain current", + ); + assert.equal( + (yield* registry.resolve(originalToken!))?.threadId, + threadId, + "the predecessor release must not revoke a credential reserved by an in-flight open", + ); + yield* manager.close(s2); + }); + + yield* effect.pipe( + Effect.provide( + makeTestLayer({ + state, + idleTimeoutMs: 1_000, + mcpConfigs, + beforeOpen: (input) => + input.providerSessionId === undefined + ? Effect.void + : Ref.get(duringOpen).pipe( + Effect.flatten, + Effect.tap(() => Ref.set(duringOpen, Effect.void)), + ), + }), + ), + ); + }), +); + +it.effect("ProviderSessionManagerV2 terminal detach revokes the thread's MCP credential", () => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + const mcpConfigs = yield* Ref.make< + ReadonlyArray + >([]); + const effect = Effect.gen(function* () { + const eventSink = yield* EventSinkV2; + const idAllocator = yield* IdAllocatorV2; + const manager = yield* ProviderSessionManagerV2; + const registry = yield* McpSessionRegistry.McpSessionRegistry; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread-provider-session-manager-terminal-detach"); + const providerSessionId = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId, + }); + + yield* eventSink.write({ + events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })], + }); + yield* manager.open({ threadId, providerSessionId, modelSelection, runtimePolicy }); + const issued = (yield* Ref.get(mcpConfigs)).at(-1); + const token = issued?.authorizationHeader.replace(/^Bearer\s+/, ""); + assert.isDefined(yield* registry.resolve(token!)); + + // Archive/delete detaches carry revokeMcpCredential: the token must die + // with the thread even though the shared provider process lives on. + yield* manager.detach({ + providerSessionId, + threadId, + detail: "Thread deleted.", + revokeMcpCredential: true, + }); + assert.isUndefined(yield* registry.resolve(token!)); + assert.isUndefined(McpProviderSession.readMcpProviderSession(threadId)); + }); + + yield* effect.pipe(Effect.provide(makeTestLayer({ state, idleTimeoutMs: 1_000, mcpConfigs }))); + }), +); + it.effect("ProviderSessionManagerV2 releases idle sessions without sweeping all sessions", () => Effect.gen(function* () { const state = yield* Ref.make(emptyState); diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index dc826ecc5d2..8eeacd7f685 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -154,6 +154,12 @@ export interface ProviderSessionManagerV2Shape { readonly providerSessionId: ProviderSessionId; readonly threadId: ThreadId; readonly detail?: string; + /** + * True for terminal detaches (thread archived or deleted): the thread's + * MCP credentials are revoked immediately instead of surviving for a + * potential re-attach. + */ + readonly revokeMcpCredential?: boolean; }) => Effect.Effect; } @@ -165,6 +171,14 @@ export class ProviderSessionManagerV2 extends Context.Service< interface LiveSessionEntry { readonly attachedThreadIds: ReadonlySet; readonly loadedProviderThreadKeyByThread: ReadonlyMap; + /** + * MCP credential session id issued for each attached thread. Revocation on + * detach/release is scoped to these ids so tearing down a superseded + * session cannot revoke a replacement session's credential for the same + * thread (the workspace-handoff sequence opens the replacement before the + * outbox executes the old session's detach). + */ + readonly mcpCredentialIdByThread: ReadonlyMap; readonly supportsMultipleProviderThreads: boolean; readonly runtime: ProviderAdapterV2SessionRuntime; readonly exposedRuntime: ProviderAdapterV2SessionRuntime; @@ -259,23 +273,116 @@ export const layerWithOptions = ( const sessionOpen = yield* makeKeyedSerialExecutor(); const idleTimeoutMs = Math.max(1, options.idleTimeoutMs ?? DEFAULT_IDLE_TIMEOUT_MS); const maxIdlePinMs = Math.max(0, options.maxIdlePinMs ?? DEFAULT_MAX_IDLE_PIN_MS); - const prepareMcpSession = (threadId: ThreadId, providerInstanceId: ProviderInstanceId) => + interface PreparedMcpCredential { + readonly mcpCredentialId: string | undefined; + /** True when this call minted the credential (vs reusing a live one). */ + readonly issued: boolean; + } + /** + * Reservations protect a credential between prepareMcpSession handing it + * out and the owning session entry becoming visible in `sessions`. + * Adapters like ACP and OpenCode consume the credential eagerly during + * openSession, so a racing release must not revoke it in that window + * (rotating afterwards cannot repair an already-configured process). + * The holder MUST drop the reservation once the entry is recorded or the + * open fails. + */ + const mcpCredentialReservations = new Map(); + const mcpReservationKey = (threadId: ThreadId, mcpCredentialId: string) => + `${threadId}${mcpCredentialId}`; + const reserveMcpCredential = (threadId: ThreadId, mcpCredentialId: string) => { + const key = mcpReservationKey(threadId, mcpCredentialId); + mcpCredentialReservations.set(key, (mcpCredentialReservations.get(key) ?? 0) + 1); + }; + const dropMcpCredentialReservation = (threadId: ThreadId, mcpCredentialId: string) => { + const key = mcpReservationKey(threadId, mcpCredentialId); + const count = mcpCredentialReservations.get(key) ?? 0; + if (count <= 1) { + mcpCredentialReservations.delete(key); + } else { + mcpCredentialReservations.set(key, count - 1); + } + }; + const isMcpCredentialReserved = (threadId: ThreadId, mcpCredentialId: string) => + (mcpCredentialReservations.get(mcpReservationKey(threadId, mcpCredentialId)) ?? 0) > 0; + const mcpPrepareLock = yield* makeKeyedSerialExecutor(); + /** + * Resolves (or mints) the thread's MCP credential and returns it with a + * reservation held; the caller must drop the reservation exactly once. + * Serialized per thread so two concurrent prepares cannot interleave + * their rotate steps and revoke each other's freshly minted credential. + */ + const prepareMcpSession = ( + threadId: ThreadId, + providerInstanceId: ProviderInstanceId, + ): Effect.Effect => options.configureMcp === false - ? Effect.sync(() => McpProviderSession.clearMcpProviderSession(threadId)) - : mcpSessionRegistry.revokeThread(threadId).pipe( - Effect.andThen(mcpSessionRegistry.issue({ threadId, providerInstanceId })), - Effect.tap((credential) => - Effect.sync(() => McpProviderSession.setMcpProviderSession(credential.config)), + ? Effect.sync((): PreparedMcpCredential => { + McpProviderSession.clearMcpProviderSession(threadId); + return { mcpCredentialId: undefined, issued: false }; + }) + : mcpPrepareLock.withLock( + threadId, + Effect.gen(function* () { + // Reuse a still-valid credential for this thread instead of + // rotating: long-lived provider processes (codex app-server) + // build their MCP client once per conversation and keep using + // the credential it started with, so a thread that detaches and + // re-attaches across a workspace handoff must come back to the + // same token or the process's tool calls fail auth. + const existing = McpProviderSession.readMcpProviderSession(threadId); + if (existing !== undefined) { + // Reserve before the async resolve so a release cannot + // revoke the credential between validation and reservation. + reserveMcpCredential(threadId, existing.providerSessionId); + const rawToken = existing.authorizationHeader.replace(/^Bearer\s+/, ""); + const resolved = yield* mcpSessionRegistry.resolve(rawToken); + if ( + resolved !== undefined && + resolved.threadId === threadId && + resolved.providerInstanceId === providerInstanceId + ) { + return { mcpCredentialId: existing.providerSessionId, issued: false }; + } + dropMcpCredentialReservation(threadId, existing.providerSessionId); + } + yield* mcpSessionRegistry.revokeThread(threadId); + const credential = yield* mcpSessionRegistry.issue({ + threadId, + providerInstanceId, + }); + McpProviderSession.setMcpProviderSession(credential.config); + reserveMcpCredential(threadId, credential.config.providerSessionId); + return { mcpCredentialId: credential.config.providerSessionId, issued: true }; + }), + ); + /** + * With a credential id, revocation is scoped to that credential and the + * config slot is cleared only while it still holds it; a replacement + * session's newer credential survives. Without one (attach failed before + * a credential was recorded), fall back to thread-wide revocation. + */ + const clearMcpSession = (threadId: ThreadId, mcpCredentialId?: string) => + mcpCredentialId === undefined + ? mcpSessionRegistry + .revokeThread(threadId) + .pipe( + Effect.tap(() => + Effect.sync(() => McpProviderSession.clearMcpProviderSession(threadId)), + ), + ) + : mcpSessionRegistry.revokeProviderSession(mcpCredentialId).pipe( + Effect.tap(() => + Effect.sync(() => { + if ( + McpProviderSession.readMcpProviderSession(threadId)?.providerSessionId === + mcpCredentialId + ) { + McpProviderSession.clearMcpProviderSession(threadId); + } + }), ), ); - const clearMcpSession = (threadId: ThreadId) => - mcpSessionRegistry - .revokeThread(threadId) - .pipe( - Effect.tap(() => - Effect.sync(() => McpProviderSession.clearMcpProviderSession(threadId)), - ), - ); const publishToSubscribers = ( subscribers: Ref.Ref< @@ -574,7 +681,40 @@ export const layerWithOptions = ( Option.match(entry, { onNone: () => Effect.void, onSome: (entry) => - Effect.forEach(entry.attachedThreadIds, clearMcpSession, { discard: true }), + // Revoke every credential this session recorded, including for + // threads that detached without re-attaching: the provider + // process is gone, so nothing holds them anymore. Skip threads + // a live replacement session took over, since credential reuse + // means the replacement may hold this very credential. + Ref.get(sessions).pipe( + Effect.flatMap((current) => + Effect.forEach( + entry.mcpCredentialIdByThread, + ([threadId, mcpCredentialId]) => { + // Id-sensitive: a stale record for the same thread but + // a DIFFERENT credential (left behind by an old session + // the thread rotated away from) must not veto revoking + // this session's own credential, or it leaks forever. + // A reservation means an in-flight open is configuring + // a provider process with this credential right now; + // revoking it here would strand that process (eager + // adapters cannot pick up a rotated token). + const heldElsewhere = + isMcpCredentialReserved(threadId, mcpCredentialId) || + Array.from(current.values()).some( + (other) => + other !== entry && + (other.attachedThreadIds.has(threadId) || + other.mcpCredentialIdByThread.get(threadId) === mcpCredentialId), + ); + return heldElsewhere + ? Effect.void + : clearMcpSession(threadId, mcpCredentialId); + }, + { discard: true }, + ), + ), + ), }), ).pipe( Effect.catchCause((cause) => @@ -822,28 +962,69 @@ export const layerWithOptions = ( readonly threadId: ThreadId; readonly providerInstanceId: ProviderInstanceId; }) => - Effect.gen(function* () { - const attached = yield* attachThread(input); - if (attached) { - yield* prepareMcpSession(input.threadId, input.providerInstanceId); - const entry = (yield* Ref.get(sessions)).get(sessionKey(input.providerSessionId)); - if (entry !== undefined) { - yield* withActivityError( - input.providerSessionId, - writeProviderSessionEvents({ - runtime: entry.runtime, - threadIds: [input.threadId], - type: "provider-session.attached", - payload: entry.runtime.providerSession, - }), - ); + Effect.suspend(() => { + let preparedForCleanup: PreparedMcpCredential | undefined; + let reservationDropped = false; + const dropReservation = () => { + if (!reservationDropped && preparedForCleanup?.mcpCredentialId !== undefined) { + reservationDropped = true; + dropMcpCredentialReservation(input.threadId, preparedForCleanup.mcpCredentialId); } - } - }).pipe( - Effect.tapError(() => - removeThreadAttachment(input).pipe(Effect.andThen(clearMcpSession(input.threadId))), - ), - ); + }; + return Effect.gen(function* () { + const attached = yield* attachThread(input); + if (attached) { + const prepared = yield* prepareMcpSession(input.threadId, input.providerInstanceId); + preparedForCleanup = prepared; + if (prepared.mcpCredentialId !== undefined) { + const mcpCredentialId = prepared.mcpCredentialId; + yield* Ref.update(sessions, (current) => { + const key = sessionKey(input.providerSessionId); + const entry = current.get(key); + if (entry === undefined) return current; + const mcpCredentialIdByThread = new Map(entry.mcpCredentialIdByThread); + mcpCredentialIdByThread.set(input.threadId, mcpCredentialId); + const updated = new Map(current); + updated.set(key, { ...entry, mcpCredentialIdByThread }); + return updated; + }); + } + const entry = (yield* Ref.get(sessions)).get(sessionKey(input.providerSessionId)); + if (entry !== undefined) { + yield* withActivityError( + input.providerSessionId, + writeProviderSessionEvents({ + runtime: entry.runtime, + threadIds: [input.threadId], + type: "provider-session.attached", + payload: entry.runtime.providerSession, + }), + ); + } + } + }).pipe( + Effect.tapError(() => + removeThreadAttachment(input).pipe( + // Revoke only a credential this attach freshly minted: a REUSED + // credential is by definition held by another live provider + // process, and revoking it thread-wide would break that + // process's MCP client mid-conversation. + Effect.andThen( + Effect.suspend(() => { + dropReservation(); + return preparedForCleanup?.issued === true + ? clearMcpSession(input.threadId, preparedForCleanup.mcpCredentialId) + : Effect.void; + }), + ), + ), + ), + // The entry's own record (written above while the thread is + // attached) guards the credential from here on; the reservation + // is only needed until then. Ensuring covers defects/interrupts. + Effect.ensuring(Effect.sync(dropReservation)), + ); + }); const markBusy = (providerSessionId: ProviderSessionId) => withActivityError( @@ -1218,7 +1399,22 @@ export const layerWithOptions = ( }), ), ); - yield* prepareMcpSession(input.threadId, input.modelSelection.instanceId); + const prepared = yield* prepareMcpSession( + input.threadId, + input.modelSelection.instanceId, + ); + const mcpCredentialId = prepared.mcpCredentialId; + // The reservation from prepare protects the credential (which + // eager adapters bake into the provider process during + // openSession) from racing releases until this session's entry + // is recorded below. Dropped exactly once on every path. + let reservationDropped = mcpCredentialId === undefined; + const dropReservation = Effect.sync(() => { + if (!reservationDropped && mcpCredentialId !== undefined) { + reservationDropped = true; + dropMcpCredentialReservation(input.threadId, mcpCredentialId); + } + }); const sessionScope = yield* Scope.make(); const runtime = yield* adapter .openSession({ @@ -1235,9 +1431,18 @@ export const layerWithOptions = ( Effect.tapError(() => Scope.close(sessionScope, Exit.void).pipe( Effect.ignore, - Effect.andThen(clearMcpSession(input.threadId)), + Effect.andThen(dropReservation), + // Revoke only a credential this open freshly minted: a + // reused credential is held by another live provider + // process and must survive this open's failure. + Effect.andThen( + prepared.issued + ? clearMcpSession(input.threadId, mcpCredentialId) + : Effect.void, + ), ), ), + Effect.onInterrupt(() => dropReservation), Effect.mapError( (cause) => new ProviderSessionOpenError({ @@ -1255,6 +1460,10 @@ export const layerWithOptions = ( const entry: LiveSessionEntry = { attachedThreadIds: new Set([input.threadId]), loadedProviderThreadKeyByThread: new Map(), + mcpCredentialIdByThread: + mcpCredentialId === undefined + ? new Map() + : new Map([[input.threadId, mcpCredentialId]]), supportsMultipleProviderThreads: runtime.providerSession.capabilities.sessions .supportsMultipleProviderThreadsPerSession, @@ -1273,6 +1482,9 @@ export const layerWithOptions = ( updated.set(key, entry); return updated; }); + // The entry now guards the credential via its recorded id, so + // the pre-open reservation can be dropped. + yield* dropReservation; yield* withActivityError( input.providerSessionId, writeProviderSessionEvents({ @@ -1376,22 +1588,45 @@ export const layerWithOptions = ( entry.loadedProviderThreadKeyByThread, ); loadedProviderThreadKeyByThread.delete(input.threadId); + // For a plain (workspace-change) detach, the credential id stays + // recorded: the thread may re-attach and reuse it, and + // releaseEntry revokes it when the provider process finally goes + // away. A terminal detach (archive/delete) prunes the record so + // nothing vetoes the revocation below. + const mcpCredentialIdByThread = + input.revokeMcpCredential === true + ? (() => { + const pruned = new Map(entry.mcpCredentialIdByThread); + pruned.delete(input.threadId); + return pruned; + })() + : entry.mcpCredentialIdByThread; const updatedEntry = { ...entry, attachedThreadIds, loadedProviderThreadKeyByThread, + mcpCredentialIdByThread, }; const updated = new Map(current); updated.set(key, updatedEntry); return [Option.some(updatedEntry), updated] as const; }); + // Plain detaches deliberately do not revoke: a detached thread's + // provider process may still be alive (shared multi-thread codex + // session across a workspace handoff) and holds its MCP client's + // credential for the thread it will re-attach with. Credentials + // are revoked when the session entry is released (process gone) + // or rotated on the next attach if they stopped resolving. + // Terminal detaches (thread archived or deleted) revoke the + // thread's credentials immediately, even on a retry where the + // entry is already gone: there is no legitimate future re-attach, + // and the token must not outlive the thread. + if (input.revokeMcpCredential === true) { + yield* clearMcpSession(input.threadId); + } if (Option.isNone(detached)) { return; } - // Detach effects are retried. Once the old entry is gone, its - // retry must not revoke credentials issued by a replacement - // session for the same app thread. - yield* clearMcpSession(input.threadId); if ( detached.value.attachedThreadIds.size === 0 && !detached.value.supportsMultipleProviderThreads diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index 3740a483832..63f2189cecf 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -35,4 +35,5 @@ export * from "./review.ts"; export * from "./preview.ts"; export * from "./previewAutomation.ts"; export * from "./scheduledTask.ts"; +export * from "./worktreeMcp.ts"; export * from "./rpc.ts"; diff --git a/packages/contracts/src/worktreeMcp.ts b/packages/contracts/src/worktreeMcp.ts new file mode 100644 index 00000000000..154839fae6b --- /dev/null +++ b/packages/contracts/src/worktreeMcp.ts @@ -0,0 +1,127 @@ +import * as Schema from "effect/Schema"; + +import { TrimmedNonEmptyString } from "./baseSchemas.ts"; + +/** + * Input for the `t3_worktree_handoff` MCP tool. + * + * Creates a git worktree for the calling agent thread and re-points the + * thread at it. Changing the thread's workspace detaches the live provider + * session, so the current turn ends shortly after the handoff is recorded; + * the conversation continues inside the worktree on the thread's next run. + */ +export const WorktreeMcpHandoffInput = Schema.Struct({ + branch: TrimmedNonEmptyString.annotate({ + description: "Branch name to create for the worktree (e.g. 'feature/my-change').", + }), + baseRef: Schema.optional( + TrimmedNonEmptyString.annotate({ + description: + "Branch or ref the worktree branch starts from. Defaults to the branch currently checked out in the project workspace.", + }), + ), + startFromOrigin: Schema.optional( + Schema.Boolean.annotate({ + description: + "Fetch origin and start the worktree branch from the remote-tracking commit of baseRef instead of the local ref. Defaults to the server's 'new worktrees start from origin' setting.", + }), + ), + path: Schema.optional( + TrimmedNonEmptyString.check( + // Absolute POSIX (/...), Windows drive (C:\ or C:/), or UNC (\\host). + Schema.isPattern(/^(?:[A-Za-z]:[\\/]|[\\/])/), + ).annotate({ + description: + "Absolute filesystem path for the new worktree. Relative paths are rejected. Defaults to the server-managed worktrees directory.", + }), + ), + runSetupScript: Schema.optional( + Schema.Boolean.annotate({ + description: + "Run the project's configured setup script in the new worktree after handoff. Defaults to true.", + }), + ), + continuationPrompt: Schema.optional( + TrimmedNonEmptyString.check(Schema.isMaxLength(120_000)).annotate({ + description: + "Message queued as the thread's next turn after the handoff. The handoff detaches the current provider session, so pass the remaining work here to automatically resume inside the worktree; omit it to stop after the handoff and wait for the next message.", + }), + ), +}); +export type WorktreeMcpHandoffInput = typeof WorktreeMcpHandoffInput.Type; + +export const WorktreeMcpSetupScriptStatus = Schema.Union([ + Schema.Struct({ + status: Schema.Literal("started"), + scriptName: TrimmedNonEmptyString, + terminalId: TrimmedNonEmptyString, + }), + Schema.Struct({ + status: Schema.Literal("no-script"), + }), + Schema.Struct({ + status: Schema.Literal("skipped"), + }), + Schema.Struct({ + status: Schema.Literal("failed"), + detail: Schema.String, + }), +]); +export type WorktreeMcpSetupScriptStatus = typeof WorktreeMcpSetupScriptStatus.Type; + +export const WorktreeMcpContinuationStatus = Schema.Union([ + Schema.Struct({ + status: Schema.Literal("scheduled"), + delivery: Schema.Literals(["started", "queued", "steered", "restarted"]), + }), + Schema.Struct({ + status: Schema.Literal("skipped"), + }), + Schema.Struct({ + status: Schema.Literal("failed"), + detail: Schema.String, + }), +]); +export type WorktreeMcpContinuationStatus = typeof WorktreeMcpContinuationStatus.Type; + +export const WorktreeMcpHandoffResult = Schema.Struct({ + worktreePath: TrimmedNonEmptyString, + branch: TrimmedNonEmptyString, + baseRef: TrimmedNonEmptyString, + startedFromOrigin: Schema.Boolean, + setupScript: WorktreeMcpSetupScriptStatus, + continuation: WorktreeMcpContinuationStatus, + note: Schema.String, +}); +export type WorktreeMcpHandoffResult = typeof WorktreeMcpHandoffResult.Type; + +export const WorktreeMcpStatusResult = Schema.Struct({ + attached: Schema.Boolean.annotate({ + description: "True when this thread is already attached to a git worktree.", + }), + worktreePath: Schema.NullOr(TrimmedNonEmptyString), + branch: Schema.NullOr(TrimmedNonEmptyString), + projectWorkspaceRoot: TrimmedNonEmptyString.annotate({ + description: "Root of the project's main workspace checkout.", + }), + defaultStartFromOrigin: Schema.Boolean.annotate({ + description: "Server default used by t3_worktree_handoff when startFromOrigin is omitted.", + }), +}); +export type WorktreeMcpStatusResult = typeof WorktreeMcpStatusResult.Type; + +export class WorktreeMcpFailure extends Schema.TaggedErrorClass()( + "WorktreeMcpFailure", + { + code: Schema.Literals([ + "capability_denied", + "thread_not_found", + "project_not_found", + "already_in_worktree", + "handoff_in_progress", + "invalid_request", + "operation_failed", + ]), + message: Schema.String, + }, +) {} diff --git a/packages/shared/src/t3McpToolPresentation.test.ts b/packages/shared/src/t3McpToolPresentation.test.ts index cc5a5035293..02a60b4e0fd 100644 --- a/packages/shared/src/t3McpToolPresentation.test.ts +++ b/packages/shared/src/t3McpToolPresentation.test.ts @@ -24,6 +24,17 @@ describe("resolveT3McpToolPresentation", () => { }); }); + it("pretty prints worktree T3 MCP tool names", () => { + expect(resolveT3McpToolPresentation("mcp__t3-code__t3_worktree_handoff")).toEqual({ + displayName: "Hand off thread to a git worktree", + logo: "t3-code", + }); + expect(resolveT3McpToolPresentation("t3-code.t3_worktree_status")).toEqual({ + displayName: "Get thread worktree status", + logo: "t3-code", + }); + }); + it("keeps unknown MCP tools on the generic renderer path", () => { expect(resolveT3McpToolPresentation("mcp__github__search_issues")).toBeNull(); }); diff --git a/packages/shared/src/t3McpToolPresentation.ts b/packages/shared/src/t3McpToolPresentation.ts index 0a92fc990cc..51b199195d3 100644 --- a/packages/shared/src/t3McpToolPresentation.ts +++ b/packages/shared/src/t3McpToolPresentation.ts @@ -23,6 +23,8 @@ const T3_MCP_TOOL_DISPLAY_NAMES: Record = { t3_thread_send: "Send to a T3 thread", t3_thread_wait: "Wait for a T3 thread", t3_thread_interrupt: "Interrupt a T3 thread", + t3_worktree_handoff: "Hand off thread to a git worktree", + t3_worktree_status: "Get thread worktree status", }; function normalizeT3McpToolLabel(value: string): string {