1import * as WorktreeSetupTracker from "../project/WorktreeSetupTracker.ts";
2import * as ProjectCloneTracker from "../project/ProjectCloneTracker.ts";
3import * as TerminalManager from "../terminal/Manager.ts";
4import { resolveProjectSettings } from "@t3tools/shared/projectSettings";
5import {
6 CommandId,
7 type ChatAttachment,
8 type MessageId,
9 type ModelSelection,
10 type OrchestrationV2Actor,
11 type OrchestrationV2CreationSource,
12 type OrchestrationV2DelegatedFrom,
13 type OrchestrationV2LinkOrigin,
14 type OrchestrationV2ProviderThreadNativeMetadata,
15 type OrchestrationV2ThreadProjection,
16 type ProviderDriverKind,
17 type ProviderInteractionMode,
18 ProjectId,
19 type RunId,
20 type RuntimeMode,
21 type ScheduledTaskId,
22 ThreadId,
23} from "@t3tools/contracts";
24import * as Cause from "effect/Cause";
25import * as Clock from "effect/Clock";
26import * as Context from "effect/Context";
27import * as DateTime from "effect/DateTime";
28import * as Effect from "effect/Effect";
29import * as Exit from "effect/Exit";
30import * as Fiber from "effect/Fiber";
31import * as Layer from "effect/Layer";
32import * as Option from "effect/Option";
33import * as Ref from "effect/Ref";
34import * as Schema from "effect/Schema";
35import * as Scope from "effect/Scope";
36import * as Stream from "effect/Stream";
37import {
38 buildTemporaryWorktreeBranchName,
39 flattenTemporaryWorktreeBranchName,
40 formatGeneratedBranchName,
41 isTemporaryWorktreeBranch,
42} from "@t3tools/shared/git";
43
44import * as GitWorkflow from "../git/GitWorkflowService.ts";
45import { parseGitCloneProgressLine } from "../project/gitCloneProgress.ts";
46import * as ProjectService from "../project/ProjectService.ts";
47import * as ProjectSetupScriptRunner from "../project/ProjectSetupScriptRunner.ts";
48import * as ManagedProjectFolders from "../project/ManagedProjectFolders.ts";
49import * as ProviderRegistry from "../provider/ProviderRegistry.ts";
50import * as ServerSettings from "../serverSettings.ts";
51import * as TextGeneration from "../textGeneration/TextGeneration.ts";
52import * as CommandReceiptStore from "./CommandReceiptStore.ts";
53import * as EventSink from "./EventSink.ts";
54import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
55import type * as Orchestrator from "./Orchestrator.ts";
56import { makeProviderFailure } from "@t3tools/provider-core/server/failure";
57import { randomUuidV4 } from "@t3tools/provider-core/server/randomUuid";
58import { DispatchModeLimit } from "./DispatchModeLimit.ts";
59import * as ThreadManagement from "./ThreadManagementService.ts";
60
61export type ThreadLaunchWorkspaceStrategy =
62 | { readonly type: "root"; readonly branch?: string | undefined }
63 | {
64 readonly type: "existing_worktree";
65 readonly worktreePath: string;
66 readonly branch?: string | undefined;
67 }
68 | {
69 readonly type: "worktree";
70 readonly baseRef: string;
71 readonly branch?: string | undefined;
72 readonly startFromOrigin?: boolean | undefined;
73 };
74
75export interface ThreadLaunchInitialMessage {
76 readonly messageId?: MessageId;
77 readonly scheduledTaskId?: ScheduledTaskId;
78 readonly senderThreadId?: ThreadId;
79 readonly text: string;
80 readonly attachments: ReadonlyArray<ChatAttachment>;
81 readonly context?: import("@t3tools/contracts").OrchestrationMessageContext | undefined;
82}
83
84export interface ThreadLaunchInput {
85 readonly commandId: CommandId;
86 readonly threadId?: ThreadId;
87 readonly reuseExistingThread?: boolean;
88 readonly projectId: ProjectId;
89 readonly title: string;
90 readonly generateTitle?: boolean;
91 readonly modelSelection: ModelSelection;
92 readonly runtimeMode: RuntimeMode;
93 readonly interactionMode: ProviderInteractionMode;
94 readonly workspaceStrategy: ThreadLaunchWorkspaceStrategy;
95 readonly initialMessage?: ThreadLaunchInitialMessage;
96 readonly importedNativeThread?: {
97 readonly ref: {
98 readonly driver: ProviderDriverKind;
99 readonly nativeId: string;
100 readonly strength: "strong";
101 };
102 readonly metadata?: OrchestrationV2ProviderThreadNativeMetadata;
103 };
104 readonly createdBy: OrchestrationV2Actor;
105 readonly creationSource: OrchestrationV2CreationSource;
106 /** The linked environment this launch is for, if one asked for it. */
107 readonly linkOrigin?: OrchestrationV2LinkOrigin;
108 /** The parent there, when that environment's agent delegated this as a task. */
109 readonly delegatedFrom?: OrchestrationV2DelegatedFrom;
110}
111
112type PreparationInput = Pick<
113 ThreadLaunchInput,
114 "commandId" | "projectId" | "workspaceStrategy" | "initialMessage"
115> & {
116 readonly reusedWorktree?: { readonly baseRef: string };
117};
118
119export interface ThreadLaunchPreparationInput {
120 readonly commandId: CommandId;
121 readonly threadId: ThreadId;
122 readonly runId: RunId;
123}
124
125export interface ThreadLaunchResult {
126 readonly threadId: ThreadId;
127 readonly projection: OrchestrationV2ThreadProjection;
128 /** The launch's last write (the initial message, else the thread claim) was accepted before this call. */
129 readonly resumed: boolean;
130 readonly sequence?: number;
131}
132
133export class ThreadLaunchError extends Schema.TaggedError<ThreadLaunchError>()(
134 "ThreadLaunchError",
135 {
136 operation: Schema.Literals([
137 "resolve-project",
138 "read-receipt",
139 "generate-metadata",
140 "provision-worktree",
141 "run-setup-script",
142 "create-thread",
143 "update-thread",
144 "dispatch-message",
145 "release-run",
146 "fail-run",
147 ]),
148 commandId: CommandId,
149 projectId: ProjectId,
150 threadId: Schema.optional(ThreadId),
151 cause: Schema.Defect(),
152 },
153) {
154 override get message(): string {
155 return `Thread launch ${this.commandId} failed during ${this.operation}.`;
156 }
157}
158
159export class ThreadLaunchBaseRefError extends Schema.TaggedError<ThreadLaunchBaseRefError>()(
160 "ThreadLaunchBaseRefError",
161 {
162 projectId: ProjectId,
163 projectTitle: Schema.String,
164 workspaceRoot: Schema.String,
165 baseRef: Schema.String,
166 },
167) {
168 override get message(): string {
169 return `Base ref "${this.baseRef}" does not resolve to a commit in project "${this.projectTitle}" (${this.workspaceRoot}).`;
170 }
171}
172
173export class ThreadLaunchBranchConflictError extends Schema.TaggedError<ThreadLaunchBranchConflictError>()(
174 "ThreadLaunchBranchConflictError",
175 {
176 projectId: ProjectId,
177 projectTitle: Schema.String,
178 branch: Schema.String,
179 conflictingBranch: Schema.String,
180 },
181) {
182 override get message(): string {
183 return `Branch "${this.branch}" cannot be created in project "${this.projectTitle}": branch "${this.conflictingBranch}" exists.`;
184 }
185}
186
187export class ThreadLaunchNotRepositoryError extends Schema.TaggedError<ThreadLaunchNotRepositoryError>()(
188 "ThreadLaunchNotRepositoryError",
189 {
190 projectId: ProjectId,
191 projectTitle: Schema.String,
192 workspaceRoot: Schema.String,
193 },
194) {
195 override get message(): string {
196 return `Project "${this.projectTitle}" (${this.workspaceRoot}) is not a Git repository.`;
197 }
198}
199
200export class ThreadLaunchService extends Context.Service<
201 ThreadLaunchService,
202 {
203 readonly launch: (
204 input: ThreadLaunchInput,
205 ) => Effect.Effect<
206 ThreadLaunchResult,
207 ThreadLaunchError | ThreadManagement.ThreadManagementProviderInstanceDisabledError
208 >;
209 /** Check a new worktree's base ref and branch name before creation for early errors; launch otherwise accepts and reports these failures inside the thread. */
210 readonly checkWorktreeBase: (
211 input: Pick<ThreadLaunchInput, "projectId" | "workspaceStrategy"> & {
212 /** Skips the check when this launch was already accepted, so its retry replays. */
213 readonly commandId?: CommandId;
214 },
215 ) => Effect.Effect<
216 void,
217 ThreadLaunchBaseRefError | ThreadLaunchBranchConflictError | ThreadLaunchNotRepositoryError
218 >;
219 readonly retryPreparation: (
220 input: ThreadLaunchPreparationInput,
221 ) => Effect.Effect<
222 Orchestrator.OrchestratorV2DispatchResult,
223 | Orchestrator.OrchestratorV2Error
224 | ThreadManagement.ThreadManagementProviderInstanceDisabledError
225 >;
226 readonly prepareRun: (input: ThreadLaunchPreparationInput) => Effect.Effect<void>;
227 }
228>()("t3/orchestration-v2/ThreadLaunchService") {}
229
230const isThreadLaunchError = Schema.is(ThreadLaunchError);
231
232function failureDetail(error: unknown): string {
233 if (isThreadLaunchError(error)) {
234 const cause = error.cause;
235 const detail = cause instanceof Error ? cause.message : String(cause);
236 return `Workspace preparation failed during ${error.operation.replaceAll("-", " ")}: ${detail}`;
237 }
238 return `Workspace preparation failed: ${error instanceof Error ? error.message : String(error)}`;
239}
240
241const make = Effect.gen(function* () {
242 const projects = yield* ProjectService.ProjectService;
243 const setupTracker = yield* WorktreeSetupTracker.WorktreeSetupTracker;
244 const cloneTracker = yield* ProjectCloneTracker.ProjectCloneTracker;
245 const terminals = yield* TerminalManager.TerminalManager;
246 const git = yield* GitWorkflow.GitWorkflowService;
247 const setupScripts = yield* ProjectSetupScriptRunner.ProjectSetupScriptRunner;
248 const providerRegistry = yield* ProviderRegistry.ProviderRegistry;
249 const serverSettings = yield* ServerSettings.ServerSettingsService;
250 const textGeneration = yield* TextGeneration.TextGeneration;
251 const receipts = yield* CommandReceiptStore.CommandReceiptStoreV2;
252 const ids = yield* IdAllocator.IdAllocatorV2;
253 const threads = yield* ThreadManagement.ThreadManagementService;
254 const eventSink = yield* EventSink.EventSinkV2;
255 const managedFolders = yield* ManagedProjectFolders.ManagedProjectFolders;
256 // A project-folder launch that names no branch records the folder's checkout, as the composer does.
257 const launchBranch = (strategy: ThreadLaunchWorkspaceStrategy, workspaceRoot: string) =>
258 strategy.type === "root" && strategy.branch === undefined
259 ? git.currentBranch(workspaceRoot).pipe(Effect.orElseSucceed(() => null))
260 : Effect.succeed(strategy.branch ?? null);
261 const preparationScope = yield* Scope.make("sequential");
262 const scheduledLaunches = yield* Ref.make<ReadonlySet<CommandId>>(new Set());
263 yield* Effect.addFinalizer(() => Scope.close(preparationScope, Exit.void));
264
265 const mapError =
266 (input: PreparationInput, operation: ThreadLaunchError["operation"], threadId?: ThreadId) =>
267 (cause: unknown) =>
268 new ThreadLaunchError({
269 operation,
270 commandId: input.commandId,
271 projectId: input.projectId,
272 ...(threadId === undefined ? {} : { threadId }),
273 cause,
274 });
275
276 const recordBranchRenameFailure = Effect.fn("ThreadLaunchService.recordBranchRenameFailure")(
277 function* (threadId: ThreadId, runId: RunId, branch: string, cause: Cause.Cause<unknown>) {
278 const error = Cause.squash(cause);
279 const now = yield* DateTime.now;
280 yield* eventSink.write({
281 events: [
282 {
283 id: yield* ids.allocate.event({ threadId }),
284 type: "turn-item.updated",
285 threadId,
286 runId,
287 occurredAt: now,
288 payload: {
289 id: ids.derive.runSignalTurnItem({ runId, signal: "branch-rename" }),
290 threadId,
291 runId,
292 nodeId: null,
293 providerThreadId: null,
294 providerTurnId: null,
295 nativeItemRef: null,
296 parentItemId: null,
297 // EventSink allocates the real position, after the run's existing items.
298 ordinal: 0,
299 status: "completed",
300 title: "Branch not renamed",
301 startedAt: now,
302 completedAt: now,
303 updatedAt: now,
304 type: "system_notice",
305 message: `Could not rename the worktree branch, so it stays ${branch}. ${error instanceof Error ? error.message : String(error)}`,
306 },
307 },
308 ],
309 });
310 },
311 Effect.catchCause((cause) =>
312 Effect.logWarning("Could not record the branch rename failure", { cause }),
313 ),
314 );
315
316 const readReceipt = (input: ThreadLaunchInput, commandId: CommandId) =>
317 receipts
318 .getByCommandId(commandId)
319 .pipe(Effect.mapError(mapError(input, "read-receipt", input.threadId)));
320
321 const validateReusableThread = Effect.fn("ThreadLaunchService.validateReusableThread")(function* (
322 input: ThreadLaunchInput,
323 threadId: ThreadId,
324 ) {
325 const projection = yield* threads
326 .getThreadRecords(threadId, ["runs"])
327 .pipe(Effect.mapError(mapError(input, "update-thread", threadId)));
328 if (
329 projection.thread.projectId !== input.projectId ||
330 projection.thread.archivedAt !== null ||
331 projection.thread.deletedAt !== null ||
332 (yield* threads
333 .getMessageCount(threadId)
334 .pipe(Effect.mapError(mapError(input, "update-thread", threadId)))) > 0 ||
335 projection.runs.length > 0
336 ) {
337 return yield* mapError(
338 input,
339 "update-thread",
340 threadId,
341 )("Only an empty active thread in the target project can change workspace during launch.");
342 }
343 });
344
345 const checkWorktreeBase: ThreadLaunchService["Service"]["checkWorktreeBase"] = Effect.fn(
346 "ThreadLaunchService.checkWorktreeBase",
347 )(function* (input) {
348 const strategy = input.workspaceStrategy;
349 // launch() turns a Scratch worktree request into a root launch.
350 if (strategy.type !== "worktree" || (yield* managedFolders.isScratchProject(input.projectId)))
351 return;
352 if (input.commandId !== undefined) {
353 const receipt = yield* receipts
354 .getByCommandId(input.commandId)
355 .pipe(Effect.orElseSucceed(() => Option.none()));
356 if (Option.isSome(receipt) && receipt.value.status === "accepted") return;
357 }
358 const project = Option.getOrUndefined(
359 yield* projects.getById(input.projectId).pipe(Effect.orElseSucceed(() => Option.none())),
360 );
361 if (project === undefined) return;
362 const cwd = project.workspaceRoot;
363 const repositoryRoot = yield* git
364 .repositoryRoot(cwd)
365 .pipe(Effect.orElseSucceed(() => undefined));
366 if (repositoryRoot === null)
367 return yield* new ThreadLaunchNotRepositoryError({
368 projectId: project.id,
369 projectTitle: project.title,
370 workspaceRoot: cwd,
371 });
372 const { baseRef, branch } = strategy;
373 const baseResolves =
374 // `:/text` is a commit search, which cannot be asked for a commit type.
375 baseRef.startsWith(":") ||
376 (yield* Effect.gen(function* () {
377 // `git worktree add` reads `-` as the previous checkout.
378 if (yield* git.hasCommit({ cwd, refName: baseRef === "-" ? "@{-1}" : baseRef }))
379 return true;
380 if (
381 strategy.startFromOrigin === true &&
382 (yield* git.resolveWorktreeBaseRemote({ cwd, refName: baseRef })) !== null
383 )
384 return true;
385 // git worktree add accepts remote-tracking branches; let Git decide when any matching ref exists.
386 if (yield* git.hasRefNamed({ cwd, refName: baseRef })) return true;
387 return !(yield* git.hasCommit({ cwd, refName: "HEAD" }));
388 }).pipe(Effect.orElseSucceed(() => true)));
389 if (!baseResolves)
390 return yield* new ThreadLaunchBaseRefError({
391 projectId: project.id,
392 projectTitle: project.title,
393 workspaceRoot: project.workspaceRoot,
394 baseRef,
395 });
396 // Provisioning flattens temporary names that collide, so only explicit names are refused.
397 if (branch === undefined || isTemporaryWorktreeBranch(branch)) return;
398 const conflictingBranch = (yield* git
399 .listLocalBranchNames(cwd)
400 .pipe(Effect.orElseSucceed((): ReadonlyArray<string> => []))).find(
401 (existing) => existing.startsWith(`${branch}/`) || branch.startsWith(`${existing}/`),
402 );
403 if (conflictingBranch !== undefined)
404 return yield* new ThreadLaunchBranchConflictError({
405 projectId: project.id,
406 projectTitle: project.title,
407 branch,
408 conflictingBranch,
409 });
410 });
411
412 const prepareInBackground = Effect.fn("ThreadLaunchService.prepareInBackground")(function* (
413 input: PreparationInput,
414 threadId: ThreadId,
415 runId: RunId | null,
416 ) {
417 const project = yield* projects.getById(input.projectId).pipe(
418 Effect.mapError(mapError(input, "resolve-project", threadId)),
419 Effect.flatMap(
420 Option.match({
421 onNone: () =>
422 Effect.fail(mapError(input, "resolve-project", threadId)("Project no longer exists.")),
423 onSome: Effect.succeed,
424 }),
425 ),
426 );
427
428 const reused = input.reusedWorktree;
429 const tracked = input.workspaceStrategy.type === "worktree" || reused !== undefined;
430 let createdWorktreePath: string | null = null;
431 let createdBranch: string | null = null;
432 let setupTerminalId: string | null = null;
433 let workspaceRecorded = false;
434 let runReleased = false;
435 // Stop background renames before forgetting worktrees; completed renames record them again.
436 let branchRename: Fiber.Fiber<unknown> | null = null;
437 if (input.workspaceStrategy.type === "worktree") {
438 yield* setupTracker.begin({
439 threadId,
440 branch: input.workspaceStrategy.branch ?? null,
441 baseRef: input.workspaceStrategy.baseRef,
442 stages: ["fetch", "checkout", "submodules", "setup-script", "agent"],
443 fiber: yield* Effect.fiber,
444 });
445 } else if (reused !== undefined) {
446 yield* setupTracker.begin({
447 threadId,
448 branch: input.workspaceStrategy.branch ?? null,
449 baseRef: reused.baseRef,
450 stages: ["setup-script", "agent"],
451 fiber: yield* Effect.fiber,
452 });
453 }
454 yield* Effect.gen(function* () {
455 const initialMessage = input.initialMessage;
456 const generateBranchNameFor = Effect.fn("ThreadLaunchService.generateBranchNameFor")(
457 function* (cwd: string, message: ThreadLaunchInitialMessage) {
458 const settings = resolveProjectSettings(
459 yield* serverSettings.getSettings,
460 input.projectId,
461 ).settings;
462 const modelSelection =
463 settings.sourceControlWriterModelSelection === null
464 ? settings.textGenerationModelSelection
465 : ServerSettings.resolveSourceControlWriterModelSelection(
466 settings,
467 yield* providerRegistry.getProviders,
468 input.projectId,
469 );
470 return yield* textGeneration
471 .generateBranchName({
472 naming: {
473 mode: settings.branchNamingMode,
474 prefix: settings.branchNamePrefix,
475 instructions: settings.branchNameInstructions,
476 },
477 cwd,
478 message: message.text,
479 attachments: message.attachments,
480 ...(message.context ? { context: message.context } : {}),
481 modelSelection,
482 projectId: input.projectId,
483 })
484 .pipe(
485 Effect.map((result) => ({
486 branch: result.branch,
487 exactName: settings.branchNamingMode === "custom",
488 })),
489 );
490 },
491 );
492
493 const requestedBranch = input.workspaceStrategy.branch;
494 const generatedTemporaryBranch =
495 input.workspaceStrategy.type === "worktree" && requestedBranch === undefined;
496 let branch: string | null;
497 if (generatedTemporaryBranch) {
498 const uuid = yield* randomUuidV4;
499 branch = buildTemporaryWorktreeBranchName(() => uuid.replaceAll("-", ""));
500 const settings = yield* serverSettings.getSettings.pipe(
501 Effect.map((settings) => resolveProjectSettings(settings, input.projectId).settings),
502 Effect.orElseSucceed(() => null),
503 );
504 if (settings?.branchNamingMode === "static") {
505 branch = formatGeneratedBranchName(branch.slice(-8), {
506 mode: "static",
507 prefix: settings.branchNamePrefix,
508 instructions: "",
509 });
510 }
511 } else {
512 branch = yield* launchBranch(input.workspaceStrategy, project.workspaceRoot);
513 }
514 let worktreePath =
515 input.workspaceStrategy.type === "existing_worktree"
516 ? input.workspaceStrategy.worktreePath
517 : null;
518 if (input.workspaceStrategy.type === "worktree") {
519 if (runId !== null) {
520 yield* threads
521 .dispatch({
522 type: "prepared-run.progress",
523 commandId: CommandId.make(`${input.commandId}:progress:worktree`),
524 threadId,
525 runId,
526 phase: "worktree",
527 })
528 .pipe(Effect.mapError(mapError(input, "update-thread", threadId)));
529 }
530 let startRef = input.workspaceStrategy.baseRef;
531 let baseRefName = input.workspaceStrategy.baseRef;
532 const remote =
533 input.workspaceStrategy.startFromOrigin === true
534 ? yield* git
535 .resolveWorktreeBaseRemote({
536 cwd: project.workspaceRoot,
537 refName: input.workspaceStrategy.baseRef,
538 })
539 .pipe(Effect.mapError(mapError(input, "provision-worktree", threadId)))
540 : null;
541 const shouldFetch = remote !== null;
542 yield* setupTracker.stageStatus(threadId, "fetch", shouldFetch ? "running" : "skipped");
543 if (remote !== null) {
544 const remoteBase = yield* git
545 .fetchWorktreeBaseCommit({
546 cwd: project.workspaceRoot,
547 refName: input.workspaceStrategy.baseRef,
548 remote,
549 })
550 .pipe(Effect.mapError(mapError(input, "provision-worktree", threadId)));
551 if (remoteBase !== null) {
552 startRef = remoteBase.commitSha;
553 baseRefName = `refs/remotes/${remoteBase.remoteRefName}`;
554 }
555 }
556 if (shouldFetch) yield* setupTracker.stageStatus(threadId, "fetch", "done");
557 if (branch !== null && (generatedTemporaryBranch || isTemporaryWorktreeBranch(branch))) {
558 const parts = branch.split("/");
559 for (let depth = 1; depth < parts.length; depth++) {
560 if (
561 yield* git
562 .hasCommit({
563 cwd: project.workspaceRoot,
564 refName: `refs/heads/${parts.slice(0, depth).join("/")}`,
565 })
566 .pipe(Effect.mapError(mapError(input, "provision-worktree", threadId)))
567 ) {
568 branch = generatedTemporaryBranch
569 ? branch.replaceAll("/", "-")
570 : flattenTemporaryWorktreeBranchName(branch);
571 break;
572 }
573 }
574 }
575 yield* setupTracker.stageStatus(threadId, "checkout", "running");
576 let submoduleProgressReportedAt = Number.NEGATIVE_INFINITY;
577 const worktree = yield* git
578 .createWorktree(
579 {
580 cwd: project.workspaceRoot,
581 refName: startRef,
582 newRefName: branch!,
583 baseRefName,
584 path: null,
585 },
586 {
587 progress: {
588 onWorktreeClaimed: (path) =>
589 Effect.sync(() => {
590 createdWorktreePath = path;
591 createdBranch = branch;
592 }),
593 onCheckoutProgress: (progress) =>
594 setupTracker.stage(threadId, "checkout", { percent: progress.percent }),
595 onSubmodulesStarted: () =>
596 setupTracker
597 .stageStatus(threadId, "checkout", "done")
598 .pipe(
599 Effect.andThen(setupTracker.stageStatus(threadId, "submodules", "running")),
600 ),
601 onSubmodulesDisabled: ({ source }) =>
602 setupTracker.stageStatus(
603 threadId,
604 "submodules",
605 "skipped",
606 `disabled in ${source}`,
607 ),
608 onSubmoduleLine: (line) => {
609 // Paths may contain quotes; git closes the path with "': ".
610 const submodulePath = /^\s*Submodule path '(.+)': /.exec(line)?.[1];
611 if (submodulePath !== undefined) {
612 return setupTracker.stage(threadId, "submodules", { detail: submodulePath });
613 }
614 if (parseGitCloneProgressLine(line) === null) return Effect.void;
615 // Git redraws on every percent; hold the websocket to one update a second.
616 return Effect.flatMap(Clock.currentTimeMillis, (now) => {
617 if (now - submoduleProgressReportedAt < 1000) return Effect.void;
618 submoduleProgressReportedAt = now;
619 return setupTracker.stage(threadId, "submodules", {
620 detail: line.trim().replace(/\s+/g, " "),
621 });
622 });
623 },
624 onSubmodulesFinished: ({ ok, detail }) =>
625 setupTracker.stageStatus(
626 threadId,
627 "submodules",
628 ok ? "done" : "warning",
629 ok ? undefined : (detail ?? "submodule checkout failed"),
630 ),
631 },
632 },
633 )
634 .pipe(Effect.mapError(mapError(input, "provision-worktree", threadId)));
635 worktreePath = worktree.worktree.path;
636 branch = worktree.worktree.refName;
637 createdWorktreePath = worktreePath;
638 createdBranch = branch;
639 yield* setupTracker.update(threadId, (snapshot) => ({
640 ...snapshot,
641 worktreePath,
642 branch,
643 stages: snapshot.stages.map((stage) =>
644 stage.id === "submodules" && stage.status === "pending"
645 ? { ...stage, status: "skipped", detail: "none" }
646 : stage,
647 ),
648 }));
649 yield* setupTracker.stageStatus(threadId, "checkout", "done");
650 }
651
652 // Do not rewrite reused worktrees; that could undo the original branch rename.
653 if (reused === undefined) {
654 yield* threads
655 .dispatch({
656 type: "thread.metadata.update",
657 commandId: CommandId.make(`${input.commandId}:workspace`),
658 threadId,
659 branch,
660 worktreePath,
661 })
662 .pipe(Effect.mapError(mapError(input, "update-thread", threadId)));
663 }
664 workspaceRecorded = true;
665
666 if (
667 reused === undefined &&
668 worktreePath !== null &&
669 branch !== null &&
670 initialMessage !== undefined &&
671 (generatedTemporaryBranch || isTemporaryWorktreeBranch(branch))
672 ) {
673 const oldBranch = branch;
674 const worktreeCwd = worktreePath;
675 branchRename = yield* generateBranchNameFor(worktreeCwd, initialMessage).pipe(
676 Effect.flatMap(({ branch: newBranch, exactName }) =>
677 git
678 .renameBranch({
679 cwd: worktreeCwd,
680 oldBranch,
681 newBranch,
682 ...(exactName ? { exactName: true } : {}),
683 })
684 .pipe(
685 // A cancel must see the renamed branch, or its cleanup deletes the old name.
686 Effect.tap((renamed) =>
687 Effect.sync(() => {
688 createdBranch = renamed.branch;
689 }),
690 ),
691 Effect.uninterruptible,
692 ),
693 ),
694 Effect.flatMap((renamed) =>
695 Effect.gen(function* () {
696 yield* threads
697 .dispatch({
698 type: "thread.metadata.update",
699 commandId: CommandId.make(`${input.commandId}:branch-rename`),
700 threadId,
701 branch: renamed.branch,
702 worktreePath: worktreeCwd,
703 expectedBranch: oldBranch,
704 expectedWorktreePath: worktreeCwd,
705 })
706 .pipe(
707 Effect.catchCause((cause) =>
708 Effect.logWarning("Parent thread worktree branch update failed", {
709 threadId,
710 cause,
711 }),
712 ),
713 );
714 // Update parents first so children inherit names; reread each pass for newly updated descendants.
715 const visited = new Set([threadId]);
716 while (true) {
717 const snapshot = yield* threads.getShellSnapshot({ worktreePath: worktreeCwd });
718 const sharing = [...snapshot.threads, ...snapshot.archivedThreads].filter(
719 (thread) => !visited.has(thread.id) && thread.branch === oldBranch,
720 );
721 if (sharing.length === 0) break;
722 for (const thread of sharing) {
723 visited.add(thread.id);
724 yield* threads
725 .dispatch({
726 type: "thread.metadata.update",
727 commandId: CommandId.make(`${input.commandId}:branch-rename:${thread.id}`),
728 threadId: thread.id,
729 branch: renamed.branch,
730 expectedBranch: oldBranch,
731 expectedWorktreePath: worktreeCwd,
732 })
733 .pipe(
734 Effect.catchCause((cause) =>
735 Effect.logWarning("Shared thread worktree branch update failed", {
736 threadId: thread.id,
737 cause,
738 }),
739 ),
740 );
741 }
742 }
743 }),
744 ),
745 Effect.catchCause((cause) =>
746 Effect.logWarning("Thread worktree branch rename failed", {
747 commandId: input.commandId,
748 threadId,
749 oldBranch,
750 cause,
751 }).pipe(
752 Effect.andThen(
753 runId === null || Cause.hasInterruptsOnly(cause)
754 ? Effect.void
755 : recordBranchRenameFailure(threadId, runId, oldBranch, cause),
756 ),
757 ),
758 ),
759 Effect.forkIn(preparationScope),
760 );
761 }
762
763 if (runId !== null && worktreePath !== null) {
764 yield* threads
765 .dispatch({
766 type: "prepared-run.progress",
767 commandId: CommandId.make(`${input.commandId}:progress:setup`),
768 threadId,
769 runId,
770 phase: "setup",
771 })
772 .pipe(Effect.mapError(mapError(input, "update-thread", threadId)));
773 }
774 yield* setupTracker.stageStatus(threadId, "setup-script", "running");
775 const setup =
776 worktreePath === null
777 ? ({ status: "no-script" } as const)
778 : yield* setupScripts
779 .runForThread({
780 threadId,
781 projectId: input.projectId,
782 projectCwd: project.workspaceRoot,
783 worktreePath,
784 ...(tracked
785 ? {
786 observeCompletion: {
787 onOutputLine: (line: string) =>
788 setupTracker.appendTail(threadId, "setup-script", line),
789 },
790 }
791 : {}),
792 project: {
793 id: project.id,
794 workspaceRoot: project.workspaceRoot,
795 scripts: project.scripts,
796 },
797 })
798 .pipe(Effect.mapError(mapError(input, "run-setup-script", threadId)));
799
800 let awaitAsyncSetup = Effect.void;
801 if (setup.status === "started") {
802 setupTerminalId = setup.terminalId;
803 yield* setupTracker.update(threadId, (snapshot) => ({
804 ...snapshot,
805 setupScript: {
806 name: setup.scriptName,
807 command: setup.scriptCommand,
808 terminalId: setup.terminalId,
809 },
810 }));
811 if (setup.completion) {
812 const awaitCompletion = Effect.gen(function* () {
813 const completion = yield* setup.completion!;
814 yield* setupTracker.stage(threadId, "setup-script", {
815 status: completion.exitCode === 0 ? "done" : "failed",
816 detail: `exited with ${completion.exitCode ?? "no exit code"}`,
817 });
818 if (completion.exitCode !== 0 && !setup.async)
819 return yield* mapError(
820 input,
821 "run-setup-script",
822 threadId,
823 )(`Setup script exited with ${completion.exitCode ?? "no exit code"}.`);
824 });
825 if (setup.async) {
826 awaitAsyncSetup = awaitCompletion.pipe(
827 Effect.catchCause((cause) =>
828 setupTracker.stage(threadId, "setup-script", {
829 status: "failed",
830 detail: failureDetail(Cause.squash(cause)),
831 }),
832 ),
833 );
834 } else {
835 yield* awaitCompletion;
836 }
837 } else {
838 yield* setupTracker.stageStatus(threadId, "setup-script", "done");
839 }
840 } else {
841 yield* setupTracker.stageStatus(threadId, "setup-script", "skipped");
842 }
843 // The agent must start on the branch name the thread will show.
844 if (branchRename !== null) yield* Fiber.join(branchRename);
845 yield* setupTracker.markUncancellable(threadId);
846 yield* setupTracker.stageStatus(threadId, "agent", "running");
847 if (runId !== null) {
848 yield* threads
849 .dispatch({
850 type: "prepared-run.release",
851 commandId: CommandId.make(`${input.commandId}:release`),
852 threadId,
853 runId,
854 })
855 .pipe(
856 Effect.mapError(mapError(input, "release-run", threadId)),
857 Effect.tap(() =>
858 Effect.sync(() => {
859 runReleased = true;
860 }),
861 ),
862 Effect.uninterruptible,
863 );
864 }
865 yield* setupTracker.stageStatus(threadId, "agent", "done");
866 yield* awaitAsyncSetup;
867 yield* setupTracker.finish(threadId, "done");
868 }).pipe(
869 Effect.onError((cause) =>
870 Effect.gen(function* () {
871 const cancelled = Cause.hasInterruptsOnly(cause);
872 yield* setupTracker.finish(
873 threadId,
874 cancelled ? "cancelled" : "failed",
875 cancelled ? null : failureDetail(Cause.squash(cause)),
876 );
877 if (cancelled && setupTerminalId)
878 yield* terminals
879 .close({ threadId, terminalId: setupTerminalId, deleteHistory: true })
880 .pipe(Effect.ignore);
881 if (
882 tracked &&
883 createdWorktreePath &&
884 ((cancelled && !runReleased) || !workspaceRecorded)
885 ) {
886 if (branchRename !== null) yield* Fiber.interrupt(branchRename);
887 const removedPath = createdWorktreePath;
888 const removedBranch = createdBranch;
889 // Forget worktrees only after removal succeeds; failed removals must not reuse partially written checkouts.
890 yield* git
891 .removeWorktree({ cwd: project.workspaceRoot, path: removedPath, force: true })
892 .pipe(
893 Effect.andThen(
894 removedBranch === null
895 ? Effect.void
896 : git
897 .deleteLocalBranch({
898 cwd: project.workspaceRoot,
899 refName: removedBranch,
900 force: true,
901 })
902 .pipe(Effect.ignoreCause({ log: true })),
903 ),
904 Effect.andThen(
905 threads
906 .dispatch({
907 type: "thread.metadata.update",
908 commandId: CommandId.make(`${input.commandId}:cancel-workspace`),
909 threadId,
910 worktreePath: null,
911 branch: null,
912 })
913 .pipe(Effect.ignore),
914 ),
915 Effect.catchCause((removeCause) =>
916 Effect.logWarning("Failed to remove an abandoned thread worktree", {
917 commandId: input.commandId,
918 threadId,
919 path: removedPath,
920 cause: removeCause,
921 }),
922 ),
923 );
924 }
925 }),
926 ),
927 );
928 });
929
930 const failPreparedRun = (
931 input: Pick<PreparationInput, "commandId">,
932 threadId: ThreadId,
933 runId: RunId | null,
934 cause: unknown,
935 ) =>
936 runId === null
937 ? Effect.logWarning("Thread workspace preparation failed", {
938 commandId: input.commandId,
939 threadId,
940 cause,
941 })
942 : threads
943 .dispatch({
944 type: "prepared-run.fail",
945 commandId: CommandId.make(`${input.commandId}:fail`),
946 threadId,
947 runId,
948 failure: makeProviderFailure({
949 cause,
950 message: failureDetail(cause),
951 class: "validation_error",
952 retryable: false,
953 }),
954 })
955 .pipe(
956 Effect.catchCause((persistCause) =>
957 Effect.logWarning("Failed to persist thread workspace preparation failure", {
958 commandId: input.commandId,
959 threadId,
960 cause,
961 persistCause,
962 }),
963 ),
964 );
965
966 const reservePreparation = (commandId: CommandId) =>
967 Ref.modify(scheduledLaunches, (scheduled) => {
968 if (scheduled.has(commandId)) return [false, scheduled] as const;
969 const next = new Set(scheduled);
970 next.add(commandId);
971 return [true, next] as const;
972 });
973
974 const releasePreparation = (commandId: CommandId) =>
975 Ref.update(scheduledLaunches, (scheduled) => {
976 const next = new Set(scheduled);
977 next.delete(commandId);
978 return next;
979 });
980
981 const schedulePreparation = Effect.fn("ThreadLaunchService.schedulePreparation")(function* (
982 input: PreparationInput,
983 threadId: ThreadId,
984 runId: RunId | null,
985 ) {
986 const preparation = Effect.gen(function* () {
987 const prepare = prepareInBackground(input, threadId, runId);
988 if (
989 runId === null ||
990 (input.workspaceStrategy.type !== "worktree" && input.reusedWorktree === undefined)
991 ) {
992 return yield* prepare;
993 }
994 const afterSequence = yield* threads.getThreadEventSequence(threadId);
995 const projection = yield* threads.getThreadRecords(threadId, ["runs"]);
996 if (projection.runs.find((run) => run.id === runId)?.status !== "preparing") return;
997 const interruption = threads
998 .streamStoredEventsFrom({ threadId, afterSequence, eventType: "run.updated" })
999 .pipe(
1000 Stream.filter(
1001 (stored) =>
1002 stored.event.type === "run.updated" &&
1003 stored.event.payload.id === runId &&
1004 (stored.event.payload.status === "interrupted" ||
1005 stored.event.payload.status === "cancelled"),
1006 ),
1007 Stream.runHead,
1008 Effect.flatMap((event) => (Option.isSome(event) ? Effect.interrupt : Effect.never)),
1009 );
1010 yield* Effect.raceFirst(prepare, interruption);
1011 });
1012 yield* preparation.pipe(
1013 Effect.onError((cause) =>
1014 failPreparedRun(
1015 input,
1016 threadId,
1017 runId,
1018 Cause.hasInterruptsOnly(cause) ? "Worktree setup cancelled." : Cause.squash(cause),
1019 ),
1020 ),
1021 Effect.ignoreCause,
1022 Effect.ensuring(releasePreparation(input.commandId)),
1023 Effect.forkIn(preparationScope),
1024 );
1025 });
1026
1027 const launch: ThreadLaunchService["Service"]["launch"] = Effect.fn("ThreadLaunchService.launch")(
1028 function* (request) {
1029 // Scratch has no Git repository. Normalize before folder allocation so
1030 // durable preparation and retries use the thread's own Scratch folder.
1031 const input: ThreadLaunchInput =
1032 request.workspaceStrategy.type === "worktree" &&
1033 (yield* managedFolders.isScratchProject(request.projectId))
1034 ? { ...request, workspaceStrategy: { type: "root" } }
1035 : request;
1036 yield* ProjectCloneTracker.rejectCommandsDuringClone(cloneTracker, {
1037 type: "thread.create",
1038 projectId: input.projectId,
1039 }).pipe(Effect.mapError(mapError(input, "resolve-project")));
1040 const project = yield* projects.getById(input.projectId).pipe(
1041 Effect.mapError(mapError(input, "resolve-project")),
1042 Effect.flatMap(
1043 Option.match({
1044 onNone: () => Effect.fail(mapError(input, "resolve-project")("Project not found.")),
1045 onSome: Effect.succeed,
1046 }),
1047 ),
1048 );
1049 if (input.reuseExistingThread === true && input.threadId === undefined) {
1050 return yield* mapError(
1051 input,
1052 "update-thread",
1053 )("Reusing an existing thread requires a thread id.");
1054 }
1055
1056 const launchReceipt = yield* readReceipt(input, input.commandId);
1057 yield* threads.assertProviderInstanceEnabled({
1058 projectId: input.projectId,
1059 commandId: input.commandId,
1060 instanceId: input.modelSelection.instanceId,
1061 });
1062 return yield* Effect.gen(function* () {
1063 // Recover ids from accepted launch receipts before allocating; retries have no client id and new ids collide with receipts.
1064 const reusableLaunchReceipt =
1065 input.threadId === undefined &&
1066 Option.isSome(launchReceipt) &&
1067 launchReceipt.value.status === "accepted" &&
1068 launchReceipt.value.commandType === "thread.create"
1069 ? launchReceipt.value
1070 : undefined;
1071 const candidateThreadId =
1072 input.threadId ??
1073 reusableLaunchReceipt?.threadId ??
1074 (yield* ids.allocate
1075 .thread({ projectId: input.projectId })
1076 .pipe(Effect.mapError(mapError(input, "create-thread"))));
1077
1078 if (reusableLaunchReceipt !== undefined) {
1079 const shell = yield* threads
1080 .getThreadShell(candidateThreadId)
1081 .pipe(Effect.mapError(mapError(input, "create-thread", candidateThreadId)));
1082 if (shell === null) {
1083 return yield* mapError(input, "create-thread", candidateThreadId)("Thread not found.");
1084 }
1085 if (shell.projectId !== input.projectId) {
1086 return yield* mapError(
1087 input,
1088 "resolve-project",
1089 candidateThreadId,
1090 )("Project identity changed.");
1091 }
1092 }
1093
1094 if (input.reuseExistingThread === true && Option.isNone(launchReceipt)) {
1095 yield* validateReusableThread(input, candidateThreadId);
1096 }
1097
1098 const workspaceStrategy: ThreadLaunchWorkspaceStrategy =
1099 input.workspaceStrategy.type === "root" && Option.isNone(launchReceipt)
1100 ? Option.match(
1101 yield* managedFolders
1102 .folderForThread({
1103 projectId: input.projectId,
1104 threadId: candidateThreadId,
1105 text: input.initialMessage?.text ?? input.title,
1106 })
1107 .pipe(Effect.mapError(mapError(input, "provision-worktree", candidateThreadId))),
1108 {
1109 onNone: () => input.workspaceStrategy,
1110 onSome: (worktreePath) => ({ type: "existing_worktree", worktreePath }),
1111 },
1112 )
1113 : input.workspaceStrategy;
1114 const initialBranch = yield* launchBranch(workspaceStrategy, project.workspaceRoot);
1115 const initialWorktreePath =
1116 workspaceStrategy.type === "existing_worktree" ? workspaceStrategy.worktreePath : null;
1117 const claimDispatch =
1118 input.reuseExistingThread === true
1119 ? threads.dispatch({
1120 type: "thread.metadata.update",
1121 commandId: input.commandId,
1122 threadId: candidateThreadId,
1123 expectedEmpty: true,
1124 })
1125 : threads.dispatch({
1126 type: "thread.create",
1127 commandId: input.commandId,
1128 threadId: candidateThreadId,
1129 projectId: input.projectId,
1130 title: input.title,
1131 modelSelection: input.modelSelection,
1132 runtimeMode: input.runtimeMode,
1133 interactionMode: input.interactionMode,
1134 branch: initialBranch,
1135 worktreePath: initialWorktreePath,
1136 ...(input.importedNativeThread === undefined
1137 ? {}
1138 : { importedNativeThread: input.importedNativeThread }),
1139 ...(input.linkOrigin === undefined ? {} : { linkOrigin: input.linkOrigin }),
1140 ...(input.delegatedFrom === undefined
1141 ? {}
1142 : { delegatedFrom: input.delegatedFrom }),
1143 createdBy: input.createdBy,
1144 creationSource: input.creationSource,
1145 });
1146 const claimed = yield* claimDispatch.pipe(
1147 Effect.mapError(
1148 mapError(
1149 input,
1150 input.reuseExistingThread === true ? "update-thread" : "create-thread",
1151 candidateThreadId,
1152 ),
1153 ),
1154 );
1155 const threadId =
1156 claimed.storedEvents.find((stored) => stored.event.type.startsWith("thread."))?.event
1157 .threadId ?? candidateThreadId;
1158 if (project.id !== input.projectId) {
1159 return yield* mapError(input, "resolve-project", threadId)("Project identity changed.");
1160 }
1161 let runId: RunId | null = null;
1162 let messageSequence: number | undefined;
1163 let messageWasAlreadyAccepted = false;
1164 if (input.initialMessage !== undefined) {
1165 const messageCommandId = CommandId.make(`${input.commandId}:initial-message`);
1166 const messageReceipt = yield* readReceipt(input, messageCommandId);
1167 messageWasAlreadyAccepted = Option.isSome(messageReceipt);
1168 const messageId =
1169 input.initialMessage.messageId ??
1170 (yield* ids.allocate
1171 .message({ threadId, ordinal: 1 })
1172 .pipe(Effect.mapError(mapError(input, "dispatch-message", threadId))));
1173 const dispatched = yield* threads
1174 .dispatch({
1175 type: "message.dispatch",
1176 commandId: messageCommandId,
1177 threadId,
1178 messageId,
1179 text: input.initialMessage.text,
1180 ...(input.initialMessage.scheduledTaskId === undefined
1181 ? {}
1182 : { scheduledTaskId: input.initialMessage.scheduledTaskId }),
1183 ...(input.initialMessage.senderThreadId === undefined
1184 ? {}
1185 : { senderThreadId: input.initialMessage.senderThreadId }),
1186 attachments: input.initialMessage.attachments,
1187 ...(input.initialMessage.context ? { context: input.initialMessage.context } : {}),
1188 ...(input.generateTitle === true ? { titleSeed: input.title } : {}),
1189 modelSelection: input.modelSelection,
1190 dispatchMode: { type: "defer_start", workspaceStrategy },
1191 createdBy: input.createdBy,
1192 creationSource: input.creationSource,
1193 })
1194 .pipe(
1195 // An agent's launch can land on a thread that a replay, or a concurrent
1196 // launch with the same key, created with broader modes. The orchestrator
1197 // refuses the message under the thread's lock; the user's own clients
1198 // launch into their own drafts, whatever modes those hold.
1199 input.createdBy === "agent"
1200 ? Effect.provideService(DispatchModeLimit, {
1201 runtimeMode: input.runtimeMode,
1202 interactionMode: input.interactionMode,
1203 })
1204 : (effect) => effect,
1205 Effect.mapError(mapError(input, "dispatch-message", threadId)),
1206 );
1207 // A concurrent retry can read no receipt and still replay the dispatch.
1208 messageWasAlreadyAccepted ||= dispatched.replayed === true;
1209 messageSequence = dispatched.sequence;
1210 const runCreated = dispatched.storedEvents.find(
1211 (stored) => stored.event.type === "run.created",
1212 );
1213 runId = runCreated?.event.type === "run.created" ? runCreated.event.payload.id : null;
1214 if (runId === null) {
1215 return yield* mapError(
1216 input,
1217 "dispatch-message",
1218 threadId,
1219 )("Initial message was accepted without a durable run.");
1220 }
1221 }
1222
1223 const projection = yield* threads
1224 .getThreadProjection(threadId)
1225 .pipe(Effect.mapError(mapError(input, "create-thread", threadId)));
1226 const runIsPreparing =
1227 runId !== null &&
1228 projection.runs.some((run) => run.id === runId && run.status === "preparing");
1229 const shouldSchedule = runId === null ? Option.isNone(launchReceipt) : runIsPreparing;
1230 const boundWorktreePath = projection.thread.worktreePath;
1231 const preparationStrategy: ThreadLaunchWorkspaceStrategy =
1232 Option.isSome(launchReceipt) &&
1233 workspaceStrategy.type === "root" &&
1234 boundWorktreePath !== null
1235 ? {
1236 type: "existing_worktree",
1237 worktreePath: boundWorktreePath,
1238 branch: workspaceStrategy.branch,
1239 }
1240 : workspaceStrategy;
1241 if (shouldSchedule) {
1242 const ownsPreparation = yield* reservePreparation(input.commandId);
1243 if (ownsPreparation) {
1244 yield* Effect.gen(function* () {
1245 const preparationStillRequired =
1246 runId === null
1247 ? true
1248 : yield* threads.getThreadRecords(threadId, ["runs"], { runIds: [runId] }).pipe(
1249 Effect.map((current) =>
1250 current.runs.some((run) => run.id === runId && run.status === "preparing"),
1251 ),
1252 Effect.mapError(mapError(input, "update-thread", threadId)),
1253 );
1254 if (preparationStillRequired) {
1255 yield* schedulePreparation(
1256 { ...input, workspaceStrategy: preparationStrategy },
1257 threadId,
1258 runId,
1259 );
1260 } else {
1261 yield* releasePreparation(input.commandId);
1262 }
1263 }).pipe(Effect.onError(() => releasePreparation(input.commandId)));
1264 }
1265 }
1266
1267 return {
1268 threadId,
1269 projection,
1270 // An attempt that failed after claiming the thread recorded nothing, so
1271 // only a repeat of the final write counts as resumed.
1272 resumed:
1273 input.initialMessage === undefined
1274 ? Option.isSome(launchReceipt) || claimed.replayed === true
1275 : messageWasAlreadyAccepted,
1276 ...(messageSequence === undefined ? {} : { sequence: messageSequence }),
1277 };
1278 });
1279 },
1280 );
1281
1282 const prepareRun: ThreadLaunchService["Service"]["prepareRun"] = Effect.fn(
1283 "ThreadLaunchService.prepareRun",
1284 )(function* (input) {
1285 const scheduled = yield* Effect.gen(function* () {
1286 const { thread, runs } = yield* threads.getThreadRecords(input.threadId, ["runs"], {
1287 runIds: [input.runId],
1288 });
1289 const run = runs.find((candidate) => candidate.id === input.runId);
1290 const workspacePreparation = run?.workspacePreparation;
1291 if (run?.status !== "preparing" || workspacePreparation === undefined) return;
1292 if (!(yield* reservePreparation(input.commandId))) return;
1293 yield* scheduleRunPreparation(input, thread, run, workspacePreparation).pipe(
1294 Effect.onError(() => releasePreparation(input.commandId)),
1295 );
1296 }).pipe(Effect.exit);
1297 if (Exit.isFailure(scheduled)) {
1298 yield* failPreparedRun(input, input.threadId, input.runId, Cause.squash(scheduled.cause));
1299 }
1300 });
1301
1302 const retryPreparation: ThreadLaunchService["Service"]["retryPreparation"] = Effect.fn(
1303 "ThreadLaunchService.retryPreparation",
1304 )(function* (input) {
1305 const dispatched = yield* threads.dispatch({
1306 type: "prepared-run.retry",
1307 commandId: input.commandId,
1308 threadId: input.threadId,
1309 runId: input.runId,
1310 });
1311 yield* prepareRun(input);
1312 return dispatched;
1313 });
1314
1315 const scheduleRunPreparation = Effect.fn("ThreadLaunchService.scheduleRunPreparation")(function* (
1316 input: ThreadLaunchPreparationInput,
1317 thread: OrchestrationV2ThreadProjection["thread"],
1318 run: OrchestrationV2ThreadProjection["runs"][number],
1319 workspacePreparation: ThreadLaunchWorkspaceStrategy,
1320 ) {
1321 const { messages } = yield* threads.getThreadRecords(input.threadId, ["messages"], {
1322 messageIds: [run.userMessageId],
1323 });
1324 const message = messages.find((candidate) => candidate.id === run.userMessageId);
1325 const reuse =
1326 workspacePreparation.type === "worktree" &&
1327 thread.worktreePath !== null &&
1328 thread.branch !== null
1329 ? {
1330 strategy: {
1331 type: "existing_worktree" as const,
1332 worktreePath: thread.worktreePath,
1333 branch: thread.branch,
1334 },
1335 reusedWorktree: { baseRef: workspacePreparation.baseRef },
1336 }
1337 : null;
1338 return yield* schedulePreparation(
1339 {
1340 commandId: input.commandId,
1341 projectId: thread.projectId,
1342 workspaceStrategy: reuse?.strategy ?? workspacePreparation,
1343 ...(reuse === null ? {} : { reusedWorktree: reuse.reusedWorktree }),
1344 ...(message === undefined
1345 ? {}
1346 : {
1347 initialMessage: {
1348 text: message.text,
1349 attachments: message.attachments,
1350 ...(message.context ? { context: message.context } : {}),
1351 },
1352 }),
1353 },
1354 input.threadId,
1355 run.id,
1356 );
1357 });
1358
1359 return ThreadLaunchService.of({ launch, checkWorktreeBase, retryPreparation, prepareRun });
1360});
1361
1362export const layer = Layer.effect(ThreadLaunchService, make);