diff --git a/apps/server/src/bin.test.ts b/apps/server/src/bin.test.ts index fcb662b9b78..d111921d322 100644 --- a/apps/server/src/bin.test.ts +++ b/apps/server/src/bin.test.ts @@ -31,6 +31,7 @@ import * as ProjectionSnapshotQuery from "./orchestration/Services/ProjectionSna import * as OrchestrationEngine from "./orchestration/Services/OrchestrationEngine.ts"; import { OrchestrationLayerLive } from "./orchestration/runtimeLayer.ts"; import { orchestrationHttpApiLayer } from "./orchestration/http.ts"; +import * as TurnStartBootstrap from "./orchestration/TurnStartBootstrap.ts"; import { layerConfig as SqlitePersistenceLayerLive } from "./persistence/Layers/Sqlite.ts"; import * as RepositoryIdentityResolver from "./project/RepositoryIdentityResolver.ts"; import { @@ -118,6 +119,8 @@ const withLiveProjectCliServer = (baseDir: string, run: () => Effect.Ef const config = yield* makeCliTestServerConfig(baseDir); const routesLayer = HttpApiBuilder.layer(ProjectCliHttpApi).pipe( Layer.provide(orchestrationHttpApiLayer), + // No CLI test here dispatches a bootstrap turn start. + Layer.provide(Layer.mock(TurnStartBootstrap.TurnStartBootstrap)({})), Layer.provide(environmentAuthenticatedAuthLayer), ); const appLayer = HttpRouter.serve(routesLayer, { diff --git a/apps/server/src/orchestration/TurnStartBootstrap.ts b/apps/server/src/orchestration/TurnStartBootstrap.ts new file mode 100644 index 00000000000..a02e0325067 --- /dev/null +++ b/apps/server/src/orchestration/TurnStartBootstrap.ts @@ -0,0 +1,397 @@ +/** + * Runs the bootstrap steps a `thread.turn.start` command can carry + * (`command.bootstrap`): create the thread, prepare its git worktree, launch the + * project setup script, then dispatch the plain turn start. A failure after the + * thread was created rolls it back with `thread.delete`. + * + * Shared by every transport that dispatches client commands (WebSocket RPC and + * the HTTP dispatch route) so both behave identically. + */ +import { + CommandId, + EventId, + type OrchestrationClientOrigin, + type OrchestrationCommand, + OrchestrationDispatchCommandError, + type ThreadId, +} from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; +import * as Context from "effect/Context"; +import * as Crypto from "effect/Crypto"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Schema from "effect/Schema"; + +import * as GitWorkflowService from "../git/GitWorkflowService.ts"; +import * as ProjectSetupScriptRunner from "../project/ProjectSetupScriptRunner.ts"; +import * as VcsStatusBroadcaster from "../vcs/VcsStatusBroadcaster.ts"; +import { OrchestrationEngineService } from "./Services/OrchestrationEngine.ts"; + +export type TurnStartCommand = Extract; + +export interface TurnStartBootstrapDispatchOptions { + readonly origin?: OrchestrationClientOrigin; +} + +export class TurnStartBootstrap extends Context.Service< + TurnStartBootstrap, + { + readonly dispatchTurnStart: ( + command: TurnStartCommand, + options?: TurnStartBootstrapDispatchOptions, + ) => Effect.Effect<{ readonly sequence: number }, OrchestrationDispatchCommandError>; + } +>()("t3/orchestration/TurnStartBootstrap") {} + +const isOrchestrationDispatchCommandError = Schema.is(OrchestrationDispatchCommandError); + +const nowIso = Effect.map(DateTime.now, DateTime.formatIso); + +/** Preserve the setup runner's broader pre-refactor message normalization. */ +function setupFailureDescription(cause: unknown): string { + if ( + typeof cause === "object" && + cause !== null && + "message" in cause && + typeof cause.message === "string" + ) { + return cause.message; + } + return String(cause); +} + +function unexpectedCompatibilityError(error: never): never { + throw new Error(`Unhandled compatibility error: ${String(error)}`); +} + +function projectSetupScriptCompatibilityDetail( + error: ProjectSetupScriptRunner.ProjectSetupScriptRunnerError, +): string { + switch (error._tag) { + case "ProjectSetupScriptOperationError": + return setupFailureDescription(error.cause); + case "ProjectSetupScriptProjectNotFoundError": + return "Project was not found for setup script execution."; + default: + return unexpectedCompatibilityError(error); + } +} + +const toDispatchCommandError = (cause: unknown, fallbackMessage: string) => + isOrchestrationDispatchCommandError(cause) + ? cause + : new OrchestrationDispatchCommandError({ + message: cause instanceof Error ? cause.message : fallbackMessage, + cause, + }); + +const toBootstrapDispatchCommandCauseError = (cause: Cause.Cause) => { + const error = Cause.squash(cause); + return isOrchestrationDispatchCommandError(error) + ? error + : new OrchestrationDispatchCommandError({ + message: error instanceof Error ? error.message : "Failed to bootstrap thread turn start.", + cause, + }); +}; + +export const make = Effect.gen(function* () { + const orchestrationEngine = yield* OrchestrationEngineService; + const gitWorkflow = yield* GitWorkflowService.GitWorkflowService; + const projectSetupScriptRunner = yield* ProjectSetupScriptRunner.ProjectSetupScriptRunner; + const vcsStatusBroadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + const crypto = yield* Crypto.Crypto; + + const randomUUID = crypto.randomUUIDv4.pipe( + Effect.mapError((cause) => + toDispatchCommandError(cause, "Failed to generate orchestration command identifier."), + ), + ); + const serverEventId = randomUUID.pipe(Effect.map(EventId.make)); + const serverCommandId = (tag: string) => + randomUUID.pipe(Effect.map((uuid) => CommandId.make(`server:${tag}:${uuid}`))); + + const refreshGitStatus = (cwd: string) => + vcsStatusBroadcaster + .refreshStatus(cwd) + .pipe(Effect.ignoreCause({ log: true }), Effect.forkDetach, Effect.asVoid); + + const dispatchTurnStart = ( + command: TurnStartCommand, + options?: TurnStartBootstrapDispatchOptions, + ): Effect.Effect<{ readonly sequence: number }, OrchestrationDispatchCommandError> => + Effect.gen(function* () { + // Every sub-command the bootstrap emits carries whatever origin the + // transport supplied: the WebSocket path passes its client origin so the + // sub-commands attribute to the request that caused them; HTTP dispatch + // passes none, as its plain dispatch path never has. + const dispatch = (subCommand: OrchestrationCommand) => + orchestrationEngine.dispatch(subCommand, options); + + const appendSetupScriptActivity = (input: { + readonly threadId: ThreadId; + readonly kind: "setup-script.requested" | "setup-script.started" | "setup-script.failed"; + readonly summary: string; + readonly createdAt: string; + readonly payload: Record; + readonly tone: "info" | "error"; + }) => + Effect.all({ + commandId: serverCommandId("setup-script-activity"), + activityId: serverEventId, + }).pipe( + Effect.flatMap(({ commandId, activityId }) => + dispatch({ + type: "thread.activity.append", + commandId, + threadId: input.threadId, + activity: { + id: activityId, + tone: input.tone, + kind: input.kind, + summary: input.summary, + payload: input.payload, + turnId: null, + createdAt: input.createdAt, + }, + createdAt: input.createdAt, + }), + ), + ); + + const bootstrap = command.bootstrap; + const { bootstrap: _bootstrap, ...finalTurnStartCommand } = command; + let createdThread = false; + let targetProjectId = bootstrap?.createThread?.projectId; + let targetProjectCwd = bootstrap?.prepareWorktree?.projectCwd; + let targetWorktreePath = bootstrap?.createThread?.worktreePath ?? null; + + const cleanupCreatedThread = () => + createdThread + ? serverCommandId("bootstrap-thread-delete").pipe( + Effect.flatMap((commandId) => + dispatch({ + type: "thread.delete", + commandId, + threadId: command.threadId, + }), + ), + Effect.as(true), + ) + : Effect.succeed(false); + + const recordSetupScriptLaunchFailure = (input: { + readonly error: ProjectSetupScriptRunner.ProjectSetupScriptRunnerError; + readonly requestedAt: string; + readonly worktreePath: string; + }) => { + const detail = projectSetupScriptCompatibilityDetail(input.error); + return appendSetupScriptActivity({ + threadId: command.threadId, + kind: "setup-script.failed", + summary: "Setup script failed to start", + createdAt: input.requestedAt, + payload: { + detail, + worktreePath: input.worktreePath, + }, + tone: "error", + }).pipe( + Effect.ignoreCause({ log: false }), + Effect.flatMap(() => + Effect.logWarning("bootstrap turn start failed to launch setup script", { + threadId: command.threadId, + worktreePath: input.worktreePath, + detail, + }), + ), + ); + }; + + const recordSetupScriptStarted = (input: { + readonly requestedAt: string; + readonly worktreePath: string; + readonly scriptId: string; + readonly scriptName: string; + readonly terminalId: string; + }) => + Effect.gen(function* () { + const startedAt = yield* nowIso; + const payload = { + scriptId: input.scriptId, + scriptName: input.scriptName, + terminalId: input.terminalId, + worktreePath: input.worktreePath, + }; + yield* Effect.all([ + appendSetupScriptActivity({ + threadId: command.threadId, + kind: "setup-script.requested", + summary: "Starting setup script", + createdAt: input.requestedAt, + payload, + tone: "info", + }), + appendSetupScriptActivity({ + threadId: command.threadId, + kind: "setup-script.started", + summary: "Setup script started", + createdAt: startedAt, + payload, + tone: "info", + }), + ]).pipe( + Effect.asVoid, + Effect.catch((error) => + Effect.logWarning( + "bootstrap turn start launched setup script but failed to record setup activity", + { + threadId: command.threadId, + worktreePath: input.worktreePath, + scriptId: input.scriptId, + terminalId: input.terminalId, + detail: error.message, + }, + ), + ), + ); + }); + + const runSetupProgram = () => + Effect.gen(function* () { + if (!bootstrap?.runSetupScript || !targetWorktreePath) { + return; + } + const worktreePath = targetWorktreePath; + const requestedAt = yield* nowIso; + yield* projectSetupScriptRunner + .runForThread({ + threadId: command.threadId, + ...(targetProjectId ? { projectId: targetProjectId } : {}), + ...(targetProjectCwd ? { projectCwd: targetProjectCwd } : {}), + worktreePath, + }) + .pipe( + Effect.matchEffect({ + onFailure: (error) => + recordSetupScriptLaunchFailure({ + error, + requestedAt, + worktreePath, + }), + onSuccess: (setupResult) => { + if (setupResult.status !== "started") { + return Effect.void; + } + return recordSetupScriptStarted({ + requestedAt, + worktreePath, + scriptId: setupResult.scriptId, + scriptName: setupResult.scriptName, + terminalId: setupResult.terminalId, + }); + }, + }), + ); + }); + + const bootstrapProgram = Effect.gen(function* () { + if (bootstrap?.createThread) { + yield* dispatch({ + type: "thread.create", + commandId: yield* serverCommandId("bootstrap-thread-create"), + threadId: command.threadId, + projectId: bootstrap.createThread.projectId, + title: bootstrap.createThread.title, + modelSelection: bootstrap.createThread.modelSelection, + runtimeMode: bootstrap.createThread.runtimeMode, + interactionMode: bootstrap.createThread.interactionMode, + branch: bootstrap.createThread.branch, + worktreePath: bootstrap.createThread.worktreePath, + createdAt: bootstrap.createThread.createdAt, + }); + createdThread = true; + } + + if (bootstrap?.prepareWorktree) { + let worktreeBaseRef = bootstrap.prepareWorktree.baseBranch; + // "Start from origin" is a stored default; repos without an + // origin remote 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 resolvedRemoteBase = yield* gitWorkflow.resolveRemoteTrackingCommit({ + cwd: bootstrap.prepareWorktree.projectCwd, + refName: bootstrap.prepareWorktree.baseBranch, + fallbackRemoteName: "origin", + }); + worktreeBaseRef = resolvedRemoteBase.commitSha; + } + const worktree = yield* gitWorkflow.createWorktree({ + cwd: bootstrap.prepareWorktree.projectCwd, + refName: worktreeBaseRef, + newRefName: bootstrap.prepareWorktree.branch, + baseRefName: bootstrap.prepareWorktree.baseBranch, + path: null, + }); + targetWorktreePath = worktree.worktree.path; + yield* dispatch({ + type: "thread.meta.update", + commandId: yield* serverCommandId("bootstrap-thread-meta-update"), + threadId: command.threadId, + branch: worktree.worktree.refName, + worktreePath: targetWorktreePath, + }); + yield* refreshGitStatus(targetWorktreePath); + } + + yield* runSetupProgram(); + + return yield* dispatch(finalTurnStartCommand); + }); + + return yield* bootstrapProgram.pipe( + Effect.catchCause((cause) => { + const dispatchError = toBootstrapDispatchCommandCauseError(cause); + if (Cause.hasInterruptsOnly(cause)) { + return Effect.fail(dispatchError); + } + 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, + ), + }), + ); + }), + ); + }); + + return TurnStartBootstrap.of({ dispatchTurnStart }); +}); + +export const layer = Layer.effect(TurnStartBootstrap, make); diff --git a/apps/server/src/orchestration/http.ts b/apps/server/src/orchestration/http.ts index 04d54ea8eff..bf4cff22b93 100644 --- a/apps/server/src/orchestration/http.ts +++ b/apps/server/src/orchestration/http.ts @@ -18,6 +18,7 @@ import { } from "../auth/http.ts"; import { OrchestrationEngineService } from "./Services/OrchestrationEngine.ts"; import { ProjectionSnapshotQuery } from "./Services/ProjectionSnapshotQuery.ts"; +import { TurnStartBootstrap } from "./TurnStartBootstrap.ts"; export const orchestrationHttpApiLayer = HttpApiBuilder.group( EnvironmentHttpApi, @@ -25,6 +26,7 @@ export const orchestrationHttpApiLayer = HttpApiBuilder.group( Effect.fnUntraced(function* (handlers) { const projectionSnapshotQuery = yield* ProjectionSnapshotQuery; const orchestrationEngine = yield* OrchestrationEngineService; + const turnStartBootstrap = yield* TurnStartBootstrap; return handlers .handle( @@ -96,13 +98,27 @@ export const orchestrationHttpApiLayer = HttpApiBuilder.group( const normalizedCommand = yield* normalizeDispatchCommand(args.payload).pipe( Effect.catch(() => failEnvironmentInvalidRequest("invalid_command")), ); - return yield* orchestrationEngine - .dispatch(normalizedCommand) - .pipe( + // Same bootstrap handling as the WebSocket dispatch path: a + // turn.start carrying `bootstrap` creates the thread (and worktree) + // before the turn itself runs. + const toDispatchFailure = (cause: unknown) => + failEnvironmentInternal("orchestration_dispatch_failed", cause); + if (normalizedCommand.type === "thread.turn.start" && normalizedCommand.bootstrap) { + return yield* turnStartBootstrap.dispatchTurnStart(normalizedCommand).pipe( + // The WebSocket path hands clients the error's + // bootstrapThreadDisposition; over HTTP that signal survives as + // its own reason, so a caller knows the thread is gone and a + // retry needs a fresh id. Effect.catch((cause) => - failEnvironmentInternal("orchestration_dispatch_failed", cause), + cause.bootstrapThreadDisposition === "deleted" + ? failEnvironmentInternal("orchestration_bootstrap_rolled_back", cause) + : toDispatchFailure(cause), ), ); + } + return yield* orchestrationEngine + .dispatch(normalizedCommand) + .pipe(Effect.catch(toDispatchFailure)); }), ); }), diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index 02a367c0879..f0d865b42cf 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -8147,6 +8147,144 @@ it.layer(NodeServices.layer)("server router seam", (it) => { }).pipe(Effect.provide(NodeHttpServer.layerTest)), ); + const httpBootstrapTurnStartCommand = (suffix: string) => { + const createdAt = "2026-01-01T00:00:00.000Z"; + return { + type: "thread.turn.start", + commandId: `cmd-http-bootstrap-${suffix}`, + threadId: `thread-http-bootstrap-${suffix}`, + message: { + messageId: `msg-http-bootstrap-${suffix}`, + 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: null, + createdAt, + }, + prepareWorktree: { + projectCwd: "/tmp/project", + baseBranch: "main", + branch: "t3code/bootstrap-refName", + }, + runSetupScript: false, + }, + createdAt, + }; + }; + + const postHttpDispatch = (command: unknown) => + Effect.gen(function* () { + const cookie = yield* getAuthenticatedSessionCookieHeader(); + const url = yield* getHttpServerUrl("/api/orchestration/dispatch"); + return yield* fetchEffect(url, { + method: "POST", + headers: { cookie, "content-type": "application/json" }, + body: jsonRequestBody(command), + }); + }); + + it.effect("http dispatch bootstraps a turn start like the websocket path", () => + Effect.gen(function* () { + const dispatchedCommands: Array = []; + const createWorktree = vi.fn( + (_: Parameters[0]) => + Effect.succeed({ + worktree: { + refName: "t3code/bootstrap-refName", + path: "/tmp/bootstrap-worktree", + }, + }), + ); + + yield* buildAppUnderTest({ + layers: { + gitVcsDriver: { + createWorktree, + }, + orchestrationEngine: { + dispatch: (command) => + Effect.sync(() => { + dispatchedCommands.push(command); + return { sequence: dispatchedCommands.length }; + }), + readEvents: () => Stream.empty, + }, + }, + }); + + const response = yield* postHttpDispatch(httpBootstrapTurnStartCommand("ok")); + assert.strictEqual(response.status, 200); + const body = yield* responseJsonEffect<{ readonly sequence: number }>(response); + assert.strictEqual(body.sequence, 3); + + assert.strictEqual(createWorktree.mock.calls.length, 1); + assert.deepEqual( + dispatchedCommands.map((command) => command.type), + ["thread.create", "thread.meta.update", "thread.turn.start"], + ); + const turnStart = dispatchedCommands[2]; + assertTrue(turnStart?.type === "thread.turn.start"); + assert.strictEqual(turnStart.threadId, "thread-http-bootstrap-ok"); + assert.isUndefined(turnStart.bootstrap); + const metaUpdate = dispatchedCommands[1]; + assertTrue(metaUpdate?.type === "thread.meta.update"); + assert.strictEqual(metaUpdate.worktreePath, "/tmp/bootstrap-worktree"); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + + it.effect("http dispatch rolls back the created thread when bootstrap fails", () => + Effect.gen(function* () { + const dispatchedCommands: Array = []; + const createWorktree = vi.fn( + (_: Parameters[0]) => + Effect.die(new Error("worktree exploded")), + ); + + yield* buildAppUnderTest({ + layers: { + gitVcsDriver: { + createWorktree, + }, + orchestrationEngine: { + dispatch: (command) => + Effect.sync(() => { + dispatchedCommands.push(command); + return { sequence: dispatchedCommands.length }; + }), + readEvents: () => Stream.empty, + }, + }, + }); + + const response = yield* postHttpDispatch(httpBootstrapTurnStartCommand("defect")); + assert.strictEqual(response.status, 500); + const body = yield* responseJsonEffect<{ readonly code: string; readonly reason: string }>( + response, + ); + assert.strictEqual(body.code, "internal_error"); + // The rollback is visible to the HTTP caller, matching the WebSocket + // error's bootstrapThreadDisposition: the thread id must not be reused. + assert.strictEqual(body.reason, "orchestration_bootstrap_rolled_back"); + + assert.deepEqual( + dispatchedCommands.map((command) => command.type), + ["thread.create", "thread.delete"], + ); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + it.effect("routes websocket rpc terminal methods", () => Effect.gen(function* () { const snapshot = { diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 3e41b4390f8..955f2eef481 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -115,6 +115,7 @@ import { persistServerRuntimeState, } from "./serverRuntimeState.ts"; import { orchestrationHttpApiLayer } from "./orchestration/http.ts"; +import * as TurnStartBootstrap from "./orchestration/TurnStartBootstrap.ts"; import * as NetService from "@t3tools/shared/Net"; import * as RelayClient from "@t3tools/shared/relayClient"; import { disableTailscaleServe, ensureTailscaleServe } from "@t3tools/tailscale"; @@ -464,6 +465,9 @@ export const makeRoutesLayer = Layer.mergeAll( // Both transports consume the same service instance, so caches single-flight across clients // and mutations observed on WebSocket invalidate patches subsequently read over HTTP. Layer.provide(PullRequestServiceLive), + // Shared by the WebSocket and HTTP dispatch routes so both bootstrap + // thread.turn.start identically. + Layer.provide(TurnStartBootstrap.layer), Layer.provide(PreviewAutomationBroker.layer), Layer.provide(ServerSelfUpdate.layer), Layer.provide(commandReadinessLayer), diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 11c659e28a7..314f28cc710 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -1,5 +1,3 @@ -import * as Cause from "effect/Cause"; -import * as Crypto from "effect/Crypto"; import * as DateTime from "effect/DateTime"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; @@ -18,7 +16,6 @@ import { ClientSurface, CommandId, type DiscoveredLocalServerList, - EventId, type OrchestrationClientOrigin, type OrchestrationCommand, type GitActionProgressEvent, @@ -100,8 +97,8 @@ import * as WorkspacePaths from "./workspace/WorkspacePaths.ts"; import * as VcsStatusBroadcaster from "./vcs/VcsStatusBroadcaster.ts"; import * as VcsProvisioningService from "./vcs/VcsProvisioningService.ts"; import * as GitWorkflowService from "./git/GitWorkflowService.ts"; +import * as TurnStartBootstrap from "./orchestration/TurnStartBootstrap.ts"; import * as ReviewService from "./review/ReviewService.ts"; -import * as ProjectSetupScriptRunner from "./project/ProjectSetupScriptRunner.ts"; import * as ServerEnvironment from "./environment/ServerEnvironment.ts"; import * as RemoteOpenTargets from "./environment/RemoteOpenTargets.ts"; import * as BackgroundPolicy from "./background/BackgroundPolicy.ts"; @@ -146,19 +143,6 @@ function unexpectedCompatibilityError(error: never): never { throw new Error(`Unhandled compatibility error: ${String(error)}`); } -/** Preserve the setup runner's broader pre-refactor message normalization. */ -function legacySetupFailureDescription(cause: unknown): string { - if ( - typeof cause === "object" && - cause !== null && - "message" in cause && - typeof cause.message === "string" - ) { - return cause.message; - } - return String(cause); -} - function projectEntriesFailureContext(error: WorkspaceEntries.WorkspaceEntriesError): { readonly failure: ProjectEntriesFailure; readonly normalizedCwd?: string; @@ -263,19 +247,6 @@ function projectFileFailureContext( } } -function projectSetupScriptCompatibilityDetail( - error: ProjectSetupScriptRunner.ProjectSetupScriptRunnerError, -): string { - switch (error._tag) { - case "ProjectSetupScriptOperationError": - return legacySetupFailureDescription(error.cause); - case "ProjectSetupScriptProjectNotFoundError": - return "Project was not found for setup script execution."; - default: - return unexpectedCompatibilityError(error); - } -} - export function isThreadDetailEvent(event: OrchestrationEvent): event is Extract< OrchestrationEvent, { @@ -390,7 +361,6 @@ const makeWsRpcLayer = ( WsRpcGroup.toLayer( Effect.gen(function* () { const currentSessionId = currentSession.sessionId; - const crypto = yield* Crypto.Crypto; const projectionSnapshotQuery = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery; const orchestrationEngine = yield* OrchestrationEngine.OrchestrationEngineService; const analytics = yield* AnalyticsService.AnalyticsService; @@ -443,7 +413,7 @@ const makeWsRpcLayer = ( const startup = yield* ServerRuntimeStartup.ServerRuntimeStartup; const workspaceEntries = yield* WorkspaceEntries.WorkspaceEntries; const workspaceFileSystem = yield* WorkspaceFileSystem.WorkspaceFileSystem; - const projectSetupScriptRunner = yield* ProjectSetupScriptRunner.ProjectSetupScriptRunner; + const turnStartBootstrap = yield* TurnStartBootstrap.TurnStartBootstrap; const serverEnvironment = yield* ServerEnvironment.ServerEnvironment; const backgroundPolicy = yield* BackgroundPolicy.BackgroundPolicy; const rpcClientIds = yield* Ref.make(new Set()); @@ -543,14 +513,6 @@ const makeWsRpcLayer = ( message: cause instanceof Error ? cause.message : fallbackMessage, cause, }); - const randomUUID = crypto.randomUUIDv4.pipe( - Effect.mapError((cause) => - toDispatchCommandError(cause, "Failed to generate orchestration command identifier."), - ), - ); - const serverEventId = randomUUID.pipe(Effect.map(EventId.make)); - const serverCommandId = (tag: string) => - randomUUID.pipe(Effect.map((uuid) => CommandId.make(`server:${tag}:${uuid}`))); const loadAuthAccessSnapshot = () => Effect.all({ @@ -565,48 +527,6 @@ const makeWsRpcLayer = ( ), ); - const appendSetupScriptActivity = (input: { - readonly threadId: ThreadId; - readonly kind: "setup-script.requested" | "setup-script.started" | "setup-script.failed"; - readonly summary: string; - readonly createdAt: string; - readonly payload: Record; - readonly tone: "info" | "error"; - }) => - Effect.all({ - commandId: serverCommandId("setup-script-activity"), - activityId: serverEventId, - }).pipe( - Effect.flatMap(({ commandId, activityId }) => - dispatchFromClient({ - type: "thread.activity.append", - commandId, - threadId: input.threadId, - activity: { - id: activityId, - tone: input.tone, - kind: input.kind, - summary: input.summary, - payload: input.payload, - turnId: null, - createdAt: input.createdAt, - }, - createdAt: input.createdAt, - }), - ), - ); - - const toBootstrapDispatchCommandCauseError = (cause: Cause.Cause) => { - const error = Cause.squash(cause); - return isOrchestrationDispatchCommandError(error) - ? error - : new OrchestrationDispatchCommandError({ - message: - error instanceof Error ? error.message : "Failed to bootstrap thread turn start.", - cause, - }); - }; - const toShellStreamEvent = ( event: OrchestrationEvent, ): Effect.Effect, never, never> => { @@ -818,247 +738,15 @@ const makeWsRpcLayer = ( Stream.flatMap((items) => Stream.fromIterable(items)), ); - const dispatchBootstrapTurnStart = ( - command: Extract, - ): Effect.Effect<{ readonly sequence: number }, OrchestrationDispatchCommandError> => - Effect.gen(function* () { - const bootstrap = command.bootstrap; - const { bootstrap: _bootstrap, ...finalTurnStartCommand } = command; - let createdThread = false; - let targetProjectId = bootstrap?.createThread?.projectId; - let targetProjectCwd = bootstrap?.prepareWorktree?.projectCwd; - let targetWorktreePath = bootstrap?.createThread?.worktreePath ?? null; - - const cleanupCreatedThread = () => - createdThread - ? serverCommandId("bootstrap-thread-delete").pipe( - Effect.flatMap((commandId) => - dispatchFromClient({ - type: "thread.delete", - commandId, - threadId: command.threadId, - }), - ), - Effect.as(true), - ) - : Effect.succeed(false); - - const recordSetupScriptLaunchFailure = (input: { - readonly error: ProjectSetupScriptRunner.ProjectSetupScriptRunnerError; - readonly requestedAt: string; - readonly worktreePath: string; - }) => { - const detail = projectSetupScriptCompatibilityDetail(input.error); - return appendSetupScriptActivity({ - threadId: command.threadId, - kind: "setup-script.failed", - summary: "Setup script failed to start", - createdAt: input.requestedAt, - payload: { - detail, - worktreePath: input.worktreePath, - }, - tone: "error", - }).pipe( - Effect.ignoreCause({ log: false }), - Effect.flatMap(() => - Effect.logWarning("bootstrap turn start failed to launch setup script", { - threadId: command.threadId, - worktreePath: input.worktreePath, - detail, - }), - ), - ); - }; - - const recordSetupScriptStarted = (input: { - readonly requestedAt: string; - readonly worktreePath: string; - readonly scriptId: string; - readonly scriptName: string; - readonly terminalId: string; - }) => - Effect.gen(function* () { - const startedAt = yield* nowIso; - const payload = { - scriptId: input.scriptId, - scriptName: input.scriptName, - terminalId: input.terminalId, - worktreePath: input.worktreePath, - }; - yield* Effect.all([ - appendSetupScriptActivity({ - threadId: command.threadId, - kind: "setup-script.requested", - summary: "Starting setup script", - createdAt: input.requestedAt, - payload, - tone: "info", - }), - appendSetupScriptActivity({ - threadId: command.threadId, - kind: "setup-script.started", - summary: "Setup script started", - createdAt: startedAt, - payload, - tone: "info", - }), - ]).pipe( - Effect.asVoid, - Effect.catch((error) => - Effect.logWarning( - "bootstrap turn start launched setup script but failed to record setup activity", - { - threadId: command.threadId, - worktreePath: input.worktreePath, - scriptId: input.scriptId, - terminalId: input.terminalId, - detail: error.message, - }, - ), - ), - ); - }); - - const runSetupProgram = () => - Effect.gen(function* () { - if (!bootstrap?.runSetupScript || !targetWorktreePath) { - return; - } - const worktreePath = targetWorktreePath; - const requestedAt = yield* nowIso; - yield* projectSetupScriptRunner - .runForThread({ - threadId: command.threadId, - ...(targetProjectId ? { projectId: targetProjectId } : {}), - ...(targetProjectCwd ? { projectCwd: targetProjectCwd } : {}), - worktreePath, - }) - .pipe( - Effect.matchEffect({ - onFailure: (error) => - recordSetupScriptLaunchFailure({ - error, - requestedAt, - worktreePath, - }), - onSuccess: (setupResult) => { - if (setupResult.status !== "started") { - return Effect.void; - } - return recordSetupScriptStarted({ - requestedAt, - worktreePath, - scriptId: setupResult.scriptId, - scriptName: setupResult.scriptName, - terminalId: setupResult.terminalId, - }); - }, - }), - ); - }); - - const bootstrapProgram = Effect.gen(function* () { - if (bootstrap?.createThread) { - yield* dispatchFromClient({ - type: "thread.create", - commandId: yield* serverCommandId("bootstrap-thread-create"), - threadId: command.threadId, - projectId: bootstrap.createThread.projectId, - title: bootstrap.createThread.title, - modelSelection: bootstrap.createThread.modelSelection, - runtimeMode: bootstrap.createThread.runtimeMode, - interactionMode: bootstrap.createThread.interactionMode, - branch: bootstrap.createThread.branch, - worktreePath: bootstrap.createThread.worktreePath, - createdAt: bootstrap.createThread.createdAt, - }); - createdThread = true; - } - - if (bootstrap?.prepareWorktree) { - let worktreeBaseRef = bootstrap.prepareWorktree.baseBranch; - // "Start from origin" is a stored default; repos without an - // origin remote 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 resolvedRemoteBase = yield* gitWorkflow.resolveRemoteTrackingCommit({ - cwd: bootstrap.prepareWorktree.projectCwd, - refName: bootstrap.prepareWorktree.baseBranch, - fallbackRemoteName: "origin", - }); - worktreeBaseRef = resolvedRemoteBase.commitSha; - } - const worktree = yield* gitWorkflow.createWorktree({ - cwd: bootstrap.prepareWorktree.projectCwd, - refName: worktreeBaseRef, - newRefName: bootstrap.prepareWorktree.branch, - baseRefName: bootstrap.prepareWorktree.baseBranch, - path: null, - }); - targetWorktreePath = worktree.worktree.path; - yield* dispatchFromClient({ - type: "thread.meta.update", - commandId: yield* serverCommandId("bootstrap-thread-meta-update"), - threadId: command.threadId, - branch: worktree.worktree.refName, - worktreePath: targetWorktreePath, - }); - yield* refreshGitStatus(targetWorktreePath); - } - - yield* runSetupProgram(); - - return yield* dispatchFromClient(finalTurnStartCommand); - }); - - return yield* bootstrapProgram.pipe( - Effect.catchCause((cause) => { - const dispatchError = toBootstrapDispatchCommandCauseError(cause); - if (Cause.hasInterruptsOnly(cause)) { - return Effect.fail(dispatchError); - } - 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, - ), - }), - ); - }), - ); - }); - const dispatchNormalizedCommand = ( normalizedCommand: OrchestrationCommand, ): Effect.Effect<{ readonly sequence: number }, OrchestrationDispatchCommandError> => { const dispatchEffect = normalizedCommand.type === "thread.turn.start" && normalizedCommand.bootstrap - ? dispatchBootstrapTurnStart(normalizedCommand) + ? turnStartBootstrap.dispatchTurnStart( + normalizedCommand, + hasClientOrigin ? { origin: clientOrigin } : undefined, + ) : dispatchFromClient(normalizedCommand).pipe( Effect.mapError((cause) => toDispatchCommandError(cause, "Failed to dispatch orchestration command"), @@ -2400,6 +2088,7 @@ export const websocketRpcRouteLayer = Layer.unwrap( const previewAutomationBroker = yield* PreviewAutomationBroker.PreviewAutomationBroker; const serverSelfUpdate = yield* ServerSelfUpdate.ServerSelfUpdate; const pullRequests = yield* PullRequestService.PullRequestService; + const turnStartBootstrap = yield* TurnStartBootstrap.TurnStartBootstrap; return HttpRouter.add( "GET", "/ws", @@ -2430,6 +2119,9 @@ export const websocketRpcRouteLayer = Layer.unwrap( // One server-lifetime service means clients share the same PR caches, and a WS // mutation invalidates the HTTP diff cache that every client reads from. Layer.provide(Layer.succeed(PullRequestService.PullRequestService, pullRequests)), + Layer.provide( + Layer.succeed(TurnStartBootstrap.TurnStartBootstrap, turnStartBootstrap), + ), Layer.provide( SourceControlDiscovery.layer.pipe( Layer.provide( diff --git a/packages/contracts/src/environmentHttp.ts b/packages/contracts/src/environmentHttp.ts index e7494862251..8de9a2f973a 100644 --- a/packages/contracts/src/environmentHttp.ts +++ b/packages/contracts/src/environmentHttp.ts @@ -90,6 +90,10 @@ export const EnvironmentInternalErrorReason = Schema.Literals([ "orchestration_snapshot_failed", "orchestration_thread_snapshot_failed", "orchestration_dispatch_failed", + // A bootstrap turn start failed after creating its thread and rolled the + // thread back: the HTTP equivalent of the WebSocket error's + // bootstrapThreadDisposition "deleted", so callers know to retry fresh. + "orchestration_bootstrap_rolled_back", "internal_error", ]); export type EnvironmentInternalErrorReason = typeof EnvironmentInternalErrorReason.Type;