diff --git a/.plans/16c-pr89-remediation-checklist.md b/.plans/16c-pr89-remediation-checklist.md index 5ebe2fe235451e5229650ef57640e690222677d2..61b8c2c9c5b99848fb7fff9d563a6b40957d947c 100644 --- a/.plans/16c-pr89-remediation-checklist.md +++ b/.plans/16c-pr89-remediation-checklist.md @@ -154,8 +154,8 @@ Counts: active `53` (`valid=34`, `partially-valid=19`), closed-invalid `5` ### Phase 2 -- [ ] `C001` Non-atomic event appending can corrupt state on retry. If an error occurs mid-loop (lines 96-102) after some events are persisted but before the receipt is written, the command appears to fail. A retry generates new UUIDs via `crypto.randomUUID()` in the decider, appending duplicate events. Consider wrapping the loop in a transaction or using deterministic event IDs derived from `commandId`.
🚀 Reply "fix it for me" or copy this AI Prompt for your agent: - - Status: `TODO` +- [x] `C001` Non-atomic event appending can corrupt state on retry. If an error occurs mid-loop (lines 96-102) after some events are persisted but before the receipt is written, the command appears to fail. A retry generates new UUIDs via `crypto.randomUUID()` in the decider, appending duplicate events. Consider wrapping the loop in a transaction or using deterministic event IDs derived from `commandId`.
🚀 Reply "fix it for me" or copy this AI Prompt for your agent: + - Status: `DONE` - Verdict: `valid` - Severity: `High` - Area: `Event ordering and state consistency` diff --git a/apps/server/src/orchestration/Layer.test.ts b/apps/server/src/orchestration/Layer.test.ts deleted file mode 100644 index 1ccf75579fa93ef7cae818baeafe2ff3d22ef5a0..0000000000000000000000000000000000000000 --- a/apps/server/src/orchestration/Layer.test.ts +++ /dev/null @@ -1,432 +0,0 @@ -import { - CheckpointRef, - CommandId, - MessageId, - ProjectId, - ThreadId, - TurnId, - type OrchestrationEvent, -} from "@t3tools/contracts"; -import { Effect, Layer, ManagedRuntime, Queue, Stream } from "effect"; -import { describe, expect, it } from "vitest"; - -import { PersistenceSqlError } from "../persistence/Errors.ts"; -import { OrchestrationCommandReceiptRepositoryLive } from "../persistence/Layers/OrchestrationCommandReceipts.ts"; -import { OrchestrationEventStoreLive } from "../persistence/Layers/OrchestrationEventStore.ts"; -import { SqlitePersistenceMemory } from "../persistence/Layers/Sqlite.ts"; -import { - OrchestrationEventStore, - type OrchestrationEventStoreShape, -} from "../persistence/Services/OrchestrationEventStore.ts"; -import { OrchestrationEngineLive } from "./Layers/OrchestrationEngine.ts"; -import { OrchestrationProjectionPipelineLive } from "./Layers/ProjectionPipeline.ts"; -import { OrchestrationEngineService } from "./Services/OrchestrationEngine.ts"; - -const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value); -const asMessageId = (value: string): MessageId => MessageId.makeUnsafe(value); -const asTurnId = (value: string): TurnId => TurnId.makeUnsafe(value); -const asCheckpointRef = (value: string): CheckpointRef => CheckpointRef.makeUnsafe(value); - -async function createOrchestrationSystem() { - const orchestrationLayer = OrchestrationEngineLive.pipe( - Layer.provide(OrchestrationProjectionPipelineLive), - Layer.provide(OrchestrationEventStoreLive), - Layer.provide(OrchestrationCommandReceiptRepositoryLive), - Layer.provide(SqlitePersistenceMemory), - ); - const runtime = ManagedRuntime.make(orchestrationLayer); - const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService)); - return { - engine, - run: (effect: Effect.Effect) => runtime.runPromise(effect), - dispose: () => runtime.dispose(), - }; -} - -function now() { - return new Date().toISOString(); -} - -describe("OrchestrationEngine", () => { - it("returns deterministic read models for repeated reads", async () => { - const createdAt = now(); - const system = await createOrchestrationSystem(); - const { engine } = system; - - await system.run( - engine.dispatch({ - type: "project.create", - commandId: CommandId.makeUnsafe("cmd-project-1-create"), - projectId: asProjectId("project-1"), - title: "Project 1", - workspaceRoot: "/tmp/project-1", - defaultModel: "gpt-5-codex", - createdAt, - }), - ); - await system.run( - engine.dispatch({ - type: "thread.create", - commandId: CommandId.makeUnsafe("cmd-thread-1-create"), - threadId: ThreadId.makeUnsafe("thread-1"), - projectId: asProjectId("project-1"), - title: "Thread", - model: "gpt-5-codex", - branch: null, - worktreePath: null, - createdAt, - }), - ); - await system.run( - engine.dispatch({ - type: "thread.turn.start", - commandId: CommandId.makeUnsafe("cmd-turn-start-1"), - threadId: ThreadId.makeUnsafe("thread-1"), - message: { - messageId: asMessageId("msg-1"), - role: "user", - text: "hello", - attachments: [], - }, - createdAt, - }), - ); - - const readModelA = await system.run(engine.getReadModel()); - const readModelB = await system.run(engine.getReadModel()); - expect(readModelB).toEqual(readModelA); - await system.dispose(); - }); - - it("replays append-only events from sequence", async () => { - const system = await createOrchestrationSystem(); - const { engine } = system; - const createdAt = now(); - - await system.run( - engine.dispatch({ - type: "project.create", - commandId: CommandId.makeUnsafe("cmd-project-replay-create"), - projectId: asProjectId("project-replay"), - title: "Replay Project", - workspaceRoot: "/tmp/project-replay", - defaultModel: "gpt-5-codex", - createdAt, - }), - ); - await system.run( - engine.dispatch({ - type: "thread.create", - commandId: CommandId.makeUnsafe("cmd-thread-replay-create"), - threadId: ThreadId.makeUnsafe("thread-replay"), - projectId: asProjectId("project-replay"), - title: "replay", - model: "gpt-5-codex", - branch: null, - worktreePath: null, - createdAt, - }), - ); - await system.run( - engine.dispatch({ - type: "thread.delete", - commandId: CommandId.makeUnsafe("cmd-thread-replay-delete"), - threadId: ThreadId.makeUnsafe("thread-replay"), - }), - ); - - const events = await system.run( - Stream.runCollect(engine.readEvents(0)).pipe( - Effect.map((chunk): OrchestrationEvent[] => Array.from(chunk)), - ), - ); - expect(events.map((event) => event.type)).toEqual([ - "project.created", - "thread.created", - "thread.deleted", - ]); - await system.dispose(); - }); - - it("streams persisted domain events in order", async () => { - const system = await createOrchestrationSystem(); - const { engine } = system; - const createdAt = now(); - - await system.run( - engine.dispatch({ - type: "project.create", - commandId: CommandId.makeUnsafe("cmd-project-stream-create"), - projectId: asProjectId("project-stream"), - title: "Stream Project", - workspaceRoot: "/tmp/project-stream", - defaultModel: "gpt-5-codex", - createdAt, - }), - ); - - const eventTypes: string[] = []; - await system.run( - Effect.gen(function* () { - const eventQueue = yield* Queue.unbounded(); - yield* Effect.forkScoped( - Stream.take(engine.streamDomainEvents, 2).pipe( - Stream.runForEach((event) => Queue.offer(eventQueue, event).pipe(Effect.asVoid)), - ), - ); - yield* Effect.sleep("10 millis"); - yield* engine.dispatch({ - type: "thread.create", - commandId: CommandId.makeUnsafe("cmd-stream-thread-create"), - threadId: ThreadId.makeUnsafe("thread-stream"), - projectId: asProjectId("project-stream"), - title: "domain-stream", - model: "gpt-5-codex", - branch: null, - worktreePath: null, - createdAt, - }); - yield* engine.dispatch({ - type: "thread.meta.update", - commandId: CommandId.makeUnsafe("cmd-stream-thread-update"), - threadId: ThreadId.makeUnsafe("thread-stream"), - title: "domain-stream-updated", - }); - eventTypes.push((yield* Queue.take(eventQueue)).type); - eventTypes.push((yield* Queue.take(eventQueue)).type); - }).pipe(Effect.scoped), - ); - - expect(eventTypes).toEqual(["thread.created", "thread.meta-updated"]); - await system.dispose(); - }); - - it("stores completed checkpoint summaries even when no files changed", async () => { - const system = await createOrchestrationSystem(); - const { engine } = system; - const createdAt = now(); - - await system.run( - engine.dispatch({ - type: "project.create", - commandId: CommandId.makeUnsafe("cmd-project-turn-diff-create"), - projectId: asProjectId("project-turn-diff"), - title: "Turn Diff Project", - workspaceRoot: "/tmp/project-turn-diff", - defaultModel: "gpt-5-codex", - createdAt, - }), - ); - await system.run( - engine.dispatch({ - type: "thread.create", - commandId: CommandId.makeUnsafe("cmd-thread-turn-diff-create"), - threadId: ThreadId.makeUnsafe("thread-turn-diff"), - projectId: asProjectId("project-turn-diff"), - title: "Turn diff thread", - model: "gpt-5-codex", - branch: null, - worktreePath: null, - createdAt, - }), - ); - await system.run( - engine.dispatch({ - type: "thread.turn.diff.complete", - commandId: CommandId.makeUnsafe("cmd-turn-diff-complete"), - threadId: ThreadId.makeUnsafe("thread-turn-diff"), - turnId: asTurnId("turn-1"), - completedAt: createdAt, - checkpointRef: asCheckpointRef("refs/t3/checkpoints/thread-turn-diff/turn/1"), - status: "ready", - files: [], - checkpointTurnCount: 1, - createdAt, - }), - ); - - const thread = (await system.run(engine.getReadModel())).threads.find( - (entry) => entry.id === "thread-turn-diff", - ); - expect(thread?.checkpoints).toEqual([ - { - turnId: asTurnId("turn-1"), - checkpointTurnCount: 1, - checkpointRef: asCheckpointRef("refs/t3/checkpoints/thread-turn-diff/turn/1"), - status: "ready", - files: [], - assistantMessageId: null, - completedAt: createdAt, - }, - ]); - await system.dispose(); - }); - - it("keeps processing queued commands after a storage failure", async () => { - type StoredEvent = - ReturnType extends Effect.Effect - ? A - : never; - const events: StoredEvent[] = []; - let nextSequence = 1; - let shouldFailFirstAppend = true; - - const flakyStore: OrchestrationEventStoreShape = { - append(event) { - if (shouldFailFirstAppend && event.commandId === CommandId.makeUnsafe("cmd-flaky-1")) { - shouldFailFirstAppend = false; - return Effect.fail( - new PersistenceSqlError({ - operation: "test.append", - detail: "append failed", - }), - ); - } - const savedEvent = { - ...event, - sequence: nextSequence, - } as StoredEvent; - nextSequence += 1; - events.push(savedEvent); - return Effect.succeed(savedEvent); - }, - readFromSequence(sequenceExclusive) { - return Stream.fromIterable(events.filter((event) => event.sequence > sequenceExclusive)); - }, - readAll() { - return Stream.fromIterable(events); - }, - }; - - const runtime = ManagedRuntime.make( - OrchestrationEngineLive.pipe( - Layer.provide(OrchestrationProjectionPipelineLive), - Layer.provide(Layer.succeed(OrchestrationEventStore, flakyStore)), - Layer.provide(OrchestrationCommandReceiptRepositoryLive), - Layer.provide(SqlitePersistenceMemory), - ), - ); - const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService)); - const createdAt = now(); - - await runtime.runPromise( - engine.dispatch({ - type: "project.create", - commandId: CommandId.makeUnsafe("cmd-project-flaky-create"), - projectId: asProjectId("project-flaky"), - title: "Flaky Project", - workspaceRoot: "/tmp/project-flaky", - defaultModel: "gpt-5-codex", - createdAt, - }), - ); - - await expect( - runtime.runPromise( - engine.dispatch({ - type: "thread.create", - commandId: CommandId.makeUnsafe("cmd-flaky-1"), - threadId: ThreadId.makeUnsafe("thread-flaky-fail"), - projectId: asProjectId("project-flaky"), - title: "flaky-fail", - model: "gpt-5-codex", - branch: null, - worktreePath: null, - createdAt, - }), - ), - ).rejects.toThrow("append failed"); - - const result = await runtime.runPromise( - engine.dispatch({ - type: "thread.create", - commandId: CommandId.makeUnsafe("cmd-flaky-2"), - threadId: ThreadId.makeUnsafe("thread-flaky-ok"), - projectId: asProjectId("project-flaky"), - title: "flaky-ok", - model: "gpt-5-codex", - branch: null, - worktreePath: null, - createdAt, - }), - ); - - expect(result.sequence).toBe(2); - expect((await runtime.runPromise(engine.getReadModel())).snapshotSequence).toBe(2); - await runtime.dispose(); - }); - - it("fails command dispatch when command invariants are violated", async () => { - const system = await createOrchestrationSystem(); - const { engine } = system; - - await expect( - system.run( - engine.dispatch({ - type: "thread.turn.start", - commandId: CommandId.makeUnsafe("cmd-invariant-missing-thread"), - threadId: ThreadId.makeUnsafe("thread-missing"), - message: { - messageId: asMessageId("msg-missing"), - role: "user", - text: "hello", - attachments: [], - }, - createdAt: now(), - }), - ), - ).rejects.toThrow("Thread 'thread-missing' does not exist"); - - await system.dispose(); - }); - - it("rejects duplicate thread creation", async () => { - const system = await createOrchestrationSystem(); - const { engine } = system; - const createdAt = now(); - - await system.run( - engine.dispatch({ - type: "project.create", - commandId: CommandId.makeUnsafe("cmd-project-duplicate-create"), - projectId: asProjectId("project-duplicate"), - title: "Duplicate Project", - workspaceRoot: "/tmp/project-duplicate", - defaultModel: "gpt-5-codex", - createdAt, - }), - ); - - await system.run( - engine.dispatch({ - type: "thread.create", - commandId: CommandId.makeUnsafe("cmd-thread-duplicate-1"), - threadId: ThreadId.makeUnsafe("thread-duplicate"), - projectId: asProjectId("project-duplicate"), - title: "duplicate", - model: "gpt-5-codex", - branch: null, - worktreePath: null, - createdAt, - }), - ); - - await expect( - system.run( - engine.dispatch({ - type: "thread.create", - commandId: CommandId.makeUnsafe("cmd-thread-duplicate-2"), - threadId: ThreadId.makeUnsafe("thread-duplicate"), - projectId: asProjectId("project-duplicate"), - title: "duplicate", - model: "gpt-5-codex", - branch: null, - worktreePath: null, - createdAt, - }), - ), - ).rejects.toThrow("already exists"); - - await system.dispose(); - }); -}); diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..e2c8ab35f5789e1a5281cce93367d7a310ada88c --- /dev/null +++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts @@ -0,0 +1,543 @@ +import { + CheckpointRef, + CommandId, + MessageId, + ProjectId, + ThreadId, + TurnId, + type OrchestrationEvent, +} from "@t3tools/contracts"; +import { Effect, Layer, ManagedRuntime, Queue, Stream } from "effect"; +import { describe, expect, it } from "vitest"; + +import { PersistenceSqlError } from "../../persistence/Errors.ts"; +import { OrchestrationCommandReceiptRepositoryLive } from "../../persistence/Layers/OrchestrationCommandReceipts.ts"; +import { OrchestrationEventStoreLive } from "../../persistence/Layers/OrchestrationEventStore.ts"; +import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts"; +import { + OrchestrationEventStore, + type OrchestrationEventStoreShape, +} from "../../persistence/Services/OrchestrationEventStore.ts"; +import { OrchestrationEngineLive } from "./OrchestrationEngine.ts"; +import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts"; +import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts"; +import { + OrchestrationProjectionPipeline, + type OrchestrationProjectionPipelineShape, +} from "../Services/ProjectionPipeline.ts"; + +const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value); +const asMessageId = (value: string): MessageId => MessageId.makeUnsafe(value); +const asTurnId = (value: string): TurnId => TurnId.makeUnsafe(value); +const asCheckpointRef = (value: string): CheckpointRef => CheckpointRef.makeUnsafe(value); + +async function createOrchestrationSystem() { + const orchestrationLayer = OrchestrationEngineLive.pipe( + Layer.provide(OrchestrationProjectionPipelineLive), + Layer.provide(OrchestrationEventStoreLive), + Layer.provide(OrchestrationCommandReceiptRepositoryLive), + Layer.provide(SqlitePersistenceMemory), + ); + const runtime = ManagedRuntime.make(orchestrationLayer); + const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService)); + return { + engine, + run: (effect: Effect.Effect) => runtime.runPromise(effect), + dispose: () => runtime.dispose(), + }; +} + +function now() { + return new Date().toISOString(); +} + +describe("OrchestrationEngine", () => { + it("returns deterministic read models for repeated reads", async () => { + const createdAt = now(); + const system = await createOrchestrationSystem(); + const { engine } = system; + + await system.run( + engine.dispatch({ + type: "project.create", + commandId: CommandId.makeUnsafe("cmd-project-1-create"), + projectId: asProjectId("project-1"), + title: "Project 1", + workspaceRoot: "/tmp/project-1", + defaultModel: "gpt-5-codex", + createdAt, + }), + ); + await system.run( + engine.dispatch({ + type: "thread.create", + commandId: CommandId.makeUnsafe("cmd-thread-1-create"), + threadId: ThreadId.makeUnsafe("thread-1"), + projectId: asProjectId("project-1"), + title: "Thread", + model: "gpt-5-codex", + branch: null, + worktreePath: null, + createdAt, + }), + ); + await system.run( + engine.dispatch({ + type: "thread.turn.start", + commandId: CommandId.makeUnsafe("cmd-turn-start-1"), + threadId: ThreadId.makeUnsafe("thread-1"), + message: { + messageId: asMessageId("msg-1"), + role: "user", + text: "hello", + attachments: [], + }, + createdAt, + }), + ); + + const readModelA = await system.run(engine.getReadModel()); + const readModelB = await system.run(engine.getReadModel()); + expect(readModelB).toEqual(readModelA); + await system.dispose(); + }); + + it("replays append-only events from sequence", async () => { + const system = await createOrchestrationSystem(); + const { engine } = system; + const createdAt = now(); + + await system.run( + engine.dispatch({ + type: "project.create", + commandId: CommandId.makeUnsafe("cmd-project-replay-create"), + projectId: asProjectId("project-replay"), + title: "Replay Project", + workspaceRoot: "/tmp/project-replay", + defaultModel: "gpt-5-codex", + createdAt, + }), + ); + await system.run( + engine.dispatch({ + type: "thread.create", + commandId: CommandId.makeUnsafe("cmd-thread-replay-create"), + threadId: ThreadId.makeUnsafe("thread-replay"), + projectId: asProjectId("project-replay"), + title: "replay", + model: "gpt-5-codex", + branch: null, + worktreePath: null, + createdAt, + }), + ); + await system.run( + engine.dispatch({ + type: "thread.delete", + commandId: CommandId.makeUnsafe("cmd-thread-replay-delete"), + threadId: ThreadId.makeUnsafe("thread-replay"), + }), + ); + + const events = await system.run( + Stream.runCollect(engine.readEvents(0)).pipe( + Effect.map((chunk): OrchestrationEvent[] => Array.from(chunk)), + ), + ); + expect(events.map((event) => event.type)).toEqual([ + "project.created", + "thread.created", + "thread.deleted", + ]); + await system.dispose(); + }); + + it("streams persisted domain events in order", async () => { + const system = await createOrchestrationSystem(); + const { engine } = system; + const createdAt = now(); + + await system.run( + engine.dispatch({ + type: "project.create", + commandId: CommandId.makeUnsafe("cmd-project-stream-create"), + projectId: asProjectId("project-stream"), + title: "Stream Project", + workspaceRoot: "/tmp/project-stream", + defaultModel: "gpt-5-codex", + createdAt, + }), + ); + + const eventTypes: string[] = []; + await system.run( + Effect.gen(function* () { + const eventQueue = yield* Queue.unbounded(); + yield* Effect.forkScoped( + Stream.take(engine.streamDomainEvents, 2).pipe( + Stream.runForEach((event) => Queue.offer(eventQueue, event).pipe(Effect.asVoid)), + ), + ); + yield* Effect.sleep("10 millis"); + yield* engine.dispatch({ + type: "thread.create", + commandId: CommandId.makeUnsafe("cmd-stream-thread-create"), + threadId: ThreadId.makeUnsafe("thread-stream"), + projectId: asProjectId("project-stream"), + title: "domain-stream", + model: "gpt-5-codex", + branch: null, + worktreePath: null, + createdAt, + }); + yield* engine.dispatch({ + type: "thread.meta.update", + commandId: CommandId.makeUnsafe("cmd-stream-thread-update"), + threadId: ThreadId.makeUnsafe("thread-stream"), + title: "domain-stream-updated", + }); + eventTypes.push((yield* Queue.take(eventQueue)).type); + eventTypes.push((yield* Queue.take(eventQueue)).type); + }).pipe(Effect.scoped), + ); + + expect(eventTypes).toEqual(["thread.created", "thread.meta-updated"]); + await system.dispose(); + }); + + it("stores completed checkpoint summaries even when no files changed", async () => { + const system = await createOrchestrationSystem(); + const { engine } = system; + const createdAt = now(); + + await system.run( + engine.dispatch({ + type: "project.create", + commandId: CommandId.makeUnsafe("cmd-project-turn-diff-create"), + projectId: asProjectId("project-turn-diff"), + title: "Turn Diff Project", + workspaceRoot: "/tmp/project-turn-diff", + defaultModel: "gpt-5-codex", + createdAt, + }), + ); + await system.run( + engine.dispatch({ + type: "thread.create", + commandId: CommandId.makeUnsafe("cmd-thread-turn-diff-create"), + threadId: ThreadId.makeUnsafe("thread-turn-diff"), + projectId: asProjectId("project-turn-diff"), + title: "Turn diff thread", + model: "gpt-5-codex", + branch: null, + worktreePath: null, + createdAt, + }), + ); + await system.run( + engine.dispatch({ + type: "thread.turn.diff.complete", + commandId: CommandId.makeUnsafe("cmd-turn-diff-complete"), + threadId: ThreadId.makeUnsafe("thread-turn-diff"), + turnId: asTurnId("turn-1"), + completedAt: createdAt, + checkpointRef: asCheckpointRef("refs/t3/checkpoints/thread-turn-diff/turn/1"), + status: "ready", + files: [], + checkpointTurnCount: 1, + createdAt, + }), + ); + + const thread = (await system.run(engine.getReadModel())).threads.find( + (entry) => entry.id === "thread-turn-diff", + ); + expect(thread?.checkpoints).toEqual([ + { + turnId: asTurnId("turn-1"), + checkpointTurnCount: 1, + checkpointRef: asCheckpointRef("refs/t3/checkpoints/thread-turn-diff/turn/1"), + status: "ready", + files: [], + assistantMessageId: null, + completedAt: createdAt, + }, + ]); + await system.dispose(); + }); + + it("keeps processing queued commands after a storage failure", async () => { + type StoredEvent = + ReturnType extends Effect.Effect + ? A + : never; + const events: StoredEvent[] = []; + let nextSequence = 1; + let shouldFailFirstAppend = true; + + const flakyStore: OrchestrationEventStoreShape = { + append(event) { + if (shouldFailFirstAppend && event.commandId === CommandId.makeUnsafe("cmd-flaky-1")) { + shouldFailFirstAppend = false; + return Effect.fail( + new PersistenceSqlError({ + operation: "test.append", + detail: "append failed", + }), + ); + } + const savedEvent = { + ...event, + sequence: nextSequence, + } as StoredEvent; + nextSequence += 1; + events.push(savedEvent); + return Effect.succeed(savedEvent); + }, + readFromSequence(sequenceExclusive) { + return Stream.fromIterable(events.filter((event) => event.sequence > sequenceExclusive)); + }, + readAll() { + return Stream.fromIterable(events); + }, + }; + + const runtime = ManagedRuntime.make( + OrchestrationEngineLive.pipe( + Layer.provide(OrchestrationProjectionPipelineLive), + Layer.provide(Layer.succeed(OrchestrationEventStore, flakyStore)), + Layer.provide(OrchestrationCommandReceiptRepositoryLive), + Layer.provide(SqlitePersistenceMemory), + ), + ); + const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService)); + const createdAt = now(); + + await runtime.runPromise( + engine.dispatch({ + type: "project.create", + commandId: CommandId.makeUnsafe("cmd-project-flaky-create"), + projectId: asProjectId("project-flaky"), + title: "Flaky Project", + workspaceRoot: "/tmp/project-flaky", + defaultModel: "gpt-5-codex", + createdAt, + }), + ); + + await expect( + runtime.runPromise( + engine.dispatch({ + type: "thread.create", + commandId: CommandId.makeUnsafe("cmd-flaky-1"), + threadId: ThreadId.makeUnsafe("thread-flaky-fail"), + projectId: asProjectId("project-flaky"), + title: "flaky-fail", + model: "gpt-5-codex", + branch: null, + worktreePath: null, + createdAt, + }), + ), + ).rejects.toThrow("append failed"); + + const result = await runtime.runPromise( + engine.dispatch({ + type: "thread.create", + commandId: CommandId.makeUnsafe("cmd-flaky-2"), + threadId: ThreadId.makeUnsafe("thread-flaky-ok"), + projectId: asProjectId("project-flaky"), + title: "flaky-ok", + model: "gpt-5-codex", + branch: null, + worktreePath: null, + createdAt, + }), + ); + + expect(result.sequence).toBe(2); + expect((await runtime.runPromise(engine.getReadModel())).snapshotSequence).toBe(2); + await runtime.dispose(); + }); + + it("rolls back all events for a multi-event command when projection fails mid-dispatch", async () => { + let shouldFailRequestedProjection = true; + const flakyProjectionPipeline: OrchestrationProjectionPipelineShape = { + bootstrap: Effect.void, + projectEvent: (event) => { + if ( + shouldFailRequestedProjection && + event.commandId === CommandId.makeUnsafe("cmd-turn-start-atomic") && + event.type === "thread.turn-start-requested" + ) { + shouldFailRequestedProjection = false; + return Effect.fail( + new PersistenceSqlError({ + operation: "test.projection", + detail: "projection failed", + }), + ); + } + return Effect.void; + }, + }; + + const runtime = ManagedRuntime.make( + OrchestrationEngineLive.pipe( + Layer.provide(Layer.succeed(OrchestrationProjectionPipeline, flakyProjectionPipeline)), + Layer.provide(OrchestrationEventStoreLive), + Layer.provide(OrchestrationCommandReceiptRepositoryLive), + Layer.provide(SqlitePersistenceMemory), + ), + ); + const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService)); + const createdAt = now(); + + await runtime.runPromise( + engine.dispatch({ + type: "project.create", + commandId: CommandId.makeUnsafe("cmd-project-atomic-create"), + projectId: asProjectId("project-atomic"), + title: "Atomic Project", + workspaceRoot: "/tmp/project-atomic", + defaultModel: "gpt-5-codex", + createdAt, + }), + ); + await runtime.runPromise( + engine.dispatch({ + type: "thread.create", + commandId: CommandId.makeUnsafe("cmd-thread-atomic-create"), + threadId: ThreadId.makeUnsafe("thread-atomic"), + projectId: asProjectId("project-atomic"), + title: "atomic", + model: "gpt-5-codex", + branch: null, + worktreePath: null, + createdAt, + }), + ); + + const turnStartCommand = { + type: "thread.turn.start" as const, + commandId: CommandId.makeUnsafe("cmd-turn-start-atomic"), + threadId: ThreadId.makeUnsafe("thread-atomic"), + message: { + messageId: asMessageId("msg-atomic-1"), + role: "user" as const, + text: "hello", + attachments: [], + }, + createdAt, + }; + + await expect(runtime.runPromise(engine.dispatch(turnStartCommand))).rejects.toThrow( + "projection failed", + ); + + const eventsAfterFailure = await runtime.runPromise( + Stream.runCollect(engine.readEvents(0)).pipe( + Effect.map((chunk): OrchestrationEvent[] => Array.from(chunk)), + ), + ); + expect(eventsAfterFailure.map((event) => event.type)).toEqual([ + "project.created", + "thread.created", + ]); + expect((await runtime.runPromise(engine.getReadModel())).snapshotSequence).toBe(2); + + const retryResult = await runtime.runPromise(engine.dispatch(turnStartCommand)); + expect(retryResult.sequence).toBe(4); + + const eventsAfterRetry = await runtime.runPromise( + Stream.runCollect(engine.readEvents(0)).pipe( + Effect.map((chunk): OrchestrationEvent[] => Array.from(chunk)), + ), + ); + expect(eventsAfterRetry.map((event) => event.type)).toEqual([ + "project.created", + "thread.created", + "thread.message-sent", + "thread.turn-start-requested", + ]); + expect( + eventsAfterRetry.filter((event) => event.commandId === turnStartCommand.commandId), + ).toHaveLength(2); + + await runtime.dispose(); + }); + + it("fails command dispatch when command invariants are violated", async () => { + const system = await createOrchestrationSystem(); + const { engine } = system; + + await expect( + system.run( + engine.dispatch({ + type: "thread.turn.start", + commandId: CommandId.makeUnsafe("cmd-invariant-missing-thread"), + threadId: ThreadId.makeUnsafe("thread-missing"), + message: { + messageId: asMessageId("msg-missing"), + role: "user", + text: "hello", + attachments: [], + }, + createdAt: now(), + }), + ), + ).rejects.toThrow("Thread 'thread-missing' does not exist"); + + await system.dispose(); + }); + + it("rejects duplicate thread creation", async () => { + const system = await createOrchestrationSystem(); + const { engine } = system; + const createdAt = now(); + + await system.run( + engine.dispatch({ + type: "project.create", + commandId: CommandId.makeUnsafe("cmd-project-duplicate-create"), + projectId: asProjectId("project-duplicate"), + title: "Duplicate Project", + workspaceRoot: "/tmp/project-duplicate", + defaultModel: "gpt-5-codex", + createdAt, + }), + ); + + await system.run( + engine.dispatch({ + type: "thread.create", + commandId: CommandId.makeUnsafe("cmd-thread-duplicate-1"), + threadId: ThreadId.makeUnsafe("thread-duplicate"), + projectId: asProjectId("project-duplicate"), + title: "duplicate", + model: "gpt-5-codex", + branch: null, + worktreePath: null, + createdAt, + }), + ); + + await expect( + system.run( + engine.dispatch({ + type: "thread.create", + commandId: CommandId.makeUnsafe("cmd-thread-duplicate-2"), + threadId: ThreadId.makeUnsafe("thread-duplicate"), + projectId: asProjectId("project-duplicate"), + title: "duplicate", + model: "gpt-5-codex", + branch: null, + worktreePath: null, + createdAt, + }), + ), + ).rejects.toThrow("already exists"); + + await system.dispose(); + }); +}); diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.ts index 52734f063fb41b7d1f5ff312a5222cfd5b1a00c7..ebe5547700892ad3deb86eacfaf83af5990f4a5e 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationEngine.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.ts @@ -6,7 +6,9 @@ import type { } from "@t3tools/contracts"; import { OrchestrationCommand } from "@t3tools/contracts"; import { Deferred, Effect, Layer, Option, PubSub, Queue, Stream } from "effect"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; +import { toPersistenceSqlError } from "../../persistence/Errors.ts"; import { OrchestrationEventStore } from "../../persistence/Services/OrchestrationEventStore.ts"; import { OrchestrationCommandReceiptRepository } from "../../persistence/Services/OrchestrationCommandReceipts.ts"; import { @@ -55,6 +57,7 @@ function formatDispatchError(error: OrchestrationDispatchError): string { } const makeOrchestrationEngine = Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; const eventStore = yield* OrchestrationEventStore; const commandReceiptRepository = yield* OrchestrationCommandReceiptRepository; const projectionPipeline = yield* OrchestrationProjectionPipeline; @@ -91,36 +94,57 @@ const makeOrchestrationEngine = Effect.gen(function* () { readModel, }); const eventBases = Array.isArray(eventBase) ? eventBase : [eventBase]; - let lastSavedEvent = null as OrchestrationEvent | null; + const committedCommand = yield* sql + .withTransaction( + Effect.gen(function* () { + const committedEvents: OrchestrationEvent[] = []; + let nextReadModel = readModel; - for (const nextEvent of eventBases) { - const savedEvent = yield* eventStore.append(nextEvent); - yield* projectionPipeline.projectEvent(savedEvent); - readModel = yield* projectEvent(readModel, savedEvent); - yield* PubSub.publish(eventPubSub, savedEvent); - lastSavedEvent = savedEvent; - } + for (const nextEvent of eventBases) { + const savedEvent = yield* eventStore.append(nextEvent); + yield* projectionPipeline.projectEvent(savedEvent); + nextReadModel = yield* projectEvent(nextReadModel, savedEvent); + committedEvents.push(savedEvent); + } - if (lastSavedEvent === null) { - return yield* Deferred.fail( - envelope.result, - new OrchestrationCommandInvariantError({ - commandType: envelope.command.type, - detail: "Command produced no events.", + const lastSavedEvent = committedEvents.at(-1) ?? null; + if (lastSavedEvent === null) { + return yield* new OrchestrationCommandInvariantError({ + commandType: envelope.command.type, + detail: "Command produced no events.", + }); + } + + yield* commandReceiptRepository.upsert({ + commandId: envelope.command.commandId, + aggregateKind: lastSavedEvent.aggregateKind, + aggregateId: lastSavedEvent.aggregateId, + acceptedAt: lastSavedEvent.occurredAt, + resultSequence: lastSavedEvent.sequence, + status: "accepted", + error: null, + }); + + return { + committedEvents, + lastSequence: lastSavedEvent.sequence, + nextReadModel, + } as const; }), + ) + .pipe( + Effect.catchTag("SqlError", (sqlError) => + Effect.fail( + toPersistenceSqlError("OrchestrationEngine.processEnvelope:transaction")(sqlError), + ), + ), ); - } - yield* commandReceiptRepository.upsert({ - commandId: envelope.command.commandId, - aggregateKind: lastSavedEvent.aggregateKind, - aggregateId: lastSavedEvent.aggregateId, - acceptedAt: lastSavedEvent.occurredAt, - resultSequence: lastSavedEvent.sequence, - status: "accepted", - error: null, - }); - yield* Deferred.succeed(envelope.result, { sequence: lastSavedEvent.sequence }); + readModel = committedCommand.nextReadModel; + for (const event of committedCommand.committedEvents) { + yield* PubSub.publish(eventPubSub, event); + } + yield* Deferred.succeed(envelope.result, { sequence: committedCommand.lastSequence }); }).pipe( Effect.catch((error) => Effect.gen(function* () {