# Files changed · PR \#8678 · pingdotgg/t3code [View on GitCafe](https://git.cafe/pingdotgg/t3code/pull/8678/diffs) Repository: [pingdotgg/t3code](https://git.cafe/pingdotgg/t3code) Visibility: public [feat\(mcp\): target thread launches by project](https://git.cafe/pingdotgg/t3code/pull/8678.md) Version: 34 Head commit: 99e60afa114984f01756807ccd61d190cec9ac22 Base commit: e3f2cdaae2c69d7d84ee05baabec32b1b26c4b10 Comparison base: e3f2cdaae2c69d7d84ee05baabec32b1b26c4b10 Showing up to 20 entries from this revision. If the pull request changes, reopen the first page before continuing. ## apps/server/src/git/GitWorkflowService.test.ts Status: modified; +78; −2 ```diff @@ -1,24 +1,69 @@ import { assert, describe, expect, it, vi } from "@effect/vitest"; +import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import { GitCommandError, VcsRepositoryDetectionError } from "@t3tools/contracts"; -import { VcsRepositoryDetectionError } from "@t3tools/contracts"; import * as GitManager from "./GitManager.ts"; import * as GitWorkflowService from "./GitWorkflowService.ts"; import * as GitVcsDriver from "../vcs/GitVcsDriver.ts"; +import * as VcsDriver from "../vcs/VcsDriver.ts"; import * as VcsDriverRegistry from "../vcs/VcsDriverRegistry.ts"; + +const unusedVcsDriver = VcsDriver.VcsDriver.of({ + capabilities: { + kind: "git", + supportsWorktrees: true, + supportsBookmarks: false, + supportsAtomicSnapshot: false, + supportsPushDefaultRemote: true, + ignoreClassifier: "native", + }, + execute: () => Effect.die("unused VCS driver execute"), + detectRepository: () => Effect.die("unused VCS repository detection"), + isInsideWorkTree: () => Effect.die("unused VCS worktree lookup"), + listWorkspaceFiles: () => Effect.die("unused VCS workspace listing"), + listRemotes: () => Effect.die("unused VCS remote listing"), + filterIgnoredPaths: () => Effect.die("unused VCS ignore classification"), + initRepository: () => Effect.die("unused VCS repository initialization"), +}); + +function gitHandle(cwd: string): VcsDriverRegistry.VcsDriverHandle { + return { + kind: "git", + repository: { + kind: "git", + rootPath: cwd, + metadataPath: `${cwd}/.git`, + freshness: { + source: "live-local", + observedAt: DateTime.makeUnsafe("2026-08-30T00:00:00.000Z"), + expiresAt: Option.none(), + }, + }, + driver: unusedVcsDriver, + }; +} function makeLayer(input: { readonly detect: VcsDriverRegistry.VcsDriverRegistry["Service"]["detect"]; + readonly resolve?: VcsDriverRegistry.VcsDriverRegistry["Service"]["resolve"]; + readonly execute?: GitVcsDriver.GitVcsDriver["Service"]["execute"]; }) { return GitWorkflowService.layer.pipe( Layer.provide( Layer.mock(VcsDriverRegistry.VcsDriverRegistry)({ detect: input.detect, + ...(input.resolve === undefined ? {} : { resolve: input.resolve }), }), ), - Layer.provide(Layer.mock(GitVcsDriver.GitVcsDriver)({})), + Layer.provide( + Layer.mock(GitVcsDriver.GitVcsDriver)( + input.execute === undefined ? {} : { execute: input.execute }, + ), + ), Layer.provide(Layer.mock(GitManager.GitManager)({})), ); } @@ -189,4 +234,35 @@ ), ); }); + + it.effect("distinguishes a detached HEAD from a failed symbolic-ref command", () => { + const detached = new GitCommandError({ + operation: "GitWorkflowService.currentBranch", + command: "git symbolic-ref", + cwd: "/detached", + exitCode: 1, + detail: "HEAD is detached.", + }); + const corrupt = new GitCommandError({ + operation: "GitWorkflowService.currentBranch", + command: "git symbolic-ref", + cwd: "/corrupt", + exitCode: 128, + detail: "HEAD cannot be read.", + }); + + return Effect.gen(function* () { + const workflow = yield* GitWorkflowService.GitWorkflowService; + assert.isNull(yield* workflow.currentBranch("/detached")); + assert.strictEqual(yield* workflow.currentBranch("/corrupt").pipe(Effect.flip), corrupt); + }).pipe( + Effect.provide( + makeLayer({ + detect: () => Effect.succeed(null), + resolve: ({ cwd }) => Effect.succeed(gitHandle(cwd)), + execute: (input) => Effect.fail(input.cwd === "/detached" ? detached : corrupt), + }), + ), + ); + }); }); ``` ## apps/server/src/git/GitWorkflowService.ts Status: modified; +19; −0 ```diff @@ -66,6 +66,7 @@ input: VcsCreateWorktreeInput, ) => Effect.Effect; readonly listLocalBranchNames: (cwd: string) => Effect.Effect; + readonly currentBranch: (cwd: string) => Effect.Effect; readonly fetchRemote: (input: { readonly cwd: string; readonly remoteName: string; @@ -319,6 +320,24 @@ ensureGitCommand("GitWorkflowService.listLocalBranchNames", cwd).pipe( Effect.andThen(git.listLocalBranchNames(cwd)), ), + currentBranch: (cwd) => + ensureGitCommand("GitWorkflowService.currentBranch", cwd).pipe( + Effect.andThen( + git.execute({ + operation: "GitWorkflowService.currentBranch", + cwd, + args: ["symbolic-ref", "--quiet", "--short", "HEAD"], + }), + ), + Effect.map((result) => { + const branch = result.stdout.trim(); + return branch.length > 0 ? branch : null; + }), + Effect.catchTags({ + GitCommandError: (error) => + error.exitCode === 1 ? Effect.succeed(null) : Effect.fail(error), + }), + ), fetchRemote: (input) => ensureGitCommand("GitWorkflowService.fetchRemote", input.cwd).pipe( Effect.andThen(git.fetchRemote(input)), ``` ## apps/server/src/mcp/OrchestratorMcpService.activity.test.ts Status: modified; +13; −0 ```diff @@ -11,11 +11,15 @@ 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 @@ // 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 @@ Layer.mock(ScheduledTaskService)({ list: () => Effect.succeed({ tasks: [] }), } satisfies Partial), + additionalDependencies, NodeCrypto.layer, ), ), @@ -184,6 +195,7 @@ Layer.mock(ScheduledTaskService)({ list: () => Effect.succeed({ tasks: [] }), } satisfies Partial), + additionalDependencies, NodeCrypto.layer, ), ), @@ -290,6 +302,7 @@ Layer.mock(ScheduledTaskService)({ list: () => Effect.succeed({ tasks: [] }), } satisfies Partial), + additionalDependencies, NodeCrypto.layer, ), ), ``` ## apps/server/src/mcp/OrchestratorMcpService.targeting.test.ts Status: added; +575; −0 ```diff @@ -1,0 +1,575 @@ +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!, + projectId: input.projectId, + title: input.title, + providerInstanceId: input.modelSelection.instanceId, + modelSelection: input.modelSelection, + runtimeMode: input.runtimeMode, + interactionMode: input.interactionMode, + }, + }, + resumed: false, + initialMessageRunId: + input.initialMessage === undefined ? null : (target.runs.at(-1)?.id ?? null), + }), + ), + }); + const dependencies = Layer.mergeAll( + commonDependencies, + launchLayer, + Layer.mock(ThreadManagementService)({ + getThreadProjection: () => Effect.succeed(parent), + withProjectCreationAdmission: (_input, effect) => effect(Option.none()), + 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", + expectedBranch: "caller-branch", + }, + modelSelection: parent.thread.modelSelection, + runtimeMode: parent.thread.runtimeMode, + }, + { + projectId: targetProjectId, + workspaceStrategy: { + type: "root", + branch: "target-main", + expectedBranch: "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 plainProjectId = ProjectId.make("project:mcp-target-plain"); + const plainProject = { + ...project, + id: plainProjectId, + title: "Plain project", + workspaceRoot: plainDirectory, + } 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 + : projectId === plainProjectId + ? plainProject + : 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, + initialMessageRunId: null, + }), + ), + }), + Layer.mock(ThreadManagementService)({ + getThreadProjection: () => Effect.succeed(parent), + withProjectCreationAdmission: (_input, effect) => effect(Option.none()), + 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", + expectedBranch: "feature", + }, + { + type: "existing_worktree", + worktreePath: canonicalSiblingWorktree, + branch: "feature", + expectedBranch: "feature", + }, + { type: "root", branch: "main", expectedBranch: "main" }, + { type: "root", branch: "main", expectedBranch: "main" }, + ], + ); + + for (const [clientRequestId, workspaceStrategy] of [ + [ + "existing-other-repository", + { type: "existing_worktree", worktreePath: otherRepository }, + ], + ["existing-plain-directory", { type: "existing_worktree", worktreePath: plainDirectory }], + ] as const) { + const error = yield* create(clientRequestId, workspaceStrategy).pipe(Effect.flip); + assert.equal(error.code, "invalid_request"); + } + const nonGitBranchError = yield* create( + "non-git-root-branch", + { type: "root", branch: "main" }, + plainProjectId, + ).pipe(Effect.flip); + assert.equal(nonGitBranchError.code, "invalid_request"); + assert.match(nonGitBranchError.message, /does not have a Git repository/u); + 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, +); ``` ## apps/server/src/mcp/OrchestratorMcpService.test.ts Status: modified; +58; −5 ```diff @@ -20,12 +20,22 @@ import * as Option from "effect/Option"; import * as Ref from "effect/Ref"; +import * as Orchestrator from "../orchestration-v2/Orchestrator.ts"; 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("preserves a structured parent admission failure", () => Effect.gen(function* () { @@ -135,6 +145,11 @@ const currentParent = yield* Ref.make(activeParent); const admissionEntered = yield* Deferred.make(); const allowAdmission = yield* Deferred.make(); + const callerRunFailure = new Orchestrator.OrchestratorCallerRunCeilingError({ + callerThreadId: parentThreadId, + callerRunId: parentRunId, + callerProviderInstanceId: providerInstanceId, + }); const dependencies = Layer.mergeAll( NodeServices.layer, Layer.mock(ThreadManagementService)({ @@ -149,6 +164,43 @@ ), dispatch: () => Effect.die("dispatch must not run after parent admission fails"), }), + Layer.mock(ThreadLaunch.ThreadLaunchService)({ + launch: (input) => + Effect.fail( + new ThreadLaunch.ThreadLaunchError({ + operation: "create-thread", + commandId: input.commandId, + projectId: input.projectId, + threadId: input.threadId, + cause: new Orchestrator.OrchestratorDispatchError({ + commandId: input.commandId, + commandType: "thread.create", + cause: callerRunFailure, + }), + }), + ), + }), + Layer.mock(ProjectService.ProjectService)({ + getById: () => + Effect.succeed( + Option.some({ + id: projectId, + title: "Admission project", + workspaceRoot: process.cwd(), + repositoryIdentity: null, + faviconPath: null, + defaultModelSelection: null, + defaultThreadEnvMode: "local", + scripts: [], + createdAt: "2026-08-30T12:00:00.000Z", + updatedAt: "2026-08-30T12:00:00.000Z", + deletedAt: null, + }), + ), + }), + Layer.mock(VcsDriverRegistry.VcsDriverRegistry)({ + detect: () => Effect.succeed(null), + }), Layer.mock(ProviderRegistry)({ getProviders: Effect.succeed([provider]) }), Layer.mock(ScheduledTaskService)({}), ); @@ -181,11 +233,8 @@ }).pipe(Effect.ensuring(Deferred.succeed(allowAdmission, undefined))); assert.equal(error.code, "parent_not_active"); - assert.equal( - error.message, + assert.match(error.message, /active/); + assert.isTrue((yield* Ref.get(projectionReads)) >= 1); - "Thread creation requires an active parent in the target project.", - ); - assert.equal(yield* Ref.get(projectionReads), 2); }), ); @@ -239,6 +288,7 @@ }), Layer.mock(ProviderRegistry)({ getProviders: Effect.succeed([]) }), Layer.mock(ScheduledTaskService)({}), + additionalDependencies, ); const scope: McpInvocationScope = { environmentId: EnvironmentId.make("environment:mcp-ack"), @@ -304,6 +354,7 @@ }), Layer.mock(ProviderRegistry)({ getProviders: Effect.succeed([]) }), Layer.mock(ScheduledTaskService)({}), + additionalDependencies, ); const scope: McpInvocationScope = { environmentId: EnvironmentId.make("environment:mcp-cancel"), @@ -366,6 +417,7 @@ }), Layer.mock(ProviderRegistry)({ getProviders: Effect.succeed([]) }), Layer.mock(ScheduledTaskService)({}), + additionalDependencies, ); const scope: McpInvocationScope = { environmentId: EnvironmentId.make("environment:mcp-cancel-failed"), @@ -435,6 +487,7 @@ }), Layer.mock(ProviderRegistry)({ getProviders: Effect.succeed([]) }), Layer.mock(ScheduledTaskService)({}), + additionalDependencies, ); const scope: McpInvocationScope = { environmentId: EnvironmentId.make("environment:mcp-cancel-dispose-failed"), ``` ## apps/server/src/mcp/OrchestratorMcpService.ts Status: modified; +481; −131 ```diff @@ -45,6 +45,8 @@ type ProviderInteractionMode, type ProviderOptionDescriptor, type ProviderOptionSelection, + type Project, + type ProjectId, type RuntimeMode, type ScheduledTask, type ScheduledTaskUpsertInput, @@ -56,12 +58,16 @@ 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 * as Orchestrator from "../orchestration-v2/Orchestrator.ts"; import { subagentResultForRun } from "../orchestration-v2/SubagentProjection.ts"; +import * as ThreadLaunch from "../orchestration-v2/ThreadLaunchService.ts"; import { isActiveRun, isTerminalRunStatus, @@ -71,7 +77,9 @@ 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; @@ -154,7 +162,12 @@ >()("t3/mcp/OrchestratorMcpService") {} const isThreadManagementError = Schema.is(ThreadManagementError); -const isOrchestratorMcpFailure = Schema.is(OrchestratorMcpFailure); +const isOrchestratorDispatchError = Schema.is(Orchestrator.OrchestratorDispatchError); +const isRuntimeModeCeilingError = Schema.is(Orchestrator.OrchestratorRuntimeModeCeilingError); +const isInteractionModeCeilingError = Schema.is( + Orchestrator.OrchestratorInteractionModeCeilingError, +); +const isCallerRunCeilingError = Schema.is(Orchestrator.OrchestratorCallerRunCeilingError); function failure(code: OrchestratorMcpFailure["code"], message: string): OrchestratorMcpFailure { return new OrchestratorMcpFailure({ code, message }); @@ -182,6 +195,32 @@ return error instanceof Error ? error.message : String(error); } +function threadLaunchPolicyFailure( + error: ThreadLaunch.ThreadLaunchError, +): OrchestratorMcpFailure | null { + if (!isOrchestratorDispatchError(error.cause)) return null; + const cause = error.cause.cause; + if (isRuntimeModeCeilingError(cause)) { + return failure( + "runtime_mode_escalation_denied", + `Child runtime mode ${cause.targetMode} exceeds the caller ceiling (captured ${cause.capturedCallerMode}; current ${cause.currentCallerMode}).`, + ); + } + if (isInteractionModeCeilingError(cause)) { + return failure( + "interaction_mode_escalation_denied", + `Child interaction mode ${cause.targetMode} exceeds the caller ceiling (captured ${cause.capturedCallerMode}; current ${cause.currentCallerMode}).`, + ); + } + if (isCallerRunCeilingError(cause)) { + return failure( + "parent_not_active", + "Thread creation requires an active run owned by this MCP provider session.", + ); + } + return null; +} + /** * Workspace strategy for a scheduled task created/updated over MCP: bound runs * post into the existing thread (the strategy is unused, keep root); unbound @@ -451,6 +490,20 @@ : 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); } @@ -716,7 +769,12 @@ 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") @@ -748,17 +806,236 @@ .getProjectThread({ projectId, threadId }) .pipe(Effect.mapError(threadManagementFailure)); + 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( - const loadScopedThread = (scope: McpInvocationScope, threadId: ThreadId) => + 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 && projectId === parent.thread.projectId - threadId === scope.threadId ? parent + : yield* loadProjectThread(projectId, threadId); + return { parent, projectId, target } as const; - : yield* loadProjectThread(parent.thread.projectId, threadId); - return { parent, 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)}`), + ), + ); + if (result.exitCode === 1) return null; + if (result.exitCode !== 0) { + return yield* failure( + "invalid_request", + `Unable to read the workspace branch because Git exited with status ${result.exitCode}.`, + ); + } + const branch = 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, + gitDetected: false, + } 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), + gitDetected: true, + } as const; + }, + ); + + 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); + const expectedBranch = explicitExisting?.branch ?? workspace.branch; + return { + type: "existing_worktree", + worktreePath: workspace.canonicalWorktreeRoot!, + ...(workspace.branch === null ? {} : { branch: workspace.branch }), + ...(workspace.gitDetected ? { expectedBranch } : {}), + }; + } + const workspace = yield* inspectLaunchWorkspace(project, project.workspaceRoot, false); + if (!workspace.gitDetected && requested?.type === "root" && requested.branch !== undefined) { + return yield* failure( + "invalid_request", + `Project ${project.id} does not have a Git repository for branch '${requested.branch}'.`, + ); + } + const expectedBranch = + requested?.type === "root" && requested.branch !== undefined + ? requested.branch + : workspace.branch; + return { + type: "root", + ...(workspace.branch === null ? {} : { branch: workspace.branch }), + ...(workspace.gitDetected ? { expectedBranch } : {}), + }; + }, + ); + const loadProviders = providerRegistry.getProviders; const resolveTarget = (input: { @@ -1178,6 +1455,13 @@ incrementalThreadRead: true, scheduledTasks: true, maxBatchThreads: 20, + projectManagement: true, + projectTargeting: true, + threadLaunchWorkspaceStrategies: [ + "root" as const, + "existing_worktree" as const, + "new_worktree" as const, + ], }, }; }), @@ -1404,140 +1688,192 @@ ); } const parentNodeId = parentRun.rootNodeId; - const providers = yield* loadProviders; const key = yield* requestKey(input.clientRequestId); const created = yield* Effect.forEach( input.threads, (request, index) => Effect.gen(function* () { - const target = yield* resolveTarget({ - parent, - target: request.target, - providers, - }); - const runtimeMode = yield* resolveRuntimeMode( - parent.thread.runtimeMode, - request.runtimeMode, - ); - const interactionMode = yield* resolveInteractionMode( - parent.thread.interactionMode, + const projectId = yield* resolveProjectId(parent, request.projectId); + const commandId = stableCommandId({ - request.interactionMode, - ); - const threadId = stableThreadId({ scope, requestKey: key, - index, - }); - const title = threadTitle({ - parentTitle: parent.thread.title, - prompt: request.prompt, - title: request.title, + operation: "create-thread", index, }); + const threadId = stableThreadId({ - const createCommandId = stableCommandId({ scope, requestKey: key, - operation: "create-thread", index, }); + const initialMessageId = stableMessageId({ scope, requestKey: key, index }); + const launchReceipt = yield* threadManagement + .withProjectCreationAdmission({ projectId, commandId }, (receipt) => - const createCommand = { + Effect.succeed(receipt), - type: "thread.create", - createdBy: "agent", - creationSource: "mcp", - commandId: createCommandId, - threadId, - projectId: parent.thread.projectId, - title, - modelSelection: target.modelSelection, - runtimeMode, - interactionMode, - branch: parent.thread.branch, - worktreePath: parent.thread.worktreePath, - } as const; - yield* threadManagement - .withProjectCreationAdmission( - { projectId: parent.thread.projectId, commandId: createCommandId }, - (receipt) => - Effect.gen(function* () { - if (Option.isNone(receipt)) { - const freshParent = yield* loadProjection(scope.threadId); - const freshParentRun = latestActiveRun(freshParent); - if ( - freshParent.thread.projectId !== parent.thread.projectId || - freshParent.thread.deletedAt !== null || - freshParent.thread.archivedAt !== null || - freshParentRun === undefined || - freshParentRun.id !== parentRun.id || - freshParentRun.rootNodeId === null || - freshParentRun.providerInstanceId !== scope.providerInstanceId - ) { - return yield* failure( - "parent_not_active", - "Thread creation requires an active parent in the target project.", - ); - } - } - return yield* threadManagement.dispatch(createCommand); - }), ) .pipe( - Effect.mapError((error) => - isOrchestratorMcpFailure(error) - ? error - : failure( - "orchestration_error", + Effect.mapError(() => + failure( + "orchestration_error", + `Unable to read the accepted launch for thread ${index + 1}.`, + ), - `Unable to create thread ${index + 1}: ${errorMessage(error)}`, - ), ), ); + if ( + Option.isSome(launchReceipt) && + (launchReceipt.value.commandType !== "thread.create" || + launchReceipt.value.threadId !== threadId) + ) { + return yield* failure( + "invalid_request", - if (request.prompt !== undefined) { - yield* threadManagement - .dispatch({ - type: "message.dispatch", - createdBy: "agent", - creationSource: "mcp", - commandId: stableCommandId({ - scope, - requestKey: key, - operation: "dispatch-thread", - index, - }), - threadId, - messageId: stableMessageId({ - scope, - requestKey: key, - index, - }), - text: request.prompt, + `clientRequestId conflicts with an existing command for thread ${index + 1}.`, + ); + } + const acceptedProjection = + Option.isSome(launchReceipt) && launchReceipt.value.status === "accepted" + ? Option.some(yield* loadProjection(threadId)) + : Option.none(); + if ( + Option.isSome(acceptedProjection) && + (acceptedProjection.value.thread.id !== threadId || + acceptedProjection.value.thread.projectId !== projectId) + ) { + return yield* failure( - attachments: [], - modelSelection: target.modelSelection, - dispatchMode: { type: "start_immediately" }, - }) - .pipe( - Effect.mapError((error) => + "invalid_request", + `clientRequestId was already accepted for a different project at thread ${index + 1}.`, + ); + } + const target = Option.isSome(acceptedProjection) + ? { modelSelection: acceptedProjection.value.thread.modelSelection } + : Option.isSome(launchReceipt) + ? { modelSelection: parent.thread.modelSelection } + : yield* resolveTarget({ + parent, + target: request.target, + providers: yield* loadProviders, + }); + const runtimeMode = Option.isSome(acceptedProjection) + ? acceptedProjection.value.thread.runtimeMode + : resolveLaunchRuntimeMode(parent.thread.runtimeMode, request.runtimeMode); + const interactionMode = Option.isSome(acceptedProjection) + ? acceptedProjection.value.thread.interactionMode + : resolveLaunchInteractionMode( + parent.thread.interactionMode, + request.interactionMode, + ); + const title = Option.isSome(acceptedProjection) + ? acceptedProjection.value.thread.title + : threadTitle({ + parentTitle: parent.thread.title, + prompt: request.prompt, + title: request.title, + index, + }); + const workspaceStrategy: ThreadLaunch.ThreadLaunchWorkspaceStrategy = Option.isSome( + acceptedProjection, + ) + ? acceptedProjection.value.thread.worktreePath !== null + ? { + type: "existing_worktree", + worktreePath: acceptedProjection.value.thread.worktreePath, + ...(acceptedProjection.value.thread.branch === null + ? {} + : { branch: acceptedProjection.value.thread.branch }), + } + : request.workspaceStrategy?.type === "new_worktree" + ? { + type: "worktree", + baseRef: request.workspaceStrategy.baseRef, + ...(acceptedProjection.value.thread.branch === null + ? request.workspaceStrategy.branch === undefined + ? {} + : { branch: request.workspaceStrategy.branch } + : { branch: acceptedProjection.value.thread.branch }), + ...(request.workspaceStrategy.startFromOrigin === undefined + ? {} + : { startFromOrigin: request.workspaceStrategy.startFromOrigin }), + } + : { + type: "root", + ...(acceptedProjection.value.thread.branch === null + ? {} + : { branch: acceptedProjection.value.thread.branch }), + } + : Option.isSome(launchReceipt) + ? { type: "root" } + : yield* resolveLaunchWorkspace( + parent, + yield* loadLaunchProject(projectId), + request.workspaceStrategy, + ); + const launched = yield* threadLaunch + .launch({ + commandId, + threadId, + projectId, + title, + modelSelection: target.modelSelection, + runtimeMode, + interactionMode, + policyCeiling: { + callerThreadId: scope.threadId, + callerRunId: parentRun.id, + callerProviderInstanceId: scope.providerInstanceId, + runtimeMode: parent.thread.runtimeMode, + interactionMode: parent.thread.interactionMode, + }, + workspaceStrategy, + ...(request.prompt === undefined + ? {} + : { + initialMessage: { + messageId: initialMessageId, + text: request.prompt, + attachments: [], + }, + }), + createdBy: "agent", + creationSource: "mcp", + }) + .pipe( + Effect.mapError( + (error) => + threadLaunchPolicyFailure(error) ?? failure( "orchestration_error", - `Unable to start thread ${index + 1}: ${errorMessage(error)}`, + `Unable to launch thread ${index + 1}: ${errorMessage(error)}`, ), + ), + ); + const projection = launched.projection; + if (projection.thread.id !== threadId || projection.thread.projectId !== projectId) { + return yield* failure( + "invalid_request", + `clientRequestId resolved to a different thread or project at thread ${index + 1}.`, - ), + ); - ); } + const run = + launched.initialMessageRunId === null + ? undefined + : projection.runs.find( + (candidate) => candidate.id === launched.initialMessageRunId, + ); - const projection = yield* loadProjection(threadId); - const run = projection.runs.at(-1); yield* threadManagement - .dispatch({ + .recordServerCreatedThread({ - 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, + targetProjectId: projection.thread.projectId, + command: { + type: "thread.created.record", + commandId: stableCommandId({ + scope, + requestKey: key, + operation: "record-created-thread", + index, + }), + parentThreadId: scope.threadId, + parentRunId: parentRun.id, + parentNodeId, + targetThreadId: projection.thread.id, + targetRunId: launched.initialMessageRunId, + }, }) .pipe( Effect.mapError((error) => @@ -1548,14 +1884,15 @@ ), ); return { - threadId, - runId: run?.id ?? null, + threadId: projection.thread.id, + projectId: projection.thread.projectId, + runId: launched.initialMessageRunId, status: run?.status ?? "idle", title: projection.thread.title, createdBy: projection.thread.createdBy, creationSource: projection.thread.creationSource, + providerInstanceId: projection.thread.modelSelection.instanceId, + model: projection.thread.modelSelection.model, - providerInstanceId: target.modelSelection.instanceId, - model: target.modelSelection.model, } satisfies OrchestratorMcpCreatedThread; }), { concurrency: 1 }, @@ -1566,9 +1903,10 @@ 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, - projectId: parent.thread.projectId, includeSubagents: input.includeSubagents !== false, }) .pipe( @@ -1593,7 +1931,7 @@ 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, @@ -1602,7 +1940,7 @@ }), 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; @@ -1658,7 +1996,11 @@ }), sendToThread: (scope, input) => Effect.gen(function* () { + const { parent, projectId, target } = yield* loadScopedThread( + scope, + input.threadId, - const { parent, target } = yield* loadScopedThread(scope, input.threadId); + input.projectId, + ); yield* resolveRuntimeMode(parent.thread.runtimeMode, target.thread.runtimeMode); yield* resolveInteractionMode(parent.thread.interactionMode, target.thread.interactionMode); @@ -1671,7 +2013,7 @@ }); const result = yield* threadManagement .sendToThread({ - projectId: parent.thread.projectId, + projectId, commandId: stableCommandId({ scope, requestKey: key, @@ -1705,10 +2047,10 @@ }), waitForThread: (scope, input) => Effect.gen(function* () { + const { projectId } = yield* loadScopedThread(scope, input.threadId, input.projectId); - const { parent } = yield* loadScopedThread(scope, input.threadId); const result = yield* threadManagement .waitForThread({ - projectId: parent.thread.projectId, + projectId, threadId: input.threadId, ...(input.runId === undefined ? {} : { runId: input.runId }), timeoutMs: Math.min( @@ -1726,11 +2068,11 @@ }), 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, - projectId: parent.thread.projectId, commandId: stableCommandId({ scope, requestKey: key, @@ -1769,5 +2111,13 @@ export const layer: Layer.Layer< OrchestratorMcpService, never, + | Crypto.Crypto + | FileSystem.FileSystem + | Path.Path + | ThreadManagementService + | ProviderRegistry + | ProjectService.ProjectService - Crypto.Crypto | ThreadManagementService | ProviderRegistry | ScheduledTaskService + | ScheduledTaskService + | ThreadLaunch.ThreadLaunchService + | VcsDriverRegistry.VcsDriverRegistry > = Layer.effect(OrchestratorMcpService, make); ``` ## apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts Status: modified; +877; −34 ```diff @@ -51,11 +51,14 @@ import { CodexProviderCapabilitiesV2 } from "../orchestration-v2/Adapters/CodexAdapterV2.ts"; import { CodexOrchestratorReplayHarness } from "../orchestration-v2/Adapters/CodexAdapterV2.testkit.ts"; import * as ClientCommandDispatch from "../orchestration-v2/ClientCommandDispatch.ts"; +import * as CommandReceiptStore from "../orchestration-v2/CommandReceiptStore.ts"; +import * as IdAllocator from "../orchestration-v2/IdAllocator.ts"; import { ThreadCommandExecutor, layer as threadCommandExecutorLayer, } from "../orchestration-v2/ThreadCommandExecutor.ts"; import { OrchestratorV2, type OrchestratorV2Shape } from "../orchestration-v2/Orchestrator.ts"; +import * as ThreadLaunch from "../orchestration-v2/ThreadLaunchService.ts"; import * as ThreadManagement from "../orchestration-v2/ThreadManagementService.ts"; import { type ProviderAdapterV2Event, @@ -77,11 +80,17 @@ decodeProviderReplayNdjson, materializeReplayTranscriptWorkspace, } from "../orchestration-v2/testkit/ReplayTranscriptNdjson.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 SourceControlRepositoryService from "../sourceControl/SourceControlRepositoryService.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"; import { delegatedTaskRun, hasPendingChildRuns } from "./OrchestratorMcpService.ts"; @@ -90,6 +99,12 @@ const projectId = ProjectId.make("project:mcp-orchestrator"); const deletionRaceProjectId = ProjectId.make("project:mcp-orchestrator-delete-race"); const deletionRaceParentThreadId = ThreadId.make("thread:mcp-orchestrator-delete-race-parent"); +const ownershipRaceParentThreadId = ThreadId.make("thread:mcp-orchestrator-ownership-race-parent"); +const lifecycleRaceParentThreadId = ThreadId.make("thread:mcp-orchestrator-lifecycle-race-parent"); +const missingRootReplayThreadId = ThreadId.make( + "thread:mcp:mcp-provider-session-parent:create-root-replay-after-removal:0", +); +const targetProjectId = ProjectId.make("project:mcp-orchestrator-target"); const codexInstanceId = ProviderInstanceId.make("codex"); const claudeInstanceId = ProviderInstanceId.make("claudeAgent"); const codexModel = "gpt-5.4"; @@ -507,13 +522,24 @@ const deletionRaceProjectState = yield* Ref.make(deletionRaceProject); const creationAdmissionEntered = yield* Deferred.make(); const allowCreationAdmission = yield* Deferred.make(); + const deletionRaceAdmissionReads = yield* Ref.make(0); const deletionLockAttemptSettled = yield* Deferred.make(); const deletionLockAcquired = yield* Deferred.make(); const allowDeletion = yield* Deferred.make(); const deletionRaceProviderStarted = yield* Deferred.make(); + const ownershipRaceProviderStarted = yield* Deferred.make(); + const lifecycleRaceProviderStarted = yield* Deferred.make(); const deletionRaceCommandId = CommandId.make( "command:mcp:mcp-provider-session-delete-race:create-thread:delete-create-race:0", ); + const missingWorkspaceReplayThreadId = ThreadId.make( + "thread:mcp:mcp-provider-session-parent:create-existing-worktree-replay-after-removal:0", + ); + const missingWorkspaceSetupEntered = yield* Deferred.make(); + const allowMissingWorkspaceSetup = yield* Deferred.make(); + const missingRootSetupEntered = yield* Deferred.make(); + const allowMissingRootSetup = yield* Deferred.make(); + const targetWorkspace = yield* checkpointWorkspace("orchestrator-mcp-target"); const capturedTurns = yield* Ref.make>([]); const parentTerminalGates = new Map>(); const deliveryTerminalGates = new Map>(); @@ -526,11 +552,17 @@ shouldComplete: (turn) => turn.threadId !== parentThreadId && turn.threadId !== deletionRaceParentThreadId && + turn.threadId !== ownershipRaceParentThreadId && + turn.threadId !== lifecycleRaceParentThreadId && turn.message.text !== cancellationPrompt, startedSignal: (turn) => turn.threadId === deletionRaceParentThreadId ? deletionRaceProviderStarted + : turn.threadId === ownershipRaceParentThreadId + ? ownershipRaceProviderStarted + : turn.threadId === lifecycleRaceParentThreadId + ? lifecycleRaceProviderStarted + : undefined, - : undefined, terminalGate: (turn) => turn.message.text.startsWith("Delegated task") || turn.message.text.startsWith("Delegated tasks") @@ -565,6 +597,16 @@ Ref.update(continuationOffers, (existing) => [...existing, request]), take: Effect.never, }); + const launchReceiptReadGates = new Map< + CommandId, + { + readonly entered: Deferred.Deferred; + readonly release: Deferred.Deferred; + } + >(); + const currentLaunchBranches = yield* Ref.make>( + new Map(), + ); // Offers land after the finalize projection writes, so poll briefly // instead of asserting counts immediately. const waitForContinuationOffers = (count: number) => @@ -587,6 +629,7 @@ expect(yield* Ref.get(continuationOffers)).toHaveLength(count); } }); + const databaseLayer = SqlitePersistenceMemory; const orchestratorLayer = makeOrchestratorV2ReplayLayerWithRegistry( { name: "orchestrator-mcp-toolkit", @@ -601,6 +644,7 @@ }, }, registryLayer, + { databaseLayer }, ).pipe(Layer.provide(continuationProbeLayer)); const threadManagementLayer = ThreadManagement.layer.pipe( Layer.provide(orchestratorLayer), @@ -613,12 +657,26 @@ ...actual, withProjectCreationAdmission: (input, effect) => actual.withProjectCreationAdmission(input, (receipt) => - input.commandId === deletionRaceCommandId && Option.isNone(receipt) - ? Deferred.succeed(creationAdmissionEntered, undefined).pipe( - Effect.andThen(Deferred.await(allowCreationAdmission)), - Effect.andThen(effect(receipt)), + Effect.gen(function* () { + const deletionRaceRead = + input.commandId === deletionRaceCommandId + ? yield* Ref.modify(deletionRaceAdmissionReads, (count) => [ - ) + count + 1, - : effect(receipt), + count + 1, + ]) + : 0; + if (deletionRaceRead === 2 && Option.isNone(receipt)) { + yield* Deferred.succeed(creationAdmissionEntered, undefined); + yield* Deferred.await(allowCreationAdmission); + } + const gate = launchReceiptReadGates.get(input.commandId); + if (gate !== undefined) { + yield* Deferred.succeed(gate.entered, undefined); + yield* Deferred.await(gate.release); + launchReceiptReadGates.delete(input.commandId); + } + return yield* effect(receipt); + }), ), }); }), @@ -688,7 +746,14 @@ create: () => Effect.die("unused"), bootstrap: () => Effect.die("unused"), update: () => Effect.die("unused"), + delete: ({ projectId: deletedProjectId }) => + deletedProjectId === deletionRaceProjectId + ? Ref.updateAndGet(deletionRaceProjectState, (project) => ({ + ...project, + deletedAt: "2026-08-30T00:01:00.000Z", + updatedAt: "2026-08-30T00:01:00.000Z", - delete: () => Effect.die("unused"), + })) + : Effect.die(`Unexpected project deletion: ${deletedProjectId}`), deleteDetailed: ({ commandId, projectId: deletedProjectId }) => Effect.gen(function* () { if (deletedProjectId !== deletionRaceProjectId) { @@ -738,16 +803,35 @@ yield* Deferred.succeed(deletionLockAttemptSettled, undefined); return yield* Fiber.join(deletion); }).pipe(Effect.orDie), + getById: (requestedProjectId, options) => { + if (requestedProjectId === deletionRaceProjectId) { + return Ref.get(deletionRaceProjectState).pipe( + Effect.map((project) => + project.deletedAt === null || options?.includeDeleted === true - getById: (requestedProjectId, options) => - requestedProjectId === deletionRaceProjectId - ? Ref.get(deletionRaceProjectState).pipe( - Effect.map((project) => - project.deletedAt === null || options?.includeDeleted === true - ? Option.some(project) - : Option.none(), + ? Option.some(project) + : Option.none(), + ), + ); - ), - ) - : Effect.succeed(Option.none()), + } + return 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(), + ); + }, getByWorkspaceRoot: () => Effect.die("unused"), snapshot: Effect.die("unused"), }); @@ -755,6 +839,62 @@ ).pipe( Layer.provide(Layer.merge(threadCommandExecutorLayer, gatedThreadManagementLayer)), ); + 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( + projectLayer, + Layer.mock(GitWorkflow.GitWorkflowService)({ + currentBranch: (workspaceRoot) => + Ref.get(currentLaunchBranches).pipe( + Effect.map((branches) => branches.get(workspaceRoot) ?? null), + ), + }), + Layer.mock(ProjectSetupScriptRunner.ProjectSetupScriptRunner)({ + runForThread: (input) => + input.threadId === missingWorkspaceReplayThreadId + ? Deferred.succeed(missingWorkspaceSetupEntered, undefined).pipe( + Effect.andThen(Deferred.await(allowMissingWorkspaceSetup)), + Effect.as({ status: "no-script" } as const), + ) + : input.threadId === missingRootReplayThreadId + ? Deferred.succeed(missingRootSetupEntered, undefined).pipe( + Effect.andThen(Deferred.await(allowMissingRootSetup)), + Effect.as({ status: "no-script" } as const), + ) + : Effect.succeed({ status: "no-script" }), + }), + Layer.mock(TextGeneration.TextGeneration)({}), + ServerSettings.layerTest({}), + providerRegistryLayer, + orchestrationLayer, + launchReceiptLayer, + IdAllocator.layer, + ), + ), + ); + const vcsProcessLayer = VcsProcess.layer.pipe(Layer.provide(NodeServices.layer)); + const vcsDriverRegistryLayer = VcsDriverRegistry.layer.pipe( + Layer.provide(vcsProcessLayer), + ); const toolkitRegistrationLayer = Layer.merge( McpHttpServer.OrchestratorToolkitRegistrationLive, McpHttpServer.ProjectToolkitRegistrationLive, @@ -769,10 +909,14 @@ Layer.mock(SourceControlRepositoryService.SourceControlRepositoryService)({}), ), Layer.provide(ServerSettings.layerTest({})), + Layer.provide(threadLaunchLayer), + Layer.provide(vcsDriverRegistryLayer), + Layer.provideMerge(vcsProcessLayer), - Layer.provide(NodeServices.layer), + Layer.provideMerge(NodeServices.layer), ); yield* Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; const orchestrator = yield* OrchestratorV2; const server = yield* McpServer.McpServer; yield* orchestrator.dispatch({ @@ -833,6 +977,18 @@ ); const invoke = (name: string, args: Record) => invokeAs(invocation, name, args); + const launchCommandId = ( + requestKey: string, + providerSessionId = "mcp-provider-session-parent", + ) => CommandId.make(`command:mcp:${providerSessionId}: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.")); @@ -1861,6 +2017,22 @@ ).toMatchObject({ state: "disposed" }); yield* expectOffersToStay(0); + const vcsProcess = yield* VcsProcess.VcsProcess; + const currentBranch = (yield* vcsProcess.run({ + operation: "OrchestratorMcpToolkit.integration.currentBranch", + command: "git", + cwd, + args: ["symbolic-ref", "--quiet", "--short", "HEAD"], + timeoutMs: 10_000, + })).stdout.trim(); + const canonicalCwd = yield* fileSystem.realPath(cwd); + yield* Ref.set( + currentLaunchBranches, + new Map([ + [cwd, currentBranch], + [canonicalCwd, currentBranch], + ]), + ); const createInput = { clientRequestId: "create-thread-batch-1", threads: [ @@ -1878,6 +2050,7 @@ }; const createCall = yield* invoke("create_threads", createInput); expect(createCall.isError).toBe(false); + expect(createCall.structuredContent).toHaveProperty("threads"); const created = yield* decodeCreateThreadsResult(createCall.structuredContent).pipe( Effect.orDie, ); @@ -2033,6 +2206,7 @@ expect( createdThreadItems.map((item) => ({ targetThreadId: item.targetThreadId, + targetProjectId: item.targetProjectId, targetRunId: item.targetRunId, title: item.title, providerInstanceId: item.targetProviderInstanceId, @@ -2041,6 +2215,7 @@ ).toEqual([ { targetThreadId: emptyThread.threadId, + targetProjectId: projectId, targetRunId: null, title: emptyThread.title, providerInstanceId: codexInstanceId, @@ -2048,6 +2223,7 @@ }, { targetThreadId: promptedThread.threadId, + targetProjectId: projectId, targetRunId: promptedThread.runId, title: promptedThread.title, providerInstanceId: claudeInstanceId, @@ -2055,6 +2231,420 @@ }, ]); + const targetBranch = (yield* vcsProcess.run({ + operation: "OrchestratorMcpToolkit.integration.currentBranch", + command: "git", + cwd: targetWorkspace, + args: ["symbolic-ref", "--quiet", "--short", "HEAD"], + timeoutMs: 10_000, + })).stdout.trim(); + const canonicalTargetWorkspace = yield* fileSystem.realPath(targetWorkspace); + const setTargetLaunchBranch = (branch: string) => + Ref.update(currentLaunchBranches, (branches) => { + const updated = new Map(branches); + updated.set(targetWorkspace, branch); + updated.set(canonicalTargetWorkspace, branch); + return updated; + }); + yield* setTargetLaunchBranch(targetBranch); + const crossProjectInput = { + clientRequestId: "create-cross-project-thread-1", + threads: [ + { + projectId: targetProjectId, + title: "Cross-project ordinary thread", + prompt: "Run in the explicitly selected project.", + runtimeMode: "full-access", + interactionMode: "default", + workspaceStrategy: { type: "root", branch: targetBranch }, + }, + ], + } as const; + const crossProjectCall = yield* invoke("create_threads", crossProjectInput); + expect(crossProjectCall.isError).toBe(false); + expect(crossProjectCall.structuredContent).toHaveProperty("threads"); + 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 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 + .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: crossProjectThread.runId, + }); + yield* vcsProcess.run({ + operation: "OrchestratorMcpToolkit.integration.changeBranch", + command: "git", + cwd: targetWorkspace, + args: ["switch", "-c", "mcp-accepted-replay-branch"], + timeoutMs: 10_000, + }); + yield* setTargetLaunchBranch("mcp-accepted-replay-branch"); + 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 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: "interaction_mode_escalation_denied", + message: + "Child interaction mode default exceeds the caller ceiling (captured default; current plan).", + }); + 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: "runtime_mode_escalation_denied", + message: + "Child runtime mode full-access exceeds the caller ceiling (captured full-access; current approval-required).", + }); + 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 laterCrossProjectDispatch = yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("command:mcp-cross-project:later-run"), + threadId: crossProjectThread.threadId, + messageId: MessageId.make("message:mcp-cross-project:later-run"), + text: "A later run must not replace the launch receipt result.", + attachments: [], + modelSelection: codexSelection, + dispatchMode: { type: "defer_start" }, + createdBy: "user", + creationSource: "web", + }); + const laterCrossProjectRun = laterCrossProjectDispatch.storedEvents.find( + (stored) => stored.event.type === "run.created", + ); + expect(laterCrossProjectRun?.event.type).toBe("run.created"); + const replayedCrossProjectCall = yield* invoke("create_threads", crossProjectInput); + expect(replayedCrossProjectCall.isError).toBe(false); + expect(replayedCrossProjectCall.structuredContent).toHaveProperty("threads"); + const replayedCrossProject = yield* decodeCreateThreadsResult( + replayedCrossProjectCall.structuredContent, + ).pipe(Effect.orDie); + expect(replayedCrossProject.threads[0]?.threadId).toBe(crossProjectThread.threadId); + expect(replayedCrossProject.threads[0]?.runId).toBe(crossProjectThread.runId); + const replayedCrossProjectProjection = yield* orchestrator.getThreadProjection( + crossProjectThread.threadId, + ); + expect(replayedCrossProjectProjection.messages).toHaveLength(2); + expect(replayedCrossProjectProjection.runs).toHaveLength(2); + expect( + (yield* orchestrator.getThreadProjection(parentThreadId)).visibleTurnItems + .map((row) => row.item) + .find( + (item) => + item.type === "thread_created" && + item.targetThreadId === crossProjectThread.threadId, + ), + ).toMatchObject({ targetRunId: crossProjectThread.runId }); + 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 freshMismatchedKey = "create-cross-project-thread-fresh-mismatch"; + const freshMismatchedCall = yield* invoke("create_threads", { + ...crossProjectInput, + clientRequestId: freshMismatchedKey, + }); + expect(freshMismatchedCall.structuredContent).toMatchObject({ + code: "orchestration_error", + }); + const freshMismatchedThreadId = ThreadId.make( + `thread:mcp:mcp-provider-session-parent:${freshMismatchedKey}:0`, + ); + expect( + Option.isNone( + yield* Effect.option(orchestrator.getThreadProjection(freshMismatchedThreadId)), + ), + ).toBe(true); + + const freshObservedBranchKey = "create-cross-project-thread-observed-mismatch"; + yield* Effect.gen(function* () { + yield* setTargetLaunchBranch("simulated-external-branch"); + const freshObservedBranchCall = yield* invoke("create_threads", { + clientRequestId: freshObservedBranchKey, + threads: [ + { + projectId: targetProjectId, + title: "Reject an observed branch race", + prompt: "Do not launch after the observed branch changes.", + workspaceStrategy: { type: "root" }, + }, + ], + }); + expect(freshObservedBranchCall.structuredContent).toMatchObject({ + code: "orchestration_error", + }); + const freshObservedBranchThreadId = ThreadId.make( + `thread:mcp:mcp-provider-session-parent:${freshObservedBranchKey}:0`, + ); + expect( + Option.isNone( + yield* Effect.option( + orchestrator.getThreadProjection(freshObservedBranchThreadId), + ), + ), + ).toBe(true); + }).pipe(Effect.ensuring(setTargetLaunchBranch("mcp-accepted-replay-branch"))); + + const missingWorkspaceReplayKey = "create-existing-worktree-replay-after-removal"; + const missingWorkspaceInput = { + clientRequestId: missingWorkspaceReplayKey, + threads: [ + { + projectId: targetProjectId, + title: "Existing worktree replay", + prompt: "Replay this accepted launch after its workspace disappears.", + target: { + providerInstanceId: codexInstanceId, + model: codexModel, + }, + workspaceStrategy: { + type: "existing_worktree", + worktreePath: targetWorkspace, + }, + }, + ], + } as const; + const originalMissingWorkspaceThread = yield* Effect.gen(function* () { + const missingWorkspaceCreateCall = yield* invoke( + "create_threads", + missingWorkspaceInput, + ); + expect(missingWorkspaceCreateCall.isError).toBe(false); + expect(missingWorkspaceCreateCall.structuredContent).toHaveProperty("threads"); + const missingWorkspaceCreated = yield* decodeCreateThreadsResult( + missingWorkspaceCreateCall.structuredContent, + ).pipe(Effect.orDie); + const original = missingWorkspaceCreated.threads[0]!; + expect(original.threadId).toBe(missingWorkspaceReplayThreadId); + yield* Deferred.await(missingWorkspaceSetupEntered); + const movedTargetWorkspace = `${targetWorkspace}-temporarily-missing`; + yield* Effect.gen(function* () { + yield* fileSystem.rename(targetWorkspace, movedTargetWorkspace); + const replayCall = yield* invoke("create_threads", missingWorkspaceInput); + expect(replayCall.structuredContent).toHaveProperty("threads"); + const replayed = yield* decodeCreateThreadsResult( + replayCall.structuredContent, + ).pipe(Effect.orDie); + expect(replayed.threads[0]).toEqual(original); + const replayedProjection = yield* orchestrator.getThreadProjection( + original.threadId, + ); + expect(replayedProjection.messages).toHaveLength(1); + expect(replayedProjection.thread.worktreePath).toBe(canonicalTargetWorkspace); + }).pipe( + Effect.ensuring( + fileSystem.rename(movedTargetWorkspace, targetWorkspace).pipe(Effect.orDie), + ), + ); + return original; + }).pipe(Effect.ensuring(Deferred.succeed(allowMissingWorkspaceSetup, undefined))); + + const missingRootReplayKey = "create-root-replay-after-removal"; + const missingRootInput = { + clientRequestId: missingRootReplayKey, + threads: [ + { + projectId: targetProjectId, + title: "Root replay", + prompt: "Replay this accepted root launch after its workspace disappears.", + workspaceStrategy: { type: "root" }, + }, + ], + } as const; + yield* Effect.gen(function* () { + const createCall = yield* invoke("create_threads", missingRootInput); + expect(createCall.isError).toBe(false); + const createdRootResult = yield* decodeCreateThreadsResult( + createCall.structuredContent, + ).pipe(Effect.orDie); + const createdRoot = createdRootResult.threads[0]!; + expect(createdRoot.threadId).toBe(missingRootReplayThreadId); + const acceptedRootProjection = yield* orchestrator.getThreadProjection( + createdRoot.threadId, + ); + const acceptedRootBranch = acceptedRootProjection.thread.branch; + expect(acceptedRootBranch).not.toBeNull(); + yield* Deferred.await(missingRootSetupEntered); + const movedTargetWorkspace = `${targetWorkspace}-root-temporarily-missing`; + yield* Effect.gen(function* () { + yield* fileSystem.rename(targetWorkspace, movedTargetWorkspace); + const replayCall = yield* invoke("create_threads", missingRootInput); + expect(replayCall.isError).toBe(false); + const replayedRootResult = yield* decodeCreateThreadsResult( + replayCall.structuredContent, + ).pipe(Effect.orDie); + const replayedRoot = replayedRootResult.threads[0]!; + expect(replayedRoot).toEqual(createdRoot); + const projection = yield* orchestrator.getThreadProjection(createdRoot.threadId); + expect(projection.messages).toHaveLength(1); + expect(projection.thread.worktreePath).toBeNull(); + expect(projection.thread.branch).toBe(acceptedRootBranch); + }).pipe( + Effect.ensuring( + fileSystem.rename(movedTargetWorkspace, targetWorkspace).pipe(Effect.orDie), + ), + ); + }).pipe(Effect.ensuring(Deferred.succeed(allowMissingRootSetup, undefined))); + + const conflictingReplayCall = yield* invoke("create_threads", { + clientRequestId: missingWorkspaceReplayKey, + threads: [ + { + projectId, + title: "Conflicting replay metadata", + prompt: "Do not replace the accepted launch identity.", + target: { + providerInstanceId: claudeInstanceId, + model: claudeModel, + }, + workspaceStrategy: { type: "root" }, + }, + ], + }); + expect(conflictingReplayCall.structuredContent).toMatchObject({ + code: "invalid_request", + message: expect.stringContaining("different project"), + }); + expect( + (yield* orchestrator.getThreadProjection(originalMissingWorkspaceThread.threadId)) + .thread.projectId, + ).toBe(targetProjectId); + const repeatedCreateCall = yield* invoke("create_threads", createInput); const repeatedCreated = yield* decodeCreateThreadsResult( repeatedCreateCall.structuredContent, @@ -2510,22 +3100,72 @@ } yield* Ref.set(continuationOffers, []); + const inactiveBeforeMessageKey = "create-before-parent-run-terminal"; + const inactiveBeforeMessageThreadId = ThreadId.make( + `thread:mcp:mcp-provider-session-parent:${inactiveBeforeMessageKey}:0`, + ); + const inactiveBeforeMessageGate = yield* gateLaunchReceipt( + CommandId.make(`${launchCommandId(inactiveBeforeMessageKey)}:initial-message`), + ); + const { parentStop, stoppedParent } = yield* Effect.gen(function* () { + const creation = yield* Effect.forkChild( + invoke("create_threads", { + clientRequestId: inactiveBeforeMessageKey, + threads: [ + { + title: "Created before its caller run terminalizes", + prompt: "Do not start this child turn after its caller run terminalizes.", + }, + ], + }), + { startImmediately: true }, + ); + yield* Deferred.await(inactiveBeforeMessageGate.entered); + expect( + (yield* orchestrator.getThreadProjection(inactiveBeforeMessageThreadId)).thread.id, + ).toBe(inactiveBeforeMessageThreadId); + + const sequenceBeforeInterrupt = + yield* orchestrator.getThreadEventSequence(parentThreadId); + const parentStop = yield* orchestrator.dispatch({ + type: "run.interrupt", + commandId: CommandId.make("command:mcp-parent:interrupt-wake"), + threadId: parentThreadId, + runId: parentRun.id, + reason: "Settle the parent before the legacy child terminalizes.", + }); + const terminalEvent = yield* orchestrator + .streamStoredEventsFrom({ + threadId: parentThreadId, + afterSequence: sequenceBeforeInterrupt, + }) + .pipe( + Stream.filter( + (stored) => + stored.event.type === "run.updated" && + stored.event.payload.id === parentRun.id && + stored.event.payload.status === "interrupted", + ), + Stream.runHead, + ); + expect(Option.isSome(terminalEvent)).toBe(true); + const stoppedParent = yield* orchestrator.getThreadProjection(parentThreadId); + yield* Deferred.succeed(inactiveBeforeMessageGate.release, undefined); + const creationCall = yield* Fiber.join(creation); + expect(creationCall.structuredContent).toMatchObject({ + code: "parent_not_active", + message: + "Thread creation requires an active run owned by this MCP provider session.", + }); + const partialChild = yield* orchestrator.getThreadProjection( + inactiveBeforeMessageThreadId, + ); + expect(partialChild.messages).toEqual([]); + expect(partialChild.runs).toEqual([]); + return { parentStop, stoppedParent }; + }).pipe( - const parentStop = yield* orchestrator.dispatch({ - type: "run.interrupt", - commandId: CommandId.make("command:mcp-parent:interrupt-wake"), + Effect.ensuring(Deferred.succeed(inactiveBeforeMessageGate.release, undefined)), - threadId: parentThreadId, - runId: parentRun.id, - reason: "Settle the parent before the legacy child terminalizes.", - }); - yield* waitForProjection(orchestrator, parentThreadId, (projection) => - projection.runs.every( - (run) => - run.status !== "preparing" && - run.status !== "starting" && - run.status !== "running", - ), ); - const stoppedParent = yield* orchestrator.getThreadProjection(parentThreadId); expect(stoppedParent.runs.some((run) => run.id === parentRun.id)).toBe(true); expect( parentStop.storedEvents.some( @@ -3032,6 +3672,206 @@ type: "thread.create", createdBy: "user", creationSource: "web", + commandId: CommandId.make("command:mcp-ownership-race-parent:create"), + threadId: ownershipRaceParentThreadId, + projectId, + title: "MCP ownership race parent", + modelSelection: codexSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: cwd, + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("command:mcp-ownership-race-parent:start"), + threadId: ownershipRaceParentThreadId, + messageId: MessageId.make("message:mcp-ownership-race-parent:start"), + text: "Stay active until child launch admission is checked.", + attachments: [], + modelSelection: codexSelection, + dispatchMode: { type: "start_immediately" }, + }); + const startedOwnershipRaceTurn = yield* Deferred.await(ownershipRaceProviderStarted); + expect(startedOwnershipRaceTurn.threadId).toBe(ownershipRaceParentThreadId); + const ownershipRaceInvocation: McpInvocationContext.McpInvocationScope = { + ...invocation, + threadId: ownershipRaceParentThreadId, + providerSessionId: "mcp-provider-session-ownership-race", + }; + const invokeOwnershipRace = (name: string, args: Record) => + server + .callTool({ name, arguments: args }) + .pipe( + Effect.provideService( + McpInvocationContext.McpInvocationContext, + ownershipRaceInvocation, + ), + Effect.provideService(McpSchema.McpServerClient, client), + ); + const ownershipRaceParent = yield* orchestrator.getThreadProjection( + ownershipRaceParentThreadId, + ); + const ownershipRaceParentRun = ownershipRaceParent.runs.find( + (run) => run.status === "running", + ); + if (ownershipRaceParentRun === undefined) { + return yield* Effect.die(new Error("Ownership-race parent run did not start.")); + } + const inactiveParentKey = "create-after-parent-run-terminal"; + const inactiveParentGate = yield* gateLaunchReceipt( + launchCommandId(inactiveParentKey, ownershipRaceInvocation.providerSessionId), + ); + yield* Effect.gen(function* () { + const creation = yield* Effect.forkChild( + invokeOwnershipRace("create_threads", { + clientRequestId: inactiveParentKey, + threads: [ + { + projectId: targetProjectId, + title: "Must not outlive its caller run", + prompt: "Do not launch after the caller run terminalizes.", + }, + ], + }), + { startImmediately: true }, + ); + yield* Deferred.await(inactiveParentGate.entered); + const sequenceBeforeInterrupt = yield* orchestrator.getThreadEventSequence( + ownershipRaceParentThreadId, + ); + yield* orchestrator.dispatch({ + type: "run.interrupt", + commandId: CommandId.make( + "command:mcp-ownership-race-parent:terminal-before-child-launch", + ), + threadId: ownershipRaceParentThreadId, + runId: ownershipRaceParentRun.id, + reason: "Prove child launch revalidates its caller run.", + }); + const terminalEvent = yield* orchestrator + .streamStoredEventsFrom({ + threadId: ownershipRaceParentThreadId, + afterSequence: sequenceBeforeInterrupt, + }) + .pipe( + Stream.filter( + (stored) => + stored.event.type === "run.updated" && + stored.event.payload.id === ownershipRaceParentRun.id && + stored.event.payload.status === "interrupted", + ), + Stream.runHead, + ); + expect(Option.isSome(terminalEvent)).toBe(true); + + yield* Deferred.succeed(inactiveParentGate.release, undefined); + const creationCall = yield* Fiber.join(creation); + expect(creationCall.structuredContent).toMatchObject({ + code: "parent_not_active", + message: + "Thread creation requires an active run owned by this MCP provider session.", + }); + const childThreadId = ThreadId.make( + `thread:mcp:${ownershipRaceInvocation.providerSessionId}:${inactiveParentKey}:0`, + ); + expect( + Option.isNone( + yield* Effect.option(orchestrator.getThreadProjection(childThreadId)), + ), + ).toBe(true); + }).pipe(Effect.ensuring(Deferred.succeed(inactiveParentGate.release, undefined))); + + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("command:mcp-lifecycle-race-parent:create"), + threadId: lifecycleRaceParentThreadId, + projectId, + title: "MCP lifecycle race parent", + modelSelection: codexSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: cwd, + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("command:mcp-lifecycle-race-parent:start"), + threadId: lifecycleRaceParentThreadId, + messageId: MessageId.make("message:mcp-lifecycle-race-parent:start"), + text: "Stay active until the caller lifecycle is rechecked.", + attachments: [], + modelSelection: codexSelection, + dispatchMode: { type: "start_immediately" }, + }); + const startedLifecycleRaceTurn = yield* Deferred.await(lifecycleRaceProviderStarted); + expect(startedLifecycleRaceTurn.threadId).toBe(lifecycleRaceParentThreadId); + const lifecycleRaceInvocation: McpInvocationContext.McpInvocationScope = { + ...invocation, + threadId: lifecycleRaceParentThreadId, + providerSessionId: "mcp-provider-session-lifecycle-race", + }; + const invokeLifecycleRace = (name: string, args: Record) => + server + .callTool({ name, arguments: args }) + .pipe( + Effect.provideService( + McpInvocationContext.McpInvocationContext, + lifecycleRaceInvocation, + ), + Effect.provideService(McpSchema.McpServerClient, client), + ); + const archivedParentKey = "create-after-parent-archived"; + const archivedParentGate = yield* gateLaunchReceipt( + launchCommandId(archivedParentKey, lifecycleRaceInvocation.providerSessionId), + ); + yield* Effect.gen(function* () { + const creation = yield* Effect.forkChild( + invokeLifecycleRace("create_threads", { + clientRequestId: archivedParentKey, + threads: [ + { + projectId: targetProjectId, + title: "Must not outlive its archived caller", + prompt: "Do not launch after the caller is archived.", + }, + ], + }), + { startImmediately: true }, + ); + yield* Deferred.await(archivedParentGate.entered); + yield* orchestrator.dispatch({ + type: "thread.archive", + commandId: CommandId.make("command:mcp-lifecycle-race-parent:archive"), + threadId: lifecycleRaceParentThreadId, + }); + yield* Deferred.succeed(archivedParentGate.release, undefined); + const creationCall = yield* Fiber.join(creation); + expect(creationCall.structuredContent).toMatchObject({ + code: "parent_not_active", + message: + "Thread creation requires an active run owned by this MCP provider session.", + }); + const childThreadId = ThreadId.make( + `thread:mcp:${lifecycleRaceInvocation.providerSessionId}:${archivedParentKey}:0`, + ); + expect( + Option.isNone( + yield* Effect.option(orchestrator.getThreadProjection(childThreadId)), + ), + ).toBe(true); + }).pipe(Effect.ensuring(Deferred.succeed(archivedParentGate.release, undefined))); + + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", commandId: CommandId.make("command:mcp-delete-race-parent:create"), threadId: deletionRaceParentThreadId, projectId: deletionRaceProjectId, @@ -3216,6 +4056,9 @@ Layer.provideMerge(orchestrationLayer), Layer.provide(providerRegistryLayer), Layer.provide(unusedScheduledTaskStubLayer), + Layer.provide(Layer.mock(ProjectService.ProjectService)({})), + Layer.provide(Layer.mock(ThreadLaunch.ThreadLaunchService)({})), + Layer.provide(Layer.mock(VcsDriverRegistry.VcsDriverRegistry)({})), Layer.provide(NodeServices.layer), ); ``` ## apps/server/src/mcp/toolkits/orchestrator/handlers.ts Status: modified; +4; −0 ```diff @@ -71,6 +71,10 @@ 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 ? {} ``` ## apps/server/src/mcp/toolkits/orchestrator/tools.ts Status: modified; +8; −8 ```diff @@ -43,7 +43,7 @@ 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", @@ -145,7 +145,7 @@ 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, @@ -158,7 +158,7 @@ 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, @@ -171,7 +171,7 @@ 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, @@ -185,7 +185,7 @@ 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, @@ -212,7 +212,7 @@ 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, @@ -225,7 +225,7 @@ 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, @@ -239,7 +239,7 @@ 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, ``` ## apps/server/src/mcp/toolkits/worktree/registration.test.ts Status: modified; +4; −0 ```diff @@ -10,6 +10,7 @@ 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 * 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 @@ Layer.mock(ProjectSetupScriptRunner.ProjectSetupScriptRunner)({}), Layer.mock(SourceControlRepositoryService.SourceControlRepositoryService)({}), Layer.mock(VcsStatusBroadcaster)({}), + Layer.mock(VcsDriverRegistry.VcsDriverRegistry)({ detect: () => Effect.succeed(null) }), ); const ToolsListPayload = Schema.fromJsonString( ``` ## apps/server/src/orchestration-v2/Orchestrator.ts Status: modified; +245; −5 ```diff @@ -24,7 +24,10 @@ type OrchestrationV2Subagent, type OrchestrationV2ThreadProjection, type OrchestrationV2TurnItem, + ProviderInteractionMode, + type ProjectId, ProviderInstanceId, + RuntimeMode, type ProviderSessionId, RunId, ThreadId, @@ -85,6 +88,47 @@ } } +export class OrchestratorRuntimeModeCeilingError extends Schema.TaggedErrorClass()( + "OrchestratorRuntimeModeCeilingError", + { + callerThreadId: ThreadId, + targetMode: RuntimeMode, + capturedCallerMode: RuntimeMode, + currentCallerMode: RuntimeMode, + }, +) { + override get message(): string { + return `Target runtime mode ${this.targetMode} exceeds the caller runtime-mode ceiling.`; + } +} + +export class OrchestratorInteractionModeCeilingError extends Schema.TaggedErrorClass()( + "OrchestratorInteractionModeCeilingError", + { + callerThreadId: ThreadId, + targetMode: ProviderInteractionMode, + capturedCallerMode: ProviderInteractionMode, + currentCallerMode: ProviderInteractionMode, + }, +) { + override get message(): string { + return `Target interaction mode ${this.targetMode} exceeds the caller interaction-mode ceiling.`; + } +} + +export class OrchestratorCallerRunCeilingError extends Schema.TaggedErrorClass()( + "OrchestratorCallerRunCeilingError", + { + callerThreadId: ThreadId, + callerRunId: RunId, + callerProviderInstanceId: ProviderInstanceId, + }, +) { + override get message(): string { + return `Caller run ${this.callerRunId} is no longer active for provider ${this.callerProviderInstanceId}.`; + } +} + export class OrchestratorProjectionError extends Schema.TaggedErrorClass()( "OrchestratorProjectionError", { @@ -176,12 +220,26 @@ 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; readonly getCommandReceipt: CommandReceiptStoreV2["Service"]["getByCommandId"]; + /** Records a cross-project MCP launch through a server-owned authorization boundary. */ + readonly recordServerCreatedThread: ( + input: OrchestratorV2ServerCreatedThreadRecordInput, + ) => Effect.Effect; readonly getThreadProjection: ( threadId: ThreadId, ) => Effect.Effect; @@ -242,6 +300,23 @@ ); } +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": @@ -640,6 +715,95 @@ 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) { + const cause = + ceiling.callerRunId !== undefined && ceiling.callerProviderInstanceId !== undefined + ? new OrchestratorCallerRunCeilingError({ + callerThreadId: ceiling.callerThreadId, + callerRunId: ceiling.callerRunId, + callerProviderInstanceId: ceiling.callerProviderInstanceId, + }) + : `Caller thread ${ceiling.callerThreadId} is not active.`; + return yield* new OrchestratorDispatchError({ + commandId: input.command.commandId, + commandType: input.command.type, + cause, + }); + } + if (ceiling.callerRunId !== undefined && ceiling.callerProviderInstanceId !== undefined) { + const callerRun = caller.runs.find((candidate) => candidate.id === ceiling.callerRunId); + if ( + callerRun === undefined || + !isBlockingRun(callerRun) || + callerRun.rootNodeId === null || + callerRun.providerInstanceId !== ceiling.callerProviderInstanceId + ) { + return yield* new OrchestratorDispatchError({ + commandId: input.command.commandId, + commandType: input.command.type, + cause: new OrchestratorCallerRunCeilingError({ + callerThreadId: ceiling.callerThreadId, + callerRunId: ceiling.callerRunId, + callerProviderInstanceId: ceiling.callerProviderInstanceId, + }), + }); + } + } + 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: new OrchestratorRuntimeModeCeilingError({ + callerThreadId: ceiling.callerThreadId, + targetMode: input.runtimeMode, + capturedCallerMode: ceiling.runtimeMode, + currentCallerMode: caller.thread.runtimeMode, + }), + }); + } + 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: new OrchestratorInteractionModeCeilingError({ + callerThreadId: ceiling.callerThreadId, + targetMode: input.interactionMode, + capturedCallerMode: ceiling.interactionMode, + currentCallerMode: caller.thread.interactionMode, + }), + }); + } + }); + const getProjectionWithPendingEvents = ( threadId: ThreadId, events: Ref.Ref>, @@ -1350,6 +1514,12 @@ "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 = { @@ -2896,6 +3066,18 @@ } } + 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 = { @@ -4922,8 +5104,9 @@ 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) @@ -4965,7 +5148,10 @@ cause: `Parent node ${command.parentNodeId} is not the root of run ${command.parentRunId}.`, }); } + if ( + authorizedTargetProjectId === undefined && + parentProjection.thread.projectId !== targetProjection.thread.projectId - if (parentProjection.thread.projectId !== targetProjection.thread.projectId) { + ) { return yield* new OrchestratorDispatchError({ commandId: command.commandId, commandType: command.type, @@ -4973,6 +5159,16 @@ }); } 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) ) { @@ -5002,6 +5198,7 @@ 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, @@ -6899,6 +7096,7 @@ const dispatchOnce = Effect.fn("orchestrationV2.dispatch.once")(function* ( command: OrchestrationV2Command, + options?: { readonly authorizedCreatedThreadProjectId?: ProjectId }, ): Effect.fn.Return< { readonly events: ReadonlyArray; @@ -7048,7 +7246,11 @@ yield* dispatchDelegatedTaskCompletionDeliveryResolution(command, events); break; case "thread.created.record": + yield* dispatchCreatedThreadRecord( + command, - yield* dispatchCreatedThreadRecord(command, events); + events, + options?.authorizedCreatedThreadProjectId, + ); break; default: return yield* dispatchUnsupported(command); @@ -7062,6 +7264,7 @@ 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, @@ -7119,7 +7322,7 @@ } 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) @@ -7198,8 +7401,36 @@ } 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) => + withDispatchLocks(dispatchLockKeys(command), dispatchWithReceiptEffect(command)); + + const recordServerCreatedThread = (input: OrchestratorV2ServerCreatedThreadRecordInput) => + threadDispatch.withLock( - threadDispatch.withLock(commandThreadId(command), dispatchWithReceiptEffect(command)); + commandThreadId(input.command), + dispatchWithReceiptEffect(input.command, { + authorizedCreatedThreadProjectId: input.targetProjectId, + }), + ); const handleTerminalRun = (stored: OrchestrationV2StoredEvent) => Effect.gen(function* () { @@ -7351,6 +7582,7 @@ resumeQueuedRuns, dispatch: dispatchWithReceipt, getCommandReceipt: commandReceipts.getByCommandId, + recordServerCreatedThread, getThreadProjection: (threadId) => projectionStore .getThreadProjection(threadId) @@ -7466,6 +7698,14 @@ 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({ ``` ## apps/server/src/orchestration-v2/ThreadLaunchService.test.ts Status: modified; +174; −1 ```diff @@ -73,12 +73,16 @@ readonly getProjectById?: ProjectService.ProjectService["Service"]["getById"]; readonly createWorktree?: GitWorkflow.GitWorkflowService["Service"]["createWorktree"]; readonly fetchRemote?: GitWorkflow.GitWorkflowService["Service"]["fetchRemote"]; + readonly currentBranch?: GitWorkflow.GitWorkflowService["Service"]["currentBranch"]; readonly renameBranch?: GitWorkflow.GitWorkflowService["Service"]["renameBranch"]; readonly runSetup?: ProjectSetupScriptRunner.ProjectSetupScriptRunner["Service"]["runForThread"]; readonly generateTitle?: TextGeneration.TextGeneration["Service"]["generateThreadTitle"]; 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 = {}) { @@ -90,7 +94,14 @@ { databaseLayer: database, runEffectWorker: false }, ); const threadManagement = ThreadManagement.layer.pipe(Layer.provide(orchestrator)); + const baseReceipts = CommandReceiptStore.layer.pipe(Layer.provide(database)); + const receipts = + options.mapCommandReceipts === undefined + ? baseReceipts + : Layer.effect( - const receipts = CommandReceiptStore.layer.pipe(Layer.provide(database)); + 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 ?? @@ -102,6 +113,7 @@ const renameBranch = vi.fn( options.renameBranch ?? ((input) => Effect.succeed({ branch: input.newBranch })), ); + const currentBranch = vi.fn(options.currentBranch ?? (() => Effect.succeed("main"))); const runSetup = vi.fn( options.runSetup ?? (() => Effect.succeed({ status: "no-script" as const })), ); @@ -126,6 +138,7 @@ }), Layer.mock(GitWorkflow.GitWorkflowService)({ createWorktree, + currentBranch, renameBranch, fetchRemote: options.fetchRemote ?? (() => Effect.void), remoteExists: () => Effect.succeed(true), @@ -172,6 +185,7 @@ return { layer: Layer.mergeAll(launch, threadManagement, titleRegeneration, outbox, database), createWorktree, + currentBranch, renameBranch, generateBranchName, generateThreadTitle, @@ -179,6 +193,105 @@ }; } +it.effect("rechecks an expected branch immediately before launch acceptance", () => + Effect.gen(function* () { + const branch = yield* Ref.make("main"); + const branchReadEntered = yield* Deferred.make(); + const allowBranchRead = yield* Deferred.make(); + const harness = makeHarness({ + currentBranch: () => + Deferred.succeed(branchReadEntered, undefined).pipe( + Effect.andThen(Deferred.await(allowBranchRead)), + Effect.andThen(Ref.get(branch)), + ), + }); + + yield* Effect.gen(function* () { + const launches = yield* ThreadLaunch.ThreadLaunchService; + const threads = yield* ThreadManagement.ThreadManagementService; + const input = launchInput({ + command: "command:launch:expected-branch-race", + thread: "thread:launch:expected-branch-race", + workspace: { type: "root", branch: "main", expectedBranch: "main" }, + }); + const launchFiber = yield* Effect.forkChild(launches.launch(input)); + + yield* Deferred.await(branchReadEntered); + yield* Ref.set(branch, "feature"); + yield* Deferred.succeed(allowBranchRead, undefined); + + const error = yield* Fiber.join(launchFiber).pipe(Effect.flip); + assert.equal(error.operation, "validate-workspace"); + assert.match(String(error.cause), /branch 'feature'.*requested branch 'main'/u); + assert.isTrue( + Exit.isFailure(yield* threads.getThreadProjection(input.threadId).pipe(Effect.exit)), + ); + }).pipe(Effect.provide(harness.layer)); + }), +); + +it.effect("rechecks a detached HEAD snapshot immediately before launch acceptance", () => + Effect.gen(function* () { + const branch = yield* Ref.make(null); + const branchReadEntered = yield* Deferred.make(); + const allowBranchRead = yield* Deferred.make(); + const harness = makeHarness({ + currentBranch: () => + Deferred.succeed(branchReadEntered, undefined).pipe( + Effect.andThen(Deferred.await(allowBranchRead)), + Effect.andThen(Ref.get(branch)), + ), + }); + + yield* Effect.gen(function* () { + const launches = yield* ThreadLaunch.ThreadLaunchService; + const threads = yield* ThreadManagement.ThreadManagementService; + const input = launchInput({ + command: "command:launch:expected-detached-race", + thread: "thread:launch:expected-detached-race", + workspace: { type: "root", expectedBranch: null }, + }); + const launchFiber = yield* Effect.forkChild(launches.launch(input)); + + yield* Deferred.await(branchReadEntered); + yield* Ref.set(branch, "main"); + yield* Deferred.succeed(allowBranchRead, undefined); + + const error = yield* Fiber.join(launchFiber).pipe(Effect.flip); + assert.equal(error.operation, "validate-workspace"); + assert.match(String(error.cause), /branch 'main'.*expected a detached HEAD/u); + assert.isTrue( + Exit.isFailure(yield* threads.getThreadProjection(input.threadId).pipe(Effect.exit)), + ); + }).pipe(Effect.provide(harness.layer)); + }), +); + +it.effect("replays an accepted launch before rechecking mutable branch state", () => + Effect.gen(function* () { + const branch = yield* Ref.make("main"); + const harness = makeHarness({ currentBranch: () => Ref.get(branch) }); + + yield* Effect.gen(function* () { + const launches = yield* ThreadLaunch.ThreadLaunchService; + const input = launchInput({ + command: "command:launch:expected-branch-replay", + thread: "thread:launch:expected-branch-replay", + workspace: { type: "root", branch: "main", expectedBranch: "main" }, + }); + + const first = yield* launches.launch(input); + yield* Ref.set(branch, "feature"); + const replay = yield* launches.launch(input); + + assert.isFalse(first.resumed); + assert.isTrue(replay.resumed); + assert.equal(replay.threadId, first.threadId); + assert.equal(harness.currentBranch.mock.calls.length, 1); + }).pipe(Effect.provide(harness.layer)); + }), +); + function launchInput(input: { readonly command: string; readonly thread: string; @@ -430,6 +543,66 @@ }), ); +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(); ``` ## apps/server/src/orchestration-v2/ThreadLaunchService.ts Status: modified; +47; −4 ```diff @@ -5,6 +5,7 @@ type ModelSelection, type OrchestrationV2Actor, type OrchestrationV2CreationSource, + type OrchestrationV2PolicyCeiling, type OrchestrationV2ThreadProjection, type ProviderInteractionMode, ProjectId, @@ -36,11 +37,16 @@ import * as ThreadManagement from "./ThreadManagementService.ts"; export type ThreadLaunchWorkspaceStrategy = + | { + readonly type: "root"; + readonly branch?: string | undefined; + readonly expectedBranch?: string | null | undefined; + } - | { readonly type: "root"; readonly branch?: string | undefined } | { readonly type: "existing_worktree"; readonly worktreePath: string; readonly branch?: string | undefined; + readonly expectedBranch?: string | null | undefined; } | { readonly type: "worktree"; @@ -65,6 +71,7 @@ readonly modelSelection: ModelSelection; readonly runtimeMode: RuntimeMode; readonly interactionMode: ProviderInteractionMode; + readonly policyCeiling?: OrchestrationV2PolicyCeiling; readonly workspaceStrategy: ThreadLaunchWorkspaceStrategy; readonly initialMessage?: ThreadLaunchInitialMessage; readonly createdBy: OrchestrationV2Actor; @@ -75,6 +82,8 @@ 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()( @@ -86,6 +95,7 @@ "generate-metadata", "provision-worktree", "run-setup-script", + "validate-workspace", "create-thread", "update-thread", "dispatch-message", @@ -456,24 +466,52 @@ .thread({ projectId: input.projectId }) .pipe(Effect.mapError(mapError(input, "create-thread")))); + let initialBranch = input.workspaceStrategy.branch ?? null; if (Option.isNone(launchReceipt)) { - yield* projects.getById(input.projectId).pipe( + const project = yield* projects.getById(input.projectId).pipe( Effect.mapError(mapError(input, "resolve-project")), Effect.flatMap( Option.match({ onNone: () => Effect.fail(mapError(input, "resolve-project")("Project not found.")), - onSome: () => Effect.void, + onSome: Effect.succeed, }), ), ); + if ( + input.workspaceStrategy.type !== "worktree" && + input.workspaceStrategy.expectedBranch !== undefined + ) { + const cwd = + input.workspaceStrategy.type === "existing_worktree" + ? input.workspaceStrategy.worktreePath + : project.workspaceRoot; + const currentBranch = yield* git + .currentBranch(cwd) + .pipe( + Effect.mapError(mapError(input, "validate-workspace", candidateThreadId)), + ); + if (currentBranch !== input.workspaceStrategy.expectedBranch) { + const expectedBranch = + input.workspaceStrategy.expectedBranch === null + ? "expected a detached HEAD" + : `requested branch '${input.workspaceStrategy.expectedBranch}'`; + return yield* mapError( + input, + "validate-workspace", + candidateThreadId, + )( + `Workspace '${cwd}' is on ${currentBranch === null ? "a detached HEAD" : `branch '${currentBranch}'`}, not ${expectedBranch}.`, + ); + } + initialBranch = currentBranch; + } } if (input.reuseExistingThread === true && Option.isNone(launchReceipt)) { yield* validateReusableThread(input, candidateThreadId); } - const initialBranch = input.workspaceStrategy.branch ?? null; const initialWorktreePath = input.workspaceStrategy.type === "existing_worktree" ? input.workspaceStrategy.worktreePath @@ -494,6 +532,9 @@ modelSelection: input.modelSelection, runtimeMode: input.runtimeMode, interactionMode: input.interactionMode, + ...(input.policyCeiling === undefined + ? {} + : { policyCeiling: input.policyCeiling }), branch: initialBranch, worktreePath: initialWorktreePath, createdBy: input.createdBy, @@ -543,6 +584,7 @@ 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, @@ -594,6 +636,7 @@ threadId, projection, resumed: Option.isSome(launchReceipt) || messageWasAlreadyAccepted, + initialMessageRunId: runId, }; }); }, ``` ## apps/server/src/orchestration-v2/ThreadManagementService.ts Status: modified; +12; −0 ```diff @@ -30,6 +30,7 @@ OrchestratorV2, type OrchestratorV2DispatchResult, type OrchestratorV2Error, + type OrchestratorV2ServerCreatedThreadRecordInput, } from "./Orchestrator.ts"; import { LegacyV1ThreadImporter, @@ -284,6 +285,9 @@ readonly dispatch: ( command: OrchestrationV2Command, ) => Effect.Effect; + readonly recordServerCreatedThread: ( + input: OrchestratorV2ServerCreatedThreadRecordInput, + ) => Effect.Effect; readonly getThreadProjection: ( threadId: ThreadId, ) => Effect.Effect; @@ -478,6 +482,13 @@ ); }; + 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( @@ -708,6 +719,7 @@ withProjectMutationLock: threadCommands.withProjectLock, ensureLegacyTranscript, dispatch, + recordServerCreatedThread, getThreadProjection, getCheckpointContext, getThreadSnapshot, ``` ## apps/server/src/relay/AgentAwarenessRelay.test.ts Status: modified; +1; −0 ```diff @@ -132,6 +132,7 @@ getCheckpointContext: unused, getThreadSnapshot: unused, getThreadSnapshotWindow: unused, + recordServerCreatedThread: unused, getProjectThread: unused, listProjectThreads: unused, sendToThread: unused, ``` ## docs/orchestration-v2/orchestrator-mcp-server.md Status: modified; +45; −9 ````diff @@ -180,7 +180,8 @@ - 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. @@ -272,6 +273,11 @@ 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"; }>; @@ -279,24 +285,45 @@ }; ``` +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 -Each entry independently resolves provider, model, and modes. The new threads +when that execution directory is a subdirectory of the physical Git worktree. -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. +Project selection does not raise the caller's authority. Provider and model resolve from an +explicit `target` when supplied and otherwise inherit from the caller. Runtime and interaction +modes inherit by default, and requested modes cannot exceed the caller's 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 an explicitly selected project in the same environment, newest +first. Omitting `projectId` keeps the calling thread's project as the scope. -Lists durable thread shells in the calling thread's project, newest first. 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` @@ -327,9 +354,14 @@ receipt. Thread list and read results expose the linked pull request, and thread detail also exposes an in-flight title regeneration. +`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; @@ -396,6 +428,10 @@ 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. ```` ## docs/user/project-settings.md Status: modified; +17; −0 ```diff @@ -26,3 +26,20 @@ 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. ``` ## packages/contracts/src/orchestrationV2.test.ts Status: modified; +2; −0 ```diff @@ -269,6 +269,7 @@ 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 @@ 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", () => { ``` ## packages/contracts/src/orchestrationV2.ts Status: modified; +13; −0 ```diff @@ -73,6 +73,15 @@ creationSource: OrchestrationV2CreationSource, } as const; +export const OrchestrationV2PolicyCeiling = Schema.Struct({ + callerThreadId: ThreadId, + callerRunId: Schema.optional(RunId), + callerProviderInstanceId: Schema.optional(ProviderInstanceId), + runtimeMode: RuntimeMode, + interactionMode: ProviderInteractionMode, +}); +export type OrchestrationV2PolicyCeiling = typeof OrchestrationV2PolicyCeiling.Type; + export const OrchestrationV2NativeRefStrength = Schema.Literals(["strong", "weak", "none"]); export type OrchestrationV2NativeRefStrength = typeof OrchestrationV2NativeRefStrength.Type; @@ -1078,6 +1087,7 @@ ...OrchestrationV2TurnItemBaseFields, type: Schema.Literal("thread_created"), targetThreadId: ThreadId, + targetProjectId: Schema.optional(ProjectId), targetRunId: Schema.NullOr(RunId), targetProviderInstanceId: ProviderInstanceId, targetModel: TrimmedNonEmptyString, @@ -1760,6 +1770,7 @@ ...OrchestrationV2TurnItemJsonBaseFields, type: Schema.Literal("thread_created"), targetThreadId: ThreadId, + targetProjectId: Schema.optional(ProjectId), targetRunId: Schema.NullOr(RunId), targetProviderInstanceId: ProviderInstanceId, targetModel: TrimmedNonEmptyString, @@ -2035,6 +2046,7 @@ modelSelection: ModelSelection, runtimeMode: RuntimeMode, interactionMode: ProviderInteractionMode, + policyCeiling: Schema.optional(OrchestrationV2PolicyCeiling), branch: Schema.NullOr(TrimmedNonEmptyString), worktreePath: Schema.NullOr(TrimmedNonEmptyString), }), @@ -2186,6 +2198,7 @@ /** 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 })), restartContinuationOfRunId: Schema.optional(RunId), delegatedCompletion: Schema.optional( ``` ## packages/contracts/src/orchestratorMcp.test.ts Status: modified; +18; −0 ```diff @@ -117,6 +117,12 @@ { prompt: "Review the API.", target: { driverKind: "claudeAgent" }, + projectId: "project-review", + workspaceStrategy: { + type: "new_worktree", + baseRef: "trunk", + branch: "review-api", + }, runtimeMode: "approval-required", }, ], @@ -125,17 +131,25 @@ 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, @@ -144,6 +158,7 @@ expect( decodeThreadReadInput({ threadId: "thread-loop-1", + projectId: "project-loop", view: "activity", afterPosition: 10, }).afterPosition, @@ -151,6 +166,7 @@ expect( decodeThreadSendInput({ threadId: "thread-loop-1", + projectId: "project-loop", message: "Continue with the next iteration.", mode: "steer", clientRequestId: "send-loop-2", @@ -159,6 +175,7 @@ expect( decodeThreadWaitInput({ threadId: "thread-loop-1", + projectId: "project-loop", runId: "run-loop-2", timeoutMs: 5_000, }).runId, @@ -166,6 +183,7 @@ expect( decodeThreadInterruptInput({ threadId: "thread-loop-1", + projectId: "project-loop", reason: "Loop converged.", }).reason, ).toBe("Loop converged."); ``` [Next page](https://git.cafe/pingdotgg/t3code/pull/8678/diffs?after=InBhY2thZ2VzL2NvbnRyYWN0cy9zcmMvb3JjaGVzdHJhdG9yTWNwLnRlc3QudHMi&expectedVersion=34&format=markdown)