diff --git a/apps/server/src/orchestration/Layers/BootstrapTurnStartRuns.ts b/apps/server/src/orchestration/Layers/BootstrapTurnStartRuns.ts new file mode 100644 index 000000000..863b12070 --- /dev/null +++ b/apps/server/src/orchestration/Layers/BootstrapTurnStartRuns.ts @@ -0,0 +1,89 @@ +/** + * BootstrapTurnStartRuns - one bootstrap turn start per command id. + * + * A bootstrap turn start (create the thread, cut a worktree, launch the setup + * script, start the turn) is several engine dispatches under one client + * command id, so the engine's per-command receipts cannot make it idempotent + * on their own. The client re-sends a command whose socket dropped or whose + * response was slow, and a second run would fail on `thread.create` after the + * first run already created the thread. + * + * This service runs each command id once. The run is forked into the + * service's own scope so a dropped socket does not abort a half-done + * bootstrap, and a retry that arrives while it is still running joins that + * run's result. Retries that arrive after the run finished are answered from + * the command receipt by the caller. + * + * @module BootstrapTurnStartRuns + */ +import type { CommandId, OrchestrationDispatchCommandError } from "@threadlines/contracts"; +import * as Context from "effect/Context"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as Layer from "effect/Layer"; + +export interface BootstrapTurnStartResult { + readonly sequence: number; +} + +type BootstrapTurnStartRun = Effect.Effect< + BootstrapTurnStartResult, + OrchestrationDispatchCommandError +>; + +export interface BootstrapTurnStartRunsShape { + /** Run `bootstrap` for `commandId`, or join the run already in flight for it. */ + readonly run: (commandId: CommandId, bootstrap: BootstrapTurnStartRun) => BootstrapTurnStartRun; +} + +export class BootstrapTurnStartRuns extends Context.Service< + BootstrapTurnStartRuns, + BootstrapTurnStartRunsShape +>()("threadlines/orchestration/BootstrapTurnStartRuns") {} + +export const makeBootstrapTurnStartRuns = Effect.gen(function* () { + const scope = yield* Effect.scope; + const inflight = new Map< + CommandId, + Deferred.Deferred + >(); + + const run: BootstrapTurnStartRunsShape["run"] = (commandId, bootstrap) => + Effect.gen(function* () { + const existing = inflight.get(commandId); + if (existing) { + return yield* Deferred.await(existing); + } + const result = yield* Deferred.make< + BootstrapTurnStartResult, + OrchestrationDispatchCommandError + >(); + inflight.set(commandId, result); + yield* bootstrap.pipe( + Effect.exit, + Effect.flatMap((exit) => + Exit.isSuccess(exit) + ? Deferred.succeed(result, exit.value) + : Deferred.failCause(result, exit.cause), + ), + // Dropped only after the result is settled (and, on success, after the + // engine wrote the receipt), so a retry never finds neither a run in + // flight nor a receipt. + Effect.ensuring( + Effect.sync(() => { + inflight.delete(commandId); + }), + ), + Effect.forkIn(scope), + ); + return yield* Deferred.await(result); + }); + + return { run } satisfies BootstrapTurnStartRunsShape; +}); + +export const BootstrapTurnStartRunsLive = Layer.effect( + BootstrapTurnStartRuns, + makeBootstrapTurnStartRuns, +); diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.ts index abfc231c0..c64fcb46b 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationEngine.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.ts @@ -307,9 +307,13 @@ const makeOrchestrationEngine = Effect.gen(function* () { return yield* Deferred.await(result); }); + const getCommandReceipt: OrchestrationEngineShape["getCommandReceipt"] = (commandId) => + commandReceiptRepository.getByCommandId({ commandId }); + return { readEvents, dispatch, + getCommandReceipt, // Each access creates a fresh PubSub subscription so that multiple // consumers (wsServer, ProviderRuntimeIngestion, CheckpointReactor, etc.) // each independently receive all domain events. diff --git a/apps/server/src/orchestration/Layers/ThreadAutoArchiveSweeper.test.ts b/apps/server/src/orchestration/Layers/ThreadAutoArchiveSweeper.test.ts index 4ffcac412..7dd8c2dc7 100644 --- a/apps/server/src/orchestration/Layers/ThreadAutoArchiveSweeper.test.ts +++ b/apps/server/src/orchestration/Layers/ThreadAutoArchiveSweeper.test.ts @@ -150,6 +150,7 @@ describe("ThreadAutoArchiveSweeper", () => { }); const orchestrationEngine: OrchestrationEngineShape = { readEvents: () => Stream.empty, + getCommandReceipt: () => Effect.succeed(Option.none()), dispatch: (command) => { if (command.type !== "thread.archive") { return Effect.die(`Unexpected command: ${command.type}`); diff --git a/apps/server/src/orchestration/Layers/ThreadDiffStatBaselineReactor.test.ts b/apps/server/src/orchestration/Layers/ThreadDiffStatBaselineReactor.test.ts index 371310794..4ca51a9bc 100644 --- a/apps/server/src/orchestration/Layers/ThreadDiffStatBaselineReactor.test.ts +++ b/apps/server/src/orchestration/Layers/ThreadDiffStatBaselineReactor.test.ts @@ -115,6 +115,7 @@ describe("ThreadDiffStatBaselineReactor", () => { const orchestrationEngine: OrchestrationEngineShape = { readEvents: () => Stream.empty, + getCommandReceipt: () => Effect.succeed(Option.none()), dispatch: (command) => { if (command.type !== "thread.diffstat.rebase") { return Effect.die(`Unexpected command: ${command.type}`); diff --git a/apps/server/src/orchestration/Services/OrchestrationEngine.ts b/apps/server/src/orchestration/Services/OrchestrationEngine.ts index 9f95fedab..c3eeb9bf2 100644 --- a/apps/server/src/orchestration/Services/OrchestrationEngine.ts +++ b/apps/server/src/orchestration/Services/OrchestrationEngine.ts @@ -10,14 +10,19 @@ * * @module OrchestrationEngineService */ -import type { OrchestrationCommand, OrchestrationEvent } from "@threadlines/contracts"; +import type { CommandId, OrchestrationCommand, OrchestrationEvent } from "@threadlines/contracts"; import * as Context from "effect/Context"; import type * as Effect from "effect/Effect"; +import type * as Option from "effect/Option"; import type * as Scope from "effect/Scope"; import type * as Stream from "effect/Stream"; import type { OrchestrationDispatchError } from "../Errors.ts"; -import type { OrchestrationEventStoreError } from "../../persistence/Errors.ts"; +import type { + OrchestrationCommandReceiptRepositoryError, + OrchestrationEventStoreError, +} from "../../persistence/Errors.ts"; +import type { OrchestrationCommandReceipt } from "../../persistence/Services/OrchestrationCommandReceipts.ts"; /** * OrchestrationEngineShape - Service API for orchestration command and event flow. @@ -46,6 +51,21 @@ export interface OrchestrationEngineShape { command: OrchestrationCommand, ) => Effect.Effect<{ sequence: number }, OrchestrationDispatchError, never>; + /** + * Read the receipt an earlier dispatch left for `commandId`, if any. + * + * Lets a caller that runs several dispatches under one client command + * (a bootstrap turn start) answer a client retry the way `dispatch` would: + * with the accepted sequence, or the original rejection. + */ + readonly getCommandReceipt: ( + commandId: CommandId, + ) => Effect.Effect< + Option.Option, + OrchestrationCommandReceiptRepositoryError, + never + >; + /** * Stream persisted domain events in dispatch order. * diff --git a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts index decc9ef5e..08c740d10 100644 --- a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts +++ b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts @@ -291,6 +291,7 @@ describe("ProviderSessionReaper", () => { Layer.succeed(OrchestrationEngineService, { readEvents: () => Stream.empty, dispatch, + getCommandReceipt: () => Effect.succeed(Option.none()), streamDomainEvents: Stream.empty, subscribeDomainEvents: Effect.succeed(Stream.empty), }), diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index d0436685f..60db81eb5 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -802,6 +802,7 @@ const buildAppUnderTest = (options?: { Layer.mock(OrchestrationEngineService)({ readEvents: () => Stream.empty, dispatch: () => Effect.succeed({ sequence: 0 }), + getCommandReceipt: () => Effect.succeed(Option.none()), streamDomainEvents: Stream.empty, subscribeDomainEvents: Effect.succeed(Stream.empty), ...options?.layers?.orchestrationEngine, @@ -4661,6 +4662,133 @@ it.layer(NodeServices.layer)("server router seam", (it) => { }).pipe(Effect.provide(NodeHttpServer.layerTest)), ); + it.effect( + "a retried bootstrap turn start joins the run in flight and later retries read the receipt", + () => + Effect.gen(function* () { + const dispatchedCommands: Array = []; + const receiptSequences = new Map(); + const worktreeRequested = yield* Deferred.make(); + const worktreeGate = yield* Deferred.make(); + const createWorktree = vi.fn( + (_: Parameters[0]) => + Deferred.succeed(worktreeRequested, undefined).pipe( + Effect.andThen(Deferred.await(worktreeGate)), + Effect.as({ + worktree: { + refName: "threadlines/bootstrap-refName", + path: "/tmp/bootstrap-worktree", + }, + }), + ), + ); + const createdAt = "2026-01-01T00:00:00.000Z"; + const threadId = ThreadId.make("thread-bootstrap-retry"); + + yield* buildAppUnderTest({ + layers: { + gitVcsDriver: { + createWorktree, + resolveFreshWorktreeBase: (input) => + Effect.succeed({ refName: input.branch, isRemote: false }), + }, + orchestrationEngine: { + // Like the real engine, every dispatch leaves a receipt under + // its command id, so the final turn start receipts the client's. + dispatch: (command) => + Effect.sync(() => { + dispatchedCommands.push(command); + receiptSequences.set(command.commandId, dispatchedCommands.length); + return { sequence: dispatchedCommands.length }; + }), + getCommandReceipt: (commandId) => + Effect.sync(() => { + const resultSequence = receiptSequences.get(commandId); + return resultSequence === undefined + ? Option.none() + : Option.some({ + commandId, + aggregateKind: "thread" as const, + aggregateId: threadId, + acceptedAt: createdAt, + resultSequence, + status: "accepted" as const, + error: null, + }); + }), + readEvents: () => Stream.empty, + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const dispatch = () => + Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.dispatchCommand]({ + type: "thread.turn.start", + commandId: CommandId.make("cmd-bootstrap-turn-start-retry"), + threadId, + message: { + messageId: MessageId.make("msg-bootstrap-retry"), + 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: "threadlines/bootstrap-refName", + }, + }, + createdAt, + }), + ), + ); + + const first = yield* Effect.forkChild(dispatch()); + yield* Deferred.await(worktreeRequested); + // The client re-sends after a reconnect, so the retry lands on a new + // socket while the first run is still cutting the worktree. + const retry = yield* Effect.forkChild(dispatch()); + yield* wallClockSleep(200); + yield* Deferred.succeed(worktreeGate, undefined); + const [firstResponse, retryResponse] = yield* Effect.all([ + Fiber.join(first), + Fiber.join(retry), + ]); + + assert.equal(firstResponse.sequence, 3); + assert.equal(retryResponse.sequence, 3); + assert.deepEqual( + dispatchedCommands.map((command) => command.type), + ["thread.create", "thread.meta.update", "thread.turn.start"], + ); + assert.equal(createWorktree.mock.calls.length, 1); + + // A retry after the run finished is answered from the receipt without + // touching the thread again. + const lateResponse = yield* dispatch(); + assert.equal(lateResponse.sequence, 3); + assert.equal(dispatchedCommands.length, 3); + assert.equal(createWorktree.mock.calls.length, 1); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + it.effect("does not misattribute setup activity dispatch failures as setup launch failures", () => Effect.gen(function* () { const dispatchedCommands: Array = []; diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 8fbc8d0af..5ac9c39f2 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -26,6 +26,7 @@ import { ProviderEventLoggersLive } from "./provider/Layers/ProviderEventLoggers import { ProviderServiceLive } from "./provider/Layers/ProviderService.ts"; import { ProviderSessionReaperLive } from "./provider/Layers/ProviderSessionReaper.ts"; import { ThreadAutoArchiveSweeperLive } from "./orchestration/Layers/ThreadAutoArchiveSweeper.ts"; +import { BootstrapTurnStartRunsLive } from "./orchestration/Layers/BootstrapTurnStartRuns.ts"; import { CheckpointDiffQueryLive } from "./checkpointing/Layers/CheckpointDiffQuery.ts"; import { CheckpointRevertLive } from "./checkpointing/Layers/CheckpointRevert.ts"; import { CheckpointStoreLive } from "./checkpointing/Layers/CheckpointStore.ts"; @@ -415,6 +416,9 @@ export const makeRoutesLayer = Layer.mergeAll( Layer.provide(SourceControlToolMaintenance.layer.pipe(Layer.provide(VcsProcess.layer))), Layer.provide(GitHubAuth.layer), Layer.provide(ProviderMaintenanceRunner.layer), + // One registry for the whole server: a retried bootstrap turn start must + // find the run in flight even when it arrives on a different socket. + Layer.provide(BootstrapTurnStartRunsLive), ); export const makeServerLayer = Layer.unwrap( diff --git a/apps/server/src/serverRuntimeStartup.test.ts b/apps/server/src/serverRuntimeStartup.test.ts index d749bf52b..4a147ea76 100644 --- a/apps/server/src/serverRuntimeStartup.test.ts +++ b/apps/server/src/serverRuntimeStartup.test.ts @@ -179,6 +179,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets returns existing project and threa }), Effect.provideService(OrchestrationEngineService, { readEvents: () => Stream.empty, + getCommandReceipt: () => Effect.succeed(Option.none()), dispatch: (command) => Ref.update(dispatchCalls, (calls) => [...calls, command.type]).pipe( Effect.as({ sequence: 1 }), @@ -225,6 +226,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets creates a project and thread when }), Effect.provideService(OrchestrationEngineService, { readEvents: () => Stream.empty, + getCommandReceipt: () => Effect.succeed(Option.none()), dispatch: (command) => Ref.update(dispatchCalls, (calls) => [...calls, command]).pipe( Effect.as({ sequence: 1 }), diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 2c2aef78b..7b3274d51 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -65,6 +65,8 @@ import { fileAttachmentMimeTypeForExtension } from "@threadlines/shared/fileAtta import { IMAGE_MIME_TYPE_BY_EXTENSION } from "./imageMime.ts"; import { Keybindings } from "./keybindings.ts"; import * as ExternalLauncher from "./process/externalLauncher.ts"; +import { OrchestrationCommandPreviouslyRejectedError } from "./orchestration/Errors.ts"; +import { BootstrapTurnStartRuns } from "./orchestration/Layers/BootstrapTurnStartRuns.ts"; import { normalizeDispatchCommand } from "./orchestration/Normalizer.ts"; import { coalesceLatestAggregateEvents } from "./orchestration/shellStreamCoalescing.ts"; import { OrchestrationEngineService } from "./orchestration/Services/OrchestrationEngine.ts"; @@ -245,6 +247,7 @@ const makeWsRpcLayer = (currentSessionId: AuthSessionId) => const threadSearch = yield* ThreadSearch; const usage = yield* UsageService; const orchestrationEngine = yield* OrchestrationEngineService; + const bootstrapTurnStartRuns = yield* BootstrapTurnStartRuns; const checkpointDiffQuery = yield* CheckpointDiffQuery; const checkpointRevert = yield* CheckpointRevert; const keybindings = yield* Keybindings; @@ -460,10 +463,35 @@ const makeWsRpcLayer = (currentSessionId: AuthSessionId) => } }; - const dispatchBootstrapTurnStart = ( + // The client re-sends a command whose socket dropped or whose response + // was slow, so the whole bootstrap has to be idempotent under the + // command id: a retry joins the run in flight, and a retry that lands + // after the run finished is answered from the receipt the final turn + // start left, exactly as a plain dispatch would answer it. + const runBootstrapTurnStart = ( command: Extract, ): Effect.Effect<{ readonly sequence: number }, OrchestrationDispatchCommandError> => Effect.gen(function* () { + const receipt = yield* orchestrationEngine + .getCommandReceipt(command.commandId) + .pipe( + Effect.mapError((cause) => + toDispatchCommandError(cause, "Failed to read orchestration command receipt"), + ), + ); + if (Option.isSome(receipt)) { + if (receipt.value.status === "accepted") { + return { sequence: receipt.value.resultSequence }; + } + return yield* toDispatchCommandError( + new OrchestrationCommandPreviouslyRejectedError({ + commandId: command.commandId, + detail: receipt.value.error ?? "Previously rejected.", + }), + "Command previously rejected.", + ); + } + const bootstrap = command.bootstrap; const { bootstrap: _bootstrap, ...finalTurnStartCommand } = command; let createdThread = false; @@ -696,6 +724,11 @@ const makeWsRpcLayer = (currentSessionId: AuthSessionId) => ); }); + const dispatchBootstrapTurnStart = ( + command: Extract, + ): Effect.Effect<{ readonly sequence: number }, OrchestrationDispatchCommandError> => + bootstrapTurnStartRuns.run(command.commandId, runBootstrapTurnStart(command)); + const dispatchNormalizedCommand = ( normalizedCommand: OrchestrationCommand, ): Effect.Effect<{ readonly sequence: number }, OrchestrationDispatchCommandError> => { @@ -2381,6 +2414,7 @@ export const websocketRpcRouteLayer = Layer.unwrap( const maintenance = yield* SourceControlToolMaintenance.SourceControlToolMaintenance; const providerMaintenance = yield* ProviderMaintenanceRunner.ProviderMaintenanceRunner; const githubSignIn = yield* GitHubAuth.GitHubAuth; + const bootstrapTurnStartRuns = yield* BootstrapTurnStartRuns; return HttpRouter.add( "GET", "/ws", @@ -2407,6 +2441,7 @@ export const websocketRpcRouteLayer = Layer.unwrap( ), ), Layer.provide(Layer.succeed(GitHubAuth.GitHubAuth, githubSignIn)), + Layer.provide(Layer.succeed(BootstrapTurnStartRuns, bootstrapTurnStartRuns)), Layer.provideMerge(RpcSerialization.layerJson), Layer.provide( SourceControlDiscoveryLayer.layer.pipe(