From 7934639aa1379f745af7db7c05c964bf6d7a7cc8 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 23 Feb 2026 10:33:45 -0800 Subject: [PATCH] es terminology --- ...er-authoritative-event-sourcing-cleanup.md | 206 ++++++++++++++++++ apps/server/src/orchestration/Errors.ts | 32 ++- apps/server/src/orchestration/Layer.test.ts | 111 ++++++++-- .../Layers/OrchestrationEngine.ts | 110 ++++++++++ .../OrchestrationEngine.ts} | 20 +- .../orchestration/commandInvariants.test.ts | 161 ++++++++++++++ .../src/orchestration/commandInvariants.ts | 80 +++++++ .../orchestration/{Layer.ts => decider.ts} | 173 ++++++--------- .../{reducer.test.ts => projector.test.ts} | 24 +- .../{reducer.ts => projector.ts} | 12 +- apps/server/src/persistence/Errors.ts | 2 +- ...est.ts => OrchestrationEventStore.test.ts} | 18 +- ...onEvents.ts => OrchestrationEventStore.ts} | 86 +++++--- ...onEvents.ts => OrchestrationEventStore.ts} | 36 +-- apps/server/src/wsServer.ts | 41 ++-- 15 files changed, 879 insertions(+), 233 deletions(-) create mode 100644 .plans/14-server-authoritative-event-sourcing-cleanup.md create mode 100644 apps/server/src/orchestration/Layers/OrchestrationEngine.ts rename apps/server/src/orchestration/{Service.ts => Services/OrchestrationEngine.ts} (79%) create mode 100644 apps/server/src/orchestration/commandInvariants.test.ts create mode 100644 apps/server/src/orchestration/commandInvariants.ts rename apps/server/src/orchestration/{Layer.ts => decider.ts} (55%) rename apps/server/src/orchestration/{reducer.test.ts => projector.test.ts} (87%) rename apps/server/src/orchestration/{reducer.ts => projector.ts} (97%) rename apps/server/src/persistence/Layers/{OrchestrationEvents.test.ts => OrchestrationEventStore.test.ts} (56%) rename apps/server/src/persistence/Layers/{OrchestrationEvents.ts => OrchestrationEventStore.ts} (66%) rename apps/server/src/persistence/Services/{OrchestrationEvents.ts => OrchestrationEventStore.ts} (50%) diff --git a/.plans/14-server-authoritative-event-sourcing-cleanup.md b/.plans/14-server-authoritative-event-sourcing-cleanup.md new file mode 100644 index 0000000000000000000000000000000000000000..e3a125b819fa3fc4028e607c796c939da9822f7c --- /dev/null +++ b/.plans/14-server-authoritative-event-sourcing-cleanup.md @@ -0,0 +1,206 @@ +# Server-Authoritative Event-Sourcing Cleanup Plan + +Goal: +- Move to a cleaner service architecture with: + - durable, server-authoritative event sourcing + - strict command routing/validation + - pluggable provider adapters + - explicit separation between transport, domain orchestration, provider runtime, and persistence + +## Target Service Graph (ASCII) + +```text + +---------------------------+ + | wsServer | + | transport | + +---------------------------+ + | orchestration.dispatchCommand + v + +-------------------------------------------+ + | OrchestrationCommandRouter | + +-------------------------------------------+ + | + v + +-------------------------------------------+ + | OrchestrationCommandHandlers | + +-------------------------------------------+ + | + v + +-------------------------------------------+ + | OrchestrationEventStore | + +-------------------------------------------+ + | + v + +-------------------------------------------+ + | OrchestrationProjectionService | + +-------------------------------------------+ + | snapshot/replay + +---------------------------> wsServer + + +wsServer -- providers.* RPC --> +---------------------------+ + | ProviderService | + +---------------------------+ + | | + v v + +-------------------+ +-------------------------+ + | ProviderSession | | ProviderAdapterRegistry | + | Registry (durable)| +-------------------------+ + +-------------------+ | + ^ v + | +-------------------------+ + | | ProviderAdapter(s) | + | +-------------------------+ + | | + | runtime events v + | +---------------------------+ + +----------| ProviderRuntimeIngestion | + +---------------------------+ + | | | + v v v + Router Session Checkpoint + Registry Service + + +-------------------------------------------+ + | CheckpointService | + +-------------------------------------------+ + | | | + v v v + +--------------------+ +-------------+ +-------------------+ + | CheckpointCatalog | | Checkpoint | | ProviderAdapter(s)| + | (durable) | | Store (git) | | (read/rollback) | + +--------------------+ +-------------+ +-------------------+ + | + v + +------+ + |SQLite| + +------+ + +OrchestrationEventStore ------> SQLite +OrchestrationProjectionService -> SQLite +ProviderSessionRegistry ------> SQLite +CheckpointCatalog ------> SQLite +``` + +## Commit Series + +### Commit 1: Split public vs system orchestration command contracts +- Create separate schemas/types: + - `ClientOrchestrationCommandSchema` + - `SystemOrchestrationCommandSchema` + - `OrchestrationCommandSchema = union(client, system)` +- Ensure client transport can only submit client commands. +- Keep system commands for server-internal workflows only. +- Expected files: + - `packages/contracts/src/orchestration.ts` + - `apps/server/src/wsServer.ts` + - orchestration/service tests +- Tests: + - reject system-only command via WS dispatch path + - preserve internal dispatch functionality for system commands + +### Commit 2: Introduce `OrchestrationCommandRouter` + handler boundary +- Add dedicated router service to validate, authorize, and route commands. +- Move command-to-event mapping out of `orchestration/Layer.ts` into handlers. +- Add aggregate-level invariant checks before append (thread exists, project exists, etc.). +- Expected files: + - `apps/server/src/orchestration/Services/CommandRouter.ts` (new) + - `apps/server/src/orchestration/Layers/CommandRouter.ts` (new) + - `apps/server/src/orchestration/Layer.ts` + - `apps/server/src/orchestration/reducer.ts` (only if needed for event payload changes) +- Tests: + - router validation and invariant failures + - handler happy-path tests per command type + +### Commit 3: Harden event store for idempotency + optimistic append metadata +- Add DB-level idempotency guard for `command_id` (`UNIQUE` where non-null). +- Extend append API to support idempotent replays and deterministic return of prior event on duplicate `commandId`. +- Add optional aggregate version metadata for future optimistic concurrency. +- Expected files: + - `apps/server/src/persistence/Migrations/00x_*.ts` (new migration) + - `apps/server/src/persistence/Services/OrchestrationEvents.ts` + - `apps/server/src/persistence/Layers/OrchestrationEvents.ts` +- Tests: + - duplicate command ID append returns same event/sequence (or explicit idempotent behavior) + - concurrent append behavior stays ordered and deterministic + +### Commit 4: Extract provider-runtime -> orchestration bridge from `wsServer` +- Create `ProviderRuntimeIngestionService` that: + - subscribes to `ProviderService.streamEvents` + - translates runtime events into orchestration commands + - dispatches through router/engine +- Remove provider-to-orchestration state mutation logic from `wsServer`. +- Expected files: + - `apps/server/src/orchestration/Services/ProviderRuntimeIngestion.ts` (new) + - `apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts` (new) + - `apps/server/src/wsServer.ts` +- Tests: + - ingestion service mapping tests (turn started/completed, message delta/completed, runtime error) + - ws integration confirms same external push behavior + +### Commit 5: Make session directory durable (`ProviderSessionRegistry`) +- Replace in-memory-only `ProviderSessionDirectoryLive` with persistence-backed registry. +- Keep in-memory cache optional, but source of truth must be persistent. +- Add startup reconciliation to prune dead sessions / keep known thread mapping. +- Expected files: + - `apps/server/src/provider/Services/ProviderSessionDirectory.ts` (or new SessionRegistry service) + - `apps/server/src/provider/Layers/ProviderSessionDirectory.ts` + - `apps/server/src/persistence/Migrations/00x_*.ts` (new table/indexes) + - provider persistence tests +- Tests: + - survives server restart with correct mapping + - stale session cleanup semantics + +### Commit 6: Re-key checkpoint metadata from session to thread identity +- Change checkpoint catalog primary identity from `provider_session_id` to durable `thread_id`. +- Keep `session_id` as nullable metadata only. +- Update checkpoint flows (`initialize`, `capture`, `list`, `diff`, `revert`) to use thread identity. +- Expected files: + - `apps/server/src/persistence/Migrations/00x_*.ts` (checkpoint schema migration) + - `apps/server/src/persistence/Services/Checkpoints.ts` + - `apps/server/src/persistence/Layers/Checkpoints.ts` + - `apps/server/src/checkpointing/Layers/CheckpointService.ts` +- Tests: + - resume/new session over same thread sees same checkpoint history + - revert/diff still work after session churn + +### Commit 7: Add durable projection persistence for orchestration read models +- Introduce projection tables/snapshots persisted in DB to avoid full replay dependency. +- Keep event stream as source of truth; projection rebuild stays deterministic. +- `getSnapshot` reads from projection store (memory cache optional). +- Expected files: + - `apps/server/src/persistence/Migrations/00x_*.ts` (projection tables) + - `apps/server/src/orchestration/*` projection service/layer + - `apps/server/src/wsServer.ts` (snapshot/replay path wiring) +- Tests: + - cold boot snapshot load without replaying full history in process + - projection rebuild from events yields same result as previous reducer semantics + +### Commit 8: Narrow `ProviderService` responsibilities +- Keep `ProviderService` focused on provider RPC/session lifecycle + unified runtime stream. +- Move checkpoint-capture side effects out of provider event worker into dedicated ingestion/checkpoint pipeline service. +- Preserve adapter pluggability and provider-neutral contracts. +- Expected files: + - `apps/server/src/provider/Layers/ProviderService.ts` + - new orchestration/checkpoint runtime coordinator service(s) +- Tests: + - provider service routing stays intact + - checkpoint capture still triggered by turn completion through new coordinator + +### Commit 9: Remove dead legacy path and finalize docs +- Remove unused legacy manager/store path from active architecture: + - `providerManager.ts` + - `filesystemCheckpointStore.ts` (if no longer needed by tests/tools) +- Update architecture docs with final service boundaries and boot/runtime graph. +- Expected files: + - legacy files + references + - `AGENTS.md`/docs as needed + - `.plans` docs linkage +- Tests: + - full server integration suite passes on Effect-only path + - no regressions in WS protocol behavior + +## Risk Controls +- Keep WS method names and payload contracts stable throughout. +- Gate each commit with targeted integration tests before moving forward. +- Avoid broad event-type churn in one step; migrate schemas incrementally with clear compatibility windows. diff --git a/apps/server/src/orchestration/Errors.ts b/apps/server/src/orchestration/Errors.ts index e7b4d8c0dd7b63ac9fd3e3a87b02ebf23c0359af..35091b3eb33b1541d6b4d80b5695492561e10fc8 100644 --- a/apps/server/src/orchestration/Errors.ts +++ b/apps/server/src/orchestration/Errors.ts @@ -1,6 +1,6 @@ import { SchemaIssue, Schema } from "effect"; -import type { OrchestrationEventRepositoryError } from "../persistence/Errors.ts"; +import type { OrchestrationEventStoreError } from "../persistence/Errors.ts"; export class OrchestrationCommandJsonParseError extends Schema.TaggedErrorClass()( "OrchestrationCommandJsonParseError", @@ -26,8 +26,21 @@ export class OrchestrationCommandDecodeError extends Schema.TaggedErrorClass()( - "OrchestrationReducerDecodeError", +export class OrchestrationCommandInvariantError extends Schema.TaggedErrorClass()( + "OrchestrationCommandInvariantError", + { + commandType: Schema.String, + detail: Schema.String, + cause: Schema.optional(Schema.Defect), + }, +) { + override get message(): string { + return `Orchestration command invariant failed (${this.commandType}): ${this.detail}`; + } +} + +export class OrchestrationProjectorDecodeError extends Schema.TaggedErrorClass()( + "OrchestrationProjectorDecodeError", { eventType: Schema.String, issue: Schema.String, @@ -35,7 +48,7 @@ export class OrchestrationReducerDecodeError extends Schema.TaggedErrorClass - new OrchestrationReducerDecodeError({ +export function toProjectorDecodeError(eventType: string) { + return (error: Schema.SchemaError): OrchestrationProjectorDecodeError => + new OrchestrationProjectorDecodeError({ eventType, issue: SchemaIssue.makeFormatterDefault()(error.issue), cause: error, diff --git a/apps/server/src/orchestration/Layer.test.ts b/apps/server/src/orchestration/Layer.test.ts index e82811c12056686e7fbfdb341aa6bb9266f24033..af5da282aea437e2f162076306122201b1bddc78 100644 --- a/apps/server/src/orchestration/Layer.test.ts +++ b/apps/server/src/orchestration/Layer.test.ts @@ -3,18 +3,18 @@ import { Effect, Layer, ManagedRuntime, Queue, Stream } from "effect"; import { describe, expect, it } from "vitest"; import { - OrchestrationEventRepository, - type OrchestrationEventRepositoryShape, -} from "../persistence/Services/OrchestrationEvents.ts"; + OrchestrationEventStore, + type OrchestrationEventStoreShape, +} from "../persistence/Services/OrchestrationEventStore.ts"; import { PersistenceSqlError } from "../persistence/Errors.ts"; import { SqlitePersistenceMemory } from "../persistence/Layers/Sqlite.ts"; -import { OrchestrationEngineLive } from "./Layer.ts"; -import { OrchestrationEngineService } from "./Service.ts"; -import { OrchestrationEventRepositoryLive } from "../persistence/Layers/OrchestrationEvents.ts"; +import { OrchestrationEngineLive } from "./Layers/OrchestrationEngine.ts"; +import { OrchestrationEngineService } from "./Services/OrchestrationEngine.ts"; +import { OrchestrationEventStoreLive } from "../persistence/Layers/OrchestrationEventStore.ts"; export async function createOrchestrationSystem() { const orchestrationLayer = OrchestrationEngineLive.pipe( - Layer.provide(OrchestrationEventRepositoryLive), + Layer.provide(OrchestrationEventStoreLive), Layer.provide(SqlitePersistenceMemory), ); const runtime = ManagedRuntime.make(orchestrationLayer); @@ -27,7 +27,7 @@ export async function createOrchestrationSystem() { } describe("OrchestrationEngine", () => { - it("returns deterministic snapshots for repeated reads", async () => { + it("returns deterministic read models for repeated reads", async () => { const createdAt = new Date().toISOString(); const projectId = "project-1"; const threadId = "thread-1"; @@ -59,9 +59,9 @@ describe("OrchestrationEngine", () => { createdAt, }), ); - const snapshotA = await system.run(engine.getSnapshot()); - const snapshotB = await system.run(engine.getSnapshot()); - expect(snapshotB).toEqual(snapshotA); + const readModelA = await system.run(engine.getReadModel()); + const readModelB = await system.run(engine.getReadModel()); + expect(readModelB).toEqual(readModelA); await system.dispose(); }); @@ -89,7 +89,11 @@ describe("OrchestrationEngine", () => { createdAt: new Date().toISOString(), }), ); - const events = await system.run(engine.replayEvents(0)); + const events = await system.run( + Stream.runCollect(engine.readEvents(0)).pipe( + Effect.map((chunk): OrchestrationEvent[] => Array.from(chunk)), + ), + ); expect(events.length).toBe(2); expect(events[0]?.type).toBe("thread.created"); expect(events[1]?.type).toBe("thread.deleted"); @@ -167,7 +171,7 @@ describe("OrchestrationEngine", () => { }), ); - const firstThread = (await system.run(engine.getSnapshot())).threads.find( + const firstThread = (await system.run(engine.getReadModel())).threads.find( (thread) => thread.id === "thread-turn-diff", ); expect(firstThread?.turnDiffSummaries).toEqual([ @@ -289,7 +293,7 @@ describe("OrchestrationEngine", () => { }), ); - const thread = (await system.run(engine.getSnapshot())).threads.find( + const thread = (await system.run(engine.getReadModel())).threads.find( (entry) => entry.id === "thread-revert", ); expect(thread?.messages.map((message) => message.id)).toEqual(["user-1", "assistant:turn-1"]); @@ -299,20 +303,20 @@ describe("OrchestrationEngine", () => { }); it("allows stop to be called multiple times", async () => { - const inMemoryStore: OrchestrationEventRepositoryShape = { + const inMemoryStore: OrchestrationEventStoreShape = { append(event) { return Effect.succeed({ ...event, sequence: 1 }); }, readFromSequence() { - return Effect.succeed([]); + return Stream.empty; }, readAll() { - return Effect.succeed([]); + return Stream.empty; }, }; const runtime = ManagedRuntime.make( OrchestrationEngineLive.pipe( - Layer.provide(Layer.succeed(OrchestrationEventRepository, inMemoryStore)), + Layer.provide(Layer.succeed(OrchestrationEventStore, inMemoryStore)), ), ); await runtime.runPromise(Effect.service(OrchestrationEngineService)); @@ -325,7 +329,7 @@ describe("OrchestrationEngine", () => { let nextSequence = 1; let shouldFailFirstAppend = true; - const flakyStore: OrchestrationEventRepositoryShape = { + const flakyStore: OrchestrationEventStoreShape = { append(event) { if (shouldFailFirstAppend) { shouldFailFirstAppend = false; @@ -345,16 +349,16 @@ describe("OrchestrationEngine", () => { return Effect.succeed(savedEvent); }, readFromSequence(sequenceExclusive) { - return Effect.succeed(events.filter((event) => event.sequence > sequenceExclusive)); + return Stream.fromIterable(events.filter((event) => event.sequence > sequenceExclusive)); }, readAll() { - return Effect.succeed(events); + return Stream.fromIterable(events); }, }; const runtime = ManagedRuntime.make( OrchestrationEngineLive.pipe( - Layer.provide(Layer.succeed(OrchestrationEventRepository, flakyStore)), + Layer.provide(Layer.succeed(OrchestrationEventStore, flakyStore)), ), ); const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService)); @@ -391,8 +395,69 @@ describe("OrchestrationEngine", () => { ); expect(result.sequence).toBe(1); - expect((await runtime.runPromise(engine.getSnapshot())).sequence).toBe(1); + expect((await runtime.runPromise(engine.getReadModel())).sequence).toBe(1); await runtime.dispose(); }); + + it("fails command dispatch when command invariants are violated", async () => { + const system = await createOrchestrationSystem(); + const engine = system.engine; + const createdAt = new Date().toISOString(); + + await expect( + system.run( + engine.dispatch({ + type: "message.send", + commandId: "cmd-invariant-missing-thread", + threadId: "thread-missing", + messageId: "msg-missing", + role: "user", + text: "hello", + streaming: false, + createdAt, + }), + ), + ).rejects.toThrow("Thread 'thread-missing' does not exist"); + + await system.dispose(); + }); + + it("rejects duplicate thread creation", async () => { + const system = await createOrchestrationSystem(); + const engine = system.engine; + const createdAt = new Date().toISOString(); + + await system.run( + engine.dispatch({ + type: "thread.create", + commandId: "cmd-thread-duplicate-1", + threadId: "thread-duplicate", + projectId: "project-duplicate", + title: "duplicate", + model: "gpt-5-codex", + branch: null, + worktreePath: null, + createdAt, + }), + ); + + await expect( + system.run( + engine.dispatch({ + type: "thread.create", + commandId: "cmd-thread-duplicate-2", + threadId: "thread-duplicate", + projectId: "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 new file mode 100644 index 0000000000000000000000000000000000000000..8dcc809fe081ca84c522faf09debf8c9fc353c46 --- /dev/null +++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.ts @@ -0,0 +1,110 @@ +import type { + OrchestrationCommand, + OrchestrationEvent, + OrchestrationReadModel, +} from "@t3tools/contracts"; +import { OrchestrationCommandSchema } from "@t3tools/contracts"; +import { Deferred, Effect, Layer, PubSub, Queue, Schema, Stream } from "effect"; + +import { createLogger } from "../../logger.ts"; +import { OrchestrationEventStore } from "../../persistence/Services/OrchestrationEventStore.ts"; +import { + toOrchestrationCommandDecodeError, + toOrchestrationJsonParseError, + type OrchestrationDispatchError, +} from "../Errors.ts"; +import { decideOrchestrationCommand } from "../decider.ts"; +import { createEmptyReadModel, projectEvent } from "../projector.ts"; +import { + OrchestrationEngineService, + type OrchestrationEngineShape, +} from "../Services/OrchestrationEngine.ts"; + +interface CommandEnvelope { + command: OrchestrationCommand; + result: Deferred.Deferred<{ sequence: number }, OrchestrationDispatchError>; +} + +const decodeUnknownCommand = Schema.decodeUnknownEffect(OrchestrationCommandSchema); + +const makeOrchestrationEngine = Effect.gen(function* () { + const logger = createLogger("orchestration"); + const eventStore = yield* OrchestrationEventStore; + + let readModel = createEmptyReadModel(new Date().toISOString()); + + const commandQueue = yield* Queue.unbounded(); + const eventPubSub = yield* PubSub.unbounded(); + + const processEnvelope = (envelope: CommandEnvelope): Effect.Effect => + Effect.gen(function* () { + const eventBase = yield* decideOrchestrationCommand({ + command: envelope.command, + readModel, + }); + const savedEvent = yield* eventStore.append(eventBase); + readModel = yield* projectEvent(readModel, savedEvent); + yield* PubSub.publish(eventPubSub, savedEvent); + + yield* Deferred.succeed(envelope.result, { sequence: savedEvent.sequence }); + }).pipe(Effect.catch((error) => Deferred.fail(envelope.result, error).pipe(Effect.asVoid))); + + // bootstrap read model from event store + yield* Stream.runForEach(eventStore.readAll(), (event) => + Effect.gen(function* () { + readModel = yield* projectEvent(readModel, event); + }), + ); + + const worker = Effect.forever(Queue.take(commandQueue).pipe(Effect.flatMap(processEnvelope))); + yield* Effect.forkScoped(worker); + logger.info("orchestration engine started", { + sequence: readModel.sequence, + }); + + const getReadModel: OrchestrationEngineShape["getReadModel"] = () => + Effect.sync((): OrchestrationReadModel => readModel); + + const readEvents: OrchestrationEngineShape["readEvents"] = (fromSequenceExclusive) => + eventStore.readFromSequence(fromSequenceExclusive); + + const dispatch: OrchestrationEngineShape["dispatch"] = (command) => + Effect.gen(function* () { + const result = yield* Deferred.make<{ sequence: number }, OrchestrationDispatchError>(); + yield* Queue.offer(commandQueue, { command, result }); + return yield* Deferred.await(result); + }); + + const dispatchUnknownCommand: OrchestrationEngineShape["dispatchUnknownCommand"] = (command) => + Effect.gen(function* () { + const payload = + typeof command === "string" + ? yield* Effect.try({ + try: () => JSON.parse(command) as unknown, + catch: toOrchestrationJsonParseError, + }) + : command; + + const decoded = yield* decodeUnknownCommand(payload).pipe( + Effect.mapError(toOrchestrationCommandDecodeError), + ); + + return yield* dispatch(decoded); + }); + + const streamDomainEvents: OrchestrationEngineShape["streamDomainEvents"] = + Stream.fromPubSub(eventPubSub); + + return { + getReadModel, + readEvents, + dispatchUnknownCommand, + dispatch, + streamDomainEvents, + } satisfies OrchestrationEngineShape; +}); + +export const OrchestrationEngineLive = Layer.effect( + OrchestrationEngineService, + makeOrchestrationEngine, +); diff --git a/apps/server/src/orchestration/Service.ts b/apps/server/src/orchestration/Services/OrchestrationEngine.ts similarity index 79% rename from apps/server/src/orchestration/Service.ts rename to apps/server/src/orchestration/Services/OrchestrationEngine.ts index bffd33e50823b42f291123199e772bbae36d7906..2287f2ae7f1686f81301c27e905f019b2c170072 100644 --- a/apps/server/src/orchestration/Service.ts +++ b/apps/server/src/orchestration/Services/OrchestrationEngine.ts @@ -2,7 +2,7 @@ * OrchestrationEngineService - Service interface for orchestration command handling. * * Owns command validation/dispatch and in-memory read-model updates backed by - * `OrchestrationEventRepository` persistence. It does not own provider process + * `OrchestrationEventStore` persistence. It does not own provider process * management or transport concerns (e.g. websocket request parsing). * * Uses Effect `ServiceMap.Service` for dependency injection. Command dispatch, @@ -18,26 +18,26 @@ import type { import { ServiceMap } from "effect"; import type { Effect, Stream } from "effect"; -import type { OrchestrationDispatchError, OrchestrationEngineError } from "./Errors.ts"; -import type { OrchestrationEventRepositoryError } from "../persistence/Errors.ts"; +import type { OrchestrationDispatchError, OrchestrationEngineError } from "../Errors.ts"; +import type { OrchestrationEventStoreError } from "../../persistence/Errors.ts"; export interface OrchestrationEngineShape { /** - * Read the current in-memory orchestration snapshot. + * Read the current in-memory orchestration read model. * * @returns Effect containing the latest read model. */ - readonly getSnapshot: () => Effect.Effect; + readonly getReadModel: () => Effect.Effect; /** * Replay persisted orchestration events from an exclusive sequence cursor. * * @param fromSequenceExclusive - Sequence cursor (exclusive). - * @returns Effect containing ordered events. + * @returns Stream containing ordered events. */ - readonly replayEvents: ( + readonly readEvents: ( fromSequenceExclusive: number, - ) => Effect.Effect; + ) => Stream.Stream; /** * Validate and dispatch an unknown command payload. @@ -47,7 +47,7 @@ export interface OrchestrationEngineShape { * @param command - Unknown command payload. * @returns Effect containing the sequence of the persisted event. */ - readonly dispatchUnknown: ( + readonly dispatchUnknownCommand: ( command: unknown, ) => Effect.Effect<{ sequence: number }, OrchestrationEngineError, never>; @@ -74,7 +74,7 @@ export interface OrchestrationEngineShape { * ```ts * const program = Effect.gen(function* () { * const engine = yield* OrchestrationEngineService - * return yield* engine.getSnapshot() + * return yield* engine.getReadModel() * }) * ``` */ diff --git a/apps/server/src/orchestration/commandInvariants.test.ts b/apps/server/src/orchestration/commandInvariants.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..7ed1c7f84887cc335600525e0cc0ffcd407334da --- /dev/null +++ b/apps/server/src/orchestration/commandInvariants.test.ts @@ -0,0 +1,161 @@ +import { describe, expect, it } from "vitest"; +import type { OrchestrationCommand, OrchestrationReadModel } from "@t3tools/contracts"; +import { Effect } from "effect"; + +import { + findThreadById, + listThreadsByProjectId, + requireNonNegativeInteger, + requireThread, + requireThreadAbsent, +} from "./commandInvariants.ts"; + +const now = new Date().toISOString(); + +const readModel: OrchestrationReadModel = { + sequence: 2, + updatedAt: now, + gitByProjectId: {}, + threads: [ + { + id: "thread-1", + projectId: "project-a", + title: "Thread A", + model: "gpt-5-codex", + branch: null, + worktreePath: null, + createdAt: now, + updatedAt: now, + latestTurnId: null, + latestTurnStartedAt: null, + latestTurnCompletedAt: null, + latestTurnDurationMs: null, + messages: [], + session: null, + turnDiffSummaries: [], + activities: [], + error: null, + }, + { + id: "thread-2", + projectId: "project-b", + title: "Thread B", + model: "gpt-5-codex", + branch: null, + worktreePath: null, + createdAt: now, + updatedAt: now, + latestTurnId: null, + latestTurnStartedAt: null, + latestTurnCompletedAt: null, + latestTurnDurationMs: null, + messages: [], + session: null, + turnDiffSummaries: [], + activities: [], + error: null, + }, + ], +}; + +const messageSendCommand: OrchestrationCommand = { + type: "message.send", + commandId: "cmd-1", + threadId: "thread-1", + messageId: "msg-1", + role: "user", + text: "hello", + streaming: false, + createdAt: now, +}; + +describe("commandInvariants", () => { + it("finds threads by id and project", () => { + expect(findThreadById(readModel, "thread-1")?.projectId).toBe("project-a"); + expect(findThreadById(readModel, "missing")).toBeUndefined(); + expect(listThreadsByProjectId(readModel, "project-b").map((thread) => thread.id)).toEqual([ + "thread-2", + ]); + }); + + it("requires existing thread", async () => { + const thread = await Effect.runPromise( + requireThread({ + readModel, + command: messageSendCommand, + threadId: "thread-1", + }), + ); + expect(thread.id).toBe("thread-1"); + + await expect( + Effect.runPromise( + requireThread({ + readModel, + command: messageSendCommand, + threadId: "missing", + }), + ), + ).rejects.toThrow("does not exist"); + }); + + it("requires missing thread for create flows", async () => { + await Effect.runPromise( + requireThreadAbsent({ + readModel, + command: { + type: "thread.create", + commandId: "cmd-2", + threadId: "thread-3", + projectId: "project-a", + title: "new", + model: "gpt-5-codex", + branch: null, + worktreePath: null, + createdAt: now, + }, + threadId: "thread-3", + }), + ); + + await expect( + Effect.runPromise( + requireThreadAbsent({ + readModel, + command: { + type: "thread.create", + commandId: "cmd-3", + threadId: "thread-1", + projectId: "project-a", + title: "dup", + model: "gpt-5-codex", + branch: null, + worktreePath: null, + createdAt: now, + }, + threadId: "thread-1", + }), + ), + ).rejects.toThrow("already exists"); + }); + + it("requires non-negative integers", async () => { + await Effect.runPromise( + requireNonNegativeInteger({ + commandType: "thread.revert", + field: "turnCount", + value: 0, + }), + ); + + await expect( + Effect.runPromise( + requireNonNegativeInteger({ + commandType: "thread.revert", + field: "turnCount", + value: -1, + }), + ), + ).rejects.toThrow("greater than or equal to 0"); + }); +}); diff --git a/apps/server/src/orchestration/commandInvariants.ts b/apps/server/src/orchestration/commandInvariants.ts new file mode 100644 index 0000000000000000000000000000000000000000..e5e0f5875ebcc1a418ea492264be6fab9b207b5c --- /dev/null +++ b/apps/server/src/orchestration/commandInvariants.ts @@ -0,0 +1,80 @@ +import type { OrchestrationCommand, OrchestrationReadModel, OrchestrationThread } from "@t3tools/contracts"; +import { Effect } from "effect"; + +import { OrchestrationCommandInvariantError } from "./Errors.ts"; + +function invariantError(commandType: string, detail: string): OrchestrationCommandInvariantError { + return new OrchestrationCommandInvariantError({ + commandType, + detail, + }); +} + +export function findThreadById( + readModel: OrchestrationReadModel, + threadId: string, +): OrchestrationThread | undefined { + return readModel.threads.find((thread) => thread.id === threadId); +} + +export function listThreadsByProjectId( + readModel: OrchestrationReadModel, + projectId: string, +): ReadonlyArray { + return readModel.threads.filter((thread) => thread.projectId === projectId); +} + +export function requireThread( + input: { + readonly readModel: OrchestrationReadModel; + readonly command: OrchestrationCommand; + readonly threadId: string; + }, +): Effect.Effect { + const thread = findThreadById(input.readModel, input.threadId); + if (thread) { + return Effect.succeed(thread); + } + return Effect.fail( + invariantError( + input.command.type, + `Thread '${input.threadId}' does not exist for command '${input.command.type}'.`, + ), + ); +} + +export function requireThreadAbsent( + input: { + readonly readModel: OrchestrationReadModel; + readonly command: OrchestrationCommand; + readonly threadId: string; + }, +): Effect.Effect { + if (!findThreadById(input.readModel, input.threadId)) { + return Effect.void; + } + return Effect.fail( + invariantError( + input.command.type, + `Thread '${input.threadId}' already exists and cannot be created twice.`, + ), + ); +} + +export function requireNonNegativeInteger( + input: { + readonly commandType: OrchestrationCommand["type"]; + readonly field: string; + readonly value: number; + }, +): Effect.Effect { + if (Number.isInteger(input.value) && input.value >= 0) { + return Effect.void; + } + return Effect.fail( + invariantError( + input.commandType, + `${input.field} must be an integer greater than or equal to 0.`, + ), + ); +} diff --git a/apps/server/src/orchestration/Layer.ts b/apps/server/src/orchestration/decider.ts similarity index 55% rename from apps/server/src/orchestration/Layer.ts rename to apps/server/src/orchestration/decider.ts index c03029b93ae7ae6be31446c1c7ce8eb323e6fd63..b1e4c5d79625ef8d19be4916f0d0d96c82ad33e6 100644 --- a/apps/server/src/orchestration/Layer.ts +++ b/apps/server/src/orchestration/decider.ts @@ -3,28 +3,31 @@ import type { OrchestrationEvent, OrchestrationReadModel, } from "@t3tools/contracts"; -import { OrchestrationCommandSchema } from "@t3tools/contracts"; -import { Deferred, Effect, Layer, PubSub, Queue, Schema, Stream } from "effect"; +import { Effect } from "effect"; -import { createLogger } from "../logger.ts"; -import { OrchestrationEventRepository } from "../persistence/Services/OrchestrationEvents.ts"; +import { OrchestrationCommandInvariantError } from "./Errors.ts"; import { - toOrchestrationCommandDecodeError, - toOrchestrationJsonParseError, - type OrchestrationDispatchError, -} from "./Errors.ts"; -import { createEmptyReadModel, reduceEvent } from "./reducer.ts"; -import { OrchestrationEngineService, type OrchestrationEngineShape } from "./Service.ts"; + requireNonNegativeInteger, + requireThread, + requireThreadAbsent, +} from "./commandInvariants.ts"; -interface CommandEnvelope { - command: OrchestrationCommand; - result: Deferred.Deferred<{ sequence: number }, OrchestrationDispatchError>; -} - -function mapCommandToEvent(command: OrchestrationCommand): Omit { +export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand")(function* ({ + command, + readModel, +}: { + readonly command: OrchestrationCommand; + readonly readModel: OrchestrationReadModel; +}): Effect.fn.Return, OrchestrationCommandInvariantError> { const eventId = crypto.randomUUID(); + switch (command.type) { case "thread.create": + yield* requireThreadAbsent({ + readModel, + command, + threadId: command.threadId, + }); return { eventId, type: "thread.created", @@ -44,6 +47,11 @@ function mapCommandToEvent(command: OrchestrationCommand): Omit(); - const eventPubSub = yield* PubSub.unbounded(); - - const processEnvelope = (envelope: CommandEnvelope): Effect.Effect => - Effect.gen(function* () { - const eventBase = mapCommandToEvent(envelope.command); - const savedEvent = yield* eventStore.append(eventBase); - readModel = yield* reduceEvent(readModel, savedEvent); - yield* PubSub.publish(eventPubSub, savedEvent); - - yield* Deferred.succeed(envelope.result, { sequence: savedEvent.sequence }); - }).pipe(Effect.catch((error) => Deferred.fail(envelope.result, error).pipe(Effect.asVoid))); - - const bootstrapReadModel: Effect.Effect = Effect.gen( - function* () { - const existingEvents = yield* eventStore.readAll(); - for (const event of existingEvents) { - readModel = yield* reduceEvent(readModel, event); - } - }, - ); - - yield* bootstrapReadModel; - - const worker = Effect.forever(Queue.take(commandQueue).pipe(Effect.flatMap(processEnvelope))); - yield* Effect.forkScoped(worker); - logger.info("orchestration engine started", { - sequence: readModel.sequence, - }); - - const getSnapshot: OrchestrationEngineShape["getSnapshot"] = () => - Effect.sync((): OrchestrationReadModel => readModel); - - const replayEvents: OrchestrationEngineShape["replayEvents"] = (fromSequenceExclusive) => - eventStore.readFromSequence(fromSequenceExclusive); - - const dispatch: OrchestrationEngineShape["dispatch"] = (command) => - Effect.gen(function* () { - const result = yield* Deferred.make<{ sequence: number }, OrchestrationDispatchError>(); - yield* Queue.offer(commandQueue, { command, result }); - return yield* Deferred.await(result); - }); - - const dispatchUnknown: OrchestrationEngineShape["dispatchUnknown"] = (command) => - Effect.gen(function* () { - const payload = - typeof command === "string" - ? yield* Effect.try({ - try: () => JSON.parse(command) as unknown, - catch: toOrchestrationJsonParseError, - }) - : command; - - const decoded = yield* decodeUnknownCommand(payload).pipe( - Effect.mapError(toOrchestrationCommandDecodeError), - ); - - return yield* dispatch(decoded); - }); - - const streamDomainEvents: OrchestrationEngineShape["streamDomainEvents"] = - Stream.fromPubSub(eventPubSub); - - return { - getSnapshot, - replayEvents, - dispatchUnknown, - dispatch, - streamDomainEvents, - } satisfies OrchestrationEngineShape; }); - -export const OrchestrationEngineLive = Layer.effect( - OrchestrationEngineService, - makeOrchestrationEngine, -); diff --git a/apps/server/src/orchestration/reducer.test.ts b/apps/server/src/orchestration/projector.test.ts similarity index 87% rename from apps/server/src/orchestration/reducer.test.ts rename to apps/server/src/orchestration/projector.test.ts index 97cb5a9d14afef839a0b941ee06308c479cbf70e..4be935ac5e23f9bb83758accd202756976350886 100644 --- a/apps/server/src/orchestration/reducer.test.ts +++ b/apps/server/src/orchestration/projector.test.ts @@ -2,9 +2,9 @@ import type { OrchestrationEvent } from "@t3tools/contracts"; import { Effect } from "effect"; import { describe, expect, it } from "vitest"; -import { createEmptyReadModel, reduceEvent } from "./reducer.ts"; +import { createEmptyReadModel, projectEvent } from "./projector.ts"; -describe("orchestration reducer", () => { +describe("orchestration projector", () => { it("applies thread.created events using runtime schema decoding", async () => { const now = new Date().toISOString(); const model = createEmptyReadModel(now); @@ -28,7 +28,7 @@ describe("orchestration reducer", () => { }, }; - const next = await Effect.runPromise(reduceEvent(model, event)); + const next = await Effect.runPromise(projectEvent(model, event)); expect(next.sequence).toBe(1); expect(next.threads).toEqual([ { @@ -73,10 +73,10 @@ describe("orchestration reducer", () => { }, }; - await expect(Effect.runPromise(reduceEvent(model, event))).rejects.toBeDefined(); + await expect(Effect.runPromise(projectEvent(model, event))).rejects.toBeDefined(); }); - it("keeps reducer forward-compatible for unknown event types", async () => { + it("keeps projector forward-compatible for unknown event types", async () => { const now = new Date().toISOString(); const model = createEmptyReadModel(now); const event: OrchestrationEvent = { @@ -90,7 +90,7 @@ describe("orchestration reducer", () => { payload: { test: true }, }; - const next = await Effect.runPromise(reduceEvent(model, event)); + const next = await Effect.runPromise(projectEvent(model, event)); expect(next.sequence).toBe(7); expect(next.updatedAt).toBe("2026-01-01T00:00:00.000Z"); expect(next.threads).toEqual([]); @@ -168,9 +168,9 @@ describe("orchestration reducer", () => { }, }; - const afterCreate = await Effect.runPromise(reduceEvent(model, createdEvent)); - const afterRunning = await Effect.runPromise(reduceEvent(afterCreate, runningSessionEvent)); - const afterReady = await Effect.runPromise(reduceEvent(afterRunning, readySessionEvent)); + const afterCreate = await Effect.runPromise(projectEvent(model, createdEvent)); + const afterRunning = await Effect.runPromise(projectEvent(afterCreate, runningSessionEvent)); + const afterReady = await Effect.runPromise(projectEvent(afterRunning, readySessionEvent)); const thread = afterReady.threads[0]; expect(thread?.latestTurnId).toBe("turn-1"); @@ -241,9 +241,9 @@ describe("orchestration reducer", () => { }, }; - const afterCreate = await Effect.runPromise(reduceEvent(model, createdEvent)); - const afterDelta = await Effect.runPromise(reduceEvent(afterCreate, deltaEvent)); - const afterComplete = await Effect.runPromise(reduceEvent(afterDelta, completionEvent)); + const afterCreate = await Effect.runPromise(projectEvent(model, createdEvent)); + const afterDelta = await Effect.runPromise(projectEvent(afterCreate, deltaEvent)); + const afterComplete = await Effect.runPromise(projectEvent(afterDelta, completionEvent)); const message = afterComplete.threads[0]?.messages[0]; expect(message).toEqual({ diff --git a/apps/server/src/orchestration/reducer.ts b/apps/server/src/orchestration/projector.ts similarity index 97% rename from apps/server/src/orchestration/reducer.ts rename to apps/server/src/orchestration/projector.ts index bf643572eebf597946d7bbe06bf41ca44bd27b4c..168e439d33d512409d0d6dec18cc00ffe5e3044b 100644 --- a/apps/server/src/orchestration/reducer.ts +++ b/apps/server/src/orchestration/projector.ts @@ -17,7 +17,7 @@ import { } from "@t3tools/contracts"; import { Effect, Schema } from "effect"; -import { toReducerDecodeError, type OrchestrationReducerDecodeError } from "./Errors.ts"; +import { toProjectorDecodeError, type OrchestrationProjectorDecodeError } from "./Errors.ts"; type ThreadPatch = Partial>; @@ -34,10 +34,10 @@ function decodeForEvent( value: unknown, eventType: string, field: string, -): Effect.Effect { +): Effect.Effect { return Schema.decodeUnknownEffect(schema)(value).pipe( - Effect.mapError(toReducerDecodeError(`${eventType}:${field}`)), - ) as Effect.Effect; + Effect.mapError(toProjectorDecodeError(`${eventType}:${field}`)), + ) as Effect.Effect; } const ThreadCreatedPayloadSchema = Schema.Struct({ @@ -104,10 +104,10 @@ export function createEmptyReadModel(nowIso: string): OrchestrationReadModel { }; } -export function reduceEvent( +export function projectEvent( model: OrchestrationReadModel, event: OrchestrationEvent, -): Effect.Effect { +): Effect.Effect { const nextBase: OrchestrationReadModel = { ...model, sequence: event.sequence, diff --git a/apps/server/src/persistence/Errors.ts b/apps/server/src/persistence/Errors.ts index 75f23a89e10f2eb6573ab3c70c6bb9cfbb4e2b00..a9073951453aaf70f639ada2bb5f2f431aaba8a9 100644 --- a/apps/server/src/persistence/Errors.ts +++ b/apps/server/src/persistence/Errors.ts @@ -158,7 +158,7 @@ export type ProjectRepositoryError = | ProjectPathMissingError | ProjectNotFoundError; -export type OrchestrationEventRepositoryError = +export type OrchestrationEventStoreError = | PersistenceSqlError | PersistenceDecodeError | PersistenceSerializationError; diff --git a/apps/server/src/persistence/Layers/OrchestrationEvents.test.ts b/apps/server/src/persistence/Layers/OrchestrationEventStore.test.ts similarity index 56% rename from apps/server/src/persistence/Layers/OrchestrationEvents.test.ts rename to apps/server/src/persistence/Layers/OrchestrationEventStore.test.ts index dddba6a727d9bc1f2a93ef7fd2b3964f0ac1fc4d..19f72ff4faf8a69ae14ed384088415208f1164f2 100644 --- a/apps/server/src/persistence/Layers/OrchestrationEvents.test.ts +++ b/apps/server/src/persistence/Layers/OrchestrationEventStore.test.ts @@ -1,25 +1,25 @@ -import { Effect, Layer } from "effect"; +import { Effect, Layer, Stream } from "effect"; import { assert, it } from "@effect/vitest"; -import { OrchestrationEventRepository } from "../Services/OrchestrationEvents.ts"; +import { OrchestrationEventStore } from "../Services/OrchestrationEventStore.ts"; import { SqlitePersistenceMemory } from "./Sqlite.ts"; -import { OrchestrationEventRepositoryLive } from "./OrchestrationEvents.ts"; +import { OrchestrationEventStoreLive } from "./OrchestrationEventStore.ts"; import { NodeServices } from "@effect/platform-node"; const layer = it.layer( Layer.mergeAll( - Layer.provide(OrchestrationEventRepositoryLive, SqlitePersistenceMemory), + Layer.provide(OrchestrationEventStoreLive, SqlitePersistenceMemory), NodeServices.layer, ), ); -layer("OrchestrationEventRepository", (it) => { +layer("OrchestrationEventStore", (it) => { it.effect("stores and replays events", () => Effect.gen(function* () { const createdAt = new Date().toISOString(); - const repository = yield* OrchestrationEventRepository; - const saved = yield* repository.append({ + const eventStore = yield* OrchestrationEventStore; + const saved = yield* eventStore.append({ eventId: "event-1", type: "thread.created", aggregateType: "thread", @@ -30,7 +30,9 @@ layer("OrchestrationEventRepository", (it) => { }); assert.equal(saved.sequence, 1); - const replayed = yield* repository.readFromSequence(0); + const replayed = yield* Stream.runCollect(eventStore.readFromSequence(0)).pipe( + Effect.map(Array.from), + ); assert.deepEqual(replayed, [saved]); }), ); diff --git a/apps/server/src/persistence/Layers/OrchestrationEvents.ts b/apps/server/src/persistence/Layers/OrchestrationEventStore.ts similarity index 66% rename from apps/server/src/persistence/Layers/OrchestrationEvents.ts rename to apps/server/src/persistence/Layers/OrchestrationEventStore.ts index 4da8c8c06323e06163844a23e83d20e53f56c293..c1a79890c2905a8c714cea11e7b92ed6581dd24c 100644 --- a/apps/server/src/persistence/Layers/OrchestrationEvents.ts +++ b/apps/server/src/persistence/Layers/OrchestrationEventStore.ts @@ -2,19 +2,19 @@ import type { OrchestrationEvent } from "@t3tools/contracts"; import { OrchestrationEventSchema } from "@t3tools/contracts"; import * as SqlClient from "effect/unstable/sql/SqlClient"; import * as SqlSchema from "effect/unstable/sql/SqlSchema"; -import { Effect, Layer, Schema } from "effect"; +import { Effect, Layer, Schema, Stream } from "effect"; import { toPersistenceDecodeCauseError, toPersistenceDecodeError, toPersistenceSerializationError, toPersistenceSqlError, - type OrchestrationEventRepositoryError, + type OrchestrationEventStoreError, } from "../Errors.ts"; import { - OrchestrationEventRepository, - type OrchestrationEventRepositoryShape, -} from "../Services/OrchestrationEvents.ts"; + OrchestrationEventStore, + type OrchestrationEventStoreShape, +} from "../Services/OrchestrationEventStore.ts"; const decodeEvent = Schema.decodeUnknownEffect(OrchestrationEventSchema); @@ -43,11 +43,13 @@ const ReadFromSequenceRequestSchema = Schema.Struct({ sequenceExclusive: Schema.Number, limit: Schema.Number, }); +const DEFAULT_READ_FROM_SEQUENCE_LIMIT = 1_000; +const READ_PAGE_SIZE = 500; function eventRowToOrchestrationEvent( row: Schema.Schema.Type, operation: string, -): Effect.Effect { +): Effect.Effect { return Effect.try({ try: () => JSON.parse(row.payloadJson) as unknown, catch: toPersistenceDecodeCauseError(`${operation}:parsePayloadJson`), @@ -67,7 +69,7 @@ function eventRowToOrchestrationEvent( ); } -const makeRepository = Effect.gen(function* () { +const makeEventStore = Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; const appendEventRow = SqlSchema.findOne({ @@ -126,10 +128,10 @@ const makeRepository = Effect.gen(function* () { `, }); - const append: OrchestrationEventRepositoryShape["append"] = (event) => + const append: OrchestrationEventStoreShape["append"] = (event) => Effect.try({ try: () => JSON.stringify(event.payload), - catch: toPersistenceSerializationError("OrchestrationEventRepository.append:encodePayload"), + catch: toPersistenceSerializationError("OrchestrationEventStore.append:encodePayload"), }).pipe( Effect.flatMap((payloadJson) => appendEventRow({ @@ -141,38 +143,68 @@ const makeRepository = Effect.gen(function* () { commandId: event.commandId, payloadJson, }).pipe( - Effect.mapError(toPersistenceSqlError("OrchestrationEventRepository.append:insert")), + Effect.mapError(toPersistenceSqlError("OrchestrationEventStore.append:insert")), Effect.flatMap((row) => - eventRowToOrchestrationEvent(row, "OrchestrationEventRepository.append:rowToEvent"), + eventRowToOrchestrationEvent(row, "OrchestrationEventStore.append:rowToEvent"), ), ), ), ); - const readFromSequence: OrchestrationEventRepositoryShape["readFromSequence"] = ( + const readFromSequence: OrchestrationEventStoreShape["readFromSequence"] = ( sequenceExclusive, - limit = 1_000, - ) => - readEventRowsFromSequence({ sequenceExclusive, limit }).pipe( - Effect.mapError(toPersistenceSqlError("OrchestrationEventRepository.readFromSequence:query")), - Effect.flatMap((rows) => - Effect.forEach(rows, (row) => - eventRowToOrchestrationEvent( - row, - "OrchestrationEventRepository.readFromSequence:rowToEvent", + limit = DEFAULT_READ_FROM_SEQUENCE_LIMIT, + ) => { + const normalizedLimit = Math.max(0, Math.floor(limit)); + if (normalizedLimit === 0) { + return Stream.empty; + } + const readPage = (cursor: number, remaining: number): Stream.Stream< + OrchestrationEvent, + OrchestrationEventStoreError + > => + Stream.fromEffect( + readEventRowsFromSequence({ + sequenceExclusive: cursor, + limit: Math.min(remaining, READ_PAGE_SIZE), + }).pipe( + Effect.mapError(toPersistenceSqlError("OrchestrationEventStore.readFromSequence:query")), + Effect.flatMap((rows) => + Effect.forEach(rows, (row) => + eventRowToOrchestrationEvent( + row, + "OrchestrationEventStore.readFromSequence:rowToEvent", + ), + ), ), ), - ), - ); + ).pipe( + Stream.flatMap((events) => { + if (events.length === 0) { + return Stream.empty; + } + const nextRemaining = remaining - events.length; + if (nextRemaining <= 0) { + return Stream.fromIterable(events); + } + return Stream.concat( + Stream.fromIterable(events), + readPage(events[events.length - 1]!.sequence, nextRemaining), + ); + }), + ); + + return readPage(sequenceExclusive, normalizedLimit); + }; return { append, readFromSequence, readAll: () => readFromSequence(0, Number.MAX_SAFE_INTEGER), - } satisfies OrchestrationEventRepositoryShape; + } satisfies OrchestrationEventStoreShape; }); -export const OrchestrationEventRepositoryLive = Layer.effect( - OrchestrationEventRepository, - makeRepository, +export const OrchestrationEventStoreLive = Layer.effect( + OrchestrationEventStore, + makeEventStore, ); diff --git a/apps/server/src/persistence/Services/OrchestrationEvents.ts b/apps/server/src/persistence/Services/OrchestrationEventStore.ts similarity index 50% rename from apps/server/src/persistence/Services/OrchestrationEvents.ts rename to apps/server/src/persistence/Services/OrchestrationEventStore.ts index 88545f142df8c50816babeb88b1e27d238572a1e..568866e55ef578f53b3e948d5209f42dde92251b 100644 --- a/apps/server/src/persistence/Services/OrchestrationEvents.ts +++ b/apps/server/src/persistence/Services/OrchestrationEventStore.ts @@ -1,5 +1,5 @@ /** - * OrchestrationEventRepository - Repository interface for orchestration events. + * OrchestrationEventStore - Event store interface for orchestration events. * * Owns durable append/replay access to the orchestration event stream. It does * not reduce events into read models or apply command validation rules. @@ -7,15 +7,15 @@ * Uses Effect `ServiceMap.Service` for dependency injection and exposes typed * persistence/decode errors for event append and replay operations. * - * @module OrchestrationEventRepository + * @module OrchestrationEventStore */ import type { OrchestrationEvent } from "@t3tools/contracts"; import { ServiceMap } from "effect"; -import type { Effect } from "effect"; +import type { Effect, Stream } from "effect"; -import type { OrchestrationEventRepositoryError } from "../Errors.ts"; +import type { OrchestrationEventStoreError } from "../Errors.ts"; -export interface OrchestrationEventRepositoryShape { +export interface OrchestrationEventStoreShape { /** * Persist a new orchestration event. * @@ -24,40 +24,40 @@ export interface OrchestrationEventRepositoryShape { */ readonly append: ( event: Omit, - ) => Effect.Effect; + ) => Effect.Effect; /** * Replay events after the provided sequence. * * @param sequenceExclusive - Sequence cursor (exclusive). - * @param limit - Maximum number of events to return. - * @returns Effect containing ordered events. + * @param limit - Maximum number of events to emit. + * @returns Stream containing ordered events. */ readonly readFromSequence: ( sequenceExclusive: number, limit?: number, - ) => Effect.Effect; + ) => Stream.Stream; /** * Read all events from the beginning of the stream. * - * @returns Effect containing all stored events. + * @returns Stream containing all stored events. */ - readonly readAll: () => Effect.Effect; + readonly readAll: () => Stream.Stream; } /** - * OrchestrationEventRepository - Service tag for orchestration event persistence. + * OrchestrationEventStore - Service tag for orchestration event persistence. * * @example * ```ts * const program = Effect.gen(function* () { - * const events = yield* OrchestrationEventRepository - * return yield* events.readAll() + * const events = yield* OrchestrationEventStore + * return yield* Stream.runCollect(events.readAll()) * }) * ``` */ -export class OrchestrationEventRepository extends ServiceMap.Service< - OrchestrationEventRepository, - OrchestrationEventRepositoryShape ->()("orchestration/EventRepository") {} +export class OrchestrationEventStore extends ServiceMap.Service< + OrchestrationEventStore, + OrchestrationEventStoreShape +>()("orchestration/EventStore") {} diff --git a/apps/server/src/wsServer.ts b/apps/server/src/wsServer.ts index b251500a9c85063239988ad54919019fd3d9e801..7fb9ab4f03a78a9925e278538661cd65f0764589 100644 --- a/apps/server/src/wsServer.ts +++ b/apps/server/src/wsServer.ts @@ -36,13 +36,16 @@ import { import { TerminalManager } from "./terminalManager"; import { loadResolvedKeybindingsConfig, upsertKeybindingRule } from "./keybindings"; import { searchWorkspaceEntries } from "./workspaceEntries"; -import { OrchestrationEngineLive } from "./orchestration/Layer"; -import { OrchestrationEngineService, type OrchestrationEngineShape } from "./orchestration/Service"; +import { OrchestrationEngineLive } from "./orchestration/Layers/OrchestrationEngine"; +import { + OrchestrationEngineService, + type OrchestrationEngineShape, +} from "./orchestration/Services/OrchestrationEngine"; import { ProjectRepositoryLive } from "./persistence/Layers/Projects"; import { makeSqlitePersistenceLive } from "./persistence/Layers/Sqlite"; import { ProjectRepository, type ProjectRepositoryShape } from "./persistence/Services/Projects"; import assert from "node:assert"; -import { OrchestrationEventRepositoryLive } from "./persistence/Layers/OrchestrationEvents"; +import { OrchestrationEventStoreLive } from "./persistence/Layers/OrchestrationEventStore"; import { makeCodexAdapterLive } from "./provider/Layers/CodexAdapter"; import { ProviderAdapterRegistryLive } from "./provider/Layers/ProviderAdapterRegistry"; import { makeProviderServiceLive } from "./provider/Layers/ProviderService"; @@ -358,8 +361,8 @@ export function createServer(options: ServerOptions) { const processEvent = (event: ProviderRuntimeEvent) => Effect.gen(function* () { - const snapshot = yield* orchestrationEngine.getSnapshot(); - const thread = snapshot.threads.find( + const readModel = yield* orchestrationEngine.getReadModel(); + const thread = readModel.threads.find( (entry) => entry.session?.sessionId === event.sessionId, ); if (!thread) return; @@ -735,8 +738,8 @@ export function createServer(options: ServerOptions) { const params = request.params as { sessionId?: string }; const sessionId = params.sessionId; if (typeof sessionId === "string") { - const snapshot = await effectRuntime.runPromise(orchestrationEngine.getSnapshot()); - const thread = snapshot.threads.find((entry) => entry.session?.sessionId === sessionId); + const readModel = await effectRuntime.runPromise(orchestrationEngine.getReadModel()); + const thread = readModel.threads.find((entry) => entry.session?.sessionId === sessionId); if (thread) { const now = new Date().toISOString(); await effectRuntime.runPromise( @@ -909,7 +912,7 @@ export function createServer(options: ServerOptions) { }; case ORCHESTRATION_WS_METHODS.getSnapshot: - return effectRuntime.runPromise(orchestrationEngine.getSnapshot()); + return effectRuntime.runPromise(orchestrationEngine.getReadModel()); case ORCHESTRATION_WS_METHODS.dispatchCommand: orchestrationCommandLogger?.write({ @@ -917,21 +920,23 @@ export function createServer(options: ServerOptions) { requestId: request.id, command: request.params, }); - return effectRuntime.runPromise(orchestrationEngine.dispatchUnknown(request.params)); + return effectRuntime.runPromise(orchestrationEngine.dispatchUnknownCommand(request.params)); case ORCHESTRATION_WS_METHODS.replayEvents: return effectRuntime.runPromise( - orchestrationEngine.replayEvents( - Math.max( - 0, - Math.floor( - Number( - (request.params as { fromSequenceExclusive?: number } | undefined) - ?.fromSequenceExclusive ?? 0, + Stream.runCollect( + orchestrationEngine.readEvents( + Math.max( + 0, + Math.floor( + Number( + (request.params as { fromSequenceExclusive?: number } | undefined) + ?.fromSequenceExclusive ?? 0, + ), ), ), ), - ), + ).pipe(Effect.map((events) => Array.from(events))), ); default: @@ -952,7 +957,7 @@ export function createServer(options: ServerOptions) { const persistenceLayer = customPersistenceLayer ?? makeSqlitePersistenceLive(dbPath); const orchestrationLayer = Layer.provide( OrchestrationEngineLive, - OrchestrationEventRepositoryLive, + OrchestrationEventStoreLive, ); const codexAdapterLayer = makeCodexAdapterLive({ nativeEventLogPath: path.join(providerLogsDir, "provider-native.ndjson"), -- 2.54.0