Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Yash-Singh1 merged 4 commits into from
Oct 14, 2026
+454 −80
Merged

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
191783b
feat(server): run delegated tasks in separate worktrees
Bil0000 Oct 13, 2026
c8165ba
fix(server): keep Git details out of delegation errors
Bil0000 Oct 13, 2026
ab14b2e
test(server): cover worktree preparation after restart
Bil0000 Oct 13, 2026
f9013fa
fix(server): stop worktree setup when delegated tasks are cancelled
Bil0000 Oct 13, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 11 additions & 4 deletions apps/server/src/mcp/OrchestratorMcpService.activity.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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");
Expand Down Expand Up @@ -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)({
Expand Down Expand Up @@ -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)({
Expand Down Expand Up @@ -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)({
Expand Down Expand Up @@ -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)({
Expand Down
55 changes: 27 additions & 28 deletions apps/server/src/mcp/OrchestratorMcpService.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand All @@ -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* () {
Expand Down Expand Up @@ -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))));
}),
);

Expand Down Expand Up @@ -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))));
}),
);

Expand Down Expand Up @@ -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))));
}),
);

Expand Down Expand Up @@ -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))));
}),
);

Expand Down Expand Up @@ -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))));
}),
);

Expand Down Expand Up @@ -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))));
}),
);

Expand Down Expand Up @@ -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", () =>
Expand Down Expand Up @@ -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))));
}),
);

Expand Down Expand Up @@ -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))));
}),
);

Expand Down Expand Up @@ -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))));
}),
);

Expand Down Expand Up @@ -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))));
}),
);

Expand Down Expand Up @@ -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))));
}),
);

Expand Down Expand Up @@ -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))));
}),
);

Expand Down Expand Up @@ -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))));
}),
);

Expand Down Expand Up @@ -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))));
}),
);

Expand Down Expand Up @@ -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))));
}),
);

Expand Down Expand Up @@ -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))));
}
}),
);
Expand Down Expand Up @@ -1972,7 +1971,7 @@ describe("OrchestratorMcpService provider resolution", () => {
boundThread: OrchestrationV2ThreadShell | null,
upserted: Ref.Ref<number>,
) =>
OrchestratorMcpService.layer.pipe(
layerOrchestratorService.pipe(
Layer.provide(
Layer.mergeAll(
NodeServices.layer,
Expand Down Expand Up @@ -2040,7 +2039,7 @@ describe("OrchestratorMcpService provider resolution", () => {
),
),
Effect.provide(
OrchestratorMcpService.layer.pipe(
layerOrchestratorService.pipe(
Layer.provide(
Layer.mergeAll(
NodeServices.layer,
Expand Down Expand Up @@ -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,
Expand Down
39 changes: 39 additions & 0 deletions apps/server/src/mcp/OrchestratorMcpService.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ import {
type OrchestrationV2Subagent,
type OrchestrationV2ThreadProjection,
type OrchestrationV2ThreadShell,
type OrchestrationV2ThreadLaunchWorkspaceStrategy,
type OrchestrationV2TurnItem,
OrchestratorMcpFailure,
type OrchestratorMcpCapabilitiesResult,
Expand Down Expand Up @@ -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";
Expand Down Expand Up @@ -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. */
Expand Down Expand Up @@ -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.
Expand All @@ -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).
Expand All @@ -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,
});
Comment thread
coderabbitai[bot] marked this conversation as resolved.
}

if (input.mode !== "wait") {
return yield* readTask(scope, taskId, false, true);
Expand Down Expand Up @@ -2660,6 +2697,8 @@ export const layer: Layer.Layer<
never,
| Crypto.Crypto
| ThreadManagementService.ThreadManagementService
| ThreadLaunchService.ThreadLaunchService
| GitVcsDriver.GitVcsDriver
| ProviderRegistry.ProviderRegistry
| ProviderAdapterRegistry.ProviderAdapterRegistryV2
| ScheduledTaskService.ScheduledTaskService
Expand Down
11 changes: 9 additions & 2 deletions apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand All @@ -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) => {
Expand Down Expand Up @@ -682,7 +689,7 @@ describe("orchestrator MCP toolkit", () => {
}),
);
const layerTest = Layer.merge(
McpHttpServer.layerOrchestratorToolkit,
layerOrchestratorToolkit,
McpHttpServer.layerThreadToolkit,
).pipe(
Layer.provideMerge(McpServer.McpServer.layer),
Expand Down Expand Up @@ -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(
Expand Down
Loading
Loading