From 074e0676b5492f12cc0bc226e1e3631ab0678b10 Mon Sep 17 00:00:00 2001 From: SAPHID Date: Sun, 11 Oct 2026 23:12:01 +0000 Subject: [PATCH 1/3] fix(server): a command id reused for two threads at once no longer succeeds for both Two concurrent commands sharing one commandId but aimed at different threads both passed the pre-planning receipt check under their own thread locks. The one that lost the receipt insert then reported the winner's receipt and events as its own success, though its mutation never happened. Check the committed receipt's thread too and reject the loser with OrchestratorCommandIdConflictError. Co-Authored-By: Claude Opus 5.5 --- .../Orchestrator.receipt-race.test.ts | 123 ++++++++++++++++++ .../src/orchestration-v2/Orchestrator.ts | 10 ++ 2 files changed, 133 insertions(+) create mode 100644 apps/server/src/orchestration-v2/Orchestrator.receipt-race.test.ts diff --git a/apps/server/src/orchestration-v2/Orchestrator.receipt-race.test.ts b/apps/server/src/orchestration-v2/Orchestrator.receipt-race.test.ts new file mode 100644 index 0000000000000..b7575234df712 --- /dev/null +++ b/apps/server/src/orchestration-v2/Orchestrator.receipt-race.test.ts @@ -0,0 +1,123 @@ +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 | 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(); + 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"); + + // 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)), +); diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index d85504bb264e7..544311ca06602 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -11077,6 +11077,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 (!canReplayCommandReceipt(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") { yield* mapDispatchError(command)(startNextQueuedRun(command.threadId)); } From 91f9d97ff0b75534380bcbe6aaa7004a075844b3 Mon Sep 17 00:00:00 2001 From: SAPHID Date: Sat, 31 Oct 2026 07:48:58 +0000 Subject: [PATCH 2/3] fix(server): preserve current receipt thread matching after commit --- apps/server/src/orchestration-v2/Orchestrator.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index bbad8ad6a10d3..35f5d55616540 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -13441,7 +13441,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio } // 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 (!canReplayCommandReceipt(committed.receipt.threadId, commandThreadId(command))) { + if (committed.receipt.threadId !== commandThreadId(command)) { return yield* new OrchestratorCommandIdConflictError({ commandId: command.commandId, commandType: command.type, From ce5c2c1d5fb5813761b020b174f1dcc77fe30c4a Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Sun, 1 Nov 2026 01:15:28 +0000 Subject: [PATCH 3/3] fix(server): validate concurrent no-op command receipts --- .../Orchestrator.receipt-race.test.ts | 45 ++++++++ .../src/orchestration-v2/Orchestrator.ts | 108 ++++++++++-------- 2 files changed, 108 insertions(+), 45 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.receipt-race.test.ts b/apps/server/src/orchestration-v2/Orchestrator.receipt-race.test.ts index b7575234df712..3e6fdd19eb9dc 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.receipt-race.test.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.receipt-race.test.ts @@ -1,3 +1,4 @@ +import { vi } from "vite-plus/test"; import { assert, it } from "@effect/vitest"; import { CommandId, @@ -109,6 +110,50 @@ it.effect("a command id reused concurrently for two threads succeeds for only on 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(); + 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)) { diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 35f5d55616540..e5316631e0932 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -104,7 +104,7 @@ import { isNativeMaintenanceCommand, resolveMessageDispatchIntent, } from "./CommandPolicy.ts"; -import { CommandReceiptStoreV2 } from "./CommandReceiptStore.ts"; +import { CommandReceiptStoreV2, type CommandReceiptV2 } from "./CommandReceiptStore.ts"; import { ContextHandoffServiceV2, readForkHistoryItems, @@ -13239,6 +13239,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 => 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, @@ -13262,49 +13310,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 => 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. @@ -13399,10 +13405,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);