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..3e6fdd19eb9dc --- /dev/null +++ b/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 | 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"); + + 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)) { + 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 b4bdbd705c290..51165474edb7b 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/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 => 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 => 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" ||