authorgravatar for 33957189+ostapondo@users.noreply.github.comOstap <33957189+ostapondo@users.noreply.github.com> 2026-08-15 14:27:19+02:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-08-15 14:27:19+02:00
log4fc80fcbdc1ed33c46296dbad08517f2968856e6
tree21ad04fb8d0ad1db2e25ec638628b31491848bc9
parent3b54a2a574b061a1ce80d7130e0776f4f5c95e20
signature Signed by PGP key B5690EEEBB952194

fix(server): stop replaying a command receipt for a different aggregate (#5246)

3 files changed, 186 insertions(+), 1 deletions(-)

apps/server/src/orchestration/Errors.ts+16
......@@ -53,6 +53,21 @@ export class OrchestrationCommandPreviouslyRejectedError extends Schema.TaggedEr
5353 }
5454}
5555
56export class OrchestrationCommandIdConflictError extends Schema.TaggedErrorClass<OrchestrationCommandIdConflictError>()(
57 "OrchestrationCommandIdConflictError",
58 {
59 commandId: Schema.String,
60 receiptAggregateKind: Schema.String,
61 receiptAggregateId: Schema.String,
62 commandAggregateKind: Schema.String,
63 commandAggregateId: Schema.String,
64 },
65) {
66 override get message(): string {
67 return `Command id '${this.commandId}' already used for ${this.receiptAggregateKind} '${this.receiptAggregateId}'; refusing to replay its receipt for ${this.commandAggregateKind} '${this.commandAggregateId}'.`;
68 }
69}
70
5671export class OrchestrationProjectorDecodeError extends Schema.TaggedErrorClass<OrchestrationProjectorDecodeError>()(
5772 "OrchestrationProjectorDecodeError",
5873 {
......@@ -82,6 +97,7 @@ export class OrchestrationListenerCallbackError extends Schema.TaggedErrorClass<
8297export type OrchestrationDispatchError =
8398 | ProjectionRepositoryError
8499 | OrchestrationCommandInvariantError
100 | OrchestrationCommandIdConflictError
85101 | OrchestrationCommandPreviouslyRejectedError
86102 | OrchestrationProjectorDecodeError
87103 | OrchestrationListenerCallbackError;
apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts+149
......@@ -1230,4 +1230,153 @@ describe("OrchestrationEngine", () => {
12301230
12311231 await system.dispose();
12321232 });
1233
1234 it("replays the accepted receipt for a genuine retry of the same command", async () => {
1235 const createdAt = now();
1236 const system = await createOrchestrationSystem();
1237 const { engine } = system;
1238
1239 await system.run(
1240 engine.dispatch({
1241 type: "project.create",
1242 commandId: CommandId.make("cmd-retry-project-create"),
1243 projectId: asProjectId("project-retry"),
1244 title: "Retry Project",
1245 workspaceRoot: "/tmp/project-retry",
1246 defaultModelSelection: {
1247 instanceId: ProviderInstanceId.make("codex"),
1248 model: "gpt-5-codex",
1249 },
1250 createdAt,
1251 }),
1252 );
1253 await system.run(
1254 engine.dispatch({
1255 type: "thread.create",
1256 commandId: CommandId.make("cmd-retry-thread-create"),
1257 threadId: ThreadId.make("thread-retry"),
1258 projectId: asProjectId("project-retry"),
1259 title: "retry",
1260 modelSelection: {
1261 instanceId: ProviderInstanceId.make("codex"),
1262 model: "gpt-5-codex",
1263 },
1264 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
1265 runtimeMode: "approval-required",
1266 branch: null,
1267 worktreePath: null,
1268 createdAt,
1269 }),
1270 );
1271
1272 const turnStart = {
1273 type: "thread.turn.start",
1274 commandId: CommandId.make("cmd-retry-turn-start"),
1275 threadId: ThreadId.make("thread-retry"),
1276 message: {
1277 messageId: asMessageId("msg-retry"),
1278 role: "user",
1279 text: "hello",
1280 attachments: [],
1281 },
1282 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
1283 runtimeMode: "approval-required",
1284 createdAt,
1285 } as const;
1286
1287 const first = await system.run(engine.dispatch(turnStart));
1288 const second = await system.run(engine.dispatch(turnStart));
1289 expect(second.sequence).toBe(first.sequence);
1290
1291 const readModel = await system.readModel();
1292 const thread = readModel.threads.find((candidate) => candidate.id === "thread-retry");
1293 expect(thread?.messages.filter((message) => message.role === "user")).toHaveLength(1);
1294
1295 await system.dispose();
1296 });
1297
1298 it("rejects reusing an accepted command id for a different aggregate", async () => {
1299 const createdAt = now();
1300 const system = await createOrchestrationSystem();
1301 const { engine } = system;
1302
1303 await system.run(
1304 engine.dispatch({
1305 type: "project.create",
1306 commandId: CommandId.make("cmd-conflict-project-create"),
1307 projectId: asProjectId("project-conflict"),
1308 title: "Conflict Project",
1309 workspaceRoot: "/tmp/project-conflict",
1310 defaultModelSelection: {
1311 instanceId: ProviderInstanceId.make("codex"),
1312 model: "gpt-5-codex",
1313 },
1314 createdAt,
1315 }),
1316 );
1317 for (const threadId of ["thread-conflict-a", "thread-conflict-b"]) {
1318 await system.run(
1319 engine.dispatch({
1320 type: "thread.create",
1321 commandId: CommandId.make(`cmd-${threadId}-create`),
1322 threadId: ThreadId.make(threadId),
1323 projectId: asProjectId("project-conflict"),
1324 title: threadId,
1325 modelSelection: {
1326 instanceId: ProviderInstanceId.make("codex"),
1327 model: "gpt-5-codex",
1328 },
1329 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
1330 runtimeMode: "approval-required",
1331 branch: null,
1332 worktreePath: null,
1333 createdAt,
1334 }),
1335 );
1336 }
1337
1338 await system.run(
1339 engine.dispatch({
1340 type: "thread.turn.start",
1341 commandId: CommandId.make("cmd-conflict-turn-start"),
1342 threadId: ThreadId.make("thread-conflict-a"),
1343 message: {
1344 messageId: asMessageId("msg-conflict-a"),
1345 role: "user",
1346 text: "hello",
1347 attachments: [],
1348 },
1349 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
1350 runtimeMode: "approval-required",
1351 createdAt,
1352 }),
1353 );
1354
1355 await expect(
1356 system.run(
1357 engine.dispatch({
1358 type: "thread.turn.start",
1359 commandId: CommandId.make("cmd-conflict-turn-start"),
1360 threadId: ThreadId.make("thread-conflict-b"),
1361 message: {
1362 messageId: asMessageId("msg-conflict-b"),
1363 role: "user",
1364 text: "hello again",
1365 attachments: [],
1366 },
1367 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
1368 runtimeMode: "approval-required",
1369 createdAt,
1370 }),
1371 ),
1372 ).rejects.toThrow("already used for thread 'thread-conflict-a'");
1373
1374 const readModel = await system.readModel();
1375 const targetThread = readModel.threads.find(
1376 (candidate) => candidate.id === "thread-conflict-b",
1377 );
1378 expect(targetThread?.messages.filter((message) => message.role === "user")).toHaveLength(0);
1379
1380 await system.dispose();
1381 });
12331382});
apps/server/src/orchestration/Layers/OrchestrationEngine.ts+21-1
......@@ -32,6 +32,7 @@ import { toPersistenceSqlError } from "../../persistence/Errors.ts";
3232import { OrchestrationEventStore } from "../../persistence/Services/OrchestrationEventStore.ts";
3333import { OrchestrationCommandReceiptRepository } from "../../persistence/Services/OrchestrationCommandReceipts.ts";
3434import {
35 OrchestrationCommandIdConflictError,
3536 OrchestrationCommandInvariantError,
3637 OrchestrationCommandPreviouslyRejectedError,
3738 type OrchestrationDispatchError,
......@@ -48,6 +49,7 @@ import {
4849const isOrchestrationCommandPreviouslyRejectedError = Schema.is(
4950 OrchestrationCommandPreviouslyRejectedError,
5051);
52const isOrchestrationCommandIdConflictError = Schema.is(OrchestrationCommandIdConflictError);
5153const isOrchestrationCommandInvariantError = Schema.is(OrchestrationCommandInvariantError);
5254
5355interface CommandEnvelope {
......@@ -139,6 +141,21 @@ const makeOrchestrationEngine = Effect.gen(function* () {
139141 commandId: envelope.command.commandId,
140142 });
141143 if (Option.isSome(existingReceipt)) {
144 // A receipt only proves this exact command was handled. Replaying it
145 // for a command aimed at another aggregate would report success for
146 // work that never happened.
147 if (
148 existingReceipt.value.aggregateKind !== aggregateRef.aggregateKind ||
149 existingReceipt.value.aggregateId !== aggregateRef.aggregateId
150 ) {
151 return yield* new OrchestrationCommandIdConflictError({
152 commandId: envelope.command.commandId,
153 receiptAggregateKind: existingReceipt.value.aggregateKind,
154 receiptAggregateId: existingReceipt.value.aggregateId,
155 commandAggregateKind: aggregateRef.aggregateKind,
156 commandAggregateId: aggregateRef.aggregateId,
157 });
158 }
142159 if (existingReceipt.value.status === "accepted") {
143160 return {
144161 sequence: existingReceipt.value.resultSequence,
......@@ -262,7 +279,10 @@ const makeOrchestrationEngine = Effect.gen(function* () {
262279 }
263280
264281 const error = Cause.squash(exit.cause) as OrchestrationDispatchError;
265 if (!isOrchestrationCommandPreviouslyRejectedError(error)) {
282 if (
283 !isOrchestrationCommandPreviouslyRejectedError(error) &&
284 !isOrchestrationCommandIdConflictError(error)
285 ) {
266286 yield* reconcileReadModelAfterDispatchFailure.pipe(
267287 Effect.catch(() =>
268288 Effect.logWarning(