diff --git a/apps/mobile/src/features/threads/thread-work-log.tsx b/apps/mobile/src/features/threads/thread-work-log.tsx index 2f06b9462b4d..f02e4b8d1123 100644 --- a/apps/mobile/src/features/threads/thread-work-log.tsx +++ b/apps/mobile/src/features/threads/thread-work-log.tsx @@ -41,7 +41,17 @@ import { View, } from "react-native"; import Svg, { Defs, LinearGradient, Rect, Stop } from "react-native-svg"; -import type { EnvironmentId, ToolActivityIcon } from "@t3tools/contracts"; +import { + ORCHESTRATION_V2_WORKSPACE_PREPARATION_FAILURE_CODE, + type EnvironmentId, + type RunId, + type ThreadId, + type ToolActivityIcon, +} from "@t3tools/contracts"; +import { useAtomValue } from "@effect/atom-react"; +import { scopeThreadRef } from "@t3tools/client-runtime/environment"; +import { environmentThreadDetails, threadEnvironment } from "../../state/threads"; +import { useAtomCommand } from "../../state/use-atom-command"; import { toolActivityFaviconUrl } from "@t3tools/shared/favicon"; import { AppText as Text } from "../../components/AppText"; @@ -770,6 +780,44 @@ function workLogRowKey(row: ThreadFeedActivity): string { return row.id; } +/** Shown while the thread's run still ends in this failed preparation. */ +function WorkspacePreparationRetryButton(props: { + readonly environmentId: EnvironmentId; + readonly threadId: ThreadId; + readonly runId: RunId; +}) { + const retryable = useAtomValue( + environmentThreadDetails.threadAtom(scopeThreadRef(props.environmentId, props.threadId)), + (thread) => { + const run = thread?.projection.runs.find((candidate) => candidate.id === props.runId); + return run?.status === "failed" && run.workspacePreparation !== undefined; + }, + ); + const retry = useAtomCommand(threadEnvironment.retryWorkspacePreparation, "retry setup"); + const [busy, setBusy] = useState(false); + if (!retryable) return null; + return ( + { + setBusy(true); + void Haptics.selectionAsync(); + void retry({ + environmentId: props.environmentId, + input: { threadId: props.threadId, runId: props.runId }, + }).finally(() => setBusy(false)); + }} + className="ml-7 mt-2 min-h-11 flex-row items-center gap-1.5 self-start rounded-full border border-border px-4" + style={{ opacity: busy ? 0.5 : 1 }} + > + + Retry + + ); +} + const ThreadWorkLogRow = memo(function ThreadWorkLogRow( props: Omit< ThreadWorkLogProps, @@ -850,6 +898,14 @@ const ThreadWorkLogRow = memo(function ThreadWorkLogRow( {failureItem.failure.message} ) : null} + {failureItem.failure.code === ORCHESTRATION_V2_WORKSPACE_PREPARATION_FAILURE_CODE && + failureItem.runId !== null ? ( + + ) : null} ); diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 2707fe5b8344..03bc55e4e536 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -11,6 +11,7 @@ import { legacyThreadPullRequestKey, } from "@t3tools/shared/threadPullRequests"; import { + ORCHESTRATION_V2_WORKSPACE_PREPARATION_FAILURE_CODE, type ChatAttachment, CommandId, isProviderNativeSubagentThread, @@ -386,6 +387,7 @@ function commandThreadId(command: OrchestrationV2ServerCommand): ThreadId { case "prepared-run.release": case "prepared-run.progress": case "prepared-run.fail": + case "prepared-run.retry": case "run.interrupt": case "queued-message.promote-to-steer": case "queue.resume": @@ -428,6 +430,14 @@ function pendingThreadTitleGenerationEffect( const WORKSPACE_PREPARATION_INPUT = "Preparing workspace"; +/** A reopened preparation item drops the output and exit code of the attempt it replaces. */ +function withoutPreparationResult( + item: Extract, +) { + const { output: _output, exitCode: _exitCode, outputIndicatesFailure: _failure, ...rest } = item; + return rest; +} + function isBlockingRun(run: OrchestrationV2Run): boolean { return ( run.status === "preparing" || @@ -5132,6 +5142,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ...(command.restartContinuationOfRunId === undefined ? {} : { restartContinuationOfRunId: command.restartContinuationOfRunId }), + ...(dispatchMode.type === "defer_start" && dispatchMode.workspaceStrategy !== undefined + ? { workspacePreparation: dispatchMode.workspaceStrategy } + : {}), ...wakeWorkStartedAt(projection.runs, command), }; const attempt: OrchestrationV2RunAttempt = { @@ -7471,7 +7484,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio command: Extract< OrchestrationV2Command, { - readonly type: "prepared-run.release" | "prepared-run.progress" | "prepared-run.fail"; + readonly type: + | "prepared-run.release" + | "prepared-run.progress" + | "prepared-run.fail" + | "prepared-run.retry"; } >, projection: Pick< @@ -7494,7 +7511,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio candidate.input === WORKSPACE_PREPARATION_INPUT, ); if ( - run?.status !== "preparing" || + run?.status !== (command.type === "prepared-run.retry" ? "failed" : "preparing") || attempt === undefined || rootNode === undefined || providerThread === undefined || @@ -7712,7 +7729,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio completedAt: now, updatedAt: now, type: "error", - failure: command.failure, + failure: { + ...command.failure, + code: ORCHESTRATION_V2_WORKSPACE_PREPARATION_FAILURE_CODE, + }, }, }); yield* emitEvent({ @@ -7726,6 +7746,90 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }); }); + /** + * Returns a run whose workspace preparation failed to preparing. The failure + * item turns cancelled so clients stop offering the retry; ThreadLaunchService + * runs the recorded preparation again once this commits. + */ + const dispatchPreparedRunRetry = ( + command: Extract, + events: Ref.Ref>, + ) => + Effect.gen(function* () { + const projection = yield* loadProjectionForCommand( + command, + ["runs", "attempts", "nodes", "providerThreads", "turnItems"], + { turnItemTypes: ["command_execution", "error"], turnItemRunId: command.runId }, + ); + const state = preparedRunState(command, projection); + const failureItem = projection.turnItems.find( + (candidate): candidate is Extract => + candidate.type === "error" && + candidate.status === "failed" && + candidate.failure.code === ORCHESTRATION_V2_WORKSPACE_PREPARATION_FAILURE_CODE, + ); + if ( + state === null || + state.run.workspacePreparation === undefined || + failureItem === undefined || + projection.thread.archivedAt !== null || + projection.thread.deletedAt !== null + ) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: `Run ${command.runId} has no failed workspace preparation to retry.`, + }); + } + if (projection.runs.some((run) => run.id !== state.run.id && isBlockingRun(run))) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: "Another run is active on this thread.", + }); + } + const now = yield* DateTime.now; + const emitEvent = emit(events, command); + const scope = { + threadId: command.threadId, + runId: state.run.id, + nodeId: state.rootNode.id, + providerInstanceId: state.run.providerInstanceId, + occurredAt: now, + }; + yield* emitEvent({ + ...scope, + type: "turn-item.updated", + payload: { ...failureItem, status: "cancelled", updatedAt: now }, + }); + yield* emitEvent({ + ...scope, + type: "turn-item.updated", + payload: { + ...withoutPreparationResult(state.preparationItem), + status: "running", + title: WORKSPACE_PREPARATION_INPUT, + completedAt: null, + updatedAt: now, + }, + }); + yield* emitEvent({ + ...scope, + type: "run-attempt.updated", + payload: { ...state.attempt, status: "pending", completedAt: null }, + }); + yield* emitEvent({ + ...scope, + type: "node.updated", + payload: { ...state.rootNode, status: "pending", completedAt: null }, + }); + yield* emitEvent({ + ...scope, + type: "run.updated", + payload: { ...state.run, status: "preparing", completedAt: null }, + }); + }); + /** * Ends the background work a settled thread still shows that no provider * process will report on: work on the provider thread whose interrupt just @@ -9323,6 +9427,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio case "prepared-run.fail": yield* dispatchPreparedRunFail(command, events); break; + case "prepared-run.retry": + yield* dispatchPreparedRunRetry(command, events); + break; case "runtime-request.respond": yield* dispatchRuntimeRequestRespond(command, events, effects); break; diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts index 10eb4ac9485e..5e53250915f8 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts @@ -10,6 +10,7 @@ import * as ThreadMessageIntake from "./ThreadMessageIntake.ts"; import { assert, it, vi } from "@effect/vitest"; import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; import { + ORCHESTRATION_V2_WORKSPACE_PREPARATION_FAILURE_CODE, ChatAttachmentId, ComposerContextId, type ChatAttachment, @@ -127,6 +128,10 @@ function makeHarness(options: HarnessOptions = {}) { const renameBranch = vi.fn( options.renameBranch ?? ((input) => Effect.succeed({ branch: input.newBranch })), ); + const removeWorktree = vi.fn( + (_input: Parameters[0]) => + Effect.void, + ); const runSetup = vi.fn( options.runSetup ?? (() => Effect.succeed({ status: "no-script" as const })), ); @@ -164,7 +169,7 @@ function makeHarness(options: HarnessOptions = {}) { fetchRemote: options.fetchRemote ?? (() => Effect.void), remoteExists: () => Effect.succeed(true), remoteBranchExists: () => Effect.succeed(true), - removeWorktree: () => Effect.void, + removeWorktree, resolveRemoteTrackingCommit: () => Effect.succeed({ commitSha: "remote-main-sha", remoteRefName: "origin/main" }), }), @@ -220,6 +225,7 @@ function makeHarness(options: HarnessOptions = {}) { externalServices, ), createWorktree, + removeWorktree, renameBranch, generateBranchName, generateThreadTitle, @@ -1293,6 +1299,172 @@ it.effect("shows the fetch diagnosis when preparing a worktree from origin fails }).pipe(Effect.provide(harness.layer)); }); +it.effect("retries a failed workspace preparation on the same run", () => { + let fetchFailures = 1; + const harness = makeHarness({ + fetchRemote: () => + fetchFailures-- > 0 + ? Effect.fail( + new GitCommandError({ + operation: "GitVcsDriver.fetchRemote", + command: "git", + cwd: project.workspaceRoot, + detail: "Git could not update a local reference.", + exitCode: 1, + }), + ) + : Effect.void, + }); + return Effect.gen(function* () { + const launches = yield* ThreadLaunch.ThreadLaunchService; + const outbox = yield* EffectOutbox.EffectOutboxV2; + const threads = yield* ThreadManagement.ThreadManagementService; + const launched = yield* launches.launch( + launchInput({ + command: "command:launch:retry", + thread: "thread:launch:retry", + message: "Retry me", + workspace: { type: "worktree", baseRef: "main", startFromOrigin: true }, + }), + ); + yield* waitUntil(() => + threads + .getThreadProjection(launched.threadId) + .pipe(Effect.map((projection) => projection.runs[0]?.status === "failed")), + ); + const failed = yield* threads.getThreadProjection(launched.threadId); + const runId = failed.runs[0]!.id; + assert.equal( + failed.turnItems.find((item) => item.type === "error")?.failure.code, + ORCHESTRATION_V2_WORKSPACE_PREPARATION_FAILURE_CODE, + ); + + const retry = { + commandId: CommandId.make("command:launch:retry:1"), + threadId: launched.threadId, + runId, + }; + yield* launches.retryPreparation(retry); + yield* waitUntil(() => + outbox + .listByCommandId(CommandId.make("command:launch:retry:1:release")) + .pipe(Effect.map((effects) => effects.length === 1)), + ); + const retried = yield* threads.getThreadProjection(launched.threadId); + assert.equal(retried.runs.length, 1); + assert.equal(retried.runs[0]?.status, "starting"); + assert.equal(retried.thread.worktreePath, "/repo-worktrees/feature"); + assert.equal(retried.turnItems.find((item) => item.type === "error")?.status, "cancelled"); + assert.equal( + retried.turnItems.find((item) => item.type === "command_execution")?.status, + "completed", + ); + assert.equal(harness.createWorktree.mock.calls.length, 1); + + // The run left preparation, so a second retry has nothing to do. + const rejected = yield* launches + .retryPreparation({ ...retry, commandId: CommandId.make("command:launch:retry:2") }) + .pipe(Effect.flip); + assert.equal(rejected._tag, "OrchestratorDispatchError"); + }).pipe(Effect.provide(harness.layer)); +}); + +it.effect("a retry reuses a recorded worktree without undoing its branch rename", () => { + let setupFailures = 1; + const harness = makeHarness({ + runSetup: () => + setupFailures-- > 0 + ? Effect.fail(new Error("setup failed") as never) + : Effect.succeed({ status: "no-script" as const }), + }); + return Effect.gen(function* () { + const launches = yield* ThreadLaunch.ThreadLaunchService; + const outbox = yield* EffectOutbox.EffectOutboxV2; + const threads = yield* ThreadManagement.ThreadManagementService; + const tracker = yield* WorktreeSetupTracker.WorktreeSetupTracker; + const launched = yield* launches.launch( + launchInput({ + command: "command:launch:reuse", + thread: "thread:launch:reuse", + message: "Reuse the worktree", + workspace: { type: "worktree", baseRef: "main" }, + }), + ); + yield* waitUntil(() => + threads + .getThreadProjection(launched.threadId) + .pipe( + Effect.map( + (projection) => + projection.runs[0]?.status === "failed" && + projection.thread.branch === "generated-branch", + ), + ), + ); + const failed = yield* threads.getThreadProjection(launched.threadId); + assert.equal(failed.thread.worktreePath, "/repo-worktrees/feature"); + + yield* launches.retryPreparation({ + commandId: CommandId.make("command:launch:reuse:retry"), + threadId: launched.threadId, + runId: failed.runs[0]!.id, + }); + yield* waitUntil(() => + outbox + .listByCommandId(CommandId.make("command:launch:reuse:retry:release")) + .pipe(Effect.map((effects) => effects.length === 1)), + ); + const retried = yield* threads.getThreadProjection(launched.threadId); + assert.equal(retried.runs[0]?.status, "starting"); + // The retry neither checks out again nor puts back the temporary branch. + assert.equal(harness.createWorktree.mock.calls.length, 1); + assert.equal(harness.renameBranch.mock.calls.length, 1); + assert.equal(retried.thread.branch, "generated-branch"); + assert.equal(retried.thread.worktreePath, "/repo-worktrees/feature"); + // Clients see the retry's setup, not the failed one it replaced. + const snapshot = yield* tracker.get(launched.threadId); + assert.equal(snapshot?.phase, "done"); + assert.deepEqual( + snapshot?.stages.map((stage) => stage.id), + ["setup-script", "agent"], + ); + }).pipe(Effect.provide(harness.layer)); +}); + +it.effect("removes a worktree that failed before the thread recorded it", () => { + const harness = makeHarness({ + // A checkout that dies after claiming its directory. + createWorktree: (_input, options) => + (options?.progress?.onWorktreeClaimed?.("/repo-worktrees/partial") ?? Effect.void).pipe( + Effect.andThen(Effect.fail(new Error("checkout failed") as never)), + ), + }); + return Effect.gen(function* () { + const launches = yield* ThreadLaunch.ThreadLaunchService; + const threads = yield* ThreadManagement.ThreadManagementService; + const launched = yield* launches.launch( + launchInput({ + command: "command:launch:partial-worktree", + thread: "thread:launch:partial-worktree", + message: "Partial checkout", + workspace: { type: "worktree", baseRef: "main" }, + }), + ); + yield* waitUntil(() => + threads + .getThreadProjection(launched.threadId) + .pipe(Effect.map((projection) => projection.runs[0]?.status === "failed")), + ); + const projection = yield* threads.getThreadProjection(launched.threadId); + // Unrecorded, so a retry would create a second checkout beside it. + assert.equal(projection.thread.worktreePath, null); + assert.deepEqual( + harness.removeWorktree.mock.calls.map(([input]) => input.path), + ["/repo-worktrees/partial"], + ); + }).pipe(Effect.provide(harness.layer)); +}); + it.effect.each(["worktree", "setup"] as const)( "%s failure keeps the thread and message visible and emits failure items", (failurePoint) => @@ -1799,6 +1971,7 @@ it.effect("shared intake preserves durable attachment bytes after a lost launch }), ), ), + retryPreparation: launches.retryPreparation, }), Effect.flip, ); diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.ts index faf613376920..555d3266eba2 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.ts @@ -39,6 +39,7 @@ import * as ServerSettings from "../serverSettings.ts"; import * as TextGeneration from "../textGeneration/TextGeneration.ts"; import * as CommandReceiptStore from "./CommandReceiptStore.ts"; import * as IdAllocator from "./IdAllocator.ts"; +import type * as Orchestrator from "./Orchestrator.ts"; import { makeProviderFailure } from "./ProviderFailure.ts"; import { randomUuidV4 } from "./RandomUuid.ts"; import * as ThreadManagement from "./ThreadManagementService.ts"; @@ -90,6 +91,25 @@ export interface ThreadLaunchInput { readonly creationSource: OrchestrationV2CreationSource; } +/** What workspace preparation reads from a launch; a retry rebuilds it from the run. */ +type PreparationInput = Pick< + ThreadLaunchInput, + "commandId" | "projectId" | "workspaceStrategy" | "initialMessage" +> & { + /** + * Set when a retry reuses the worktree its failed attempt created and + * recorded. Its setup is tracked like a new one, but the thread already + * records the workspace, and a branch rename may still be running. + */ + readonly reusedWorktree?: { readonly baseRef: string }; +}; + +export interface ThreadLaunchRetryInput { + readonly commandId: CommandId; + readonly threadId: ThreadId; + readonly runId: RunId; +} + export interface ThreadLaunchResult { readonly threadId: ThreadId; readonly projection: OrchestrationV2ThreadProjection; @@ -128,6 +148,10 @@ export class ThreadLaunchService extends Context.Service< readonly launch: ( input: ThreadLaunchInput, ) => Effect.Effect; + /** Dispatches prepared-run.retry and prepares the run's workspace again. */ + readonly retryPreparation: ( + input: ThreadLaunchRetryInput, + ) => Effect.Effect; } >()("t3/orchestration-v2/ThreadLaunchService") {} @@ -161,7 +185,7 @@ const make = Effect.gen(function* () { yield* Effect.addFinalizer(() => Scope.close(preparationScope, Exit.void)); const mapError = - (input: ThreadLaunchInput, operation: ThreadLaunchError["operation"], threadId?: ThreadId) => + (input: PreparationInput, operation: ThreadLaunchError["operation"], threadId?: ThreadId) => (cause: unknown) => new ThreadLaunchError({ operation, @@ -201,7 +225,7 @@ const make = Effect.gen(function* () { }); const prepareInBackground = Effect.fn("ThreadLaunchService.prepareInBackground")(function* ( - input: ThreadLaunchInput, + input: PreparationInput, threadId: ThreadId, runId: RunId | null, ) { @@ -216,10 +240,12 @@ const make = Effect.gen(function* () { ), ); - const tracked = input.workspaceStrategy.type === "worktree"; + const reused = input.reusedWorktree; + const tracked = input.workspaceStrategy.type === "worktree" || reused !== undefined; let createdWorktreePath: string | null = null; let setupTerminalId: string | null = null; - if (tracked) { + let workspaceRecorded = false; + if (input.workspaceStrategy.type === "worktree") { yield* setupTracker.begin({ threadId, branch: input.workspaceStrategy.branch ?? null, @@ -227,6 +253,14 @@ const make = Effect.gen(function* () { stages: ["fetch", "checkout", "setup-script", "agent"], fiber: yield* Effect.fiber, }); + } else if (reused !== undefined) { + yield* setupTracker.begin({ + threadId, + branch: input.workspaceStrategy.branch ?? null, + baseRef: reused.baseRef, + stages: ["setup-script", "agent"], + fiber: yield* Effect.fiber, + }); } yield* Effect.gen(function* () { const initialMessage = input.initialMessage; @@ -358,21 +392,27 @@ const make = Effect.gen(function* () { yield* setupTracker.stageStatus(threadId, "checkout", "done"); } - yield* threads - .dispatch({ - type: "thread.metadata.update", - commandId: CommandId.make(`${input.commandId}:workspace`), - threadId, - branch, - worktreePath, - }) - .pipe(Effect.mapError(mapError(input, "update-thread", threadId))); + // A reused worktree is already recorded, and rewriting it could undo + // the first attempt's branch rename. + if (reused === undefined) { + yield* threads + .dispatch({ + type: "thread.metadata.update", + commandId: CommandId.make(`${input.commandId}:workspace`), + threadId, + branch, + worktreePath, + }) + .pipe(Effect.mapError(mapError(input, "update-thread", threadId))); + } + workspaceRecorded = true; // Rename temporary branches (server-invented above, or sent by clients // that name worktrees themselves) in the background so generation latency // never delays provisioning or the provider turn. The temporary name // simply sticks if generation or the rename fails. if ( + reused === undefined && worktreePath !== null && branch !== null && initialMessage !== undefined && @@ -512,27 +552,41 @@ const make = Effect.gen(function* () { cancelled ? "cancelled" : "failed", cancelled ? null : failureDetail(Cause.squash(cause)), ); - if (cancelled && tracked && createdWorktreePath) { + // A cancelled setup leaves nothing behind. A failed one keeps a worktree + // the thread recorded, so a retry reuses it, and removes one it never + // recorded, which a retry would otherwise duplicate. + if (tracked && createdWorktreePath && (cancelled || !workspaceRecorded)) { if (setupTerminalId) yield* terminals .close({ threadId, terminalId: setupTerminalId, deleteHistory: true }) .pipe(Effect.ignore); + const removedPath = createdWorktreePath; + // The thread forgets the worktree only once it is gone; a failed + // removal leaves the directory for the user to clean up rather than + // reusing a checkout that may be half written. yield* git - .removeWorktree({ - cwd: project.workspaceRoot, - path: createdWorktreePath, - force: true, - }) - .pipe(Effect.ignore); - yield* threads - .dispatch({ - type: "thread.metadata.update", - commandId: CommandId.make(`${input.commandId}:cancel-workspace`), - threadId, - worktreePath: null, - branch: null, - }) - .pipe(Effect.ignore); + .removeWorktree({ cwd: project.workspaceRoot, path: removedPath, force: true }) + .pipe( + Effect.andThen( + threads + .dispatch({ + type: "thread.metadata.update", + commandId: CommandId.make(`${input.commandId}:cancel-workspace`), + threadId, + worktreePath: null, + branch: null, + }) + .pipe(Effect.ignore), + ), + Effect.catchCause((removeCause) => + Effect.logWarning("Failed to remove an abandoned thread worktree", { + commandId: input.commandId, + threadId, + path: removedPath, + cause: removeCause, + }), + ), + ); } }), ), @@ -540,7 +594,7 @@ const make = Effect.gen(function* () { }); const failPreparedRun = ( - input: ThreadLaunchInput, + input: Pick, threadId: ThreadId, runId: RunId | null, cause: unknown, @@ -565,7 +619,6 @@ const make = Effect.gen(function* () { }), }) .pipe( - Effect.mapError(mapError(input, "fail-run", threadId)), Effect.catchCause((persistCause) => Effect.logWarning("Failed to persist thread workspace preparation failure", { commandId: input.commandId, @@ -592,7 +645,7 @@ const make = Effect.gen(function* () { }); const schedulePreparation = Effect.fn("ThreadLaunchService.schedulePreparation")(function* ( - input: ThreadLaunchInput, + input: PreparationInput, threadId: ThreadId, runId: RunId | null, ) { @@ -763,7 +816,7 @@ const make = Effect.gen(function* () { ...(input.initialMessage.context ? { context: input.initialMessage.context } : {}), ...(input.generateTitle === true ? { titleSeed: input.title } : {}), modelSelection: input.modelSelection, - dispatchMode: { type: "defer_start" }, + dispatchMode: { type: "defer_start", workspaceStrategy }, createdBy: input.createdBy, creationSource: input.creationSource, }) @@ -836,7 +889,78 @@ const make = Effect.gen(function* () { }, ); - return ThreadLaunchService.of({ launch }); + const retryPreparation: ThreadLaunchService["Service"]["retryPreparation"] = Effect.fn( + "ThreadLaunchService.retryPreparation", + )(function* (input) { + const dispatched = yield* threads.dispatch({ + type: "prepared-run.retry", + commandId: input.commandId, + threadId: input.threadId, + runId: input.runId, + }); + // A replayed retry finds the run already past preparation, or prepared by + // the attempt that first reserved this command. + // From here the run is preparing again; anything that stops preparation + // from being scheduled must fail it, or it would wait in preparing forever. + const scheduled = yield* Effect.gen(function* () { + const projection = yield* threads.getThreadProjection(input.threadId); + const run = projection.runs.find((candidate) => candidate.id === input.runId); + const workspacePreparation = run?.workspacePreparation; + if (run?.status !== "preparing" || workspacePreparation === undefined) return; + if (!(yield* reservePreparation(input.commandId))) return; + yield* scheduleRetriedPreparation(input, projection, run, workspacePreparation).pipe( + Effect.onError(() => releasePreparation(input.commandId)), + ); + }).pipe(Effect.exit); + if (Exit.isFailure(scheduled)) { + yield* failPreparedRun(input, input.threadId, input.runId, Cause.squash(scheduled.cause)); + } + return dispatched; + }); + + const scheduleRetriedPreparation = ( + input: ThreadLaunchRetryInput, + projection: OrchestrationV2ThreadProjection, + run: OrchestrationV2ThreadProjection["runs"][number], + workspacePreparation: ThreadLaunchWorkspaceStrategy, + ) => { + const message = projection.messages.find((candidate) => candidate.id === run.userMessageId); + // A worktree the failed attempt already created is reused, not created again. + const reuse = + workspacePreparation.type === "worktree" && + projection.thread.worktreePath !== null && + projection.thread.branch !== null + ? { + strategy: { + type: "existing_worktree" as const, + worktreePath: projection.thread.worktreePath, + branch: projection.thread.branch, + }, + reusedWorktree: { baseRef: workspacePreparation.baseRef }, + } + : null; + return schedulePreparation( + { + commandId: input.commandId, + projectId: projection.thread.projectId, + workspaceStrategy: reuse?.strategy ?? workspacePreparation, + ...(reuse === null ? {} : { reusedWorktree: reuse.reusedWorktree }), + ...(message === undefined + ? {} + : { + initialMessage: { + text: message.text, + attachments: message.attachments, + ...(message.context ? { context: message.context } : {}), + }, + }), + }, + input.threadId, + run.id, + ); + }; + + return ThreadLaunchService.of({ launch, retryPreparation }); }); export const layer = Layer.effect(ThreadLaunchService, make); diff --git a/apps/server/src/orchestration-v2/testkit/OrchestratorScenario.ts b/apps/server/src/orchestration-v2/testkit/OrchestratorScenario.ts index b9cad107409a..0bb2d5053b4b 100644 --- a/apps/server/src/orchestration-v2/testkit/OrchestratorScenario.ts +++ b/apps/server/src/orchestration-v2/testkit/OrchestratorScenario.ts @@ -165,6 +165,7 @@ function commandThreadIds(command: OrchestrationV2Command): ReadonlyArray(); + export default function ChatView(props: ChatViewProps) { const { environmentId, @@ -3866,6 +3870,34 @@ export default function ChatView(props: ChatViewProps) { const cancelWorktreeSetup = useAtomCommand(vcsEnvironment.cancelWorktreeSetup, { reportFailure: false, }); + const retryWorkspacePreparation = useAtomCommand(threadEnvironment.retryWorkspacePreparation); + const retryableRunIdsKey = useMemo( + () => + [ + ...workspacePreparationRetryRunIds( + serverProjection?.runs ?? [], + serverProjection?.turnItems ?? [], + ), + ].join("\n"), + [serverProjection?.runs, serverProjection?.turnItems], + ); + // Keyed by content so the timeline context only changes when a retry appears or clears. + const retryableWorkspacePreparationRunIds = useMemo( + () => new Set(retryableRunIdsKey === "" ? [] : (retryableRunIdsKey.split("\n") as RunId[])), + [retryableRunIdsKey], + ); + const onRetryWorkspacePreparation = useCallback( + (runId: RunId) => { + // One retry per failed run: a second click lands after the run is preparing again. + if (!activeThreadRef || retryingWorkspacePreparationRunIds.has(runId)) return; + retryingWorkspacePreparationRunIds.add(runId); + void retryWorkspacePreparation({ + environmentId: activeThreadRef.environmentId, + input: { threadId: activeThreadRef.threadId, runId }, + }).finally(() => retryingWorkspacePreparationRunIds.delete(runId)); + }, + [activeThreadRef, retryWorkspacePreparation], + ); const onCancelWorktreeSetup = useCallback(() => { if (!worktreeSetup || worktreeSetup.phase !== "running") return; void cancelWorktreeSetup({ @@ -10689,6 +10721,9 @@ export default function ChatView(props: ChatViewProps) { activeTurnStartedAt={paintOnlyDisplayedTimeline ? null : activeWorkStartedAt} worktreeSetup={paintOnlyDisplayedTimeline ? null : worktreeSetup} onCancelWorktreeSetup={onCancelWorktreeSetup} + {...(paintOnlyDisplayedTimeline + ? {} + : { retryableWorkspacePreparationRunIds, onRetryWorkspacePreparation })} {...(draftId ? { onWorktreeSetupWorkLocally } : {})} {...(onOpenWorktreeSetupTerminal ? { onOpenWorktreeSetupTerminal } : {})} isPreparingWorktree={!paintOnlyDisplayedTimeline && isPreparingWorktree} diff --git a/apps/web/src/components/chat/MessagesTimeline.tsx b/apps/web/src/components/chat/MessagesTimeline.tsx index 8b2808b3a18d..5c0adecfea80 100644 --- a/apps/web/src/components/chat/MessagesTimeline.tsx +++ b/apps/web/src/components/chat/MessagesTimeline.tsx @@ -18,6 +18,7 @@ import { } from "./timelineMinimapItems"; import { COMPOSER_CONTEXT_KINDS, + ORCHESTRATION_V2_WORKSPACE_PREPARATION_FAILURE_CODE, type AssistantCitation, type EnvironmentId, type MessageId, @@ -139,6 +140,7 @@ import { WrenchIcon, XIcon, ZapIcon, + RotateCcwIcon, } from "lucide-react"; import type { ComposerContextId, @@ -322,6 +324,8 @@ interface TimelineRowSharedState { onToggleWorkGroup: (groupId: string, anchorKey: string) => void; onToggleWorkEntry: (anchorKey: string, collapsed: boolean) => void; onCancelWorktreeSetup: (() => void) | null; + retryableWorkspacePreparationRunIds: ReadonlySet; + onRetryWorkspacePreparation: ((runId: RunId) => void) | null; onWorktreeSetupWorkLocally: (() => void) | null; onOpenWorktreeSetupTerminal: ((terminalId: string) => void) | null; workGroupViewState: WorkGroupViewState; @@ -380,6 +384,7 @@ const TIMELINE_MAINTAIN_SCROLL_AT_END = { }, } as const satisfies MaintainScrollAtEndOptions; const EMPTY_TIMELINE_RUNS: ReadonlyArray = []; +const EMPTY_RUN_IDS: ReadonlySet = new Set(); // Streamed text lands a paragraph at a time. A smooth scroll to the end // turns each landing into a short glide instead of a jump. Thread switches // and layout settles keep the instant variant so nothing visibly travels. @@ -413,6 +418,9 @@ interface MessagesTimelineProps { activeTurnStartedAt?: string | null; worktreeSetup?: WorktreeSetupSnapshot | null; onCancelWorktreeSetup?: () => void; + /** Runs whose failed workspace preparation can be retried, keyed by run id. */ + retryableWorkspacePreparationRunIds?: ReadonlySet; + onRetryWorkspacePreparation?: (runId: RunId) => void; onWorktreeSetupWorkLocally?: () => void; onOpenWorktreeSetupTerminal?: (terminalId: string) => void; isPreparingWorktree?: boolean; @@ -496,6 +504,8 @@ export const MessagesTimeline = memo(function MessagesTimeline({ activeTurnStartedAt = null, worktreeSetup = null, onCancelWorktreeSetup, + retryableWorkspacePreparationRunIds = EMPTY_RUN_IDS, + onRetryWorkspacePreparation, onWorktreeSetupWorkLocally, onOpenWorktreeSetupTerminal, isPreparingWorktree = false, @@ -1161,6 +1171,8 @@ export const MessagesTimeline = memo(function MessagesTimeline({ onToggleWorkGroup, onToggleWorkEntry: suspendEndScrollMaintenanceForDisclosure, onCancelWorktreeSetup: onCancelWorktreeSetup ?? null, + retryableWorkspacePreparationRunIds, + onRetryWorkspacePreparation: onRetryWorkspacePreparation ?? null, onWorktreeSetupWorkLocally: onWorktreeSetupWorkLocally ?? null, onOpenWorktreeSetupTerminal: onOpenWorktreeSetupTerminal ?? null, workGroupViewState, @@ -1194,6 +1206,8 @@ export const MessagesTimeline = memo(function MessagesTimeline({ onToggleWorkGroup, suspendEndScrollMaintenanceForDisclosure, onCancelWorktreeSetup, + retryableWorkspacePreparationRunIds, + onRetryWorkspacePreparation, onWorktreeSetupWorkLocally, onOpenWorktreeSetupTerminal, workGroupViewState, @@ -4952,6 +4966,7 @@ const SimpleWorkEntryRow = memo(function SimpleWorkEntryRow(props: { const { workEntry, workspaceRoot, displayLabel } = props; const ctx = use(TimelineRowCtx); const { threadRef, onImageExpand, timestampFormat } = ctx; + const { retryableWorkspacePreparationRunIds, onRetryWorkspacePreparation } = ctx; const createdThread = workEntry.projectedItem?.item.type === "thread_created" ? workEntry.projectedItem.item @@ -4983,6 +4998,12 @@ const SimpleWorkEntryRow = memo(function SimpleWorkEntryRow(props: { const label = warning ? `Usage limit reached.${resetTime ? ` Retry after ${resetTime}.` : ""}` : workEntry.label; + const retryRunId = + failureItem.runId !== null && + retryableWorkspacePreparationRunIds.has(failureItem.runId) && + failureItem.failure.code === ORCHESTRATION_V2_WORKSPACE_PREPARATION_FAILURE_CODE + ? failureItem.runId + : null; return ( ) : null} + {retryRunId !== null && onRetryWorkspacePreparation ? ( +
+ +
+ ) : null} ); } diff --git a/packages/client-runtime/src/operations/commands.ts b/packages/client-runtime/src/operations/commands.ts index c0cb8e67504c..bedf552f9b20 100644 --- a/packages/client-runtime/src/operations/commands.ts +++ b/packages/client-runtime/src/operations/commands.ts @@ -237,6 +237,10 @@ export interface CancelQueuedRunInput extends ThreadCommandInput { readonly runId: RunId; } +export interface RetryWorkspacePreparationInput extends ThreadCommandInput { + readonly runId: RunId; +} + export interface EditQueuedRunInput extends ThreadCommandInput { readonly runId: RunId; readonly text: string; @@ -982,6 +986,17 @@ export const cancelQueuedRun = Effect.fn("EnvironmentCommands.cancelQueuedRun")( }); }); +export const retryWorkspacePreparation = Effect.fn("EnvironmentCommands.retryWorkspacePreparation")( + function* (input: RetryWorkspacePreparationInput) { + return yield* dispatch({ + type: "prepared-run.retry", + commandId: yield* allocateCommandId(input), + threadId: input.threadId, + runId: input.runId, + }); + }, +); + export const editQueuedRun = Effect.fn("EnvironmentCommands.editQueuedRun")(function* ( input: EditQueuedRunInput, ) { diff --git a/packages/client-runtime/src/state/threadCommands.ts b/packages/client-runtime/src/state/threadCommands.ts index e3ad5c07f583..ed8cc1ec97cc 100644 --- a/packages/client-runtime/src/state/threadCommands.ts +++ b/packages/client-runtime/src/state/threadCommands.ts @@ -21,6 +21,7 @@ import { type ThreadCommandInput, type ArchiveThreadInput, type CancelQueuedRunInput, + type RetryWorkspacePreparationInput, type CreateThreadInput, type DeleteThreadInput, type EditQueuedRunInput, @@ -65,6 +66,7 @@ import { promoteQueuedRun, reorderQueuedRun, resumeThreadQueue, + retryWorkspacePreparation, linkThreadPullRequest, respondToThreadApproval, respondToThreadUserInput, @@ -357,6 +359,12 @@ export function createThreadEnvironmentAtoms( scheduler, concurrency, }), + retryWorkspacePreparation: createEnvironmentCommand(runtime, { + label: "environment-data:commands:thread:retry-workspace-preparation", + execute: (input: RetryWorkspacePreparationInput) => retryWorkspacePreparation(input), + scheduler, + concurrency, + }), editQueuedRun: createEnvironmentCommand(runtime, { label: "environment-data:commands:thread:edit-queued-run", execute: (input: EditQueuedRunInput) => editQueuedRun(input), diff --git a/packages/client-runtime/src/state/turnItemPresentation.test.ts b/packages/client-runtime/src/state/turnItemPresentation.test.ts index af2b9fcbe83b..bfccbe63cf5f 100644 --- a/packages/client-runtime/src/state/turnItemPresentation.test.ts +++ b/packages/client-runtime/src/state/turnItemPresentation.test.ts @@ -1,8 +1,18 @@ -import { RunId, ThreadId, TurnItemId, type OrchestrationV2TurnItem } from "@t3tools/contracts"; +import { + ORCHESTRATION_V2_WORKSPACE_PREPARATION_FAILURE_CODE, + RunId, + ThreadId, + TurnItemId, + type OrchestrationV2Run, + type OrchestrationV2TurnItem, +} from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; import { describe, expect, it } from "vite-plus/test"; -import { turnItemIsWorkspacePreparation } from "./turnItemPresentation.ts"; +import { + turnItemIsWorkspacePreparation, + workspacePreparationRetryRunIds, +} from "./turnItemPresentation.ts"; function command(input: string): OrchestrationV2TurnItem { const now = DateTime.makeUnsafe("2026-08-03T00:00:00.000Z"); @@ -28,9 +38,67 @@ function command(input: string): OrchestrationV2TurnItem { }; } +function preparationFailure( + status: OrchestrationV2TurnItem["status"], + code: string | null = ORCHESTRATION_V2_WORKSPACE_PREPARATION_FAILURE_CODE, +): OrchestrationV2TurnItem { + const { + input: _input, + output: _output, + exitCode: _exitCode, + ...base + } = command("") as Extract; + return { + ...base, + id: TurnItemId.make("item-error"), + status, + title: "Workspace preparation failed", + type: "error", + failure: { class: "validation_error", message: "fetch failed", code, retryable: false }, + }; +} + +const worktree = { type: "worktree", baseRef: "main" } as const; +function run( + status: OrchestrationV2Run["status"], + workspacePreparation?: typeof worktree, +): Pick { + return { + id: RunId.make("run-1"), + status, + ...(workspacePreparation === undefined ? {} : { workspacePreparation }), + }; +} + describe("turnItemIsWorkspacePreparation", () => { it("identifies the synthetic workspace preparation command", () => { expect(turnItemIsWorkspacePreparation(command("Preparing workspace"))).toBe(true); expect(turnItemIsWorkspacePreparation(command("prepare workspace"))).toBe(false); }); + + it("hides a preparation failure only once a retry cancelled it", () => { + expect(turnItemIsWorkspacePreparation(preparationFailure("failed"))).toBe(false); + expect(turnItemIsWorkspacePreparation(preparationFailure("cancelled"))).toBe(true); + expect(turnItemIsWorkspacePreparation(preparationFailure("cancelled", null))).toBe(false); + }); +}); + +describe("workspacePreparationRetryRunIds", () => { + it("offers a retry while the run still ends in its failed preparation", () => { + const failure = [preparationFailure("failed")]; + expect([...workspacePreparationRetryRunIds([run("failed", worktree)], failure)]).toEqual([ + RunId.make("run-1"), + ]); + // Retried: the run is preparing again and the old failure is cancelled. + expect(workspacePreparationRetryRunIds([run("preparing", worktree)], failure).size).toBe(0); + // An older server records no preparation, so there is nothing to repeat. + expect(workspacePreparationRetryRunIds([run("failed")], failure).size).toBe(0); + // A provider error on a run that did reach the provider is not a preparation failure. + expect( + workspacePreparationRetryRunIds( + [run("failed", worktree)], + [preparationFailure("failed", "provider_crashed")], + ).size, + ).toBe(0); + }); }); diff --git a/packages/client-runtime/src/state/turnItemPresentation.ts b/packages/client-runtime/src/state/turnItemPresentation.ts index fe306c92ae34..5268aee52321 100644 --- a/packages/client-runtime/src/state/turnItemPresentation.ts +++ b/packages/client-runtime/src/state/turnItemPresentation.ts @@ -1,8 +1,49 @@ -import type { OrchestrationV2TurnItem } from "@t3tools/contracts"; +import { + ORCHESTRATION_V2_WORKSPACE_PREPARATION_FAILURE_CODE, + type OrchestrationV2Run, + type OrchestrationV2TurnItem, +} from "@t3tools/contracts"; const WORKSPACE_PREPARATION_INPUT = "Preparing workspace"; -/** Workspace setup is client bookkeeping; preparation failures have their own error item. */ +/** + * Workspace setup is client bookkeeping; preparation failures have their own + * error item. A retry cancels that item, which then has nothing left to say. + */ export function turnItemIsWorkspacePreparation(item: OrchestrationV2TurnItem): boolean { - return item.type === "command_execution" && item.input === WORKSPACE_PREPARATION_INPUT; + return ( + (item.type === "command_execution" && item.input === WORKSPACE_PREPARATION_INPUT) || + (item.type === "error" && + item.status === "cancelled" && + item.failure.code === ORCHESTRATION_V2_WORKSPACE_PREPARATION_FAILURE_CODE) + ); +} + +/** + * Runs a Retry can prepare again: their workspace preparation failed and the + * run still ended there. Older servers record no preparation on the run, so + * they never offer it. + */ +export function workspacePreparationRetryRunIds( + runs: ReadonlyArray>, + items: ReadonlyArray, +): ReadonlySet { + const failedRuns = new Set( + runs.flatMap((run) => + run.status === "failed" && run.workspacePreparation !== undefined ? [run.id] : [], + ), + ); + const retryable = new Set(); + if (failedRuns.size === 0) return retryable; + for (const item of items) { + if ( + item.type === "error" && + item.status === "failed" && + item.failure.code === ORCHESTRATION_V2_WORKSPACE_PREPARATION_FAILURE_CODE && + item.runId !== null && + failedRuns.has(item.runId) + ) + retryable.add(item.runId); + } + return retryable; } diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 6004ab119917..a2cdc71368eb 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -508,6 +508,29 @@ export const OrchestrationV2RunBackgroundWorkCancelled = Schema.Struct({ export type OrchestrationV2RunBackgroundWorkCancelled = typeof OrchestrationV2RunBackgroundWorkCancelled.Type; +export const OrchestrationV2ThreadLaunchWorkspaceStrategy = Schema.Union([ + Schema.Struct({ + type: Schema.Literal("root"), + branch: Schema.optional(TrimmedNonEmptyString), + }), + Schema.Struct({ + type: Schema.Literal("existing_worktree"), + worktreePath: TrimmedNonEmptyString, + branch: Schema.optional(TrimmedNonEmptyString), + }), + Schema.Struct({ + type: Schema.Literal("worktree"), + baseRef: TrimmedNonEmptyString, + branch: Schema.optional(TrimmedNonEmptyString), + startFromOrigin: Schema.optional(Schema.Boolean), + }), +]); +export type OrchestrationV2ThreadLaunchWorkspaceStrategy = + typeof OrchestrationV2ThreadLaunchWorkspaceStrategy.Type; + +/** Failure code on the error item a failed workspace preparation leaves. */ +export const ORCHESTRATION_V2_WORKSPACE_PREPARATION_FAILURE_CODE = "workspace_preparation_failed"; + export const OrchestrationV2Run = Schema.Struct({ id: RunId, threadId: ThreadId, @@ -549,6 +572,8 @@ export const OrchestrationV2Run = Schema.Struct({ }), ), delegatedCompletion: Schema.optional(OrchestrationV2DelegatedCompletionCohort), + /** How a launch prepares this run's workspace; prepared-run.retry repeats it. */ + workspacePreparation: Schema.optional(OrchestrationV2ThreadLaunchWorkspaceStrategy), }); export type OrchestrationV2Run = typeof OrchestrationV2Run.Type; @@ -2691,7 +2716,10 @@ export const OrchestrationV2Command = Schema.Union([ }), ), dispatchMode: Schema.Union([ - Schema.Struct({ type: Schema.Literal("defer_start") }), + Schema.Struct({ + type: Schema.Literal("defer_start"), + workspaceStrategy: Schema.optional(OrchestrationV2ThreadLaunchWorkspaceStrategy), + }), Schema.Struct({ type: Schema.Literal("steer_active"), targetRunId: RunId }), Schema.Struct({ type: Schema.Literal("restart_active"), targetRunId: RunId }), Schema.Struct({ type: Schema.Literal("queue_after_active") }), @@ -2725,6 +2753,13 @@ export const OrchestrationV2Command = Schema.Union([ runId: RunId, failure: OrchestrationV2ProviderFailure, }), + /** Puts a run whose workspace preparation failed back into preparation. */ + Schema.Struct({ + type: Schema.Literal("prepared-run.retry"), + commandId: CommandId, + threadId: ThreadId, + runId: RunId, + }), Schema.Struct({ type: Schema.Literal("run.interrupt"), commandId: CommandId, @@ -2961,26 +2996,6 @@ export const OrchestrationV2ArchivedShellStreamItem = Schema.Union([ export type OrchestrationV2ArchivedShellStreamItem = typeof OrchestrationV2ArchivedShellStreamItem.Type; -export const OrchestrationV2ThreadLaunchWorkspaceStrategy = Schema.Union([ - Schema.Struct({ - type: Schema.Literal("root"), - branch: Schema.optional(TrimmedNonEmptyString), - }), - Schema.Struct({ - type: Schema.Literal("existing_worktree"), - worktreePath: TrimmedNonEmptyString, - branch: Schema.optional(TrimmedNonEmptyString), - }), - Schema.Struct({ - type: Schema.Literal("worktree"), - baseRef: TrimmedNonEmptyString, - branch: Schema.optional(TrimmedNonEmptyString), - startFromOrigin: Schema.optional(Schema.Boolean), - }), -]); -export type OrchestrationV2ThreadLaunchWorkspaceStrategy = - typeof OrchestrationV2ThreadLaunchWorkspaceStrategy.Type; - export const OrchestrationV2ThreadLaunchInput = Schema.Struct({ commandId: CommandId, creationSource: Schema.optional(OrchestrationV2CreationSource),