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/Layer.ts b/apps/server/src/orchestration/Layer.ts deleted file mode 100644 index c03029b93ae7ae6be31446c1c7ce8eb323e6fd63..0000000000000000000000000000000000000000 --- a/apps/server/src/orchestration/Layer.ts +++ /dev/null @@ -1,258 +0,0 @@ -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 { OrchestrationEventRepository } from "../persistence/Services/OrchestrationEvents.ts"; -import { - toOrchestrationCommandDecodeError, - toOrchestrationJsonParseError, - type OrchestrationDispatchError, -} from "./Errors.ts"; -import { createEmptyReadModel, reduceEvent } from "./reducer.ts"; -import { OrchestrationEngineService, type OrchestrationEngineShape } from "./Service.ts"; - -interface CommandEnvelope { - command: OrchestrationCommand; - result: Deferred.Deferred<{ sequence: number }, OrchestrationDispatchError>; -} - -function mapCommandToEvent(command: OrchestrationCommand): Omit { - const eventId = crypto.randomUUID(); - switch (command.type) { - case "thread.create": - return { - eventId, - type: "thread.created", - aggregateType: "thread", - aggregateId: command.threadId, - occurredAt: command.createdAt, - commandId: command.commandId, - payload: { - id: command.threadId, - projectId: command.projectId, - title: command.title, - model: command.model, - branch: command.branch, - worktreePath: command.worktreePath, - createdAt: command.createdAt, - updatedAt: command.createdAt, - }, - }; - case "thread.delete": - return { - eventId, - type: "thread.deleted", - aggregateType: "thread", - aggregateId: command.threadId, - occurredAt: command.createdAt, - commandId: command.commandId, - payload: { - id: command.threadId, - deletedAt: command.createdAt, - }, - }; - case "thread.meta.update": - return { - eventId, - type: "thread.meta-updated", - aggregateType: "thread", - aggregateId: command.threadId, - occurredAt: command.createdAt, - commandId: command.commandId, - payload: { - threadId: command.threadId, - ...(command.title !== undefined ? { title: command.title } : {}), - ...(command.model !== undefined ? { model: command.model } : {}), - ...(command.branch !== undefined ? { branch: command.branch } : {}), - ...(command.worktreePath !== undefined ? { worktreePath: command.worktreePath } : {}), - updatedAt: command.createdAt, - }, - }; - case "message.send": - return { - eventId, - type: "message.sent", - aggregateType: "thread", - aggregateId: command.threadId, - occurredAt: command.createdAt, - commandId: command.commandId, - payload: { - id: command.messageId, - role: command.role, - text: command.text, - threadId: command.threadId, - createdAt: command.createdAt, - streaming: command.streaming === true, - }, - }; - case "thread.session": - return { - eventId, - type: "thread.session-set", - aggregateType: "thread", - aggregateId: command.threadId, - occurredAt: command.createdAt, - commandId: command.commandId, - payload: { - threadId: command.threadId, - session: command.session, - }, - }; - case "git.readModel.upsert": - return { - eventId, - type: "git.read-model-upsert", - aggregateType: "project", - aggregateId: command.projectId, - occurredAt: command.createdAt, - commandId: command.commandId, - payload: { - projectId: command.projectId, - branch: command.branch, - hasWorkingTreeChanges: command.hasWorkingTreeChanges, - aheadCount: command.aheadCount, - behindCount: command.behindCount, - updatedAt: command.createdAt, - }, - }; - case "thread.turnDiff.complete": - return { - eventId, - type: "thread.turn-diff-completed", - aggregateType: "thread", - aggregateId: command.threadId, - occurredAt: command.createdAt, - commandId: command.commandId, - payload: { - threadId: command.threadId, - turnId: command.turnId, - completedAt: command.completedAt, - ...(command.status !== undefined ? { status: command.status } : {}), - files: command.files, - ...(command.assistantMessageId !== undefined - ? { assistantMessageId: command.assistantMessageId } - : {}), - ...(command.checkpointTurnCount !== undefined - ? { checkpointTurnCount: command.checkpointTurnCount } - : {}), - }, - }; - case "thread.activity.append": - return { - eventId, - type: "thread.activity-appended", - aggregateType: "thread", - aggregateId: command.threadId, - occurredAt: command.createdAt, - commandId: command.commandId, - payload: { - threadId: command.threadId, - activity: command.activity, - }, - }; - case "thread.revert": - return { - eventId, - type: "thread.reverted", - aggregateType: "thread", - aggregateId: command.threadId, - occurredAt: command.createdAt, - commandId: command.commandId, - payload: { - threadId: command.threadId, - turnCount: command.turnCount, - messageCount: command.messageCount, - }, - }; - } -} - -const decodeUnknownCommand = Schema.decodeUnknownEffect(OrchestrationCommandSchema); - -const makeOrchestrationEngine = Effect.gen(function* () { - const logger = createLogger("orchestration"); - const eventStore = yield* OrchestrationEventRepository; - - 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 = 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/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/Service.ts deleted file mode 100644 index bffd33e50823b42f291123199e772bbae36d7906..0000000000000000000000000000000000000000 --- a/apps/server/src/orchestration/Service.ts +++ /dev/null @@ -1,84 +0,0 @@ -/** - * 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 - * management or transport concerns (e.g. websocket request parsing). - * - * Uses Effect `ServiceMap.Service` for dependency injection. Command dispatch, - * replay, and unknown-input decoding all return typed domain errors. - * - * @module OrchestrationEngineService - */ -import type { - OrchestrationCommand, - OrchestrationEvent, - OrchestrationReadModel, -} from "@t3tools/contracts"; -import { ServiceMap } from "effect"; -import type { Effect, Stream } from "effect"; - -import type { OrchestrationDispatchError, OrchestrationEngineError } from "./Errors.ts"; -import type { OrchestrationEventRepositoryError } from "../persistence/Errors.ts"; - -export interface OrchestrationEngineShape { - /** - * Read the current in-memory orchestration snapshot. - * - * @returns Effect containing the latest read model. - */ - readonly getSnapshot: () => Effect.Effect; - - /** - * Replay persisted orchestration events from an exclusive sequence cursor. - * - * @param fromSequenceExclusive - Sequence cursor (exclusive). - * @returns Effect containing ordered events. - */ - readonly replayEvents: ( - fromSequenceExclusive: number, - ) => Effect.Effect; - - /** - * Validate and dispatch an unknown command payload. - * - * Accepts either parsed values or raw JSON strings. - * - * @param command - Unknown command payload. - * @returns Effect containing the sequence of the persisted event. - */ - readonly dispatchUnknown: ( - command: unknown, - ) => Effect.Effect<{ sequence: number }, OrchestrationEngineError, never>; - - /** - * Dispatch a validated orchestration command. - * - * @param command - Valid orchestration command. - * @returns Effect containing the sequence of the persisted event. - */ - readonly dispatch: ( - command: OrchestrationCommand, - ) => Effect.Effect<{ sequence: number }, OrchestrationDispatchError, never>; - - /** - * Stream persisted domain events in dispatch order. - */ - readonly streamDomainEvents: Stream.Stream; -} - -/** - * OrchestrationEngineService - Service tag for orchestration engine access. - * - * @example - * ```ts - * const program = Effect.gen(function* () { - * const engine = yield* OrchestrationEngineService - * return yield* engine.getSnapshot() - * }) - * ``` - */ -export class OrchestrationEngineService extends ServiceMap.Service< - OrchestrationEngineService, - OrchestrationEngineShape ->()("orchestration/Engine") {} diff --git a/apps/server/src/orchestration/Services/OrchestrationEngine.ts b/apps/server/src/orchestration/Services/OrchestrationEngine.ts new file mode 100644 index 0000000000000000000000000000000000000000..2287f2ae7f1686f81301c27e905f019b2c170072 --- /dev/null +++ b/apps/server/src/orchestration/Services/OrchestrationEngine.ts @@ -0,0 +1,84 @@ +/** + * OrchestrationEngineService - Service interface for orchestration command handling. + * + * Owns command validation/dispatch and in-memory read-model updates backed by + * `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, + * replay, and unknown-input decoding all return typed domain errors. + * + * @module OrchestrationEngineService + */ +import type { + OrchestrationCommand, + OrchestrationEvent, + OrchestrationReadModel, +} from "@t3tools/contracts"; +import { ServiceMap } from "effect"; +import type { Effect, Stream } from "effect"; + +import type { OrchestrationDispatchError, OrchestrationEngineError } from "../Errors.ts"; +import type { OrchestrationEventStoreError } from "../../persistence/Errors.ts"; + +export interface OrchestrationEngineShape { + /** + * Read the current in-memory orchestration read model. + * + * @returns Effect containing the latest read model. + */ + readonly getReadModel: () => Effect.Effect; + + /** + * Replay persisted orchestration events from an exclusive sequence cursor. + * + * @param fromSequenceExclusive - Sequence cursor (exclusive). + * @returns Stream containing ordered events. + */ + readonly readEvents: ( + fromSequenceExclusive: number, + ) => Stream.Stream; + + /** + * Validate and dispatch an unknown command payload. + * + * Accepts either parsed values or raw JSON strings. + * + * @param command - Unknown command payload. + * @returns Effect containing the sequence of the persisted event. + */ + readonly dispatchUnknownCommand: ( + command: unknown, + ) => Effect.Effect<{ sequence: number }, OrchestrationEngineError, never>; + + /** + * Dispatch a validated orchestration command. + * + * @param command - Valid orchestration command. + * @returns Effect containing the sequence of the persisted event. + */ + readonly dispatch: ( + command: OrchestrationCommand, + ) => Effect.Effect<{ sequence: number }, OrchestrationDispatchError, never>; + + /** + * Stream persisted domain events in dispatch order. + */ + readonly streamDomainEvents: Stream.Stream; +} + +/** + * OrchestrationEngineService - Service tag for orchestration engine access. + * + * @example + * ```ts + * const program = Effect.gen(function* () { + * const engine = yield* OrchestrationEngineService + * return yield* engine.getReadModel() + * }) + * ``` + */ +export class OrchestrationEngineService extends ServiceMap.Service< + OrchestrationEngineService, + OrchestrationEngineShape +>()("orchestration/Engine") {} 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/decider.ts b/apps/server/src/orchestration/decider.ts new file mode 100644 index 0000000000000000000000000000000000000000..b1e4c5d79625ef8d19be4916f0d0d96c82ad33e6 --- /dev/null +++ b/apps/server/src/orchestration/decider.ts @@ -0,0 +1,229 @@ +import type { + OrchestrationCommand, + OrchestrationEvent, + OrchestrationReadModel, +} from "@t3tools/contracts"; +import { Effect } from "effect"; + +import { OrchestrationCommandInvariantError } from "./Errors.ts"; +import { + requireNonNegativeInteger, + requireThread, + requireThreadAbsent, +} from "./commandInvariants.ts"; + +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", + aggregateType: "thread", + aggregateId: command.threadId, + occurredAt: command.createdAt, + commandId: command.commandId, + payload: { + id: command.threadId, + projectId: command.projectId, + title: command.title, + model: command.model, + branch: command.branch, + worktreePath: command.worktreePath, + createdAt: command.createdAt, + updatedAt: command.createdAt, + }, + }; + case "thread.delete": + yield* requireThread({ + readModel, + command, + threadId: command.threadId, + }); + return { + eventId, + type: "thread.deleted", + aggregateType: "thread", + aggregateId: command.threadId, + occurredAt: command.createdAt, + commandId: command.commandId, + payload: { + id: command.threadId, + deletedAt: command.createdAt, + }, + }; + case "thread.meta.update": + yield* requireThread({ + readModel, + command, + threadId: command.threadId, + }); + return { + eventId, + type: "thread.meta-updated", + aggregateType: "thread", + aggregateId: command.threadId, + occurredAt: command.createdAt, + commandId: command.commandId, + payload: { + threadId: command.threadId, + ...(command.title !== undefined ? { title: command.title } : {}), + ...(command.model !== undefined ? { model: command.model } : {}), + ...(command.branch !== undefined ? { branch: command.branch } : {}), + ...(command.worktreePath !== undefined ? { worktreePath: command.worktreePath } : {}), + updatedAt: command.createdAt, + }, + }; + case "message.send": + yield* requireThread({ + readModel, + command, + threadId: command.threadId, + }); + return { + eventId, + type: "message.sent", + aggregateType: "thread", + aggregateId: command.threadId, + occurredAt: command.createdAt, + commandId: command.commandId, + payload: { + id: command.messageId, + role: command.role, + text: command.text, + threadId: command.threadId, + createdAt: command.createdAt, + streaming: command.streaming === true, + }, + }; + case "thread.session": + yield* requireThread({ + readModel, + command, + threadId: command.threadId, + }); + return { + eventId, + type: "thread.session-set", + aggregateType: "thread", + aggregateId: command.threadId, + occurredAt: command.createdAt, + commandId: command.commandId, + payload: { + threadId: command.threadId, + session: command.session, + }, + }; + case "git.readModel.upsert": + return { + eventId, + type: "git.read-model-upsert", + aggregateType: "project", + aggregateId: command.projectId, + occurredAt: command.createdAt, + commandId: command.commandId, + payload: { + projectId: command.projectId, + branch: command.branch, + hasWorkingTreeChanges: command.hasWorkingTreeChanges, + aheadCount: command.aheadCount, + behindCount: command.behindCount, + updatedAt: command.createdAt, + }, + }; + case "thread.turnDiff.complete": + yield* requireThread({ + readModel, + command, + threadId: command.threadId, + }); + return { + eventId, + type: "thread.turn-diff-completed", + aggregateType: "thread", + aggregateId: command.threadId, + occurredAt: command.createdAt, + commandId: command.commandId, + payload: { + threadId: command.threadId, + turnId: command.turnId, + completedAt: command.completedAt, + ...(command.status !== undefined ? { status: command.status } : {}), + files: command.files, + ...(command.assistantMessageId !== undefined + ? { assistantMessageId: command.assistantMessageId } + : {}), + ...(command.checkpointTurnCount !== undefined + ? { checkpointTurnCount: command.checkpointTurnCount } + : {}), + }, + }; + case "thread.activity.append": + yield* requireThread({ + readModel, + command, + threadId: command.threadId, + }); + return { + eventId, + type: "thread.activity-appended", + aggregateType: "thread", + aggregateId: command.threadId, + occurredAt: command.createdAt, + commandId: command.commandId, + payload: { + threadId: command.threadId, + activity: command.activity, + }, + }; + case "thread.revert": + yield* requireThread({ + readModel, + command, + threadId: command.threadId, + }); + yield* requireNonNegativeInteger({ + commandType: command.type, + field: "turnCount", + value: command.turnCount, + }); + yield* requireNonNegativeInteger({ + commandType: command.type, + field: "messageCount", + value: command.messageCount, + }); + return { + eventId, + type: "thread.reverted", + aggregateType: "thread", + aggregateId: command.threadId, + occurredAt: command.createdAt, + commandId: command.commandId, + payload: { + threadId: command.threadId, + turnCount: command.turnCount, + messageCount: command.messageCount, + }, + }; + default: { + command satisfies never; + const fallback = command as any; + return yield* new OrchestrationCommandInvariantError({ + commandType: fallback.type, + detail: `Unknown command type: ${fallback.type}`, + }); + } + } +}); diff --git a/apps/server/src/orchestration/projector.test.ts b/apps/server/src/orchestration/projector.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..4be935ac5e23f9bb83758accd202756976350886 --- /dev/null +++ b/apps/server/src/orchestration/projector.test.ts @@ -0,0 +1,258 @@ +import type { OrchestrationEvent } from "@t3tools/contracts"; +import { Effect } from "effect"; +import { describe, expect, it } from "vitest"; + +import { createEmptyReadModel, projectEvent } from "./projector.ts"; + +describe("orchestration projector", () => { + it("applies thread.created events using runtime schema decoding", async () => { + const now = new Date().toISOString(); + const model = createEmptyReadModel(now); + const event: OrchestrationEvent = { + sequence: 1, + eventId: "event-thread-created", + type: "thread.created", + aggregateType: "thread", + aggregateId: "thread-1", + occurredAt: now, + commandId: "cmd-1", + payload: { + id: "thread-1", + projectId: "project-1", + title: "demo", + model: "gpt-5-codex", + branch: null, + worktreePath: null, + createdAt: now, + updatedAt: now, + }, + }; + + const next = await Effect.runPromise(projectEvent(model, event)); + expect(next.sequence).toBe(1); + expect(next.threads).toEqual([ + { + id: "thread-1", + projectId: "project-1", + title: "demo", + model: "gpt-5-codex", + branch: null, + worktreePath: null, + createdAt: now, + updatedAt: now, + latestTurnId: null, + latestTurnStartedAt: null, + latestTurnCompletedAt: null, + latestTurnDurationMs: null, + messages: [], + session: null, + activities: [], + turnDiffSummaries: [], + error: null, + }, + ]); + }); + + it("fails when event payload cannot be decoded by runtime schema", async () => { + const now = new Date().toISOString(); + const model = createEmptyReadModel(now); + const event: OrchestrationEvent = { + sequence: 1, + eventId: "event-message-invalid", + type: "message.sent", + aggregateType: "thread", + aggregateId: "thread-1", + occurredAt: now, + commandId: "cmd-1", + payload: { + id: "message-1", + threadId: "thread-1", + text: "hello", + createdAt: now, + streaming: false, + }, + }; + + await expect(Effect.runPromise(projectEvent(model, event))).rejects.toBeDefined(); + }); + + it("keeps projector forward-compatible for unknown event types", async () => { + const now = new Date().toISOString(); + const model = createEmptyReadModel(now); + const event: OrchestrationEvent = { + sequence: 7, + eventId: "event-unknown", + type: "custom.event.ignored", + aggregateType: "unknown", + aggregateId: "aggregate-1", + occurredAt: "2026-01-01T00:00:00.000Z", + commandId: null, + payload: { test: true }, + }; + + 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([]); + }); + + it("tracks latest turn timing from session lifecycle events", async () => { + const createdAt = "2026-02-23T08:00:00.000Z"; + const startedAt = "2026-02-23T08:00:05.000Z"; + const completedAt = "2026-02-23T08:00:11.250Z"; + const model = createEmptyReadModel(createdAt); + + const createdEvent: OrchestrationEvent = { + sequence: 1, + eventId: "thread-created", + type: "thread.created", + aggregateType: "thread", + aggregateId: "thread-1", + occurredAt: createdAt, + commandId: "cmd-create", + payload: { + id: "thread-1", + projectId: "project-1", + title: "demo", + model: "gpt-5.3-codex", + branch: null, + worktreePath: null, + createdAt, + updatedAt: createdAt, + }, + }; + + const runningSessionEvent: OrchestrationEvent = { + sequence: 2, + eventId: "session-running", + type: "thread.session-set", + aggregateType: "thread", + aggregateId: "thread-1", + occurredAt: startedAt, + commandId: "cmd-running", + payload: { + threadId: "thread-1", + session: { + sessionId: "session-1", + status: "running", + provider: "codex", + threadId: "thread-1", + activeTurnId: "turn-1", + createdAt, + updatedAt: startedAt, + lastError: null, + }, + }, + }; + + const readySessionEvent: OrchestrationEvent = { + sequence: 3, + eventId: "session-ready", + type: "thread.session-set", + aggregateType: "thread", + aggregateId: "thread-1", + occurredAt: completedAt, + commandId: "cmd-ready", + payload: { + threadId: "thread-1", + session: { + sessionId: "session-1", + status: "ready", + provider: "codex", + threadId: "thread-1", + activeTurnId: null, + createdAt, + updatedAt: completedAt, + lastError: null, + }, + }, + }; + + 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"); + expect(thread?.latestTurnStartedAt).toBe(startedAt); + expect(thread?.latestTurnCompletedAt).toBe(completedAt); + expect(thread?.latestTurnDurationMs).toBe(6250); + }); + + it("marks assistant messages completed using non-streaming updates", async () => { + const createdAt = "2026-02-23T09:00:00.000Z"; + const deltaAt = "2026-02-23T09:00:01.000Z"; + const completeAt = "2026-02-23T09:00:03.500Z"; + const model = createEmptyReadModel(createdAt); + + const createdEvent: OrchestrationEvent = { + sequence: 1, + eventId: "thread-created", + type: "thread.created", + aggregateType: "thread", + aggregateId: "thread-1", + occurredAt: createdAt, + commandId: "cmd-create", + payload: { + id: "thread-1", + projectId: "project-1", + title: "demo", + model: "gpt-5.3-codex", + branch: null, + worktreePath: null, + createdAt, + updatedAt: createdAt, + }, + }; + + const deltaEvent: OrchestrationEvent = { + sequence: 2, + eventId: "assistant-delta", + type: "message.sent", + aggregateType: "thread", + aggregateId: "thread-1", + occurredAt: deltaAt, + commandId: "cmd-delta", + payload: { + id: "assistant:msg-1", + role: "assistant", + text: "hello", + threadId: "thread-1", + createdAt: deltaAt, + streaming: true, + }, + }; + + const completionEvent: OrchestrationEvent = { + sequence: 3, + eventId: "assistant-complete", + type: "message.sent", + aggregateType: "thread", + aggregateId: "thread-1", + occurredAt: completeAt, + commandId: "cmd-complete", + payload: { + id: "assistant:msg-1", + role: "assistant", + text: "", + threadId: "thread-1", + createdAt: completeAt, + streaming: false, + }, + }; + + 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({ + id: "assistant:msg-1", + role: "assistant", + text: "hello", + createdAt: deltaAt, + completedAt: completeAt, + streaming: false, + }); + }); +}); diff --git a/apps/server/src/orchestration/projector.ts b/apps/server/src/orchestration/projector.ts new file mode 100644 index 0000000000000000000000000000000000000000..168e439d33d512409d0d6dec18cc00ffe5e3044b --- /dev/null +++ b/apps/server/src/orchestration/projector.ts @@ -0,0 +1,475 @@ +import type { + OrchestrationEvent, + OrchestrationGitReadModel, + OrchestrationMessage, + OrchestrationReadModel, + OrchestrationSession, + OrchestrationThread, +} from "@t3tools/contracts"; +import { + OrchestrationThreadActivitySchema, + OrchestrationGitReadModelSchema, + OrchestrationMessageSchema, + OrchestrationSessionSchema, + OrchestrationThreadSchema, + OrchestrationTurnDiffFileSchema, + OrchestrationTurnDiffSummarySchema, +} from "@t3tools/contracts"; +import { Effect, Schema } from "effect"; + +import { toProjectorDecodeError, type OrchestrationProjectorDecodeError } from "./Errors.ts"; + +type ThreadPatch = Partial>; + +function updateThread( + threads: ReadonlyArray, + threadId: string, + patch: ThreadPatch, +): OrchestrationThread[] { + return threads.map((thread) => (thread.id === threadId ? { ...thread, ...patch } : thread)); +} + +function decodeForEvent( + schema: Schema.Schema, + value: unknown, + eventType: string, + field: string, +): Effect.Effect { + return Schema.decodeUnknownEffect(schema)(value).pipe( + Effect.mapError(toProjectorDecodeError(`${eventType}:${field}`)), + ) as Effect.Effect; +} + +const ThreadCreatedPayloadSchema = Schema.Struct({ + id: Schema.String, + projectId: Schema.String, + title: Schema.String, + model: Schema.String, + branch: Schema.NullOr(Schema.String), + worktreePath: Schema.NullOr(Schema.String), + createdAt: Schema.String, + updatedAt: Schema.String, +}); + +const ThreadMetaUpdatedPayloadSchema = Schema.Struct({ + threadId: Schema.String, + title: Schema.optional(Schema.String), + model: Schema.optional(Schema.String), + branch: Schema.optional(Schema.NullOr(Schema.String)), + worktreePath: Schema.optional(Schema.NullOr(Schema.String)), + updatedAt: Schema.String, +}); + +const MessageSentPayloadSchema = Schema.Struct({ + id: Schema.String, + role: Schema.Literals(["user", "assistant"]), + text: Schema.String, + threadId: Schema.String, + createdAt: Schema.String, + streaming: Schema.Boolean, +}); + +const ThreadSessionSetPayloadSchema = Schema.Struct({ + threadId: Schema.String, + session: OrchestrationSessionSchema, +}); + +const ThreadTurnDiffCompletedPayloadSchema = Schema.Struct({ + threadId: Schema.String, + turnId: Schema.String, + completedAt: Schema.String, + status: Schema.optional(Schema.String), + files: Schema.Array(OrchestrationTurnDiffFileSchema), + assistantMessageId: Schema.optional(Schema.String), + checkpointTurnCount: Schema.optional(Schema.Number), +}); + +const ThreadRevertedPayloadSchema = Schema.Struct({ + threadId: Schema.String, + turnCount: Schema.Number, + messageCount: Schema.Number, +}); + +const ThreadActivityAppendedPayloadSchema = Schema.Struct({ + threadId: Schema.String, + activity: OrchestrationThreadActivitySchema, +}); + +export function createEmptyReadModel(nowIso: string): OrchestrationReadModel { + return { + sequence: 0, + threads: [], + gitByProjectId: {}, + updatedAt: nowIso, + }; +} + +export function projectEvent( + model: OrchestrationReadModel, + event: OrchestrationEvent, +): Effect.Effect { + const nextBase: OrchestrationReadModel = { + ...model, + sequence: event.sequence, + updatedAt: event.occurredAt, + }; + + switch (event.type) { + case "thread.created": + return Effect.gen(function* () { + const payload = yield* decodeForEvent( + ThreadCreatedPayloadSchema, + event.payload, + event.type, + "payload", + ); + const thread: OrchestrationThread = yield* decodeForEvent( + OrchestrationThreadSchema, + { + id: payload.id, + projectId: payload.projectId, + title: payload.title, + model: payload.model, + branch: payload.branch, + worktreePath: payload.worktreePath, + createdAt: payload.createdAt, + updatedAt: payload.updatedAt, + latestTurnId: null, + latestTurnStartedAt: null, + latestTurnCompletedAt: null, + latestTurnDurationMs: null, + messages: [], + session: null, + turnDiffSummaries: [], + activities: [], + error: null, + }, + event.type, + "thread", + ); + const existing = nextBase.threads.find((entry) => entry.id === thread.id); + return { + ...nextBase, + threads: existing + ? nextBase.threads.map((entry) => (entry.id === thread.id ? thread : entry)) + : [...nextBase.threads, thread], + }; + }); + case "thread.deleted": + return Effect.succeed({ + ...nextBase, + threads: nextBase.threads.filter((thread) => thread.id !== event.aggregateId), + }); + case "thread.meta-updated": + return decodeForEvent( + ThreadMetaUpdatedPayloadSchema, + event.payload, + event.type, + "payload", + ).pipe( + Effect.map((payload) => { + const thread = nextBase.threads.find((entry) => entry.id === payload.threadId); + if (!thread) { + return nextBase; + } + return { + ...nextBase, + threads: updateThread(nextBase.threads, payload.threadId, { + ...(payload.title !== undefined ? { title: payload.title } : {}), + ...(payload.model !== undefined ? { model: payload.model } : {}), + ...(payload.branch !== undefined ? { branch: payload.branch } : {}), + ...(payload.worktreePath !== undefined ? { worktreePath: payload.worktreePath } : {}), + updatedAt: payload.updatedAt, + }), + }; + }), + ); + case "message.sent": + return Effect.gen(function* () { + const payload = yield* decodeForEvent( + MessageSentPayloadSchema, + event.payload, + event.type, + "payload", + ); + const targetThread = nextBase.threads.find((thread) => thread.id === payload.threadId); + if (!targetThread) { + return nextBase; + } + + const message: OrchestrationMessage = yield* decodeForEvent( + OrchestrationMessageSchema, + { + id: payload.id, + role: payload.role, + text: payload.text, + createdAt: payload.createdAt, + ...(payload.streaming ? {} : { completedAt: event.occurredAt }), + streaming: payload.streaming, + }, + event.type, + "message", + ); + const existingMessage = targetThread.messages.find((entry) => entry.id === message.id); + const nextMessages = existingMessage + ? targetThread.messages.map((entry) => + entry.id === message.id + ? { + ...entry, + text: message.streaming + ? `${entry.text}${message.text}` + : message.text.length > 0 + ? message.text + : entry.text, + streaming: message.streaming, + createdAt: entry.createdAt, + ...(message.streaming + ? { completedAt: undefined } + : { completedAt: message.completedAt ?? event.occurredAt }), + } + : entry, + ) + : !message.streaming && message.text.length === 0 + ? targetThread.messages + : [...targetThread.messages, message]; + + return { + ...nextBase, + threads: updateThread(nextBase.threads, payload.threadId, { + messages: nextMessages, + updatedAt: event.occurredAt, + }), + }; + }); + case "thread.session-set": + return Effect.gen(function* () { + const payload = yield* decodeForEvent( + ThreadSessionSetPayloadSchema, + event.payload, + event.type, + "payload", + ); + const thread = nextBase.threads.find((entry) => entry.id === payload.threadId); + if (!thread) { + return nextBase; + } + + const session: OrchestrationSession = yield* decodeForEvent( + OrchestrationSessionSchema, + { + ...payload.session, + threadId: payload.threadId, + }, + event.type, + "session", + ); + + const sessionUpdatedAt = session.updatedAt; + const turnTimingPatch: ThreadPatch = + session.status === "running" && session.activeTurnId + ? (() => { + const isSameTurn = thread.latestTurnId === session.activeTurnId; + return { + latestTurnId: session.activeTurnId, + latestTurnStartedAt: + isSameTurn && thread.latestTurnStartedAt + ? thread.latestTurnStartedAt + : sessionUpdatedAt, + ...(isSameTurn + ? {} + : { + latestTurnCompletedAt: null, + latestTurnDurationMs: null, + }), + }; + })() + : session.activeTurnId === null && thread.latestTurnId && thread.latestTurnStartedAt + ? (() => { + const latestTurnCompletedAt = thread.latestTurnCompletedAt ?? sessionUpdatedAt; + const startedAtMs = Date.parse(thread.latestTurnStartedAt); + const completedAtMs = Date.parse(latestTurnCompletedAt); + const latestTurnDurationMs = + !Number.isNaN(startedAtMs) && + !Number.isNaN(completedAtMs) && + completedAtMs >= startedAtMs + ? completedAtMs - startedAtMs + : thread.latestTurnDurationMs; + + return { + latestTurnCompletedAt, + latestTurnDurationMs, + }; + })() + : {}; + + return { + ...nextBase, + threads: updateThread(nextBase.threads, payload.threadId, { + session, + updatedAt: event.occurredAt, + error: session.lastError, + ...turnTimingPatch, + }), + }; + }); + case "thread.turn-diff-completed": + return Effect.gen(function* () { + const payload = yield* decodeForEvent( + ThreadTurnDiffCompletedPayloadSchema, + event.payload, + event.type, + "payload", + ); + const thread = nextBase.threads.find((entry) => entry.id === payload.threadId); + if (!thread) { + return nextBase; + } + + const nextSummary = yield* decodeForEvent( + OrchestrationTurnDiffSummarySchema, + { + turnId: payload.turnId, + completedAt: payload.completedAt, + ...(payload.status !== undefined ? { status: payload.status } : {}), + files: payload.files, + ...(payload.assistantMessageId !== undefined + ? { assistantMessageId: payload.assistantMessageId } + : {}), + ...(payload.checkpointTurnCount !== undefined + ? { checkpointTurnCount: payload.checkpointTurnCount } + : {}), + }, + event.type, + "turnDiffSummary", + ); + const turnDiffSummaries = [ + ...thread.turnDiffSummaries.filter((summary) => summary.turnId !== payload.turnId), + nextSummary, + ].toSorted((left, right) => left.completedAt.localeCompare(right.completedAt)); + const startedAtMs = thread.latestTurnStartedAt + ? Date.parse(thread.latestTurnStartedAt) + : Number.NaN; + const completedAtMs = Date.parse(nextSummary.completedAt); + const latestTurnDurationMs = + !Number.isNaN(startedAtMs) && + !Number.isNaN(completedAtMs) && + completedAtMs >= startedAtMs + ? completedAtMs - startedAtMs + : thread.latestTurnDurationMs; + + return { + ...nextBase, + threads: updateThread(nextBase.threads, payload.threadId, { + latestTurnId: payload.turnId, + latestTurnCompletedAt: nextSummary.completedAt, + latestTurnDurationMs, + turnDiffSummaries, + updatedAt: event.occurredAt, + }), + }; + }); + case "thread.reverted": + return decodeForEvent(ThreadRevertedPayloadSchema, event.payload, event.type, "payload").pipe( + Effect.map((payload) => { + const thread = nextBase.threads.find((entry) => entry.id === payload.threadId); + if (!thread) { + return nextBase; + } + + const targetTurnCount = Math.max(0, Math.floor(payload.turnCount)); + const targetMessageCount = Math.max(0, Math.floor(payload.messageCount)); + const sortedSummaries = [...thread.turnDiffSummaries].toSorted((left, right) => + left.completedAt.localeCompare(right.completedAt), + ); + const inferredTurnCountByTurnId = new Map(); + for (let index = 0; index < sortedSummaries.length; index += 1) { + const summary = sortedSummaries[index]; + if (!summary) continue; + inferredTurnCountByTurnId.set(summary.turnId, index + 1); + } + + const turnDiffSummaries = thread.turnDiffSummaries + .filter((summary) => { + const checkpointTurnCount = + summary.checkpointTurnCount ?? inferredTurnCountByTurnId.get(summary.turnId) ?? 0; + return checkpointTurnCount <= targetTurnCount; + }) + .toSorted((left, right) => left.completedAt.localeCompare(right.completedAt)); + const latestSummary = + turnDiffSummaries.length > 0 ? turnDiffSummaries[turnDiffSummaries.length - 1] : null; + const activities = thread.activities.filter((activity) => { + if (!activity.turnId) { + return true; + } + const checkpointTurnCount = inferredTurnCountByTurnId.get(activity.turnId); + if (checkpointTurnCount === undefined) { + return true; + } + return checkpointTurnCount <= targetTurnCount; + }); + + return { + ...nextBase, + threads: updateThread(nextBase.threads, payload.threadId, { + messages: thread.messages.slice( + 0, + Math.min(targetMessageCount, thread.messages.length), + ), + turnDiffSummaries, + latestTurnId: latestSummary?.turnId ?? null, + latestTurnStartedAt: null, + latestTurnCompletedAt: latestSummary?.completedAt ?? null, + latestTurnDurationMs: null, + activities, + updatedAt: event.occurredAt, + }), + }; + }), + ); + case "thread.activity-appended": + return decodeForEvent( + ThreadActivityAppendedPayloadSchema, + event.payload, + event.type, + "payload", + ).pipe( + Effect.map((payload) => { + const thread = nextBase.threads.find((entry) => entry.id === payload.threadId); + if (!thread) { + return nextBase; + } + const activities = [ + ...thread.activities.filter((activity) => activity.id !== payload.activity.id), + payload.activity, + ] + .toSorted((left, right) => left.createdAt.localeCompare(right.createdAt)) + .slice(-500); + + return { + ...nextBase, + threads: updateThread(nextBase.threads, payload.threadId, { + activities, + updatedAt: event.occurredAt, + }), + }; + }), + ); + case "git.read-model-upsert": + return decodeForEvent( + OrchestrationGitReadModelSchema, + event.payload, + event.type, + "payload", + ).pipe( + Effect.map((gitReadModel: OrchestrationGitReadModel) => ({ + ...nextBase, + gitByProjectId: { + ...nextBase.gitByProjectId, + [gitReadModel.projectId]: gitReadModel, + }, + })), + ); + default: + return Effect.succeed(nextBase); + } +} diff --git a/apps/server/src/orchestration/reducer.test.ts b/apps/server/src/orchestration/reducer.test.ts deleted file mode 100644 index 97cb5a9d14afef839a0b941ee06308c479cbf70e..0000000000000000000000000000000000000000 --- a/apps/server/src/orchestration/reducer.test.ts +++ /dev/null @@ -1,258 +0,0 @@ -import type { OrchestrationEvent } from "@t3tools/contracts"; -import { Effect } from "effect"; -import { describe, expect, it } from "vitest"; - -import { createEmptyReadModel, reduceEvent } from "./reducer.ts"; - -describe("orchestration reducer", () => { - it("applies thread.created events using runtime schema decoding", async () => { - const now = new Date().toISOString(); - const model = createEmptyReadModel(now); - const event: OrchestrationEvent = { - sequence: 1, - eventId: "event-thread-created", - type: "thread.created", - aggregateType: "thread", - aggregateId: "thread-1", - occurredAt: now, - commandId: "cmd-1", - payload: { - id: "thread-1", - projectId: "project-1", - title: "demo", - model: "gpt-5-codex", - branch: null, - worktreePath: null, - createdAt: now, - updatedAt: now, - }, - }; - - const next = await Effect.runPromise(reduceEvent(model, event)); - expect(next.sequence).toBe(1); - expect(next.threads).toEqual([ - { - id: "thread-1", - projectId: "project-1", - title: "demo", - model: "gpt-5-codex", - branch: null, - worktreePath: null, - createdAt: now, - updatedAt: now, - latestTurnId: null, - latestTurnStartedAt: null, - latestTurnCompletedAt: null, - latestTurnDurationMs: null, - messages: [], - session: null, - activities: [], - turnDiffSummaries: [], - error: null, - }, - ]); - }); - - it("fails when event payload cannot be decoded by runtime schema", async () => { - const now = new Date().toISOString(); - const model = createEmptyReadModel(now); - const event: OrchestrationEvent = { - sequence: 1, - eventId: "event-message-invalid", - type: "message.sent", - aggregateType: "thread", - aggregateId: "thread-1", - occurredAt: now, - commandId: "cmd-1", - payload: { - id: "message-1", - threadId: "thread-1", - text: "hello", - createdAt: now, - streaming: false, - }, - }; - - await expect(Effect.runPromise(reduceEvent(model, event))).rejects.toBeDefined(); - }); - - it("keeps reducer forward-compatible for unknown event types", async () => { - const now = new Date().toISOString(); - const model = createEmptyReadModel(now); - const event: OrchestrationEvent = { - sequence: 7, - eventId: "event-unknown", - type: "custom.event.ignored", - aggregateType: "unknown", - aggregateId: "aggregate-1", - occurredAt: "2026-01-01T00:00:00.000Z", - commandId: null, - payload: { test: true }, - }; - - const next = await Effect.runPromise(reduceEvent(model, event)); - expect(next.sequence).toBe(7); - expect(next.updatedAt).toBe("2026-01-01T00:00:00.000Z"); - expect(next.threads).toEqual([]); - }); - - it("tracks latest turn timing from session lifecycle events", async () => { - const createdAt = "2026-02-23T08:00:00.000Z"; - const startedAt = "2026-02-23T08:00:05.000Z"; - const completedAt = "2026-02-23T08:00:11.250Z"; - const model = createEmptyReadModel(createdAt); - - const createdEvent: OrchestrationEvent = { - sequence: 1, - eventId: "thread-created", - type: "thread.created", - aggregateType: "thread", - aggregateId: "thread-1", - occurredAt: createdAt, - commandId: "cmd-create", - payload: { - id: "thread-1", - projectId: "project-1", - title: "demo", - model: "gpt-5.3-codex", - branch: null, - worktreePath: null, - createdAt, - updatedAt: createdAt, - }, - }; - - const runningSessionEvent: OrchestrationEvent = { - sequence: 2, - eventId: "session-running", - type: "thread.session-set", - aggregateType: "thread", - aggregateId: "thread-1", - occurredAt: startedAt, - commandId: "cmd-running", - payload: { - threadId: "thread-1", - session: { - sessionId: "session-1", - status: "running", - provider: "codex", - threadId: "thread-1", - activeTurnId: "turn-1", - createdAt, - updatedAt: startedAt, - lastError: null, - }, - }, - }; - - const readySessionEvent: OrchestrationEvent = { - sequence: 3, - eventId: "session-ready", - type: "thread.session-set", - aggregateType: "thread", - aggregateId: "thread-1", - occurredAt: completedAt, - commandId: "cmd-ready", - payload: { - threadId: "thread-1", - session: { - sessionId: "session-1", - status: "ready", - provider: "codex", - threadId: "thread-1", - activeTurnId: null, - createdAt, - updatedAt: completedAt, - lastError: null, - }, - }, - }; - - 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 thread = afterReady.threads[0]; - - expect(thread?.latestTurnId).toBe("turn-1"); - expect(thread?.latestTurnStartedAt).toBe(startedAt); - expect(thread?.latestTurnCompletedAt).toBe(completedAt); - expect(thread?.latestTurnDurationMs).toBe(6250); - }); - - it("marks assistant messages completed using non-streaming updates", async () => { - const createdAt = "2026-02-23T09:00:00.000Z"; - const deltaAt = "2026-02-23T09:00:01.000Z"; - const completeAt = "2026-02-23T09:00:03.500Z"; - const model = createEmptyReadModel(createdAt); - - const createdEvent: OrchestrationEvent = { - sequence: 1, - eventId: "thread-created", - type: "thread.created", - aggregateType: "thread", - aggregateId: "thread-1", - occurredAt: createdAt, - commandId: "cmd-create", - payload: { - id: "thread-1", - projectId: "project-1", - title: "demo", - model: "gpt-5.3-codex", - branch: null, - worktreePath: null, - createdAt, - updatedAt: createdAt, - }, - }; - - const deltaEvent: OrchestrationEvent = { - sequence: 2, - eventId: "assistant-delta", - type: "message.sent", - aggregateType: "thread", - aggregateId: "thread-1", - occurredAt: deltaAt, - commandId: "cmd-delta", - payload: { - id: "assistant:msg-1", - role: "assistant", - text: "hello", - threadId: "thread-1", - createdAt: deltaAt, - streaming: true, - }, - }; - - const completionEvent: OrchestrationEvent = { - sequence: 3, - eventId: "assistant-complete", - type: "message.sent", - aggregateType: "thread", - aggregateId: "thread-1", - occurredAt: completeAt, - commandId: "cmd-complete", - payload: { - id: "assistant:msg-1", - role: "assistant", - text: "", - threadId: "thread-1", - createdAt: completeAt, - streaming: false, - }, - }; - - 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 message = afterComplete.threads[0]?.messages[0]; - - expect(message).toEqual({ - id: "assistant:msg-1", - role: "assistant", - text: "hello", - createdAt: deltaAt, - completedAt: completeAt, - streaming: false, - }); - }); -}); diff --git a/apps/server/src/orchestration/reducer.ts b/apps/server/src/orchestration/reducer.ts deleted file mode 100644 index bf643572eebf597946d7bbe06bf41ca44bd27b4c..0000000000000000000000000000000000000000 --- a/apps/server/src/orchestration/reducer.ts +++ /dev/null @@ -1,475 +0,0 @@ -import type { - OrchestrationEvent, - OrchestrationGitReadModel, - OrchestrationMessage, - OrchestrationReadModel, - OrchestrationSession, - OrchestrationThread, -} from "@t3tools/contracts"; -import { - OrchestrationThreadActivitySchema, - OrchestrationGitReadModelSchema, - OrchestrationMessageSchema, - OrchestrationSessionSchema, - OrchestrationThreadSchema, - OrchestrationTurnDiffFileSchema, - OrchestrationTurnDiffSummarySchema, -} from "@t3tools/contracts"; -import { Effect, Schema } from "effect"; - -import { toReducerDecodeError, type OrchestrationReducerDecodeError } from "./Errors.ts"; - -type ThreadPatch = Partial>; - -function updateThread( - threads: ReadonlyArray, - threadId: string, - patch: ThreadPatch, -): OrchestrationThread[] { - return threads.map((thread) => (thread.id === threadId ? { ...thread, ...patch } : thread)); -} - -function decodeForEvent( - schema: Schema.Schema, - value: unknown, - eventType: string, - field: string, -): Effect.Effect { - return Schema.decodeUnknownEffect(schema)(value).pipe( - Effect.mapError(toReducerDecodeError(`${eventType}:${field}`)), - ) as Effect.Effect; -} - -const ThreadCreatedPayloadSchema = Schema.Struct({ - id: Schema.String, - projectId: Schema.String, - title: Schema.String, - model: Schema.String, - branch: Schema.NullOr(Schema.String), - worktreePath: Schema.NullOr(Schema.String), - createdAt: Schema.String, - updatedAt: Schema.String, -}); - -const ThreadMetaUpdatedPayloadSchema = Schema.Struct({ - threadId: Schema.String, - title: Schema.optional(Schema.String), - model: Schema.optional(Schema.String), - branch: Schema.optional(Schema.NullOr(Schema.String)), - worktreePath: Schema.optional(Schema.NullOr(Schema.String)), - updatedAt: Schema.String, -}); - -const MessageSentPayloadSchema = Schema.Struct({ - id: Schema.String, - role: Schema.Literals(["user", "assistant"]), - text: Schema.String, - threadId: Schema.String, - createdAt: Schema.String, - streaming: Schema.Boolean, -}); - -const ThreadSessionSetPayloadSchema = Schema.Struct({ - threadId: Schema.String, - session: OrchestrationSessionSchema, -}); - -const ThreadTurnDiffCompletedPayloadSchema = Schema.Struct({ - threadId: Schema.String, - turnId: Schema.String, - completedAt: Schema.String, - status: Schema.optional(Schema.String), - files: Schema.Array(OrchestrationTurnDiffFileSchema), - assistantMessageId: Schema.optional(Schema.String), - checkpointTurnCount: Schema.optional(Schema.Number), -}); - -const ThreadRevertedPayloadSchema = Schema.Struct({ - threadId: Schema.String, - turnCount: Schema.Number, - messageCount: Schema.Number, -}); - -const ThreadActivityAppendedPayloadSchema = Schema.Struct({ - threadId: Schema.String, - activity: OrchestrationThreadActivitySchema, -}); - -export function createEmptyReadModel(nowIso: string): OrchestrationReadModel { - return { - sequence: 0, - threads: [], - gitByProjectId: {}, - updatedAt: nowIso, - }; -} - -export function reduceEvent( - model: OrchestrationReadModel, - event: OrchestrationEvent, -): Effect.Effect { - const nextBase: OrchestrationReadModel = { - ...model, - sequence: event.sequence, - updatedAt: event.occurredAt, - }; - - switch (event.type) { - case "thread.created": - return Effect.gen(function* () { - const payload = yield* decodeForEvent( - ThreadCreatedPayloadSchema, - event.payload, - event.type, - "payload", - ); - const thread: OrchestrationThread = yield* decodeForEvent( - OrchestrationThreadSchema, - { - id: payload.id, - projectId: payload.projectId, - title: payload.title, - model: payload.model, - branch: payload.branch, - worktreePath: payload.worktreePath, - createdAt: payload.createdAt, - updatedAt: payload.updatedAt, - latestTurnId: null, - latestTurnStartedAt: null, - latestTurnCompletedAt: null, - latestTurnDurationMs: null, - messages: [], - session: null, - turnDiffSummaries: [], - activities: [], - error: null, - }, - event.type, - "thread", - ); - const existing = nextBase.threads.find((entry) => entry.id === thread.id); - return { - ...nextBase, - threads: existing - ? nextBase.threads.map((entry) => (entry.id === thread.id ? thread : entry)) - : [...nextBase.threads, thread], - }; - }); - case "thread.deleted": - return Effect.succeed({ - ...nextBase, - threads: nextBase.threads.filter((thread) => thread.id !== event.aggregateId), - }); - case "thread.meta-updated": - return decodeForEvent( - ThreadMetaUpdatedPayloadSchema, - event.payload, - event.type, - "payload", - ).pipe( - Effect.map((payload) => { - const thread = nextBase.threads.find((entry) => entry.id === payload.threadId); - if (!thread) { - return nextBase; - } - return { - ...nextBase, - threads: updateThread(nextBase.threads, payload.threadId, { - ...(payload.title !== undefined ? { title: payload.title } : {}), - ...(payload.model !== undefined ? { model: payload.model } : {}), - ...(payload.branch !== undefined ? { branch: payload.branch } : {}), - ...(payload.worktreePath !== undefined ? { worktreePath: payload.worktreePath } : {}), - updatedAt: payload.updatedAt, - }), - }; - }), - ); - case "message.sent": - return Effect.gen(function* () { - const payload = yield* decodeForEvent( - MessageSentPayloadSchema, - event.payload, - event.type, - "payload", - ); - const targetThread = nextBase.threads.find((thread) => thread.id === payload.threadId); - if (!targetThread) { - return nextBase; - } - - const message: OrchestrationMessage = yield* decodeForEvent( - OrchestrationMessageSchema, - { - id: payload.id, - role: payload.role, - text: payload.text, - createdAt: payload.createdAt, - ...(payload.streaming ? {} : { completedAt: event.occurredAt }), - streaming: payload.streaming, - }, - event.type, - "message", - ); - const existingMessage = targetThread.messages.find((entry) => entry.id === message.id); - const nextMessages = existingMessage - ? targetThread.messages.map((entry) => - entry.id === message.id - ? { - ...entry, - text: message.streaming - ? `${entry.text}${message.text}` - : message.text.length > 0 - ? message.text - : entry.text, - streaming: message.streaming, - createdAt: entry.createdAt, - ...(message.streaming - ? { completedAt: undefined } - : { completedAt: message.completedAt ?? event.occurredAt }), - } - : entry, - ) - : !message.streaming && message.text.length === 0 - ? targetThread.messages - : [...targetThread.messages, message]; - - return { - ...nextBase, - threads: updateThread(nextBase.threads, payload.threadId, { - messages: nextMessages, - updatedAt: event.occurredAt, - }), - }; - }); - case "thread.session-set": - return Effect.gen(function* () { - const payload = yield* decodeForEvent( - ThreadSessionSetPayloadSchema, - event.payload, - event.type, - "payload", - ); - const thread = nextBase.threads.find((entry) => entry.id === payload.threadId); - if (!thread) { - return nextBase; - } - - const session: OrchestrationSession = yield* decodeForEvent( - OrchestrationSessionSchema, - { - ...payload.session, - threadId: payload.threadId, - }, - event.type, - "session", - ); - - const sessionUpdatedAt = session.updatedAt; - const turnTimingPatch: ThreadPatch = - session.status === "running" && session.activeTurnId - ? (() => { - const isSameTurn = thread.latestTurnId === session.activeTurnId; - return { - latestTurnId: session.activeTurnId, - latestTurnStartedAt: - isSameTurn && thread.latestTurnStartedAt - ? thread.latestTurnStartedAt - : sessionUpdatedAt, - ...(isSameTurn - ? {} - : { - latestTurnCompletedAt: null, - latestTurnDurationMs: null, - }), - }; - })() - : session.activeTurnId === null && thread.latestTurnId && thread.latestTurnStartedAt - ? (() => { - const latestTurnCompletedAt = thread.latestTurnCompletedAt ?? sessionUpdatedAt; - const startedAtMs = Date.parse(thread.latestTurnStartedAt); - const completedAtMs = Date.parse(latestTurnCompletedAt); - const latestTurnDurationMs = - !Number.isNaN(startedAtMs) && - !Number.isNaN(completedAtMs) && - completedAtMs >= startedAtMs - ? completedAtMs - startedAtMs - : thread.latestTurnDurationMs; - - return { - latestTurnCompletedAt, - latestTurnDurationMs, - }; - })() - : {}; - - return { - ...nextBase, - threads: updateThread(nextBase.threads, payload.threadId, { - session, - updatedAt: event.occurredAt, - error: session.lastError, - ...turnTimingPatch, - }), - }; - }); - case "thread.turn-diff-completed": - return Effect.gen(function* () { - const payload = yield* decodeForEvent( - ThreadTurnDiffCompletedPayloadSchema, - event.payload, - event.type, - "payload", - ); - const thread = nextBase.threads.find((entry) => entry.id === payload.threadId); - if (!thread) { - return nextBase; - } - - const nextSummary = yield* decodeForEvent( - OrchestrationTurnDiffSummarySchema, - { - turnId: payload.turnId, - completedAt: payload.completedAt, - ...(payload.status !== undefined ? { status: payload.status } : {}), - files: payload.files, - ...(payload.assistantMessageId !== undefined - ? { assistantMessageId: payload.assistantMessageId } - : {}), - ...(payload.checkpointTurnCount !== undefined - ? { checkpointTurnCount: payload.checkpointTurnCount } - : {}), - }, - event.type, - "turnDiffSummary", - ); - const turnDiffSummaries = [ - ...thread.turnDiffSummaries.filter((summary) => summary.turnId !== payload.turnId), - nextSummary, - ].toSorted((left, right) => left.completedAt.localeCompare(right.completedAt)); - const startedAtMs = thread.latestTurnStartedAt - ? Date.parse(thread.latestTurnStartedAt) - : Number.NaN; - const completedAtMs = Date.parse(nextSummary.completedAt); - const latestTurnDurationMs = - !Number.isNaN(startedAtMs) && - !Number.isNaN(completedAtMs) && - completedAtMs >= startedAtMs - ? completedAtMs - startedAtMs - : thread.latestTurnDurationMs; - - return { - ...nextBase, - threads: updateThread(nextBase.threads, payload.threadId, { - latestTurnId: payload.turnId, - latestTurnCompletedAt: nextSummary.completedAt, - latestTurnDurationMs, - turnDiffSummaries, - updatedAt: event.occurredAt, - }), - }; - }); - case "thread.reverted": - return decodeForEvent(ThreadRevertedPayloadSchema, event.payload, event.type, "payload").pipe( - Effect.map((payload) => { - const thread = nextBase.threads.find((entry) => entry.id === payload.threadId); - if (!thread) { - return nextBase; - } - - const targetTurnCount = Math.max(0, Math.floor(payload.turnCount)); - const targetMessageCount = Math.max(0, Math.floor(payload.messageCount)); - const sortedSummaries = [...thread.turnDiffSummaries].toSorted((left, right) => - left.completedAt.localeCompare(right.completedAt), - ); - const inferredTurnCountByTurnId = new Map(); - for (let index = 0; index < sortedSummaries.length; index += 1) { - const summary = sortedSummaries[index]; - if (!summary) continue; - inferredTurnCountByTurnId.set(summary.turnId, index + 1); - } - - const turnDiffSummaries = thread.turnDiffSummaries - .filter((summary) => { - const checkpointTurnCount = - summary.checkpointTurnCount ?? inferredTurnCountByTurnId.get(summary.turnId) ?? 0; - return checkpointTurnCount <= targetTurnCount; - }) - .toSorted((left, right) => left.completedAt.localeCompare(right.completedAt)); - const latestSummary = - turnDiffSummaries.length > 0 ? turnDiffSummaries[turnDiffSummaries.length - 1] : null; - const activities = thread.activities.filter((activity) => { - if (!activity.turnId) { - return true; - } - const checkpointTurnCount = inferredTurnCountByTurnId.get(activity.turnId); - if (checkpointTurnCount === undefined) { - return true; - } - return checkpointTurnCount <= targetTurnCount; - }); - - return { - ...nextBase, - threads: updateThread(nextBase.threads, payload.threadId, { - messages: thread.messages.slice( - 0, - Math.min(targetMessageCount, thread.messages.length), - ), - turnDiffSummaries, - latestTurnId: latestSummary?.turnId ?? null, - latestTurnStartedAt: null, - latestTurnCompletedAt: latestSummary?.completedAt ?? null, - latestTurnDurationMs: null, - activities, - updatedAt: event.occurredAt, - }), - }; - }), - ); - case "thread.activity-appended": - return decodeForEvent( - ThreadActivityAppendedPayloadSchema, - event.payload, - event.type, - "payload", - ).pipe( - Effect.map((payload) => { - const thread = nextBase.threads.find((entry) => entry.id === payload.threadId); - if (!thread) { - return nextBase; - } - const activities = [ - ...thread.activities.filter((activity) => activity.id !== payload.activity.id), - payload.activity, - ] - .toSorted((left, right) => left.createdAt.localeCompare(right.createdAt)) - .slice(-500); - - return { - ...nextBase, - threads: updateThread(nextBase.threads, payload.threadId, { - activities, - updatedAt: event.occurredAt, - }), - }; - }), - ); - case "git.read-model-upsert": - return decodeForEvent( - OrchestrationGitReadModelSchema, - event.payload, - event.type, - "payload", - ).pipe( - Effect.map((gitReadModel: OrchestrationGitReadModel) => ({ - ...nextBase, - gitByProjectId: { - ...nextBase.gitByProjectId, - [gitReadModel.projectId]: gitReadModel, - }, - })), - ); - default: - return Effect.succeed(nextBase); - } -} 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/OrchestrationEventStore.test.ts b/apps/server/src/persistence/Layers/OrchestrationEventStore.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..19f72ff4faf8a69ae14ed384088415208f1164f2 --- /dev/null +++ b/apps/server/src/persistence/Layers/OrchestrationEventStore.test.ts @@ -0,0 +1,39 @@ +import { Effect, Layer, Stream } from "effect"; +import { assert, it } from "@effect/vitest"; + +import { OrchestrationEventStore } from "../Services/OrchestrationEventStore.ts"; +import { SqlitePersistenceMemory } from "./Sqlite.ts"; +import { OrchestrationEventStoreLive } from "./OrchestrationEventStore.ts"; +import { NodeServices } from "@effect/platform-node"; + +const layer = it.layer( + Layer.mergeAll( + Layer.provide(OrchestrationEventStoreLive, SqlitePersistenceMemory), + NodeServices.layer, + ), +); + +layer("OrchestrationEventStore", (it) => { + it.effect("stores and replays events", () => + Effect.gen(function* () { + const createdAt = new Date().toISOString(); + + const eventStore = yield* OrchestrationEventStore; + const saved = yield* eventStore.append({ + eventId: "event-1", + type: "thread.created", + aggregateType: "thread", + aggregateId: "thread-1", + occurredAt: createdAt, + commandId: "cmd-1", + payload: { id: "thread-1", projectId: "project-1", title: "demo" }, + }); + assert.equal(saved.sequence, 1); + + 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/OrchestrationEventStore.ts b/apps/server/src/persistence/Layers/OrchestrationEventStore.ts new file mode 100644 index 0000000000000000000000000000000000000000..c1a79890c2905a8c714cea11e7b92ed6581dd24c --- /dev/null +++ b/apps/server/src/persistence/Layers/OrchestrationEventStore.ts @@ -0,0 +1,210 @@ +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, Stream } from "effect"; + +import { + toPersistenceDecodeCauseError, + toPersistenceDecodeError, + toPersistenceSerializationError, + toPersistenceSqlError, + type OrchestrationEventStoreError, +} from "../Errors.ts"; +import { + OrchestrationEventStore, + type OrchestrationEventStoreShape, +} from "../Services/OrchestrationEventStore.ts"; + +const decodeEvent = Schema.decodeUnknownEffect(OrchestrationEventSchema); + +const EventRowSchema = Schema.Struct({ + sequence: Schema.Number, + eventId: Schema.String, + type: Schema.String, + aggregateType: Schema.String, + aggregateId: Schema.String, + occurredAt: Schema.String, + commandId: Schema.NullOr(Schema.String), + payloadJson: Schema.String, +}); + +const AppendEventRequestSchema = Schema.Struct({ + eventId: Schema.String, + type: Schema.String, + aggregateType: Schema.String, + aggregateId: Schema.String, + occurredAt: Schema.String, + commandId: Schema.NullOr(Schema.String), + payloadJson: Schema.String, +}); + +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 { + return Effect.try({ + try: () => JSON.parse(row.payloadJson) as unknown, + catch: toPersistenceDecodeCauseError(`${operation}:parsePayloadJson`), + }).pipe( + Effect.flatMap((payload) => + decodeEvent({ + sequence: row.sequence, + eventId: row.eventId, + type: row.type, + aggregateType: row.aggregateType, + aggregateId: row.aggregateId, + occurredAt: row.occurredAt, + commandId: row.commandId, + payload, + }).pipe(Effect.mapError(toPersistenceDecodeError(`${operation}:decodeEvent`))), + ), + ); +} + +const makeEventStore = Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + const appendEventRow = SqlSchema.findOne({ + Request: AppendEventRequestSchema, + Result: EventRowSchema, + execute: (request) => + sql` + INSERT INTO orchestration_events ( + event_id, + event_type, + aggregate_type, + aggregate_id, + occurred_at, + command_id, + payload_json + ) + VALUES ( + ${request.eventId}, + ${request.type}, + ${request.aggregateType}, + ${request.aggregateId}, + ${request.occurredAt}, + ${request.commandId}, + ${request.payloadJson} + ) + RETURNING + sequence, + event_id AS "eventId", + event_type AS "type", + aggregate_type AS "aggregateType", + aggregate_id AS "aggregateId", + occurred_at AS "occurredAt", + command_id AS "commandId", + payload_json AS "payloadJson" + `, + }); + + const readEventRowsFromSequence = SqlSchema.findAll({ + Request: ReadFromSequenceRequestSchema, + Result: EventRowSchema, + execute: (request) => + sql` + SELECT + sequence, + event_id AS "eventId", + event_type AS "type", + aggregate_type AS "aggregateType", + aggregate_id AS "aggregateId", + occurred_at AS "occurredAt", + command_id AS "commandId", + payload_json AS "payloadJson" + FROM orchestration_events + WHERE sequence > ${request.sequenceExclusive} + ORDER BY sequence ASC + LIMIT ${request.limit} + `, + }); + + const append: OrchestrationEventStoreShape["append"] = (event) => + Effect.try({ + try: () => JSON.stringify(event.payload), + catch: toPersistenceSerializationError("OrchestrationEventStore.append:encodePayload"), + }).pipe( + Effect.flatMap((payloadJson) => + appendEventRow({ + eventId: event.eventId, + type: event.type, + aggregateType: event.aggregateType, + aggregateId: event.aggregateId, + occurredAt: event.occurredAt, + commandId: event.commandId, + payloadJson, + }).pipe( + Effect.mapError(toPersistenceSqlError("OrchestrationEventStore.append:insert")), + Effect.flatMap((row) => + eventRowToOrchestrationEvent(row, "OrchestrationEventStore.append:rowToEvent"), + ), + ), + ), + ); + + const readFromSequence: OrchestrationEventStoreShape["readFromSequence"] = ( + sequenceExclusive, + 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 OrchestrationEventStoreShape; +}); + +export const OrchestrationEventStoreLive = Layer.effect( + OrchestrationEventStore, + makeEventStore, +); diff --git a/apps/server/src/persistence/Layers/OrchestrationEvents.test.ts b/apps/server/src/persistence/Layers/OrchestrationEvents.test.ts deleted file mode 100644 index dddba6a727d9bc1f2a93ef7fd2b3964f0ac1fc4d..0000000000000000000000000000000000000000 --- a/apps/server/src/persistence/Layers/OrchestrationEvents.test.ts +++ /dev/null @@ -1,37 +0,0 @@ -import { Effect, Layer } from "effect"; -import { assert, it } from "@effect/vitest"; - -import { OrchestrationEventRepository } from "../Services/OrchestrationEvents.ts"; -import { SqlitePersistenceMemory } from "./Sqlite.ts"; -import { OrchestrationEventRepositoryLive } from "./OrchestrationEvents.ts"; -import { NodeServices } from "@effect/platform-node"; - -const layer = it.layer( - Layer.mergeAll( - Layer.provide(OrchestrationEventRepositoryLive, SqlitePersistenceMemory), - NodeServices.layer, - ), -); - -layer("OrchestrationEventRepository", (it) => { - it.effect("stores and replays events", () => - Effect.gen(function* () { - const createdAt = new Date().toISOString(); - - const repository = yield* OrchestrationEventRepository; - const saved = yield* repository.append({ - eventId: "event-1", - type: "thread.created", - aggregateType: "thread", - aggregateId: "thread-1", - occurredAt: createdAt, - commandId: "cmd-1", - payload: { id: "thread-1", projectId: "project-1", title: "demo" }, - }); - assert.equal(saved.sequence, 1); - - const replayed = yield* repository.readFromSequence(0); - assert.deepEqual(replayed, [saved]); - }), - ); -}); diff --git a/apps/server/src/persistence/Layers/OrchestrationEvents.ts b/apps/server/src/persistence/Layers/OrchestrationEvents.ts deleted file mode 100644 index 4da8c8c06323e06163844a23e83d20e53f56c293..0000000000000000000000000000000000000000 --- a/apps/server/src/persistence/Layers/OrchestrationEvents.ts +++ /dev/null @@ -1,178 +0,0 @@ -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 { - toPersistenceDecodeCauseError, - toPersistenceDecodeError, - toPersistenceSerializationError, - toPersistenceSqlError, - type OrchestrationEventRepositoryError, -} from "../Errors.ts"; -import { - OrchestrationEventRepository, - type OrchestrationEventRepositoryShape, -} from "../Services/OrchestrationEvents.ts"; - -const decodeEvent = Schema.decodeUnknownEffect(OrchestrationEventSchema); - -const EventRowSchema = Schema.Struct({ - sequence: Schema.Number, - eventId: Schema.String, - type: Schema.String, - aggregateType: Schema.String, - aggregateId: Schema.String, - occurredAt: Schema.String, - commandId: Schema.NullOr(Schema.String), - payloadJson: Schema.String, -}); - -const AppendEventRequestSchema = Schema.Struct({ - eventId: Schema.String, - type: Schema.String, - aggregateType: Schema.String, - aggregateId: Schema.String, - occurredAt: Schema.String, - commandId: Schema.NullOr(Schema.String), - payloadJson: Schema.String, -}); - -const ReadFromSequenceRequestSchema = Schema.Struct({ - sequenceExclusive: Schema.Number, - limit: Schema.Number, -}); - -function eventRowToOrchestrationEvent( - row: Schema.Schema.Type, - operation: string, -): Effect.Effect { - return Effect.try({ - try: () => JSON.parse(row.payloadJson) as unknown, - catch: toPersistenceDecodeCauseError(`${operation}:parsePayloadJson`), - }).pipe( - Effect.flatMap((payload) => - decodeEvent({ - sequence: row.sequence, - eventId: row.eventId, - type: row.type, - aggregateType: row.aggregateType, - aggregateId: row.aggregateId, - occurredAt: row.occurredAt, - commandId: row.commandId, - payload, - }).pipe(Effect.mapError(toPersistenceDecodeError(`${operation}:decodeEvent`))), - ), - ); -} - -const makeRepository = Effect.gen(function* () { - const sql = yield* SqlClient.SqlClient; - - const appendEventRow = SqlSchema.findOne({ - Request: AppendEventRequestSchema, - Result: EventRowSchema, - execute: (request) => - sql` - INSERT INTO orchestration_events ( - event_id, - event_type, - aggregate_type, - aggregate_id, - occurred_at, - command_id, - payload_json - ) - VALUES ( - ${request.eventId}, - ${request.type}, - ${request.aggregateType}, - ${request.aggregateId}, - ${request.occurredAt}, - ${request.commandId}, - ${request.payloadJson} - ) - RETURNING - sequence, - event_id AS "eventId", - event_type AS "type", - aggregate_type AS "aggregateType", - aggregate_id AS "aggregateId", - occurred_at AS "occurredAt", - command_id AS "commandId", - payload_json AS "payloadJson" - `, - }); - - const readEventRowsFromSequence = SqlSchema.findAll({ - Request: ReadFromSequenceRequestSchema, - Result: EventRowSchema, - execute: (request) => - sql` - SELECT - sequence, - event_id AS "eventId", - event_type AS "type", - aggregate_type AS "aggregateType", - aggregate_id AS "aggregateId", - occurred_at AS "occurredAt", - command_id AS "commandId", - payload_json AS "payloadJson" - FROM orchestration_events - WHERE sequence > ${request.sequenceExclusive} - ORDER BY sequence ASC - LIMIT ${request.limit} - `, - }); - - const append: OrchestrationEventRepositoryShape["append"] = (event) => - Effect.try({ - try: () => JSON.stringify(event.payload), - catch: toPersistenceSerializationError("OrchestrationEventRepository.append:encodePayload"), - }).pipe( - Effect.flatMap((payloadJson) => - appendEventRow({ - eventId: event.eventId, - type: event.type, - aggregateType: event.aggregateType, - aggregateId: event.aggregateId, - occurredAt: event.occurredAt, - commandId: event.commandId, - payloadJson, - }).pipe( - Effect.mapError(toPersistenceSqlError("OrchestrationEventRepository.append:insert")), - Effect.flatMap((row) => - eventRowToOrchestrationEvent(row, "OrchestrationEventRepository.append:rowToEvent"), - ), - ), - ), - ); - - const readFromSequence: OrchestrationEventRepositoryShape["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", - ), - ), - ), - ); - - return { - append, - readFromSequence, - readAll: () => readFromSequence(0, Number.MAX_SAFE_INTEGER), - } satisfies OrchestrationEventRepositoryShape; -}); - -export const OrchestrationEventRepositoryLive = Layer.effect( - OrchestrationEventRepository, - makeRepository, -); diff --git a/apps/server/src/persistence/Services/OrchestrationEventStore.ts b/apps/server/src/persistence/Services/OrchestrationEventStore.ts new file mode 100644 index 0000000000000000000000000000000000000000..568866e55ef578f53b3e948d5209f42dde92251b --- /dev/null +++ b/apps/server/src/persistence/Services/OrchestrationEventStore.ts @@ -0,0 +1,63 @@ +/** + * 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. + * + * Uses Effect `ServiceMap.Service` for dependency injection and exposes typed + * persistence/decode errors for event append and replay operations. + * + * @module OrchestrationEventStore + */ +import type { OrchestrationEvent } from "@t3tools/contracts"; +import { ServiceMap } from "effect"; +import type { Effect, Stream } from "effect"; + +import type { OrchestrationEventStoreError } from "../Errors.ts"; + +export interface OrchestrationEventStoreShape { + /** + * Persist a new orchestration event. + * + * @param event - Event payload without sequence (assigned by storage). + * @returns Effect containing the stored event with assigned sequence. + */ + readonly append: ( + event: Omit, + ) => Effect.Effect; + + /** + * Replay events after the provided sequence. + * + * @param sequenceExclusive - Sequence cursor (exclusive). + * @param limit - Maximum number of events to emit. + * @returns Stream containing ordered events. + */ + readonly readFromSequence: ( + sequenceExclusive: number, + limit?: number, + ) => Stream.Stream; + + /** + * Read all events from the beginning of the stream. + * + * @returns Stream containing all stored events. + */ + readonly readAll: () => Stream.Stream; +} + +/** + * OrchestrationEventStore - Service tag for orchestration event persistence. + * + * @example + * ```ts + * const program = Effect.gen(function* () { + * const events = yield* OrchestrationEventStore + * return yield* Stream.runCollect(events.readAll()) + * }) + * ``` + */ +export class OrchestrationEventStore extends ServiceMap.Service< + OrchestrationEventStore, + OrchestrationEventStoreShape +>()("orchestration/EventStore") {} diff --git a/apps/server/src/persistence/Services/OrchestrationEvents.ts b/apps/server/src/persistence/Services/OrchestrationEvents.ts deleted file mode 100644 index 88545f142df8c50816babeb88b1e27d238572a1e..0000000000000000000000000000000000000000 --- a/apps/server/src/persistence/Services/OrchestrationEvents.ts +++ /dev/null @@ -1,63 +0,0 @@ -/** - * OrchestrationEventRepository - Repository 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. - * - * Uses Effect `ServiceMap.Service` for dependency injection and exposes typed - * persistence/decode errors for event append and replay operations. - * - * @module OrchestrationEventRepository - */ -import type { OrchestrationEvent } from "@t3tools/contracts"; -import { ServiceMap } from "effect"; -import type { Effect } from "effect"; - -import type { OrchestrationEventRepositoryError } from "../Errors.ts"; - -export interface OrchestrationEventRepositoryShape { - /** - * Persist a new orchestration event. - * - * @param event - Event payload without sequence (assigned by storage). - * @returns Effect containing the stored event with assigned sequence. - */ - readonly append: ( - event: Omit, - ) => 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. - */ - readonly readFromSequence: ( - sequenceExclusive: number, - limit?: number, - ) => Effect.Effect; - - /** - * Read all events from the beginning of the stream. - * - * @returns Effect containing all stored events. - */ - readonly readAll: () => Effect.Effect; -} - -/** - * OrchestrationEventRepository - Service tag for orchestration event persistence. - * - * @example - * ```ts - * const program = Effect.gen(function* () { - * const events = yield* OrchestrationEventRepository - * return yield* events.readAll() - * }) - * ``` - */ -export class OrchestrationEventRepository extends ServiceMap.Service< - OrchestrationEventRepository, - OrchestrationEventRepositoryShape ->()("orchestration/EventRepository") {} 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"),