Skip to content

Commit d70bfe6

Browse files
juliusmarmingejuliusmarminge
authored
fix(server): a send to a thread whose worktree can't be recreated now fails instead of reporting started (#36251)
1 parent 9b5d997 commit d70bfe6

7 files changed

Lines changed: 229 additions & 7 deletions

File tree

‎apps/server/src/mcp/OrchestratorMcpService.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -251,6 +251,7 @@ function threadManagementFailure(error: unknown): OrchestratorMcpFailure {
251251
return failure("run_not_found", error.message);
252252
case "ThreadManagementThreadArchivedError":
253253
case "ThreadManagementNoSteerableRunError":
254+
case "ThreadManagementWorktreeRestoreError":
254255
return failure("thread_not_sendable", error.message);
255256
case "ThreadManagementThreadNotInterruptibleError":
256257
return failure("thread_not_interruptible", error.message);

‎apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts‎

Lines changed: 48 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,10 @@ import {
4040
withWorkspaceLease,
4141
} from "../workspace/workspaceLease.ts";
4242
import * as GitWorkflow from "../git/GitWorkflowService.ts";
43+
import {
44+
layerThreadWorktreeRestore,
45+
ThreadWorktreeRestore,
46+
} from "../workspace/restoreMissingWorktree.ts";
4347
import * as GitVcsDriver from "../vcs/GitVcsDriver.ts";
4448
import { layerRealGit } from "./ProviderTurnStartService.testkit.ts";
4549
import * as ProjectService from "../project/ProjectService.ts";
@@ -272,6 +276,50 @@ it.each([
272276
expect(harness.createWorktree).toHaveBeenCalledTimes(input.recreated ? 1 : 0);
273277
});
274278

279+
const restoreLeaseExists = vi.fn(() => Effect.succeed(false));
280+
const restoreLeaseCreateWorktree = vi.fn(() => Effect.succeed({} as never));
281+
const layerRestoreLeaseTest = layerThreadWorktreeRestore.pipe(
282+
Layer.provide(
283+
Layer.mergeAll(
284+
Layer.succeed(FileSystem.FileSystem, {
285+
exists: restoreLeaseExists,
286+
realPath: (target: string) => Effect.succeed(normalize(target)),
287+
} as never),
288+
layerWorktreeLocation(),
289+
Layer.mock(GitWorkflow.GitWorkflowService)({
290+
pruneWorktrees: () => Effect.void,
291+
createWorktree: restoreLeaseCreateWorktree,
292+
listLocalBranchNames: () => Effect.succeed(["feature/restore"]),
293+
}),
294+
Layer.mock(ProjectService.ProjectService)({
295+
getById: () =>
296+
Effect.succeedSome({ workspaceRoot: "/tmp/provider-turn-start-project" } as never),
297+
}),
298+
),
299+
),
300+
);
301+
302+
effectIt.effect("restores a missing worktree before a send only under the workspace lease", () =>
303+
Effect.gen(function* () {
304+
const worktreePath = `${managedWorktreesDir}/project/feature-restore`;
305+
const { restore } = yield* ThreadWorktreeRestore;
306+
yield* withWorkspaceLease(
307+
worktreePath,
308+
Effect.gen(function* () {
309+
const restoring = yield* restore({
310+
id: ThreadId.make("thread_send_restore_lease"),
311+
projectId: ProjectId.make("project_send_restore_lease"),
312+
worktreePath,
313+
branch: "feature/restore",
314+
}).pipe(Effect.forkChild({ startImmediately: true }));
315+
expect(restoreLeaseExists).not.toHaveBeenCalled();
316+
return restoring;
317+
}),
318+
).pipe(Effect.flatMap(Fiber.join));
319+
expect(restoreLeaseCreateWorktree).toHaveBeenCalledTimes(1);
320+
}).pipe(Effect.provide(layerRestoreLeaseTest)),
321+
);
322+
275323
function makeLocalCommandHarness(input: {
276324
readonly text: string;
277325
readonly nativeGoals?: boolean;

‎apps/server/src/orchestration-v2/ThreadManagementService.test.ts‎

Lines changed: 100 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,11 @@ import * as ProviderSessionManager from "./ProviderSessionManager.ts";
3838
import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
3939
import * as ThreadManagementService from "./ThreadManagementService.ts";
4040
import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts";
41+
import * as LegacyV1ThreadImporter from "./legacy/LegacyV1ThreadImporter.ts";
42+
import {
43+
ThreadWorktreeRestore,
44+
WorktreeRecreateError,
45+
} from "../workspace/restoreMissingWorktree.ts";
4146

4247
it("stamps authoritative provenance on commands that create threads or messages", () => {
4348
const command: OrchestrationV2Command = {
@@ -199,6 +204,101 @@ it.effect(
199204
},
200205
);
201206

207+
it.effect.each([
208+
{ label: "fails a send that starts a turn", running: false },
209+
{ label: "still steers a running turn", running: true },
210+
])("sendToThread with a missing worktree that cannot be recreated $label", ({ running }) => {
211+
const projectId = ProjectId.make("project:thread-management:worktree-gone");
212+
const threadId = ThreadId.make("thread:thread-management:worktree-gone");
213+
const dispatched: Array<string> = [];
214+
const layerTest = ThreadManagementService.layerWithLegacyImporter.pipe(
215+
Layer.provide(
216+
Layer.succeed(ThreadWorktreeRestore, {
217+
restore: () =>
218+
Effect.fail(
219+
new WorktreeRecreateError({
220+
worktreePath: "/worktrees/gone",
221+
branch: "t3/gone",
222+
cause: new Error("branch_checked_out_in_worktree"),
223+
}),
224+
),
225+
}),
226+
),
227+
Layer.provide(
228+
Layer.mock(LegacyV1ThreadImporter.LegacyV1ThreadImporter)({
229+
ensureTranscript: () => Effect.succeed({ importedThreadCount: 0, importedMessageCount: 0 }),
230+
}),
231+
),
232+
Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})),
233+
Layer.provide(
234+
Layer.mock(Orchestrator.OrchestratorV2)({
235+
getThreadRecords: () =>
236+
Effect.succeed({
237+
thread: {
238+
id: threadId,
239+
projectId,
240+
deletedAt: null,
241+
archivedAt: null,
242+
worktreePath: "/worktrees/gone",
243+
branch: "t3/gone",
244+
modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "test" },
245+
},
246+
runs: running
247+
? [
248+
{
249+
id: RunId.make("run:thread-management:worktree-gone"),
250+
status: "running",
251+
activeAttemptId: "attempt:thread-management:worktree-gone",
252+
},
253+
]
254+
: [],
255+
providerTurns: running
256+
? [
257+
{
258+
runAttemptId: "attempt:thread-management:worktree-gone",
259+
status: "running",
260+
},
261+
]
262+
: [],
263+
messages: [],
264+
turnItems: [],
265+
} as unknown as OrchestrationV2ThreadProjection),
266+
dispatch: (command) =>
267+
Effect.sync(() => {
268+
dispatched.push(command.type);
269+
return { sequence: 1, storedEvents: [] } as never;
270+
}),
271+
}),
272+
),
273+
Layer.provide(ServerSettings.layerTest()),
274+
Layer.provideMerge(layerReceiptsTest),
275+
);
276+
277+
return Effect.gen(function* () {
278+
const threads = yield* ThreadManagementService.ThreadManagementService;
279+
const error = yield* Effect.flip(
280+
threads.sendToThread({
281+
projectId,
282+
threadId,
283+
commandId: CommandId.make("command:thread-management:worktree-gone"),
284+
messageId: MessageId.make("message:thread-management:worktree-gone"),
285+
text: "Reply with ok.",
286+
attachments: [],
287+
mode: "auto",
288+
createdBy: "agent",
289+
creationSource: "mcp",
290+
}),
291+
);
292+
// The mock records no message, so a dispatched send ends in the projection read.
293+
expect(error._tag).toBe(
294+
running
295+
? "ThreadManagementDurableRunProjectionError"
296+
: "ThreadManagementWorktreeRestoreError",
297+
);
298+
expect(dispatched).toEqual(running ? ["message.dispatch"] : []);
299+
}).pipe(Effect.provide(layerTest));
300+
});
301+
202302
it.effect("waitForThread follows a restart continuation instead of the cut run", () =>
203303
Effect.gen(function* () {
204304
const projectId = ProjectId.make("project:thread-management:wait-restart");

‎apps/server/src/orchestration-v2/ThreadManagementService.ts‎

Lines changed: 44 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -69,6 +69,7 @@ import { DispatchModeLimit, exceededDispatchModeLimit } from "./DispatchModeLimi
6969
import { projectTurnItemForDetail } from "./WireProjection.ts";
7070
import * as LegacyV1ThreadImporter from "./legacy/LegacyV1ThreadImporter.ts";
7171
import * as ServerSettings from "../serverSettings.ts";
72+
import { ThreadWorktreeRestore } from "../workspace/restoreMissingWorktree.ts";
7273

7374
export type ThreadManagementSendMode = "auto" | "queue" | "steer" | "restart";
7475

@@ -316,6 +317,19 @@ export class ThreadManagementDurableRunProjectionError extends Schema.TaggedErro
316317
}
317318
}
318319

320+
export class ThreadManagementWorktreeRestoreError extends Schema.TaggedError<ThreadManagementWorktreeRestoreError>()(
321+
"ThreadManagementWorktreeRestoreError",
322+
{
323+
threadId: ThreadId,
324+
cause: Schema.Defect(),
325+
},
326+
) {
327+
override get message(): string {
328+
const detail = this.cause instanceof Error ? ` ${this.cause.message}` : "";
329+
return `Couldn't recreate the missing workspace of thread ${this.threadId}, so the message was not sent.${detail}`;
330+
}
331+
}
332+
319333
export class ThreadManagementProviderInstanceDisabledError extends Schema.TaggedError<ThreadManagementProviderInstanceDisabledError>()(
320334
"ThreadManagementProviderInstanceDisabledError",
321335
{
@@ -360,6 +374,7 @@ export const ThreadManagementError = Schema.Union([
360374
ThreadManagementProjectThreadsListError,
361375
ThreadManagementDurableRunProjectionError,
362376
ThreadManagementProviderInstanceDisabledError,
377+
ThreadManagementWorktreeRestoreError,
363378
]);
364379
export type ThreadManagementError = typeof ThreadManagementError.Type;
365380

@@ -592,6 +607,7 @@ const make = Effect.gen(function* () {
592607
const legacyImporter = yield* LegacyV1ThreadImporter.LegacyV1ThreadImporter;
593608
const serverSettings = yield* ServerSettings.ServerSettingsService;
594609
const receipts = yield* CommandReceiptStore.CommandReceiptStoreV2;
610+
const worktreeRestore = yield* ThreadWorktreeRestore;
595611

596612
const assertProviderInstanceEnabled: ThreadManagementServiceShape["assertProviderInstanceEnabled"] =
597613
(input) =>
@@ -873,6 +889,26 @@ const make = Effect.gen(function* () {
873889
};
874890
}
875891

892+
// Turn start would fail on its own after the send reported success; a replay keeps its answer.
893+
// A live steer starts no turn, so it needs no workspace.
894+
if (dispatchMode.type !== "steer_active") {
895+
yield* worktreeRestore
896+
.restore(target.thread)
897+
.pipe(
898+
Effect.catch((cause) =>
899+
hasReceipt(input.commandId).pipe(
900+
Effect.flatMap((replay) =>
901+
replay
902+
? Effect.void
903+
: Effect.fail(
904+
new ThreadManagementWorktreeRestoreError({ threadId: input.threadId, cause }),
905+
),
906+
),
907+
),
908+
),
909+
);
910+
}
911+
876912
const dispatched = yield* dispatch({
877913
type: "message.dispatch",
878914
commandId: input.commandId,
@@ -1719,6 +1755,12 @@ const make = Effect.gen(function* () {
17191755
});
17201756
});
17211757

1758+
const layerThreadWorktreeRestoreNoop = Layer.succeed(ThreadWorktreeRestore, {
1759+
restore: Effect.fn("ThreadWorktreeRestore.restore")(function () {
1760+
return Effect.void;
1761+
}),
1762+
});
1763+
17221764
const layerLegacyV1ThreadImporterNoop = Layer.succeed(
17231765
LegacyV1ThreadImporter.LegacyV1ThreadImporter,
17241766
LegacyV1ThreadImporter.LegacyV1ThreadImporter.of({
@@ -1740,13 +1782,15 @@ export const layer: Layer.Layer<
17401782
| CommandReceiptStore.CommandReceiptStoreV2
17411783
> = Layer.effect(ThreadManagementService, make).pipe(
17421784
Layer.provide(layerLegacyV1ThreadImporterNoop),
1785+
Layer.provide(layerThreadWorktreeRestoreNoop),
17431786
Layer.provide(ThreadCommandExecutor.layer),
17441787
);
17451788

17461789
export const layerWithLegacyImporter: Layer.Layer<
17471790
ThreadManagementService,
17481791
never,
17491792
| LegacyV1ThreadImporter.LegacyV1ThreadImporter
1793+
| ThreadWorktreeRestore
17501794
| Orchestrator.OrchestratorV2
17511795
| ProviderSessionManager.ProviderSessionManagerV2
17521796
| ServerSettings.ServerSettingsService

‎apps/server/src/orchestration-v2/ThreadMessageIntake.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@ function dispatchWasNotAccepted(
2727
case "OrchestratorProviderSelectionBusyError":
2828
case "OrchestratorRuntimeModeAboveCeilingError":
2929
case "ThreadManagementProviderInstanceDisabledError":
30+
case "ThreadManagementWorktreeRestoreError":
3031
return true;
3132
default:
3233
return false;

‎apps/server/src/orchestration-v2/runtimeLayer.ts‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -47,6 +47,7 @@ import * as ProviderRuntimeRecoveryService from "./ProviderRuntimeRecoveryServic
4747
import * as ProviderSwitchService from "./ProviderSwitchService.ts";
4848
import * as ProviderTurnControlService from "./ProviderTurnControlService.ts";
4949
import * as ProviderTurnStartService from "./ProviderTurnStartService.ts";
50+
import { layerThreadWorktreeRestore } from "../workspace/restoreMissingWorktree.ts";
5051
import * as RunExecutionService from "./RunExecutionService.ts";
5152
import * as RunFinalizationService from "./RunFinalizationService.ts";
5253
import * as ResourceCleanupService from "./ResourceCleanupService.ts";
@@ -267,6 +268,7 @@ const layerThreadManagementProvided = ThreadManagementService.layerWithLegacyImp
267268
layerLegacyV1ThreadImporterProvided,
268269
layerCommandReceiptStoreProvided,
269270
layerProviderSessionManagerProvided,
271+
layerThreadWorktreeRestore,
270272
),
271273
),
272274
);
@@ -281,14 +283,13 @@ const layerManagedProjectFoldersProvided = ManagedProjectFolders.layer.pipe(
281283
const layerThreadLaunchProvided = ThreadLaunchService.layer.pipe(
282284
Layer.provide(
283285
Layer.mergeAll(
284-
layerProjectService,
285286
layerProjectSetupScriptRunner,
286287
layerManagedProjectFoldersProvided,
287288
layerThreadManagementProvided,
288289
layerCommandReceiptStoreProvided,
289290
layerEventSink,
290291
IdAllocator.layer,
291-
),
292+
).pipe(Layer.provideMerge(layerProjectService)),
292293
),
293294
);
294295
const layerSecretRequestsProvided = SecretRequests.layer.pipe(
@@ -307,11 +308,10 @@ const layerScheduledTaskProvided = ScheduledTaskService.layer.pipe(
307308
const layerScratchThreadPromotionProvided = ScratchThreadPromotion.layer.pipe(
308309
Layer.provide(
309310
Layer.mergeAll(
310-
layerProjectService,
311311
layerManagedProjectFoldersProvided,
312312
layerThreadManagementProvided,
313313
layerScheduledTaskProvided,
314-
),
314+
).pipe(Layer.provideMerge(layerProjectService)),
315315
),
316316
);
317317
const layerProviderContinuationWorkerProvided = ProviderContinuationService.layer.pipe(
@@ -389,8 +389,7 @@ export const layerProduction = Layer.mergeAll(
389389
WorktreeBranchSync.layer.pipe(
390390
Layer.provide(Layer.merge(layerThreadManagementProvided, ProjectionStore.layer)),
391391
),
392-
layer.pipe(Layer.provide(layerProjectService)),
393-
layerProjectService,
392+
layer,
394393
layerManagedProjectFoldersProvided,
395394
layerThreadLaunchProvided,
396395
layerScheduledTaskProvided,
@@ -414,6 +413,7 @@ export const layerProduction = Layer.mergeAll(
414413
EffectOutbox.layerPruneWorker.pipe(Layer.provide(EffectOutbox.layer)),
415414
).pipe(
416415
Layer.provide(ResourceCleanupService.layer.pipe(Layer.provide(layerScheduledTaskProvided))),
416+
Layer.provideMerge(layerProjectService),
417417
Layer.provide(Scheduler.layer),
418418
Layer.provideMerge(layerEventInfrastructure),
419419
);

0 commit comments

Comments
 (0)