diff --git a/apps/server/src/orchestration/Errors.ts b/apps/server/src/orchestration/Errors.ts index be7943f78a6..7abd567704f 100644 --- a/apps/server/src/orchestration/Errors.ts +++ b/apps/server/src/orchestration/Errors.ts @@ -53,6 +53,21 @@ export class OrchestrationCommandPreviouslyRejectedError extends Schema.TaggedEr } } +export class OrchestrationCommandIdConflictError extends Schema.TaggedErrorClass()( + "OrchestrationCommandIdConflictError", + { + commandId: Schema.String, + receiptAggregateKind: Schema.String, + receiptAggregateId: Schema.String, + commandAggregateKind: Schema.String, + commandAggregateId: Schema.String, + }, +) { + override get message(): string { + return `Command id '${this.commandId}' already used for ${this.receiptAggregateKind} '${this.receiptAggregateId}'; refusing to replay its receipt for ${this.commandAggregateKind} '${this.commandAggregateId}'.`; + } +} + export class OrchestrationProjectorDecodeError extends Schema.TaggedErrorClass()( "OrchestrationProjectorDecodeError", { @@ -82,6 +97,7 @@ export class OrchestrationListenerCallbackError extends Schema.TaggedErrorClass< export type OrchestrationDispatchError = | ProjectionRepositoryError | OrchestrationCommandInvariantError + | OrchestrationCommandIdConflictError | OrchestrationCommandPreviouslyRejectedError | OrchestrationProjectorDecodeError | OrchestrationListenerCallbackError; diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts index 9ffe50d1341..41624a14e4c 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts @@ -1220,4 +1220,153 @@ describe("OrchestrationEngine", () => { await system.dispose(); }); + + it("replays the accepted receipt for a genuine retry of the same command", async () => { + const createdAt = now(); + const system = await createOrchestrationSystem(); + const { engine } = system; + + await system.run( + engine.dispatch({ + type: "project.create", + commandId: CommandId.make("cmd-retry-project-create"), + projectId: asProjectId("project-retry"), + title: "Retry Project", + workspaceRoot: "/tmp/project-retry", + defaultModelSelection: { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5-codex", + }, + createdAt, + }), + ); + await system.run( + engine.dispatch({ + type: "thread.create", + commandId: CommandId.make("cmd-retry-thread-create"), + threadId: ThreadId.make("thread-retry"), + projectId: asProjectId("project-retry"), + title: "retry", + modelSelection: { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5-codex", + }, + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + runtimeMode: "approval-required", + branch: null, + worktreePath: null, + createdAt, + }), + ); + + const turnStart = { + type: "thread.turn.start", + commandId: CommandId.make("cmd-retry-turn-start"), + threadId: ThreadId.make("thread-retry"), + message: { + messageId: asMessageId("msg-retry"), + role: "user", + text: "hello", + attachments: [], + }, + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + runtimeMode: "approval-required", + createdAt, + } as const; + + const first = await system.run(engine.dispatch(turnStart)); + const second = await system.run(engine.dispatch(turnStart)); + expect(second.sequence).toBe(first.sequence); + + const readModel = await system.readModel(); + const thread = readModel.threads.find((candidate) => candidate.id === "thread-retry"); + expect(thread?.messages.filter((message) => message.role === "user")).toHaveLength(1); + + await system.dispose(); + }); + + it("rejects reusing an accepted command id for a different aggregate", async () => { + const createdAt = now(); + const system = await createOrchestrationSystem(); + const { engine } = system; + + await system.run( + engine.dispatch({ + type: "project.create", + commandId: CommandId.make("cmd-conflict-project-create"), + projectId: asProjectId("project-conflict"), + title: "Conflict Project", + workspaceRoot: "/tmp/project-conflict", + defaultModelSelection: { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5-codex", + }, + createdAt, + }), + ); + for (const threadId of ["thread-conflict-a", "thread-conflict-b"]) { + await system.run( + engine.dispatch({ + type: "thread.create", + commandId: CommandId.make(`cmd-${threadId}-create`), + threadId: ThreadId.make(threadId), + projectId: asProjectId("project-conflict"), + title: threadId, + modelSelection: { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5-codex", + }, + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + runtimeMode: "approval-required", + branch: null, + worktreePath: null, + createdAt, + }), + ); + } + + await system.run( + engine.dispatch({ + type: "thread.turn.start", + commandId: CommandId.make("cmd-conflict-turn-start"), + threadId: ThreadId.make("thread-conflict-a"), + message: { + messageId: asMessageId("msg-conflict-a"), + role: "user", + text: "hello", + attachments: [], + }, + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + runtimeMode: "approval-required", + createdAt, + }), + ); + + await expect( + system.run( + engine.dispatch({ + type: "thread.turn.start", + commandId: CommandId.make("cmd-conflict-turn-start"), + threadId: ThreadId.make("thread-conflict-b"), + message: { + messageId: asMessageId("msg-conflict-b"), + role: "user", + text: "hello again", + attachments: [], + }, + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + runtimeMode: "approval-required", + createdAt, + }), + ), + ).rejects.toThrow("already used for thread 'thread-conflict-a'"); + + const readModel = await system.readModel(); + const targetThread = readModel.threads.find( + (candidate) => candidate.id === "thread-conflict-b", + ); + expect(targetThread?.messages.filter((message) => message.role === "user")).toHaveLength(0); + + await system.dispose(); + }); }); diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.ts index 19184915ac7..da79b4395ac 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationEngine.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.ts @@ -32,6 +32,7 @@ import { toPersistenceSqlError } from "../../persistence/Errors.ts"; import { OrchestrationEventStore } from "../../persistence/Services/OrchestrationEventStore.ts"; import { OrchestrationCommandReceiptRepository } from "../../persistence/Services/OrchestrationCommandReceipts.ts"; import { + OrchestrationCommandIdConflictError, OrchestrationCommandInvariantError, OrchestrationCommandPreviouslyRejectedError, type OrchestrationDispatchError, @@ -48,6 +49,7 @@ import { const isOrchestrationCommandPreviouslyRejectedError = Schema.is( OrchestrationCommandPreviouslyRejectedError, ); +const isOrchestrationCommandIdConflictError = Schema.is(OrchestrationCommandIdConflictError); const isOrchestrationCommandInvariantError = Schema.is(OrchestrationCommandInvariantError); interface CommandEnvelope { @@ -139,6 +141,21 @@ const makeOrchestrationEngine = Effect.gen(function* () { commandId: envelope.command.commandId, }); if (Option.isSome(existingReceipt)) { + // A receipt only proves this exact command was handled. Replaying it + // for a command aimed at another aggregate would report success for + // work that never happened. + if ( + existingReceipt.value.aggregateKind !== aggregateRef.aggregateKind || + existingReceipt.value.aggregateId !== aggregateRef.aggregateId + ) { + return yield* new OrchestrationCommandIdConflictError({ + commandId: envelope.command.commandId, + receiptAggregateKind: existingReceipt.value.aggregateKind, + receiptAggregateId: existingReceipt.value.aggregateId, + commandAggregateKind: aggregateRef.aggregateKind, + commandAggregateId: aggregateRef.aggregateId, + }); + } if (existingReceipt.value.status === "accepted") { return { sequence: existingReceipt.value.resultSequence, @@ -262,7 +279,10 @@ const makeOrchestrationEngine = Effect.gen(function* () { } const error = Cause.squash(exit.cause) as OrchestrationDispatchError; - if (!isOrchestrationCommandPreviouslyRejectedError(error)) { + if ( + !isOrchestrationCommandPreviouslyRejectedError(error) && + !isOrchestrationCommandIdConflictError(error) + ) { yield* reconcileReadModelAfterDispatchFailure.pipe( Effect.catch(() => Effect.logWarning(