From 13bd326fdcbe6765fbbdd5a8793fdfcd0dd7901b Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 29 Aug 2026 14:46:44 -0700 Subject: [PATCH 1/4] feat(mcp): target thread launches by project --- .../OrchestratorMcpService.activity.test.ts | 13 + .../OrchestratorMcpService.targeting.test.ts | 544 ++++++++++++++++++ .../src/mcp/OrchestratorMcpService.test.ts | 13 + apps/server/src/mcp/OrchestratorMcpService.ts | 364 +++++++++--- ...OrchestratorMcpToolkit.integration.test.ts | 107 ++++ .../src/mcp/toolkits/orchestrator/handlers.ts | 4 + .../src/mcp/toolkits/orchestrator/tools.ts | 16 +- .../toolkits/worktree/registration.test.ts | 4 + .../src/orchestration-v2/Orchestrator.ts | 61 +- .../ThreadManagementService.ts | 12 + .../orchestrator-mcp-server.md | 49 +- docs/user/project-settings.md | 17 + .../contracts/src/orchestrationV2.test.ts | 2 + packages/contracts/src/orchestrationV2.ts | 2 + .../contracts/src/orchestratorMcp.test.ts | 18 + packages/contracts/src/orchestratorMcp.ts | 39 ++ 16 files changed, 1178 insertions(+), 87 deletions(-) create mode 100644 apps/server/src/mcp/OrchestratorMcpService.targeting.test.ts diff --git a/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts b/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts index ccb4393684cf..9373ec4021eb 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts @@ -11,11 +11,15 @@ import { import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; +import * as NodeServices from "@effect/platform-node/NodeServices"; import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; import { expect, it } from "vite-plus/test"; import { ProviderRegistry } from "../provider/Services/ProviderRegistry.ts"; +import * as ProjectService from "../project/ProjectService.ts"; import { ScheduledTaskService } from "../scheduledTasks/ScheduledTaskService.ts"; +import * as VcsDriverRegistry from "../vcs/VcsDriverRegistry.ts"; +import * as ThreadLaunch from "../orchestration-v2/ThreadLaunchService.ts"; import { ThreadManagementService } from "../orchestration-v2/ThreadManagementService.ts"; import type * as McpInvocationContext from "./McpInvocationContext.ts"; import { @@ -36,6 +40,12 @@ const codexDriver = ProviderDriverKind.make("codex"); // Distinct from driver kind so a regression that re-derives from driver fails. const customCodexInstanceId = ProviderInstanceId.make("codex-custom-workspace"); const parentInstanceId = ProviderInstanceId.make("codex"); +const additionalDependencies = Layer.mergeAll( + Layer.mock(ProjectService.ProjectService)({}), + Layer.mock(ThreadLaunch.ThreadLaunchService)({}), + Layer.mock(VcsDriverRegistry.VcsDriverRegistry)({ detect: () => Effect.succeed(null) }), + NodeServices.layer, +); const makeScope = (): McpInvocationContext.McpInvocationScope => ({ environmentId, @@ -135,6 +145,7 @@ it("readThread prefers activity-run status over a newer cancelled queued run", a Layer.mock(ScheduledTaskService)({ list: () => Effect.succeed({ tasks: [] }), } satisfies Partial), + additionalDependencies, NodeCrypto.layer, ), ), @@ -184,6 +195,7 @@ it("readThread prefers waiting activity status over a newer cancelled queued run Layer.mock(ScheduledTaskService)({ list: () => Effect.succeed({ tasks: [] }), } satisfies Partial), + additionalDependencies, NodeCrypto.layer, ), ), @@ -290,6 +302,7 @@ it("taskStatus returns task.providerInstanceId rather than the driver kind", asy Layer.mock(ScheduledTaskService)({ list: () => Effect.succeed({ tasks: [] }), } satisfies Partial), + additionalDependencies, NodeCrypto.layer, ), ), diff --git a/apps/server/src/mcp/OrchestratorMcpService.targeting.test.ts b/apps/server/src/mcp/OrchestratorMcpService.targeting.test.ts new file mode 100644 index 000000000000..acb4f08da2e7 --- /dev/null +++ b/apps/server/src/mcp/OrchestratorMcpService.targeting.test.ts @@ -0,0 +1,544 @@ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, describe, it } from "@effect/vitest"; +import { + EnvironmentId, + NodeId, + ProjectId, + ProviderDriverKind, + ProviderInstanceId, + RunId, + ThreadId, + type OrchestrationV2ThreadProjection, + type Project, + type ServerProvider, +} from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Path from "effect/Path"; +import * as Ref from "effect/Ref"; + +import * as ThreadLaunch from "../orchestration-v2/ThreadLaunchService.ts"; +import { ThreadManagementService } from "../orchestration-v2/ThreadManagementService.ts"; +import * as ProjectService from "../project/ProjectService.ts"; +import { ProviderRegistry } from "../provider/Services/ProviderRegistry.ts"; +import { ScheduledTaskService } from "../scheduledTasks/ScheduledTaskService.ts"; +import * as VcsDriverRegistry from "../vcs/VcsDriverRegistry.ts"; +import * as VcsProcess from "../vcs/VcsProcess.ts"; +import type { McpInvocationScope } from "./McpInvocationContext.ts"; +import * as OrchestratorMcp from "./OrchestratorMcpService.ts"; + +const parentProjectId = ProjectId.make("project:mcp-parent"); +const targetProjectId = ProjectId.make("project:mcp-target"); +const nestedProjectId = ProjectId.make("project:mcp-nested-target"); +const parentThreadId = ThreadId.make("thread:mcp-parent"); +const targetThreadId = ThreadId.make("thread:mcp-target"); +const providerInstanceId = ProviderInstanceId.make("codex"); +const nowIso = "2026-08-29T12:00:00.000Z" as const; +const now = DateTime.makeUnsafe(nowIso); + +const scope: McpInvocationScope = { + environmentId: EnvironmentId.make("environment:mcp-project-targeting"), + threadId: parentThreadId, + providerSessionId: "provider-session:mcp-project-targeting", + providerInstanceId, + capabilities: new Set(["orchestration"]), + issuedAt: 1, +}; + +const targetProject = { + id: targetProjectId, + title: "Target project", + workspaceRoot: "/target/project", + repositoryIdentity: null, + faviconPath: null, + defaultModelSelection: null, + defaultThreadEnvMode: "worktree", + scripts: [], + createdAt: nowIso, + updatedAt: nowIso, + deletedAt: null, +} satisfies Project; + +const parentProject = { + ...targetProject, + id: parentProjectId, + title: "Parent project", + workspaceRoot: "/caller/project", +} satisfies Project; + +const fakeGitHandle = (cwd: string): VcsDriverRegistry.VcsDriverHandle => { + const target = cwd.startsWith("/target/"); + const rootPath = target + ? "/target/project" + : cwd === "/caller/worktree" + ? "/caller/worktree" + : "/caller/project"; + return { + kind: "git", + repository: { + kind: "git", + rootPath, + metadataPath: target ? "/target/project/.git" : "/caller/project/.git", + freshness: { source: "live-local", observedAt: now, expiresAt: Option.none() }, + }, + driver: { + execute: ({ cwd: executionCwd }: { readonly cwd: string }) => + Effect.succeed({ + exitCode: 0, + stdout: executionCwd.startsWith("/target/") ? "target-main\n" : "caller-branch\n", + stderr: "", + stdoutTruncated: false, + stderrTruncated: false, + }), + } as never, + }; +}; + +const provider = { + instanceId: providerInstanceId, + driver: ProviderDriverKind.make("codex"), + enabled: true, + installed: true, + version: "test", + status: "ready", + auth: { status: "authenticated" }, + checkedAt: "2026-08-29T12:00:00.000Z", + models: [{ slug: "gpt-test", name: "GPT Test", isCustom: false, capabilities: null }], + slashCommands: [], + skills: [], +} satisfies ServerProvider; + +function projection(input: { + readonly threadId: ThreadId; + readonly projectId: ProjectId; + readonly title: string; + readonly worktreePath: string | null; + readonly active?: boolean; +}): OrchestrationV2ThreadProjection { + const runId = RunId.make(`run:${input.threadId}`); + return { + thread: { + id: input.threadId, + projectId: input.projectId, + title: input.title, + createdBy: "agent", + creationSource: "mcp", + modelSelection: { instanceId: providerInstanceId, model: "gpt-test" }, + runtimeMode: input.threadId === parentThreadId ? "full-access" : "approval-required", + interactionMode: "default", + branch: "caller-branch", + worktreePath: input.worktreePath, + lineage: { + parentThreadId: null, + relationshipToParent: null, + rootThreadId: input.threadId, + }, + archivedAt: null, + deletedAt: null, + providerInstanceId, + createdAt: now, + updatedAt: now, + }, + runs: input.active + ? [ + { + id: runId, + ordinal: 1, + status: "running", + rootNodeId: NodeId.make(`node:${input.threadId}`), + providerInstanceId, + modelSelection: { instanceId: providerInstanceId, model: "gpt-test" }, + requestedAt: now, + startedAt: now, + completedAt: null, + } as never, + ] + : [], + visibleTurnItems: [], + runtimeRequests: [], + messages: [], + contextTransfers: [], + subagents: [], + updatedAt: now, + } as unknown as OrchestrationV2ThreadProjection; +} + +const parent = projection({ + threadId: parentThreadId, + projectId: parentProjectId, + title: "Parent", + worktreePath: "/caller/worktree", + active: true, +}); + +const target = projection({ + threadId: targetThreadId, + projectId: targetProjectId, + title: "Target", + worktreePath: null, +}); + +const commonDependencies = Layer.mergeAll( + NodeServices.layer, + Layer.succeed( + FileSystem.FileSystem, + FileSystem.makeNoop({ realPath: (path) => Effect.succeed(path) }), + ), + Layer.mock(ProviderRegistry)({ getProviders: Effect.succeed([provider]) }), + Layer.mock(ProjectService.ProjectService)({ + getById: (projectId) => + Effect.succeed( + projectId === targetProjectId + ? Option.some(targetProject) + : projectId === parentProjectId + ? Option.some(parentProject) + : Option.none(), + ), + }), + Layer.mock(ScheduledTaskService)({}), + Layer.mock(VcsDriverRegistry.VcsDriverRegistry)({ + detect: ({ cwd }) => Effect.succeed(fakeGitHandle(cwd)), + }), +); + +describe("OrchestratorMcpService project targeting", () => { + it.effect( + "launches cross-project threads at the target root unless a workspace is explicit", + () => + Effect.gen(function* () { + const launches = yield* Ref.make>([]); + const launchLayer = Layer.mock(ThreadLaunch.ThreadLaunchService)({ + launch: (input) => + Ref.update(launches, (all) => [...all, input]).pipe( + Effect.as({ + threadId: input.threadId!, + projection: { + ...target, + thread: { + ...target.thread, + id: input.threadId!, + title: input.title, + modelSelection: input.modelSelection, + runtimeMode: input.runtimeMode, + interactionMode: input.interactionMode, + }, + }, + resumed: false, + }), + ), + }); + const dependencies = Layer.mergeAll( + commonDependencies, + launchLayer, + Layer.mock(ThreadManagementService)({ + getThreadProjection: () => Effect.succeed(parent), + dispatch: () => Effect.succeed({} as never), + recordServerCreatedThread: () => Effect.succeed({} as never), + }), + ); + + const result = yield* Effect.gen(function* () { + const service = yield* OrchestratorMcp.OrchestratorMcpService; + return yield* service.createThreads(scope, { + clientRequestId: "cross-project-launch", + threads: [ + { prompt: "Reuse the current checkout" }, + { projectId: targetProjectId, prompt: "Use the project root" }, + { + projectId: targetProjectId, + prompt: "Create a worktree", + workspaceStrategy: { + type: "new_worktree", + baseRef: "trunk", + startFromOrigin: false, + }, + }, + ], + }); + }).pipe(Effect.provide(OrchestratorMcp.layer.pipe(Layer.provide(dependencies)))); + + assert.deepEqual( + (yield* Ref.get(launches)).map((launch) => ({ + projectId: launch.projectId, + workspaceStrategy: launch.workspaceStrategy, + modelSelection: launch.modelSelection, + runtimeMode: launch.runtimeMode, + })), + [ + { + projectId: parentProjectId, + workspaceStrategy: { + type: "existing_worktree", + worktreePath: "/caller/worktree", + branch: "caller-branch", + }, + modelSelection: parent.thread.modelSelection, + runtimeMode: parent.thread.runtimeMode, + }, + { + projectId: targetProjectId, + workspaceStrategy: { type: "root", branch: "target-main" }, + modelSelection: parent.thread.modelSelection, + runtimeMode: parent.thread.runtimeMode, + }, + { + projectId: targetProjectId, + workspaceStrategy: { type: "worktree", baseRef: "trunk", startFromOrigin: false }, + modelSelection: parent.thread.modelSelection, + runtimeMode: parent.thread.runtimeMode, + }, + ], + ); + assert.deepEqual( + result.threads.map((thread) => thread.projectId), + [parentProjectId, targetProjectId, targetProjectId], + ); + }), + ); + + it.effect("routes existing thread operations through the explicitly selected project", () => + Effect.gen(function* () { + const routedProjectIds = yield* Ref.make>([]); + const recordProject = (projectId: ProjectId) => + Ref.update(routedProjectIds, (all) => [...all, projectId]); + const dependencies = Layer.mergeAll( + commonDependencies, + Layer.mock(ThreadLaunch.ThreadLaunchService)({}), + Layer.mock(ThreadManagementService)({ + getThreadProjection: () => Effect.succeed(parent), + getProjectThread: ({ projectId }) => recordProject(projectId).pipe(Effect.as(target)), + listProjectThreads: ({ projectId }) => recordProject(projectId).pipe(Effect.as([])), + sendToThread: ({ projectId }) => + recordProject(projectId).pipe( + Effect.as({ + run: { id: RunId.make("run:send"), status: "running" }, + delivery: "started", + } as never), + ), + waitForThread: ({ projectId }) => + recordProject(projectId).pipe( + Effect.as({ threadId: targetThreadId, run: null, timedOut: false }), + ), + interruptThread: ({ projectId }) => + recordProject(projectId).pipe(Effect.as({ type: "no_active_run" })), + }), + ); + + yield* Effect.gen(function* () { + const service = yield* OrchestratorMcp.OrchestratorMcpService; + yield* service.listThreads(scope, { projectId: targetProjectId }); + yield* service.readThread(scope, { projectId: targetProjectId, threadId: targetThreadId }); + yield* service.sendToThread(scope, { + projectId: targetProjectId, + threadId: targetThreadId, + message: "Continue", + }); + yield* service.waitForThread(scope, { + projectId: targetProjectId, + threadId: targetThreadId, + }); + yield* service.interruptThread(scope, { + projectId: targetProjectId, + threadId: targetThreadId, + }); + }).pipe(Effect.provide(OrchestratorMcp.layer.pipe(Layer.provide(dependencies)))); + + assert.deepEqual(yield* Ref.get(routedProjectIds), [ + targetProjectId, + targetProjectId, + targetProjectId, + targetProjectId, + targetProjectId, + targetProjectId, + targetProjectId, + targetProjectId, + ]); + }), + ); + + it.effect("accepts only canonical Git workspaces on their actual branch", () => + Effect.scoped( + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const vcsProcess = yield* VcsProcess.VcsProcess; + const tempRoot = yield* fileSystem.makeTempDirectoryScoped({ + prefix: "orchestrator-mcp-workspaces-", + }); + const repositoryRoot = path.join(tempRoot, "repository"); + const siblingWorktree = path.join(tempRoot, "feature-worktree"); + const linkedWorktree = path.join(tempRoot, "linked-worktree"); + const nestedProjectRoot = path.join(repositoryRoot, "packages", "nested-project"); + const otherRepository = path.join(tempRoot, "other-repository"); + const plainDirectory = path.join(tempRoot, "plain-directory"); + + const git = (cwd: string, args: ReadonlyArray) => + vcsProcess.run({ + operation: "OrchestratorMcpService.targeting.test", + command: "git", + cwd, + args, + timeoutMs: 10_000, + }); + const initializeRepository = (cwd: string) => + Effect.gen(function* () { + yield* fileSystem.makeDirectory(cwd); + yield* git(cwd, ["init", "--initial-branch=main"]); + yield* git(cwd, ["config", "user.email", "mcp-test@example.com"]); + yield* git(cwd, ["config", "user.name", "MCP Test"]); + yield* fileSystem.writeFileString(path.join(cwd, "README.md"), "workspace test\n"); + yield* git(cwd, ["add", "README.md"]); + yield* git(cwd, ["commit", "-m", "initial"]); + }); + + yield* initializeRepository(repositoryRoot); + yield* fileSystem.makeDirectory(nestedProjectRoot, { recursive: true }); + yield* git(repositoryRoot, ["worktree", "add", "-b", "feature", siblingWorktree]); + yield* fileSystem.symlink(siblingWorktree, linkedWorktree); + const canonicalSiblingWorktree = yield* fileSystem.realPath(siblingWorktree); + yield* initializeRepository(otherRepository); + yield* fileSystem.makeDirectory(plainDirectory); + + const project = { + ...targetProject, + workspaceRoot: repositoryRoot, + } satisfies Project; + const nestedProject = { + ...project, + id: nestedProjectId, + title: "Nested project", + workspaceRoot: nestedProjectRoot, + } satisfies Project; + const launches = yield* Ref.make>([]); + const dependencies = Layer.mergeAll( + NodeServices.layer, + Layer.mock(ProviderRegistry)({ getProviders: Effect.succeed([provider]) }), + Layer.mock(ProjectService.ProjectService)({ + getById: (projectId) => + Effect.succeed(Option.some(projectId === nestedProjectId ? nestedProject : project)), + }), + Layer.mock(ScheduledTaskService)({}), + Layer.mock(ThreadLaunch.ThreadLaunchService)({ + launch: (input) => + Ref.update(launches, (all) => [...all, input]).pipe( + Effect.as({ + threadId: input.threadId!, + projection: { + ...target, + thread: { + ...target.thread, + id: input.threadId!, + projectId: input.projectId, + branch: input.workspaceStrategy.branch ?? null, + worktreePath: + input.workspaceStrategy.type === "existing_worktree" + ? input.workspaceStrategy.worktreePath + : null, + }, + }, + resumed: false, + }), + ), + }), + Layer.mock(ThreadManagementService)({ + getThreadProjection: () => Effect.succeed(parent), + dispatch: () => Effect.succeed({} as never), + recordServerCreatedThread: () => Effect.succeed({} as never), + }), + realVcsDriverRegistryLayer, + ); + + const create = ( + clientRequestId: string, + workspaceStrategy: + | { + readonly type: "root"; + readonly branch?: string; + } + | { + readonly type: "existing_worktree"; + readonly worktreePath: string; + readonly branch?: string; + } + | undefined, + projectId: ProjectId = targetProjectId, + ) => + Effect.gen(function* () { + const service = yield* OrchestratorMcp.OrchestratorMcpService; + return yield* service.createThreads(scope, { + clientRequestId, + threads: [ + { + projectId, + ...(workspaceStrategy === undefined ? {} : { workspaceStrategy }), + }, + ], + }); + }).pipe(Effect.provide(OrchestratorMcp.layer.pipe(Layer.provide(dependencies)))); + + yield* create("existing-sibling", { + type: "existing_worktree", + worktreePath: siblingWorktree, + branch: "feature", + }); + yield* create("existing-symlink", { + type: "existing_worktree", + worktreePath: linkedWorktree, + branch: "feature", + }); + yield* create("root-actual-branch", { type: "root", branch: "main" }); + yield* create("nested-project-default-root", undefined, nestedProjectId); + + const accepted = yield* Ref.get(launches); + assert.deepEqual( + accepted.map((launch) => launch.workspaceStrategy), + [ + { + type: "existing_worktree", + worktreePath: canonicalSiblingWorktree, + branch: "feature", + }, + { + type: "existing_worktree", + worktreePath: canonicalSiblingWorktree, + branch: "feature", + }, + { type: "root", branch: "main" }, + { type: "root", branch: "main" }, + ], + ); + + for (const [clientRequestId, workspaceStrategy] of [ + [ + "existing-other-repository", + { type: "existing_worktree", worktreePath: otherRepository }, + ], + ["existing-plain-directory", { type: "existing_worktree", worktreePath: plainDirectory }], + [ + "existing-wrong-branch", + { type: "existing_worktree", worktreePath: siblingWorktree, branch: "main" }, + ], + ["root-wrong-branch", { type: "root", branch: "feature" }], + ] as const) { + const error = yield* create(clientRequestId, workspaceStrategy).pipe(Effect.flip); + assert.equal(error.code, "invalid_request"); + } + assert.equal((yield* Ref.get(launches)).length, 4); + }).pipe(Effect.provide(realVcsInfrastructureLayer)), + ), + ); +}); + +const realVcsProcessLayer = VcsProcess.layer.pipe(Layer.provide(NodeServices.layer)); +const realVcsDriverRegistryLayer = VcsDriverRegistry.layer.pipe( + Layer.provide(realVcsProcessLayer), + Layer.provide(NodeServices.layer), +); +const realVcsInfrastructureLayer = Layer.mergeAll( + NodeServices.layer, + realVcsProcessLayer, + realVcsDriverRegistryLayer, +); diff --git a/apps/server/src/mcp/OrchestratorMcpService.test.ts b/apps/server/src/mcp/OrchestratorMcpService.test.ts index d1fd1975003a..5ea2bd6360b3 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.test.ts @@ -13,11 +13,20 @@ import * as Layer from "effect/Layer"; import * as Ref from "effect/Ref"; import { ThreadManagementService } from "../orchestration-v2/ThreadManagementService.ts"; +import * as ThreadLaunch from "../orchestration-v2/ThreadLaunchService.ts"; +import * as ProjectService from "../project/ProjectService.ts"; import { ProviderRegistry } from "../provider/Services/ProviderRegistry.ts"; import { ScheduledTaskService } from "../scheduledTasks/ScheduledTaskService.ts"; +import * as VcsDriverRegistry from "../vcs/VcsDriverRegistry.ts"; import type { McpInvocationScope } from "./McpInvocationContext.ts"; import * as OrchestratorMcpService from "./OrchestratorMcpService.ts"; +const additionalDependencies = Layer.mergeAll( + Layer.mock(ProjectService.ProjectService)({}), + Layer.mock(ThreadLaunch.ThreadLaunchService)({}), + Layer.mock(VcsDriverRegistry.VcsDriverRegistry)({ detect: () => Effect.succeed(null) }), +); + describe("OrchestratorMcpService", () => { it.effect("retries terminal acknowledgement with a fresh command id", () => Effect.gen(function* () { @@ -69,6 +78,7 @@ describe("OrchestratorMcpService", () => { }), Layer.mock(ProviderRegistry)({ getProviders: Effect.succeed([]) }), Layer.mock(ScheduledTaskService)({}), + additionalDependencies, ); const scope: McpInvocationScope = { environmentId: EnvironmentId.make("environment:mcp-ack"), @@ -134,6 +144,7 @@ describe("OrchestratorMcpService", () => { }), Layer.mock(ProviderRegistry)({ getProviders: Effect.succeed([]) }), Layer.mock(ScheduledTaskService)({}), + additionalDependencies, ); const scope: McpInvocationScope = { environmentId: EnvironmentId.make("environment:mcp-cancel"), @@ -196,6 +207,7 @@ describe("OrchestratorMcpService", () => { }), Layer.mock(ProviderRegistry)({ getProviders: Effect.succeed([]) }), Layer.mock(ScheduledTaskService)({}), + additionalDependencies, ); const scope: McpInvocationScope = { environmentId: EnvironmentId.make("environment:mcp-cancel-failed"), @@ -265,6 +277,7 @@ describe("OrchestratorMcpService", () => { }), Layer.mock(ProviderRegistry)({ getProviders: Effect.succeed([]) }), Layer.mock(ScheduledTaskService)({}), + additionalDependencies, ); const scope: McpInvocationScope = { environmentId: EnvironmentId.make("environment:mcp-cancel-dispose-failed"), diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 7d7bd24f02de..3d77f19111b2 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -45,6 +45,8 @@ import { type ProviderInteractionMode, type ProviderOptionDescriptor, type ProviderOptionSelection, + type Project, + type ProjectId, type RuntimeMode, type ScheduledTask, type ScheduledTaskUpsertInput, @@ -56,12 +58,15 @@ import * as Crypto from "effect/Crypto"; import * as DateTime from "effect/DateTime"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; 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 { isBuiltInProviderAdapterDriverV2 } from "../orchestration-v2/builtInProviderAdapterDrivers.ts"; import { subagentResultForRun } from "../orchestration-v2/SubagentProjection.ts"; +import * as ThreadLaunch from "../orchestration-v2/ThreadLaunchService.ts"; import { isActiveRun, latestActiveRun, @@ -70,7 +75,9 @@ import { ThreadManagementService, } from "../orchestration-v2/ThreadManagementService.ts"; import { ProviderRegistry } from "../provider/Services/ProviderRegistry.ts"; +import * as ProjectService from "../project/ProjectService.ts"; import { ScheduledTaskService } from "../scheduledTasks/ScheduledTaskService.ts"; +import * as VcsDriverRegistry from "../vcs/VcsDriverRegistry.ts"; import type { McpInvocationScope } from "./McpInvocationContext.ts"; const DEFAULT_WAIT_TIMEOUT_MS = 10 * 60 * 1_000; @@ -672,7 +679,12 @@ const make = Effect.gen(function* () { const crypto = yield* Crypto.Crypto; const threadManagement = yield* ThreadManagementService; const providerRegistry = yield* ProviderRegistry; + const projects = yield* ProjectService.ProjectService; const scheduledTasks = yield* ScheduledTaskService; + const threadLaunch = yield* ThreadLaunch.ThreadLaunchService; + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const vcsRegistry = yield* VcsDriverRegistry.VcsDriverRegistry; const requireCapability = (scope: McpInvocationScope) => scope.capabilities.has("orchestration") @@ -704,17 +716,231 @@ const make = Effect.gen(function* () { .getProjectThread({ projectId, threadId }) .pipe(Effect.mapError(threadManagementFailure)); - const loadScopedThread = (scope: McpInvocationScope, threadId: ThreadId) => + const resolveProjectId = ( + parent: OrchestrationV2ThreadProjection, + requestedProjectId: ProjectId | undefined, + ): Effect.Effect => + Effect.gen(function* () { + const projectId = requestedProjectId ?? parent.thread.projectId; + if (projectId === parent.thread.projectId) return projectId; + const project = yield* projects + .getById(projectId) + .pipe( + Effect.mapError((error) => + failure( + "orchestration_error", + `Unable to read project ${projectId}: ${errorMessage(error)}`, + ), + ), + ); + if (Option.isNone(project)) { + return yield* failure("project_not_found", `Project ${projectId} was not found.`); + } + return projectId; + }); + + const loadLaunchProject = (projectId: ProjectId) => + projects.getById(projectId).pipe( + Effect.mapError((error) => + failure( + "orchestration_error", + `Unable to read project ${projectId}: ${errorMessage(error)}`, + ), + ), + Effect.flatMap( + Option.match({ + onNone: () => + Effect.fail(failure("project_not_found", `Project ${projectId} was not found.`)), + onSome: Effect.succeed, + }), + ), + ); + + const loadScopedThread = ( + scope: McpInvocationScope, + threadId: ThreadId, + requestedProjectId?: ProjectId, + ) => Effect.gen(function* () { yield* requireCapability(scope); const parent = yield* loadProjection(scope.threadId); + const projectId = yield* resolveProjectId(parent, requestedProjectId); const target = - threadId === scope.threadId + threadId === scope.threadId && projectId === parent.thread.projectId ? parent - : yield* loadProjectThread(parent.thread.projectId, threadId); - return { parent, target } as const; + : yield* loadProjectThread(projectId, threadId); + return { parent, projectId, target } as const; }); + const canonicalPath = (value: string, label: string) => + fileSystem + .realPath(value) + .pipe( + Effect.mapError((error) => + failure("invalid_request", `Unable to resolve ${label}: ${errorMessage(error)}`), + ), + ); + + const detectRepository = (cwd: string, label: string) => + vcsRegistry + .detect({ cwd }) + .pipe( + Effect.mapError((error) => + failure("invalid_request", `Unable to inspect ${label}: ${errorMessage(error)}`), + ), + ); + + const canonicalRepositoryMetadata = Effect.fn( + "OrchestratorMcpService.canonicalRepositoryMetadata", + )(function* (input: { readonly detectionCwd: string; readonly metadataPath: string | null }) { + if (input.metadataPath === null) return null; + return yield* canonicalPath( + path.isAbsolute(input.metadataPath) + ? input.metadataPath + : path.resolve(input.detectionCwd, input.metadataPath), + "repository metadata", + ); + }); + + const actualGitBranch = Effect.fn("OrchestratorMcpService.actualGitBranch")(function* ( + handle: VcsDriverRegistry.VcsDriverHandle, + cwd: string, + ) { + const result = yield* handle.driver + .execute({ + operation: "OrchestratorMcpService.actualGitBranch", + cwd, + args: ["symbolic-ref", "--quiet", "--short", "HEAD"], + allowNonZeroExit: true, + timeoutMs: 5_000, + maxOutputBytes: 16 * 1024, + }) + .pipe( + Effect.mapError((error) => + failure("invalid_request", `Unable to read the workspace branch: ${errorMessage(error)}`), + ), + ); + const branch = result.exitCode === 0 ? result.stdout.trim() : ""; + return branch.length === 0 ? null : branch; + }); + + const inspectLaunchWorkspace = Effect.fn("OrchestratorMcpService.inspectLaunchWorkspace")( + function* (project: Project, candidatePath: string, requireWorktreeRoot: boolean) { + const projectRepository = yield* detectRepository(project.workspaceRoot, "project root"); + if (projectRepository === null) { + if (requireWorktreeRoot) { + return yield* failure( + "invalid_request", + `Project ${project.id} does not have a repository with linked worktrees.`, + ); + } + return { + canonicalExecutionPath: yield* canonicalPath(project.workspaceRoot, "project root"), + canonicalWorktreeRoot: null, + branch: null, + } as const; + } + if (projectRepository.kind !== "git") { + return yield* failure( + "invalid_request", + `Project ${project.id} uses ${projectRepository.kind}; this launch strategy requires Git.`, + ); + } + + const candidateRepository = yield* detectRepository(candidatePath, "workspace path"); + if (candidateRepository === null || candidateRepository.kind !== "git") { + return yield* failure( + "invalid_request", + `Workspace path '${candidatePath}' is not a Git workspace for project ${project.id}.`, + ); + } + const [canonicalCandidate, canonicalCandidateRoot, projectMetadata, candidateMetadata] = + yield* Effect.all([ + canonicalPath(candidatePath, "workspace path"), + canonicalPath(candidateRepository.repository.rootPath, "workspace root"), + canonicalRepositoryMetadata({ + detectionCwd: project.workspaceRoot, + metadataPath: projectRepository.repository.metadataPath, + }), + canonicalRepositoryMetadata({ + detectionCwd: candidatePath, + metadataPath: candidateRepository.repository.metadataPath, + }), + ]); + if ( + (requireWorktreeRoot && canonicalCandidate !== canonicalCandidateRoot) || + projectMetadata === null || + candidateMetadata === null || + projectMetadata !== candidateMetadata + ) { + return yield* failure( + "invalid_request", + `Workspace path '${candidatePath}' is not a workspace of project ${project.id}.`, + ); + } + return { + canonicalExecutionPath: canonicalCandidate, + canonicalWorktreeRoot: canonicalCandidateRoot, + branch: yield* actualGitBranch(candidateRepository, canonicalCandidate), + } as const; + }, + ); + + const enforceExpectedBranch = ( + expected: string | undefined, + actual: string | null, + cwd: string, + ): Effect.Effect => + expected === undefined || expected === actual + ? Effect.void + : Effect.fail( + failure( + "invalid_request", + `Workspace '${cwd}' is on ${actual === null ? "a detached HEAD" : `branch '${actual}'`}, not requested branch '${expected}'.`, + ), + ); + + const resolveLaunchWorkspace = Effect.fn("OrchestratorMcpService.resolveLaunchWorkspace")( + function* ( + parent: OrchestrationV2ThreadProjection, + project: Project, + requested: OrchestratorMcpCreateThreadsInput["threads"][number]["workspaceStrategy"], + ): Effect.fn.Return { + if (requested?.type === "new_worktree") { + return { + type: "worktree", + baseRef: requested.baseRef, + ...(requested.branch === undefined ? {} : { branch: requested.branch }), + ...(requested.startFromOrigin === undefined + ? {} + : { startFromOrigin: requested.startFromOrigin }), + }; + } + const explicitExisting = requested?.type === "existing_worktree" ? requested : undefined; + const useParentWorktree = + requested === undefined && + project.id === parent.thread.projectId && + parent.thread.worktreePath !== null; + if (explicitExisting !== undefined || useParentWorktree) { + const requestedPath = explicitExisting?.worktreePath ?? parent.thread.worktreePath!; + const workspace = yield* inspectLaunchWorkspace(project, requestedPath, true); + yield* enforceExpectedBranch(explicitExisting?.branch, workspace.branch, requestedPath); + return { + type: "existing_worktree", + worktreePath: workspace.canonicalWorktreeRoot!, + ...(workspace.branch === null ? {} : { branch: workspace.branch }), + }; + } + const workspace = yield* inspectLaunchWorkspace(project, project.workspaceRoot, false); + const expectedBranch = requested?.type === "root" ? requested.branch : undefined; + yield* enforceExpectedBranch(expectedBranch, workspace.branch, project.workspaceRoot); + return { + type: "root", + ...(workspace.branch === null ? {} : { branch: workspace.branch }), + }; + }, + ); + const loadProviders = providerRegistry.getProviders; const resolveTarget = (input: { @@ -1114,6 +1340,13 @@ const make = Effect.gen(function* () { incrementalThreadRead: true, scheduledTasks: true, maxBatchThreads: 20, + projectManagement: true, + projectTargeting: true, + threadLaunchWorkspaceStrategies: [ + "root" as const, + "existing_worktree" as const, + "new_worktree" as const, + ], }, }; }), @@ -1359,6 +1592,13 @@ const make = Effect.gen(function* () { parent.thread.interactionMode, request.interactionMode, ); + const projectId = yield* resolveProjectId(parent, request.projectId); + const project = yield* loadLaunchProject(projectId); + const workspaceStrategy = yield* resolveLaunchWorkspace( + parent, + project, + request.workspaceStrategy, + ); const threadId = stableThreadId({ scope, requestKey: key, @@ -1370,11 +1610,8 @@ const make = Effect.gen(function* () { title: request.title, index, }); - yield* threadManagement - .dispatch({ - type: "thread.create", - createdBy: "agent", - creationSource: "mcp", + const launched = yield* threadLaunch + .launch({ commandId: stableCommandId({ scope, requestKey: key, @@ -1382,70 +1619,51 @@ const make = Effect.gen(function* () { index, }), threadId, - projectId: parent.thread.projectId, + projectId, title, modelSelection: target.modelSelection, runtimeMode, interactionMode, - branch: parent.thread.branch, - worktreePath: parent.thread.worktreePath, + workspaceStrategy, + ...(request.prompt === undefined + ? {} + : { + initialMessage: { + messageId: stableMessageId({ scope, requestKey: key, index }), + text: request.prompt, + attachments: [], + }, + }), + createdBy: "agent", + creationSource: "mcp", }) .pipe( Effect.mapError((error) => failure( "orchestration_error", - `Unable to create thread ${index + 1}: ${errorMessage(error)}`, + `Unable to launch thread ${index + 1}: ${errorMessage(error)}`, ), ), ); - if (request.prompt !== undefined) { - yield* threadManagement - .dispatch({ - type: "message.dispatch", - createdBy: "agent", - creationSource: "mcp", + const projection = launched.projection; + const run = projection.runs.at(-1); + yield* threadManagement + .recordServerCreatedThread({ + targetProjectId: projectId, + command: { + type: "thread.created.record", commandId: stableCommandId({ scope, requestKey: key, - operation: "dispatch-thread", - index, - }), - threadId, - messageId: stableMessageId({ - scope, - requestKey: key, + operation: "record-created-thread", index, }), - text: request.prompt, - attachments: [], - modelSelection: target.modelSelection, - dispatchMode: { type: "start_immediately" }, - }) - .pipe( - Effect.mapError((error) => - failure( - "orchestration_error", - `Unable to start thread ${index + 1}: ${errorMessage(error)}`, - ), - ), - ); - } - const projection = yield* loadProjection(threadId); - const run = projection.runs.at(-1); - yield* threadManagement - .dispatch({ - type: "thread.created.record", - commandId: stableCommandId({ - scope, - requestKey: key, - operation: "record-created-thread", - index, - }), - parentThreadId: scope.threadId, - parentRunId: parentRun.id, - parentNodeId, - targetThreadId: threadId, - targetRunId: run?.id ?? null, + parentThreadId: scope.threadId, + parentRunId: parentRun.id, + parentNodeId, + targetThreadId: threadId, + targetRunId: run?.id ?? null, + }, }) .pipe( Effect.mapError((error) => @@ -1457,6 +1675,7 @@ const make = Effect.gen(function* () { ); return { threadId, + projectId, runId: run?.id ?? null, status: run?.status ?? "idle", title: projection.thread.title, @@ -1474,9 +1693,10 @@ const make = Effect.gen(function* () { Effect.gen(function* () { yield* requireCapability(scope); const parent = yield* loadProjection(scope.threadId); + const projectId = yield* resolveProjectId(parent, input.projectId); const projectThreads = yield* threadManagement .listProjectThreads({ - projectId: parent.thread.projectId, + projectId, includeSubagents: input.includeSubagents !== false, }) .pipe( @@ -1501,7 +1721,7 @@ const make = Effect.gen(function* () { const page = filtered.slice(cursor, cursor + limit); const nextCursor = cursor + page.length < filtered.length ? cursor + page.length : null; return { - projectId: parent.thread.projectId, + projectId, currentThreadId: scope.threadId, threads: page.map(listItemFromShell), nextCursor, @@ -1510,7 +1730,7 @@ const make = Effect.gen(function* () { }), readThread: (scope, input) => Effect.gen(function* () { - const { parent, target } = yield* loadScopedThread(scope, input.threadId); + const { parent, target } = yield* loadScopedThread(scope, input.threadId, input.projectId); const view = input.view ?? "messages"; const afterPosition = input.afterPosition ?? -1; const limit = input.limit ?? DEFAULT_THREAD_READ_LIMIT; @@ -1566,7 +1786,11 @@ const make = Effect.gen(function* () { }), sendToThread: (scope, input) => Effect.gen(function* () { - const { parent, target } = yield* loadScopedThread(scope, input.threadId); + const { parent, projectId, target } = yield* loadScopedThread( + scope, + input.threadId, + input.projectId, + ); yield* resolveRuntimeMode(parent.thread.runtimeMode, target.thread.runtimeMode); yield* resolveInteractionMode(parent.thread.interactionMode, target.thread.interactionMode); @@ -1579,7 +1803,7 @@ const make = Effect.gen(function* () { }); const result = yield* threadManagement .sendToThread({ - projectId: parent.thread.projectId, + projectId, commandId: stableCommandId({ scope, requestKey: key, @@ -1613,10 +1837,10 @@ const make = Effect.gen(function* () { }), waitForThread: (scope, input) => Effect.gen(function* () { - const { parent } = yield* loadScopedThread(scope, input.threadId); + const { projectId } = yield* loadScopedThread(scope, input.threadId, input.projectId); const result = yield* threadManagement .waitForThread({ - projectId: parent.thread.projectId, + projectId, threadId: input.threadId, ...(input.runId === undefined ? {} : { runId: input.runId }), timeoutMs: Math.min( @@ -1634,11 +1858,11 @@ const make = Effect.gen(function* () { }), interruptThread: (scope, input) => Effect.gen(function* () { - const { parent } = yield* loadScopedThread(scope, input.threadId); + const { projectId } = yield* loadScopedThread(scope, input.threadId, input.projectId); const key = yield* requestKey(input.clientRequestId); const result = yield* threadManagement .interruptThread({ - projectId: parent.thread.projectId, + projectId, commandId: stableCommandId({ scope, requestKey: key, @@ -1677,5 +1901,13 @@ const make = Effect.gen(function* () { export const layer: Layer.Layer< OrchestratorMcpService, never, - Crypto.Crypto | ThreadManagementService | ProviderRegistry | ScheduledTaskService + | Crypto.Crypto + | FileSystem.FileSystem + | Path.Path + | ThreadManagementService + | ProviderRegistry + | ProjectService.ProjectService + | ScheduledTaskService + | ThreadLaunch.ThreadLaunchService + | VcsDriverRegistry.VcsDriverRegistry > = Layer.effect(OrchestratorMcpService, make); diff --git a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts index ad6ea73c7688..1786b9a8345d 100644 --- a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts @@ -36,6 +36,7 @@ import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; import * as PubSub from "effect/PubSub"; import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; @@ -44,7 +45,10 @@ import { McpSchema, McpServer } from "effect/unstable/ai"; import { ClaudeProviderCapabilitiesV2 } from "../orchestration-v2/Adapters/ClaudeAdapterV2.ts"; import { CodexProviderCapabilitiesV2 } from "../orchestration-v2/Adapters/CodexAdapterV2.ts"; +import * as CommandReceiptStore from "../orchestration-v2/CommandReceiptStore.ts"; +import * as IdAllocator from "../orchestration-v2/IdAllocator.ts"; import { OrchestratorV2, type OrchestratorV2Shape } from "../orchestration-v2/Orchestrator.ts"; +import * as ThreadLaunch from "../orchestration-v2/ThreadLaunchService.ts"; import { layer as threadManagementServiceLayer } from "../orchestration-v2/ThreadManagementService.ts"; import { type ProviderAdapterV2Event, @@ -59,13 +63,22 @@ import { } from "../orchestration-v2/ProviderContinuationRequests.ts"; import { checkpointWorkspace } from "../orchestration-v2/testkit/ReplayFixtureWorkspace.ts"; import { makeOrchestratorV2ReplayLayerWithRegistry } from "../orchestration-v2/testkit/ProviderReplayHarness.ts"; +import * as GitWorkflow from "../git/GitWorkflowService.ts"; +import { SqlitePersistenceMemory } from "../persistence/Layers/Sqlite.ts"; import { makeProviderRegistryLayer } from "../provider/testUtils/providerRegistryMock.ts"; +import * as ProjectService from "../project/ProjectService.ts"; +import * as ProjectSetupScriptRunner from "../project/ProjectSetupScriptRunner.ts"; import { ScheduledTaskService } from "../scheduledTasks/ScheduledTaskService.ts"; +import * as ServerSettings from "../serverSettings.ts"; +import * as TextGeneration from "../textGeneration/TextGeneration.ts"; +import * as VcsDriverRegistry from "../vcs/VcsDriverRegistry.ts"; +import * as VcsProcess from "../vcs/VcsProcess.ts"; import * as McpHttpServer from "./McpHttpServer.ts"; import * as McpInvocationContext from "./McpInvocationContext.ts"; const parentThreadId = ThreadId.make("thread:mcp-orchestrator-parent"); const projectId = ProjectId.make("project:mcp-orchestrator"); +const targetProjectId = ProjectId.make("project:mcp-orchestrator-target"); const codexInstanceId = ProviderInstanceId.make("codex"); const claudeInstanceId = ProviderInstanceId.make("claudeAgent"); const codexModel = "gpt-5.4"; @@ -429,6 +442,7 @@ describe("orchestrator MCP toolkit", () => { Effect.scoped( Effect.gen(function* () { const cwd = yield* checkpointWorkspace("orchestrator-mcp-toolkit"); + const targetWorkspace = yield* checkpointWorkspace("orchestrator-mcp-target"); const capturedTurns = yield* Ref.make>([]); const parentTerminalGates = new Map>(); const deliveryTerminalGates = new Map>(); @@ -496,6 +510,7 @@ describe("orchestrator MCP toolkit", () => { expect(yield* Ref.get(continuationOffers)).toHaveLength(count); } }); + const databaseLayer = SqlitePersistenceMemory; const orchestratorLayer = makeOrchestratorV2ReplayLayerWithRegistry( { name: "orchestrator-mcp-toolkit", @@ -510,6 +525,7 @@ describe("orchestrator MCP toolkit", () => { }, }, registryLayer, + { databaseLayer }, ).pipe(Layer.provide(continuationProbeLayer)); const orchestrationLayer = Layer.merge( orchestratorLayer, @@ -544,6 +560,27 @@ describe("orchestrator MCP toolkit", () => { model: "opencode/test", }), ]); + const projectLayer = Layer.mock(ProjectService.ProjectService)({ + getById: (requestedProjectId) => + Effect.succeed( + requestedProjectId === projectId || requestedProjectId === targetProjectId + ? Option.some({ + id: requestedProjectId, + title: + requestedProjectId === projectId ? "MCP project" : "MCP target project", + workspaceRoot: requestedProjectId === projectId ? cwd : targetWorkspace, + repositoryIdentity: null, + faviconPath: null, + defaultModelSelection: codexSelection, + defaultThreadEnvMode: "worktree", + scripts: [], + createdAt: "2026-08-29T12:00:00.000Z", + updatedAt: "2026-08-29T12:00:00.000Z", + deletedAt: null, + }) + : Option.none(), + ), + }); // In-memory ScheduledTaskService stub so the schedule/list/update/ // delete tools can be exercised without SQL/launch wiring. const scheduledStore = yield* Ref.make>([]); @@ -570,11 +607,36 @@ describe("orchestrator MCP toolkit", () => { runNow: () => Effect.die("ScheduledTaskService.runNow is unused in this test"), }), ); + const receiptLayer = CommandReceiptStore.layer.pipe(Layer.provide(databaseLayer)); + const threadLaunchLayer = ThreadLaunch.layer.pipe( + Layer.provide( + Layer.mergeAll( + projectLayer, + Layer.mock(GitWorkflow.GitWorkflowService)({}), + Layer.mock(ProjectSetupScriptRunner.ProjectSetupScriptRunner)({ + runForThread: () => Effect.succeed({ status: "no-script" }), + }), + Layer.mock(TextGeneration.TextGeneration)({}), + ServerSettings.layerTest({}), + providerRegistryLayer, + orchestrationLayer, + receiptLayer, + IdAllocator.layer, + ), + ), + ); + const vcsProcessLayer = VcsProcess.layer.pipe(Layer.provide(NodeServices.layer)); + const vcsDriverRegistryLayer = VcsDriverRegistry.layer.pipe( + Layer.provide(vcsProcessLayer), + ); const testLayer = McpHttpServer.OrchestratorToolkitRegistrationLive.pipe( Layer.provideMerge(McpServer.McpServer.layer), Layer.provideMerge(orchestrationLayer), Layer.provide(providerRegistryLayer), Layer.provide(scheduledTaskStubLayer), + Layer.provide(projectLayer), + Layer.provide(threadLaunchLayer), + Layer.provide(vcsDriverRegistryLayer), Layer.provide(NodeServices.layer), ); @@ -1628,6 +1690,7 @@ describe("orchestrator MCP toolkit", () => { expect( createdThreadItems.map((item) => ({ targetThreadId: item.targetThreadId, + targetProjectId: item.targetProjectId, targetRunId: item.targetRunId, title: item.title, providerInstanceId: item.targetProviderInstanceId, @@ -1636,6 +1699,7 @@ describe("orchestrator MCP toolkit", () => { ).toEqual([ { targetThreadId: emptyThread.threadId, + targetProjectId: projectId, targetRunId: null, title: emptyThread.title, providerInstanceId: codexInstanceId, @@ -1643,6 +1707,7 @@ describe("orchestrator MCP toolkit", () => { }, { targetThreadId: promptedThread.threadId, + targetProjectId: projectId, targetRunId: promptedThread.runId, title: promptedThread.title, providerInstanceId: claudeInstanceId, @@ -1650,6 +1715,48 @@ describe("orchestrator MCP toolkit", () => { }, ]); + const crossProjectCall = yield* invoke("create_threads", { + clientRequestId: "create-cross-project-thread-1", + threads: [{ projectId: targetProjectId, title: "Cross-project ordinary thread" }], + }); + expect(crossProjectCall.isError).toBe(false); + const crossProjectCreated = yield* decodeCreateThreadsResult( + crossProjectCall.structuredContent, + ).pipe(Effect.orDie); + const crossProjectThread = crossProjectCreated.threads[0]!; + expect(crossProjectThread.projectId).toBe(targetProjectId); + expect( + (yield* orchestrator.getThreadProjection(crossProjectThread.threadId)).thread + .projectId, + ).toBe(targetProjectId); + const crossProjectItem = (yield* orchestrator.getThreadProjection( + parentThreadId, + )).visibleTurnItems + .map((row) => row.item) + .find( + (item) => + item.type === "thread_created" && + item.targetThreadId === crossProjectThread.threadId, + ); + expect(crossProjectItem).toMatchObject({ + type: "thread_created", + targetThreadId: crossProjectThread.threadId, + targetProjectId, + targetRunId: null, + }); + const untrustedCrossProjectRecord = yield* orchestrator + .dispatch({ + type: "thread.created.record", + commandId: CommandId.make("command:mcp-parent:untrusted-cross-project-record"), + parentThreadId, + parentRunId: parentRun.id, + parentNodeId: parentRootNodeId, + targetThreadId: crossProjectThread.threadId, + targetRunId: null, + }) + .pipe(Effect.flip); + expect(untrustedCrossProjectRecord._tag).toBe("OrchestratorDispatchError"); + const repeatedCreateCall = yield* invoke("create_threads", createInput); const repeatedCreated = yield* decodeCreateThreadsResult( repeatedCreateCall.structuredContent, diff --git a/apps/server/src/mcp/toolkits/orchestrator/handlers.ts b/apps/server/src/mcp/toolkits/orchestrator/handlers.ts index ab9f55341822..fbd3c553d3a6 100644 --- a/apps/server/src/mcp/toolkits/orchestrator/handlers.ts +++ b/apps/server/src/mcp/toolkits/orchestrator/handlers.ts @@ -70,6 +70,10 @@ const handlers = { prompt: input.prompt, ...(input.title === undefined ? {} : { title: input.title }), ...(input.target === undefined ? {} : { target: input.target }), + ...(input.projectId === undefined ? {} : { projectId: input.projectId }), + ...(input.workspaceStrategy === undefined + ? {} + : { workspaceStrategy: input.workspaceStrategy }), ...(input.runtimeMode === undefined ? {} : { runtimeMode: input.runtimeMode }), ...(input.interactionMode === undefined ? {} diff --git a/apps/server/src/mcp/toolkits/orchestrator/tools.ts b/apps/server/src/mcp/toolkits/orchestrator/tools.ts index 9a453695ad30..ace5e67f8b2d 100644 --- a/apps/server/src/mcp/toolkits/orchestrator/tools.ts +++ b/apps/server/src/mcp/toolkits/orchestrator/tools.ts @@ -36,7 +36,7 @@ const dependencies = [McpInvocationContext.McpInvocationContext, OrchestratorMcp export const OrchestratorCapabilitiesTool = Tool.make("orchestrator_capabilities", { description: - "List the V2 provider instances, models, inherited runtime settings, and app-owned orchestration features available to this T3 thread.", + "List the V2 provider instances, models, inherited runtime settings, project targeting, workspace launch strategies, and app-owned orchestration features available to this T3 thread.", success: OrchestratorMcpCapabilitiesResult, failure: OrchestratorMcpFailure, failureMode: "return", @@ -138,7 +138,7 @@ export const DeleteScheduledTaskTool = Tool.make("delete_scheduled_task", { export const CreateThreadsTool = Tool.make("create_threads", { description: - "Create one or more ORDINARY TOP-LEVEL T3 conversations. This is not delegation and does not create child agents/subagents. If the user asks for agents, subagents, workers, delegation, or parallel help, call delegate_task once per child instead—even when selecting different providers. Use create_threads only when the user explicitly asks for separate/new/top-level threads or conversations. Each entry may override provider, model, options, runtime mode, and interaction mode; omitted settings inherit.", + "Create one or more ORDINARY TOP-LEVEL T3 conversations. This is not delegation and does not create child agents/subagents. If the user asks for agents, subagents, workers, delegation, or parallel help, call delegate_task once per child instead—even when selecting different providers. Use create_threads only when the user explicitly asks for separate/new/top-level threads or conversations. Each entry may select a project and root, existing_worktree, or new_worktree workspace strategy, and may override provider, model, options, runtime mode, and interaction mode. Omitted project uses the caller's project; omitted workspace reuses the caller's checkout only in that project and otherwise uses the selected project's root. Provider and runtime settings inherit from the caller and remain within its permission ceiling.", parameters: OrchestratorMcpCreateThreadsInput, success: OrchestratorMcpCreateThreadsResult, failure: OrchestratorMcpFailure, @@ -151,7 +151,7 @@ export const CreateThreadsTool = Tool.make("create_threads", { export const ThreadStartTool = Tool.make("t3_thread_start", { description: - "Create an ordinary TOP-LEVEL T3 conversation and immediately start its first turn. This is not a child agent/subagent; use delegate_task for delegated work. The new thread inherits this thread's project, checkout, provider, model, and runtime settings unless overridden. Use t3_thread_wait and t3_thread_read to collect its result.", + "Create an ordinary TOP-LEVEL T3 conversation and immediately start its first turn. This is not a child agent/subagent; use delegate_task for delegated work. projectId may select another project in this environment; workspaceStrategy may use its root, an existing worktree, or a new worktree. Omitted workspace reuses the caller's checkout only in the current project and otherwise uses the target project's root. Provider, model, and runtime settings inherit from the caller unless overridden, within the caller's permission ceiling. Use t3_thread_wait and t3_thread_read to collect its result.", parameters: OrchestratorMcpThreadStartInput, success: OrchestratorMcpCreatedThread, failure: OrchestratorMcpFailure, @@ -164,7 +164,7 @@ export const ThreadStartTool = Tool.make("t3_thread_start", { export const ThreadListTool = Tool.make("t3_thread_list", { description: - "List T3 threads in the calling thread's project, newest first. Filter by durable run status or title and paginate with the returned cursor. Threads from other projects are never exposed.", + "List T3 threads in the selected project in this environment, newest first. Omit projectId for the calling thread's project. Filter by durable run status or title and paginate with the returned cursor.", parameters: OrchestratorMcpThreadListInput, success: OrchestratorMcpThreadListResult, failure: OrchestratorMcpFailure, @@ -178,7 +178,7 @@ export const ThreadListTool = Tool.make("t3_thread_list", { export const ThreadReadTool = Tool.make("t3_thread_read", { description: - "Read durable state and a paginated timeline from a T3 thread in the calling project. The default messages view returns user messages, assistant messages, and proposed plans; activity returns all summarized timeline items. Reading an untruncated terminal assistant result from this parent thread's direct app-owned child acknowledges that child's automatic completion delivery. Continue with afterPosition=nextPosition.", + "Read durable state and a paginated timeline from a T3 thread in the selected project in this environment. Omit projectId for the calling thread's project. The default messages view returns user messages, assistant messages, and proposed plans; activity returns all summarized timeline items. Reading an untruncated terminal assistant result from this parent thread's direct app-owned child acknowledges that child's automatic completion delivery. Continue with afterPosition=nextPosition.", parameters: OrchestratorMcpThreadReadInput, success: OrchestratorMcpThreadReadResult, failure: OrchestratorMcpFailure, @@ -192,7 +192,7 @@ export const ThreadReadTool = Tool.make("t3_thread_read", { export const ThreadSendTool = Tool.make("t3_thread_send", { description: - "Send a message to a T3 thread in the calling project. mode='auto' starts an idle thread, steers a fully active turn, or queues behind a turn that is not yet steerable. Use queue for a separate follow-up turn, steer for an in-flight update, or restart to interrupt-and-restart the active turn. clientRequestId makes retries idempotent.", + "Send a message to a T3 thread in the selected project in this environment. Omit projectId for the calling thread's project. mode='auto' starts an idle thread, steers a fully active turn, or queues behind a turn that is not yet steerable. Use queue for a separate follow-up turn, steer for an in-flight update, or restart to interrupt-and-restart the active turn. clientRequestId makes retries idempotent.", parameters: OrchestratorMcpThreadSendInput, success: OrchestratorMcpThreadSendResult, failure: OrchestratorMcpFailure, @@ -205,7 +205,7 @@ export const ThreadSendTool = Tool.make("t3_thread_send", { export const ThreadWaitTool = Tool.make("t3_thread_wait", { description: - "Wait for a T3 thread run to reach a terminal durable state. Without runId, the latest run at call time is selected; an idle thread returns immediately. Timeout does not interrupt work, so call again or use t3_thread_read/list after timedOut=true. Waiting reports status only and does not acknowledge a delegated result.", + "Wait for a T3 thread run in the selected project in this environment to reach a terminal durable state. Omit projectId for the calling thread's project. Without runId, the latest run at call time is selected; an idle thread returns immediately. Timeout does not interrupt work, so call again or use t3_thread_read/list after timedOut=true. Waiting reports status only and does not acknowledge a delegated result.", parameters: OrchestratorMcpThreadWaitInput, success: OrchestratorMcpThreadWaitResult, failure: OrchestratorMcpFailure, @@ -219,7 +219,7 @@ export const ThreadWaitTool = Tool.make("t3_thread_wait", { export const ThreadInterruptTool = Tool.make("t3_thread_interrupt", { description: - "Request interruption of a running turn in a T3 thread in the calling project. Without runId, the newest interruptible run is selected. Terminal runs and threads without an active turn return without another side effect. clientRequestId makes retries idempotent.", + "Request interruption of a running turn in a T3 thread in the selected project in this environment. Omit projectId for the calling thread's project. Without runId, the newest interruptible run is selected. Terminal runs and threads without an active turn return without another side effect. clientRequestId makes retries idempotent.", parameters: OrchestratorMcpThreadInterruptInput, success: OrchestratorMcpThreadInterruptResult, failure: OrchestratorMcpFailure, diff --git a/apps/server/src/mcp/toolkits/worktree/registration.test.ts b/apps/server/src/mcp/toolkits/worktree/registration.test.ts index f49f6e2def6c..5583c2d8a154 100644 --- a/apps/server/src/mcp/toolkits/worktree/registration.test.ts +++ b/apps/server/src/mcp/toolkits/worktree/registration.test.ts @@ -10,6 +10,7 @@ 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 ThreadLaunch from "../../../orchestration-v2/ThreadLaunchService.ts"; import * as ProjectService from "../../../project/ProjectService.ts"; import * as ProjectSetupScriptRunner from "../../../project/ProjectSetupScriptRunner.ts"; import { ProviderRegistry } from "../../../provider/Services/ProviderRegistry.ts"; @@ -17,12 +18,14 @@ import { ScheduledTaskService } from "../../../scheduledTasks/ScheduledTaskServi import * as ServerSettings from "../../../serverSettings.ts"; import * as SourceControlRepositoryService from "../../../sourceControl/SourceControlRepositoryService.ts"; import { VcsStatusBroadcaster } from "../../../vcs/VcsStatusBroadcaster.ts"; +import * as VcsDriverRegistry from "../../../vcs/VcsDriverRegistry.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(ThreadLaunch.ThreadLaunchService)({}), Layer.mock(ProviderRegistry)({}), Layer.mock(ScheduledTaskService)({}), Layer.mock(ProjectService.ProjectService)({}), @@ -31,6 +34,7 @@ const StubServicesLive = Layer.mergeAll( Layer.mock(ProjectSetupScriptRunner.ProjectSetupScriptRunner)({}), Layer.mock(SourceControlRepositoryService.SourceControlRepositoryService)({}), Layer.mock(VcsStatusBroadcaster)({}), + Layer.mock(VcsDriverRegistry.VcsDriverRegistry)({ detect: () => Effect.succeed(null) }), ); const ToolsListPayload = Schema.fromJsonString( diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 1cb8fe1a8565..a2e640236276 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -24,6 +24,7 @@ import { type OrchestrationV2Subagent, type OrchestrationV2ThreadProjection, type OrchestrationV2TurnItem, + type ProjectId, ProviderInstanceId, type ProviderSessionId, RunId, @@ -171,11 +172,25 @@ export interface OrchestratorV2DispatchResult { readonly storedEvents: ReadonlyArray; } +export type OrchestratorV2CreatedThreadRecordCommand = Extract< + OrchestrationV2Command, + { readonly type: "thread.created.record" } +>; + +export interface OrchestratorV2ServerCreatedThreadRecordInput { + readonly command: OrchestratorV2CreatedThreadRecordCommand; + readonly targetProjectId: ProjectId; +} + export interface OrchestratorV2Shape { readonly resumeQueuedRuns: Effect.Effect; readonly dispatch: ( command: OrchestrationV2Command, ) => Effect.Effect; + /** Records a cross-project MCP launch through a server-owned authorization boundary. */ + readonly recordServerCreatedThread: ( + input: OrchestratorV2ServerCreatedThreadRecordInput, + ) => Effect.Effect; readonly getThreadProjection: ( threadId: ThreadId, ) => Effect.Effect; @@ -4867,8 +4882,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const dispatchCreatedThreadRecord = Effect.fn("orchestrationV2.dispatch.createdThreadRecord")( function* ( - command: Extract, + command: OrchestratorV2CreatedThreadRecordCommand, events: Ref.Ref>, + authorizedTargetProjectId: ProjectId | undefined, ) { const parentProjection = yield* projectionStore .getThreadProjection(command.parentThreadId) @@ -4910,13 +4926,26 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio cause: `Parent node ${command.parentNodeId} is not the root of run ${command.parentRunId}.`, }); } - if (parentProjection.thread.projectId !== targetProjection.thread.projectId) { + if ( + authorizedTargetProjectId === undefined && + parentProjection.thread.projectId !== targetProjection.thread.projectId + ) { return yield* new OrchestratorDispatchError({ commandId: command.commandId, commandType: command.type, cause: `Target thread ${command.targetThreadId} belongs to another project.`, }); } + if ( + authorizedTargetProjectId !== undefined && + authorizedTargetProjectId !== targetProjection.thread.projectId + ) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: `Target thread ${command.targetThreadId} does not belong to authorized project ${authorizedTargetProjectId}.`, + }); + } if ( command.targetRunId !== null && !targetProjection.runs.some((candidate) => candidate.id === command.targetRunId) @@ -4947,6 +4976,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio updatedAt: now, type: "thread_created", targetThreadId: command.targetThreadId, + targetProjectId: targetProjection.thread.projectId, targetRunId: command.targetRunId, targetProviderInstanceId: targetProjection.thread.modelSelection.instanceId, targetModel: targetProjection.thread.modelSelection.model, @@ -6748,6 +6778,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const dispatchOnce = Effect.fn("orchestrationV2.dispatch.once")(function* ( command: OrchestrationV2Command, + options?: { readonly authorizedCreatedThreadProjectId?: ProjectId }, ): Effect.fn.Return< { readonly events: ReadonlyArray; @@ -6850,7 +6881,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio yield* dispatchDelegatedTaskCompletionDeliveryResolution(command, events); break; case "thread.created.record": - yield* dispatchCreatedThreadRecord(command, events); + yield* dispatchCreatedThreadRecord( + command, + events, + options?.authorizedCreatedThreadProjectId, + ); break; default: return yield* dispatchUnsupported(command); @@ -6864,6 +6899,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const dispatchWithReceiptEffect = Effect.fn("orchestrationV2.dispatch.withReceipt")(function* ( command: OrchestrationV2Command, + options?: { readonly authorizedCreatedThreadProjectId?: ProjectId }, ): Effect.fn.Return { yield* Effect.annotateCurrentSpan({ "orchestration_v2.command_id": command.commandId, @@ -6921,7 +6957,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio } satisfies OrchestratorV2DispatchResult; } - const plan = yield* dispatchOnce(command).pipe( + const plan = yield* dispatchOnce(command, options).pipe( Effect.flatMap((planned) => planned.events.length > 0 ? Effect.succeed(planned) @@ -7003,6 +7039,14 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const dispatchWithReceipt = (command: OrchestrationV2Command) => threadDispatch.withLock(commandThreadId(command), dispatchWithReceiptEffect(command)); + const recordServerCreatedThread = (input: OrchestratorV2ServerCreatedThreadRecordInput) => + threadDispatch.withLock( + commandThreadId(input.command), + dispatchWithReceiptEffect(input.command, { + authorizedCreatedThreadProjectId: input.targetProjectId, + }), + ); + const handleTerminalRun = (stored: OrchestrationV2StoredEvent) => Effect.gen(function* () { const threadId = stored.event.threadId; @@ -7157,6 +7201,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio return OrchestratorV2.of({ resumeQueuedRuns, dispatch: dispatchWithReceipt, + recordServerCreatedThread, getThreadProjection: (threadId) => projectionStore .getThreadProjection(threadId) @@ -7255,6 +7300,14 @@ export const layerUnavailable: Layer.Layer = Layer.succeed( cause: "Orchestration V2 live runtime is not configured.", }), ), + recordServerCreatedThread: ({ command }) => + Effect.fail( + new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: "Orchestration V2 live runtime is not configured.", + }), + ), getThreadProjection: (threadId) => Effect.fail( new OrchestratorProjectionError({ diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index b51eec7ba188..8c51616d833a 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -30,6 +30,7 @@ import { OrchestratorV2, type OrchestratorV2DispatchResult, type OrchestratorV2Error, + type OrchestratorV2ServerCreatedThreadRecordInput, } from "./Orchestrator.ts"; import { LegacyV1ThreadImporter, @@ -276,6 +277,9 @@ export interface ThreadManagementServiceShape { readonly dispatch: ( command: OrchestrationV2Command, ) => Effect.Effect; + readonly recordServerCreatedThread: ( + input: OrchestratorV2ServerCreatedThreadRecordInput, + ) => Effect.Effect; readonly getThreadProjection: ( threadId: ThreadId, ) => Effect.Effect; @@ -439,6 +443,13 @@ const make = Effect.gen(function* () { ); }; + const recordServerCreatedThread: ThreadManagementServiceShape["recordServerCreatedThread"] = ( + input, + ) => + ensureCommandTranscripts(input.command).pipe( + Effect.andThen(orchestrator.recordServerCreatedThread(input)), + ); + const getProjectThread: ThreadManagementServiceShape["getProjectThread"] = (input) => getThreadProjection(input.threadId).pipe( Effect.mapError( @@ -668,6 +679,7 @@ const make = Effect.gen(function* () { withProjectMutationLock: projectMutations.withLock, ensureLegacyTranscript, dispatch, + recordServerCreatedThread, getThreadProjection, getThreadSnapshot, getProjectThread, diff --git a/docs/orchestration-v2/orchestrator-mcp-server.md b/docs/orchestration-v2/orchestrator-mcp-server.md index 8e452649f17f..a4ca6aa49143 100644 --- a/docs/orchestration-v2/orchestrator-mcp-server.md +++ b/docs/orchestration-v2/orchestrator-mcp-server.md @@ -179,7 +179,8 @@ Returns: - the parent runtime and interaction modes; - registered provider instances and advertised models; - whether each provider can run a child task; and -- feature flags for polling, cancellation, and batch thread creation. +- feature flags for polling, cancellation, batch thread creation, project targeting, and the + supported thread workspace strategies. Unavailable providers include model-visible constraints such as missing V2 adapter support, disabled state, missing executable, or missing authentication. @@ -261,6 +262,11 @@ type CreateThreadsInput = { driverKind?: string; model?: string; }; + projectId?: string; + workspaceStrategy?: + | { type: "root"; branch?: string } + | { type: "existing_worktree"; worktreePath: string; branch?: string } + | { type: "new_worktree"; baseRef: string; branch?: string; startFromOrigin?: boolean }; runtimeMode?: "inherit" | "approval-required" | "auto-accept-edits" | "full-access"; interactionMode?: "inherit" | "plan" | "default"; }>; @@ -268,24 +274,44 @@ type CreateThreadsInput = { }; ``` -Each entry independently resolves provider, model, and modes. The new threads -inherit the parent's project, branch, and worktree path, but they have no -sub-agent lineage. Entries with a prompt immediately dispatch a run; entries -without a prompt remain idle. +Each entry independently resolves its project, workspace, provider, model, and modes. With no +`projectId`, a thread uses the parent's project. With no `workspaceStrategy`, a thread in the +parent's project reuses the parent's checkout, while a thread in another project uses that +project's root and never inherits the parent's worktree path. An explicit strategy can use the +project root, register a known existing worktree for the new thread, or ask `ThreadLaunchService` +to provision a new worktree. Root and existing-worktree launches inspect the checkout before +launch. A supplied branch is an expected-branch constraint, not a request to switch branches, and +the call fails when the checkout is detached or on another branch. Existing-worktree paths are +canonicalized and must resolve to the physical root of a Git worktree belonging to the selected +project's repository. A root launch still uses the project's configured workspace root, including +when that execution directory is a subdirectory of the physical Git worktree. + +Project selection does not raise the caller's authority. Provider, model, runtime mode, and +interaction mode still inherit from the caller, and requested modes cannot exceed its permission +ceiling. Project model and workspace-mode defaults remain client defaults; the MCP request's +explicit strategy wins, followed by current-checkout reuse in the current project, then target-root +fallback for a different project. The new threads have no sub-agent lineage. Entries with a prompt +use the normal deferred launch path; entries without a prompt remain idle. + +The parent timeline records the selected project alongside each created thread. Cross-project +recording uses a server-owned orchestration entry point that checks the MCP-selected project +against the durable target-thread projection. The public `thread.created.record` command retains +its same-project restriction and cannot promote a caller-supplied project ID into authorization. ### `t3_thread_start` Creates one ordinary top-level thread and immediately dispatches its first prompt. It is the single-thread convenience form of `create_threads` and returns the created thread and run IDs. Use `clientRequestId` when a caller may -retry the request. +retry the request. It accepts the same optional `projectId` and `workspaceStrategy` fields. ### `t3_thread_list` -Lists durable thread shells in the calling thread's project, newest first. +Lists durable thread shells in an explicitly selected project in the same environment, newest +first. Omitting `projectId` keeps the calling thread's project as the scope. Callers can filter by title, run status, and whether app-owned sub-agent threads are included. Results are bounded and offset-paginated. Deleted threads and -threads from other projects are never exposed. +threads outside the selected project are never exposed. ### `t3_thread_read` @@ -302,9 +328,14 @@ and `creationSource: "mcp"`; provider output uses `creationSource: "provider"`. Actor and ingress are separate so agent-authored user-role messages remain distinguishable from human-authored messages. +`t3_thread_read`, `t3_thread_send`, `t3_thread_wait`, and `t3_thread_interrupt` accept an optional +`projectId`. The server resolves it through the local `ProjectService`, then verifies that the +thread belongs to that project. No tool accepts an arbitrary environment ID. Organization remains +scoped to the current project until its shared resolver is adopted there. + ### `t3_thread_send` -Sends a message to an ordinary or delegated thread in the calling project: +Sends a message to an ordinary or delegated thread in the selected project: - `auto` starts an idle thread, steers a fully active turn, or queues behind a turn that is not yet steerable; diff --git a/docs/user/project-settings.md b/docs/user/project-settings.md index 43f4dd6ebf8e..acc25ebd67c2 100644 --- a/docs/user/project-settings.md +++ b/docs/user/project-settings.md @@ -24,3 +24,20 @@ or update the same project settings available in the app. Removing a project does not remove its directory, repository, worktrees, or other workspace files. If the project still has threads, the agent must explicitly request that T3 Code delete those thread records first. This prevents a project removal from silently discarding conversations. + +# Start a thread in another project + +An agent can start an ordinary top-level conversation in any project registered in the same T3 +Code environment. It can use the project root, an existing worktree, or ask T3 Code to create a new +worktree. If it selects another project without choosing a workspace, T3 Code uses that project's +root and does not reuse the calling thread's checkout. + +Choosing an existing worktree does not switch its branch. T3 Code verifies that the path is a real +worktree root of the selected project's repository, resolves symlinks to the canonical checkout, +and rejects a requested branch when the checkout is currently on a different branch. Projects +whose configured root is a folder inside a repository continue to launch from that folder. + +The new thread keeps the calling agent's provider, model, runtime mode, and interaction mode unless +the request supplies a narrower supported override. Selecting a project does not grant broader +permissions. Agents can also list, read, message, wait for, and interrupt threads in that selected +project; leaving the project unspecified keeps the current project as the default. diff --git a/packages/contracts/src/orchestrationV2.test.ts b/packages/contracts/src/orchestrationV2.test.ts index 21ec7e75e214..c6716880d9a3 100644 --- a/packages/contracts/src/orchestrationV2.test.ts +++ b/packages/contracts/src/orchestrationV2.test.ts @@ -269,6 +269,7 @@ describe("orchestration V2 contracts", () => { status: "completed", title: "Child thread", targetThreadId: "thread-child-1", + targetProjectId: "project-child-1", targetRunId: "run-child-1", targetProviderInstanceId: "claude-default", targetModel: "claude-sonnet-4-6", @@ -287,6 +288,7 @@ describe("orchestration V2 contracts", () => { throw new Error("expected thread_created"); } expect(item.targetRunId).toBe(RunId.make("run-child-1")); + expect(item.targetProjectId).toBe(ProjectId.make("project-child-1")); }); it("decodes provider-neutral replay transcripts", () => { diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 801ed5f4d29a..25f8ea0ca9f2 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -1046,6 +1046,7 @@ export const OrchestrationV2TurnItem = Schema.Union([ ...OrchestrationV2TurnItemBaseFields, type: Schema.Literal("thread_created"), targetThreadId: ThreadId, + targetProjectId: Schema.optional(ProjectId), targetRunId: Schema.NullOr(RunId), targetProviderInstanceId: ProviderInstanceId, targetModel: TrimmedNonEmptyString, @@ -1719,6 +1720,7 @@ export const OrchestrationV2TurnItemJson = Schema.Union([ ...OrchestrationV2TurnItemJsonBaseFields, type: Schema.Literal("thread_created"), targetThreadId: ThreadId, + targetProjectId: Schema.optional(ProjectId), targetRunId: Schema.NullOr(RunId), targetProviderInstanceId: ProviderInstanceId, targetModel: TrimmedNonEmptyString, diff --git a/packages/contracts/src/orchestratorMcp.test.ts b/packages/contracts/src/orchestratorMcp.test.ts index c668de287a92..178fadc70db0 100644 --- a/packages/contracts/src/orchestratorMcp.test.ts +++ b/packages/contracts/src/orchestratorMcp.test.ts @@ -112,6 +112,12 @@ describe("orchestrator MCP contracts", () => { { prompt: "Review the API.", target: { driverKind: "claudeAgent" }, + projectId: "project-review", + workspaceStrategy: { + type: "new_worktree", + baseRef: "trunk", + branch: "review-api", + }, runtimeMode: "approval-required", }, ], @@ -120,17 +126,25 @@ describe("orchestrator MCP contracts", () => { expect(request.threads).toHaveLength(2); expect(request.threads[0]?.prompt).toBeUndefined(); expect(request.threads[1]?.target?.driverKind).toBe("claudeAgent"); + expect(request.threads[1]?.workspaceStrategy).toEqual({ + type: "new_worktree", + baseRef: "trunk", + branch: "review-api", + }); }); it("decodes project-scoped thread orchestration requests", () => { expect( decodeThreadStartInput({ prompt: "Run the first loop iteration.", + projectId: "project-loop", + workspaceStrategy: { type: "root" }, clientRequestId: "start-loop-1", }).prompt, ).toBe("Run the first loop iteration."); expect( decodeThreadListInput({ + projectId: "project-loop", statuses: ["running", "completed"], includeSubagents: false, limit: 25, @@ -139,6 +153,7 @@ describe("orchestrator MCP contracts", () => { expect( decodeThreadReadInput({ threadId: "thread-loop-1", + projectId: "project-loop", view: "activity", afterPosition: 10, }).afterPosition, @@ -146,6 +161,7 @@ describe("orchestrator MCP contracts", () => { expect( decodeThreadSendInput({ threadId: "thread-loop-1", + projectId: "project-loop", message: "Continue with the next iteration.", mode: "steer", clientRequestId: "send-loop-2", @@ -154,6 +170,7 @@ describe("orchestrator MCP contracts", () => { expect( decodeThreadWaitInput({ threadId: "thread-loop-1", + projectId: "project-loop", runId: "run-loop-2", timeoutMs: 5_000, }).runId, @@ -161,6 +178,7 @@ describe("orchestrator MCP contracts", () => { expect( decodeThreadInterruptInput({ threadId: "thread-loop-1", + projectId: "project-loop", reason: "Loop converged.", }).reason, ).toBe("Loop converged."); diff --git a/packages/contracts/src/orchestratorMcp.ts b/packages/contracts/src/orchestratorMcp.ts index 7e62e9351d31..7ceaf56795e5 100644 --- a/packages/contracts/src/orchestratorMcp.ts +++ b/packages/contracts/src/orchestratorMcp.ts @@ -202,10 +202,35 @@ export const OrchestratorMcpTaskCancelResult = Schema.Struct({ }); export type OrchestratorMcpTaskCancelResult = typeof OrchestratorMcpTaskCancelResult.Type; +export const OrchestratorMcpThreadWorkspaceStrategy = Schema.Union([ + Schema.Struct({ + type: Schema.Literal("root"), + branch: Schema.optional(TrimmedNonEmptyString), + }), + Schema.Struct({ + type: Schema.Literal("existing_worktree"), + worktreePath: TrimmedNonEmptyString, + branch: Schema.optional(TrimmedNonEmptyString), + }), + Schema.Struct({ + type: Schema.Literal("new_worktree"), + baseRef: TrimmedNonEmptyString, + branch: Schema.optional(TrimmedNonEmptyString), + startFromOrigin: Schema.optional(Schema.Boolean), + }), +]).annotate({ + description: + "Workspace for the new thread. Omit to reuse the caller's checkout in the current project, or the target project's root when projectId selects another project.", +}); +export type OrchestratorMcpThreadWorkspaceStrategy = + typeof OrchestratorMcpThreadWorkspaceStrategy.Type; + export const OrchestratorMcpCreateThreadRequest = Schema.Struct({ prompt: Schema.optional(OrchestratorMcpPrompt), title: Schema.optional(OrchestratorMcpTitle), target: Schema.optional(OrchestratorMcpTarget), + projectId: Schema.optional(ProjectId), + workspaceStrategy: Schema.optional(OrchestratorMcpThreadWorkspaceStrategy), runtimeMode: Schema.optional(OrchestratorMcpRuntimeMode), interactionMode: Schema.optional(OrchestratorMcpInteractionMode), }); @@ -231,6 +256,7 @@ export type OrchestratorMcpCreatedThreadStatus = typeof OrchestratorMcpCreatedTh export const OrchestratorMcpCreatedThread = Schema.Struct({ threadId: ThreadId, + projectId: Schema.optional(ProjectId), runId: Schema.NullOr(RunId), status: OrchestratorMcpCreatedThreadStatus, title: Schema.String, @@ -250,6 +276,8 @@ export const OrchestratorMcpThreadStartInput = Schema.Struct({ prompt: OrchestratorMcpPrompt, title: Schema.optional(OrchestratorMcpTitle), target: Schema.optional(OrchestratorMcpTarget), + projectId: Schema.optional(ProjectId), + workspaceStrategy: Schema.optional(OrchestratorMcpThreadWorkspaceStrategy), clientRequestId: Schema.optional(OrchestratorMcpClientRequestId), runtimeMode: Schema.optional(OrchestratorMcpRuntimeMode), interactionMode: Schema.optional(OrchestratorMcpInteractionMode), @@ -263,6 +291,7 @@ export const OrchestratorMcpThreadStatus = Schema.Union([ export type OrchestratorMcpThreadStatus = typeof OrchestratorMcpThreadStatus.Type; export const OrchestratorMcpThreadListInput = Schema.Struct({ + projectId: Schema.optional(ProjectId), statuses: Schema.optional( Schema.Array(OrchestratorMcpThreadStatus).check(Schema.isMaxLength(10)), ), @@ -303,6 +332,7 @@ export type OrchestratorMcpThreadListResult = typeof OrchestratorMcpThreadListRe export const OrchestratorMcpThreadReadInput = Schema.Struct({ threadId: ThreadId, + projectId: Schema.optional(ProjectId), view: Schema.optional(Schema.Literals(["messages", "activity"])), afterPosition: Schema.optional(NonNegativeInt), limit: Schema.optional(PositiveInt.check(Schema.isLessThanOrEqualTo(100))), @@ -378,6 +408,7 @@ export type OrchestratorMcpThreadReadResult = typeof OrchestratorMcpThreadReadRe export const OrchestratorMcpThreadSendInput = Schema.Struct({ threadId: ThreadId, + projectId: Schema.optional(ProjectId), message: OrchestratorMcpPrompt, mode: Schema.optional(Schema.Literals(["auto", "queue", "steer", "restart"])), clientRequestId: Schema.optional(OrchestratorMcpClientRequestId), @@ -395,6 +426,7 @@ export type OrchestratorMcpThreadSendResult = typeof OrchestratorMcpThreadSendRe export const OrchestratorMcpThreadWaitInput = Schema.Struct({ threadId: ThreadId, + projectId: Schema.optional(ProjectId), runId: Schema.optional(RunId), timeoutMs: Schema.optional(Schema.Number), }); @@ -410,6 +442,7 @@ export type OrchestratorMcpThreadWaitResult = typeof OrchestratorMcpThreadWaitRe export const OrchestratorMcpThreadInterruptInput = Schema.Struct({ threadId: ThreadId, + projectId: Schema.optional(ProjectId), runId: Schema.optional(RunId), reason: Schema.optional(Schema.String.check(Schema.isMaxLength(2_000))), clientRequestId: Schema.optional(OrchestratorMcpClientRequestId), @@ -461,6 +494,11 @@ export const OrchestratorMcpCapabilitiesResult = Schema.Struct({ incrementalThreadRead: Schema.Boolean, scheduledTasks: Schema.Boolean, maxBatchThreads: Schema.Number, + projectManagement: Schema.optional(Schema.Boolean), + projectTargeting: Schema.optional(Schema.Boolean), + threadLaunchWorkspaceStrategies: Schema.optional( + Schema.Array(Schema.Literals(["root", "existing_worktree", "new_worktree"])), + ), }), }); export type OrchestratorMcpCapabilitiesResult = typeof OrchestratorMcpCapabilitiesResult.Type; @@ -549,6 +587,7 @@ export class OrchestratorMcpFailure extends Schema.TaggedErrorClass Date: Sat, 29 Aug 2026 16:33:48 -0700 Subject: [PATCH 2/4] fix(mcp): keep workspace strategy schema descriptions concrete --- packages/contracts/src/orchestratorMcp.ts | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/packages/contracts/src/orchestratorMcp.ts b/packages/contracts/src/orchestratorMcp.ts index 7ceaf56795e5..569f9c4968fb 100644 --- a/packages/contracts/src/orchestratorMcp.ts +++ b/packages/contracts/src/orchestratorMcp.ts @@ -218,10 +218,7 @@ export const OrchestratorMcpThreadWorkspaceStrategy = Schema.Union([ branch: Schema.optional(TrimmedNonEmptyString), startFromOrigin: Schema.optional(Schema.Boolean), }), -]).annotate({ - description: - "Workspace for the new thread. Omit to reuse the caller's checkout in the current project, or the target project's root when projectId selects another project.", -}); +]); export type OrchestratorMcpThreadWorkspaceStrategy = typeof OrchestratorMcpThreadWorkspaceStrategy.Type; @@ -230,7 +227,10 @@ export const OrchestratorMcpCreateThreadRequest = Schema.Struct({ title: Schema.optional(OrchestratorMcpTitle), target: Schema.optional(OrchestratorMcpTarget), projectId: Schema.optional(ProjectId), - workspaceStrategy: Schema.optional(OrchestratorMcpThreadWorkspaceStrategy), + workspaceStrategy: Schema.optional(OrchestratorMcpThreadWorkspaceStrategy).annotate({ + description: + "Workspace for the new thread. Omit to reuse the caller's checkout in the current project, or the target project's root when projectId selects another project.", + }), runtimeMode: Schema.optional(OrchestratorMcpRuntimeMode), interactionMode: Schema.optional(OrchestratorMcpInteractionMode), }); @@ -277,7 +277,10 @@ export const OrchestratorMcpThreadStartInput = Schema.Struct({ title: Schema.optional(OrchestratorMcpTitle), target: Schema.optional(OrchestratorMcpTarget), projectId: Schema.optional(ProjectId), - workspaceStrategy: Schema.optional(OrchestratorMcpThreadWorkspaceStrategy), + workspaceStrategy: Schema.optional(OrchestratorMcpThreadWorkspaceStrategy).annotate({ + description: + "Workspace for the new thread. Omit to reuse the caller's checkout in the current project, or the target project's root when projectId selects another project.", + }), clientRequestId: Schema.optional(OrchestratorMcpClientRequestId), runtimeMode: Schema.optional(OrchestratorMcpRuntimeMode), interactionMode: Schema.optional(OrchestratorMcpInteractionMode), From 0a6656974cf2137d46c62c0b2882a233913e4d72 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 29 Aug 2026 20:12:03 -0700 Subject: [PATCH 3/4] fix(server): enforce fresh thread launch ceilings --- apps/server/src/mcp/OrchestratorMcpService.ts | 23 ++- ...OrchestratorMcpToolkit.integration.test.ts | 193 +++++++++++++++++- .../src/orchestration-v2/Orchestrator.ts | 109 +++++++++- .../orchestration-v2/ThreadLaunchService.ts | 6 + .../orchestrator-mcp-server.md | 4 + packages/contracts/src/orchestrationV2.ts | 9 + 6 files changed, 336 insertions(+), 8 deletions(-) diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 3d77f19111b2..a6a3a5755c93 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -425,6 +425,20 @@ function resolveInteractionMode( : Effect.succeed(resolved); } +function resolveLaunchRuntimeMode( + parentMode: RuntimeMode, + requested: OrchestratorMcpRuntimeMode | undefined, +): RuntimeMode { + return requested === undefined || requested === "inherit" ? parentMode : requested; +} + +function resolveLaunchInteractionMode( + parentMode: ProviderInteractionMode, + requested: OrchestratorMcpInteractionMode | undefined, +): ProviderInteractionMode { + return requested === undefined || requested === "inherit" ? parentMode : requested; +} + function stablePart(value: string): string { return encodeURIComponent(value); } @@ -1584,11 +1598,11 @@ const make = Effect.gen(function* () { target: request.target, providers, }); - const runtimeMode = yield* resolveRuntimeMode( + const runtimeMode = resolveLaunchRuntimeMode( parent.thread.runtimeMode, request.runtimeMode, ); - const interactionMode = yield* resolveInteractionMode( + const interactionMode = resolveLaunchInteractionMode( parent.thread.interactionMode, request.interactionMode, ); @@ -1624,6 +1638,11 @@ const make = Effect.gen(function* () { modelSelection: target.modelSelection, runtimeMode, interactionMode, + policyCeiling: { + callerThreadId: scope.threadId, + runtimeMode: parent.thread.runtimeMode, + interactionMode: parent.thread.interactionMode, + }, workspaceStrategy, ...(request.prompt === undefined ? {} diff --git a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts index 1786b9a8345d..0f12c0e2d8fa 100644 --- a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts @@ -35,6 +35,7 @@ import { import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as PubSub from "effect/PubSub"; @@ -488,6 +489,13 @@ describe("orchestrator MCP toolkit", () => { Ref.update(continuationOffers, (existing) => [...existing, request]), take: Effect.never, }); + const launchReceiptReadGates = new Map< + CommandId, + { + readonly entered: Deferred.Deferred; + readonly release: Deferred.Deferred; + } + >(); // Offers land after the finalize projection writes, so poll briefly // instead of asserting counts immediately. const waitForContinuationOffers = (count: number) => @@ -608,6 +616,24 @@ describe("orchestrator MCP toolkit", () => { }), ); const receiptLayer = CommandReceiptStore.layer.pipe(Layer.provide(databaseLayer)); + const launchReceiptLayer = Layer.effect( + CommandReceiptStore.CommandReceiptStoreV2, + Effect.map(CommandReceiptStore.CommandReceiptStoreV2, (receipts) => + CommandReceiptStore.CommandReceiptStoreV2.of({ + ...receipts, + getByCommandId: (commandId) => + Effect.gen(function* () { + const gate = launchReceiptReadGates.get(commandId); + if (gate !== undefined) { + yield* Deferred.succeed(gate.entered, undefined); + yield* Deferred.await(gate.release); + launchReceiptReadGates.delete(commandId); + } + return yield* receipts.getByCommandId(commandId); + }), + }), + ), + ).pipe(Layer.provide(receiptLayer)); const threadLaunchLayer = ThreadLaunch.layer.pipe( Layer.provide( Layer.mergeAll( @@ -620,7 +646,7 @@ describe("orchestrator MCP toolkit", () => { ServerSettings.layerTest({}), providerRegistryLayer, orchestrationLayer, - receiptLayer, + launchReceiptLayer, IdAllocator.layer, ), ), @@ -695,6 +721,18 @@ describe("orchestrator MCP toolkit", () => { Effect.provideService(McpInvocationContext.McpInvocationContext, invocation), Effect.provideService(McpSchema.McpServerClient, client), ); + const launchCommandId = (requestKey: string) => + CommandId.make( + `command:mcp:mcp-provider-session-parent:create-thread:${requestKey}:0`, + ); + const gateLaunchReceipt = Effect.fn("test.gateLaunchReceipt")(function* ( + commandId: CommandId, + ) { + const entered = yield* Deferred.make(); + const release = yield* Deferred.make(); + launchReceiptReadGates.set(commandId, { entered, release }); + return { entered, release } as const; + }); if (parentRun === undefined || parentRun.rootNodeId === null) { return yield* Effect.die(new Error("Parent run missing.")); @@ -1715,10 +1753,19 @@ describe("orchestrator MCP toolkit", () => { }, ]); - const crossProjectCall = yield* invoke("create_threads", { + const crossProjectInput = { clientRequestId: "create-cross-project-thread-1", - threads: [{ projectId: targetProjectId, title: "Cross-project ordinary thread" }], - }); + threads: [ + { + projectId: targetProjectId, + title: "Cross-project ordinary thread", + prompt: "Run in the explicitly selected project.", + runtimeMode: "full-access", + interactionMode: "default", + }, + ], + } as const; + const crossProjectCall = yield* invoke("create_threads", crossProjectInput); expect(crossProjectCall.isError).toBe(false); const crossProjectCreated = yield* decodeCreateThreadsResult( crossProjectCall.structuredContent, @@ -1729,6 +1776,12 @@ describe("orchestrator MCP toolkit", () => { (yield* orchestrator.getThreadProjection(crossProjectThread.threadId)).thread .projectId, ).toBe(targetProjectId); + const crossProjectProjection = yield* orchestrator.getThreadProjection( + crossProjectThread.threadId, + ); + expect(crossProjectProjection.messages.map((message) => message.text)).toEqual([ + "Run in the explicitly selected project.", + ]); const crossProjectItem = (yield* orchestrator.getThreadProjection( parentThreadId, )).visibleTurnItems @@ -1742,7 +1795,7 @@ describe("orchestrator MCP toolkit", () => { type: "thread_created", targetThreadId: crossProjectThread.threadId, targetProjectId, - targetRunId: null, + targetRunId: crossProjectThread.runId, }); const untrustedCrossProjectRecord = yield* orchestrator .dispatch({ @@ -1757,6 +1810,136 @@ describe("orchestrator MCP toolkit", () => { .pipe(Effect.flip); expect(untrustedCrossProjectRecord._tag).toBe("OrchestratorDispatchError"); + const preCreateKey = "cross-project-policy-before-create"; + const preCreateGate = yield* gateLaunchReceipt(launchCommandId(preCreateKey)); + const preCreateFiber = yield* Effect.forkChild( + invoke("create_threads", { + clientRequestId: preCreateKey, + threads: [ + { + projectId: targetProjectId, + prompt: "Do not accept after the caller narrows to plan mode.", + interactionMode: "default", + }, + ], + }), + { startImmediately: true }, + ); + yield* Deferred.await(preCreateGate.entered); + yield* orchestrator.dispatch({ + type: "thread.interaction-mode.set", + commandId: CommandId.make("command:mcp-parent:interaction-plan-before-create"), + threadId: parentThreadId, + interactionMode: "plan", + }); + expect( + (yield* orchestrator.getThreadProjection(parentThreadId)).thread.interactionMode, + ).toBe("plan"); + yield* Deferred.succeed(preCreateGate.release, undefined); + const preCreateCall = yield* Fiber.join(preCreateFiber); + expect(preCreateCall.structuredContent).toMatchObject({ + code: "orchestration_error", + }); + const preCreateThreadId = ThreadId.make( + `thread:mcp:mcp-provider-session-parent:${preCreateKey}:0`, + ); + expect( + Option.isNone( + yield* Effect.option(orchestrator.getThreadProjection(preCreateThreadId)), + ), + ).toBe(true); + yield* orchestrator.dispatch({ + type: "thread.interaction-mode.set", + commandId: CommandId.make("command:mcp-parent:interaction-default-after-create-race"), + threadId: parentThreadId, + interactionMode: "default", + }); + + const betweenCreateAndMessageKey = "cross-project-policy-before-message"; + const messageReceiptId = CommandId.make( + `${launchCommandId(betweenCreateAndMessageKey)}:initial-message`, + ); + const messageGate = yield* gateLaunchReceipt(messageReceiptId); + const betweenCreateAndMessageFiber = yield* Effect.forkChild( + invoke("create_threads", { + clientRequestId: betweenCreateAndMessageKey, + threads: [ + { + projectId: targetProjectId, + prompt: "Do not accept this message after the caller runtime narrows.", + runtimeMode: "full-access", + }, + ], + }), + { startImmediately: true }, + ); + yield* Deferred.await(messageGate.entered); + const betweenCreateAndMessageThreadId = ThreadId.make( + `thread:mcp:mcp-provider-session-parent:${betweenCreateAndMessageKey}:0`, + ); + expect( + (yield* orchestrator.getThreadProjection(betweenCreateAndMessageThreadId)).thread + .projectId, + ).toBe(targetProjectId); + yield* orchestrator.dispatch({ + type: "thread.runtime-mode.set", + commandId: CommandId.make("command:mcp-parent:runtime-narrow-before-message"), + threadId: parentThreadId, + runtimeMode: "approval-required", + }); + yield* Deferred.succeed(messageGate.release, undefined); + const betweenCreateAndMessageCall = yield* Fiber.join(betweenCreateAndMessageFiber); + expect(betweenCreateAndMessageCall.structuredContent).toMatchObject({ + code: "orchestration_error", + }); + const partialProjection = yield* orchestrator.getThreadProjection( + betweenCreateAndMessageThreadId, + ); + expect(partialProjection.messages).toEqual([]); + expect(partialProjection.runs).toEqual([]); + yield* orchestrator.dispatch({ + type: "thread.runtime-mode.set", + commandId: CommandId.make("command:mcp-parent:runtime-restore-after-message-race"), + threadId: parentThreadId, + runtimeMode: "full-access", + }); + + yield* orchestrator.dispatch({ + type: "thread.runtime-mode.set", + commandId: CommandId.make("command:mcp-parent:runtime-narrow-before-replay"), + threadId: parentThreadId, + runtimeMode: "approval-required", + }); + yield* orchestrator.dispatch({ + type: "thread.interaction-mode.set", + commandId: CommandId.make("command:mcp-parent:interaction-plan-before-replay"), + threadId: parentThreadId, + interactionMode: "plan", + }); + const replayedCrossProjectCall = yield* invoke("create_threads", crossProjectInput); + expect(replayedCrossProjectCall.isError).toBe(false); + const replayedCrossProject = yield* decodeCreateThreadsResult( + replayedCrossProjectCall.structuredContent, + ).pipe(Effect.orDie); + expect(replayedCrossProject.threads[0]?.threadId).toBe(crossProjectThread.threadId); + const replayedCrossProjectProjection = yield* orchestrator.getThreadProjection( + crossProjectThread.threadId, + ); + expect(replayedCrossProjectProjection.messages).toHaveLength(1); + expect(replayedCrossProjectProjection.runs).toHaveLength(1); + yield* orchestrator.dispatch({ + type: "thread.runtime-mode.set", + commandId: CommandId.make("command:mcp-parent:runtime-restore-after-replay"), + threadId: parentThreadId, + runtimeMode: "full-access", + }); + yield* orchestrator.dispatch({ + type: "thread.interaction-mode.set", + commandId: CommandId.make("command:mcp-parent:interaction-default-after-replay"), + threadId: parentThreadId, + interactionMode: "default", + }); + const repeatedCreateCall = yield* invoke("create_threads", createInput); const repeatedCreated = yield* decodeCreateThreadsResult( repeatedCreateCall.structuredContent, diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index a2e640236276..9fc9a896d11a 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -227,6 +227,23 @@ function nextRunOrdinal(projection: OrchestrationV2ThreadProjection): number { return projection.runs.length + 1; } +function runtimeModeRank(mode: OrchestrationV2AppThread["runtimeMode"]): number { + switch (mode) { + case "approval-required": + return 0; + case "auto-accept-edits": + return 1; + case "auto": + return 2; + case "full-access": + return 3; + } +} + +function interactionModeRank(mode: OrchestrationV2AppThread["interactionMode"]): number { + return mode === "plan" ? 0 : 1; +} + function commandThreadId(command: OrchestrationV2Command): ThreadId { switch (command.type) { case "thread.create": @@ -624,6 +641,58 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio return withId; }); + const enforcePolicyCeiling = Effect.fn("orchestrationV2.enforcePolicyCeiling")(function* (input: { + readonly command: Extract< + OrchestrationV2Command, + { readonly type: "thread.create" | "message.dispatch" } + >; + readonly runtimeMode: OrchestrationV2AppThread["runtimeMode"]; + readonly interactionMode: OrchestrationV2AppThread["interactionMode"]; + }) { + // Project selection is authorized by the caller-facing service before it + // constructs this command. This shared boundary owns only lifecycle and + // permission ceilings so an authorized cross-project launch remains valid. + const ceiling = input.command.policyCeiling; + if (ceiling === undefined) return; + const caller = yield* projectionStore.getThreadProjection(ceiling.callerThreadId).pipe( + Effect.mapError( + (cause) => + new OrchestratorProjectionError({ + threadId: ceiling.callerThreadId, + cause, + }), + ), + ); + if (caller.thread.deletedAt !== null || caller.thread.archivedAt !== null) { + return yield* new OrchestratorDispatchError({ + commandId: input.command.commandId, + commandType: input.command.type, + cause: `Caller thread ${ceiling.callerThreadId} is not active.`, + }); + } + if ( + runtimeModeRank(input.runtimeMode) > runtimeModeRank(ceiling.runtimeMode) || + runtimeModeRank(input.runtimeMode) > runtimeModeRank(caller.thread.runtimeMode) + ) { + return yield* new OrchestratorDispatchError({ + commandId: input.command.commandId, + commandType: input.command.type, + cause: `Target runtime mode ${input.runtimeMode} exceeds the caller ceiling.`, + }); + } + if ( + interactionModeRank(input.interactionMode) > interactionModeRank(ceiling.interactionMode) || + interactionModeRank(input.interactionMode) > + interactionModeRank(caller.thread.interactionMode) + ) { + return yield* new OrchestratorDispatchError({ + commandId: input.command.commandId, + commandType: input.command.type, + cause: `Target interaction mode ${input.interactionMode} exceeds the caller ceiling.`, + }); + } + }); + const getProjectionWithPendingEvents = ( threadId: ThreadId, events: Ref.Ref>, @@ -1334,6 +1403,12 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio "orchestration_v2.driver": command.modelSelection.instanceId, }); + yield* enforcePolicyCeiling({ + command, + runtimeMode: command.runtimeMode, + interactionMode: command.interactionMode, + }); + const now = yield* DateTime.now; const emitEvent = emit(events, command); const thread: OrchestrationV2AppThread = { @@ -2876,6 +2951,18 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ) => Effect.gen(function* () { let projection = yield* getProjectionWithPendingEvents(command.threadId, events); + if (projection.thread.deletedAt !== null || projection.thread.archivedAt !== null) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: `Thread ${command.threadId} is ${projection.thread.deletedAt !== null ? "deleted" : "archived"}.`, + }); + } + yield* enforcePolicyCeiling({ + command, + runtimeMode: projection.thread.runtimeMode, + interactionMode: projection.thread.interactionMode, + }); if (projection.thread.settledOverride !== null) { const now = yield* DateTime.now; const thread: OrchestrationV2AppThread = { @@ -7036,8 +7123,28 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio } satisfies OrchestratorV2DispatchResult; }); + const dispatchLockKeys = (command: OrchestrationV2Command): ReadonlyArray => { + const targetThreadId = commandThreadId(command); + const keys = + (command.type === "thread.create" || command.type === "message.dispatch") && + command.policyCeiling !== undefined + ? [targetThreadId, command.policyCeiling.callerThreadId] + : [targetThreadId]; + return [...new Set(keys)].toSorted((left, right) => (left < right ? -1 : left > right ? 1 : 0)); + }; + + const withDispatchLocks = ( + keys: ReadonlyArray, + effect: Effect.Effect, + ): Effect.Effect => { + const [key, ...remaining] = keys; + return key === undefined + ? effect + : threadDispatch.withLock(key, withDispatchLocks(remaining, effect)); + }; + const dispatchWithReceipt = (command: OrchestrationV2Command) => - threadDispatch.withLock(commandThreadId(command), dispatchWithReceiptEffect(command)); + withDispatchLocks(dispatchLockKeys(command), dispatchWithReceiptEffect(command)); const recordServerCreatedThread = (input: OrchestratorV2ServerCreatedThreadRecordInput) => threadDispatch.withLock( diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.ts index 7e0bbc0c3230..6b54b37b265d 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.ts @@ -5,6 +5,7 @@ import { type ModelSelection, type OrchestrationV2Actor, type OrchestrationV2CreationSource, + type OrchestrationV2PolicyCeiling, type OrchestrationV2ThreadProjection, type ProviderInteractionMode, ProjectId, @@ -65,6 +66,7 @@ export interface ThreadLaunchInput { readonly modelSelection: ModelSelection; readonly runtimeMode: RuntimeMode; readonly interactionMode: ProviderInteractionMode; + readonly policyCeiling?: OrchestrationV2PolicyCeiling; readonly workspaceStrategy: ThreadLaunchWorkspaceStrategy; readonly initialMessage?: ThreadLaunchInitialMessage; readonly createdBy: OrchestrationV2Actor; @@ -473,6 +475,9 @@ export const make = Effect.gen(function* () { modelSelection: input.modelSelection, runtimeMode: input.runtimeMode, interactionMode: input.interactionMode, + ...(input.policyCeiling === undefined + ? {} + : { policyCeiling: input.policyCeiling }), branch: initialBranch, worktreePath: initialWorktreePath, createdBy: input.createdBy, @@ -520,6 +525,7 @@ export const make = Effect.gen(function* () { attachments: input.initialMessage.attachments, ...(input.generateTitle === true ? { titleSeed: input.title } : {}), modelSelection: input.modelSelection, + ...(input.policyCeiling === undefined ? {} : { policyCeiling: input.policyCeiling }), dispatchMode: { type: "defer_start" }, createdBy: input.createdBy, creationSource: input.creationSource, diff --git a/docs/orchestration-v2/orchestrator-mcp-server.md b/docs/orchestration-v2/orchestrator-mcp-server.md index a4ca6aa49143..451e972eec86 100644 --- a/docs/orchestration-v2/orchestrator-mcp-server.md +++ b/docs/orchestration-v2/orchestrator-mcp-server.md @@ -402,6 +402,10 @@ falls back to a terminal-status message when no assistant text exists. mode. It may not escalate privileges. - A child interaction mode may stay equal to or narrow from `default` to `plan`. It may not escalate from `plan` to `default`. +- Ordinary thread launches capture those ceilings and recheck the calling + thread inside serialized thread creation and initial-message acceptance, so + a concurrent permission downgrade wins before new work is accepted. An + already accepted stable command remains replayable. - General thread management is limited to the calling thread's project. Send additionally enforces the same runtime and interaction privilege ceiling as child creation. diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 25f8ea0ca9f2..2f86925cfbc7 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -67,6 +67,13 @@ const OrchestrationV2CreationFields = { creationSource: OrchestrationV2CreationSource, } as const; +export const OrchestrationV2PolicyCeiling = Schema.Struct({ + callerThreadId: ThreadId, + runtimeMode: RuntimeMode, + interactionMode: ProviderInteractionMode, +}); +export type OrchestrationV2PolicyCeiling = typeof OrchestrationV2PolicyCeiling.Type; + export const OrchestrationV2NativeRefStrength = Schema.Literals(["strong", "weak", "none"]); export type OrchestrationV2NativeRefStrength = typeof OrchestrationV2NativeRefStrength.Type; @@ -1996,6 +2003,7 @@ export const OrchestrationV2Command = Schema.Union([ modelSelection: ModelSelection, runtimeMode: RuntimeMode, interactionMode: ProviderInteractionMode, + policyCeiling: Schema.optional(OrchestrationV2PolicyCeiling), branch: Schema.NullOr(TrimmedNonEmptyString), worktreePath: Schema.NullOr(TrimmedNonEmptyString), }), @@ -2133,6 +2141,7 @@ export const OrchestrationV2Command = Schema.Union([ /** Seed the temporary title and generate a durable replacement for the first message. */ titleSeed: Schema.optional(TrimmedNonEmptyString), modelSelection: Schema.optional(ModelSelection), + policyCeiling: Schema.optional(OrchestrationV2PolicyCeiling), sourcePlanRef: Schema.optional(Schema.Struct({ threadId: ThreadId, planId: PlanId })), delegatedCompletion: Schema.optional( Schema.Struct({ From d842f7c03cc45c293f04352aed5835309778b61f Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 29 Aug 2026 20:28:34 -0700 Subject: [PATCH 4/4] fix(mcp): preserve launched run identity --- .../OrchestratorMcpService.targeting.test.ts | 3 + apps/server/src/mcp/OrchestratorMcpService.ts | 11 ++- .../ThreadLaunchService.test.ts | 72 ++++++++++++++++++- .../orchestration-v2/ThreadLaunchService.ts | 3 + 4 files changed, 85 insertions(+), 4 deletions(-) diff --git a/apps/server/src/mcp/OrchestratorMcpService.targeting.test.ts b/apps/server/src/mcp/OrchestratorMcpService.targeting.test.ts index acb4f08da2e7..330a6f46af50 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.targeting.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.targeting.test.ts @@ -227,6 +227,8 @@ describe("OrchestratorMcpService project targeting", () => { }, }, resumed: false, + initialMessageRunId: + input.initialMessage === undefined ? null : (target.runs.at(-1)?.id ?? null), }), ), }); @@ -440,6 +442,7 @@ describe("OrchestratorMcpService project targeting", () => { }, }, resumed: false, + initialMessageRunId: null, }), ), }), diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index a6a3a5755c93..e6de94d297ac 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -1665,7 +1665,12 @@ const make = Effect.gen(function* () { ), ); const projection = launched.projection; - const run = projection.runs.at(-1); + const run = + launched.initialMessageRunId === null + ? undefined + : projection.runs.find( + (candidate) => candidate.id === launched.initialMessageRunId, + ); yield* threadManagement .recordServerCreatedThread({ targetProjectId: projectId, @@ -1681,7 +1686,7 @@ const make = Effect.gen(function* () { parentRunId: parentRun.id, parentNodeId, targetThreadId: threadId, - targetRunId: run?.id ?? null, + targetRunId: launched.initialMessageRunId, }, }) .pipe( @@ -1695,7 +1700,7 @@ const make = Effect.gen(function* () { return { threadId, projectId, - runId: run?.id ?? null, + runId: launched.initialMessageRunId, status: run?.status ?? "idle", title: projection.thread.title, createdBy: projection.thread.createdBy, diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts index de07f06c2e07..8bf2b8c6ef28 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts @@ -76,6 +76,9 @@ interface HarnessOptions { readonly generateBranchName?: TextGeneration.TextGeneration["Service"]["generateBranchName"]; readonly serverSettings?: Parameters[0]; readonly providers?: ReadonlyArray; + readonly mapCommandReceipts?: ( + service: CommandReceiptStore.CommandReceiptStoreV2["Service"], + ) => CommandReceiptStore.CommandReceiptStoreV2["Service"]; } function makeHarness(options: HarnessOptions = {}) { @@ -87,7 +90,14 @@ function makeHarness(options: HarnessOptions = {}) { { databaseLayer: database, runEffectWorker: false }, ); const threadManagement = ThreadManagement.layer.pipe(Layer.provide(orchestrator)); - const receipts = CommandReceiptStore.layer.pipe(Layer.provide(database)); + const baseReceipts = CommandReceiptStore.layer.pipe(Layer.provide(database)); + const receipts = + options.mapCommandReceipts === undefined + ? baseReceipts + : Layer.effect( + CommandReceiptStore.CommandReceiptStoreV2, + Effect.map(CommandReceiptStore.CommandReceiptStoreV2, options.mapCommandReceipts), + ).pipe(Layer.provide(baseReceipts)); const outbox = EffectOutbox.layer.pipe(Layer.provide(database)); const createWorktree = vi.fn( options.createWorktree ?? @@ -376,6 +386,66 @@ it.effect("provisions independent launches concurrently instead of behind a glob }), ); +it.effect("keeps the initial run identity when replay resumes after a later run", () => + Effect.gen(function* () { + const receiptLookupCompleted = yield* Deferred.make(); + const allowReceiptLookup = yield* Deferred.make(); + const messageCommandId = CommandId.make("command:launch:message-replay:initial-message"); + let gated = false; + const harness = makeHarness({ + mapCommandReceipts: (service) => ({ + ...service, + getByCommandId: (commandId) => + service.getByCommandId(commandId).pipe( + Effect.flatMap((receipt) => { + if (gated || commandId !== messageCommandId || Option.isSome(receipt)) { + return Effect.succeed(receipt); + } + gated = true; + return Deferred.succeed(receiptLookupCompleted, undefined).pipe( + Effect.andThen(Deferred.await(allowReceiptLookup)), + Effect.as(receipt), + ); + }), + ), + }), + }); + yield* Effect.gen(function* () { + const launches = yield* ThreadLaunch.ThreadLaunchService; + const threads = yield* ThreadManagement.ThreadManagementService; + const input = launchInput({ + command: "command:launch:message-replay", + thread: "thread:launch:message-replay", + message: "Use the durable message once", + }); + + const delayedReplay = yield* launches.launch(input).pipe(Effect.forkChild); + yield* Deferred.await(receiptLookupCompleted); + const accepted = yield* launches.launch(input); + const acceptedRunId = accepted.initialMessageRunId; + assert.isNotNull(acceptedRunId); + yield* threads.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("command:launch:message-replay:later-message"), + threadId: accepted.threadId, + messageId: MessageId.make("message:launch:message-replay:later-message"), + text: "A later queued message", + attachments: [], + modelSelection, + dispatchMode: { type: "queue_after_active" }, + createdBy: "user", + creationSource: "web", + }); + yield* Deferred.succeed(allowReceiptLookup, undefined); + const replayed = yield* Fiber.join(delayedReplay); + + assert.equal(replayed.initialMessageRunId, acceptedRunId); + assert.lengthOf(replayed.projection.runs, 2); + assert.notEqual(replayed.projection.runs.at(-1)?.id, replayed.initialMessageRunId); + }).pipe(Effect.provide(harness.layer)); + }), +); + it.effect("enqueues provider work only after setup has been initiated", () => Effect.gen(function* () { const setupEntered = yield* Deferred.make(); diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.ts index 6b54b37b265d..6144eea29ae1 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.ts @@ -77,6 +77,8 @@ export interface ThreadLaunchResult { readonly threadId: ThreadId; readonly projection: OrchestrationV2ThreadProjection; readonly resumed: boolean; + /** The durable run created for `initialMessage`, or null when no initial message was requested. */ + readonly initialMessageRunId: RunId | null; } export class ThreadLaunchError extends Schema.TaggedErrorClass()( @@ -577,6 +579,7 @@ export const make = Effect.gen(function* () { threadId, projection, resumed: Option.isSome(launchReceipt) || messageWasAlreadyAccepted, + initialMessageRunId: runId, }; }); },