fix(server): make bootstrap turn starts idempotent

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<BootstrapTurnStartResult, OrchestrationDispatchCommandError>
+  >();
+
+  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,
+);

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.

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}`);

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}`);

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<OrchestrationCommandReceipt>,
+    OrchestrationCommandReceiptRepositoryError,
+    never
+  >;
+
   /**
    * Stream persisted domain events in dispatch order.
    *

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),
         }),

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<OrchestrationCommand> = [];
+        const receiptSequences = new Map<CommandId, number>();
+        const worktreeRequested = yield* Deferred.make<void>();
+        const worktreeGate = yield* Deferred.make<void>();
+        const createWorktree = vi.fn(
+          (_: Parameters<GitVcsDriver.GitVcsDriverShape["createWorktree"]>[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<OrchestrationCommand> = [];

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(

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 }),