clover's git

the-knife

1	import * as WorktreeSetupTracker from "../project/WorktreeSetupTracker.ts";
2	import * as ProjectCloneTracker from "../project/ProjectCloneTracker.ts";
3	import * as TerminalManager from "../terminal/Manager.ts";
4	import { resolveProjectSettings } from "@t3tools/shared/projectSettings";
5	import {
6	CommandId,
7	type ChatAttachment,
8	type MessageId,
9	type ModelSelection,
10	type OrchestrationV2Actor,
11	type OrchestrationV2CreationSource,
12	type OrchestrationV2ProviderThreadNativeMetadata,
13	type OrchestrationV2ThreadProjection,
14	type ProviderDriverKind,
15	type ProviderInteractionMode,
16	ProjectId,
17	type RunId,
18	type RuntimeMode,
19	type ScheduledTaskId,
20	ThreadId,
21	} from "@t3tools/contracts";
22	import * as Cause from "effect/Cause";
23	import * as Context from "effect/Context";
24	import * as Effect from "effect/Effect";
25	import * as Exit from "effect/Exit";
26	import * as Layer from "effect/Layer";
27	import * as Option from "effect/Option";
28	import * as Ref from "effect/Ref";
29	import * as Schema from "effect/Schema";
30	import * as Scope from "effect/Scope";
31	import * as Stream from "effect/Stream";
32	import {
33	buildTemporaryWorktreeBranchName,
34	flattenTemporaryWorktreeBranchName,
35	isTemporaryWorktreeBranch,
36	WORKTREE_BRANCH_PREFIX,
37	} from "@t3tools/shared/git";
38
39	import * as GitWorkflow from "../git/GitWorkflowService.ts";
40	import * as ProjectService from "../project/ProjectService.ts";
41	import * as ProjectSetupScriptRunner from "../project/ProjectSetupScriptRunner.ts";
42	import * as ManagedProjectFolders from "../project/ManagedProjectFolders.ts";
43	import * as ProviderRegistry from "../provider/ProviderRegistry.ts";
44	import * as ServerSettings from "../serverSettings.ts";
45	import * as TextGeneration from "../textGeneration/TextGeneration.ts";
46	import * as CommandReceiptStore from "./CommandReceiptStore.ts";
47	import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
48	import type * as Orchestrator from "./Orchestrator.ts";
49	import { makeProviderFailure } from "@t3tools/provider-core/server/failure";
50	import { randomUuidV4 } from "@t3tools/provider-core/server/randomUuid";
51	import * as ThreadManagement from "./ThreadManagementService.ts";
52
53	export type ThreadLaunchWorkspaceStrategy =
54	| { readonly type: "root"; readonly branch?: string | undefined }
55	| {
56	readonly type: "existing_worktree";
57	readonly worktreePath: string;
58	readonly branch?: string | undefined;
59	}
60	| {
61	readonly type: "worktree";
62	readonly baseRef: string;
63	readonly branch?: string | undefined;
64	readonly startFromOrigin?: boolean | undefined;
65	};
66
67	export interface ThreadLaunchInitialMessage {
68	readonly messageId?: MessageId;
69	readonly scheduledTaskId?: ScheduledTaskId;
70	readonly senderThreadId?: ThreadId;
71	readonly text: string;
72	readonly attachments: ReadonlyArray<ChatAttachment>;
73	readonly context?: import("@t3tools/contracts").OrchestrationMessageContext | undefined;
74	}
75
76	export interface ThreadLaunchInput {
77	readonly commandId: CommandId;
78	readonly threadId?: ThreadId;
79	readonly reuseExistingThread?: boolean;
80	readonly projectId: ProjectId;
81	readonly title: string;
82	readonly generateTitle?: boolean;
83	readonly modelSelection: ModelSelection;
84	readonly runtimeMode: RuntimeMode;
85	readonly interactionMode: ProviderInteractionMode;
86	readonly workspaceStrategy: ThreadLaunchWorkspaceStrategy;
87	readonly initialMessage?: ThreadLaunchInitialMessage;
88	readonly importedNativeThread?: {
89	readonly ref: {
90	readonly driver: ProviderDriverKind;
91	readonly nativeId: string;
92	readonly strength: "strong";
93	};
94	readonly metadata?: OrchestrationV2ProviderThreadNativeMetadata;
95	};
96	readonly createdBy: OrchestrationV2Actor;
97	readonly creationSource: OrchestrationV2CreationSource;
98	}
99
100	/** What workspace preparation reads from a launch; a retry rebuilds it from the run. */
101	type PreparationInput = Pick<
102	ThreadLaunchInput,
103	"commandId" | "projectId" | "workspaceStrategy" | "initialMessage"
104	> & {
105	/**
106	* Set when a retry reuses the worktree its failed attempt created and
107	* recorded. Its setup is tracked like a new one, but the thread already
108	* records the workspace, and a branch rename may still be running.
109	*/
110	readonly reusedWorktree?: { readonly baseRef: string };
111	};
112
113	export interface ThreadLaunchPreparationInput {
114	readonly commandId: CommandId;
115	readonly threadId: ThreadId;
116	readonly runId: RunId;
117	}
118
119	export interface ThreadLaunchResult {
120	readonly threadId: ThreadId;
121	readonly projection: OrchestrationV2ThreadProjection;
122	readonly resumed: boolean;
123	}
124
125	export class ThreadLaunchError extends Schema.TaggedError<ThreadLaunchError>()(
126	"ThreadLaunchError",
127	{
128	operation: Schema.Literals([
129	"resolve-project",
130	"read-receipt",
131	"generate-metadata",
132	"provision-worktree",
133	"run-setup-script",
134	"create-thread",
135	"update-thread",
136	"dispatch-message",
137	"release-run",
138	"fail-run",
139	]),
140	commandId: CommandId,
141	projectId: ProjectId,
142	threadId: Schema.optional(ThreadId),
143	cause: Schema.Defect(),
144	},
145	) {
146	override get message(): string {
147	return `Thread launch ${this.commandId} failed during ${this.operation}.`;
148	}
149	}
150
151	/** A new worktree was asked to start from a ref that gives it no commit. */
152	export class ThreadLaunchBaseRefError extends Schema.TaggedError<ThreadLaunchBaseRefError>()(
153	"ThreadLaunchBaseRefError",
154	{
155	projectId: ProjectId,
156	projectTitle: Schema.String,
157	workspaceRoot: Schema.String,
158	baseRef: Schema.String,
159	},
160	) {
161	override get message(): string {
162	return `Base ref "${this.baseRef}" does not resolve to a commit in project "${this.projectTitle}" (${this.workspaceRoot}).`;
163	}
164	}
165
166	export class ThreadLaunchService extends Context.Service<
167	ThreadLaunchService,
168	{
169	readonly launch: (
170	input: ThreadLaunchInput,
171	) => Effect.Effect<
172	ThreadLaunchResult,
173	ThreadLaunchError | ThreadManagement.ThreadManagementProviderInstanceDisabledError
174	>;
175	/**
176	* Fails when a new worktree's base ref cannot give provisioning a commit.
177	* `launch` accepts such a launch and reports the failure inside the
178	* thread; a caller that needs it before a thread exists checks first.
179	*/
180	readonly checkWorktreeBase: (
181	input: Pick<ThreadLaunchInput, "projectId" | "workspaceStrategy">,
182	) => Effect.Effect<void, ThreadLaunchBaseRefError>;
183	/** Dispatches prepared-run.retry and prepares the run's workspace again. */
184	readonly retryPreparation: (input: ThreadLaunchPreparationInput) => Effect.Effect<
185	Orchestrator.OrchestratorV2DispatchResult,
186	// Dispatches prepared-run.retry, a command the enablement chokepoint
187	// never gates; the type stays accurate anyway.
188	| Orchestrator.OrchestratorV2Error
189	| ThreadManagement.ThreadManagementProviderInstanceDisabledError
190	>;
191	readonly prepareRun: (input: ThreadLaunchPreparationInput) => Effect.Effect<void>;
192	}
193	>()("t3/orchestration-v2/ThreadLaunchService") {}
194
195	const isThreadLaunchError = Schema.is(ThreadLaunchError);
196
197	function failureDetail(error: unknown): string {
198	if (isThreadLaunchError(error)) {
199	const cause = error.cause;
200	const detail = cause instanceof Error ? cause.message : String(cause);
201	return `Workspace preparation failed during ${error.operation.replaceAll("-", " ")}: ${detail}`;
202	}
203	return `Workspace preparation failed: ${error instanceof Error ? error.message : String(error)}`;
204	}
205
206	const make = Effect.gen(function* () {
207	const projects = yield* ProjectService.ProjectService;
208	const setupTracker = yield* WorktreeSetupTracker.WorktreeSetupTracker;
209	const cloneTracker = yield* ProjectCloneTracker.ProjectCloneTracker;
210	const terminals = yield* TerminalManager.TerminalManager;
211	const git = yield* GitWorkflow.GitWorkflowService;
212	const setupScripts = yield* ProjectSetupScriptRunner.ProjectSetupScriptRunner;
213	const providerRegistry = yield* ProviderRegistry.ProviderRegistry;
214	const serverSettings = yield* ServerSettings.ServerSettingsService;
215	const textGeneration = yield* TextGeneration.TextGeneration;
216	const receipts = yield* CommandReceiptStore.CommandReceiptStoreV2;
217	const ids = yield* IdAllocator.IdAllocatorV2;
218	const threads = yield* ThreadManagement.ThreadManagementService;
219	const managedFolders = yield* ManagedProjectFolders.ManagedProjectFolders;
220	const preparationScope = yield* Scope.make("sequential");
221	const scheduledLaunches = yield* Ref.make<ReadonlySet<CommandId>>(new Set());
222	yield* Effect.addFinalizer(() => Scope.close(preparationScope, Exit.void));
223
224	const mapError =
225	(input: PreparationInput, operation: ThreadLaunchError["operation"], threadId?: ThreadId) =>
226	(cause: unknown) =>
227	new ThreadLaunchError({
228	operation,
229	commandId: input.commandId,
230	projectId: input.projectId,
231	...(threadId === undefined ? {} : { threadId }),
232	cause,
233	});
234
235	const readReceipt = (input: ThreadLaunchInput, commandId: CommandId) =>
236	receipts
237	.getByCommandId(commandId)
238	.pipe(Effect.mapError(mapError(input, "read-receipt", input.threadId)));
239
240	const validateReusableThread = Effect.fn("ThreadLaunchService.validateReusableThread")(function* (
241	input: ThreadLaunchInput,
242	threadId: ThreadId,
243	) {
244	const projection = yield* threads
245	.getThreadRecords(threadId, ["runs"])
246	.pipe(Effect.mapError(mapError(input, "update-thread", threadId)));
247	if (
248	projection.thread.projectId !== input.projectId ||
249	projection.thread.archivedAt !== null ||
250	projection.thread.deletedAt !== null ||
251	(yield* threads
252	.getMessageCount(threadId)
253	.pipe(Effect.mapError(mapError(input, "update-thread", threadId)))) > 0 ||
254	projection.runs.length > 0
255	) {
256	return yield* mapError(
257	input,
258	"update-thread",
259	threadId,
260	)("Only an empty active thread in the target project can change workspace during launch.");
261	}
262	});
263
264	const checkWorktreeBase: ThreadLaunchService["Service"]["checkWorktreeBase"] = Effect.fn(
265	"ThreadLaunchService.checkWorktreeBase",
266	)(function* (input) {
267	const strategy = input.workspaceStrategy;
268	// `:/text` is a commit search, which cannot be asked for a commit type.
269	if (strategy.type !== "worktree" || strategy.baseRef.startsWith(":")) return;
270	const { baseRef } = strategy;
271	// Anything this cannot establish is left for the launch to report: a
272	// missing project, a folder that is not a repository, a failing Git.
273	const unresolved = yield* Effect.gen(function* () {
274	const project = Option.getOrUndefined(yield* projects.getById(input.projectId));
275	if (project === undefined) return undefined;
276	const cwd = project.workspaceRoot;
277	// `git worktree add` reads `-` as the previous checkout.
278	if (yield* git.hasCommit({ cwd, refName: baseRef === "-" ? "@{-1}" : baseRef }))
279	return undefined;
280	// Provisioning fetches the base from origin when it can, and runs a
281	// repository with no commits yet without a worktree.
282	if (
283	strategy.startFromOrigin === true &&
284	(yield* git.remoteExists({ cwd, remoteName: "origin" }))
285	)
286	return undefined;
287	// `git worktree add` also starts from a remote-tracking branch of that
288	// name; any ref carrying it is reason enough to let Git decide.
289	if (yield* git.hasRefNamed({ cwd, refName: baseRef })) return undefined;
290	return (yield* git.hasCommit({ cwd, refName: "HEAD" })) ? project : undefined;
291	}).pipe(Effect.orElseSucceed(() => undefined));
292	if (unresolved !== undefined)
293	return yield* new ThreadLaunchBaseRefError({
294	projectId: unresolved.id,
295	projectTitle: unresolved.title,
296	workspaceRoot: unresolved.workspaceRoot,
297	baseRef,
298	});
299	});
300
301	const prepareInBackground = Effect.fn("ThreadLaunchService.prepareInBackground")(function* (
302	input: PreparationInput,
303	threadId: ThreadId,
304	runId: RunId | null,
305	) {
306	const project = yield* projects.getById(input.projectId).pipe(
307	Effect.mapError(mapError(input, "resolve-project", threadId)),
308	Effect.flatMap(
309	Option.match({
310	onNone: () =>
311	Effect.fail(mapError(input, "resolve-project", threadId)("Project no longer exists.")),
312	onSome: Effect.succeed,
313	}),
314	),
315	);
316
317	const reused = input.reusedWorktree;
318	const tracked = input.workspaceStrategy.type === "worktree" || reused !== undefined;
319	let createdWorktreePath: string | null = null;
320	let setupTerminalId: string | null = null;
321	let workspaceRecorded = false;
322	let runReleased = false;
323	if (input.workspaceStrategy.type === "worktree") {
324	yield* setupTracker.begin({
325	threadId,
326	branch: input.workspaceStrategy.branch ?? null,
327	baseRef: input.workspaceStrategy.baseRef,
328	stages: ["fetch", "checkout", "setup-script", "agent"],
329	fiber: yield* Effect.fiber,
330	});
331	} else if (reused !== undefined) {
332	yield* setupTracker.begin({
333	threadId,
334	branch: input.workspaceStrategy.branch ?? null,
335	baseRef: reused.baseRef,
336	stages: ["setup-script", "agent"],
337	fiber: yield* Effect.fiber,
338	});
339	}
340	yield* Effect.gen(function* () {
341	const initialMessage = input.initialMessage;
342	const generateBranchNameFor = (cwd: string, message: ThreadLaunchInitialMessage) =>
343	Effect.gen(function* () {
344	const settings = resolveProjectSettings(
345	yield* serverSettings.getSettings,
346	input.projectId,
347	).settings;
348	const modelSelection =
349	settings.sourceControlWriterModelSelection === null
350	? settings.textGenerationModelSelection
351	: ServerSettings.resolveSourceControlWriterModelSelection(
352	settings,
353	yield* providerRegistry.getProviders,
354	input.projectId,
355	);
356	return yield* textGeneration
357	.generateBranchName({
358	naming: {
359	mode: settings.branchNamingMode,
360	prefix: settings.branchNamePrefix,
361	instructions: settings.branchNameInstructions,
362	},
363	cwd,
364	message: message.text,
365	attachments: message.attachments,
366	...(message.context ? { context: message.context } : {}),
367	modelSelection,
368	})
369	.pipe(
370	Effect.map((result) => ({
371	branch: result.branch,
372	exactName: settings.branchNamingMode === "custom",
373	})),
374	);
375	});
376
377	// The server owns worktree naming: without an explicit branch, provision
378	// under a temporary `t3/<hash>` name so the worktree never waits on
379	// name generation, then rename in the background below.
380	const requestedBranch = input.workspaceStrategy.branch;
381	let branch: string | null;
382	if (input.workspaceStrategy.type === "worktree" && requestedBranch === undefined) {
383	const uuid = yield* randomUuidV4;
384	branch = buildTemporaryWorktreeBranchName(() => uuid.replaceAll("-", ""));
385	} else {
386	branch = requestedBranch ?? null;
387	}
388	let worktreePath =
389	input.workspaceStrategy.type === "existing_worktree"
390	? input.workspaceStrategy.worktreePath
391	: null;
392	if (input.workspaceStrategy.type === "worktree") {
393	if (runId !== null) {
394	yield* threads
395	.dispatch({
396	type: "prepared-run.progress",
397	commandId: CommandId.make(`${input.commandId}:progress:worktree`),
398	threadId,
399	runId,
400	phase: "worktree",
401	})
402	.pipe(Effect.mapError(mapError(input, "update-thread", threadId)));
403	}
404	let startRef = input.workspaceStrategy.baseRef;
405	// "Start from origin" is a stored default; repos without the requested
406	// remote branch fall back to the local base branch.
407	const startFromOrigin =
408	input.workspaceStrategy.startFromOrigin === true &&
409	(yield* git
410	.remoteExists({ cwd: project.workspaceRoot, remoteName: "origin" })
411	.pipe(Effect.mapError(mapError(input, "provision-worktree", threadId))));
412	yield* setupTracker.stageStatus(threadId, "fetch", startFromOrigin ? "running" : "skipped");
413	if (startFromOrigin) {
414	yield* git
415	.fetchRemote({
416	cwd: project.workspaceRoot,
417	remoteName: "origin",
418	refName: input.workspaceStrategy.baseRef,
419	})
420	.pipe(Effect.mapError(mapError(input, "provision-worktree", threadId)));
421	const remoteBaseExists = yield* git
422	.remoteBranchExists({
423	cwd: project.workspaceRoot,
424	refName: input.workspaceStrategy.baseRef,
425	remoteName: "origin",
426	})
427	.pipe(Effect.mapError(mapError(input, "provision-worktree", threadId)));
428	if (remoteBaseExists) {
429	startRef = yield* git
430	.resolveRemoteTrackingCommit({
431	cwd: project.workspaceRoot,
432	refName: input.workspaceStrategy.baseRef,
433	fallbackRemoteName: "origin",
434	})
435	.pipe(
436	Effect.map((resolved) => resolved.commitSha),
437	Effect.mapError(mapError(input, "provision-worktree", threadId)),
438	);
439	}
440	}
441	if (startFromOrigin) yield* setupTracker.stageStatus(threadId, "fetch", "done");
442	if (
443	branch !== null &&
444	isTemporaryWorktreeBranch(branch) &&
445	(yield* git
446	.hasCommit({
447	cwd: project.workspaceRoot,
448	refName: `refs/heads/${WORKTREE_BRANCH_PREFIX}`,
449	})
450	.pipe(Effect.mapError(mapError(input, "provision-worktree", threadId))))
451	) {
452	branch = flattenTemporaryWorktreeBranchName(branch);
453	}
454	yield* setupTracker.stageStatus(threadId, "checkout", "running");
455	const worktree = yield* git
456	.createWorktree(
457	{
458	cwd: project.workspaceRoot,
459	refName: startRef,
460	newRefName: branch!,
461	baseRefName: input.workspaceStrategy.baseRef,
462	path: null,
463	},
464	{
465	progress: {
466	onWorktreeClaimed: (path) =>
467	Effect.sync(() => {
468	createdWorktreePath = path;
469	}),
470	onCheckoutProgress: (progress) =>
471	setupTracker.stage(threadId, "checkout", { percent: progress.percent }),
472	},
473	},
474	)
475	.pipe(Effect.mapError(mapError(input, "provision-worktree", threadId)));
476	worktreePath = worktree.worktree.path;
477	branch = worktree.worktree.refName;
478	createdWorktreePath = worktreePath;
479	yield* setupTracker.update(threadId, (snapshot) => ({ ...snapshot, worktreePath, branch }));
480	yield* setupTracker.stageStatus(threadId, "checkout", "done");
481	}
482
483	// A reused worktree is already recorded, and rewriting it could undo
484	// the first attempt's branch rename.
485	if (reused === undefined) {
486	yield* threads
487	.dispatch({
488	type: "thread.metadata.update",
489	commandId: CommandId.make(`${input.commandId}:workspace`),
490	threadId,
491	branch,
492	worktreePath,
493	})
494	.pipe(Effect.mapError(mapError(input, "update-thread", threadId)));
495	}
496	workspaceRecorded = true;
497
498	// Rename temporary branches (server-invented above, or sent by clients
499	// that name worktrees themselves) in the background so generation latency
500	// never delays provisioning or the provider turn. The temporary name
501	// simply sticks if generation or the rename fails.
502	if (
503	reused === undefined &&
504	worktreePath !== null &&
505	branch !== null &&
506	initialMessage !== undefined &&
507	isTemporaryWorktreeBranch(branch)
508	) {
509	const oldBranch = branch;
510	const worktreeCwd = worktreePath;
511	yield* generateBranchNameFor(worktreeCwd, initialMessage).pipe(
512	Effect.flatMap(({ branch: newBranch, exactName }) =>
513	git.renameBranch({
514	cwd: worktreeCwd,
515	oldBranch,
516	newBranch,
517	...(exactName ? { exactName: true } : {}),
518	}),
519	),
520	Effect.flatMap((renamed) =>
521	threads.dispatch({
522	type: "thread.metadata.update",
523	commandId: CommandId.make(`${input.commandId}:branch-rename`),
524	threadId,
525	branch: renamed.branch,
526	worktreePath: worktreeCwd,
527	}),
528	),
529	Effect.catchCause((cause) =>
530	Effect.logWarning("Thread worktree branch rename failed", {
531	commandId: input.commandId,
532	threadId,
533	oldBranch,
534	cause,
535	}),
536	),
537	Effect.forkIn(preparationScope),
538	);
539	}
540
541	const cwd = worktreePath ?? project.workspaceRoot;
542	if (runId !== null) {
543	yield* threads
544	.dispatch({
545	type: "prepared-run.progress",
546	commandId: CommandId.make(`${input.commandId}:progress:setup`),
547	threadId,
548	runId,
549	phase: "setup",
550	})
551	.pipe(Effect.mapError(mapError(input, "update-thread", threadId)));
552	}
553	yield* setupTracker.stageStatus(threadId, "setup-script", "running");
554	const setup = yield* setupScripts
555	.runForThread({
556	threadId,
557	projectId: input.projectId,
558	projectCwd: project.workspaceRoot,
559	worktreePath: cwd,
560	...(tracked
561	? {
562	observeCompletion: {
563	onOutputLine: (line: string) =>
564	setupTracker.appendTail(threadId, "setup-script", line),
565	},
566	}
567	: {}),
568	project: {
569	id: project.id,
570	workspaceRoot: project.workspaceRoot,
571	scripts: project.scripts,
572	},
573	})
574	.pipe(Effect.mapError(mapError(input, "run-setup-script", threadId)));
575
576	let awaitAsyncSetup = Effect.void;
577	if (setup.status === "started") {
578	setupTerminalId = setup.terminalId;
579	yield* setupTracker.update(threadId, (snapshot) => ({
580	...snapshot,
581	setupScript: {
582	name: setup.scriptName,
583	command: setup.scriptCommand,
584	terminalId: setup.terminalId,
585	},
586	}));
587	if (setup.completion) {
588	const awaitCompletion = Effect.gen(function* () {
589	const completion = yield* setup.completion!;
590	yield* setupTracker.stage(threadId, "setup-script", {
591	status: completion.exitCode === 0 ? "done" : "failed",
592	detail: `exited with ${completion.exitCode ?? "no exit code"}`,
593	});
594	if (completion.exitCode !== 0 && !setup.async)
595	return yield* mapError(
596	input,
597	"run-setup-script",
598	threadId,
599	)(`Setup script exited with ${completion.exitCode ?? "no exit code"}.`);
600	});
601	if (setup.async) {
602	awaitAsyncSetup = awaitCompletion.pipe(
603	Effect.catchCause((cause) =>
604	setupTracker.stage(threadId, "setup-script", {
605	status: "failed",
606	detail: failureDetail(Cause.squash(cause)),
607	}),
608	),
609	);
610	} else {
611	yield* awaitCompletion;
612	}
613	} else {
614	yield* setupTracker.stageStatus(threadId, "setup-script", "done");
615	}
616	} else {
617	yield* setupTracker.stageStatus(threadId, "setup-script", "skipped");
618	}
619	yield* setupTracker.markUncancellable(threadId);
620	yield* setupTracker.stageStatus(threadId, "agent", "running");
621	if (runId !== null) {
622	yield* threads
623	.dispatch({
624	type: "prepared-run.release",
625	commandId: CommandId.make(`${input.commandId}:release`),
626	threadId,
627	runId,
628	})
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	);
638	}
639	yield* setupTracker.stageStatus(threadId, "agent", "done");
640	yield* awaitAsyncSetup;
641	yield* setupTracker.finish(threadId, "done");
642	}).pipe(
643	Effect.onError((cause) =>
644	Effect.gen(function* () {
645	const cancelled = Cause.hasInterruptsOnly(cause);
646	yield* setupTracker.finish(
647	threadId,
648	cancelled ? "cancelled" : "failed",
649	cancelled ? null : failureDetail(Cause.squash(cause)),
650	);
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	) {
660	const removedPath = createdWorktreePath;
661	// The thread forgets the worktree only once it is gone; a failed
662	// removal leaves the directory for the user to clean up rather than
663	// reusing a checkout that may be half written.
664	yield* git
665	.removeWorktree({ cwd: project.workspaceRoot, path: removedPath, force: true })
666	.pipe(
667	Effect.andThen(
668	threads
669	.dispatch({
670	type: "thread.metadata.update",
671	commandId: CommandId.make(`${input.commandId}:cancel-workspace`),
672	threadId,
673	worktreePath: null,
674	branch: null,
675	})
676	.pipe(Effect.ignore),
677	),
678	Effect.catchCause((removeCause) =>
679	Effect.logWarning("Failed to remove an abandoned thread worktree", {
680	commandId: input.commandId,
681	threadId,
682	path: removedPath,
683	cause: removeCause,
684	}),
685	),
686	);
687	}
688	}),
689	),
690	);
691	});
692
693	const failPreparedRun = (
694	input: Pick<PreparationInput, "commandId">,
695	threadId: ThreadId,
696	runId: RunId | null,
697	cause: unknown,
698	) =>
699	runId === null
700	? Effect.logWarning("Thread workspace preparation failed", {
701	commandId: input.commandId,
702	threadId,
703	cause,
704	})
705	: threads
706	.dispatch({
707	type: "prepared-run.fail",
708	commandId: CommandId.make(`${input.commandId}:fail`),
709	threadId,
710	runId,
711	failure: makeProviderFailure({
712	cause,
713	message: failureDetail(cause),
714	class: "validation_error",
715	retryable: false,
716	}),
717	})
718	.pipe(
719	Effect.catchCause((persistCause) =>
720	Effect.logWarning("Failed to persist thread workspace preparation failure", {
721	commandId: input.commandId,
722	threadId,
723	cause,
724	persistCause,
725	}),
726	),
727	);
728
729	const reservePreparation = (commandId: CommandId) =>
730	Ref.modify(scheduledLaunches, (scheduled) => {
731	if (scheduled.has(commandId)) return [false, scheduled] as const;
732	const next = new Set(scheduled);
733	next.add(commandId);
734	return [true, next] as const;
735	});
736
737	const releasePreparation = (commandId: CommandId) =>
738	Ref.update(scheduledLaunches, (scheduled) => {
739	const next = new Set(scheduled);
740	next.delete(commandId);
741	return next;
742	});
743
744	const schedulePreparation = Effect.fn("ThreadLaunchService.schedulePreparation")(function* (
745	input: PreparationInput,
746	threadId: ThreadId,
747	runId: RunId | null,
748	) {
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(
776	Effect.onError((cause) =>
777	failPreparedRun(
778	input,
779	threadId,
780	runId,
781	Cause.hasInterruptsOnly(cause) ? "Worktree setup cancelled." : Cause.squash(cause),
782	),
783	),
784	Effect.ignoreCause,
785	Effect.ensuring(releasePreparation(input.commandId)),
786	Effect.forkIn(preparationScope),
787	);
788	});
789
790	const launch: ThreadLaunchService["Service"]["launch"] = Effect.fn("ThreadLaunchService.launch")(
791	function* (input) {
792	yield* ProjectCloneTracker.rejectCommandsDuringClone(cloneTracker, {
793	type: "thread.create",
794	projectId: input.projectId,
795	}).pipe(Effect.mapError(mapError(input, "resolve-project")));
796	const project = yield* projects.getById(input.projectId).pipe(
797	Effect.mapError(mapError(input, "resolve-project")),
798	Effect.flatMap(
799	Option.match({
800	onNone: () => Effect.fail(mapError(input, "resolve-project")("Project not found.")),
801	onSome: Effect.succeed,
802	}),
803	),
804	);
805	if (input.reuseExistingThread === true && input.threadId === undefined) {
806	return yield* mapError(
807	input,
808	"update-thread",
809	)("Reusing an existing thread requires a thread id.");
810	}
811
812	const launchReceipt = yield* readReceipt(input, input.commandId);
813	yield* threads.assertProviderInstanceEnabled({
814	projectId: input.projectId,
815	commandId: input.commandId,
816	instanceId: input.modelSelection.instanceId,
817	});
818	return yield* Effect.gen(function* () {
819	// A retried launch has no client-supplied id to replay against, so
820	// recover the thread id its accepted create was recorded under before
821	// allocating another one; a fresh id would only collide with the
822	// recorded receipt.
823	const reusableLaunchReceipt =
824	input.threadId === undefined &&
825	Option.isSome(launchReceipt) &&
826	launchReceipt.value.status === "accepted" &&
827	launchReceipt.value.commandType === "thread.create"
828	? launchReceipt.value
829	: undefined;
830	const candidateThreadId =
831	input.threadId ??
832	reusableLaunchReceipt?.threadId ??
833	(yield* ids.allocate
834	.thread({ projectId: input.projectId })
835	.pipe(Effect.mapError(mapError(input, "create-thread"))));
836
837	if (reusableLaunchReceipt !== undefined) {
838	const shell = yield* threads
839	.getThreadShell(candidateThreadId)
840	.pipe(Effect.mapError(mapError(input, "create-thread", candidateThreadId)));
841	if (shell === null) {
842	return yield* mapError(input, "create-thread", candidateThreadId)("Thread not found.");
843	}
844	if (shell.projectId !== input.projectId) {
845	return yield* mapError(
846	input,
847	"resolve-project",
848	candidateThreadId,
849	)("Project identity changed.");
850	}
851	}
852
853	if (input.reuseExistingThread === true && Option.isNone(launchReceipt)) {
854	yield* validateReusableThread(input, candidateThreadId);
855	}
856
857	// A Scratch thread launched at the project root runs in a folder of its
858	// own. Only the first attempt claims one; a retry replays its create.
859	const workspaceStrategy: ThreadLaunchWorkspaceStrategy =
860	input.workspaceStrategy.type === "root" && Option.isNone(launchReceipt)
861	? Option.match(
862	yield* managedFolders
863	.folderForThread({
864	projectId: input.projectId,
865	threadId: candidateThreadId,
866	text: input.initialMessage?.text ?? input.title,
867	})
868	.pipe(Effect.mapError(mapError(input, "provision-worktree", candidateThreadId))),
869	{
870	onNone: () => input.workspaceStrategy,
871	onSome: (worktreePath) => ({ type: "existing_worktree", worktreePath }),
872	},
873	)
874	: input.workspaceStrategy;
875	const initialBranch = workspaceStrategy.branch ?? null;
876	const initialWorktreePath =
877	workspaceStrategy.type === "existing_worktree" ? workspaceStrategy.worktreePath : null;
878	const claimDispatch =
879	input.reuseExistingThread === true
880	? threads.dispatch({
881	type: "thread.metadata.update",
882	commandId: input.commandId,
883	threadId: candidateThreadId,
884	expectedEmpty: true,
885	})
886	: threads.dispatch({
887	type: "thread.create",
888	commandId: input.commandId,
889	threadId: candidateThreadId,
890	projectId: input.projectId,
891	title: input.title,
892	modelSelection: input.modelSelection,
893	runtimeMode: input.runtimeMode,
894	interactionMode: input.interactionMode,
895	branch: initialBranch,
896	worktreePath: initialWorktreePath,
897	...(input.importedNativeThread === undefined
898	? {}
899	: { importedNativeThread: input.importedNativeThread }),
900	createdBy: input.createdBy,
901	creationSource: input.creationSource,
902	});
903	const claimed = yield* claimDispatch.pipe(
904	Effect.mapError(
905	mapError(
906	input,
907	input.reuseExistingThread === true ? "update-thread" : "create-thread",
908	candidateThreadId,
909	),
910	),
911	);
912	const threadId =
913	claimed.storedEvents.find((stored) => stored.event.type.startsWith("thread."))?.event
914	.threadId ?? candidateThreadId;
915	if (project.id !== input.projectId) {
916	return yield* mapError(input, "resolve-project", threadId)("Project identity changed.");
917	}
918
919	let runId: RunId | null = null;
920	let messageWasAlreadyAccepted = false;
921	if (input.initialMessage !== undefined) {
922	const messageCommandId = CommandId.make(`${input.commandId}:initial-message`);
923	const messageReceipt = yield* readReceipt(input, messageCommandId);
924	messageWasAlreadyAccepted = Option.isSome(messageReceipt);
925	const messageId =
926	input.initialMessage.messageId ??
927	(yield* ids.allocate
928	.message({ threadId, ordinal: 1 })
929	.pipe(Effect.mapError(mapError(input, "dispatch-message", threadId))));
930	const dispatched = yield* threads
931	.dispatch({
932	type: "message.dispatch",
933	commandId: messageCommandId,
934	threadId,
935	messageId,
936	text: input.initialMessage.text,
937	...(input.initialMessage.scheduledTaskId === undefined
938	? {}
939	: { scheduledTaskId: input.initialMessage.scheduledTaskId }),
940	...(input.initialMessage.senderThreadId === undefined
941	? {}
942	: { senderThreadId: input.initialMessage.senderThreadId }),
943	attachments: input.initialMessage.attachments,
944	...(input.initialMessage.context ? { context: input.initialMessage.context } : {}),
945	...(input.generateTitle === true ? { titleSeed: input.title } : {}),
946	modelSelection: input.modelSelection,
947	dispatchMode: { type: "defer_start", workspaceStrategy },
948	createdBy: input.createdBy,
949	creationSource: input.creationSource,
950	})
951	.pipe(Effect.mapError(mapError(input, "dispatch-message", threadId)));
952	const runCreated = dispatched.storedEvents.find(
953	(stored) => stored.event.type === "run.created",
954	);
955	runId = runCreated?.event.type === "run.created" ? runCreated.event.payload.id : null;
956	if (runId === null) {
957	return yield* mapError(
958	input,
959	"dispatch-message",
960	threadId,
961	)("Initial message was accepted without a durable run.");
962	}
963	}
964
965	const projection = yield* threads
966	.getThreadProjection(threadId)
967	.pipe(Effect.mapError(mapError(input, "create-thread", threadId)));
968	const runIsPreparing =
969	runId !== null &&
970	projection.runs.some((run) => run.id === runId && run.status === "preparing");
971	const shouldSchedule = runId === null ? Option.isNone(launchReceipt) : runIsPreparing;
972	// A retried root launch prepares the folder its first attempt bound, so
973	// a Scratch thread keeps its own. Other root launches bind no folder.
974	const boundWorktreePath = projection.thread.worktreePath;
975	const preparationStrategy: ThreadLaunchWorkspaceStrategy =
976	Option.isSome(launchReceipt) &&
977	workspaceStrategy.type === "root" &&
978	boundWorktreePath !== null
979	? {
980	type: "existing_worktree",
981	worktreePath: boundWorktreePath,
982	branch: workspaceStrategy.branch,
983	}
984	: workspaceStrategy;
985	if (shouldSchedule) {
986	const ownsPreparation = yield* reservePreparation(input.commandId);
987	if (ownsPreparation) {
988	yield* Effect.gen(function* () {
989	const preparationStillRequired =
990	runId === null
991	? true
992	: yield* threads.getThreadRecords(threadId, ["runs"], { runIds: [runId] }).pipe(
993	Effect.map((current) =>
994	current.runs.some((run) => run.id === runId && run.status === "preparing"),
995	),
996	Effect.mapError(mapError(input, "update-thread", threadId)),
997	);
998	if (preparationStillRequired) {
999	yield* schedulePreparation(
1000	{ ...input, workspaceStrategy: preparationStrategy },
1001	threadId,
1002	runId,
1003	);
1004	} else {
1005	yield* releasePreparation(input.commandId);
1006	}
1007	}).pipe(Effect.onError(() => releasePreparation(input.commandId)));
1008	}
1009	}
1010
1011	return {
1012	threadId,
1013	projection,
1014	resumed: Option.isSome(launchReceipt) || messageWasAlreadyAccepted,
1015	};
1016	});
1017	},
1018	);
1019
1020	const prepareRun: ThreadLaunchService["Service"]["prepareRun"] = Effect.fn(
1021	"ThreadLaunchService.prepareRun",
1022	)(function* (input) {
1023	const scheduled = yield* Effect.gen(function* () {
1024	const projection = yield* threads.getThreadProjection(input.threadId);
1025	const run = projection.runs.find((candidate) => candidate.id === input.runId);
1026	const workspacePreparation = run?.workspacePreparation;
1027	if (run?.status !== "preparing" || workspacePreparation === undefined) return;
1028	if (!(yield* reservePreparation(input.commandId))) return;
1029	yield* scheduleRunPreparation(input, projection, run, workspacePreparation).pipe(
1030	Effect.onError(() => releasePreparation(input.commandId)),
1031	);
1032	}).pipe(Effect.exit);
1033	if (Exit.isFailure(scheduled)) {
1034	yield* failPreparedRun(input, input.threadId, input.runId, Cause.squash(scheduled.cause));
1035	}
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);
1048	return dispatched;
1049	});
1050
1051	const scheduleRunPreparation = (
1052	input: ThreadLaunchPreparationInput,
1053	projection: OrchestrationV2ThreadProjection,
1054	run: OrchestrationV2ThreadProjection["runs"][number],
1055	workspacePreparation: ThreadLaunchWorkspaceStrategy,
1056	) => {
1057	const message = projection.messages.find((candidate) => candidate.id === run.userMessageId);
1058	// A worktree the failed attempt already created is reused, not created again.
1059	const reuse =
1060	workspacePreparation.type === "worktree" &&
1061	projection.thread.worktreePath !== null &&
1062	projection.thread.branch !== null
1063	? {
1064	strategy: {
1065	type: "existing_worktree" as const,
1066	worktreePath: projection.thread.worktreePath,
1067	branch: projection.thread.branch,
1068	},
1069	reusedWorktree: { baseRef: workspacePreparation.baseRef },
1070	}
1071	: null;
1072	return schedulePreparation(
1073	{
1074	commandId: input.commandId,
1075	projectId: projection.thread.projectId,
1076	workspaceStrategy: reuse?.strategy ?? workspacePreparation,
1077	...(reuse === null ? {} : { reusedWorktree: reuse.reusedWorktree }),
1078	...(message === undefined
1079	? {}
1080	: {
1081	initialMessage: {
1082	text: message.text,
1083	attachments: message.attachments,
1084	...(message.context ? { context: message.context } : {}),
1085	},
1086	}),
1087	},
1088	input.threadId,
1089	run.id,
1090	);
1091	};
1092
1093	return ThreadLaunchService.of({ launch, checkWorktreeBase, retryPreparation, prepareRun });
1094	});
1095
1096	export const layer = Layer.effect(ThreadLaunchService, make);