From b39ef2805380ce256c859518bdb474598d0e9a51 Mon Sep 17 00:00:00 2001 From: ostapondo <33957189+ostapondo@users.noreply.github.com> Date: Sun, 2 Aug 2026 21:38:30 +0200 Subject: [PATCH 1/2] fix(server): stop replaying a command receipt for a different aggregate The orchestration engine deduplicates commands by command id alone: when a receipt exists, dispatch returns the stored result sequence without checking which aggregate the receipt belongs to. Reusing an accepted command id against another thread therefore reports success while creating nothing on the target thread. Compare the receipt's stored aggregate with the incoming command's aggregate and fail dispatch with a typed conflict error on mismatch. A genuine retry (same command id, same aggregate) still replays the stored receipt. --- apps/server/src/orchestration/Errors.ts | 16 ++ .../Layers/OrchestrationEngine.test.ts | 143 ++++++++++++++++++ .../Layers/OrchestrationEngine.ts | 22 ++- 3 files changed, 180 insertions(+), 1 deletion(-) 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..ddd2f7324c6 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts @@ -1220,4 +1220,147 @@ 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'"); + + 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( From 4c0aa1eec97bb9181571330a97214ec1479ff38b Mon Sep 17 00:00:00 2001 From: ostapondo <33957189+ostapondo@users.noreply.github.com> Date: Sun, 2 Aug 2026 22:15:31 +0200 Subject: [PATCH 2/2] test(server): assert the conflicted dispatch leaves the target thread untouched --- .../src/orchestration/Layers/OrchestrationEngine.test.ts | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts index ddd2f7324c6..41624a14e4c 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts @@ -1361,6 +1361,12 @@ describe("OrchestrationEngine", () => { ), ).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(); }); });