authorgravatar for 62337003+Bil0000@users.noreply.github.comBilal Bakr <62337003+Bil0000@users.noreply.github.com> 2026-10-14 03:39:02+03:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-13 19:39:02-05:00
log25629a76a0458bdeb707483722894158af090b41
treed51ab305c1b78d6a1e5994d53af98eef70f5f2c3
parent3feae3a94077317e78b7fd4591454ed68f9d4fd2
signature Signed by PGP key B5690EEEBB952194

feat(server): run delegated tasks in separate worktrees (#19102)

15 files changed, 454 insertions(+), 80 deletions(-)

apps/server/src/mcp/OrchestratorMcpService.activity.test.ts+11-4
......@@ -22,10 +22,17 @@ import * as ProjectService from "../project/ProjectService.ts";
2222import * as ScheduledTaskService from "../scheduledTasks/ScheduledTaskService.ts";
2323import * as ServerSettings from "../serverSettings.ts";
2424import * as SecretRequests from "../secrets/SecretRequests.ts";
25import * as ThreadLaunchService from "../orchestration-v2/ThreadLaunchService.ts";
26import * as GitVcsDriver from "../vcs/GitVcsDriver.ts";
2527import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts";
2628import type * as McpInvocationContext from "./McpInvocationContext.ts";
2729import * as OrchestratorMcpService from "./OrchestratorMcpService.ts";
2830
31const layerOrchestratorService = OrchestratorMcpService.layer.pipe(
32 Layer.provide(Layer.mock(ThreadLaunchService.ThreadLaunchService)({})),
33 Layer.provide(Layer.mock(GitVcsDriver.GitVcsDriver)({})),
34);
35
2936const environmentId = EnvironmentId.make("environment-mcp-orchestrator-detail");
3037const projectId = ProjectId.make("project-mcp-orchestrator-detail");
3138const parentThreadId = ThreadId.make("thread-mcp-orchestrator-parent");
......@@ -127,7 +134,7 @@ it("readThread prefers activity-run status over a newer cancelled queued run", a
127134 updatedAt: now,
128135 } as unknown as OrchestrationV2ThreadProjection;
129136
130 const layer = OrchestratorMcpService.layer.pipe(
137 const layer = layerOrchestratorService.pipe(
131138 Layer.provide(
132139 Layer.mergeAll(
133140 Layer.mock(ThreadManagementService.ThreadManagementService)({
......@@ -193,7 +200,7 @@ it("readThread prefers waiting activity status over a newer cancelled queued run
193200 updatedAt: now,
194201 } as unknown as OrchestrationV2ThreadProjection;
195202
196 const layer = OrchestratorMcpService.layer.pipe(
203 const layer = layerOrchestratorService.pipe(
197204 Layer.provide(
198205 Layer.mergeAll(
199206 Layer.mock(ThreadManagementService.ThreadManagementService)({
......@@ -316,7 +323,7 @@ it("taskStatus returns task.providerInstanceId rather than the driver kind", asy
316323 updatedAt: now,
317324 } as unknown as OrchestrationV2ThreadProjection;
318325
319 const layer = OrchestratorMcpService.layer.pipe(
326 const layer = layerOrchestratorService.pipe(
320327 Layer.provide(
321328 Layer.mergeAll(
322329 Layer.mock(ThreadManagementService.ThreadManagementService)({
......@@ -415,7 +422,7 @@ it("readThread and sendToThread reach threads in other projects", async () => {
415422 updatedAt: now,
416423 }) as unknown as OrchestrationV2ThreadProjection;
417424
418 const layer = OrchestratorMcpService.layer.pipe(
425 const layer = layerOrchestratorService.pipe(
419426 Layer.provide(
420427 Layer.mergeAll(
421428 Layer.mock(ThreadManagementService.ThreadManagementService)({
apps/server/src/mcp/OrchestratorMcpService.test.ts+27-28
......@@ -29,6 +29,8 @@ import {
2929 OrchestratorThreadAboveModeLimitError,
3030} from "../orchestration-v2/Orchestrator.ts";
3131import * as ProviderAdapterRegistry from "../orchestration-v2/ProviderAdapterRegistry.ts";
32import * as ThreadLaunchService from "../orchestration-v2/ThreadLaunchService.ts";
33import * as GitVcsDriver from "../vcs/GitVcsDriver.ts";
3234import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts";
3335import * as ProviderRegistry from "../provider/ProviderRegistry.ts";
3436import { buildUnavailableProviderSnapshot } from "../provider/unavailableProviderSnapshot.ts";
......@@ -41,6 +43,11 @@ import { idleThreadProjection, liveThreadShell } from "./McpToolAccess.testkit.t
4143import * as OrchestratorMcpService from "./OrchestratorMcpService.ts";
4244import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
4345
46const layerOrchestratorService = OrchestratorMcpService.layer.pipe(
47 Layer.provide(Layer.mock(ThreadLaunchService.ThreadLaunchService)({})),
48 Layer.provide(Layer.mock(GitVcsDriver.GitVcsDriver)({})),
49);
50
4451describe("OrchestratorMcpService", () => {
4552 it.effect("retries terminal acknowledgement with a fresh command id", () =>
4653 Effect.gen(function* () {
......@@ -170,7 +177,7 @@ describe("OrchestratorMcpService", () => {
170177 const commandIds = yield* Ref.get(acknowledgementCommandIds);
171178 assert.equal(commandIds.length, 2);
172179 assert.notEqual(commandIds[0], commandIds[1]);
173 }).pipe(Effect.provide(OrchestratorMcpService.layer.pipe(Layer.provide(layerDependencies))));
180 }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies))));
174181 }),
175182 );
176183
......@@ -267,7 +274,7 @@ describe("OrchestratorMcpService", () => {
267274 const settled = yield* service.taskStatus(scope, taskId);
268275 assert.equal(settled.status, "cancelled");
269276 assert.equal(yield* Ref.get(dispatched), 1);
270 }).pipe(Effect.provide(OrchestratorMcpService.layer.pipe(Layer.provide(layerDependencies))));
277 }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies))));
271278 }),
272279 );
273280
......@@ -343,7 +350,7 @@ describe("OrchestratorMcpService", () => {
343350 .pipe(Effect.flip);
344351 assert.equal(error.code, "task_not_cancellable");
345352 assert.deepEqual(yield* Ref.get(dispatched), []);
346 }).pipe(Effect.provide(OrchestratorMcpService.layer.pipe(Layer.provide(layerDependencies))));
353 }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies))));
347354 }),
348355 );
349356
......@@ -423,7 +430,7 @@ describe("OrchestratorMcpService", () => {
423430 (yield* Ref.get(dispatched)).map((command) => (command as { type: string }).type),
424431 ["thread.stop"],
425432 );
426 }).pipe(Effect.provide(OrchestratorMcpService.layer.pipe(Layer.provide(layerDependencies))));
433 }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies))));
427434 }),
428435 );
429436
......@@ -509,7 +516,7 @@ describe("OrchestratorMcpService", () => {
509516 (yield* Ref.get(dispatched)).map((command) => (command as { type: string }).type),
510517 ["thread.stop", "delegated_task.completion-delivery.dispose"],
511518 );
512 }).pipe(Effect.provide(OrchestratorMcpService.layer.pipe(Layer.provide(layerDependencies))));
519 }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies))));
513520 }),
514521 );
515522
......@@ -592,7 +599,7 @@ describe("OrchestratorMcpService", () => {
592599 .pipe(Effect.flip);
593600 assert.equal(error.code, "runtime_mode_escalation_denied");
594601 assert.deepEqual(yield* Ref.get(dispatched), []);
595 }).pipe(Effect.provide(OrchestratorMcpService.layer.pipe(Layer.provide(layerDependencies))));
602 }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies))));
596603 }),
597604 );
598605
......@@ -736,7 +743,7 @@ describe("OrchestratorMcpService", () => {
736743 dispatched: yield* Ref.get(dispatched),
737744 stoppedBelow: yield* Ref.get(stoppedBelow),
738745 };
739 }).pipe(Effect.provide(OrchestratorMcpService.layer.pipe(Layer.provide(layerDependencies))));
746 }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies))));
740747 });
741748
742749 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", () => {
10191026 assert.isTrue(
10201027 fork!.constraints.includes("Driver 'forkOnly' is not registered in this build."),
10211028 );
1022 }).pipe(
1023 Effect.provide(OrchestratorMcpService.layer.pipe(Layer.provide(layerDependencies))),
1024 );
1029 }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies))));
10251030 }),
10261031 );
10271032
......@@ -1066,7 +1071,7 @@ describe("OrchestratorMcpService provider resolution", () => {
10661071 capabilities.providers[0]?.models.map((model) => model.id),
10671072 ["gpt-5.4", "my-model"],
10681073 );
1069 }).pipe(Effect.provide(OrchestratorMcpService.layer.pipe(Layer.provide(layerDependencies))));
1074 }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies))));
10701075 }),
10711076 );
10721077
......@@ -1166,9 +1171,7 @@ describe("OrchestratorMcpService provider resolution", () => {
11661171 assert.equal(request.type, "delegated_task.request");
11671172 assert.equal(request.modelSelection.instanceId, antigravityInstanceId);
11681173 assert.equal(request.modelSelection.model, "ant-model");
1169 }).pipe(
1170 Effect.provide(OrchestratorMcpService.layer.pipe(Layer.provide(layerDependencies))),
1171 );
1174 }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies))));
11721175 }),
11731176 );
11741177
......@@ -1264,7 +1267,7 @@ describe("OrchestratorMcpService provider resolution", () => {
12641267 };
12651268 assert.equal(request.modelSelection.instanceId, antigravityInstanceId);
12661269 assert.equal(request.modelSelection.model, "ant-model");
1267 }).pipe(Effect.provide(OrchestratorMcpService.layer.pipe(Layer.provide(layerDependencies))));
1270 }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies))));
12681271 }),
12691272 );
12701273
......@@ -1323,7 +1326,7 @@ describe("OrchestratorMcpService provider resolution", () => {
13231326 assert.isTrue(
13241327 byDriver.message.includes("No V2 provider adapter is registered for driver forkOnly."),
13251328 );
1326 }).pipe(Effect.provide(OrchestratorMcpService.layer.pipe(Layer.provide(layerDependencies))));
1329 }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies))));
13271330 }),
13281331 );
13291332
......@@ -1369,7 +1372,7 @@ describe("OrchestratorMcpService provider resolution", () => {
13691372 })
13701373 .pipe(Effect.flip);
13711374 assert.equal(error.code, "invalid_request");
1372 }).pipe(Effect.provide(OrchestratorMcpService.layer.pipe(Layer.provide(layerDependencies))));
1375 }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies))));
13731376 }),
13741377 );
13751378
......@@ -1474,7 +1477,7 @@ describe("OrchestratorMcpService provider resolution", () => {
14741477 modelSelection: { instanceId: string; model: string };
14751478 };
14761479 assert.equal(request.modelSelection.instanceId, altInstanceId);
1477 }).pipe(Effect.provide(OrchestratorMcpService.layer.pipe(Layer.provide(layerDependencies))));
1480 }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies))));
14781481 }),
14791482 );
14801483
......@@ -1599,9 +1602,7 @@ describe("OrchestratorMcpService provider resolution", () => {
15991602 modelSelection: { instanceId: string; model: string };
16001603 };
16011604 assert.equal(retryCommand.modelSelection.instanceId, codexInstanceId);
1602 }).pipe(
1603 Effect.provide(OrchestratorMcpService.layer.pipe(Layer.provide(layerDependencies))),
1604 );
1605 }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies))));
16051606 }),
16061607 );
16071608
......@@ -1706,7 +1707,7 @@ describe("OrchestratorMcpService provider resolution", () => {
17061707 assert.equal(result.providerInstanceId, claudeInstanceId);
17071708 assert.equal(yield* Ref.get(probes), 2);
17081709 assert.equal(yield* Ref.get(dispatched), 1);
1709 }).pipe(Effect.provide(OrchestratorMcpService.layer.pipe(Layer.provide(layerDependencies))));
1710 }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies))));
17101711 }),
17111712 );
17121713
......@@ -1917,9 +1918,7 @@ describe("OrchestratorMcpService provider resolution", () => {
19171918 } else {
19181919 assert.equal(request.modelSelection.model, "codex-alt-model", testCase.name);
19191920 }
1920 }).pipe(
1921 Effect.provide(OrchestratorMcpService.layer.pipe(Layer.provide(layerDependencies))),
1922 );
1921 }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies))));
19231922 }
19241923 }),
19251924 );
......@@ -1972,7 +1971,7 @@ describe("OrchestratorMcpService provider resolution", () => {
19721971 boundThread: OrchestrationV2ThreadShell | null,
19731972 upserted: Ref.Ref<number>,
19741973 ) =>
1975 OrchestratorMcpService.layer.pipe(
1974 layerOrchestratorService.pipe(
19761975 Layer.provide(
19771976 Layer.mergeAll(
19781977 NodeServices.layer,
......@@ -2040,7 +2039,7 @@ describe("OrchestratorMcpService provider resolution", () => {
20402039 ),
20412040 ),
20422041 Effect.provide(
2043 OrchestratorMcpService.layer.pipe(
2042 layerOrchestratorService.pipe(
20442043 Layer.provide(
20452044 Layer.mergeAll(
20462045 NodeServices.layer,
......@@ -2123,7 +2122,7 @@ describe("OrchestratorMcpService provider resolution", () => {
21232122 // The edit's own check reads the thread; the read after the save fails.
21242123 const mcp = yield* OrchestratorMcpService.OrchestratorMcpService.pipe(
21252124 Effect.provide(
2126 OrchestratorMcpService.layer.pipe(
2125 layerOrchestratorService.pipe(
21272126 Layer.provide(
21282127 Layer.mergeAll(
21292128 NodeServices.layer,
apps/server/src/mcp/OrchestratorMcpService.ts+39
......@@ -10,6 +10,7 @@ import {
1010 type OrchestrationV2Subagent,
1111 type OrchestrationV2ThreadProjection,
1212 type OrchestrationV2ThreadShell,
13 type OrchestrationV2ThreadLaunchWorkspaceStrategy,
1314 type OrchestrationV2TurnItem,
1415 OrchestratorMcpFailure,
1516 type OrchestratorMcpCapabilitiesResult,
......@@ -81,6 +82,8 @@ import {
8182 type DispatchModeRefusal,
8283} from "../orchestration-v2/DispatchModeLimit.ts";
8384import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts";
85import * as ThreadLaunchService from "../orchestration-v2/ThreadLaunchService.ts";
86import * as GitVcsDriver from "../vcs/GitVcsDriver.ts";
8487import { isSnoozed } from "../orchestration-v2/ThreadSettlementService.ts";
8588import * as ProjectService from "../project/ProjectService.ts";
8689import * as ProviderRegistry from "../provider/ProviderRegistry.ts";
......@@ -904,6 +907,8 @@ const make = Effect.gen(function* () {
904907 const providerAdapters = yield* ProviderAdapterRegistry.ProviderAdapterRegistryV2;
905908 const scheduledTasks = yield* ScheduledTaskService.ScheduledTaskService;
906909 const projects = yield* ProjectService.ProjectService;
910 const threadLaunch = yield* ThreadLaunchService.ThreadLaunchService;
911 const gitDriver = yield* GitVcsDriver.GitVcsDriver;
907912 const serverSettings = yield* ServerSettings.ServerSettingsService;
908913
909914 /** A caller-named project, which must exist before anything is recorded against it. */
......@@ -1993,6 +1998,21 @@ const make = Effect.gen(function* () {
19931998 parent.thread.interactionMode,
19941999 input.interactionMode,
19952000 );
2001 let workspaceStrategy: OrchestrationV2ThreadLaunchWorkspaceStrategy | undefined;
2002 if (input.isolation === "worktree" && !isRetryOfAcceptedDelegation) {
2003 const project = yield* requireProject(parent.thread.projectId);
2004 const base = yield* gitDriver
2005 .resolveCommit({
2006 cwd: parent.thread.worktreePath ?? project.workspaceRoot,
2007 revision: "HEAD",
2008 })
2009 .pipe(
2010 Effect.mapError(() =>
2011 failure("invalid_request", "Unable to resolve the parent checkout commit."),
2012 ),
2013 );
2014 workspaceStrategy = { type: "worktree", baseRef: base.commitSha, startFromOrigin: false };
2015 }
19962016 // ThreadManagementService.dispatch is the chokepoint for this command
19972017 // type's own enablement check, using the same command id and its
19982018 // receipt-replay exemption, so no separate assert runs here.
......@@ -2010,6 +2030,7 @@ const make = Effect.gen(function* () {
20102030 modelSelection: target.modelSelection,
20112031 runtimeMode,
20122032 interactionMode,
2033 ...(workspaceStrategy === undefined ? {} : { workspaceStrategy }),
20132034 // Async delegations wake the parent on every child terminal; wait
20142035 // delegations deliver through the blocking tool call, so a wake is
20152036 // only needed if the parent settled first (timeout, disconnect).
......@@ -2036,6 +2057,22 @@ const make = Effect.gen(function* () {
20362057 );
20372058 }
20382059 const taskId = taskEvent.event.payload.id;
2060 const childThreadId = taskEvent.event.payload.childThreadId;
2061 const childRunEvent = result.storedEvents.find(
2062 (stored) =>
2063 stored.event.type === "run.created" && stored.event.threadId === childThreadId,
2064 );
2065 if (
2066 childThreadId !== null &&
2067 childRunEvent?.event.type === "run.created" &&
2068 childRunEvent.event.payload.workspacePreparation !== undefined
2069 ) {
2070 yield* threadLaunch.prepareRun({
2071 commandId,
2072 threadId: childThreadId,
2073 runId: childRunEvent.event.payload.id,
2074 });
2075 }
20392076
20402077 if (input.mode !== "wait") {
20412078 return yield* readTask(scope, taskId, false, true);
......@@ -2660,6 +2697,8 @@ export const layer: Layer.Layer<
26602697 never,
26612698 | Crypto.Crypto
26622699 | ThreadManagementService.ThreadManagementService
2700 | ThreadLaunchService.ThreadLaunchService
2701 | GitVcsDriver.GitVcsDriver
26632702 | ProviderRegistry.ProviderRegistry
26642703 | ProviderAdapterRegistry.ProviderAdapterRegistryV2
26652704 | ScheduledTaskService.ScheduledTaskService
apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts+9-2
......@@ -57,6 +57,8 @@ import { threadShellFromProjection } from "../orchestration-v2/ProjectionStore.t
5757import * as CommandReceiptStore from "../orchestration-v2/CommandReceiptStore.ts";
5858import * as EventSink from "../orchestration-v2/EventSink.ts";
5959import * as Orchestrator from "../orchestration-v2/Orchestrator.ts";
60import * as ThreadLaunchService from "../orchestration-v2/ThreadLaunchService.ts";
61import * as GitVcsDriver from "../vcs/GitVcsDriver.ts";
6062import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts";
6163import * as ServerSettings from "../serverSettings.ts";
6264import * as SqlitePersistence from "../persistence/Sqlite.ts";
......@@ -80,6 +82,11 @@ import * as McpInvocationContext from "./McpInvocationContext.ts";
8082import { delegatedTaskRun, hasPendingChildRuns } from "./OrchestratorMcpService.ts";
8183import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
8284
85const layerOrchestratorToolkit = McpHttpServer.layerOrchestratorToolkit.pipe(
86 Layer.provide(Layer.mock(ThreadLaunchService.ThreadLaunchService)({})),
87 Layer.provide(Layer.mock(GitVcsDriver.GitVcsDriver)({})),
88);
89
8390// Effect returns a declared tool failure as `isError` with its encoded payload
8491// as JSON text, never as `structuredContent`.
8592const declaredFailure = (result: McpSchema.CallToolResult) => {
......@@ -682,7 +689,7 @@ describe("orchestrator MCP toolkit", () => {
682689 }),
683690 );
684691 const layerTest = Layer.merge(
685 McpHttpServer.layerOrchestratorToolkit,
692 layerOrchestratorToolkit,
686693 McpHttpServer.layerThreadToolkit,
687694 ).pipe(
688695 Layer.provideMerge(McpServer.McpServer.layer),
......@@ -3838,7 +3845,7 @@ describe("orchestrator MCP toolkit", () => {
38383845 model: codexModel,
38393846 }),
38403847 ]);
3841 const layerTest = McpHttpServer.layerOrchestratorToolkit.pipe(
3848 const layerTest = layerOrchestratorToolkit.pipe(
38423849 Layer.provideMerge(McpServer.McpServer.layer),
38433850 Layer.provideMerge(layerOrchestration),
38443851 Layer.provide(
apps/server/src/mcp/TaskCancelNativeSubagent.integration.test.ts+8-1
......@@ -27,6 +27,8 @@ import * as EffectWorker from "../orchestration-v2/EffectWorker.ts";
2727import * as Orchestrator from "../orchestration-v2/Orchestrator.ts";
2828import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
2929import * as ProviderAdapterRegistry from "../orchestration-v2/ProviderAdapterRegistry.ts";
30import * as ThreadLaunchService from "../orchestration-v2/ThreadLaunchService.ts";
31import * as GitVcsDriver from "../vcs/GitVcsDriver.ts";
3032import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts";
3133import * as ServerSettings from "../serverSettings.ts";
3234import * as ProviderReplayHarness from "../orchestration-v2/testkit/ProviderReplayHarness.ts";
......@@ -39,6 +41,11 @@ import * as SecretRequests from "../secrets/SecretRequests.ts";
3941import type { McpInvocationScope } from "./McpInvocationContext.ts";
4042import * as OrchestratorMcpService from "./OrchestratorMcpService.ts";
4143
44const layerOrchestratorService = OrchestratorMcpService.layer.pipe(
45 Layer.provide(Layer.mock(ThreadLaunchService.ThreadLaunchService)({})),
46 Layer.provide(Layer.mock(GitVcsDriver.GitVcsDriver)({})),
47);
48
4249const driver = ProviderDriverKind.make("claudeAgent");
4350const instanceId = ProviderInstanceId.make("claudeAgent");
4451const modelSelection = { instanceId, model: "claude-test" };
......@@ -143,7 +150,7 @@ it.effect("task_cancel ends a delegated child's background native subagent", ()
143150 ),
144151 ),
145152 );
146 const layerTest = OrchestratorMcpService.layer.pipe(
153 const layerTest = layerOrchestratorService.pipe(
147154 Layer.provideMerge(layerOrchestration),
148155 Layer.provide(
149156 Layer.mergeAll(
apps/server/src/mcp/toolkits/core.test.ts+11-4
......@@ -35,6 +35,8 @@ import {
3535import * as ServerSecretStore from "../../auth/ServerSecretStore.ts";
3636import * as ServerConfig from "../../config.ts";
3737import * as ProviderAdapterRegistry from "../../orchestration-v2/ProviderAdapterRegistry.ts";
38import * as ThreadLaunchService from "../../orchestration-v2/ThreadLaunchService.ts";
39import * as GitVcsDriver from "../../vcs/GitVcsDriver.ts";
3840import * as ThreadManagement from "../../orchestration-v2/ThreadManagementService.ts";
3941import * as ThreadSearch from "../../orchestration-v2/ThreadSearch.ts";
4042import * as PreviewBrowser from "../../preview/PreviewBrowser.ts";
......@@ -59,6 +61,11 @@ import { ThreadToolkit } from "./thread/tools.ts";
5961import { WorktreeToolkit } from "./worktree/tools.ts";
6062import { DeviceToolkit } from "./device/tools.ts";
6163
64const layerOrchestratorToolkit = McpHttpServer.layerOrchestratorToolkit.pipe(
65 Layer.provide(Layer.mock(ThreadLaunchService.ThreadLaunchService)({})),
66 Layer.provide(Layer.mock(GitVcsDriver.GitVcsDriver)({})),
67);
68
6269// Effect returns a declared tool failure as `isError` with its encoded payload
6370// as JSON text, never as `structuredContent`.
6471const declaredFailure = (result: McpSchema.CallToolResult) => {
......@@ -578,7 +585,7 @@ it.effect("refuses act-as-caller tools to a client caller", () =>
578585 expect(declaredFailure(result)).toMatchObject({ code: "thread_credential_required" });
579586 }).pipe(
580587 Effect.provide(
581 McpHttpServer.layerOrchestratorToolkit.pipe(
588 layerOrchestratorToolkit.pipe(
582589 Layer.provideMerge(McpServer.McpServer.layer),
583590 Layer.provide(NodeCrypto.layer),
584591 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
617624 expect(allowed.isError).toBe(false);
618625 }).pipe(
619626 Effect.provide(
620 McpHttpServer.layerOrchestratorToolkit.pipe(
627 layerOrchestratorToolkit.pipe(
621628 Layer.provideMerge(McpServer.McpServer.layer),
622629 Layer.provide(NodeCrypto.layer),
623630 Layer.provide(Layer.mock(ThreadManagement.ThreadManagementService)({})),
......@@ -692,7 +699,7 @@ it.effect.each(["t3_thread_interrupt", "t3_workflow_stop"])(
692699 expect(declaredFailure(result)).toMatchObject({ code: "runtime_mode_escalation_denied" });
693700 }).pipe(
694701 Effect.provide(
695 McpHttpServer.layerOrchestratorToolkit.pipe(
702 layerOrchestratorToolkit.pipe(
696703 Layer.provideMerge(McpServer.McpServer.layer),
697704 Layer.provide(NodeCrypto.layer),
698705 Layer.provide(
......@@ -796,7 +803,7 @@ it.effect("discovers workflow stop targets and requires orchestration capability
796803 expect(stopped).toEqual([{ threadId: "workflow-parent", subagentId: "workflow-coordinator" }]);
797804 }).pipe(
798805 Effect.provide(
799 McpHttpServer.layerOrchestratorToolkit.pipe(
806 layerOrchestratorToolkit.pipe(
800807 Layer.provideMerge(McpServer.McpServer.layer),
801808 Layer.provide(NodeCrypto.layer),
802809 Layer.provide(
apps/server/src/mcp/toolkits/orchestrator/tools.ts+1-1
......@@ -65,7 +65,7 @@ const OrchestratorCapabilitiesTool = Tool.make("orchestrator_capabilities", {
6565
6666export const DelegateTaskTool = Tool.make("delegate_task", {
6767 description:
68 "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.",
68 "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.",
6969 parameters: OrchestratorMcpDelegateTaskInput,
7070 success: OrchestratorMcpDelegateTaskResult,
7171 failure: OrchestratorMcpFailure,
apps/server/src/orchestration-v2/Orchestrator.ts+5-1
......@@ -6666,6 +6666,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
66666666 }),
66676667 runtimeMode: command.runtimeMode,
66686668 interactionMode: command.interactionMode,
6669 ...(command.workspaceStrategy === undefined ? {} : { branch: null, worktreePath: null }),
66696670 };
66706671 const task: OrchestrationV2Subagent = {
66716672 id: taskNodeId,
......@@ -6783,7 +6784,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
67836784 text: command.task,
67846785 attachments: [],
67856786 modelSelection: command.modelSelection,
6786 dispatchMode: { type: "start_immediately" },
6787 dispatchMode:
6788 command.workspaceStrategy === undefined
6789 ? { type: "start_immediately" }
6790 : { type: "defer_start", workspaceStrategy: command.workspaceStrategy },
67876791 } satisfies Extract<OrchestrationV2Command, { readonly type: "message.dispatch" }>;
67886792 yield* dispatchMessage(childMessageCommand, events, effects);
67896793
apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts+32-8
......@@ -643,7 +643,7 @@ it.effect("closes a secret request card's form when its run is recovered", () =>
643643 }).pipe(Effect.provide(layer));
644644});
645645
646it.effect("holds accepted queued work without cancelling its execution state after restart", () => {
646it.effect.each(["queued", "preparing"] as const)("recovers $0 work after restart", (status) => {
647647 const threadId = ThreadId.make("thread_queued_restart");
648648 const runId = RunId.make("run_queued_restart");
649649 const attemptId = RunAttemptId.make("attempt_queued_restart");
......@@ -659,9 +659,18 @@ it.effect("holds accepted queued work without cancelling its execution state aft
659659 runs: [
660660 {
661661 id: runId,
662 status: "queued",
662 status,
663663 queuePosition: 1,
664664 providerInstanceId: ProviderInstanceId.make("codex"),
665 ...(status === "preparing"
666 ? {
667 workspacePreparation: {
668 type: "worktree",
669 baseRef: "parent-local-commit",
670 startFromOrigin: false,
671 },
672 }
673 : {}),
665674 },
666675 ],
667676 attempts: [
......@@ -712,18 +721,33 @@ it.effect("holds accepted queued work without cancelling its execution state aft
712721 return Effect.gen(function* () {
713722 const summary =
714723 yield* (yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService).reconcile("startup");
715 assert.equal(summary.terminalizedRuns, 0);
724 assert.equal(summary.terminalizedRuns, status === "preparing" ? 1 : 0);
716725 const command = committedInput;
717726 assert.isNotNull(command);
718727 if (command === null) return;
719728 const runEvent = command.events.find((event) => event.type === "run.updated");
720729 const attemptEvent = command.events.find((event) => event.type === "run-attempt.updated");
721730 const nodeEvent = command.events.find((event) => event.type === "node.updated");
722 assert.equal(runEvent?.type === "run.updated" ? runEvent.payload.status : null, "queued");
723 assert.equal(runEvent?.type === "run.updated" ? runEvent.payload.queuePosition : null, 1);
724 assert.equal(runEvent?.type === "run.updated" ? runEvent.payload.queueHeld : false, true);
725 assert.isUndefined(attemptEvent);
726 assert.isUndefined(nodeEvent);
731 assert.equal(
732 runEvent?.type === "run.updated" ? runEvent.payload.status : null,
733 status === "preparing" ? "cancelled" : "queued",
734 );
735 assert.isEmpty(command.effects);
736 if (status === "preparing") {
737 assert.equal(
738 attemptEvent?.type === "run-attempt.updated" ? attemptEvent.payload.status : null,
739 "cancelled",
740 );
741 assert.equal(
742 nodeEvent?.type === "node.updated" ? nodeEvent.payload.status : null,
743 "cancelled",
744 );
745 } else {
746 assert.equal(runEvent?.type === "run.updated" ? runEvent.payload.queuePosition : null, 1);
747 assert.equal(runEvent?.type === "run.updated" ? runEvent.payload.queueHeld : false, true);
748 assert.isUndefined(attemptEvent);
749 assert.isUndefined(nodeEvent);
750 }
727751 }).pipe(Effect.provide(layer));
728752});
729753
apps/server/src/orchestration-v2/ThreadLaunchService.test.ts+228-2
......@@ -15,6 +15,7 @@ import {
1515 ComposerContextId,
1616 type ChatAttachment,
1717 CommandId,
18 EnvironmentId,
1819 DEFAULT_SERVER_SETTINGS,
1920 GitCommandError,
2021 MessageId,
......@@ -56,6 +57,8 @@ import * as EffectOutbox from "./EffectOutbox.ts";
5657import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
5758import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts";
5859import * as ThreadLaunch from "./ThreadLaunchService.ts";
60import * as OrchestratorMcpService from "../mcp/OrchestratorMcpService.ts";
61import * as GitVcsDriver from "../vcs/GitVcsDriver.ts";
5962import * as ThreadManagement from "./ThreadManagementService.ts";
6063import * as ThreadTitleRegeneration from "./ThreadTitleRegenerationService.ts";
6164import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts";
......@@ -97,6 +100,8 @@ const adapter = {
97100} as ProviderAdapter.ProviderAdapterV2["Service"];
98101
99102interface HarnessOptions {
103 readonly closeTerminal?: TerminalManager.TerminalManager["Service"]["close"];
104 readonly adapters?: ReadonlyArray<ProviderAdapter.ProviderAdapterV2["Service"]>;
100105 readonly managedFolders?: Layer.Layer<ManagedProjectFolders.ManagedProjectFolders>;
101106 readonly createWorktree?: GitWorkflow.GitWorkflowService["Service"]["createWorktree"];
102107 readonly fetchRemote?: GitWorkflow.GitWorkflowService["Service"]["fetchRemote"];
......@@ -112,7 +117,7 @@ interface HarnessOptions {
112117
113118function makeHarness(options: HarnessOptions = {}) {
114119 const layerDatabase = SqlitePersistence.layerMemory;
115 const layerRegistry = ProviderAdapterRegistry.layerFromAdapters([adapter]);
120 const layerRegistry = ProviderAdapterRegistry.layerFromAdapters(options.adapters ?? [adapter]);
116121 const layerOrchestrator = ProviderReplayHarness.layerWithRegistry(
117122 { name: "thread-launch" },
118123 layerRegistry,
......@@ -143,6 +148,7 @@ function makeHarness(options: HarnessOptions = {}) {
143148 const runSetup = vi.fn(
144149 options.runSetup ?? (() => Effect.succeed({ status: "no-script" as const })),
145150 );
151 const closeTerminal = vi.fn(options.closeTerminal ?? (() => Effect.void));
146152 const generateBranchName = vi.fn(
147153 options.generateBranchName ?? (() => Effect.succeed({ branch: "generated-branch" })),
148154 );
......@@ -152,7 +158,7 @@ function makeHarness(options: HarnessOptions = {}) {
152158 const layerExternalServices = Layer.mergeAll(
153159 WorktreeSetupTracker.layer,
154160 Layer.mock(ProjectCloneTracker.ProjectCloneTracker)({ get: () => Effect.succeed(null) }),
155 Layer.mock(TerminalManager.TerminalManager)({ close: () => Effect.void }),
161 Layer.mock(TerminalManager.TerminalManager)({ close: closeTerminal }),
156162 Layer.succeed(ProjectService.ProjectService, {
157163 create: () => Effect.die("unused"),
158164 bootstrap: () => Effect.die("unused"),
......@@ -237,6 +243,7 @@ function makeHarness(options: HarnessOptions = {}) {
237243 return {
238244 layer: Layer.mergeAll(
239245 layerLaunch,
246 layerRegistry,
240247 layerThreadManagement,
241248 layerTitleRegeneration,
242249 layerOutbox,
......@@ -249,6 +256,7 @@ function makeHarness(options: HarnessOptions = {}) {
249256 generateBranchName,
250257 generateThreadTitle,
251258 runSetup,
259 closeTerminal,
252260 };
253261}
254262
......@@ -2263,6 +2271,7 @@ it.effect("shared intake preserves durable attachment bytes after a lost launch
22632271 ),
22642272 checkWorktreeBase: launches.checkWorktreeBase,
22652273 retryPreparation: launches.retryPreparation,
2274 prepareRun: launches.prepareRun,
22662275 }),
22672276 Effect.flip,
22682277 );
......@@ -2705,3 +2714,220 @@ it.effect("replays an accepted message.dispatch after the project disables the i
27052714 }
27062715 }).pipe(Effect.provide(Layer.mergeAll(harness.layer, layerFiles)));
27072716});
2717
2718it.effect.each(
2719 (["codex", "claudeAgent"] as const).flatMap((driver) =>
2720 [
2721 { failSetup: false, failResolve: false },
2722 { failSetup: true, failResolve: false },
2723 { failSetup: false, failResolve: true },
2724 { failSetup: false, failResolve: false, cancelSetup: true },
2725 { failSetup: false, failResolve: false, cancelSetup: true, cancelAsync: true },
2726 ].map((failures) => ({ driver, cancelSetup: false, cancelAsync: false, ...failures })),
2727 ),
2728)(
2729 "prepares $driver delegated worktrees before execution, setup failure=$failSetup, commit failure=$failResolve, cancellation=$cancelSetup, async=$cancelAsync",
2730 ({ driver, failSetup, failResolve, cancelSetup, cancelAsync }) =>
2731 Effect.gen(function* () {
2732 const setupEntered = yield* Deferred.make<void>();
2733 const allowSetup = yield* Deferred.make<void>();
2734 const terminalClosed = yield* Deferred.make<void>();
2735 const childInstanceId = ProviderInstanceId.make(`child-${driver}`);
2736 const parentWorktree = driver === "codex" ? null : "/parent-worktree";
2737 const resolveCommit = vi.fn(() =>
2738 failResolve
2739 ? Effect.fail(
2740 new GitCommandError({
2741 operation: "resolveCommit",
2742 command: "git",
2743 cwd: "/private-checkout",
2744 detail: "private Git failure details",
2745 }),
2746 )
2747 : Effect.succeed({ commitSha: "parent-local-commit" }),
2748 );
2749 const harness = makeHarness({
2750 closeTerminal: () => Deferred.succeed(terminalClosed, undefined).pipe(Effect.asVoid),
2751 adapters: [
2752 adapter,
2753 { ...adapter, instanceId: childInstanceId, driver: ProviderDriverKind.make(driver) },
2754 ],
2755 serverSettings: {
2756 providerInstances: {
2757 [childInstanceId]: { driver: ProviderDriverKind.make(driver), config: {} },
2758 },
2759 },
2760 hasCommit: () => Effect.succeed(true),
2761 providers: [
2762 {
2763 instanceId: childInstanceId,
2764 driver: ProviderDriverKind.make(driver),
2765 enabled: true,
2766 installed: true,
2767 version: "test",
2768 checkedAt: "2026-10-13T00:00:00.000Z",
2769 slashCommands: [],
2770 skills: [],
2771 status: "ready",
2772 auth: { status: "authenticated" },
2773 models: [{ slug: "child-model", name: "Child", isCustom: true, capabilities: null }],
2774 },
2775 ],
2776 runSetup: () =>
2777 cancelSetup
2778 ? Effect.succeed({
2779 status: "started" as const,
2780 async: cancelAsync,
2781 scriptId: "setup",
2782 scriptName: "Setup",
2783 scriptCommand: "setup",
2784 terminalId: "setup",
2785 cwd: "/repo-worktrees/feature",
2786 completion: Deferred.succeed(setupEntered, undefined).pipe(
2787 Effect.andThen(Deferred.await(allowSetup)),
2788 Effect.as({ exitCode: 0, durationMs: 0 }),
2789 ),
2790 })
2791 : Deferred.succeed(setupEntered, undefined).pipe(
2792 Effect.andThen(Deferred.await(allowSetup)),
2793 Effect.andThen(
2794 failSetup
2795 ? Effect.fail(
2796 new ProjectSetupScriptRunner.ProjectSetupScriptOperationError({
2797 threadId: "child",
2798 worktreePath: "/repo-worktrees/feature",
2799 operation: "writeCommand",
2800 cause: "setup failed",
2801 }),
2802 )
2803 : Effect.succeed({ status: "no-script" as const }),
2804 ),
2805 ),
2806 });
2807 const layerService = OrchestratorMcpService.layer.pipe(
2808 Layer.provide(NodeCrypto.layer),
2809 Layer.provide(harness.layer),
2810 Layer.provide(Layer.mock(GitVcsDriver.GitVcsDriver)({ resolveCommit })),
2811 Layer.provide(Layer.mock(ScheduledTasks.ScheduledTaskService)({})),
2812 Layer.provide(Layer.mock(SecretRequests.SecretRequests)({})),
2813 );
2814 yield* Effect.gen(function* () {
2815 const service = yield* OrchestratorMcpService.OrchestratorMcpService;
2816 const threads = yield* ThreadManagement.ThreadManagementService;
2817 const parentThreadId = ThreadId.make("thread:delegated-worktree-parent");
2818 yield* threads.dispatch({
2819 type: "thread.create",
2820 commandId: CommandId.make("create-parent"),
2821 threadId: parentThreadId,
2822 projectId,
2823 title: "Parent",
2824 modelSelection,
2825 runtimeMode: "full-access",
2826 interactionMode: "default",
2827 branch: "parent-feature",
2828 worktreePath: parentWorktree,
2829 createdBy: "user",
2830 creationSource: "web",
2831 });
2832 yield* threads.dispatch({
2833 type: "message.dispatch",
2834 commandId: CommandId.make("start-parent"),
2835 threadId: parentThreadId,
2836 messageId: MessageId.make("parent-message"),
2837 text: "Delegate work",
2838 attachments: [],
2839 dispatchMode: { type: "start_immediately" },
2840 createdBy: "user",
2841 creationSource: "web",
2842 });
2843 const scope = {
2844 environmentId: EnvironmentId.make("delegated-worktree-test"),
2845 requestNamespace: "parent-session",
2846 thread: {
2847 threadId: parentThreadId,
2848 providerSessionId: "parent-session",
2849 providerInstanceId: modelSelection.instanceId,
2850 },
2851 client: undefined,
2852 capabilities: new Set(["orchestration"] as const),
2853 issuedAt: 1,
2854 };
2855 const input = {
2856 task: "Implement the child task",
2857 target: { providerInstanceId: childInstanceId, model: "child-model" },
2858 isolation: "worktree" as const,
2859 clientRequestId: "isolated-child",
2860 };
2861 if (failResolve) {
2862 const error = yield* service.delegateTask(scope, input).pipe(Effect.flip);
2863 assert.equal(error.code, "invalid_request");
2864 assert.equal(error.message, "Unable to resolve the parent checkout commit.");
2865 assert.isEmpty((yield* threads.getThreadProjection(parentThreadId)).subagents);
2866 assert.equal(harness.createWorktree.mock.calls.length, 0);
2867 return;
2868 }
2869 const delegated = yield* service.delegateTask(scope, input);
2870 yield* Deferred.await(setupEntered);
2871 const child = yield* threads.getThreadProjection(delegated.childThreadId);
2872 assert.equal(child.thread.lineage.parentThreadId, parentThreadId);
2873 assert.equal(child.thread.worktreePath, "/repo-worktrees/feature");
2874 assert.equal(child.runs[0]?.status, cancelAsync ? "starting" : "preparing");
2875 assert.equal(child.thread.modelSelection.instanceId, childInstanceId);
2876 assert.equal(child.thread.modelSelection.model, "child-model");
2877 assert.deepEqual(resolveCommit.mock.calls, [
2878 [{ cwd: parentWorktree ?? "/repo", revision: "HEAD" }],
2879 ]);
2880 assert.equal(harness.createWorktree.mock.calls[0]?.[0].refName, "parent-local-commit");
2881 assert.equal((yield* service.delegateTask(scope, input)).taskId, delegated.taskId);
2882 assert.equal(harness.createWorktree.mock.calls.length, 1);
2883 assert.equal(resolveCommit.mock.calls.length, 1);
2884 if (!cancelAsync) assert.isEmpty(child.checkpointScopes);
2885 const parent = yield* threads.getThreadProjection(parentThreadId);
2886 assert.equal(parent.thread.worktreePath, parentWorktree);
2887 assert.equal(parent.thread.branch, "parent-feature");
2888 if (cancelSetup) {
2889 yield* service.cancelTask(scope, { taskId: delegated.taskId });
2890 yield* Deferred.await(terminalClosed);
2891 if (!cancelAsync) {
2892 yield* threads.streamStoredEventsFrom({ threadId: delegated.childThreadId }).pipe(
2893 Stream.filter(
2894 (stored) =>
2895 stored.event.type === "thread.metadata-updated" &&
2896 stored.event.payload.worktreePath === null,
2897 ),
2898 Stream.runHead,
2899 );
2900 }
2901 assert.equal(harness.closeTerminal.mock.calls.length, 1);
2902 assert.equal(harness.removeWorktree.mock.calls.length, cancelAsync ? 0 : 1);
2903 const cancelled = yield* threads.getThreadProjection(delegated.childThreadId);
2904 assert.equal(cancelled.runs[0]?.status, "interrupted");
2905 if (cancelAsync) {
2906 assert.equal(cancelled.thread.worktreePath, "/repo-worktrees/feature");
2907 assert.equal(cancelled.checkpointScopes[0]?.cwd, "/repo-worktrees/feature");
2908 } else {
2909 assert.isNull(cancelled.thread.worktreePath);
2910 assert.isEmpty(cancelled.checkpointScopes);
2911 }
2912 return;
2913 }
2914 yield* Deferred.succeed(allowSetup, undefined);
2915 yield* threads
2916 .streamStoredEventsFrom({ threadId: delegated.childThreadId, eventType: "run.updated" })
2917 .pipe(
2918 Stream.filter(
2919 (stored) =>
2920 stored.event.type === "run.updated" &&
2921 stored.event.payload.status === (failSetup ? "failed" : "starting"),
2922 ),
2923 Stream.runHead,
2924 );
2925 const prepared = yield* threads.getThreadProjection(delegated.childThreadId);
2926 if (failSetup) {
2927 assert.isEmpty(prepared.checkpointScopes);
2928 } else {
2929 assert.equal(prepared.checkpointScopes[0]?.cwd, "/repo-worktrees/feature");
2930 }
2931 }).pipe(Effect.provide(Layer.merge(harness.layer, layerService)));
2932 }),
2933);
apps/server/src/orchestration-v2/ThreadLaunchService.ts+68-28
......@@ -28,6 +28,7 @@ import * as Option from "effect/Option";
2828import * as Ref from "effect/Ref";
2929import * as Schema from "effect/Schema";
3030import * as Scope from "effect/Scope";
31import * as Stream from "effect/Stream";
3132import {
3233 buildTemporaryWorktreeBranchName,
3334 flattenTemporaryWorktreeBranchName,
......@@ -109,7 +110,7 @@ type PreparationInput = Pick<
109110 readonly reusedWorktree?: { readonly baseRef: string };
110111};
111112
112export interface ThreadLaunchRetryInput {
113export interface ThreadLaunchPreparationInput {
113114 readonly commandId: CommandId;
114115 readonly threadId: ThreadId;
115116 readonly runId: RunId;
......@@ -180,13 +181,14 @@ export class ThreadLaunchService extends Context.Service<
180181 input: Pick<ThreadLaunchInput, "projectId" | "workspaceStrategy">,
181182 ) => Effect.Effect<void, ThreadLaunchBaseRefError>;
182183 /** Dispatches prepared-run.retry and prepares the run's workspace again. */
183 readonly retryPreparation: (input: ThreadLaunchRetryInput) => Effect.Effect<
184 readonly retryPreparation: (input: ThreadLaunchPreparationInput) => Effect.Effect<
184185 Orchestrator.OrchestratorV2DispatchResult,
185186 // Dispatches prepared-run.retry, a command the enablement chokepoint
186187 // never gates; the type stays accurate anyway.
187188 | Orchestrator.OrchestratorV2Error
188189 | ThreadManagement.ThreadManagementProviderInstanceDisabledError
189190 >;
191 readonly prepareRun: (input: ThreadLaunchPreparationInput) => Effect.Effect<void>;
190192 }
191193>()("t3/orchestration-v2/ThreadLaunchService") {}
192194
......@@ -317,6 +319,7 @@ const make = Effect.gen(function* () {
317319 let createdWorktreePath: string | null = null;
318320 let setupTerminalId: string | null = null;
319321 let workspaceRecorded = false;
322 let runReleased = false;
320323 if (input.workspaceStrategy.type === "worktree") {
321324 yield* setupTracker.begin({
322325 threadId,
......@@ -623,7 +626,15 @@ const make = Effect.gen(function* () {
623626 threadId,
624627 runId,
625628 })
626 .pipe(Effect.mapError(mapError(input, "release-run", threadId)));
629 .pipe(
630 Effect.mapError(mapError(input, "release-run", threadId)),
631 Effect.tap(() =>
632 Effect.sync(() => {
633 runReleased = true;
634 }),
635 ),
636 Effect.uninterruptible,
637 );
627638 }
628639 yield* setupTracker.stageStatus(threadId, "agent", "done");
629640 yield* awaitAsyncSetup;
......@@ -637,14 +648,15 @@ const make = Effect.gen(function* () {
637648 cancelled ? "cancelled" : "failed",
638649 cancelled ? null : failureDetail(Cause.squash(cause)),
639650 );
640 // A cancelled setup leaves nothing behind. A failed one keeps a worktree
641 // the thread recorded, so a retry reuses it, and removes one it never
642 // recorded, which a retry would otherwise duplicate.
643 if (tracked && createdWorktreePath && (cancelled || !workspaceRecorded)) {
644 if (setupTerminalId)
645 yield* terminals
646 .close({ threadId, terminalId: setupTerminalId, deleteHistory: true })
647 .pipe(Effect.ignore);
651 if (cancelled && setupTerminalId)
652 yield* terminals
653 .close({ threadId, terminalId: setupTerminalId, deleteHistory: true })
654 .pipe(Effect.ignore);
655 if (
656 tracked &&
657 createdWorktreePath &&
658 ((cancelled && !runReleased) || !workspaceRecorded)
659 ) {
648660 const removedPath = createdWorktreePath;
649661 // The thread forgets the worktree only once it is gone; a failed
650662 // removal leaves the directory for the user to clean up rather than
......@@ -734,7 +746,33 @@ const make = Effect.gen(function* () {
734746 threadId: ThreadId,
735747 runId: RunId | null,
736748 ) {
737 yield* prepareInBackground(input, threadId, runId).pipe(
749 const preparation = Effect.gen(function* () {
750 const prepare = prepareInBackground(input, threadId, runId);
751 if (
752 runId === null ||
753 (input.workspaceStrategy.type !== "worktree" && input.reusedWorktree === undefined)
754 ) {
755 return yield* prepare;
756 }
757 const afterSequence = yield* threads.getThreadEventSequence(threadId);
758 const projection = yield* threads.getThreadRecords(threadId, ["runs"]);
759 if (projection.runs.find((run) => run.id === runId)?.status !== "preparing") return;
760 const interruption = threads
761 .streamStoredEventsFrom({ threadId, afterSequence, eventType: "run.updated" })
762 .pipe(
763 Stream.filter(
764 (stored) =>
765 stored.event.type === "run.updated" &&
766 stored.event.payload.id === runId &&
767 (stored.event.payload.status === "interrupted" ||
768 stored.event.payload.status === "cancelled"),
769 ),
770 Stream.runHead,
771 Effect.flatMap((event) => (Option.isSome(event) ? Effect.interrupt : Effect.never)),
772 );
773 yield* Effect.raceFirst(prepare, interruption);
774 });
775 yield* preparation.pipe(
738776 Effect.onError((cause) =>
739777 failPreparedRun(
740778 input,
......@@ -979,37 +1017,39 @@ const make = Effect.gen(function* () {
9791017 },
9801018 );
9811019
982 const retryPreparation: ThreadLaunchService["Service"]["retryPreparation"] = Effect.fn(
983 "ThreadLaunchService.retryPreparation",
1020 const prepareRun: ThreadLaunchService["Service"]["prepareRun"] = Effect.fn(
1021 "ThreadLaunchService.prepareRun",
9841022 )(function* (input) {
985 const dispatched = yield* threads.dispatch({
986 type: "prepared-run.retry",
987 commandId: input.commandId,
988 threadId: input.threadId,
989 runId: input.runId,
990 });
991 // A replayed retry finds the run already past preparation, or prepared by
992 // the attempt that first reserved this command.
993 // From here the run is preparing again; anything that stops preparation
994 // from being scheduled must fail it, or it would wait in preparing forever.
9951023 const scheduled = yield* Effect.gen(function* () {
9961024 const projection = yield* threads.getThreadProjection(input.threadId);
9971025 const run = projection.runs.find((candidate) => candidate.id === input.runId);
9981026 const workspacePreparation = run?.workspacePreparation;
9991027 if (run?.status !== "preparing" || workspacePreparation === undefined) return;
10001028 if (!(yield* reservePreparation(input.commandId))) return;
1001 yield* scheduleRetriedPreparation(input, projection, run, workspacePreparation).pipe(
1029 yield* scheduleRunPreparation(input, projection, run, workspacePreparation).pipe(
10021030 Effect.onError(() => releasePreparation(input.commandId)),
10031031 );
10041032 }).pipe(Effect.exit);
10051033 if (Exit.isFailure(scheduled)) {
10061034 yield* failPreparedRun(input, input.threadId, input.runId, Cause.squash(scheduled.cause));
10071035 }
1036 });
1037
1038 const retryPreparation: ThreadLaunchService["Service"]["retryPreparation"] = Effect.fn(
1039 "ThreadLaunchService.retryPreparation",
1040 )(function* (input) {
1041 const dispatched = yield* threads.dispatch({
1042 type: "prepared-run.retry",
1043 commandId: input.commandId,
1044 threadId: input.threadId,
1045 runId: input.runId,
1046 });
1047 yield* prepareRun(input);
10081048 return dispatched;
10091049 });
10101050
1011 const scheduleRetriedPreparation = (
1012 input: ThreadLaunchRetryInput,
1051 const scheduleRunPreparation = (
1052 input: ThreadLaunchPreparationInput,
10131053 projection: OrchestrationV2ThreadProjection,
10141054 run: OrchestrationV2ThreadProjection["runs"][number],
10151055 workspacePreparation: ThreadLaunchWorkspaceStrategy,
......@@ -1050,7 +1090,7 @@ const make = Effect.gen(function* () {
10501090 );
10511091 };
10521092
1053 return ThreadLaunchService.of({ launch, checkWorktreeBase, retryPreparation });
1093 return ThreadLaunchService.of({ launch, checkWorktreeBase, retryPreparation, prepareRun });
10541094});
10551095
10561096export const layer = Layer.effect(ThreadLaunchService, make);
packages/contracts/src/orchestrationV2.ts+1
......@@ -3110,6 +3110,7 @@ export const OrchestrationV2Command = Schema.Union([
31103110 modelSelection: ModelSelection,
31113111 runtimeMode: RuntimeMode,
31123112 interactionMode: ProviderInteractionMode,
3113 workspaceStrategy: Schema.optional(OrchestrationV2ThreadLaunchWorkspaceStrategy),
31133114 // Omitted behaves as "settled_only" (no wake while the parent has a live
31143115 // run); producers that want fire-and-forget wakes must set "always".
31153116 completionWake: Schema.optional(Schema.Literals(["always", "settled_only"])),
packages/contracts/src/orchestratorMcp.test.ts+7
......@@ -22,6 +22,13 @@ const decodeThreadSendInput = Schema.decodeUnknownSync(OrchestratorMcpThreadSend
2222const decodeThreadWaitInput = Schema.decodeUnknownSync(OrchestratorMcpThreadWaitInput);
2323
2424describe("orchestrator MCP contracts", () => {
25 it("accepts explicit worktree isolation and keeps shared delegation as the default", () => {
26 expect(decodeDelegateTaskInput({ task: "Review changes" }).isolation).toBeUndefined();
27 expect(
28 decodeDelegateTaskInput({ task: "Review changes", isolation: "worktree" }).isolation,
29 ).toBe("worktree");
30 expect(() => decodeDelegateTaskInput({ task: "Review changes", isolation: "copy" })).toThrow();
31 });
2532 it("decodes cross-provider delegated task requests and durable results", () => {
2633 const request = decodeDelegateTaskInput({
2734 task: "Inspect the workspace and report the result.",
packages/contracts/src/orchestratorMcp.ts+6
......@@ -174,6 +174,12 @@ export const OrchestratorMcpDelegateTaskInput = Schema.Struct({
174174 target: Schema.optional(OrchestratorMcpTarget),
175175 title: Schema.optional(OrchestratorMcpTitle),
176176 role: Schema.optional(OrchestratorMcpTaskRole),
177 isolation: Schema.optional(
178 Schema.Literals(["shared", "worktree"]).annotate({
179 description:
180 "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.",
181 }),
182 ),
177183 mode: Schema.optional(
178184 Schema.Literals(["async", "wait"]).annotate({
179185 description:
packages/provider-core/src/server/orchestrationInstructions.ts+1-1
......@@ -6,7 +6,7 @@ export const T3_CODE_ORCHESTRATION_INSTRUCTIONS = `
66
77The \`t3-code\` MCP server provides app-owned orchestration. Treat these concepts distinctly:
88
9- 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.
9- 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.
1010- \`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.
1111- 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.
1212- \`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.