diff --git a/apps/server/src/orchestration/Layers/ThreadBootstrap.ts b/apps/server/src/orchestration/Layers/ThreadBootstrap.ts index 8627b2d024d7..b73090961428 100644 --- a/apps/server/src/orchestration/Layers/ThreadBootstrap.ts +++ b/apps/server/src/orchestration/Layers/ThreadBootstrap.ts @@ -3,16 +3,23 @@ import { EventId, OrchestrationDispatchCommandError, type ThreadId, + WORKTREE_SETUP_ACTIVITY_KIND, + worktreeSetupActivityId, + type WorktreeSetupSnapshot, } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; import * as Crypto from "effect/Crypto"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; +import * as Schedule from "effect/Schedule"; import * as Schema from "effect/Schema"; import { GitWorkflowService } from "../../git/GitWorkflowService.ts"; import * as ProjectSetupScriptRunner from "../../project/ProjectSetupScriptRunner.ts"; +import * as WorktreeSetupTracker from "../../project/WorktreeSetupTracker.ts"; +import * as TerminalManager from "../../terminal/Manager.ts"; import { VcsStatusBroadcaster } from "../../vcs/VcsStatusBroadcaster.ts"; import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts"; import { ThreadDeletionReactor } from "../Services/ThreadDeletionReactor.ts"; @@ -63,6 +70,8 @@ const makeThreadBootstrap = Effect.gen(function* () { const orchestrationEngine = yield* OrchestrationEngineService; const gitWorkflow = yield* GitWorkflowService; const projectSetupScriptRunner = yield* ProjectSetupScriptRunner.ProjectSetupScriptRunner; + const worktreeSetupTracker = yield* WorktreeSetupTracker.WorktreeSetupTracker; + const terminalManager = yield* TerminalManager.TerminalManager; const vcsStatusBroadcaster = yield* VcsStatusBroadcaster; const threadDeletionReactor = yield* ThreadDeletionReactor; const crypto = yield* Crypto.Crypto; @@ -125,6 +134,47 @@ const makeThreadBootstrap = Effect.gen(function* () { ), ); + // The worktree setup's durable record: one activity per thread, upserted + // by a fixed id when the setup starts and again when it settles. Live + // progress keeps streaming from the tracker; this is what a reload or + // another client reads. Best effort: the thread may already be gone + // after a failed bootstrap. + const recordWorktreeSetup = (snapshot: WorktreeSetupSnapshot, options: DispatchOptions) => + serverCommandId("worktree-setup-activity").pipe( + Effect.flatMap((commandId) => + orchestrationEngine.dispatch( + { + type: "thread.activity.append", + commandId, + threadId: snapshot.threadId, + activity: { + id: EventId.make(worktreeSetupActivityId(snapshot.threadId)), + tone: + snapshot.phase === "failed" || + snapshot.stages.some((stage) => stage.status === "failed") + ? "error" + : "info", + kind: WORKTREE_SETUP_ACTIVITY_KIND, + summary: + snapshot.phase === "running" + ? "Setting up worktree" + : snapshot.phase === "done" + ? "Worktree ready" + : snapshot.phase === "cancelled" + ? "Worktree setup cancelled" + : "Worktree setup failed", + payload: snapshot, + turnId: null, + createdAt: snapshot.startedAt, + }, + createdAt: snapshot.endedAt ?? snapshot.startedAt, + }, + options, + ), + ), + Effect.ignoreCause({ log: true }), + ); + const toBootstrapDispatchCommandCauseError = (cause: Cause.Cause) => { const error = Cause.squash(cause); return isOrchestrationDispatchCommandError(error) @@ -143,10 +193,44 @@ const makeThreadBootstrap = Effect.gen(function* () { Effect.gen(function* () { const bootstrap = command.bootstrap; const { bootstrap: _bootstrap, ...finalTurnStartCommand } = command; + const threadId = command.threadId; + const tracked = bootstrap?.prepareWorktree !== undefined; + const track = (effect: Effect.Effect) => (tracked ? effect : Effect.void); let createdThread = false; let targetProjectId = bootstrap?.createThread?.projectId; let targetProjectCwd = bootstrap?.prepareWorktree?.projectCwd; let targetWorktreePath = bootstrap?.createThread?.worktreePath ?? null; + // The setup script's terminal, once started. Cancel closes only this + // one so terminals the user opened meanwhile survive. + let setupTerminalId: string | null = null; + + // Set once the checkout starts; see the session.set below. + let preparingSessionSet = false; + const markPreparingSessionFailed = (detail: string) => + Effect.gen(function* () { + const failedAt = yield* nowIso; + yield* orchestrationEngine.dispatch( + { + type: "thread.session.set", + commandId: yield* serverCommandId("bootstrap-thread-preparing-failed"), + threadId, + session: { + threadId, + status: "error", + providerName: null, + providerInstanceId: + bootstrap?.createThread?.modelSelection.instanceId ?? + command.modelSelection?.instanceId, + runtimeMode: command.runtimeMode, + activeTurnId: null, + lastError: detail.trim().length > 0 ? detail : "Worktree setup failed.", + updatedAt: failedAt, + }, + createdAt: failedAt, + }, + options, + ); + }); const cleanupCreatedThread = () => createdThread @@ -251,19 +335,34 @@ const makeThreadBootstrap = Effect.gen(function* () { ); }); + // Starts the setup script. For tracked bootstraps it returns the + // effect that waits for the script to exit and records the outcome + // on the card; whether the agent stage waits on it depends on the + // script's `async` flag. Returns null when nothing is left to await. + // Untracked callers keep the old fire-and-forget behavior. const runSetupProgram = () => Effect.gen(function* () { if (!bootstrap?.runSetupScript || !targetWorktreePath) { - return; + yield* track(worktreeSetupTracker.stageStatus(threadId, "setup-script", "skipped")); + return null; } const worktreePath = targetWorktreePath; const requestedAt = yield* nowIso; - yield* projectSetupScriptRunner + yield* track(worktreeSetupTracker.stageStatus(threadId, "setup-script", "running")); + const setupResult = yield* projectSetupScriptRunner .runForThread({ - threadId: command.threadId, + threadId, ...(targetProjectId ? { projectId: targetProjectId } : {}), ...(targetProjectCwd ? { projectCwd: targetProjectCwd } : {}), worktreePath, + ...(tracked + ? { + observeCompletion: { + onOutputLine: (line) => + worktreeSetupTracker.appendTail(threadId, "setup-script", line), + }, + } + : {}), }) .pipe( Effect.matchEffect({ @@ -272,24 +371,177 @@ const makeThreadBootstrap = Effect.gen(function* () { error, requestedAt, worktreePath, - }), + }).pipe( + Effect.andThen( + track( + worktreeSetupTracker.stageStatus( + threadId, + "setup-script", + "failed", + "failed to start", + ), + ), + ), + Effect.as(null), + ), onSuccess: (setupResult) => { if (setupResult.status !== "started") { - return Effect.void; + return track( + worktreeSetupTracker.stageStatus( + threadId, + "setup-script", + "skipped", + "no setup script", + ), + ).pipe(Effect.as(null)); } + setupTerminalId = setupResult.terminalId; return recordSetupScriptStarted({ requestedAt, worktreePath, scriptId: setupResult.scriptId, scriptName: setupResult.scriptName, terminalId: setupResult.terminalId, - }); + }).pipe( + Effect.andThen( + track( + worktreeSetupTracker.update(threadId, (snapshot) => ({ + ...snapshot, + setupScript: { + name: setupResult.scriptName, + command: setupResult.scriptCommand, + terminalId: setupResult.terminalId, + }, + })), + ), + ), + Effect.as(setupResult), + ); }, }), ); + if (!tracked || !setupResult?.completion) { + return null; + } + // The setup script is best effort, like the untracked path: a + // failed install must not throw away the worktree the user just + // waited for. The card keeps the failed stage and its terminal. + // Forked right away so the terminal listener behind `completion` + // is always consumed, even when the turn dispatch fails before + // anyone would otherwise wait on it. The tracker update is a + // no-op once the snapshot has been dropped. + const completionFiber = yield* setupResult.completion.pipe( + Effect.flatMap((completion) => { + if (completion.exitCode === 0) { + return worktreeSetupTracker.stageStatus(threadId, "setup-script", "done"); + } + const detail = + completion.exitCode === null + ? "terminal closed before the script finished" + : `exit ${completion.exitCode}`; + return worktreeSetupTracker.stageStatus(threadId, "setup-script", "failed", detail); + }), + Effect.forkDetach, + ); + if (!setupResult.async) { + yield* Fiber.join(completionFiber); + return null; + } + return completionFiber; }); const bootstrapProgram = Effect.gen(function* () { + const prepareWorktree = bootstrap?.prepareWorktree; + let shouldPrepareWorktree = prepareWorktree + ? yield* gitWorkflow.isRepository(prepareWorktree.projectCwd) + : false; + let worktreeBaseRef = prepareWorktree?.baseBranch ?? null; + + if (prepareWorktree && shouldPrepareWorktree) { + // "Start from origin" is a stored default; repos without the + // requested remote branch fall back to the local base branch. + const startFromOrigin = + prepareWorktree.startFromOrigin === true && + (yield* gitWorkflow.remoteExists({ + cwd: prepareWorktree.projectCwd, + remoteName: "origin", + })); + if (startFromOrigin) { + yield* track(worktreeSetupTracker.stageStatus(threadId, "fetch", "running")); + yield* gitWorkflow.fetchRemote({ + cwd: prepareWorktree.projectCwd, + remoteName: "origin", + refName: prepareWorktree.baseBranch, + }); + const remoteBaseExists = yield* gitWorkflow.remoteBranchExists({ + cwd: prepareWorktree.projectCwd, + refName: prepareWorktree.baseBranch, + remoteName: "origin", + }); + if (remoteBaseExists) { + const resolvedRemoteBase = yield* gitWorkflow.resolveRemoteTrackingCommit({ + cwd: prepareWorktree.projectCwd, + refName: prepareWorktree.baseBranch, + fallbackRemoteName: "origin", + }); + worktreeBaseRef = resolvedRemoteBase.commitSha; + yield* track( + worktreeSetupTracker.stageStatus( + threadId, + "fetch", + "done", + `origin/${prepareWorktree.baseBranch} at ${resolvedRemoteBase.commitSha.slice(0, 7)}`, + ), + ); + } else { + yield* track( + worktreeSetupTracker.stageStatus( + threadId, + "fetch", + "warning", + `origin/${prepareWorktree.baseBranch} not found, using local branch`, + ), + ); + } + } else { + yield* track(worktreeSetupTracker.stageStatus(threadId, "fetch", "skipped")); + } + + const resolvedWorktreeBaseRef = worktreeBaseRef ?? prepareWorktree.baseBranch; + shouldPrepareWorktree = yield* gitWorkflow.hasCommit({ + cwd: prepareWorktree.projectCwd, + refName: resolvedWorktreeBaseRef, + }); + worktreeBaseRef = resolvedWorktreeBaseRef; + yield* track( + worktreeSetupTracker.update(threadId, (snapshot) => ({ + ...snapshot, + baseRef: resolvedWorktreeBaseRef, + })), + ); + } + + if (prepareWorktree && !shouldPrepareWorktree) { + if (prepareWorktree.requireWorktree) { + return yield* new OrchestrationDispatchCommandError({ + message: + "A separate worktree requires a Git repository and a base branch with a commit.", + }); + } + // Not a git repo, or the base has no commit: the thread runs in + // the project checkout instead. The card says so and moves on. + yield* track( + worktreeSetupTracker.update(threadId, (snapshot) => ({ + ...snapshot, + stages: snapshot.stages.map((stage) => + stage.id === "fetch" || stage.id === "checkout" || stage.id === "submodules" + ? { ...stage, status: "skipped", detail: "using project checkout" } + : stage, + ), + })), + ); + } + if (bootstrap?.createThread) { const created = yield* orchestrationEngine.dispatch( { @@ -312,54 +564,152 @@ const makeThreadBootstrap = Effect.gen(function* () { // every delete for the prior incarnation committed before it. // Drain through that event before setup or turn start can own // terminals and provider sessions under the reused thread id. - yield* threadDeletionReactor.drainThrough(created.sequence); createdThread = true; + yield* threadDeletionReactor.drainThrough(created.sequence); + // Persist the send now rather than with the turn: the thread is + // real from here on, so any client (or a reload) sees the message + // while the worktree is still being prepared. The turn start + // later references this id instead of re-sending the text. + yield* orchestrationEngine.dispatch( + { + type: "thread.message.user.append", + commandId: yield* serverCommandId("bootstrap-thread-message"), + threadId: command.threadId, + message: { + messageId: command.message.messageId, + text: command.message.text, + attachments: command.message.attachments, + ...(command.message.context !== undefined + ? { context: command.message.context } + : {}), + }, + createdAt: command.createdAt, + }, + options, + ); + if (tracked) { + const running = yield* worktreeSetupTracker.get(threadId); + if (running) yield* recordWorktreeSetup(running, options); + } } - if (bootstrap?.prepareWorktree) { - let worktreeBaseRef = bootstrap.prepareWorktree.baseBranch; - // "Start from origin" is a stored default; repos without the - // requested remote branch, or without an origin remote at all, - // fall back to the local base branch instead of failing the whole - // bootstrap on `git fetch origin`. - const startFromOrigin = - bootstrap.prepareWorktree.startFromOrigin === true && - (yield* gitWorkflow.remoteExists({ - cwd: bootstrap.prepareWorktree.projectCwd, - remoteName: "origin", - })); - if (startFromOrigin) { - yield* gitWorkflow.fetchRemote({ - cwd: bootstrap.prepareWorktree.projectCwd, - remoteName: "origin", - }); - const remoteBaseExists = yield* gitWorkflow.remoteBranchExists({ - cwd: bootstrap.prepareWorktree.projectCwd, - refName: bootstrap.prepareWorktree.baseBranch, - remoteName: "origin", - }); - if (remoteBaseExists) { - const resolvedRemoteBase = yield* gitWorkflow.resolveRemoteTrackingCommit({ - cwd: bootstrap.prepareWorktree.projectCwd, - refName: bootstrap.prepareWorktree.baseBranch, - fallbackRemoteName: "origin", - }); - worktreeBaseRef = resolvedRemoteBase.commitSha; - } + if (prepareWorktree && shouldPrepareWorktree && worktreeBaseRef) { + if (bootstrap?.createThread && createdThread) { + // The checkout and setup script can run for minutes before the + // turn starts, and the created thread carries no message or + // turn until then. Project a starting session now so every + // client lists the thread as working and a reopened thread + // knows to follow the setup stream. A failed or cancelled setup + // deletes the thread, so nothing lingers. + const preparingAt = yield* nowIso; + yield* orchestrationEngine.dispatch( + { + type: "thread.session.set", + commandId: yield* serverCommandId("bootstrap-thread-preparing"), + threadId, + session: { + threadId, + status: "starting", + providerName: null, + providerInstanceId: bootstrap.createThread.modelSelection.instanceId, + runtimeMode: command.runtimeMode, + activeTurnId: null, + lastError: null, + updatedAt: preparingAt, + }, + createdAt: preparingAt, + }, + options, + ); + preparingSessionSet = true; } - const worktree = yield* gitWorkflow.createWorktree({ - cwd: bootstrap.prepareWorktree.projectCwd, - refName: worktreeBaseRef, - newRefName: bootstrap.prepareWorktree.branch, - baseRefName: bootstrap.prepareWorktree.baseBranch, - path: null, - }); + yield* worktreeSetupTracker.stageStatus(threadId, "checkout", "running"); + let checkoutTotal: number | null = null; + const worktree = yield* gitWorkflow.createWorktree( + { + cwd: prepareWorktree.projectCwd, + refName: worktreeBaseRef, + newRefName: prepareWorktree.branch, + baseRefName: prepareWorktree.baseBranch, + path: null, + }, + { + progress: { + // Git has registered the directory at this point, so a + // cancel during the submodule step can still remove it. + onWorktreeClaimed: (path) => + Effect.sync(() => { + targetWorktreePath = path; + }), + onCheckoutProgress: ({ percent, completed, total }) => { + checkoutTotal = total; + return worktreeSetupTracker.stage(threadId, "checkout", { + percent, + detail: `${completed.toLocaleString("en-US")} / ${total.toLocaleString("en-US")} files`, + }); + }, + onSubmodulesStarted: () => + worktreeSetupTracker + .stageStatus( + threadId, + "checkout", + "done", + checkoutTotal === null + ? null + : `${checkoutTotal.toLocaleString("en-US")} files`, + ) + .pipe( + Effect.andThen( + worktreeSetupTracker.stageStatus(threadId, "submodules", "running"), + ), + ), + onSubmoduleLine: (line) => { + const submodulePath = /Submodule path '([^']+)'/.exec(line)?.[1]; + return submodulePath === undefined + ? Effect.void + : worktreeSetupTracker.stage(threadId, "submodules", { + detail: submodulePath, + }); + }, + onSubmodulesFinished: ({ ok, detail }) => + worktreeSetupTracker.stageStatus( + threadId, + "submodules", + ok ? "done" : "warning", + ok ? undefined : (detail ?? "submodule checkout failed"), + ), + }, + }, + ); + const checkoutEndedAt = yield* nowIso; + yield* worktreeSetupTracker.update(threadId, (snapshot) => ({ + ...snapshot, + worktreePath: worktree.worktree.path, + stages: snapshot.stages.map((stage) => { + if (stage.id === "checkout" && stage.status === "running") { + return { + ...stage, + status: "done", + percent: 100, + endedAt: checkoutEndedAt, + detail: + checkoutTotal === null + ? stage.detail + : `${checkoutTotal.toLocaleString("en-US")} files`, + }; + } + if (stage.id === "submodules" && stage.status === "pending") { + return { ...stage, status: "skipped", detail: "none" }; + } + return stage; + }), + })); targetWorktreePath = worktree.worktree.path; yield* orchestrationEngine.dispatch( { type: "thread.meta.update", commandId: yield* serverCommandId("bootstrap-thread-meta-update"), - threadId: command.threadId, + threadId, branch: worktree.worktree.refName, worktreePath: targetWorktreePath, }, @@ -368,39 +718,180 @@ const makeThreadBootstrap = Effect.gen(function* () { yield* refreshGitStatus(targetWorktreePath); } - yield* runSetupProgram(); + const pendingSetupScript = yield* runSetupProgram(); - return yield* orchestrationEngine.dispatch(finalTurnStartCommand, options); + yield* track(worktreeSetupTracker.stageStatus(threadId, "agent", "running")); + // Past this point a cancel would roll back a thread whose turn has + // started. Drop the cancel handle and make the handoff atomic. + yield* track(worktreeSetupTracker.markUncancellable(threadId)); + const started = yield* Effect.uninterruptible( + orchestrationEngine.dispatch(finalTurnStartCommand, options), + ); + yield* track(worktreeSetupTracker.stageStatus(threadId, "agent", "done")); + // An async setup script outlives the handoff: the snapshot stays + // running so the client keeps its row next to the agent's work, + // and settles when the script exits. The turn already started, so + // the wait cannot fail the dispatch. + const settle = tracked + ? worktreeSetupTracker + .finish(threadId, "done") + .pipe( + Effect.flatMap((snapshot) => + snapshot ? recordWorktreeSetup(snapshot, options) : Effect.void, + ), + ) + : Effect.void; + if (pendingSetupScript) { + yield* Fiber.join(pendingSetupScript).pipe( + Effect.ignoreCause({ log: true }), + Effect.andThen(settle), + Effect.forkDetach, + ); + } else { + yield* settle; + } + return started; }); - return yield* bootstrapProgram.pipe( + const cleanupAndFail = ( + cause: Cause.Cause, + dispatchError: OrchestrationDispatchCommandError, + ) => + Effect.uninterruptible(cleanupCreatedThread()).pipe( + Effect.matchCauseEffect({ + onFailure: (cleanupCause) => + Effect.logWarning("bootstrap thread cleanup failed", { + threadId, + detail: Cause.pretty(cleanupCause), + }).pipe( + // The thread outlived its setup. Its preparing session + // must not read as working forever, so record the failure + // on it instead. + Effect.andThen( + preparingSessionSet + ? markPreparingSessionFailed(dispatchError.message).pipe( + Effect.ignoreCause({ log: true }), + ) + : Effect.void, + ), + Effect.flatMap(() => Effect.fail(dispatchError)), + ), + onSuccess: (threadDeleted) => + Effect.fail( + threadDeleted || (bootstrap?.createThread && !createdThread) + ? new OrchestrationDispatchCommandError({ + message: dispatchError.message, + ...(dispatchError.cause !== undefined ? { cause: dispatchError.cause } : {}), + bootstrapThreadDisposition: threadDeleted ? "deleted" : "not-created", + }) + : dispatchError, + ), + }), + ); + + const settledBootstrapProgram = bootstrapProgram.pipe( + Effect.interruptible, Effect.catchCause((cause) => { const dispatchError = toBootstrapDispatchCommandCauseError(cause); - // Interruptions roll back too; a created-but-never-started thread - // must not outlive its bootstrap. - return Effect.uninterruptible(cleanupCreatedThread()).pipe( - Effect.matchCauseEffect({ - onFailure: (cleanupCause) => - Effect.logWarning("bootstrap thread cleanup failed", { - threadId: command.threadId, - detail: Cause.pretty(cleanupCause), - }).pipe(Effect.flatMap(() => Effect.fail(dispatchError))), - onSuccess: (threadDeleted) => - Effect.fail( - threadDeleted + if (Cause.hasInterruptsOnly(cause)) { + // A user cancel interrupts the forked bootstrap fiber. The + // created thread is rolled back like any other failure so the + // draft returns to the composer. The setup terminal is closed + // first so a still-running script cannot hold files open in + // the worktree while git removes it. Closing kills the + // process asynchronously, so the removal retries briefly. + const closeSetupTerminal = setupTerminalId + ? terminalManager.close({ + threadId, + terminalId: setupTerminalId, + deleteHistory: true, + }) + : Effect.void; + const removeCreatedWorktree = + tracked && targetWorktreePath && bootstrap?.prepareWorktree + ? closeSetupTerminal.pipe( + Effect.ignoreCause({ log: true }), + Effect.andThen( + gitWorkflow + .removeWorktree({ + cwd: bootstrap.prepareWorktree.projectCwd, + path: targetWorktreePath, + force: true, + }) + .pipe(Effect.retry({ times: 4, schedule: Schedule.spaced("500 millis") })), + ), + Effect.ignoreCause({ log: true }), + Effect.uninterruptible, + ) + : Effect.void; + return track( + worktreeSetupTracker + .finish(threadId, "cancelled") + .pipe( + Effect.flatMap((snapshot) => + snapshot ? recordWorktreeSetup(snapshot, options) : Effect.void, + ), + ), + ).pipe( + Effect.andThen(removeCreatedWorktree), + // Interruptions roll back too, tracked or not: a + // created-but-never-started thread must not outlive its + // bootstrap. Only the message is specific to a cancelled + // worktree setup. + Effect.andThen( + cleanupAndFail( + cause, + tracked ? new OrchestrationDispatchCommandError({ - message: dispatchError.message, - ...(dispatchError.cause !== undefined - ? { cause: dispatchError.cause } - : {}), - bootstrapThreadDisposition: "deleted", + message: "Worktree setup cancelled.", }) : dispatchError, ), - }), - ); + ), + ); + } + return track( + worktreeSetupTracker + .finish(threadId, "failed", dispatchError.message) + .pipe( + Effect.flatMap((snapshot) => + snapshot ? recordWorktreeSetup(snapshot, options) : Effect.void, + ), + ), + ).pipe(Effect.andThen(cleanupAndFail(cause, dispatchError))); }), + // Cancellation must finish recording and rollback after the bootstrap is interrupted. + Effect.uninterruptible, ); + + // The bootstrap outlives the connection that asked for it: a reload + // or a dropped socket must not abandon a half-made worktree, and + // the thread it created is already visible to every client. The + // caller only waits on the detached fiber; a user cancel interrupts + // it through the tracker. + const runBootstrap = tracked + ? Effect.gen(function* () { + // Fork and register as one step: a detached fiber keeps going + // if the caller is interrupted, so it must never exist without + // the tracker entry that cancel and the stage updates key on. + const fiber = yield* Effect.uninterruptible( + Effect.gen(function* () { + const fiber = yield* Effect.forkDetach(settledBootstrapProgram); + yield* worktreeSetupTracker.begin({ + threadId, + branch: bootstrap?.prepareWorktree?.branch ?? null, + baseRef: bootstrap?.prepareWorktree?.baseBranch ?? null, + stages: ["fetch", "checkout", "submodules", "setup-script", "agent"], + fiber, + }); + return fiber; + }), + ); + return yield* Fiber.join(fiber); + }) + : settledBootstrapProgram; + + return yield* runBootstrap; }); return { diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index 16daf83ef370..c324a0d9579b 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -11268,12 +11268,25 @@ it.layer(NodeServices.layer)("server router seam", (it) => { ); it.effect.each([ - { caseName: "a non-repository", isRepository: false, failFetch: false }, - { caseName: "a base without a commit", isRepository: true, failFetch: false }, - { caseName: "a fetch failure", isRepository: true, failFetch: true }, + { caseName: "a non-repository", isRepository: false, failFetch: false, requireWorktree: true }, + { + caseName: "a base without a commit", + isRepository: true, + failFetch: false, + requireWorktree: true, + }, + { caseName: "a fetch failure", isRepository: true, failFetch: true, requireWorktree: true }, + // Nothing was created, so the draft goes back to the composer whether or + // not the worktree was the part that had to work. + { + caseName: "a fetch failure the worktree did not depend on", + isRepository: true, + failFetch: true, + requireWorktree: false, + }, ])( - "rejects required worktree bootstrap before creating a thread for $caseName", - ({ isRepository, failFetch }) => + "rejects worktree bootstrap before creating a thread for $caseName", + ({ isRepository, failFetch, requireWorktree }) => Effect.gen(function* () { const dispatchedCommands: Array = []; const createWorktree = vi.fn( @@ -11339,7 +11352,7 @@ it.layer(NodeServices.layer)("server router seam", (it) => { prepareWorktree: { projectCwd: "/tmp/project", baseBranch: "main", - requireWorktree: true, + ...(requireWorktree ? { requireWorktree: true } : {}), startFromOrigin: failFetch, }, }, @@ -11979,6 +11992,108 @@ it.layer(NodeServices.layer)("server router seam", (it) => { }).pipe(Effect.provide(NodeHttpServer.layerTest)), ); + // A bootstrap without prepareWorktree is not tracked, so there is no setup + // card to cancel from and no detached fiber: it lives and dies with the + // request. Losing that request must still roll the created thread back. + it.effect("rolls back a created thread when an untracked bootstrap is interrupted", () => + Effect.gen(function* () { + const dispatchedCommands: Array = []; + const setupEntered = yield* Deferred.make(); + const setupRelease = yield* Deferred.make(); + const threadDeleted = yield* Deferred.make(); + const runForThread = vi.fn( + ( + _: Parameters< + ProjectSetupScriptRunner.ProjectSetupScriptRunner["Service"]["runForThread"] + >[0], + ) => + Deferred.succeed(setupEntered, undefined).pipe( + Effect.andThen(Deferred.await(setupRelease)), + Effect.as({ + status: "started" as const, + scriptId: "setup", + scriptName: "Setup", + scriptCommand: "npm install", + terminalId: "setup-setup", + cwd: "/tmp/existing-worktree", + async: true, + }), + ), + ); + + yield* buildAppUnderTest({ + layers: { + vcsDriver: { + isInsideWorkTree: () => Effect.succeed(true), + }, + gitVcsDriver: { + execute: () => Effect.succeed(SUCCESSFUL_GIT_EXECUTION), + }, + orchestrationEngine: { + dispatch: (command) => + Effect.gen(function* () { + dispatchedCommands.push(command); + if (command.type === "thread.delete") { + yield* Deferred.succeed(threadDeleted, undefined); + } + return { sequence: dispatchedCommands.length }; + }), + readEvents: () => Stream.empty, + }, + projectSetupScriptRunner: { + runForThread, + }, + }, + }); + + const createdAt = "2026-01-01T00:00:00.000Z"; + const threadId = ThreadId.make("thread-bootstrap-untracked-interrupt"); + const wsUrl = yield* getWsServerUrl("/ws"); + const dispatchFiber = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.dispatchCommand]({ + type: "thread.turn.start", + commandId: CommandId.make("cmd-bootstrap-untracked-interrupt"), + threadId, + message: { + messageId: MessageId.make("msg-bootstrap-untracked-interrupt"), + role: "user", + text: "hello", + attachments: [], + }, + modelSelection: defaultModelSelection, + runtimeMode: "full-access", + interactionMode: "default", + bootstrap: { + createThread: { + projectId: defaultProjectId, + title: "Bootstrap Thread", + modelSelection: defaultModelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: "main", + worktreePath: "/tmp/existing-worktree", + createdAt, + }, + runSetupScript: true, + }, + createdAt, + }), + ), + ).pipe(Effect.forkChild); + + yield* Deferred.await(setupEntered); + yield* Fiber.interrupt(dispatchFiber); + yield* Deferred.await(threadDeleted); + yield* Deferred.succeed(setupRelease, undefined); + + assert.deepEqual( + dispatchedCommands.map((command) => command.type), + ["thread.create", "thread.message.user.append", "thread.delete"], + ); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + it.effect("cleans up created bootstrap threads when worktree creation defects", () => Effect.gen(function* () { const dispatchedCommands: Array = [];