From f6cf6723d59fa3a044c84c86ea4026f398e31d91 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Thu, 17 Sep 2026 15:59:59 -0400 Subject: [PATCH 1/3] fix(server): worktree setup progress reaches clients during bootstrap The fork owns bootstrap dispatch in ThreadBootstrapService, so upstream's new setup tracking arrived with its subscription and card but nothing feeding them: no stage ever ran, cancel had no fiber to interrupt, and a reload found no durable record of the setup. Signed-off-by: Yordis Prieto --- .../orchestration/Layers/ThreadBootstrap.ts | 630 ++++++++++++++++-- 1 file changed, 560 insertions(+), 70 deletions(-) diff --git a/apps/server/src/orchestration/Layers/ThreadBootstrap.ts b/apps/server/src/orchestration/Layers/ThreadBootstrap.ts index 8627b2d024d7..12fee7f68d94 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,179 @@ 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 && + bootstrap.prepareWorktree?.requireWorktree === true && + !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 - ? new OrchestrationDispatchCommandError({ - message: dispatchError.message, - ...(dispatchError.cause !== undefined - ? { cause: dispatchError.cause } - : {}), - bootstrapThreadDisposition: "deleted", - }) - : dispatchError, + 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), + Effect.andThen( + tracked + ? cleanupAndFail( + cause, + new OrchestrationDispatchCommandError({ + message: "Worktree setup cancelled.", + }), + ) + : Effect.fail(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 { From 9786725c5f0a0e43523ef1a0ca94897eedfa97b2 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Thu, 17 Sep 2026 16:20:55 -0400 Subject: [PATCH 2/3] fix(server): a cancelled bootstrap must not leave its thread behind An untracked bootstrap lives and dies with its request, so losing that request has to roll the created thread back too. Signed-off-by: Yordis Prieto --- .../orchestration/Layers/ThreadBootstrap.ts | 18 ++-- apps/server/src/server.test.ts | 102 ++++++++++++++++++ 2 files changed, 113 insertions(+), 7 deletions(-) diff --git a/apps/server/src/orchestration/Layers/ThreadBootstrap.ts b/apps/server/src/orchestration/Layers/ThreadBootstrap.ts index 12fee7f68d94..f41afe393955 100644 --- a/apps/server/src/orchestration/Layers/ThreadBootstrap.ts +++ b/apps/server/src/orchestration/Layers/ThreadBootstrap.ts @@ -837,15 +837,19 @@ const makeThreadBootstrap = Effect.gen(function* () { ), ).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( - tracked - ? cleanupAndFail( - cause, - new OrchestrationDispatchCommandError({ + cleanupAndFail( + cause, + tracked + ? new OrchestrationDispatchCommandError({ message: "Worktree setup cancelled.", - }), - ) - : Effect.fail(dispatchError), + }) + : dispatchError, + ), ), ); } diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index 16daf83ef370..9c9439bfd4f0 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -11979,6 +11979,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 = []; From 6aeae45a2e594e5ede7f1d2274cdc70814b1c80a Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Thu, 17 Sep 2026 16:46:34 -0400 Subject: [PATCH 3/3] fix(server): say a thread was never created whenever it was not A bootstrap can fail before the create dispatch for reasons that have nothing to do with the worktree being required, and the client needs the draft back in the composer in those cases too. Signed-off-by: Yordis Prieto --- .../orchestration/Layers/ThreadBootstrap.ts | 5 +--- apps/server/src/server.test.ts | 25 ++++++++++++++----- 2 files changed, 20 insertions(+), 10 deletions(-) diff --git a/apps/server/src/orchestration/Layers/ThreadBootstrap.ts b/apps/server/src/orchestration/Layers/ThreadBootstrap.ts index f41afe393955..b73090961428 100644 --- a/apps/server/src/orchestration/Layers/ThreadBootstrap.ts +++ b/apps/server/src/orchestration/Layers/ThreadBootstrap.ts @@ -778,10 +778,7 @@ const makeThreadBootstrap = Effect.gen(function* () { ), onSuccess: (threadDeleted) => Effect.fail( - threadDeleted || - (bootstrap?.createThread && - bootstrap.prepareWorktree?.requireWorktree === true && - !createdThread) + threadDeleted || (bootstrap?.createThread && !createdThread) ? new OrchestrationDispatchCommandError({ message: dispatchError.message, ...(dispatchError.cause !== undefined ? { cause: dispatchError.cause } : {}), diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index 9c9439bfd4f0..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, }, },