fix(server): a command id reused for two threads at once no longer succeeds for both

apps/server/src/orchestration-v2/Orchestrator.receipt-race.test.ts

@@ -0,0 +1,168 @@
+import { vi } from "vite-plus/test";
+import { assert, it } from "@effect/vitest";
+import {
+  CommandId,
+  ProjectId,
+  ProviderDriverKind,
+  ProviderInstanceId,
+  ThreadId,
+} from "@t3tools/contracts";
+import * as Cause from "effect/Cause";
+import * as Deferred from "effect/Deferred";
+import * as Effect from "effect/Effect";
+import * as Exit from "effect/Exit";
+import * as Layer from "effect/Layer";
+import * as Ref from "effect/Ref";
+import * as SqlitePersistence from "../persistence/Sqlite.ts";
+import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";
+import * as Orchestrator from "./Orchestrator.ts";
+import * as ProjectionStore from "./ProjectionStore.ts";
+import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts";
+import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts";
+import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
+
+const instanceId = ProviderInstanceId.make("codex");
+const modelSelection = { instanceId, model: "gpt-5.1-codex" };
+// Holds model-selection planning until both racing commands have read no receipt.
+let capabilityGate: Effect.Effect<void> | undefined;
+const adapter = {
+  instanceId,
+  driver: ProviderDriverKind.make("codex"),
+  getCapabilities: () =>
+    (capabilityGate ?? Effect.void).pipe(Effect.as(CodexProviderCapabilitiesV2)),
+  planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" as const }),
+  openSession: () => Effect.die("No provider process needed for metadata controls"),
+} as ProviderAdapter.ProviderAdapterV2["Service"];
+const layerDatabase = SqlitePersistence.layerMemory;
+const layerTest = Layer.mergeAll(
+  layerDatabase,
+  ProjectionStore.layer.pipe(Layer.provide(layerDatabase)),
+  ProviderReplayHarness.layerWithRegistry(
+    { name: "receipt-race" },
+    ProviderAdapterRegistry.layerFromAdapters([adapter]),
+    { databaseLayer: layerDatabase, runEffectWorker: false },
+  ),
+);
+
+it.effect("a command id reused concurrently for two threads succeeds for only one", () =>
+  Effect.gen(function* () {
+    const orchestrator = yield* Orchestrator.OrchestratorV2;
+    const projections = yield* ProjectionStore.ProjectionStoreV2;
+    const threadIds = [ThreadId.make("thread:race-a"), ThreadId.make("thread:race-b")];
+    for (const threadId of threadIds) {
+      yield* orchestrator.dispatch({
+        type: "thread.create",
+        commandId: CommandId.make(`create:${threadId}`),
+        threadId,
+        projectId: ProjectId.make("project:receipt-race"),
+        title: threadId,
+        modelSelection,
+        runtimeMode: "full-access",
+        interactionMode: "default",
+        branch: null,
+        worktreePath: null,
+        createdBy: "user",
+        creationSource: "web",
+      });
+    }
+
+    const arrived = yield* Ref.make(0);
+    const bothArrived = yield* Deferred.make<void>();
+    capabilityGate = Effect.gen(function* () {
+      if ((yield* Ref.updateAndGet(arrived, (count) => count + 1)) === 2) {
+        yield* Deferred.succeed(bothArrived, undefined);
+      }
+      yield* Deferred.await(bothArrived);
+    });
+    const setModel = (threadId: ThreadId) =>
+      orchestrator.dispatch({
+        type: "thread.model-selection.set",
+        commandId: CommandId.make("command:shared"),
+        threadId,
+        modelSelection: { ...modelSelection, model: "gpt-6" },
+      });
+    const exits = yield* Effect.forEach(threadIds, (threadId) => Effect.exit(setModel(threadId)), {
+      concurrency: "unbounded",
+    }).pipe(Effect.ensuring(Effect.sync(() => (capabilityGate = undefined))));
+
+    const winner = exits.findIndex(Exit.isSuccess);
+    const loser = exits.findIndex(Exit.isFailure);
+    assert.isAtLeast(winner, 0);
+    assert.isAtLeast(loser, 0);
+    const winnerExit = exits[winner]!;
+    const loserExit = exits[loser]!;
+    assert.isTrue(Exit.isSuccess(winnerExit) && Exit.isFailure(loserExit));
+    if (Exit.isFailure(loserExit)) {
+      const error = Cause.findErrorOption(loserExit.cause);
+      assert.isTrue(error._tag === "Some");
+      if (error._tag === "Some") {
+        assert.equal(error.value._tag, "OrchestratorCommandIdConflictError");
+        if (error.value._tag === "OrchestratorCommandIdConflictError") {
+          assert.equal(error.value.receiptThreadId, threadIds[winner]);
+          assert.equal(error.value.commandThreadId, threadIds[loser]);
+        }
+      }
+    }
+
+    const models = yield* Effect.forEach(threadIds, (threadId) =>
+      projections.getThreadShell(threadId).pipe(Effect.map((shell) => shell?.modelSelection.model)),
+    );
+    assert.equal(models[winner], "gpt-6");
+    assert.equal(models[loser], "gpt-5.1-codex");
+
+    const noOpArrived = yield* Ref.make(0);
+    const bothNoOpsArrived = yield* Deferred.make<void>();
+    const getThreadRecords = projections.getThreadRecords;
+    using _records = vi.spyOn(projections, "getThreadRecords").mockImplementation((...args) =>
+      Effect.gen(function* () {
+        if ((yield* Ref.updateAndGet(noOpArrived, (count) => count + 1)) === 2) {
+          yield* Deferred.succeed(bothNoOpsArrived, undefined);
+        }
+        yield* Deferred.await(bothNoOpsArrived);
+        return yield* getThreadRecords(...args);
+      }),
+    );
+    const closeShare = (threadId: ThreadId) =>
+      orchestrator.dispatch({
+        type: "thread.guest-share.close",
+        commandId: CommandId.make("command:shared-noop"),
+        threadId,
+      });
+    const noOpExits = yield* Effect.forEach(
+      threadIds,
+      (threadId) => Effect.exit(closeShare(threadId)),
+      {
+        concurrency: "unbounded",
+      },
+    );
+    const noOpWinner = noOpExits.findIndex(Exit.isSuccess);
+    const noOpLoser = noOpExits.findIndex(Exit.isFailure);
+    assert.isAtLeast(noOpWinner, 0);
+    assert.isAtLeast(noOpLoser, 0);
+    const noOpFailure = noOpExits[noOpLoser]!;
+    if (Exit.isFailure(noOpFailure)) {
+      const error = Cause.findErrorOption(noOpFailure.cause);
+      assert.isTrue(error._tag === "Some");
+      if (error._tag === "Some") {
+        assert.equal(error.value._tag, "OrchestratorCommandIdConflictError");
+      }
+    }
+    const noOpReplay = yield* closeShare(threadIds[noOpWinner]!);
+    const noOpSuccess = noOpExits[noOpWinner]!;
+    if (Exit.isSuccess(noOpSuccess)) {
+      assert.equal(noOpReplay.sequence, noOpSuccess.value.sequence);
+      assert.deepEqual(noOpReplay.storedEvents, []);
+    }
+
+    // The winner's own thread still replays its receipt.
+    const replay = yield* setModel(threadIds[winner]!);
+    if (Exit.isSuccess(winnerExit)) {
+      assert.equal(replay.sequence, winnerExit.value.sequence);
+      assert.isAbove(replay.storedEvents.length, 0);
+      assert.deepEqual(
+        replay.storedEvents.map((stored) => stored.sequence),
+        winnerExit.value.storedEvents.map((stored) => stored.sequence),
+      );
+    }
+  }).pipe(Effect.provide(layerTest)),
+);

apps/server/src/orchestration-v2/Orchestrator.ts

@@ -109,7 +109,7 @@ import {
   isNativeMaintenanceCommand,
   resolveMessageDispatchIntent,
 } from "./CommandPolicy.ts";
-import { CommandReceiptStoreV2 } from "./CommandReceiptStore.ts";
+import { CommandReceiptStoreV2, type CommandReceiptV2 } from "./CommandReceiptStore.ts";
 import {
   ContextHandoffServiceV2,
   readForkHistoryItems,
@@ -13496,6 +13496,54 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
     };
   });
 
+  const replayCommandReceipt = Effect.fn("orchestrationV2.dispatch.replayReceipt")(function* (
+    command: OrchestrationV2ServerCommand,
+    receipt: CommandReceiptV2,
+  ) {
+    if (receipt.status === "rejected") {
+      return yield* new OrchestratorCommandPreviouslyRejectedError({
+        commandId: command.commandId,
+        commandType: command.type,
+        detail: receipt.error ?? "Previously rejected.",
+      });
+    }
+    const dispatchThreadId = commandThreadId(command);
+    // Receipts only prove execution for the thread they were recorded against.
+    if (receipt.threadId !== dispatchThreadId) {
+      return yield* new OrchestratorCommandIdConflictError({
+        commandId: command.commandId,
+        commandType: command.type,
+        receiptThreadId: receipt.threadId,
+        commandThreadId: dispatchThreadId,
+      });
+    }
+    const storedEvents = yield* eventSink.readByCommandId({ commandId: command.commandId }).pipe(
+      Stream.runCollect,
+      Effect.map((events): ReadonlyArray<OrchestrationV2StoredEvent> => Array.from(events)),
+      Effect.mapError(
+        (cause) =>
+          new OrchestratorDispatchError({
+            commandId: command.commandId,
+            commandType: command.type,
+            cause,
+          }),
+      ),
+    );
+    if (command.type === "queue.resume" || command.type === "queued-run.start") {
+      yield* mapDispatchError(command)(
+        startNextQueuedRun(
+          command.threadId,
+          command.type === "queued-run.start" ? { queuedRunId: command.runId } : undefined,
+        ),
+      );
+    }
+    return {
+      sequence: receipt.resultSequence,
+      storedEvents,
+      replayed: true,
+    } satisfies OrchestratorV2DispatchResult;
+  });
+
   /** `sideChatIds` are the live side chats a `thread.delete` cascades to, read under their locks. */
   const dispatchWithReceiptEffect = Effect.fn("orchestrationV2.dispatch.withReceipt")(function* (
     command: OrchestrationV2ServerCommand,
@@ -13519,49 +13567,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
     );
 
     if (Option.isSome(existingReceipt)) {
-      const receipt = existingReceipt.value;
-      if (receipt.status === "rejected") {
-        return yield* new OrchestratorCommandPreviouslyRejectedError({
-          commandId: command.commandId,
-          commandType: command.type,
-          detail: receipt.error ?? "Previously rejected.",
-        });
-      }
-      const dispatchThreadId = commandThreadId(command);
-      // Receipts only prove execution for the thread they were recorded against.
-      if (receipt.threadId !== dispatchThreadId) {
-        return yield* new OrchestratorCommandIdConflictError({
-          commandId: command.commandId,
-          commandType: command.type,
-          receiptThreadId: receipt.threadId,
-          commandThreadId: dispatchThreadId,
-        });
-      }
-      const storedEvents = yield* eventSink.readByCommandId({ commandId: command.commandId }).pipe(
-        Stream.runCollect,
-        Effect.map((events): ReadonlyArray<OrchestrationV2StoredEvent> => Array.from(events)),
-        Effect.mapError(
-          (cause) =>
-            new OrchestratorDispatchError({
-              commandId: command.commandId,
-              commandType: command.type,
-              cause,
-            }),
-        ),
-      );
-      if (command.type === "queue.resume" || command.type === "queued-run.start") {
-        yield* mapDispatchError(command)(
-          startNextQueuedRun(
-            command.threadId,
-            command.type === "queued-run.start" ? { queuedRunId: command.runId } : undefined,
-          ),
-        );
-      }
-      return {
-        sequence: receipt.resultSequence,
-        storedEvents,
-        replayed: true,
-      } satisfies OrchestratorV2DispatchResult;
+      return yield* replayCommandReceipt(command, existingReceipt.value);
     }
 
     // Recheck under the lock; the user may have raised modes since sender preflight.
@@ -13676,10 +13682,22 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
             }),
         ),
       );
+      if (!inserted) {
+        const receipt = yield* mapDispatchError(command)(
+          commandReceipts.getByCommandId(command.commandId),
+        );
+        if (Option.isNone(receipt)) {
+          return yield* new OrchestratorDispatchError({
+            commandId: command.commandId,
+            commandType: command.type,
+            cause: "Conflicting receipt is not a thread command receipt.",
+          });
+        }
+        return yield* replayCommandReceipt(command, receipt.value);
+      }
       return {
         sequence: resultSequence,
         storedEvents: [],
-        ...(inserted ? {} : { replayed: true as const }),
       } satisfies OrchestratorV2DispatchResult;
     }
     const acceptedAt = plan.events.at(-1)?.occurredAt ?? (yield* DateTime.now);
@@ -13716,6 +13734,16 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
         detail: committed.receipt.error ?? "Previously rejected.",
       });
     }
+    // A concurrent command with this id for another thread can commit first,
+    // under that thread's lock; its receipt is not this command's result.
+    if (committed.receipt.threadId !== commandThreadId(command)) {
+      return yield* new OrchestratorCommandIdConflictError({
+        commandId: command.commandId,
+        commandType: command.type,
+        receiptThreadId: committed.receipt.threadId,
+        commandThreadId: commandThreadId(command),
+      });
+    }
     if (
       command.type === "queue.resume" ||
       command.type === "queued-run.start" ||