From 25629a76a0458bdeb707483722894158af090b41 Mon Sep 17 00:00:00 2001 From: Bilal Bakr <62337003+Bil0000@users.noreply.github.com> Date: Wed, 14 Oct 2026 03:39:02 +0300 Subject: [PATCH] feat(server): run delegated tasks in separate worktrees (#19102) --- .../OrchestratorMcpService.activity.test.ts | 15 +- .../src/mcp/OrchestratorMcpService.test.ts | 55 ++--- apps/server/src/mcp/OrchestratorMcpService.ts | 39 +++ ...OrchestratorMcpToolkit.integration.test.ts | 11 +- ...skCancelNativeSubagent.integration.test.ts | 9 +- apps/server/src/mcp/toolkits/core.test.ts | 15 +- .../src/mcp/toolkits/orchestrator/tools.ts | 2 +- .../src/orchestration-v2/Orchestrator.ts | 6 +- .../ProviderRuntimeRecoveryService.test.ts | 40 ++- .../ThreadLaunchService.test.ts | 230 +++++++++++++++++- .../orchestration-v2/ThreadLaunchService.ts | 96 +++++--- packages/contracts/src/orchestrationV2.ts | 1 + .../contracts/src/orchestratorMcp.test.ts | 7 + packages/contracts/src/orchestratorMcp.ts | 6 + .../src/server/orchestrationInstructions.ts | 2 +- 15 files changed, 454 insertions(+), 80 deletions(-) diff --git a/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts b/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts index ce6921d02da42dbbdc16283f5e3c98db14efbe9f..1c81d0ddb26400c0119894c2db90d16f3cbc85d2 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts @@ -22,10 +22,17 @@ import * as ProjectService from "../project/ProjectService.ts"; import * as ScheduledTaskService from "../scheduledTasks/ScheduledTaskService.ts"; import * as ServerSettings from "../serverSettings.ts"; import * as SecretRequests from "../secrets/SecretRequests.ts"; +import * as ThreadLaunchService from "../orchestration-v2/ThreadLaunchService.ts"; +import * as GitVcsDriver from "../vcs/GitVcsDriver.ts"; import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts"; import type * as McpInvocationContext from "./McpInvocationContext.ts"; import * as OrchestratorMcpService from "./OrchestratorMcpService.ts"; +const layerOrchestratorService = OrchestratorMcpService.layer.pipe( + Layer.provide(Layer.mock(ThreadLaunchService.ThreadLaunchService)({})), + Layer.provide(Layer.mock(GitVcsDriver.GitVcsDriver)({})), +); + const environmentId = EnvironmentId.make("environment-mcp-orchestrator-detail"); const projectId = ProjectId.make("project-mcp-orchestrator-detail"); const parentThreadId = ThreadId.make("thread-mcp-orchestrator-parent"); @@ -127,7 +134,7 @@ it("readThread prefers activity-run status over a newer cancelled queued run", a updatedAt: now, } as unknown as OrchestrationV2ThreadProjection; - const layer = OrchestratorMcpService.layer.pipe( + const layer = layerOrchestratorService.pipe( Layer.provide( Layer.mergeAll( Layer.mock(ThreadManagementService.ThreadManagementService)({ @@ -193,7 +200,7 @@ it("readThread prefers waiting activity status over a newer cancelled queued run updatedAt: now, } as unknown as OrchestrationV2ThreadProjection; - const layer = OrchestratorMcpService.layer.pipe( + const layer = layerOrchestratorService.pipe( Layer.provide( Layer.mergeAll( Layer.mock(ThreadManagementService.ThreadManagementService)({ @@ -316,7 +323,7 @@ it("taskStatus returns task.providerInstanceId rather than the driver kind", asy updatedAt: now, } as unknown as OrchestrationV2ThreadProjection; - const layer = OrchestratorMcpService.layer.pipe( + const layer = layerOrchestratorService.pipe( Layer.provide( Layer.mergeAll( Layer.mock(ThreadManagementService.ThreadManagementService)({ @@ -415,7 +422,7 @@ it("readThread and sendToThread reach threads in other projects", async () => { updatedAt: now, }) as unknown as OrchestrationV2ThreadProjection; - const layer = OrchestratorMcpService.layer.pipe( + const layer = layerOrchestratorService.pipe( Layer.provide( Layer.mergeAll( Layer.mock(ThreadManagementService.ThreadManagementService)({ diff --git a/apps/server/src/mcp/OrchestratorMcpService.test.ts b/apps/server/src/mcp/OrchestratorMcpService.test.ts index 1b68f710d6df75d319ac0d1f9aefcddd5d0bf7b5..b0688fd217bb803ce2793caa8313efd69198d776 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.test.ts @@ -29,6 +29,8 @@ import { OrchestratorThreadAboveModeLimitError, } from "../orchestration-v2/Orchestrator.ts"; import * as ProviderAdapterRegistry from "../orchestration-v2/ProviderAdapterRegistry.ts"; +import * as ThreadLaunchService from "../orchestration-v2/ThreadLaunchService.ts"; +import * as GitVcsDriver from "../vcs/GitVcsDriver.ts"; import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts"; import * as ProviderRegistry from "../provider/ProviderRegistry.ts"; import { buildUnavailableProviderSnapshot } from "../provider/unavailableProviderSnapshot.ts"; @@ -41,6 +43,11 @@ import { idleThreadProjection, liveThreadShell } from "./McpToolAccess.testkit.t import * as OrchestratorMcpService from "./OrchestratorMcpService.ts"; import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; +const layerOrchestratorService = OrchestratorMcpService.layer.pipe( + Layer.provide(Layer.mock(ThreadLaunchService.ThreadLaunchService)({})), + Layer.provide(Layer.mock(GitVcsDriver.GitVcsDriver)({})), +); + describe("OrchestratorMcpService", () => { it.effect("retries terminal acknowledgement with a fresh command id", () => Effect.gen(function* () { @@ -170,7 +177,7 @@ describe("OrchestratorMcpService", () => { const commandIds = yield* Ref.get(acknowledgementCommandIds); assert.equal(commandIds.length, 2); assert.notEqual(commandIds[0], commandIds[1]); - }).pipe(Effect.provide(OrchestratorMcpService.layer.pipe(Layer.provide(layerDependencies)))); + }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies)))); }), ); @@ -267,7 +274,7 @@ describe("OrchestratorMcpService", () => { const settled = yield* service.taskStatus(scope, taskId); assert.equal(settled.status, "cancelled"); assert.equal(yield* Ref.get(dispatched), 1); - }).pipe(Effect.provide(OrchestratorMcpService.layer.pipe(Layer.provide(layerDependencies)))); + }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies)))); }), ); @@ -343,7 +350,7 @@ describe("OrchestratorMcpService", () => { .pipe(Effect.flip); assert.equal(error.code, "task_not_cancellable"); assert.deepEqual(yield* Ref.get(dispatched), []); - }).pipe(Effect.provide(OrchestratorMcpService.layer.pipe(Layer.provide(layerDependencies)))); + }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies)))); }), ); @@ -423,7 +430,7 @@ describe("OrchestratorMcpService", () => { (yield* Ref.get(dispatched)).map((command) => (command as { type: string }).type), ["thread.stop"], ); - }).pipe(Effect.provide(OrchestratorMcpService.layer.pipe(Layer.provide(layerDependencies)))); + }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies)))); }), ); @@ -509,7 +516,7 @@ describe("OrchestratorMcpService", () => { (yield* Ref.get(dispatched)).map((command) => (command as { type: string }).type), ["thread.stop", "delegated_task.completion-delivery.dispose"], ); - }).pipe(Effect.provide(OrchestratorMcpService.layer.pipe(Layer.provide(layerDependencies)))); + }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies)))); }), ); @@ -592,7 +599,7 @@ describe("OrchestratorMcpService", () => { .pipe(Effect.flip); assert.equal(error.code, "runtime_mode_escalation_denied"); assert.deepEqual(yield* Ref.get(dispatched), []); - }).pipe(Effect.provide(OrchestratorMcpService.layer.pipe(Layer.provide(layerDependencies)))); + }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies)))); }), ); @@ -736,7 +743,7 @@ describe("OrchestratorMcpService", () => { dispatched: yield* Ref.get(dispatched), stoppedBelow: yield* Ref.get(stoppedBelow), }; - }).pipe(Effect.provide(OrchestratorMcpService.layer.pipe(Layer.provide(layerDependencies)))); + }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies)))); }); it.effect("refuses to cancel a task when a task under it now runs above the parent's modes", () => @@ -1019,9 +1026,7 @@ describe("OrchestratorMcpService provider resolution", () => { assert.isTrue( fork!.constraints.includes("Driver 'forkOnly' is not registered in this build."), ); - }).pipe( - Effect.provide(OrchestratorMcpService.layer.pipe(Layer.provide(layerDependencies))), - ); + }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies)))); }), ); @@ -1066,7 +1071,7 @@ describe("OrchestratorMcpService provider resolution", () => { capabilities.providers[0]?.models.map((model) => model.id), ["gpt-5.4", "my-model"], ); - }).pipe(Effect.provide(OrchestratorMcpService.layer.pipe(Layer.provide(layerDependencies)))); + }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies)))); }), ); @@ -1166,9 +1171,7 @@ describe("OrchestratorMcpService provider resolution", () => { assert.equal(request.type, "delegated_task.request"); assert.equal(request.modelSelection.instanceId, antigravityInstanceId); assert.equal(request.modelSelection.model, "ant-model"); - }).pipe( - Effect.provide(OrchestratorMcpService.layer.pipe(Layer.provide(layerDependencies))), - ); + }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies)))); }), ); @@ -1264,7 +1267,7 @@ describe("OrchestratorMcpService provider resolution", () => { }; assert.equal(request.modelSelection.instanceId, antigravityInstanceId); assert.equal(request.modelSelection.model, "ant-model"); - }).pipe(Effect.provide(OrchestratorMcpService.layer.pipe(Layer.provide(layerDependencies)))); + }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies)))); }), ); @@ -1323,7 +1326,7 @@ describe("OrchestratorMcpService provider resolution", () => { assert.isTrue( byDriver.message.includes("No V2 provider adapter is registered for driver forkOnly."), ); - }).pipe(Effect.provide(OrchestratorMcpService.layer.pipe(Layer.provide(layerDependencies)))); + }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies)))); }), ); @@ -1369,7 +1372,7 @@ describe("OrchestratorMcpService provider resolution", () => { }) .pipe(Effect.flip); assert.equal(error.code, "invalid_request"); - }).pipe(Effect.provide(OrchestratorMcpService.layer.pipe(Layer.provide(layerDependencies)))); + }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies)))); }), ); @@ -1474,7 +1477,7 @@ describe("OrchestratorMcpService provider resolution", () => { modelSelection: { instanceId: string; model: string }; }; assert.equal(request.modelSelection.instanceId, altInstanceId); - }).pipe(Effect.provide(OrchestratorMcpService.layer.pipe(Layer.provide(layerDependencies)))); + }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies)))); }), ); @@ -1599,9 +1602,7 @@ describe("OrchestratorMcpService provider resolution", () => { modelSelection: { instanceId: string; model: string }; }; assert.equal(retryCommand.modelSelection.instanceId, codexInstanceId); - }).pipe( - Effect.provide(OrchestratorMcpService.layer.pipe(Layer.provide(layerDependencies))), - ); + }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies)))); }), ); @@ -1706,7 +1707,7 @@ describe("OrchestratorMcpService provider resolution", () => { assert.equal(result.providerInstanceId, claudeInstanceId); assert.equal(yield* Ref.get(probes), 2); assert.equal(yield* Ref.get(dispatched), 1); - }).pipe(Effect.provide(OrchestratorMcpService.layer.pipe(Layer.provide(layerDependencies)))); + }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies)))); }), ); @@ -1917,9 +1918,7 @@ describe("OrchestratorMcpService provider resolution", () => { } else { assert.equal(request.modelSelection.model, "codex-alt-model", testCase.name); } - }).pipe( - Effect.provide(OrchestratorMcpService.layer.pipe(Layer.provide(layerDependencies))), - ); + }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies)))); } }), ); @@ -1972,7 +1971,7 @@ describe("OrchestratorMcpService provider resolution", () => { boundThread: OrchestrationV2ThreadShell | null, upserted: Ref.Ref, ) => - OrchestratorMcpService.layer.pipe( + layerOrchestratorService.pipe( Layer.provide( Layer.mergeAll( NodeServices.layer, @@ -2040,7 +2039,7 @@ describe("OrchestratorMcpService provider resolution", () => { ), ), Effect.provide( - OrchestratorMcpService.layer.pipe( + layerOrchestratorService.pipe( Layer.provide( Layer.mergeAll( NodeServices.layer, @@ -2123,7 +2122,7 @@ describe("OrchestratorMcpService provider resolution", () => { // The edit's own check reads the thread; the read after the save fails. const mcp = yield* OrchestratorMcpService.OrchestratorMcpService.pipe( Effect.provide( - OrchestratorMcpService.layer.pipe( + layerOrchestratorService.pipe( Layer.provide( Layer.mergeAll( NodeServices.layer, diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index f9bb61fa810e9c485b11729131b3f10ae6772126..9a0575a85e302cf9da10800edf7862d0718f35e2 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -10,6 +10,7 @@ import { type OrchestrationV2Subagent, type OrchestrationV2ThreadProjection, type OrchestrationV2ThreadShell, + type OrchestrationV2ThreadLaunchWorkspaceStrategy, type OrchestrationV2TurnItem, OrchestratorMcpFailure, type OrchestratorMcpCapabilitiesResult, @@ -81,6 +82,8 @@ import { type DispatchModeRefusal, } from "../orchestration-v2/DispatchModeLimit.ts"; import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts"; +import * as ThreadLaunchService from "../orchestration-v2/ThreadLaunchService.ts"; +import * as GitVcsDriver from "../vcs/GitVcsDriver.ts"; import { isSnoozed } from "../orchestration-v2/ThreadSettlementService.ts"; import * as ProjectService from "../project/ProjectService.ts"; import * as ProviderRegistry from "../provider/ProviderRegistry.ts"; @@ -904,6 +907,8 @@ const make = Effect.gen(function* () { const providerAdapters = yield* ProviderAdapterRegistry.ProviderAdapterRegistryV2; const scheduledTasks = yield* ScheduledTaskService.ScheduledTaskService; const projects = yield* ProjectService.ProjectService; + const threadLaunch = yield* ThreadLaunchService.ThreadLaunchService; + const gitDriver = yield* GitVcsDriver.GitVcsDriver; const serverSettings = yield* ServerSettings.ServerSettingsService; /** A caller-named project, which must exist before anything is recorded against it. */ @@ -1993,6 +1998,21 @@ const make = Effect.gen(function* () { parent.thread.interactionMode, input.interactionMode, ); + let workspaceStrategy: OrchestrationV2ThreadLaunchWorkspaceStrategy | undefined; + if (input.isolation === "worktree" && !isRetryOfAcceptedDelegation) { + const project = yield* requireProject(parent.thread.projectId); + const base = yield* gitDriver + .resolveCommit({ + cwd: parent.thread.worktreePath ?? project.workspaceRoot, + revision: "HEAD", + }) + .pipe( + Effect.mapError(() => + failure("invalid_request", "Unable to resolve the parent checkout commit."), + ), + ); + workspaceStrategy = { type: "worktree", baseRef: base.commitSha, startFromOrigin: false }; + } // ThreadManagementService.dispatch is the chokepoint for this command // type's own enablement check, using the same command id and its // receipt-replay exemption, so no separate assert runs here. @@ -2010,6 +2030,7 @@ const make = Effect.gen(function* () { modelSelection: target.modelSelection, runtimeMode, interactionMode, + ...(workspaceStrategy === undefined ? {} : { workspaceStrategy }), // Async delegations wake the parent on every child terminal; wait // delegations deliver through the blocking tool call, so a wake is // only needed if the parent settled first (timeout, disconnect). @@ -2036,6 +2057,22 @@ const make = Effect.gen(function* () { ); } const taskId = taskEvent.event.payload.id; + const childThreadId = taskEvent.event.payload.childThreadId; + const childRunEvent = result.storedEvents.find( + (stored) => + stored.event.type === "run.created" && stored.event.threadId === childThreadId, + ); + if ( + childThreadId !== null && + childRunEvent?.event.type === "run.created" && + childRunEvent.event.payload.workspacePreparation !== undefined + ) { + yield* threadLaunch.prepareRun({ + commandId, + threadId: childThreadId, + runId: childRunEvent.event.payload.id, + }); + } if (input.mode !== "wait") { return yield* readTask(scope, taskId, false, true); @@ -2660,6 +2697,8 @@ export const layer: Layer.Layer< never, | Crypto.Crypto | ThreadManagementService.ThreadManagementService + | ThreadLaunchService.ThreadLaunchService + | GitVcsDriver.GitVcsDriver | ProviderRegistry.ProviderRegistry | ProviderAdapterRegistry.ProviderAdapterRegistryV2 | ScheduledTaskService.ScheduledTaskService diff --git a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts index 0b3d1b0e18cf2c1c892fbad8d7d4de2a63e43e55..b68f3f29c8da95a02f8b12864ebaf9455f0469ba 100644 --- a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts @@ -57,6 +57,8 @@ import { threadShellFromProjection } from "../orchestration-v2/ProjectionStore.t import * as CommandReceiptStore from "../orchestration-v2/CommandReceiptStore.ts"; import * as EventSink from "../orchestration-v2/EventSink.ts"; import * as Orchestrator from "../orchestration-v2/Orchestrator.ts"; +import * as ThreadLaunchService from "../orchestration-v2/ThreadLaunchService.ts"; +import * as GitVcsDriver from "../vcs/GitVcsDriver.ts"; import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts"; import * as ServerSettings from "../serverSettings.ts"; import * as SqlitePersistence from "../persistence/Sqlite.ts"; @@ -80,6 +82,11 @@ import * as McpInvocationContext from "./McpInvocationContext.ts"; import { delegatedTaskRun, hasPendingChildRuns } from "./OrchestratorMcpService.ts"; import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; +const layerOrchestratorToolkit = McpHttpServer.layerOrchestratorToolkit.pipe( + Layer.provide(Layer.mock(ThreadLaunchService.ThreadLaunchService)({})), + Layer.provide(Layer.mock(GitVcsDriver.GitVcsDriver)({})), +); + // Effect returns a declared tool failure as `isError` with its encoded payload // as JSON text, never as `structuredContent`. const declaredFailure = (result: McpSchema.CallToolResult) => { @@ -682,7 +689,7 @@ describe("orchestrator MCP toolkit", () => { }), ); const layerTest = Layer.merge( - McpHttpServer.layerOrchestratorToolkit, + layerOrchestratorToolkit, McpHttpServer.layerThreadToolkit, ).pipe( Layer.provideMerge(McpServer.McpServer.layer), @@ -3838,7 +3845,7 @@ describe("orchestrator MCP toolkit", () => { model: codexModel, }), ]); - const layerTest = McpHttpServer.layerOrchestratorToolkit.pipe( + const layerTest = layerOrchestratorToolkit.pipe( Layer.provideMerge(McpServer.McpServer.layer), Layer.provideMerge(layerOrchestration), Layer.provide( diff --git a/apps/server/src/mcp/TaskCancelNativeSubagent.integration.test.ts b/apps/server/src/mcp/TaskCancelNativeSubagent.integration.test.ts index c3bd9f0af5b400c60a77a3a6a2117767e043ddb9..9f58cb799150e916d375b2cee94aa55e09aa13e9 100644 --- a/apps/server/src/mcp/TaskCancelNativeSubagent.integration.test.ts +++ b/apps/server/src/mcp/TaskCancelNativeSubagent.integration.test.ts @@ -27,6 +27,8 @@ import * as EffectWorker from "../orchestration-v2/EffectWorker.ts"; import * as Orchestrator from "../orchestration-v2/Orchestrator.ts"; import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; import * as ProviderAdapterRegistry from "../orchestration-v2/ProviderAdapterRegistry.ts"; +import * as ThreadLaunchService from "../orchestration-v2/ThreadLaunchService.ts"; +import * as GitVcsDriver from "../vcs/GitVcsDriver.ts"; import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts"; import * as ServerSettings from "../serverSettings.ts"; import * as ProviderReplayHarness from "../orchestration-v2/testkit/ProviderReplayHarness.ts"; @@ -39,6 +41,11 @@ import * as SecretRequests from "../secrets/SecretRequests.ts"; import type { McpInvocationScope } from "./McpInvocationContext.ts"; import * as OrchestratorMcpService from "./OrchestratorMcpService.ts"; +const layerOrchestratorService = OrchestratorMcpService.layer.pipe( + Layer.provide(Layer.mock(ThreadLaunchService.ThreadLaunchService)({})), + Layer.provide(Layer.mock(GitVcsDriver.GitVcsDriver)({})), +); + const driver = ProviderDriverKind.make("claudeAgent"); const instanceId = ProviderInstanceId.make("claudeAgent"); const modelSelection = { instanceId, model: "claude-test" }; @@ -143,7 +150,7 @@ it.effect("task_cancel ends a delegated child's background native subagent", () ), ), ); - const layerTest = OrchestratorMcpService.layer.pipe( + const layerTest = layerOrchestratorService.pipe( Layer.provideMerge(layerOrchestration), Layer.provide( Layer.mergeAll( diff --git a/apps/server/src/mcp/toolkits/core.test.ts b/apps/server/src/mcp/toolkits/core.test.ts index a9e2997282edb0a1ba7953250c3da00d27f09d4c..92a179c1138d4acd5ce8598ecf98bba1280c9715 100644 --- a/apps/server/src/mcp/toolkits/core.test.ts +++ b/apps/server/src/mcp/toolkits/core.test.ts @@ -35,6 +35,8 @@ import { import * as ServerSecretStore from "../../auth/ServerSecretStore.ts"; import * as ServerConfig from "../../config.ts"; import * as ProviderAdapterRegistry from "../../orchestration-v2/ProviderAdapterRegistry.ts"; +import * as ThreadLaunchService from "../../orchestration-v2/ThreadLaunchService.ts"; +import * as GitVcsDriver from "../../vcs/GitVcsDriver.ts"; import * as ThreadManagement from "../../orchestration-v2/ThreadManagementService.ts"; import * as ThreadSearch from "../../orchestration-v2/ThreadSearch.ts"; import * as PreviewBrowser from "../../preview/PreviewBrowser.ts"; @@ -59,6 +61,11 @@ import { ThreadToolkit } from "./thread/tools.ts"; import { WorktreeToolkit } from "./worktree/tools.ts"; import { DeviceToolkit } from "./device/tools.ts"; +const layerOrchestratorToolkit = McpHttpServer.layerOrchestratorToolkit.pipe( + Layer.provide(Layer.mock(ThreadLaunchService.ThreadLaunchService)({})), + Layer.provide(Layer.mock(GitVcsDriver.GitVcsDriver)({})), +); + // Effect returns a declared tool failure as `isError` with its encoded payload // as JSON text, never as `structuredContent`. const declaredFailure = (result: McpSchema.CallToolResult) => { @@ -578,7 +585,7 @@ it.effect("refuses act-as-caller tools to a client caller", () => expect(declaredFailure(result)).toMatchObject({ code: "thread_credential_required" }); }).pipe( Effect.provide( - McpHttpServer.layerOrchestratorToolkit.pipe( + layerOrchestratorToolkit.pipe( Layer.provideMerge(McpServer.McpServer.layer), Layer.provide(NodeCrypto.layer), Layer.provide(Layer.mock(ThreadManagement.ThreadManagementService)({})), @@ -617,7 +624,7 @@ it.effect("a caller cannot rewrite a scheduled task that runs above its own mode expect(allowed.isError).toBe(false); }).pipe( Effect.provide( - McpHttpServer.layerOrchestratorToolkit.pipe( + layerOrchestratorToolkit.pipe( Layer.provideMerge(McpServer.McpServer.layer), Layer.provide(NodeCrypto.layer), Layer.provide(Layer.mock(ThreadManagement.ThreadManagementService)({})), @@ -692,7 +699,7 @@ it.effect.each(["t3_thread_interrupt", "t3_workflow_stop"])( expect(declaredFailure(result)).toMatchObject({ code: "runtime_mode_escalation_denied" }); }).pipe( Effect.provide( - McpHttpServer.layerOrchestratorToolkit.pipe( + layerOrchestratorToolkit.pipe( Layer.provideMerge(McpServer.McpServer.layer), Layer.provide(NodeCrypto.layer), Layer.provide( @@ -796,7 +803,7 @@ it.effect("discovers workflow stop targets and requires orchestration capability expect(stopped).toEqual([{ threadId: "workflow-parent", subagentId: "workflow-coordinator" }]); }).pipe( Effect.provide( - McpHttpServer.layerOrchestratorToolkit.pipe( + layerOrchestratorToolkit.pipe( Layer.provideMerge(McpServer.McpServer.layer), Layer.provide(NodeCrypto.layer), Layer.provide( diff --git a/apps/server/src/mcp/toolkits/orchestrator/tools.ts b/apps/server/src/mcp/toolkits/orchestrator/tools.ts index f2c34bb22cbe151b496d7ec28a0cf33a68bb8520..8984cb1a47af3218bace53f04f6146372ca0ab5f 100644 --- a/apps/server/src/mcp/toolkits/orchestrator/tools.ts +++ b/apps/server/src/mcp/toolkits/orchestrator/tools.ts @@ -65,7 +65,7 @@ const OrchestratorCapabilitiesTool = Tool.make("orchestrator_capabilities", { export const DelegateTaskTool = Tool.make("delegate_task", { description: - "Needs an agent running inside a T3 thread. Delegate one task to a T3-owned child agent/subagent of THIS thread and run it with only the supplied task prompt, without copying parent conversation history. Choose providers and models from orchestrator_capabilities, which uses the same live catalog as the composer. Prefer native subagent tools for same-provider work only when they support the chosen model. Use this for any model missing from the native tool, including same-provider work, for cross-provider work, or for explicitly T3-owned child tasks. For every T3 delegated review round, call delegate_task again with the original brief, prior findings, responses, and unresolved objections in the task prompt. Track each round by its own taskId and use a distinct clientRequestId per round, stable across retries of that round. The childThreadId is backing storage, not the target for starting another delegated review round through t3_thread_send. Provider, model, model options (see orchestrator_capabilities), runtime mode, and interaction mode inherit unless target overrides them. A requested runtimeMode must be one the target provider lists in orchestrator_capabilities runtimeModes; an inherited one steps down to the broadest mode it offers. Prefer mode='async' for long work; mode='wait' blocks until completion or timeout. timeoutMs on mode=wait is only the parent's wait budget and does not cancel the child. waitTimedOut on that wait call means the timeout fired; keep that taskId and read status on later task_status. An async child's completion wakes this thread through a notification, steered into active turns where supported or queued otherwise, so end the turn instead of polling or spawning watchers; use task_status only when the result is needed mid-turn.", + "Needs an agent running inside a T3 thread. Delegate one task to a T3-owned child agent/subagent of THIS thread and run it with only the supplied task prompt, without copying parent conversation history. Choose providers and models from orchestrator_capabilities, which uses the same live catalog as the composer. Prefer native subagent tools for same-provider work only when they support the chosen model. Use this for any model missing from the native tool, including same-provider work, for cross-provider work, or for explicitly T3-owned child tasks. For every T3 delegated review round, call delegate_task again with the original brief, prior findings, responses, and unresolved objections in the task prompt. Track each round by its own taskId and use a distinct clientRequestId per round, stable across retries of that round. The childThreadId is backing storage, not the target for starting another delegated review round through t3_thread_send. Set isolation to worktree for a separate checkout based on the parent local commit, with project setup completed before the child starts. Uncommitted edits are not copied; omitted isolation shares the parent checkout. Provider, model, model options (see orchestrator_capabilities), runtime mode, and interaction mode inherit unless target overrides them. A requested runtimeMode must be one the target provider lists in orchestrator_capabilities runtimeModes; an inherited one steps down to the broadest mode it offers. Prefer mode='async' for long work; mode='wait' blocks until completion or timeout. timeoutMs on mode=wait is only the parent's wait budget and does not cancel the child. waitTimedOut on that wait call means the timeout fired; keep that taskId and read status on later task_status. An async child's completion wakes this thread through a notification, steered into active turns where supported or queued otherwise, so end the turn instead of polling or spawning watchers; use task_status only when the result is needed mid-turn.", parameters: OrchestratorMcpDelegateTaskInput, success: OrchestratorMcpDelegateTaskResult, failure: OrchestratorMcpFailure, diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 12d4df5001e30c9af54950ab1fd0d0b2105f12d0..0756fdd59cdafe8767f2b9c123455f73bfb84e62 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -6666,6 +6666,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }), runtimeMode: command.runtimeMode, interactionMode: command.interactionMode, + ...(command.workspaceStrategy === undefined ? {} : { branch: null, worktreePath: null }), }; const task: OrchestrationV2Subagent = { id: taskNodeId, @@ -6783,7 +6784,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio text: command.task, attachments: [], modelSelection: command.modelSelection, - dispatchMode: { type: "start_immediately" }, + dispatchMode: + command.workspaceStrategy === undefined + ? { type: "start_immediately" } + : { type: "defer_start", workspaceStrategy: command.workspaceStrategy }, } satisfies Extract; yield* dispatchMessage(childMessageCommand, events, effects); diff --git a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts index 1e2fcc44248de0692f5f8ce8e7fca2ba55504da6..d6f8672ca52f258d244491ba147fb26fdff4a55f 100644 --- a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts @@ -643,7 +643,7 @@ it.effect("closes a secret request card's form when its run is recovered", () => }).pipe(Effect.provide(layer)); }); -it.effect("holds accepted queued work without cancelling its execution state after restart", () => { +it.effect.each(["queued", "preparing"] as const)("recovers $0 work after restart", (status) => { const threadId = ThreadId.make("thread_queued_restart"); const runId = RunId.make("run_queued_restart"); const attemptId = RunAttemptId.make("attempt_queued_restart"); @@ -659,9 +659,18 @@ it.effect("holds accepted queued work without cancelling its execution state aft runs: [ { id: runId, - status: "queued", + status, queuePosition: 1, providerInstanceId: ProviderInstanceId.make("codex"), + ...(status === "preparing" + ? { + workspacePreparation: { + type: "worktree", + baseRef: "parent-local-commit", + startFromOrigin: false, + }, + } + : {}), }, ], attempts: [ @@ -712,18 +721,33 @@ it.effect("holds accepted queued work without cancelling its execution state aft return Effect.gen(function* () { const summary = yield* (yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService).reconcile("startup"); - assert.equal(summary.terminalizedRuns, 0); + assert.equal(summary.terminalizedRuns, status === "preparing" ? 1 : 0); const command = committedInput; assert.isNotNull(command); if (command === null) return; const runEvent = command.events.find((event) => event.type === "run.updated"); const attemptEvent = command.events.find((event) => event.type === "run-attempt.updated"); const nodeEvent = command.events.find((event) => event.type === "node.updated"); - assert.equal(runEvent?.type === "run.updated" ? runEvent.payload.status : null, "queued"); - assert.equal(runEvent?.type === "run.updated" ? runEvent.payload.queuePosition : null, 1); - assert.equal(runEvent?.type === "run.updated" ? runEvent.payload.queueHeld : false, true); - assert.isUndefined(attemptEvent); - assert.isUndefined(nodeEvent); + assert.equal( + runEvent?.type === "run.updated" ? runEvent.payload.status : null, + status === "preparing" ? "cancelled" : "queued", + ); + assert.isEmpty(command.effects); + if (status === "preparing") { + assert.equal( + attemptEvent?.type === "run-attempt.updated" ? attemptEvent.payload.status : null, + "cancelled", + ); + assert.equal( + nodeEvent?.type === "node.updated" ? nodeEvent.payload.status : null, + "cancelled", + ); + } else { + assert.equal(runEvent?.type === "run.updated" ? runEvent.payload.queuePosition : null, 1); + assert.equal(runEvent?.type === "run.updated" ? runEvent.payload.queueHeld : false, true); + assert.isUndefined(attemptEvent); + assert.isUndefined(nodeEvent); + } }).pipe(Effect.provide(layer)); }); diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts index 3012e09411ac1dca1da3f1805f097348448bac46..26e9b8c8d78d4aed6cf05d39a46bfad75de0d877 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts @@ -15,6 +15,7 @@ import { ComposerContextId, type ChatAttachment, CommandId, + EnvironmentId, DEFAULT_SERVER_SETTINGS, GitCommandError, MessageId, @@ -56,6 +57,8 @@ import * as EffectOutbox from "./EffectOutbox.ts"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; import * as ThreadLaunch from "./ThreadLaunchService.ts"; +import * as OrchestratorMcpService from "../mcp/OrchestratorMcpService.ts"; +import * as GitVcsDriver from "../vcs/GitVcsDriver.ts"; import * as ThreadManagement from "./ThreadManagementService.ts"; import * as ThreadTitleRegeneration from "./ThreadTitleRegenerationService.ts"; import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts"; @@ -97,6 +100,8 @@ const adapter = { } as ProviderAdapter.ProviderAdapterV2["Service"]; interface HarnessOptions { + readonly closeTerminal?: TerminalManager.TerminalManager["Service"]["close"]; + readonly adapters?: ReadonlyArray; readonly managedFolders?: Layer.Layer; readonly createWorktree?: GitWorkflow.GitWorkflowService["Service"]["createWorktree"]; readonly fetchRemote?: GitWorkflow.GitWorkflowService["Service"]["fetchRemote"]; @@ -112,7 +117,7 @@ interface HarnessOptions { function makeHarness(options: HarnessOptions = {}) { const layerDatabase = SqlitePersistence.layerMemory; - const layerRegistry = ProviderAdapterRegistry.layerFromAdapters([adapter]); + const layerRegistry = ProviderAdapterRegistry.layerFromAdapters(options.adapters ?? [adapter]); const layerOrchestrator = ProviderReplayHarness.layerWithRegistry( { name: "thread-launch" }, layerRegistry, @@ -143,6 +148,7 @@ function makeHarness(options: HarnessOptions = {}) { const runSetup = vi.fn( options.runSetup ?? (() => Effect.succeed({ status: "no-script" as const })), ); + const closeTerminal = vi.fn(options.closeTerminal ?? (() => Effect.void)); const generateBranchName = vi.fn( options.generateBranchName ?? (() => Effect.succeed({ branch: "generated-branch" })), ); @@ -152,7 +158,7 @@ function makeHarness(options: HarnessOptions = {}) { const layerExternalServices = Layer.mergeAll( WorktreeSetupTracker.layer, Layer.mock(ProjectCloneTracker.ProjectCloneTracker)({ get: () => Effect.succeed(null) }), - Layer.mock(TerminalManager.TerminalManager)({ close: () => Effect.void }), + Layer.mock(TerminalManager.TerminalManager)({ close: closeTerminal }), Layer.succeed(ProjectService.ProjectService, { create: () => Effect.die("unused"), bootstrap: () => Effect.die("unused"), @@ -237,6 +243,7 @@ function makeHarness(options: HarnessOptions = {}) { return { layer: Layer.mergeAll( layerLaunch, + layerRegistry, layerThreadManagement, layerTitleRegeneration, layerOutbox, @@ -249,6 +256,7 @@ function makeHarness(options: HarnessOptions = {}) { generateBranchName, generateThreadTitle, runSetup, + closeTerminal, }; } @@ -2263,6 +2271,7 @@ it.effect("shared intake preserves durable attachment bytes after a lost launch ), checkWorktreeBase: launches.checkWorktreeBase, retryPreparation: launches.retryPreparation, + prepareRun: launches.prepareRun, }), Effect.flip, ); @@ -2705,3 +2714,220 @@ it.effect("replays an accepted message.dispatch after the project disables the i } }).pipe(Effect.provide(Layer.mergeAll(harness.layer, layerFiles))); }); + +it.effect.each( + (["codex", "claudeAgent"] as const).flatMap((driver) => + [ + { failSetup: false, failResolve: false }, + { failSetup: true, failResolve: false }, + { failSetup: false, failResolve: true }, + { failSetup: false, failResolve: false, cancelSetup: true }, + { failSetup: false, failResolve: false, cancelSetup: true, cancelAsync: true }, + ].map((failures) => ({ driver, cancelSetup: false, cancelAsync: false, ...failures })), + ), +)( + "prepares $driver delegated worktrees before execution, setup failure=$failSetup, commit failure=$failResolve, cancellation=$cancelSetup, async=$cancelAsync", + ({ driver, failSetup, failResolve, cancelSetup, cancelAsync }) => + Effect.gen(function* () { + const setupEntered = yield* Deferred.make(); + const allowSetup = yield* Deferred.make(); + const terminalClosed = yield* Deferred.make(); + const childInstanceId = ProviderInstanceId.make(`child-${driver}`); + const parentWorktree = driver === "codex" ? null : "/parent-worktree"; + const resolveCommit = vi.fn(() => + failResolve + ? Effect.fail( + new GitCommandError({ + operation: "resolveCommit", + command: "git", + cwd: "/private-checkout", + detail: "private Git failure details", + }), + ) + : Effect.succeed({ commitSha: "parent-local-commit" }), + ); + const harness = makeHarness({ + closeTerminal: () => Deferred.succeed(terminalClosed, undefined).pipe(Effect.asVoid), + adapters: [ + adapter, + { ...adapter, instanceId: childInstanceId, driver: ProviderDriverKind.make(driver) }, + ], + serverSettings: { + providerInstances: { + [childInstanceId]: { driver: ProviderDriverKind.make(driver), config: {} }, + }, + }, + hasCommit: () => Effect.succeed(true), + providers: [ + { + instanceId: childInstanceId, + driver: ProviderDriverKind.make(driver), + enabled: true, + installed: true, + version: "test", + checkedAt: "2026-10-13T00:00:00.000Z", + slashCommands: [], + skills: [], + status: "ready", + auth: { status: "authenticated" }, + models: [{ slug: "child-model", name: "Child", isCustom: true, capabilities: null }], + }, + ], + runSetup: () => + cancelSetup + ? Effect.succeed({ + status: "started" as const, + async: cancelAsync, + scriptId: "setup", + scriptName: "Setup", + scriptCommand: "setup", + terminalId: "setup", + cwd: "/repo-worktrees/feature", + completion: Deferred.succeed(setupEntered, undefined).pipe( + Effect.andThen(Deferred.await(allowSetup)), + Effect.as({ exitCode: 0, durationMs: 0 }), + ), + }) + : Deferred.succeed(setupEntered, undefined).pipe( + Effect.andThen(Deferred.await(allowSetup)), + Effect.andThen( + failSetup + ? Effect.fail( + new ProjectSetupScriptRunner.ProjectSetupScriptOperationError({ + threadId: "child", + worktreePath: "/repo-worktrees/feature", + operation: "writeCommand", + cause: "setup failed", + }), + ) + : Effect.succeed({ status: "no-script" as const }), + ), + ), + }); + const layerService = OrchestratorMcpService.layer.pipe( + Layer.provide(NodeCrypto.layer), + Layer.provide(harness.layer), + Layer.provide(Layer.mock(GitVcsDriver.GitVcsDriver)({ resolveCommit })), + Layer.provide(Layer.mock(ScheduledTasks.ScheduledTaskService)({})), + Layer.provide(Layer.mock(SecretRequests.SecretRequests)({})), + ); + yield* Effect.gen(function* () { + const service = yield* OrchestratorMcpService.OrchestratorMcpService; + const threads = yield* ThreadManagement.ThreadManagementService; + const parentThreadId = ThreadId.make("thread:delegated-worktree-parent"); + yield* threads.dispatch({ + type: "thread.create", + commandId: CommandId.make("create-parent"), + threadId: parentThreadId, + projectId, + title: "Parent", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: "parent-feature", + worktreePath: parentWorktree, + createdBy: "user", + creationSource: "web", + }); + yield* threads.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("start-parent"), + threadId: parentThreadId, + messageId: MessageId.make("parent-message"), + text: "Delegate work", + attachments: [], + dispatchMode: { type: "start_immediately" }, + createdBy: "user", + creationSource: "web", + }); + const scope = { + environmentId: EnvironmentId.make("delegated-worktree-test"), + requestNamespace: "parent-session", + thread: { + threadId: parentThreadId, + providerSessionId: "parent-session", + providerInstanceId: modelSelection.instanceId, + }, + client: undefined, + capabilities: new Set(["orchestration"] as const), + issuedAt: 1, + }; + const input = { + task: "Implement the child task", + target: { providerInstanceId: childInstanceId, model: "child-model" }, + isolation: "worktree" as const, + clientRequestId: "isolated-child", + }; + if (failResolve) { + const error = yield* service.delegateTask(scope, input).pipe(Effect.flip); + assert.equal(error.code, "invalid_request"); + assert.equal(error.message, "Unable to resolve the parent checkout commit."); + assert.isEmpty((yield* threads.getThreadProjection(parentThreadId)).subagents); + assert.equal(harness.createWorktree.mock.calls.length, 0); + return; + } + const delegated = yield* service.delegateTask(scope, input); + yield* Deferred.await(setupEntered); + const child = yield* threads.getThreadProjection(delegated.childThreadId); + assert.equal(child.thread.lineage.parentThreadId, parentThreadId); + assert.equal(child.thread.worktreePath, "/repo-worktrees/feature"); + assert.equal(child.runs[0]?.status, cancelAsync ? "starting" : "preparing"); + assert.equal(child.thread.modelSelection.instanceId, childInstanceId); + assert.equal(child.thread.modelSelection.model, "child-model"); + assert.deepEqual(resolveCommit.mock.calls, [ + [{ cwd: parentWorktree ?? "/repo", revision: "HEAD" }], + ]); + assert.equal(harness.createWorktree.mock.calls[0]?.[0].refName, "parent-local-commit"); + assert.equal((yield* service.delegateTask(scope, input)).taskId, delegated.taskId); + assert.equal(harness.createWorktree.mock.calls.length, 1); + assert.equal(resolveCommit.mock.calls.length, 1); + if (!cancelAsync) assert.isEmpty(child.checkpointScopes); + const parent = yield* threads.getThreadProjection(parentThreadId); + assert.equal(parent.thread.worktreePath, parentWorktree); + assert.equal(parent.thread.branch, "parent-feature"); + if (cancelSetup) { + yield* service.cancelTask(scope, { taskId: delegated.taskId }); + yield* Deferred.await(terminalClosed); + if (!cancelAsync) { + yield* threads.streamStoredEventsFrom({ threadId: delegated.childThreadId }).pipe( + Stream.filter( + (stored) => + stored.event.type === "thread.metadata-updated" && + stored.event.payload.worktreePath === null, + ), + Stream.runHead, + ); + } + assert.equal(harness.closeTerminal.mock.calls.length, 1); + assert.equal(harness.removeWorktree.mock.calls.length, cancelAsync ? 0 : 1); + const cancelled = yield* threads.getThreadProjection(delegated.childThreadId); + assert.equal(cancelled.runs[0]?.status, "interrupted"); + if (cancelAsync) { + assert.equal(cancelled.thread.worktreePath, "/repo-worktrees/feature"); + assert.equal(cancelled.checkpointScopes[0]?.cwd, "/repo-worktrees/feature"); + } else { + assert.isNull(cancelled.thread.worktreePath); + assert.isEmpty(cancelled.checkpointScopes); + } + return; + } + yield* Deferred.succeed(allowSetup, undefined); + yield* threads + .streamStoredEventsFrom({ threadId: delegated.childThreadId, eventType: "run.updated" }) + .pipe( + Stream.filter( + (stored) => + stored.event.type === "run.updated" && + stored.event.payload.status === (failSetup ? "failed" : "starting"), + ), + Stream.runHead, + ); + const prepared = yield* threads.getThreadProjection(delegated.childThreadId); + if (failSetup) { + assert.isEmpty(prepared.checkpointScopes); + } else { + assert.equal(prepared.checkpointScopes[0]?.cwd, "/repo-worktrees/feature"); + } + }).pipe(Effect.provide(Layer.merge(harness.layer, layerService))); + }), +); diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.ts index 42b1d9cef1e42302db8436a758fac01ce473575f..a4d94977097864d94e52b4d5cd609b3372701358 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.ts @@ -28,6 +28,7 @@ import * as Option from "effect/Option"; import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; import * as Scope from "effect/Scope"; +import * as Stream from "effect/Stream"; import { buildTemporaryWorktreeBranchName, flattenTemporaryWorktreeBranchName, @@ -109,7 +110,7 @@ type PreparationInput = Pick< readonly reusedWorktree?: { readonly baseRef: string }; }; -export interface ThreadLaunchRetryInput { +export interface ThreadLaunchPreparationInput { readonly commandId: CommandId; readonly threadId: ThreadId; readonly runId: RunId; @@ -180,13 +181,14 @@ export class ThreadLaunchService extends Context.Service< input: Pick, ) => Effect.Effect; /** Dispatches prepared-run.retry and prepares the run's workspace again. */ - readonly retryPreparation: (input: ThreadLaunchRetryInput) => Effect.Effect< + readonly retryPreparation: (input: ThreadLaunchPreparationInput) => Effect.Effect< Orchestrator.OrchestratorV2DispatchResult, // Dispatches prepared-run.retry, a command the enablement chokepoint // never gates; the type stays accurate anyway. | Orchestrator.OrchestratorV2Error | ThreadManagement.ThreadManagementProviderInstanceDisabledError >; + readonly prepareRun: (input: ThreadLaunchPreparationInput) => Effect.Effect; } >()("t3/orchestration-v2/ThreadLaunchService") {} @@ -317,6 +319,7 @@ const make = Effect.gen(function* () { let createdWorktreePath: string | null = null; let setupTerminalId: string | null = null; let workspaceRecorded = false; + let runReleased = false; if (input.workspaceStrategy.type === "worktree") { yield* setupTracker.begin({ threadId, @@ -623,7 +626,15 @@ const make = Effect.gen(function* () { threadId, runId, }) - .pipe(Effect.mapError(mapError(input, "release-run", threadId))); + .pipe( + Effect.mapError(mapError(input, "release-run", threadId)), + Effect.tap(() => + Effect.sync(() => { + runReleased = true; + }), + ), + Effect.uninterruptible, + ); } yield* setupTracker.stageStatus(threadId, "agent", "done"); yield* awaitAsyncSetup; @@ -637,14 +648,15 @@ const make = Effect.gen(function* () { cancelled ? "cancelled" : "failed", cancelled ? null : failureDetail(Cause.squash(cause)), ); - // A cancelled setup leaves nothing behind. A failed one keeps a worktree - // the thread recorded, so a retry reuses it, and removes one it never - // recorded, which a retry would otherwise duplicate. - if (tracked && createdWorktreePath && (cancelled || !workspaceRecorded)) { - if (setupTerminalId) - yield* terminals - .close({ threadId, terminalId: setupTerminalId, deleteHistory: true }) - .pipe(Effect.ignore); + if (cancelled && setupTerminalId) + yield* terminals + .close({ threadId, terminalId: setupTerminalId, deleteHistory: true }) + .pipe(Effect.ignore); + if ( + tracked && + createdWorktreePath && + ((cancelled && !runReleased) || !workspaceRecorded) + ) { const removedPath = createdWorktreePath; // The thread forgets the worktree only once it is gone; a failed // removal leaves the directory for the user to clean up rather than @@ -734,7 +746,33 @@ const make = Effect.gen(function* () { threadId: ThreadId, runId: RunId | null, ) { - yield* prepareInBackground(input, threadId, runId).pipe( + const preparation = Effect.gen(function* () { + const prepare = prepareInBackground(input, threadId, runId); + if ( + runId === null || + (input.workspaceStrategy.type !== "worktree" && input.reusedWorktree === undefined) + ) { + return yield* prepare; + } + const afterSequence = yield* threads.getThreadEventSequence(threadId); + const projection = yield* threads.getThreadRecords(threadId, ["runs"]); + if (projection.runs.find((run) => run.id === runId)?.status !== "preparing") return; + const interruption = threads + .streamStoredEventsFrom({ threadId, afterSequence, eventType: "run.updated" }) + .pipe( + Stream.filter( + (stored) => + stored.event.type === "run.updated" && + stored.event.payload.id === runId && + (stored.event.payload.status === "interrupted" || + stored.event.payload.status === "cancelled"), + ), + Stream.runHead, + Effect.flatMap((event) => (Option.isSome(event) ? Effect.interrupt : Effect.never)), + ); + yield* Effect.raceFirst(prepare, interruption); + }); + yield* preparation.pipe( Effect.onError((cause) => failPreparedRun( input, @@ -979,37 +1017,39 @@ const make = Effect.gen(function* () { }, ); - const retryPreparation: ThreadLaunchService["Service"]["retryPreparation"] = Effect.fn( - "ThreadLaunchService.retryPreparation", + const prepareRun: ThreadLaunchService["Service"]["prepareRun"] = Effect.fn( + "ThreadLaunchService.prepareRun", )(function* (input) { - const dispatched = yield* threads.dispatch({ - type: "prepared-run.retry", - commandId: input.commandId, - threadId: input.threadId, - runId: input.runId, - }); - // A replayed retry finds the run already past preparation, or prepared by - // the attempt that first reserved this command. - // From here the run is preparing again; anything that stops preparation - // from being scheduled must fail it, or it would wait in preparing forever. const scheduled = yield* Effect.gen(function* () { const projection = yield* threads.getThreadProjection(input.threadId); const run = projection.runs.find((candidate) => candidate.id === input.runId); const workspacePreparation = run?.workspacePreparation; if (run?.status !== "preparing" || workspacePreparation === undefined) return; if (!(yield* reservePreparation(input.commandId))) return; - yield* scheduleRetriedPreparation(input, projection, run, workspacePreparation).pipe( + yield* scheduleRunPreparation(input, projection, run, workspacePreparation).pipe( Effect.onError(() => releasePreparation(input.commandId)), ); }).pipe(Effect.exit); if (Exit.isFailure(scheduled)) { yield* failPreparedRun(input, input.threadId, input.runId, Cause.squash(scheduled.cause)); } + }); + + const retryPreparation: ThreadLaunchService["Service"]["retryPreparation"] = Effect.fn( + "ThreadLaunchService.retryPreparation", + )(function* (input) { + const dispatched = yield* threads.dispatch({ + type: "prepared-run.retry", + commandId: input.commandId, + threadId: input.threadId, + runId: input.runId, + }); + yield* prepareRun(input); return dispatched; }); - const scheduleRetriedPreparation = ( - input: ThreadLaunchRetryInput, + const scheduleRunPreparation = ( + input: ThreadLaunchPreparationInput, projection: OrchestrationV2ThreadProjection, run: OrchestrationV2ThreadProjection["runs"][number], workspacePreparation: ThreadLaunchWorkspaceStrategy, @@ -1050,7 +1090,7 @@ const make = Effect.gen(function* () { ); }; - return ThreadLaunchService.of({ launch, checkWorktreeBase, retryPreparation }); + return ThreadLaunchService.of({ launch, checkWorktreeBase, retryPreparation, prepareRun }); }); export const layer = Layer.effect(ThreadLaunchService, make); diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 8228e289a47351a598a7d4f2018441dfc7ef03c8..41eabae0aea6d77d00790ab1190e400e09c53512 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -3110,6 +3110,7 @@ export const OrchestrationV2Command = Schema.Union([ modelSelection: ModelSelection, runtimeMode: RuntimeMode, interactionMode: ProviderInteractionMode, + workspaceStrategy: Schema.optional(OrchestrationV2ThreadLaunchWorkspaceStrategy), // Omitted behaves as "settled_only" (no wake while the parent has a live // run); producers that want fire-and-forget wakes must set "always". completionWake: Schema.optional(Schema.Literals(["always", "settled_only"])), diff --git a/packages/contracts/src/orchestratorMcp.test.ts b/packages/contracts/src/orchestratorMcp.test.ts index 617442ecea9d5b3297042cbfb0e485d2ed717793..4c4a9a4a5aac92cd68235a2ea3d69c8d1cddec8a 100644 --- a/packages/contracts/src/orchestratorMcp.test.ts +++ b/packages/contracts/src/orchestratorMcp.test.ts @@ -22,6 +22,13 @@ const decodeThreadSendInput = Schema.decodeUnknownSync(OrchestratorMcpThreadSend const decodeThreadWaitInput = Schema.decodeUnknownSync(OrchestratorMcpThreadWaitInput); describe("orchestrator MCP contracts", () => { + it("accepts explicit worktree isolation and keeps shared delegation as the default", () => { + expect(decodeDelegateTaskInput({ task: "Review changes" }).isolation).toBeUndefined(); + expect( + decodeDelegateTaskInput({ task: "Review changes", isolation: "worktree" }).isolation, + ).toBe("worktree"); + expect(() => decodeDelegateTaskInput({ task: "Review changes", isolation: "copy" })).toThrow(); + }); it("decodes cross-provider delegated task requests and durable results", () => { const request = decodeDelegateTaskInput({ task: "Inspect the workspace and report the result.", diff --git a/packages/contracts/src/orchestratorMcp.ts b/packages/contracts/src/orchestratorMcp.ts index 0738426a4449caadf19f228c5d0668dc40b43a45..4641a79e3863ef6b639990960690c3320e3cef7e 100644 --- a/packages/contracts/src/orchestratorMcp.ts +++ b/packages/contracts/src/orchestratorMcp.ts @@ -174,6 +174,12 @@ export const OrchestratorMcpDelegateTaskInput = Schema.Struct({ target: Schema.optional(OrchestratorMcpTarget), title: Schema.optional(OrchestratorMcpTitle), role: Schema.optional(OrchestratorMcpTaskRole), + isolation: Schema.optional( + Schema.Literals(["shared", "worktree"]).annotate({ + description: + "Defaults to shared: use the parent's checkout. Worktree creates a separate checkout from the parent's current local commit before starting the child. Uncommitted edits are not copied.", + }), + ), mode: Schema.optional( Schema.Literals(["async", "wait"]).annotate({ description: diff --git a/packages/provider-core/src/server/orchestrationInstructions.ts b/packages/provider-core/src/server/orchestrationInstructions.ts index 6dcf3875ff250e1ca525dd95ee22a837a0c73ed4..821ada6f952fc23a57ba1f2bb5d3bc585b6469d0 100644 --- a/packages/provider-core/src/server/orchestrationInstructions.ts +++ b/packages/provider-core/src/server/orchestrationInstructions.ts @@ -6,7 +6,7 @@ export const T3_CODE_ORCHESTRATION_INSTRUCTIONS = ` The \`t3-code\` MCP server provides app-owned orchestration. Treat these concepts distinctly: -- A delegated task/subagent is child work owned by the current thread. Use \`orchestrator_capabilities\` to discover the current provider/model IDs from the same live catalog as the composer, including configured custom models. Do not treat a native tool's model list as the full list of available subagent models. Prefer native subagent tools for same-provider work only when they support the chosen model. Use \`delegate_task\` with that provider instance and model when native tools cannot, including for same-provider work. Also use \`delegate_task\` for cross-provider or explicitly T3-owned child tasks. Retain each returned \`taskId\`, and use \`task_status\` or \`task_cancel\` to manage it. The returned \`childThreadId\` is backing storage for the subagent, not the target for starting another delegated review round. +- A delegated task/subagent is child work owned by the current thread. Use \`orchestrator_capabilities\` to discover the current provider/model IDs from the same live catalog as the composer, including configured custom models. Do not treat a native tool's model list as the full list of available subagent models. Prefer native subagent tools for same-provider work only when they support the chosen model. Use \`delegate_task\` with that provider instance and model when native tools cannot, including for same-provider work. Also use \`delegate_task\` for cross-provider or explicitly T3-owned child tasks. For a child task that needs a separate checkout, use \`delegate_task\` with \`isolation:"worktree"\` across any provider or model. It starts from the parent local commit and runs project setup; uncommitted edits are not copied. Retain each returned \`taskId\`, and use \`task_status\` or \`task_cancel\` to manage it. The returned \`childThreadId\` is backing storage for the subagent, not the target for starting another delegated review round. - \`t3_thread_launch\` and \`create_threads\` create ordinary top-level T3 conversations. Use them only when the user explicitly asks for separate/new/top-level threads or conversations. Never use them merely because the user said "subagent" or requested parallel delegated work. - For every T3 delegated review round, call \`delegate_task\` again. Include the original brief, prior findings, responses, and unresolved objections in each new task prompt. Track each round by its own \`taskId\`. Use a distinct \`clientRequestId\` per round, stable across retries of that round. Do not use \`t3_thread_send\` on \`childThreadId\` to continue a delegated review. - \`schedule_task\` creates persistent recurring work in the app scheduler. Pass \`schedule\` as a structured object, never as JSON text: \`{"type":"interval","everyMs":3600000}\` for an interval, or \`{"type":"fixed_time","timeOfDay":"09:00","weekdays":[1,2,3,4,5]}\` for a wall-clock schedule, or \`{"type":"webhook"}\` to run on each request to the returned \`webhookUrl\` (the run sees the request only through \`{{body.path}}\`-style placeholders in the prompt). By default runs return to the current thread, which suits orchestrating: each trigger arrives here and you delegate or dedupe; set \`bindToCurrentThread=false\` only when the user wants a fresh thread for every run. After scheduling a timer, report the returned cadence and next run time; for a webhook, report its \`webhookUrl\`, or say T3 Connect remote access is needed if it is missing. -- 2.54.0