authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-02-23 10:33:45-08:00
committergravatar for jmarminge@gmail.comJulius Marminge <jmarminge@gmail.com> 2026-02-26 13:20:36-08:00
log7934639aa1379f745af7db7c05c964bf6d7a7cc8
treea9f0fde49d03d0433d8747ee7bf46e89af65e712
parent20668e49a934ce0bb02fe1eb8b57846fe36959ad

es terminology

22 files changed, 2050 insertions(+), 1404 deletions(-)

.plans/14-server-authoritative-event-sourcing-cleanup.md created+206
...@@ -0,0 +1,206 @@
1# Server-Authoritative Event-Sourcing Cleanup Plan
2
3Goal:
4- Move to a cleaner service architecture with:
5 - durable, server-authoritative event sourcing
6 - strict command routing/validation
7 - pluggable provider adapters
8 - explicit separation between transport, domain orchestration, provider runtime, and persistence
9
10## Target Service Graph (ASCII)
11
12```text
13 +---------------------------+
14 | wsServer |
15 | transport |
16 +---------------------------+
17 | orchestration.dispatchCommand
18 v
19 +-------------------------------------------+
20 | OrchestrationCommandRouter |
21 +-------------------------------------------+
22 |
23 v
24 +-------------------------------------------+
25 | OrchestrationCommandHandlers |
26 +-------------------------------------------+
27 |
28 v
29 +-------------------------------------------+
30 | OrchestrationEventStore |
31 +-------------------------------------------+
32 |
33 v
34 +-------------------------------------------+
35 | OrchestrationProjectionService |
36 +-------------------------------------------+
37 | snapshot/replay
38 +---------------------------> wsServer
39
40
41wsServer -- providers.* RPC --> +---------------------------+
42 | ProviderService |
43 +---------------------------+
44 | |
45 v v
46 +-------------------+ +-------------------------+
47 | ProviderSession | | ProviderAdapterRegistry |
48 | Registry (durable)| +-------------------------+
49 +-------------------+ |
50 ^ v
51 | +-------------------------+
52 | | ProviderAdapter(s) |
53 | +-------------------------+
54 | |
55 | runtime events v
56 | +---------------------------+
57 +----------| ProviderRuntimeIngestion |
58 +---------------------------+
59 | | |
60 v v v
61 Router Session Checkpoint
62 Registry Service
63
64 +-------------------------------------------+
65 | CheckpointService |
66 +-------------------------------------------+
67 | | |
68 v v v
69 +--------------------+ +-------------+ +-------------------+
70 | CheckpointCatalog | | Checkpoint | | ProviderAdapter(s)|
71 | (durable) | | Store (git) | | (read/rollback) |
72 +--------------------+ +-------------+ +-------------------+
73 |
74 v
75 +------+
76 |SQLite|
77 +------+
78
79OrchestrationEventStore ------> SQLite
80OrchestrationProjectionService -> SQLite
81ProviderSessionRegistry ------> SQLite
82CheckpointCatalog ------> SQLite
83```
84
85## Commit Series
86
87### Commit 1: Split public vs system orchestration command contracts
88- Create separate schemas/types:
89 - `ClientOrchestrationCommandSchema`
90 - `SystemOrchestrationCommandSchema`
91 - `OrchestrationCommandSchema = union(client, system)`
92- Ensure client transport can only submit client commands.
93- Keep system commands for server-internal workflows only.
94- Expected files:
95 - `packages/contracts/src/orchestration.ts`
96 - `apps/server/src/wsServer.ts`
97 - orchestration/service tests
98- Tests:
99 - reject system-only command via WS dispatch path
100 - preserve internal dispatch functionality for system commands
101
102### Commit 2: Introduce `OrchestrationCommandRouter` + handler boundary
103- Add dedicated router service to validate, authorize, and route commands.
104- Move command-to-event mapping out of `orchestration/Layer.ts` into handlers.
105- Add aggregate-level invariant checks before append (thread exists, project exists, etc.).
106- Expected files:
107 - `apps/server/src/orchestration/Services/CommandRouter.ts` (new)
108 - `apps/server/src/orchestration/Layers/CommandRouter.ts` (new)
109 - `apps/server/src/orchestration/Layer.ts`
110 - `apps/server/src/orchestration/reducer.ts` (only if needed for event payload changes)
111- Tests:
112 - router validation and invariant failures
113 - handler happy-path tests per command type
114
115### Commit 3: Harden event store for idempotency + optimistic append metadata
116- Add DB-level idempotency guard for `command_id` (`UNIQUE` where non-null).
117- Extend append API to support idempotent replays and deterministic return of prior event on duplicate `commandId`.
118- Add optional aggregate version metadata for future optimistic concurrency.
119- Expected files:
120 - `apps/server/src/persistence/Migrations/00x_*.ts` (new migration)
121 - `apps/server/src/persistence/Services/OrchestrationEvents.ts`
122 - `apps/server/src/persistence/Layers/OrchestrationEvents.ts`
123- Tests:
124 - duplicate command ID append returns same event/sequence (or explicit idempotent behavior)
125 - concurrent append behavior stays ordered and deterministic
126
127### Commit 4: Extract provider-runtime -> orchestration bridge from `wsServer`
128- Create `ProviderRuntimeIngestionService` that:
129 - subscribes to `ProviderService.streamEvents`
130 - translates runtime events into orchestration commands
131 - dispatches through router/engine
132- Remove provider-to-orchestration state mutation logic from `wsServer`.
133- Expected files:
134 - `apps/server/src/orchestration/Services/ProviderRuntimeIngestion.ts` (new)
135 - `apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts` (new)
136 - `apps/server/src/wsServer.ts`
137- Tests:
138 - ingestion service mapping tests (turn started/completed, message delta/completed, runtime error)
139 - ws integration confirms same external push behavior
140
141### Commit 5: Make session directory durable (`ProviderSessionRegistry`)
142- Replace in-memory-only `ProviderSessionDirectoryLive` with persistence-backed registry.
143- Keep in-memory cache optional, but source of truth must be persistent.
144- Add startup reconciliation to prune dead sessions / keep known thread mapping.
145- Expected files:
146 - `apps/server/src/provider/Services/ProviderSessionDirectory.ts` (or new SessionRegistry service)
147 - `apps/server/src/provider/Layers/ProviderSessionDirectory.ts`
148 - `apps/server/src/persistence/Migrations/00x_*.ts` (new table/indexes)
149 - provider persistence tests
150- Tests:
151 - survives server restart with correct mapping
152 - stale session cleanup semantics
153
154### Commit 6: Re-key checkpoint metadata from session to thread identity
155- Change checkpoint catalog primary identity from `provider_session_id` to durable `thread_id`.
156- Keep `session_id` as nullable metadata only.
157- Update checkpoint flows (`initialize`, `capture`, `list`, `diff`, `revert`) to use thread identity.
158- Expected files:
159 - `apps/server/src/persistence/Migrations/00x_*.ts` (checkpoint schema migration)
160 - `apps/server/src/persistence/Services/Checkpoints.ts`
161 - `apps/server/src/persistence/Layers/Checkpoints.ts`
162 - `apps/server/src/checkpointing/Layers/CheckpointService.ts`
163- Tests:
164 - resume/new session over same thread sees same checkpoint history
165 - revert/diff still work after session churn
166
167### Commit 7: Add durable projection persistence for orchestration read models
168- Introduce projection tables/snapshots persisted in DB to avoid full replay dependency.
169- Keep event stream as source of truth; projection rebuild stays deterministic.
170- `getSnapshot` reads from projection store (memory cache optional).
171- Expected files:
172 - `apps/server/src/persistence/Migrations/00x_*.ts` (projection tables)
173 - `apps/server/src/orchestration/*` projection service/layer
174 - `apps/server/src/wsServer.ts` (snapshot/replay path wiring)
175- Tests:
176 - cold boot snapshot load without replaying full history in process
177 - projection rebuild from events yields same result as previous reducer semantics
178
179### Commit 8: Narrow `ProviderService` responsibilities
180- Keep `ProviderService` focused on provider RPC/session lifecycle + unified runtime stream.
181- Move checkpoint-capture side effects out of provider event worker into dedicated ingestion/checkpoint pipeline service.
182- Preserve adapter pluggability and provider-neutral contracts.
183- Expected files:
184 - `apps/server/src/provider/Layers/ProviderService.ts`
185 - new orchestration/checkpoint runtime coordinator service(s)
186- Tests:
187 - provider service routing stays intact
188 - checkpoint capture still triggered by turn completion through new coordinator
189
190### Commit 9: Remove dead legacy path and finalize docs
191- Remove unused legacy manager/store path from active architecture:
192 - `providerManager.ts`
193 - `filesystemCheckpointStore.ts` (if no longer needed by tests/tools)
194- Update architecture docs with final service boundaries and boot/runtime graph.
195- Expected files:
196 - legacy files + references
197 - `AGENTS.md`/docs as needed
198 - `.plans` docs linkage
199- Tests:
200 - full server integration suite passes on Effect-only path
201 - no regressions in WS protocol behavior
202
203## Risk Controls
204- Keep WS method names and payload contracts stable throughout.
205- Gate each commit with targeted integration tests before moving forward.
206- Avoid broad event-type churn in one step; migrate schemas incrementally with clear compatibility windows.
apps/server/src/orchestration/Errors.ts+23-9
...@@ -1,6 +1,6 @@...@@ -1,6 +1,6 @@
1import { SchemaIssue, Schema } from "effect";1import { SchemaIssue, Schema } from "effect";
22
3import type { OrchestrationEventRepositoryError } from "../persistence/Errors.ts";3import type { OrchestrationEventStoreError } from "../persistence/Errors.ts";
44
5export class OrchestrationCommandJsonParseError extends Schema.TaggedErrorClass<OrchestrationCommandJsonParseError>()(5export class OrchestrationCommandJsonParseError extends Schema.TaggedErrorClass<OrchestrationCommandJsonParseError>()(
6 "OrchestrationCommandJsonParseError",6 "OrchestrationCommandJsonParseError",
...@@ -26,8 +26,21 @@ export class OrchestrationCommandDecodeError extends Schema.TaggedErrorClass<Orc...@@ -26,8 +26,21 @@ export class OrchestrationCommandDecodeError extends Schema.TaggedErrorClass<Orc
26 }26 }
27}27}
2828
29export class OrchestrationReducerDecodeError extends Schema.TaggedErrorClass<OrchestrationReducerDecodeError>()(29export class OrchestrationCommandInvariantError extends Schema.TaggedErrorClass<OrchestrationCommandInvariantError>()(
30 "OrchestrationReducerDecodeError",30 "OrchestrationCommandInvariantError",
31 {
32 commandType: Schema.String,
33 detail: Schema.String,
34 cause: Schema.optional(Schema.Defect),
35 },
36) {
37 override get message(): string {
38 return `Orchestration command invariant failed (${this.commandType}): ${this.detail}`;
39 }
40}
41
42export class OrchestrationProjectorDecodeError extends Schema.TaggedErrorClass<OrchestrationProjectorDecodeError>()(
43 "OrchestrationProjectorDecodeError",
31 {44 {
32 eventType: Schema.String,45 eventType: Schema.String,
33 issue: Schema.String,46 issue: Schema.String,
...@@ -35,7 +48,7 @@ export class OrchestrationReducerDecodeError extends Schema.TaggedErrorClass<Orc...@@ -35,7 +48,7 @@ export class OrchestrationReducerDecodeError extends Schema.TaggedErrorClass<Orc
35 },48 },
36) {49) {
37 override get message(): string {50 override get message(): string {
38 return `Reducer decode failed for ${this.eventType}: ${this.issue}`;51 return `Projector decode failed for ${this.eventType}: ${this.issue}`;
39 }52 }
40}53}
4154
...@@ -53,8 +66,9 @@ export class OrchestrationListenerCallbackError extends Schema.TaggedErrorClass<...@@ -53,8 +66,9 @@ export class OrchestrationListenerCallbackError extends Schema.TaggedErrorClass<
53}66}
5467
55export type OrchestrationDispatchError =68export type OrchestrationDispatchError =
56 | OrchestrationEventRepositoryError69 | OrchestrationEventStoreError
57 | OrchestrationReducerDecodeError70 | OrchestrationCommandInvariantError
71 | OrchestrationProjectorDecodeError
58 | OrchestrationListenerCallbackError;72 | OrchestrationListenerCallbackError;
5973
60export type OrchestrationEngineError =74export type OrchestrationEngineError =
...@@ -69,9 +83,9 @@ export function toOrchestrationCommandDecodeError(error: Schema.SchemaError) {...@@ -69,9 +83,9 @@ export function toOrchestrationCommandDecodeError(error: Schema.SchemaError) {
69 });83 });
70}84}
7185
72export function toReducerDecodeError(eventType: string) {86export function toProjectorDecodeError(eventType: string) {
73 return (error: Schema.SchemaError): OrchestrationReducerDecodeError =>87 return (error: Schema.SchemaError): OrchestrationProjectorDecodeError =>
74 new OrchestrationReducerDecodeError({88 new OrchestrationProjectorDecodeError({
75 eventType,89 eventType,
76 issue: SchemaIssue.makeFormatterDefault()(error.issue),90 issue: SchemaIssue.makeFormatterDefault()(error.issue),
77 cause: error,91 cause: error,
apps/server/src/orchestration/Layer.test.ts+88-23
...@@ -3,18 +3,18 @@ import { Effect, Layer, ManagedRuntime, Queue, Stream } from "effect";...@@ -3,18 +3,18 @@ import { Effect, Layer, ManagedRuntime, Queue, Stream } from "effect";
3import { describe, expect, it } from "vitest";3import { describe, expect, it } from "vitest";
44
5import {5import {
6 OrchestrationEventRepository,6 OrchestrationEventStore,
7 type OrchestrationEventRepositoryShape,7 type OrchestrationEventStoreShape,
8} from "../persistence/Services/OrchestrationEvents.ts";8} from "../persistence/Services/OrchestrationEventStore.ts";
9import { PersistenceSqlError } from "../persistence/Errors.ts";9import { PersistenceSqlError } from "../persistence/Errors.ts";
10import { SqlitePersistenceMemory } from "../persistence/Layers/Sqlite.ts";10import { SqlitePersistenceMemory } from "../persistence/Layers/Sqlite.ts";
11import { OrchestrationEngineLive } from "./Layer.ts";11import { OrchestrationEngineLive } from "./Layers/OrchestrationEngine.ts";
12import { OrchestrationEngineService } from "./Service.ts";12import { OrchestrationEngineService } from "./Services/OrchestrationEngine.ts";
13import { OrchestrationEventRepositoryLive } from "../persistence/Layers/OrchestrationEvents.ts";13import { OrchestrationEventStoreLive } from "../persistence/Layers/OrchestrationEventStore.ts";
1414
15export async function createOrchestrationSystem() {15export async function createOrchestrationSystem() {
16 const orchestrationLayer = OrchestrationEngineLive.pipe(16 const orchestrationLayer = OrchestrationEngineLive.pipe(
17 Layer.provide(OrchestrationEventRepositoryLive),17 Layer.provide(OrchestrationEventStoreLive),
18 Layer.provide(SqlitePersistenceMemory),18 Layer.provide(SqlitePersistenceMemory),
19 );19 );
20 const runtime = ManagedRuntime.make(orchestrationLayer);20 const runtime = ManagedRuntime.make(orchestrationLayer);
...@@ -27,7 +27,7 @@ export async function createOrchestrationSystem() {...@@ -27,7 +27,7 @@ export async function createOrchestrationSystem() {
27}27}
2828
29describe("OrchestrationEngine", () => {29describe("OrchestrationEngine", () => {
30 it("returns deterministic snapshots for repeated reads", async () => {30 it("returns deterministic read models for repeated reads", async () => {
31 const createdAt = new Date().toISOString();31 const createdAt = new Date().toISOString();
32 const projectId = "project-1";32 const projectId = "project-1";
33 const threadId = "thread-1";33 const threadId = "thread-1";
...@@ -59,9 +59,9 @@ describe("OrchestrationEngine", () => {...@@ -59,9 +59,9 @@ describe("OrchestrationEngine", () => {
59 createdAt,59 createdAt,
60 }),60 }),
61 );61 );
62 const snapshotA = await system.run(engine.getSnapshot());62 const readModelA = await system.run(engine.getReadModel());
63 const snapshotB = await system.run(engine.getSnapshot());63 const readModelB = await system.run(engine.getReadModel());
64 expect(snapshotB).toEqual(snapshotA);64 expect(readModelB).toEqual(readModelA);
65 await system.dispose();65 await system.dispose();
66 });66 });
6767
...@@ -89,7 +89,11 @@ describe("OrchestrationEngine", () => {...@@ -89,7 +89,11 @@ describe("OrchestrationEngine", () => {
89 createdAt: new Date().toISOString(),89 createdAt: new Date().toISOString(),
90 }),90 }),
91 );91 );
92 const events = await system.run(engine.replayEvents(0));92 const events = await system.run(
93 Stream.runCollect(engine.readEvents(0)).pipe(
94 Effect.map((chunk): OrchestrationEvent[] => Array.from(chunk)),
95 ),
96 );
93 expect(events.length).toBe(2);97 expect(events.length).toBe(2);
94 expect(events[0]?.type).toBe("thread.created");98 expect(events[0]?.type).toBe("thread.created");
95 expect(events[1]?.type).toBe("thread.deleted");99 expect(events[1]?.type).toBe("thread.deleted");
...@@ -167,7 +171,7 @@ describe("OrchestrationEngine", () => {...@@ -167,7 +171,7 @@ describe("OrchestrationEngine", () => {
167 }),171 }),
168 );172 );
169173
170 const firstThread = (await system.run(engine.getSnapshot())).threads.find(174 const firstThread = (await system.run(engine.getReadModel())).threads.find(
171 (thread) => thread.id === "thread-turn-diff",175 (thread) => thread.id === "thread-turn-diff",
172 );176 );
173 expect(firstThread?.turnDiffSummaries).toEqual([177 expect(firstThread?.turnDiffSummaries).toEqual([
...@@ -289,7 +293,7 @@ describe("OrchestrationEngine", () => {...@@ -289,7 +293,7 @@ describe("OrchestrationEngine", () => {
289 }),293 }),
290 );294 );
291295
292 const thread = (await system.run(engine.getSnapshot())).threads.find(296 const thread = (await system.run(engine.getReadModel())).threads.find(
293 (entry) => entry.id === "thread-revert",297 (entry) => entry.id === "thread-revert",
294 );298 );
295 expect(thread?.messages.map((message) => message.id)).toEqual(["user-1", "assistant:turn-1"]);299 expect(thread?.messages.map((message) => message.id)).toEqual(["user-1", "assistant:turn-1"]);
...@@ -299,20 +303,20 @@ describe("OrchestrationEngine", () => {...@@ -299,20 +303,20 @@ describe("OrchestrationEngine", () => {
299 });303 });
300304
301 it("allows stop to be called multiple times", async () => {305 it("allows stop to be called multiple times", async () => {
302 const inMemoryStore: OrchestrationEventRepositoryShape = {306 const inMemoryStore: OrchestrationEventStoreShape = {
303 append(event) {307 append(event) {
304 return Effect.succeed({ ...event, sequence: 1 });308 return Effect.succeed({ ...event, sequence: 1 });
305 },309 },
306 readFromSequence() {310 readFromSequence() {
307 return Effect.succeed([]);311 return Stream.empty;
308 },312 },
309 readAll() {313 readAll() {
310 return Effect.succeed([]);314 return Stream.empty;
311 },315 },
312 };316 };
313 const runtime = ManagedRuntime.make(317 const runtime = ManagedRuntime.make(
314 OrchestrationEngineLive.pipe(318 OrchestrationEngineLive.pipe(
315 Layer.provide(Layer.succeed(OrchestrationEventRepository, inMemoryStore)),319 Layer.provide(Layer.succeed(OrchestrationEventStore, inMemoryStore)),
316 ),320 ),
317 );321 );
318 await runtime.runPromise(Effect.service(OrchestrationEngineService));322 await runtime.runPromise(Effect.service(OrchestrationEngineService));
...@@ -325,7 +329,7 @@ describe("OrchestrationEngine", () => {...@@ -325,7 +329,7 @@ describe("OrchestrationEngine", () => {
325 let nextSequence = 1;329 let nextSequence = 1;
326 let shouldFailFirstAppend = true;330 let shouldFailFirstAppend = true;
327331
328 const flakyStore: OrchestrationEventRepositoryShape = {332 const flakyStore: OrchestrationEventStoreShape = {
329 append(event) {333 append(event) {
330 if (shouldFailFirstAppend) {334 if (shouldFailFirstAppend) {
331 shouldFailFirstAppend = false;335 shouldFailFirstAppend = false;
...@@ -345,16 +349,16 @@ describe("OrchestrationEngine", () => {...@@ -345,16 +349,16 @@ describe("OrchestrationEngine", () => {
345 return Effect.succeed(savedEvent);349 return Effect.succeed(savedEvent);
346 },350 },
347 readFromSequence(sequenceExclusive) {351 readFromSequence(sequenceExclusive) {
348 return Effect.succeed(events.filter((event) => event.sequence > sequenceExclusive));352 return Stream.fromIterable(events.filter((event) => event.sequence > sequenceExclusive));
349 },353 },
350 readAll() {354 readAll() {
351 return Effect.succeed(events);355 return Stream.fromIterable(events);
352 },356 },
353 };357 };
354358
355 const runtime = ManagedRuntime.make(359 const runtime = ManagedRuntime.make(
356 OrchestrationEngineLive.pipe(360 OrchestrationEngineLive.pipe(
357 Layer.provide(Layer.succeed(OrchestrationEventRepository, flakyStore)),361 Layer.provide(Layer.succeed(OrchestrationEventStore, flakyStore)),
358 ),362 ),
359 );363 );
360 const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService));364 const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService));
...@@ -391,8 +395,69 @@ describe("OrchestrationEngine", () => {...@@ -391,8 +395,69 @@ describe("OrchestrationEngine", () => {
391 );395 );
392396
393 expect(result.sequence).toBe(1);397 expect(result.sequence).toBe(1);
394 expect((await runtime.runPromise(engine.getSnapshot())).sequence).toBe(1);398 expect((await runtime.runPromise(engine.getReadModel())).sequence).toBe(1);
395399
396 await runtime.dispose();400 await runtime.dispose();
397 });401 });
402
403 it("fails command dispatch when command invariants are violated", async () => {
404 const system = await createOrchestrationSystem();
405 const engine = system.engine;
406 const createdAt = new Date().toISOString();
407
408 await expect(
409 system.run(
410 engine.dispatch({
411 type: "message.send",
412 commandId: "cmd-invariant-missing-thread",
413 threadId: "thread-missing",
414 messageId: "msg-missing",
415 role: "user",
416 text: "hello",
417 streaming: false,
418 createdAt,
419 }),
420 ),
421 ).rejects.toThrow("Thread 'thread-missing' does not exist");
422
423 await system.dispose();
424 });
425
426 it("rejects duplicate thread creation", async () => {
427 const system = await createOrchestrationSystem();
428 const engine = system.engine;
429 const createdAt = new Date().toISOString();
430
431 await system.run(
432 engine.dispatch({
433 type: "thread.create",
434 commandId: "cmd-thread-duplicate-1",
435 threadId: "thread-duplicate",
436 projectId: "project-duplicate",
437 title: "duplicate",
438 model: "gpt-5-codex",
439 branch: null,
440 worktreePath: null,
441 createdAt,
442 }),
443 );
444
445 await expect(
446 system.run(
447 engine.dispatch({
448 type: "thread.create",
449 commandId: "cmd-thread-duplicate-2",
450 threadId: "thread-duplicate",
451 projectId: "project-duplicate",
452 title: "duplicate",
453 model: "gpt-5-codex",
454 branch: null,
455 worktreePath: null,
456 createdAt,
457 }),
458 ),
459 ).rejects.toThrow("already exists");
460
461 await system.dispose();
462 });
398});463});
apps/server/src/orchestration/Layer.ts deleted-258
...@@ -1,258 +0,0 @@
1import type {
2 OrchestrationCommand,
3 OrchestrationEvent,
4 OrchestrationReadModel,
5} from "@t3tools/contracts";
6import { OrchestrationCommandSchema } from "@t3tools/contracts";
7import { Deferred, Effect, Layer, PubSub, Queue, Schema, Stream } from "effect";
8
9import { createLogger } from "../logger.ts";
10import { OrchestrationEventRepository } from "../persistence/Services/OrchestrationEvents.ts";
11import {
12 toOrchestrationCommandDecodeError,
13 toOrchestrationJsonParseError,
14 type OrchestrationDispatchError,
15} from "./Errors.ts";
16import { createEmptyReadModel, reduceEvent } from "./reducer.ts";
17import { OrchestrationEngineService, type OrchestrationEngineShape } from "./Service.ts";
18
19interface CommandEnvelope {
20 command: OrchestrationCommand;
21 result: Deferred.Deferred<{ sequence: number }, OrchestrationDispatchError>;
22}
23
24function mapCommandToEvent(command: OrchestrationCommand): Omit<OrchestrationEvent, "sequence"> {
25 const eventId = crypto.randomUUID();
26 switch (command.type) {
27 case "thread.create":
28 return {
29 eventId,
30 type: "thread.created",
31 aggregateType: "thread",
32 aggregateId: command.threadId,
33 occurredAt: command.createdAt,
34 commandId: command.commandId,
35 payload: {
36 id: command.threadId,
37 projectId: command.projectId,
38 title: command.title,
39 model: command.model,
40 branch: command.branch,
41 worktreePath: command.worktreePath,
42 createdAt: command.createdAt,
43 updatedAt: command.createdAt,
44 },
45 };
46 case "thread.delete":
47 return {
48 eventId,
49 type: "thread.deleted",
50 aggregateType: "thread",
51 aggregateId: command.threadId,
52 occurredAt: command.createdAt,
53 commandId: command.commandId,
54 payload: {
55 id: command.threadId,
56 deletedAt: command.createdAt,
57 },
58 };
59 case "thread.meta.update":
60 return {
61 eventId,
62 type: "thread.meta-updated",
63 aggregateType: "thread",
64 aggregateId: command.threadId,
65 occurredAt: command.createdAt,
66 commandId: command.commandId,
67 payload: {
68 threadId: command.threadId,
69 ...(command.title !== undefined ? { title: command.title } : {}),
70 ...(command.model !== undefined ? { model: command.model } : {}),
71 ...(command.branch !== undefined ? { branch: command.branch } : {}),
72 ...(command.worktreePath !== undefined ? { worktreePath: command.worktreePath } : {}),
73 updatedAt: command.createdAt,
74 },
75 };
76 case "message.send":
77 return {
78 eventId,
79 type: "message.sent",
80 aggregateType: "thread",
81 aggregateId: command.threadId,
82 occurredAt: command.createdAt,
83 commandId: command.commandId,
84 payload: {
85 id: command.messageId,
86 role: command.role,
87 text: command.text,
88 threadId: command.threadId,
89 createdAt: command.createdAt,
90 streaming: command.streaming === true,
91 },
92 };
93 case "thread.session":
94 return {
95 eventId,
96 type: "thread.session-set",
97 aggregateType: "thread",
98 aggregateId: command.threadId,
99 occurredAt: command.createdAt,
100 commandId: command.commandId,
101 payload: {
102 threadId: command.threadId,
103 session: command.session,
104 },
105 };
106 case "git.readModel.upsert":
107 return {
108 eventId,
109 type: "git.read-model-upsert",
110 aggregateType: "project",
111 aggregateId: command.projectId,
112 occurredAt: command.createdAt,
113 commandId: command.commandId,
114 payload: {
115 projectId: command.projectId,
116 branch: command.branch,
117 hasWorkingTreeChanges: command.hasWorkingTreeChanges,
118 aheadCount: command.aheadCount,
119 behindCount: command.behindCount,
120 updatedAt: command.createdAt,
121 },
122 };
123 case "thread.turnDiff.complete":
124 return {
125 eventId,
126 type: "thread.turn-diff-completed",
127 aggregateType: "thread",
128 aggregateId: command.threadId,
129 occurredAt: command.createdAt,
130 commandId: command.commandId,
131 payload: {
132 threadId: command.threadId,
133 turnId: command.turnId,
134 completedAt: command.completedAt,
135 ...(command.status !== undefined ? { status: command.status } : {}),
136 files: command.files,
137 ...(command.assistantMessageId !== undefined
138 ? { assistantMessageId: command.assistantMessageId }
139 : {}),
140 ...(command.checkpointTurnCount !== undefined
141 ? { checkpointTurnCount: command.checkpointTurnCount }
142 : {}),
143 },
144 };
145 case "thread.activity.append":
146 return {
147 eventId,
148 type: "thread.activity-appended",
149 aggregateType: "thread",
150 aggregateId: command.threadId,
151 occurredAt: command.createdAt,
152 commandId: command.commandId,
153 payload: {
154 threadId: command.threadId,
155 activity: command.activity,
156 },
157 };
158 case "thread.revert":
159 return {
160 eventId,
161 type: "thread.reverted",
162 aggregateType: "thread",
163 aggregateId: command.threadId,
164 occurredAt: command.createdAt,
165 commandId: command.commandId,
166 payload: {
167 threadId: command.threadId,
168 turnCount: command.turnCount,
169 messageCount: command.messageCount,
170 },
171 };
172 }
173}
174
175const decodeUnknownCommand = Schema.decodeUnknownEffect(OrchestrationCommandSchema);
176
177const makeOrchestrationEngine = Effect.gen(function* () {
178 const logger = createLogger("orchestration");
179 const eventStore = yield* OrchestrationEventRepository;
180
181 let readModel = createEmptyReadModel(new Date().toISOString());
182
183 const commandQueue = yield* Queue.unbounded<CommandEnvelope>();
184 const eventPubSub = yield* PubSub.unbounded<OrchestrationEvent>();
185
186 const processEnvelope = (envelope: CommandEnvelope): Effect.Effect<void> =>
187 Effect.gen(function* () {
188 const eventBase = mapCommandToEvent(envelope.command);
189 const savedEvent = yield* eventStore.append(eventBase);
190 readModel = yield* reduceEvent(readModel, savedEvent);
191 yield* PubSub.publish(eventPubSub, savedEvent);
192
193 yield* Deferred.succeed(envelope.result, { sequence: savedEvent.sequence });
194 }).pipe(Effect.catch((error) => Deferred.fail(envelope.result, error).pipe(Effect.asVoid)));
195
196 const bootstrapReadModel: Effect.Effect<void, OrchestrationDispatchError> = Effect.gen(
197 function* () {
198 const existingEvents = yield* eventStore.readAll();
199 for (const event of existingEvents) {
200 readModel = yield* reduceEvent(readModel, event);
201 }
202 },
203 );
204
205 yield* bootstrapReadModel;
206
207 const worker = Effect.forever(Queue.take(commandQueue).pipe(Effect.flatMap(processEnvelope)));
208 yield* Effect.forkScoped(worker);
209 logger.info("orchestration engine started", {
210 sequence: readModel.sequence,
211 });
212
213 const getSnapshot: OrchestrationEngineShape["getSnapshot"] = () =>
214 Effect.sync((): OrchestrationReadModel => readModel);
215
216 const replayEvents: OrchestrationEngineShape["replayEvents"] = (fromSequenceExclusive) =>
217 eventStore.readFromSequence(fromSequenceExclusive);
218
219 const dispatch: OrchestrationEngineShape["dispatch"] = (command) =>
220 Effect.gen(function* () {
221 const result = yield* Deferred.make<{ sequence: number }, OrchestrationDispatchError>();
222 yield* Queue.offer(commandQueue, { command, result });
223 return yield* Deferred.await(result);
224 });
225
226 const dispatchUnknown: OrchestrationEngineShape["dispatchUnknown"] = (command) =>
227 Effect.gen(function* () {
228 const payload =
229 typeof command === "string"
230 ? yield* Effect.try({
231 try: () => JSON.parse(command) as unknown,
232 catch: toOrchestrationJsonParseError,
233 })
234 : command;
235
236 const decoded = yield* decodeUnknownCommand(payload).pipe(
237 Effect.mapError(toOrchestrationCommandDecodeError),
238 );
239
240 return yield* dispatch(decoded);
241 });
242
243 const streamDomainEvents: OrchestrationEngineShape["streamDomainEvents"] =
244 Stream.fromPubSub(eventPubSub);
245
246 return {
247 getSnapshot,
248 replayEvents,
249 dispatchUnknown,
250 dispatch,
251 streamDomainEvents,
252 } satisfies OrchestrationEngineShape;
253});
254
255export const OrchestrationEngineLive = Layer.effect(
256 OrchestrationEngineService,
257 makeOrchestrationEngine,
258);
apps/server/src/orchestration/Layers/OrchestrationEngine.ts created+110
...@@ -0,0 +1,110 @@
1import type {
2 OrchestrationCommand,
3 OrchestrationEvent,
4 OrchestrationReadModel,
5} from "@t3tools/contracts";
6import { OrchestrationCommandSchema } from "@t3tools/contracts";
7import { Deferred, Effect, Layer, PubSub, Queue, Schema, Stream } from "effect";
8
9import { createLogger } from "../../logger.ts";
10import { OrchestrationEventStore } from "../../persistence/Services/OrchestrationEventStore.ts";
11import {
12 toOrchestrationCommandDecodeError,
13 toOrchestrationJsonParseError,
14 type OrchestrationDispatchError,
15} from "../Errors.ts";
16import { decideOrchestrationCommand } from "../decider.ts";
17import { createEmptyReadModel, projectEvent } from "../projector.ts";
18import {
19 OrchestrationEngineService,
20 type OrchestrationEngineShape,
21} from "../Services/OrchestrationEngine.ts";
22
23interface CommandEnvelope {
24 command: OrchestrationCommand;
25 result: Deferred.Deferred<{ sequence: number }, OrchestrationDispatchError>;
26}
27
28const decodeUnknownCommand = Schema.decodeUnknownEffect(OrchestrationCommandSchema);
29
30const makeOrchestrationEngine = Effect.gen(function* () {
31 const logger = createLogger("orchestration");
32 const eventStore = yield* OrchestrationEventStore;
33
34 let readModel = createEmptyReadModel(new Date().toISOString());
35
36 const commandQueue = yield* Queue.unbounded<CommandEnvelope>();
37 const eventPubSub = yield* PubSub.unbounded<OrchestrationEvent>();
38
39 const processEnvelope = (envelope: CommandEnvelope): Effect.Effect<void> =>
40 Effect.gen(function* () {
41 const eventBase = yield* decideOrchestrationCommand({
42 command: envelope.command,
43 readModel,
44 });
45 const savedEvent = yield* eventStore.append(eventBase);
46 readModel = yield* projectEvent(readModel, savedEvent);
47 yield* PubSub.publish(eventPubSub, savedEvent);
48
49 yield* Deferred.succeed(envelope.result, { sequence: savedEvent.sequence });
50 }).pipe(Effect.catch((error) => Deferred.fail(envelope.result, error).pipe(Effect.asVoid)));
51
52 // bootstrap read model from event store
53 yield* Stream.runForEach(eventStore.readAll(), (event) =>
54 Effect.gen(function* () {
55 readModel = yield* projectEvent(readModel, event);
56 }),
57 );
58
59 const worker = Effect.forever(Queue.take(commandQueue).pipe(Effect.flatMap(processEnvelope)));
60 yield* Effect.forkScoped(worker);
61 logger.info("orchestration engine started", {
62 sequence: readModel.sequence,
63 });
64
65 const getReadModel: OrchestrationEngineShape["getReadModel"] = () =>
66 Effect.sync((): OrchestrationReadModel => readModel);
67
68 const readEvents: OrchestrationEngineShape["readEvents"] = (fromSequenceExclusive) =>
69 eventStore.readFromSequence(fromSequenceExclusive);
70
71 const dispatch: OrchestrationEngineShape["dispatch"] = (command) =>
72 Effect.gen(function* () {
73 const result = yield* Deferred.make<{ sequence: number }, OrchestrationDispatchError>();
74 yield* Queue.offer(commandQueue, { command, result });
75 return yield* Deferred.await(result);
76 });
77
78 const dispatchUnknownCommand: OrchestrationEngineShape["dispatchUnknownCommand"] = (command) =>
79 Effect.gen(function* () {
80 const payload =
81 typeof command === "string"
82 ? yield* Effect.try({
83 try: () => JSON.parse(command) as unknown,
84 catch: toOrchestrationJsonParseError,
85 })
86 : command;
87
88 const decoded = yield* decodeUnknownCommand(payload).pipe(
89 Effect.mapError(toOrchestrationCommandDecodeError),
90 );
91
92 return yield* dispatch(decoded);
93 });
94
95 const streamDomainEvents: OrchestrationEngineShape["streamDomainEvents"] =
96 Stream.fromPubSub(eventPubSub);
97
98 return {
99 getReadModel,
100 readEvents,
101 dispatchUnknownCommand,
102 dispatch,
103 streamDomainEvents,
104 } satisfies OrchestrationEngineShape;
105});
106
107export const OrchestrationEngineLive = Layer.effect(
108 OrchestrationEngineService,
109 makeOrchestrationEngine,
110);
apps/server/src/orchestration/Service.ts deleted-84
...@@ -1,84 +0,0 @@
1/**
2 * OrchestrationEngineService - Service interface for orchestration command handling.
3 *
4 * Owns command validation/dispatch and in-memory read-model updates backed by
5 * `OrchestrationEventRepository` persistence. It does not own provider process
6 * management or transport concerns (e.g. websocket request parsing).
7 *
8 * Uses Effect `ServiceMap.Service` for dependency injection. Command dispatch,
9 * replay, and unknown-input decoding all return typed domain errors.
10 *
11 * @module OrchestrationEngineService
12 */
13import type {
14 OrchestrationCommand,
15 OrchestrationEvent,
16 OrchestrationReadModel,
17} from "@t3tools/contracts";
18import { ServiceMap } from "effect";
19import type { Effect, Stream } from "effect";
20
21import type { OrchestrationDispatchError, OrchestrationEngineError } from "./Errors.ts";
22import type { OrchestrationEventRepositoryError } from "../persistence/Errors.ts";
23
24export interface OrchestrationEngineShape {
25 /**
26 * Read the current in-memory orchestration snapshot.
27 *
28 * @returns Effect containing the latest read model.
29 */
30 readonly getSnapshot: () => Effect.Effect<OrchestrationReadModel, never, never>;
31
32 /**
33 * Replay persisted orchestration events from an exclusive sequence cursor.
34 *
35 * @param fromSequenceExclusive - Sequence cursor (exclusive).
36 * @returns Effect containing ordered events.
37 */
38 readonly replayEvents: (
39 fromSequenceExclusive: number,
40 ) => Effect.Effect<OrchestrationEvent[], OrchestrationEventRepositoryError, never>;
41
42 /**
43 * Validate and dispatch an unknown command payload.
44 *
45 * Accepts either parsed values or raw JSON strings.
46 *
47 * @param command - Unknown command payload.
48 * @returns Effect containing the sequence of the persisted event.
49 */
50 readonly dispatchUnknown: (
51 command: unknown,
52 ) => Effect.Effect<{ sequence: number }, OrchestrationEngineError, never>;
53
54 /**
55 * Dispatch a validated orchestration command.
56 *
57 * @param command - Valid orchestration command.
58 * @returns Effect containing the sequence of the persisted event.
59 */
60 readonly dispatch: (
61 command: OrchestrationCommand,
62 ) => Effect.Effect<{ sequence: number }, OrchestrationDispatchError, never>;
63
64 /**
65 * Stream persisted domain events in dispatch order.
66 */
67 readonly streamDomainEvents: Stream.Stream<OrchestrationEvent>;
68}
69
70/**
71 * OrchestrationEngineService - Service tag for orchestration engine access.
72 *
73 * @example
74 * ```ts
75 * const program = Effect.gen(function* () {
76 * const engine = yield* OrchestrationEngineService
77 * return yield* engine.getSnapshot()
78 * })
79 * ```
80 */
81export class OrchestrationEngineService extends ServiceMap.Service<
82 OrchestrationEngineService,
83 OrchestrationEngineShape
84>()("orchestration/Engine") {}
apps/server/src/orchestration/Services/OrchestrationEngine.ts created+84
...@@ -0,0 +1,84 @@
1/**
2 * OrchestrationEngineService - Service interface for orchestration command handling.
3 *
4 * Owns command validation/dispatch and in-memory read-model updates backed by
5 * `OrchestrationEventStore` persistence. It does not own provider process
6 * management or transport concerns (e.g. websocket request parsing).
7 *
8 * Uses Effect `ServiceMap.Service` for dependency injection. Command dispatch,
9 * replay, and unknown-input decoding all return typed domain errors.
10 *
11 * @module OrchestrationEngineService
12 */
13import type {
14 OrchestrationCommand,
15 OrchestrationEvent,
16 OrchestrationReadModel,
17} from "@t3tools/contracts";
18import { ServiceMap } from "effect";
19import type { Effect, Stream } from "effect";
20
21import type { OrchestrationDispatchError, OrchestrationEngineError } from "../Errors.ts";
22import type { OrchestrationEventStoreError } from "../../persistence/Errors.ts";
23
24export interface OrchestrationEngineShape {
25 /**
26 * Read the current in-memory orchestration read model.
27 *
28 * @returns Effect containing the latest read model.
29 */
30 readonly getReadModel: () => Effect.Effect<OrchestrationReadModel, never, never>;
31
32 /**
33 * Replay persisted orchestration events from an exclusive sequence cursor.
34 *
35 * @param fromSequenceExclusive - Sequence cursor (exclusive).
36 * @returns Stream containing ordered events.
37 */
38 readonly readEvents: (
39 fromSequenceExclusive: number,
40 ) => Stream.Stream<OrchestrationEvent, OrchestrationEventStoreError, never>;
41
42 /**
43 * Validate and dispatch an unknown command payload.
44 *
45 * Accepts either parsed values or raw JSON strings.
46 *
47 * @param command - Unknown command payload.
48 * @returns Effect containing the sequence of the persisted event.
49 */
50 readonly dispatchUnknownCommand: (
51 command: unknown,
52 ) => Effect.Effect<{ sequence: number }, OrchestrationEngineError, never>;
53
54 /**
55 * Dispatch a validated orchestration command.
56 *
57 * @param command - Valid orchestration command.
58 * @returns Effect containing the sequence of the persisted event.
59 */
60 readonly dispatch: (
61 command: OrchestrationCommand,
62 ) => Effect.Effect<{ sequence: number }, OrchestrationDispatchError, never>;
63
64 /**
65 * Stream persisted domain events in dispatch order.
66 */
67 readonly streamDomainEvents: Stream.Stream<OrchestrationEvent>;
68}
69
70/**
71 * OrchestrationEngineService - Service tag for orchestration engine access.
72 *
73 * @example
74 * ```ts
75 * const program = Effect.gen(function* () {
76 * const engine = yield* OrchestrationEngineService
77 * return yield* engine.getReadModel()
78 * })
79 * ```
80 */
81export class OrchestrationEngineService extends ServiceMap.Service<
82 OrchestrationEngineService,
83 OrchestrationEngineShape
84>()("orchestration/Engine") {}
apps/server/src/orchestration/commandInvariants.test.ts created+161
...@@ -0,0 +1,161 @@
1import { describe, expect, it } from "vitest";
2import type { OrchestrationCommand, OrchestrationReadModel } from "@t3tools/contracts";
3import { Effect } from "effect";
4
5import {
6 findThreadById,
7 listThreadsByProjectId,
8 requireNonNegativeInteger,
9 requireThread,
10 requireThreadAbsent,
11} from "./commandInvariants.ts";
12
13const now = new Date().toISOString();
14
15const readModel: OrchestrationReadModel = {
16 sequence: 2,
17 updatedAt: now,
18 gitByProjectId: {},
19 threads: [
20 {
21 id: "thread-1",
22 projectId: "project-a",
23 title: "Thread A",
24 model: "gpt-5-codex",
25 branch: null,
26 worktreePath: null,
27 createdAt: now,
28 updatedAt: now,
29 latestTurnId: null,
30 latestTurnStartedAt: null,
31 latestTurnCompletedAt: null,
32 latestTurnDurationMs: null,
33 messages: [],
34 session: null,
35 turnDiffSummaries: [],
36 activities: [],
37 error: null,
38 },
39 {
40 id: "thread-2",
41 projectId: "project-b",
42 title: "Thread B",
43 model: "gpt-5-codex",
44 branch: null,
45 worktreePath: null,
46 createdAt: now,
47 updatedAt: now,
48 latestTurnId: null,
49 latestTurnStartedAt: null,
50 latestTurnCompletedAt: null,
51 latestTurnDurationMs: null,
52 messages: [],
53 session: null,
54 turnDiffSummaries: [],
55 activities: [],
56 error: null,
57 },
58 ],
59};
60
61const messageSendCommand: OrchestrationCommand = {
62 type: "message.send",
63 commandId: "cmd-1",
64 threadId: "thread-1",
65 messageId: "msg-1",
66 role: "user",
67 text: "hello",
68 streaming: false,
69 createdAt: now,
70};
71
72describe("commandInvariants", () => {
73 it("finds threads by id and project", () => {
74 expect(findThreadById(readModel, "thread-1")?.projectId).toBe("project-a");
75 expect(findThreadById(readModel, "missing")).toBeUndefined();
76 expect(listThreadsByProjectId(readModel, "project-b").map((thread) => thread.id)).toEqual([
77 "thread-2",
78 ]);
79 });
80
81 it("requires existing thread", async () => {
82 const thread = await Effect.runPromise(
83 requireThread({
84 readModel,
85 command: messageSendCommand,
86 threadId: "thread-1",
87 }),
88 );
89 expect(thread.id).toBe("thread-1");
90
91 await expect(
92 Effect.runPromise(
93 requireThread({
94 readModel,
95 command: messageSendCommand,
96 threadId: "missing",
97 }),
98 ),
99 ).rejects.toThrow("does not exist");
100 });
101
102 it("requires missing thread for create flows", async () => {
103 await Effect.runPromise(
104 requireThreadAbsent({
105 readModel,
106 command: {
107 type: "thread.create",
108 commandId: "cmd-2",
109 threadId: "thread-3",
110 projectId: "project-a",
111 title: "new",
112 model: "gpt-5-codex",
113 branch: null,
114 worktreePath: null,
115 createdAt: now,
116 },
117 threadId: "thread-3",
118 }),
119 );
120
121 await expect(
122 Effect.runPromise(
123 requireThreadAbsent({
124 readModel,
125 command: {
126 type: "thread.create",
127 commandId: "cmd-3",
128 threadId: "thread-1",
129 projectId: "project-a",
130 title: "dup",
131 model: "gpt-5-codex",
132 branch: null,
133 worktreePath: null,
134 createdAt: now,
135 },
136 threadId: "thread-1",
137 }),
138 ),
139 ).rejects.toThrow("already exists");
140 });
141
142 it("requires non-negative integers", async () => {
143 await Effect.runPromise(
144 requireNonNegativeInteger({
145 commandType: "thread.revert",
146 field: "turnCount",
147 value: 0,
148 }),
149 );
150
151 await expect(
152 Effect.runPromise(
153 requireNonNegativeInteger({
154 commandType: "thread.revert",
155 field: "turnCount",
156 value: -1,
157 }),
158 ),
159 ).rejects.toThrow("greater than or equal to 0");
160 });
161});
apps/server/src/orchestration/commandInvariants.ts created+80
...@@ -0,0 +1,80 @@
1import type { OrchestrationCommand, OrchestrationReadModel, OrchestrationThread } from "@t3tools/contracts";
2import { Effect } from "effect";
3
4import { OrchestrationCommandInvariantError } from "./Errors.ts";
5
6function invariantError(commandType: string, detail: string): OrchestrationCommandInvariantError {
7 return new OrchestrationCommandInvariantError({
8 commandType,
9 detail,
10 });
11}
12
13export function findThreadById(
14 readModel: OrchestrationReadModel,
15 threadId: string,
16): OrchestrationThread | undefined {
17 return readModel.threads.find((thread) => thread.id === threadId);
18}
19
20export function listThreadsByProjectId(
21 readModel: OrchestrationReadModel,
22 projectId: string,
23): ReadonlyArray<OrchestrationThread> {
24 return readModel.threads.filter((thread) => thread.projectId === projectId);
25}
26
27export function requireThread(
28 input: {
29 readonly readModel: OrchestrationReadModel;
30 readonly command: OrchestrationCommand;
31 readonly threadId: string;
32 },
33): Effect.Effect<OrchestrationThread, OrchestrationCommandInvariantError> {
34 const thread = findThreadById(input.readModel, input.threadId);
35 if (thread) {
36 return Effect.succeed(thread);
37 }
38 return Effect.fail(
39 invariantError(
40 input.command.type,
41 `Thread '${input.threadId}' does not exist for command '${input.command.type}'.`,
42 ),
43 );
44}
45
46export function requireThreadAbsent(
47 input: {
48 readonly readModel: OrchestrationReadModel;
49 readonly command: OrchestrationCommand;
50 readonly threadId: string;
51 },
52): Effect.Effect<void, OrchestrationCommandInvariantError> {
53 if (!findThreadById(input.readModel, input.threadId)) {
54 return Effect.void;
55 }
56 return Effect.fail(
57 invariantError(
58 input.command.type,
59 `Thread '${input.threadId}' already exists and cannot be created twice.`,
60 ),
61 );
62}
63
64export function requireNonNegativeInteger(
65 input: {
66 readonly commandType: OrchestrationCommand["type"];
67 readonly field: string;
68 readonly value: number;
69 },
70): Effect.Effect<void, OrchestrationCommandInvariantError> {
71 if (Number.isInteger(input.value) && input.value >= 0) {
72 return Effect.void;
73 }
74 return Effect.fail(
75 invariantError(
76 input.commandType,
77 `${input.field} must be an integer greater than or equal to 0.`,
78 ),
79 );
80}
apps/server/src/orchestration/decider.ts created+229
...@@ -0,0 +1,229 @@
1import type {
2 OrchestrationCommand,
3 OrchestrationEvent,
4 OrchestrationReadModel,
5} from "@t3tools/contracts";
6import { Effect } from "effect";
7
8import { OrchestrationCommandInvariantError } from "./Errors.ts";
9import {
10 requireNonNegativeInteger,
11 requireThread,
12 requireThreadAbsent,
13} from "./commandInvariants.ts";
14
15export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand")(function* ({
16 command,
17 readModel,
18}: {
19 readonly command: OrchestrationCommand;
20 readonly readModel: OrchestrationReadModel;
21}): Effect.fn.Return<Omit<OrchestrationEvent, "sequence">, OrchestrationCommandInvariantError> {
22 const eventId = crypto.randomUUID();
23
24 switch (command.type) {
25 case "thread.create":
26 yield* requireThreadAbsent({
27 readModel,
28 command,
29 threadId: command.threadId,
30 });
31 return {
32 eventId,
33 type: "thread.created",
34 aggregateType: "thread",
35 aggregateId: command.threadId,
36 occurredAt: command.createdAt,
37 commandId: command.commandId,
38 payload: {
39 id: command.threadId,
40 projectId: command.projectId,
41 title: command.title,
42 model: command.model,
43 branch: command.branch,
44 worktreePath: command.worktreePath,
45 createdAt: command.createdAt,
46 updatedAt: command.createdAt,
47 },
48 };
49 case "thread.delete":
50 yield* requireThread({
51 readModel,
52 command,
53 threadId: command.threadId,
54 });
55 return {
56 eventId,
57 type: "thread.deleted",
58 aggregateType: "thread",
59 aggregateId: command.threadId,
60 occurredAt: command.createdAt,
61 commandId: command.commandId,
62 payload: {
63 id: command.threadId,
64 deletedAt: command.createdAt,
65 },
66 };
67 case "thread.meta.update":
68 yield* requireThread({
69 readModel,
70 command,
71 threadId: command.threadId,
72 });
73 return {
74 eventId,
75 type: "thread.meta-updated",
76 aggregateType: "thread",
77 aggregateId: command.threadId,
78 occurredAt: command.createdAt,
79 commandId: command.commandId,
80 payload: {
81 threadId: command.threadId,
82 ...(command.title !== undefined ? { title: command.title } : {}),
83 ...(command.model !== undefined ? { model: command.model } : {}),
84 ...(command.branch !== undefined ? { branch: command.branch } : {}),
85 ...(command.worktreePath !== undefined ? { worktreePath: command.worktreePath } : {}),
86 updatedAt: command.createdAt,
87 },
88 };
89 case "message.send":
90 yield* requireThread({
91 readModel,
92 command,
93 threadId: command.threadId,
94 });
95 return {
96 eventId,
97 type: "message.sent",
98 aggregateType: "thread",
99 aggregateId: command.threadId,
100 occurredAt: command.createdAt,
101 commandId: command.commandId,
102 payload: {
103 id: command.messageId,
104 role: command.role,
105 text: command.text,
106 threadId: command.threadId,
107 createdAt: command.createdAt,
108 streaming: command.streaming === true,
109 },
110 };
111 case "thread.session":
112 yield* requireThread({
113 readModel,
114 command,
115 threadId: command.threadId,
116 });
117 return {
118 eventId,
119 type: "thread.session-set",
120 aggregateType: "thread",
121 aggregateId: command.threadId,
122 occurredAt: command.createdAt,
123 commandId: command.commandId,
124 payload: {
125 threadId: command.threadId,
126 session: command.session,
127 },
128 };
129 case "git.readModel.upsert":
130 return {
131 eventId,
132 type: "git.read-model-upsert",
133 aggregateType: "project",
134 aggregateId: command.projectId,
135 occurredAt: command.createdAt,
136 commandId: command.commandId,
137 payload: {
138 projectId: command.projectId,
139 branch: command.branch,
140 hasWorkingTreeChanges: command.hasWorkingTreeChanges,
141 aheadCount: command.aheadCount,
142 behindCount: command.behindCount,
143 updatedAt: command.createdAt,
144 },
145 };
146 case "thread.turnDiff.complete":
147 yield* requireThread({
148 readModel,
149 command,
150 threadId: command.threadId,
151 });
152 return {
153 eventId,
154 type: "thread.turn-diff-completed",
155 aggregateType: "thread",
156 aggregateId: command.threadId,
157 occurredAt: command.createdAt,
158 commandId: command.commandId,
159 payload: {
160 threadId: command.threadId,
161 turnId: command.turnId,
162 completedAt: command.completedAt,
163 ...(command.status !== undefined ? { status: command.status } : {}),
164 files: command.files,
165 ...(command.assistantMessageId !== undefined
166 ? { assistantMessageId: command.assistantMessageId }
167 : {}),
168 ...(command.checkpointTurnCount !== undefined
169 ? { checkpointTurnCount: command.checkpointTurnCount }
170 : {}),
171 },
172 };
173 case "thread.activity.append":
174 yield* requireThread({
175 readModel,
176 command,
177 threadId: command.threadId,
178 });
179 return {
180 eventId,
181 type: "thread.activity-appended",
182 aggregateType: "thread",
183 aggregateId: command.threadId,
184 occurredAt: command.createdAt,
185 commandId: command.commandId,
186 payload: {
187 threadId: command.threadId,
188 activity: command.activity,
189 },
190 };
191 case "thread.revert":
192 yield* requireThread({
193 readModel,
194 command,
195 threadId: command.threadId,
196 });
197 yield* requireNonNegativeInteger({
198 commandType: command.type,
199 field: "turnCount",
200 value: command.turnCount,
201 });
202 yield* requireNonNegativeInteger({
203 commandType: command.type,
204 field: "messageCount",
205 value: command.messageCount,
206 });
207 return {
208 eventId,
209 type: "thread.reverted",
210 aggregateType: "thread",
211 aggregateId: command.threadId,
212 occurredAt: command.createdAt,
213 commandId: command.commandId,
214 payload: {
215 threadId: command.threadId,
216 turnCount: command.turnCount,
217 messageCount: command.messageCount,
218 },
219 };
220 default: {
221 command satisfies never;
222 const fallback = command as any;
223 return yield* new OrchestrationCommandInvariantError({
224 commandType: fallback.type,
225 detail: `Unknown command type: ${fallback.type}`,
226 });
227 }
228 }
229});
apps/server/src/orchestration/projector.test.ts created+258
...@@ -0,0 +1,258 @@
1import type { OrchestrationEvent } from "@t3tools/contracts";
2import { Effect } from "effect";
3import { describe, expect, it } from "vitest";
4
5import { createEmptyReadModel, projectEvent } from "./projector.ts";
6
7describe("orchestration projector", () => {
8 it("applies thread.created events using runtime schema decoding", async () => {
9 const now = new Date().toISOString();
10 const model = createEmptyReadModel(now);
11 const event: OrchestrationEvent = {
12 sequence: 1,
13 eventId: "event-thread-created",
14 type: "thread.created",
15 aggregateType: "thread",
16 aggregateId: "thread-1",
17 occurredAt: now,
18 commandId: "cmd-1",
19 payload: {
20 id: "thread-1",
21 projectId: "project-1",
22 title: "demo",
23 model: "gpt-5-codex",
24 branch: null,
25 worktreePath: null,
26 createdAt: now,
27 updatedAt: now,
28 },
29 };
30
31 const next = await Effect.runPromise(projectEvent(model, event));
32 expect(next.sequence).toBe(1);
33 expect(next.threads).toEqual([
34 {
35 id: "thread-1",
36 projectId: "project-1",
37 title: "demo",
38 model: "gpt-5-codex",
39 branch: null,
40 worktreePath: null,
41 createdAt: now,
42 updatedAt: now,
43 latestTurnId: null,
44 latestTurnStartedAt: null,
45 latestTurnCompletedAt: null,
46 latestTurnDurationMs: null,
47 messages: [],
48 session: null,
49 activities: [],
50 turnDiffSummaries: [],
51 error: null,
52 },
53 ]);
54 });
55
56 it("fails when event payload cannot be decoded by runtime schema", async () => {
57 const now = new Date().toISOString();
58 const model = createEmptyReadModel(now);
59 const event: OrchestrationEvent = {
60 sequence: 1,
61 eventId: "event-message-invalid",
62 type: "message.sent",
63 aggregateType: "thread",
64 aggregateId: "thread-1",
65 occurredAt: now,
66 commandId: "cmd-1",
67 payload: {
68 id: "message-1",
69 threadId: "thread-1",
70 text: "hello",
71 createdAt: now,
72 streaming: false,
73 },
74 };
75
76 await expect(Effect.runPromise(projectEvent(model, event))).rejects.toBeDefined();
77 });
78
79 it("keeps projector forward-compatible for unknown event types", async () => {
80 const now = new Date().toISOString();
81 const model = createEmptyReadModel(now);
82 const event: OrchestrationEvent = {
83 sequence: 7,
84 eventId: "event-unknown",
85 type: "custom.event.ignored",
86 aggregateType: "unknown",
87 aggregateId: "aggregate-1",
88 occurredAt: "2026-01-01T00:00:00.000Z",
89 commandId: null,
90 payload: { test: true },
91 };
92
93 const next = await Effect.runPromise(projectEvent(model, event));
94 expect(next.sequence).toBe(7);
95 expect(next.updatedAt).toBe("2026-01-01T00:00:00.000Z");
96 expect(next.threads).toEqual([]);
97 });
98
99 it("tracks latest turn timing from session lifecycle events", async () => {
100 const createdAt = "2026-02-23T08:00:00.000Z";
101 const startedAt = "2026-02-23T08:00:05.000Z";
102 const completedAt = "2026-02-23T08:00:11.250Z";
103 const model = createEmptyReadModel(createdAt);
104
105 const createdEvent: OrchestrationEvent = {
106 sequence: 1,
107 eventId: "thread-created",
108 type: "thread.created",
109 aggregateType: "thread",
110 aggregateId: "thread-1",
111 occurredAt: createdAt,
112 commandId: "cmd-create",
113 payload: {
114 id: "thread-1",
115 projectId: "project-1",
116 title: "demo",
117 model: "gpt-5.3-codex",
118 branch: null,
119 worktreePath: null,
120 createdAt,
121 updatedAt: createdAt,
122 },
123 };
124
125 const runningSessionEvent: OrchestrationEvent = {
126 sequence: 2,
127 eventId: "session-running",
128 type: "thread.session-set",
129 aggregateType: "thread",
130 aggregateId: "thread-1",
131 occurredAt: startedAt,
132 commandId: "cmd-running",
133 payload: {
134 threadId: "thread-1",
135 session: {
136 sessionId: "session-1",
137 status: "running",
138 provider: "codex",
139 threadId: "thread-1",
140 activeTurnId: "turn-1",
141 createdAt,
142 updatedAt: startedAt,
143 lastError: null,
144 },
145 },
146 };
147
148 const readySessionEvent: OrchestrationEvent = {
149 sequence: 3,
150 eventId: "session-ready",
151 type: "thread.session-set",
152 aggregateType: "thread",
153 aggregateId: "thread-1",
154 occurredAt: completedAt,
155 commandId: "cmd-ready",
156 payload: {
157 threadId: "thread-1",
158 session: {
159 sessionId: "session-1",
160 status: "ready",
161 provider: "codex",
162 threadId: "thread-1",
163 activeTurnId: null,
164 createdAt,
165 updatedAt: completedAt,
166 lastError: null,
167 },
168 },
169 };
170
171 const afterCreate = await Effect.runPromise(projectEvent(model, createdEvent));
172 const afterRunning = await Effect.runPromise(projectEvent(afterCreate, runningSessionEvent));
173 const afterReady = await Effect.runPromise(projectEvent(afterRunning, readySessionEvent));
174 const thread = afterReady.threads[0];
175
176 expect(thread?.latestTurnId).toBe("turn-1");
177 expect(thread?.latestTurnStartedAt).toBe(startedAt);
178 expect(thread?.latestTurnCompletedAt).toBe(completedAt);
179 expect(thread?.latestTurnDurationMs).toBe(6250);
180 });
181
182 it("marks assistant messages completed using non-streaming updates", async () => {
183 const createdAt = "2026-02-23T09:00:00.000Z";
184 const deltaAt = "2026-02-23T09:00:01.000Z";
185 const completeAt = "2026-02-23T09:00:03.500Z";
186 const model = createEmptyReadModel(createdAt);
187
188 const createdEvent: OrchestrationEvent = {
189 sequence: 1,
190 eventId: "thread-created",
191 type: "thread.created",
192 aggregateType: "thread",
193 aggregateId: "thread-1",
194 occurredAt: createdAt,
195 commandId: "cmd-create",
196 payload: {
197 id: "thread-1",
198 projectId: "project-1",
199 title: "demo",
200 model: "gpt-5.3-codex",
201 branch: null,
202 worktreePath: null,
203 createdAt,
204 updatedAt: createdAt,
205 },
206 };
207
208 const deltaEvent: OrchestrationEvent = {
209 sequence: 2,
210 eventId: "assistant-delta",
211 type: "message.sent",
212 aggregateType: "thread",
213 aggregateId: "thread-1",
214 occurredAt: deltaAt,
215 commandId: "cmd-delta",
216 payload: {
217 id: "assistant:msg-1",
218 role: "assistant",
219 text: "hello",
220 threadId: "thread-1",
221 createdAt: deltaAt,
222 streaming: true,
223 },
224 };
225
226 const completionEvent: OrchestrationEvent = {
227 sequence: 3,
228 eventId: "assistant-complete",
229 type: "message.sent",
230 aggregateType: "thread",
231 aggregateId: "thread-1",
232 occurredAt: completeAt,
233 commandId: "cmd-complete",
234 payload: {
235 id: "assistant:msg-1",
236 role: "assistant",
237 text: "",
238 threadId: "thread-1",
239 createdAt: completeAt,
240 streaming: false,
241 },
242 };
243
244 const afterCreate = await Effect.runPromise(projectEvent(model, createdEvent));
245 const afterDelta = await Effect.runPromise(projectEvent(afterCreate, deltaEvent));
246 const afterComplete = await Effect.runPromise(projectEvent(afterDelta, completionEvent));
247 const message = afterComplete.threads[0]?.messages[0];
248
249 expect(message).toEqual({
250 id: "assistant:msg-1",
251 role: "assistant",
252 text: "hello",
253 createdAt: deltaAt,
254 completedAt: completeAt,
255 streaming: false,
256 });
257 });
258});
apps/server/src/orchestration/projector.ts created+475
...@@ -0,0 +1,475 @@
1import type {
2 OrchestrationEvent,
3 OrchestrationGitReadModel,
4 OrchestrationMessage,
5 OrchestrationReadModel,
6 OrchestrationSession,
7 OrchestrationThread,
8} from "@t3tools/contracts";
9import {
10 OrchestrationThreadActivitySchema,
11 OrchestrationGitReadModelSchema,
12 OrchestrationMessageSchema,
13 OrchestrationSessionSchema,
14 OrchestrationThreadSchema,
15 OrchestrationTurnDiffFileSchema,
16 OrchestrationTurnDiffSummarySchema,
17} from "@t3tools/contracts";
18import { Effect, Schema } from "effect";
19
20import { toProjectorDecodeError, type OrchestrationProjectorDecodeError } from "./Errors.ts";
21
22type ThreadPatch = Partial<Omit<OrchestrationThread, "id" | "projectId">>;
23
24function updateThread(
25 threads: ReadonlyArray<OrchestrationThread>,
26 threadId: string,
27 patch: ThreadPatch,
28): OrchestrationThread[] {
29 return threads.map((thread) => (thread.id === threadId ? { ...thread, ...patch } : thread));
30}
31
32function decodeForEvent<A>(
33 schema: Schema.Schema<A>,
34 value: unknown,
35 eventType: string,
36 field: string,
37): Effect.Effect<A, OrchestrationProjectorDecodeError> {
38 return Schema.decodeUnknownEffect(schema)(value).pipe(
39 Effect.mapError(toProjectorDecodeError(`${eventType}:${field}`)),
40 ) as Effect.Effect<A, OrchestrationProjectorDecodeError>;
41}
42
43const ThreadCreatedPayloadSchema = Schema.Struct({
44 id: Schema.String,
45 projectId: Schema.String,
46 title: Schema.String,
47 model: Schema.String,
48 branch: Schema.NullOr(Schema.String),
49 worktreePath: Schema.NullOr(Schema.String),
50 createdAt: Schema.String,
51 updatedAt: Schema.String,
52});
53
54const ThreadMetaUpdatedPayloadSchema = Schema.Struct({
55 threadId: Schema.String,
56 title: Schema.optional(Schema.String),
57 model: Schema.optional(Schema.String),
58 branch: Schema.optional(Schema.NullOr(Schema.String)),
59 worktreePath: Schema.optional(Schema.NullOr(Schema.String)),
60 updatedAt: Schema.String,
61});
62
63const MessageSentPayloadSchema = Schema.Struct({
64 id: Schema.String,
65 role: Schema.Literals(["user", "assistant"]),
66 text: Schema.String,
67 threadId: Schema.String,
68 createdAt: Schema.String,
69 streaming: Schema.Boolean,
70});
71
72const ThreadSessionSetPayloadSchema = Schema.Struct({
73 threadId: Schema.String,
74 session: OrchestrationSessionSchema,
75});
76
77const ThreadTurnDiffCompletedPayloadSchema = Schema.Struct({
78 threadId: Schema.String,
79 turnId: Schema.String,
80 completedAt: Schema.String,
81 status: Schema.optional(Schema.String),
82 files: Schema.Array(OrchestrationTurnDiffFileSchema),
83 assistantMessageId: Schema.optional(Schema.String),
84 checkpointTurnCount: Schema.optional(Schema.Number),
85});
86
87const ThreadRevertedPayloadSchema = Schema.Struct({
88 threadId: Schema.String,
89 turnCount: Schema.Number,
90 messageCount: Schema.Number,
91});
92
93const ThreadActivityAppendedPayloadSchema = Schema.Struct({
94 threadId: Schema.String,
95 activity: OrchestrationThreadActivitySchema,
96});
97
98export function createEmptyReadModel(nowIso: string): OrchestrationReadModel {
99 return {
100 sequence: 0,
101 threads: [],
102 gitByProjectId: {},
103 updatedAt: nowIso,
104 };
105}
106
107export function projectEvent(
108 model: OrchestrationReadModel,
109 event: OrchestrationEvent,
110): Effect.Effect<OrchestrationReadModel, OrchestrationProjectorDecodeError> {
111 const nextBase: OrchestrationReadModel = {
112 ...model,
113 sequence: event.sequence,
114 updatedAt: event.occurredAt,
115 };
116
117 switch (event.type) {
118 case "thread.created":
119 return Effect.gen(function* () {
120 const payload = yield* decodeForEvent(
121 ThreadCreatedPayloadSchema,
122 event.payload,
123 event.type,
124 "payload",
125 );
126 const thread: OrchestrationThread = yield* decodeForEvent(
127 OrchestrationThreadSchema,
128 {
129 id: payload.id,
130 projectId: payload.projectId,
131 title: payload.title,
132 model: payload.model,
133 branch: payload.branch,
134 worktreePath: payload.worktreePath,
135 createdAt: payload.createdAt,
136 updatedAt: payload.updatedAt,
137 latestTurnId: null,
138 latestTurnStartedAt: null,
139 latestTurnCompletedAt: null,
140 latestTurnDurationMs: null,
141 messages: [],
142 session: null,
143 turnDiffSummaries: [],
144 activities: [],
145 error: null,
146 },
147 event.type,
148 "thread",
149 );
150 const existing = nextBase.threads.find((entry) => entry.id === thread.id);
151 return {
152 ...nextBase,
153 threads: existing
154 ? nextBase.threads.map((entry) => (entry.id === thread.id ? thread : entry))
155 : [...nextBase.threads, thread],
156 };
157 });
158 case "thread.deleted":
159 return Effect.succeed({
160 ...nextBase,
161 threads: nextBase.threads.filter((thread) => thread.id !== event.aggregateId),
162 });
163 case "thread.meta-updated":
164 return decodeForEvent(
165 ThreadMetaUpdatedPayloadSchema,
166 event.payload,
167 event.type,
168 "payload",
169 ).pipe(
170 Effect.map((payload) => {
171 const thread = nextBase.threads.find((entry) => entry.id === payload.threadId);
172 if (!thread) {
173 return nextBase;
174 }
175 return {
176 ...nextBase,
177 threads: updateThread(nextBase.threads, payload.threadId, {
178 ...(payload.title !== undefined ? { title: payload.title } : {}),
179 ...(payload.model !== undefined ? { model: payload.model } : {}),
180 ...(payload.branch !== undefined ? { branch: payload.branch } : {}),
181 ...(payload.worktreePath !== undefined ? { worktreePath: payload.worktreePath } : {}),
182 updatedAt: payload.updatedAt,
183 }),
184 };
185 }),
186 );
187 case "message.sent":
188 return Effect.gen(function* () {
189 const payload = yield* decodeForEvent(
190 MessageSentPayloadSchema,
191 event.payload,
192 event.type,
193 "payload",
194 );
195 const targetThread = nextBase.threads.find((thread) => thread.id === payload.threadId);
196 if (!targetThread) {
197 return nextBase;
198 }
199
200 const message: OrchestrationMessage = yield* decodeForEvent(
201 OrchestrationMessageSchema,
202 {
203 id: payload.id,
204 role: payload.role,
205 text: payload.text,
206 createdAt: payload.createdAt,
207 ...(payload.streaming ? {} : { completedAt: event.occurredAt }),
208 streaming: payload.streaming,
209 },
210 event.type,
211 "message",
212 );
213 const existingMessage = targetThread.messages.find((entry) => entry.id === message.id);
214 const nextMessages = existingMessage
215 ? targetThread.messages.map((entry) =>
216 entry.id === message.id
217 ? {
218 ...entry,
219 text: message.streaming
220 ? `${entry.text}${message.text}`
221 : message.text.length > 0
222 ? message.text
223 : entry.text,
224 streaming: message.streaming,
225 createdAt: entry.createdAt,
226 ...(message.streaming
227 ? { completedAt: undefined }
228 : { completedAt: message.completedAt ?? event.occurredAt }),
229 }
230 : entry,
231 )
232 : !message.streaming && message.text.length === 0
233 ? targetThread.messages
234 : [...targetThread.messages, message];
235
236 return {
237 ...nextBase,
238 threads: updateThread(nextBase.threads, payload.threadId, {
239 messages: nextMessages,
240 updatedAt: event.occurredAt,
241 }),
242 };
243 });
244 case "thread.session-set":
245 return Effect.gen(function* () {
246 const payload = yield* decodeForEvent(
247 ThreadSessionSetPayloadSchema,
248 event.payload,
249 event.type,
250 "payload",
251 );
252 const thread = nextBase.threads.find((entry) => entry.id === payload.threadId);
253 if (!thread) {
254 return nextBase;
255 }
256
257 const session: OrchestrationSession = yield* decodeForEvent(
258 OrchestrationSessionSchema,
259 {
260 ...payload.session,
261 threadId: payload.threadId,
262 },
263 event.type,
264 "session",
265 );
266
267 const sessionUpdatedAt = session.updatedAt;
268 const turnTimingPatch: ThreadPatch =
269 session.status === "running" && session.activeTurnId
270 ? (() => {
271 const isSameTurn = thread.latestTurnId === session.activeTurnId;
272 return {
273 latestTurnId: session.activeTurnId,
274 latestTurnStartedAt:
275 isSameTurn && thread.latestTurnStartedAt
276 ? thread.latestTurnStartedAt
277 : sessionUpdatedAt,
278 ...(isSameTurn
279 ? {}
280 : {
281 latestTurnCompletedAt: null,
282 latestTurnDurationMs: null,
283 }),
284 };
285 })()
286 : session.activeTurnId === null && thread.latestTurnId && thread.latestTurnStartedAt
287 ? (() => {
288 const latestTurnCompletedAt = thread.latestTurnCompletedAt ?? sessionUpdatedAt;
289 const startedAtMs = Date.parse(thread.latestTurnStartedAt);
290 const completedAtMs = Date.parse(latestTurnCompletedAt);
291 const latestTurnDurationMs =
292 !Number.isNaN(startedAtMs) &&
293 !Number.isNaN(completedAtMs) &&
294 completedAtMs >= startedAtMs
295 ? completedAtMs - startedAtMs
296 : thread.latestTurnDurationMs;
297
298 return {
299 latestTurnCompletedAt,
300 latestTurnDurationMs,
301 };
302 })()
303 : {};
304
305 return {
306 ...nextBase,
307 threads: updateThread(nextBase.threads, payload.threadId, {
308 session,
309 updatedAt: event.occurredAt,
310 error: session.lastError,
311 ...turnTimingPatch,
312 }),
313 };
314 });
315 case "thread.turn-diff-completed":
316 return Effect.gen(function* () {
317 const payload = yield* decodeForEvent(
318 ThreadTurnDiffCompletedPayloadSchema,
319 event.payload,
320 event.type,
321 "payload",
322 );
323 const thread = nextBase.threads.find((entry) => entry.id === payload.threadId);
324 if (!thread) {
325 return nextBase;
326 }
327
328 const nextSummary = yield* decodeForEvent(
329 OrchestrationTurnDiffSummarySchema,
330 {
331 turnId: payload.turnId,
332 completedAt: payload.completedAt,
333 ...(payload.status !== undefined ? { status: payload.status } : {}),
334 files: payload.files,
335 ...(payload.assistantMessageId !== undefined
336 ? { assistantMessageId: payload.assistantMessageId }
337 : {}),
338 ...(payload.checkpointTurnCount !== undefined
339 ? { checkpointTurnCount: payload.checkpointTurnCount }
340 : {}),
341 },
342 event.type,
343 "turnDiffSummary",
344 );
345 const turnDiffSummaries = [
346 ...thread.turnDiffSummaries.filter((summary) => summary.turnId !== payload.turnId),
347 nextSummary,
348 ].toSorted((left, right) => left.completedAt.localeCompare(right.completedAt));
349 const startedAtMs = thread.latestTurnStartedAt
350 ? Date.parse(thread.latestTurnStartedAt)
351 : Number.NaN;
352 const completedAtMs = Date.parse(nextSummary.completedAt);
353 const latestTurnDurationMs =
354 !Number.isNaN(startedAtMs) &&
355 !Number.isNaN(completedAtMs) &&
356 completedAtMs >= startedAtMs
357 ? completedAtMs - startedAtMs
358 : thread.latestTurnDurationMs;
359
360 return {
361 ...nextBase,
362 threads: updateThread(nextBase.threads, payload.threadId, {
363 latestTurnId: payload.turnId,
364 latestTurnCompletedAt: nextSummary.completedAt,
365 latestTurnDurationMs,
366 turnDiffSummaries,
367 updatedAt: event.occurredAt,
368 }),
369 };
370 });
371 case "thread.reverted":
372 return decodeForEvent(ThreadRevertedPayloadSchema, event.payload, event.type, "payload").pipe(
373 Effect.map((payload) => {
374 const thread = nextBase.threads.find((entry) => entry.id === payload.threadId);
375 if (!thread) {
376 return nextBase;
377 }
378
379 const targetTurnCount = Math.max(0, Math.floor(payload.turnCount));
380 const targetMessageCount = Math.max(0, Math.floor(payload.messageCount));
381 const sortedSummaries = [...thread.turnDiffSummaries].toSorted((left, right) =>
382 left.completedAt.localeCompare(right.completedAt),
383 );
384 const inferredTurnCountByTurnId = new Map<string, number>();
385 for (let index = 0; index < sortedSummaries.length; index += 1) {
386 const summary = sortedSummaries[index];
387 if (!summary) continue;
388 inferredTurnCountByTurnId.set(summary.turnId, index + 1);
389 }
390
391 const turnDiffSummaries = thread.turnDiffSummaries
392 .filter((summary) => {
393 const checkpointTurnCount =
394 summary.checkpointTurnCount ?? inferredTurnCountByTurnId.get(summary.turnId) ?? 0;
395 return checkpointTurnCount <= targetTurnCount;
396 })
397 .toSorted((left, right) => left.completedAt.localeCompare(right.completedAt));
398 const latestSummary =
399 turnDiffSummaries.length > 0 ? turnDiffSummaries[turnDiffSummaries.length - 1] : null;
400 const activities = thread.activities.filter((activity) => {
401 if (!activity.turnId) {
402 return true;
403 }
404 const checkpointTurnCount = inferredTurnCountByTurnId.get(activity.turnId);
405 if (checkpointTurnCount === undefined) {
406 return true;
407 }
408 return checkpointTurnCount <= targetTurnCount;
409 });
410
411 return {
412 ...nextBase,
413 threads: updateThread(nextBase.threads, payload.threadId, {
414 messages: thread.messages.slice(
415 0,
416 Math.min(targetMessageCount, thread.messages.length),
417 ),
418 turnDiffSummaries,
419 latestTurnId: latestSummary?.turnId ?? null,
420 latestTurnStartedAt: null,
421 latestTurnCompletedAt: latestSummary?.completedAt ?? null,
422 latestTurnDurationMs: null,
423 activities,
424 updatedAt: event.occurredAt,
425 }),
426 };
427 }),
428 );
429 case "thread.activity-appended":
430 return decodeForEvent(
431 ThreadActivityAppendedPayloadSchema,
432 event.payload,
433 event.type,
434 "payload",
435 ).pipe(
436 Effect.map((payload) => {
437 const thread = nextBase.threads.find((entry) => entry.id === payload.threadId);
438 if (!thread) {
439 return nextBase;
440 }
441 const activities = [
442 ...thread.activities.filter((activity) => activity.id !== payload.activity.id),
443 payload.activity,
444 ]
445 .toSorted((left, right) => left.createdAt.localeCompare(right.createdAt))
446 .slice(-500);
447
448 return {
449 ...nextBase,
450 threads: updateThread(nextBase.threads, payload.threadId, {
451 activities,
452 updatedAt: event.occurredAt,
453 }),
454 };
455 }),
456 );
457 case "git.read-model-upsert":
458 return decodeForEvent(
459 OrchestrationGitReadModelSchema,
460 event.payload,
461 event.type,
462 "payload",
463 ).pipe(
464 Effect.map((gitReadModel: OrchestrationGitReadModel) => ({
465 ...nextBase,
466 gitByProjectId: {
467 ...nextBase.gitByProjectId,
468 [gitReadModel.projectId]: gitReadModel,
469 },
470 })),
471 );
472 default:
473 return Effect.succeed(nextBase);
474 }
475}
apps/server/src/orchestration/reducer.test.ts deleted-258
...@@ -1,258 +0,0 @@
1import type { OrchestrationEvent } from "@t3tools/contracts";
2import { Effect } from "effect";
3import { describe, expect, it } from "vitest";
4
5import { createEmptyReadModel, reduceEvent } from "./reducer.ts";
6
7describe("orchestration reducer", () => {
8 it("applies thread.created events using runtime schema decoding", async () => {
9 const now = new Date().toISOString();
10 const model = createEmptyReadModel(now);
11 const event: OrchestrationEvent = {
12 sequence: 1,
13 eventId: "event-thread-created",
14 type: "thread.created",
15 aggregateType: "thread",
16 aggregateId: "thread-1",
17 occurredAt: now,
18 commandId: "cmd-1",
19 payload: {
20 id: "thread-1",
21 projectId: "project-1",
22 title: "demo",
23 model: "gpt-5-codex",
24 branch: null,
25 worktreePath: null,
26 createdAt: now,
27 updatedAt: now,
28 },
29 };
30
31 const next = await Effect.runPromise(reduceEvent(model, event));
32 expect(next.sequence).toBe(1);
33 expect(next.threads).toEqual([
34 {
35 id: "thread-1",
36 projectId: "project-1",
37 title: "demo",
38 model: "gpt-5-codex",
39 branch: null,
40 worktreePath: null,
41 createdAt: now,
42 updatedAt: now,
43 latestTurnId: null,
44 latestTurnStartedAt: null,
45 latestTurnCompletedAt: null,
46 latestTurnDurationMs: null,
47 messages: [],
48 session: null,
49 activities: [],
50 turnDiffSummaries: [],
51 error: null,
52 },
53 ]);
54 });
55
56 it("fails when event payload cannot be decoded by runtime schema", async () => {
57 const now = new Date().toISOString();
58 const model = createEmptyReadModel(now);
59 const event: OrchestrationEvent = {
60 sequence: 1,
61 eventId: "event-message-invalid",
62 type: "message.sent",
63 aggregateType: "thread",
64 aggregateId: "thread-1",
65 occurredAt: now,
66 commandId: "cmd-1",
67 payload: {
68 id: "message-1",
69 threadId: "thread-1",
70 text: "hello",
71 createdAt: now,
72 streaming: false,
73 },
74 };
75
76 await expect(Effect.runPromise(reduceEvent(model, event))).rejects.toBeDefined();
77 });
78
79 it("keeps reducer forward-compatible for unknown event types", async () => {
80 const now = new Date().toISOString();
81 const model = createEmptyReadModel(now);
82 const event: OrchestrationEvent = {
83 sequence: 7,
84 eventId: "event-unknown",
85 type: "custom.event.ignored",
86 aggregateType: "unknown",
87 aggregateId: "aggregate-1",
88 occurredAt: "2026-01-01T00:00:00.000Z",
89 commandId: null,
90 payload: { test: true },
91 };
92
93 const next = await Effect.runPromise(reduceEvent(model, event));
94 expect(next.sequence).toBe(7);
95 expect(next.updatedAt).toBe("2026-01-01T00:00:00.000Z");
96 expect(next.threads).toEqual([]);
97 });
98
99 it("tracks latest turn timing from session lifecycle events", async () => {
100 const createdAt = "2026-02-23T08:00:00.000Z";
101 const startedAt = "2026-02-23T08:00:05.000Z";
102 const completedAt = "2026-02-23T08:00:11.250Z";
103 const model = createEmptyReadModel(createdAt);
104
105 const createdEvent: OrchestrationEvent = {
106 sequence: 1,
107 eventId: "thread-created",
108 type: "thread.created",
109 aggregateType: "thread",
110 aggregateId: "thread-1",
111 occurredAt: createdAt,
112 commandId: "cmd-create",
113 payload: {
114 id: "thread-1",
115 projectId: "project-1",
116 title: "demo",
117 model: "gpt-5.3-codex",
118 branch: null,
119 worktreePath: null,
120 createdAt,
121 updatedAt: createdAt,
122 },
123 };
124
125 const runningSessionEvent: OrchestrationEvent = {
126 sequence: 2,
127 eventId: "session-running",
128 type: "thread.session-set",
129 aggregateType: "thread",
130 aggregateId: "thread-1",
131 occurredAt: startedAt,
132 commandId: "cmd-running",
133 payload: {
134 threadId: "thread-1",
135 session: {
136 sessionId: "session-1",
137 status: "running",
138 provider: "codex",
139 threadId: "thread-1",
140 activeTurnId: "turn-1",
141 createdAt,
142 updatedAt: startedAt,
143 lastError: null,
144 },
145 },
146 };
147
148 const readySessionEvent: OrchestrationEvent = {
149 sequence: 3,
150 eventId: "session-ready",
151 type: "thread.session-set",
152 aggregateType: "thread",
153 aggregateId: "thread-1",
154 occurredAt: completedAt,
155 commandId: "cmd-ready",
156 payload: {
157 threadId: "thread-1",
158 session: {
159 sessionId: "session-1",
160 status: "ready",
161 provider: "codex",
162 threadId: "thread-1",
163 activeTurnId: null,
164 createdAt,
165 updatedAt: completedAt,
166 lastError: null,
167 },
168 },
169 };
170
171 const afterCreate = await Effect.runPromise(reduceEvent(model, createdEvent));
172 const afterRunning = await Effect.runPromise(reduceEvent(afterCreate, runningSessionEvent));
173 const afterReady = await Effect.runPromise(reduceEvent(afterRunning, readySessionEvent));
174 const thread = afterReady.threads[0];
175
176 expect(thread?.latestTurnId).toBe("turn-1");
177 expect(thread?.latestTurnStartedAt).toBe(startedAt);
178 expect(thread?.latestTurnCompletedAt).toBe(completedAt);
179 expect(thread?.latestTurnDurationMs).toBe(6250);
180 });
181
182 it("marks assistant messages completed using non-streaming updates", async () => {
183 const createdAt = "2026-02-23T09:00:00.000Z";
184 const deltaAt = "2026-02-23T09:00:01.000Z";
185 const completeAt = "2026-02-23T09:00:03.500Z";
186 const model = createEmptyReadModel(createdAt);
187
188 const createdEvent: OrchestrationEvent = {
189 sequence: 1,
190 eventId: "thread-created",
191 type: "thread.created",
192 aggregateType: "thread",
193 aggregateId: "thread-1",
194 occurredAt: createdAt,
195 commandId: "cmd-create",
196 payload: {
197 id: "thread-1",
198 projectId: "project-1",
199 title: "demo",
200 model: "gpt-5.3-codex",
201 branch: null,
202 worktreePath: null,
203 createdAt,
204 updatedAt: createdAt,
205 },
206 };
207
208 const deltaEvent: OrchestrationEvent = {
209 sequence: 2,
210 eventId: "assistant-delta",
211 type: "message.sent",
212 aggregateType: "thread",
213 aggregateId: "thread-1",
214 occurredAt: deltaAt,
215 commandId: "cmd-delta",
216 payload: {
217 id: "assistant:msg-1",
218 role: "assistant",
219 text: "hello",
220 threadId: "thread-1",
221 createdAt: deltaAt,
222 streaming: true,
223 },
224 };
225
226 const completionEvent: OrchestrationEvent = {
227 sequence: 3,
228 eventId: "assistant-complete",
229 type: "message.sent",
230 aggregateType: "thread",
231 aggregateId: "thread-1",
232 occurredAt: completeAt,
233 commandId: "cmd-complete",
234 payload: {
235 id: "assistant:msg-1",
236 role: "assistant",
237 text: "",
238 threadId: "thread-1",
239 createdAt: completeAt,
240 streaming: false,
241 },
242 };
243
244 const afterCreate = await Effect.runPromise(reduceEvent(model, createdEvent));
245 const afterDelta = await Effect.runPromise(reduceEvent(afterCreate, deltaEvent));
246 const afterComplete = await Effect.runPromise(reduceEvent(afterDelta, completionEvent));
247 const message = afterComplete.threads[0]?.messages[0];
248
249 expect(message).toEqual({
250 id: "assistant:msg-1",
251 role: "assistant",
252 text: "hello",
253 createdAt: deltaAt,
254 completedAt: completeAt,
255 streaming: false,
256 });
257 });
258});
apps/server/src/orchestration/reducer.ts deleted-475
...@@ -1,475 +0,0 @@
1import type {
2 OrchestrationEvent,
3 OrchestrationGitReadModel,
4 OrchestrationMessage,
5 OrchestrationReadModel,
6 OrchestrationSession,
7 OrchestrationThread,
8} from "@t3tools/contracts";
9import {
10 OrchestrationThreadActivitySchema,
11 OrchestrationGitReadModelSchema,
12 OrchestrationMessageSchema,
13 OrchestrationSessionSchema,
14 OrchestrationThreadSchema,
15 OrchestrationTurnDiffFileSchema,
16 OrchestrationTurnDiffSummarySchema,
17} from "@t3tools/contracts";
18import { Effect, Schema } from "effect";
19
20import { toReducerDecodeError, type OrchestrationReducerDecodeError } from "./Errors.ts";
21
22type ThreadPatch = Partial<Omit<OrchestrationThread, "id" | "projectId">>;
23
24function updateThread(
25 threads: ReadonlyArray<OrchestrationThread>,
26 threadId: string,
27 patch: ThreadPatch,
28): OrchestrationThread[] {
29 return threads.map((thread) => (thread.id === threadId ? { ...thread, ...patch } : thread));
30}
31
32function decodeForEvent<A>(
33 schema: Schema.Schema<A>,
34 value: unknown,
35 eventType: string,
36 field: string,
37): Effect.Effect<A, OrchestrationReducerDecodeError> {
38 return Schema.decodeUnknownEffect(schema)(value).pipe(
39 Effect.mapError(toReducerDecodeError(`${eventType}:${field}`)),
40 ) as Effect.Effect<A, OrchestrationReducerDecodeError>;
41}
42
43const ThreadCreatedPayloadSchema = Schema.Struct({
44 id: Schema.String,
45 projectId: Schema.String,
46 title: Schema.String,
47 model: Schema.String,
48 branch: Schema.NullOr(Schema.String),
49 worktreePath: Schema.NullOr(Schema.String),
50 createdAt: Schema.String,
51 updatedAt: Schema.String,
52});
53
54const ThreadMetaUpdatedPayloadSchema = Schema.Struct({
55 threadId: Schema.String,
56 title: Schema.optional(Schema.String),
57 model: Schema.optional(Schema.String),
58 branch: Schema.optional(Schema.NullOr(Schema.String)),
59 worktreePath: Schema.optional(Schema.NullOr(Schema.String)),
60 updatedAt: Schema.String,
61});
62
63const MessageSentPayloadSchema = Schema.Struct({
64 id: Schema.String,
65 role: Schema.Literals(["user", "assistant"]),
66 text: Schema.String,
67 threadId: Schema.String,
68 createdAt: Schema.String,
69 streaming: Schema.Boolean,
70});
71
72const ThreadSessionSetPayloadSchema = Schema.Struct({
73 threadId: Schema.String,
74 session: OrchestrationSessionSchema,
75});
76
77const ThreadTurnDiffCompletedPayloadSchema = Schema.Struct({
78 threadId: Schema.String,
79 turnId: Schema.String,
80 completedAt: Schema.String,
81 status: Schema.optional(Schema.String),
82 files: Schema.Array(OrchestrationTurnDiffFileSchema),
83 assistantMessageId: Schema.optional(Schema.String),
84 checkpointTurnCount: Schema.optional(Schema.Number),
85});
86
87const ThreadRevertedPayloadSchema = Schema.Struct({
88 threadId: Schema.String,
89 turnCount: Schema.Number,
90 messageCount: Schema.Number,
91});
92
93const ThreadActivityAppendedPayloadSchema = Schema.Struct({
94 threadId: Schema.String,
95 activity: OrchestrationThreadActivitySchema,
96});
97
98export function createEmptyReadModel(nowIso: string): OrchestrationReadModel {
99 return {
100 sequence: 0,
101 threads: [],
102 gitByProjectId: {},
103 updatedAt: nowIso,
104 };
105}
106
107export function reduceEvent(
108 model: OrchestrationReadModel,
109 event: OrchestrationEvent,
110): Effect.Effect<OrchestrationReadModel, OrchestrationReducerDecodeError> {
111 const nextBase: OrchestrationReadModel = {
112 ...model,
113 sequence: event.sequence,
114 updatedAt: event.occurredAt,
115 };
116
117 switch (event.type) {
118 case "thread.created":
119 return Effect.gen(function* () {
120 const payload = yield* decodeForEvent(
121 ThreadCreatedPayloadSchema,
122 event.payload,
123 event.type,
124 "payload",
125 );
126 const thread: OrchestrationThread = yield* decodeForEvent(
127 OrchestrationThreadSchema,
128 {
129 id: payload.id,
130 projectId: payload.projectId,
131 title: payload.title,
132 model: payload.model,
133 branch: payload.branch,
134 worktreePath: payload.worktreePath,
135 createdAt: payload.createdAt,
136 updatedAt: payload.updatedAt,
137 latestTurnId: null,
138 latestTurnStartedAt: null,
139 latestTurnCompletedAt: null,
140 latestTurnDurationMs: null,
141 messages: [],
142 session: null,
143 turnDiffSummaries: [],
144 activities: [],
145 error: null,
146 },
147 event.type,
148 "thread",
149 );
150 const existing = nextBase.threads.find((entry) => entry.id === thread.id);
151 return {
152 ...nextBase,
153 threads: existing
154 ? nextBase.threads.map((entry) => (entry.id === thread.id ? thread : entry))
155 : [...nextBase.threads, thread],
156 };
157 });
158 case "thread.deleted":
159 return Effect.succeed({
160 ...nextBase,
161 threads: nextBase.threads.filter((thread) => thread.id !== event.aggregateId),
162 });
163 case "thread.meta-updated":
164 return decodeForEvent(
165 ThreadMetaUpdatedPayloadSchema,
166 event.payload,
167 event.type,
168 "payload",
169 ).pipe(
170 Effect.map((payload) => {
171 const thread = nextBase.threads.find((entry) => entry.id === payload.threadId);
172 if (!thread) {
173 return nextBase;
174 }
175 return {
176 ...nextBase,
177 threads: updateThread(nextBase.threads, payload.threadId, {
178 ...(payload.title !== undefined ? { title: payload.title } : {}),
179 ...(payload.model !== undefined ? { model: payload.model } : {}),
180 ...(payload.branch !== undefined ? { branch: payload.branch } : {}),
181 ...(payload.worktreePath !== undefined ? { worktreePath: payload.worktreePath } : {}),
182 updatedAt: payload.updatedAt,
183 }),
184 };
185 }),
186 );
187 case "message.sent":
188 return Effect.gen(function* () {
189 const payload = yield* decodeForEvent(
190 MessageSentPayloadSchema,
191 event.payload,
192 event.type,
193 "payload",
194 );
195 const targetThread = nextBase.threads.find((thread) => thread.id === payload.threadId);
196 if (!targetThread) {
197 return nextBase;
198 }
199
200 const message: OrchestrationMessage = yield* decodeForEvent(
201 OrchestrationMessageSchema,
202 {
203 id: payload.id,
204 role: payload.role,
205 text: payload.text,
206 createdAt: payload.createdAt,
207 ...(payload.streaming ? {} : { completedAt: event.occurredAt }),
208 streaming: payload.streaming,
209 },
210 event.type,
211 "message",
212 );
213 const existingMessage = targetThread.messages.find((entry) => entry.id === message.id);
214 const nextMessages = existingMessage
215 ? targetThread.messages.map((entry) =>
216 entry.id === message.id
217 ? {
218 ...entry,
219 text: message.streaming
220 ? `${entry.text}${message.text}`
221 : message.text.length > 0
222 ? message.text
223 : entry.text,
224 streaming: message.streaming,
225 createdAt: entry.createdAt,
226 ...(message.streaming
227 ? { completedAt: undefined }
228 : { completedAt: message.completedAt ?? event.occurredAt }),
229 }
230 : entry,
231 )
232 : !message.streaming && message.text.length === 0
233 ? targetThread.messages
234 : [...targetThread.messages, message];
235
236 return {
237 ...nextBase,
238 threads: updateThread(nextBase.threads, payload.threadId, {
239 messages: nextMessages,
240 updatedAt: event.occurredAt,
241 }),
242 };
243 });
244 case "thread.session-set":
245 return Effect.gen(function* () {
246 const payload = yield* decodeForEvent(
247 ThreadSessionSetPayloadSchema,
248 event.payload,
249 event.type,
250 "payload",
251 );
252 const thread = nextBase.threads.find((entry) => entry.id === payload.threadId);
253 if (!thread) {
254 return nextBase;
255 }
256
257 const session: OrchestrationSession = yield* decodeForEvent(
258 OrchestrationSessionSchema,
259 {
260 ...payload.session,
261 threadId: payload.threadId,
262 },
263 event.type,
264 "session",
265 );
266
267 const sessionUpdatedAt = session.updatedAt;
268 const turnTimingPatch: ThreadPatch =
269 session.status === "running" && session.activeTurnId
270 ? (() => {
271 const isSameTurn = thread.latestTurnId === session.activeTurnId;
272 return {
273 latestTurnId: session.activeTurnId,
274 latestTurnStartedAt:
275 isSameTurn && thread.latestTurnStartedAt
276 ? thread.latestTurnStartedAt
277 : sessionUpdatedAt,
278 ...(isSameTurn
279 ? {}
280 : {
281 latestTurnCompletedAt: null,
282 latestTurnDurationMs: null,
283 }),
284 };
285 })()
286 : session.activeTurnId === null && thread.latestTurnId && thread.latestTurnStartedAt
287 ? (() => {
288 const latestTurnCompletedAt = thread.latestTurnCompletedAt ?? sessionUpdatedAt;
289 const startedAtMs = Date.parse(thread.latestTurnStartedAt);
290 const completedAtMs = Date.parse(latestTurnCompletedAt);
291 const latestTurnDurationMs =
292 !Number.isNaN(startedAtMs) &&
293 !Number.isNaN(completedAtMs) &&
294 completedAtMs >= startedAtMs
295 ? completedAtMs - startedAtMs
296 : thread.latestTurnDurationMs;
297
298 return {
299 latestTurnCompletedAt,
300 latestTurnDurationMs,
301 };
302 })()
303 : {};
304
305 return {
306 ...nextBase,
307 threads: updateThread(nextBase.threads, payload.threadId, {
308 session,
309 updatedAt: event.occurredAt,
310 error: session.lastError,
311 ...turnTimingPatch,
312 }),
313 };
314 });
315 case "thread.turn-diff-completed":
316 return Effect.gen(function* () {
317 const payload = yield* decodeForEvent(
318 ThreadTurnDiffCompletedPayloadSchema,
319 event.payload,
320 event.type,
321 "payload",
322 );
323 const thread = nextBase.threads.find((entry) => entry.id === payload.threadId);
324 if (!thread) {
325 return nextBase;
326 }
327
328 const nextSummary = yield* decodeForEvent(
329 OrchestrationTurnDiffSummarySchema,
330 {
331 turnId: payload.turnId,
332 completedAt: payload.completedAt,
333 ...(payload.status !== undefined ? { status: payload.status } : {}),
334 files: payload.files,
335 ...(payload.assistantMessageId !== undefined
336 ? { assistantMessageId: payload.assistantMessageId }
337 : {}),
338 ...(payload.checkpointTurnCount !== undefined
339 ? { checkpointTurnCount: payload.checkpointTurnCount }
340 : {}),
341 },
342 event.type,
343 "turnDiffSummary",
344 );
345 const turnDiffSummaries = [
346 ...thread.turnDiffSummaries.filter((summary) => summary.turnId !== payload.turnId),
347 nextSummary,
348 ].toSorted((left, right) => left.completedAt.localeCompare(right.completedAt));
349 const startedAtMs = thread.latestTurnStartedAt
350 ? Date.parse(thread.latestTurnStartedAt)
351 : Number.NaN;
352 const completedAtMs = Date.parse(nextSummary.completedAt);
353 const latestTurnDurationMs =
354 !Number.isNaN(startedAtMs) &&
355 !Number.isNaN(completedAtMs) &&
356 completedAtMs >= startedAtMs
357 ? completedAtMs - startedAtMs
358 : thread.latestTurnDurationMs;
359
360 return {
361 ...nextBase,
362 threads: updateThread(nextBase.threads, payload.threadId, {
363 latestTurnId: payload.turnId,
364 latestTurnCompletedAt: nextSummary.completedAt,
365 latestTurnDurationMs,
366 turnDiffSummaries,
367 updatedAt: event.occurredAt,
368 }),
369 };
370 });
371 case "thread.reverted":
372 return decodeForEvent(ThreadRevertedPayloadSchema, event.payload, event.type, "payload").pipe(
373 Effect.map((payload) => {
374 const thread = nextBase.threads.find((entry) => entry.id === payload.threadId);
375 if (!thread) {
376 return nextBase;
377 }
378
379 const targetTurnCount = Math.max(0, Math.floor(payload.turnCount));
380 const targetMessageCount = Math.max(0, Math.floor(payload.messageCount));
381 const sortedSummaries = [...thread.turnDiffSummaries].toSorted((left, right) =>
382 left.completedAt.localeCompare(right.completedAt),
383 );
384 const inferredTurnCountByTurnId = new Map<string, number>();
385 for (let index = 0; index < sortedSummaries.length; index += 1) {
386 const summary = sortedSummaries[index];
387 if (!summary) continue;
388 inferredTurnCountByTurnId.set(summary.turnId, index + 1);
389 }
390
391 const turnDiffSummaries = thread.turnDiffSummaries
392 .filter((summary) => {
393 const checkpointTurnCount =
394 summary.checkpointTurnCount ?? inferredTurnCountByTurnId.get(summary.turnId) ?? 0;
395 return checkpointTurnCount <= targetTurnCount;
396 })
397 .toSorted((left, right) => left.completedAt.localeCompare(right.completedAt));
398 const latestSummary =
399 turnDiffSummaries.length > 0 ? turnDiffSummaries[turnDiffSummaries.length - 1] : null;
400 const activities = thread.activities.filter((activity) => {
401 if (!activity.turnId) {
402 return true;
403 }
404 const checkpointTurnCount = inferredTurnCountByTurnId.get(activity.turnId);
405 if (checkpointTurnCount === undefined) {
406 return true;
407 }
408 return checkpointTurnCount <= targetTurnCount;
409 });
410
411 return {
412 ...nextBase,
413 threads: updateThread(nextBase.threads, payload.threadId, {
414 messages: thread.messages.slice(
415 0,
416 Math.min(targetMessageCount, thread.messages.length),
417 ),
418 turnDiffSummaries,
419 latestTurnId: latestSummary?.turnId ?? null,
420 latestTurnStartedAt: null,
421 latestTurnCompletedAt: latestSummary?.completedAt ?? null,
422 latestTurnDurationMs: null,
423 activities,
424 updatedAt: event.occurredAt,
425 }),
426 };
427 }),
428 );
429 case "thread.activity-appended":
430 return decodeForEvent(
431 ThreadActivityAppendedPayloadSchema,
432 event.payload,
433 event.type,
434 "payload",
435 ).pipe(
436 Effect.map((payload) => {
437 const thread = nextBase.threads.find((entry) => entry.id === payload.threadId);
438 if (!thread) {
439 return nextBase;
440 }
441 const activities = [
442 ...thread.activities.filter((activity) => activity.id !== payload.activity.id),
443 payload.activity,
444 ]
445 .toSorted((left, right) => left.createdAt.localeCompare(right.createdAt))
446 .slice(-500);
447
448 return {
449 ...nextBase,
450 threads: updateThread(nextBase.threads, payload.threadId, {
451 activities,
452 updatedAt: event.occurredAt,
453 }),
454 };
455 }),
456 );
457 case "git.read-model-upsert":
458 return decodeForEvent(
459 OrchestrationGitReadModelSchema,
460 event.payload,
461 event.type,
462 "payload",
463 ).pipe(
464 Effect.map((gitReadModel: OrchestrationGitReadModel) => ({
465 ...nextBase,
466 gitByProjectId: {
467 ...nextBase.gitByProjectId,
468 [gitReadModel.projectId]: gitReadModel,
469 },
470 })),
471 );
472 default:
473 return Effect.succeed(nextBase);
474 }
475}
apps/server/src/persistence/Errors.ts+1-1
...@@ -158,7 +158,7 @@ export type ProjectRepositoryError =...@@ -158,7 +158,7 @@ export type ProjectRepositoryError =
158 | ProjectPathMissingError158 | ProjectPathMissingError
159 | ProjectNotFoundError;159 | ProjectNotFoundError;
160160
161export type OrchestrationEventRepositoryError =161export type OrchestrationEventStoreError =
162 | PersistenceSqlError162 | PersistenceSqlError
163 | PersistenceDecodeError163 | PersistenceDecodeError
164 | PersistenceSerializationError;164 | PersistenceSerializationError;
apps/server/src/persistence/Layers/OrchestrationEventStore.test.ts created+39
...@@ -0,0 +1,39 @@
1import { Effect, Layer, Stream } from "effect";
2import { assert, it } from "@effect/vitest";
3
4import { OrchestrationEventStore } from "../Services/OrchestrationEventStore.ts";
5import { SqlitePersistenceMemory } from "./Sqlite.ts";
6import { OrchestrationEventStoreLive } from "./OrchestrationEventStore.ts";
7import { NodeServices } from "@effect/platform-node";
8
9const layer = it.layer(
10 Layer.mergeAll(
11 Layer.provide(OrchestrationEventStoreLive, SqlitePersistenceMemory),
12 NodeServices.layer,
13 ),
14);
15
16layer("OrchestrationEventStore", (it) => {
17 it.effect("stores and replays events", () =>
18 Effect.gen(function* () {
19 const createdAt = new Date().toISOString();
20
21 const eventStore = yield* OrchestrationEventStore;
22 const saved = yield* eventStore.append({
23 eventId: "event-1",
24 type: "thread.created",
25 aggregateType: "thread",
26 aggregateId: "thread-1",
27 occurredAt: createdAt,
28 commandId: "cmd-1",
29 payload: { id: "thread-1", projectId: "project-1", title: "demo" },
30 });
31 assert.equal(saved.sequence, 1);
32
33 const replayed = yield* Stream.runCollect(eventStore.readFromSequence(0)).pipe(
34 Effect.map(Array.from),
35 );
36 assert.deepEqual(replayed, [saved]);
37 }),
38 );
39});
apps/server/src/persistence/Layers/OrchestrationEventStore.ts created+210
...@@ -0,0 +1,210 @@
1import type { OrchestrationEvent } from "@t3tools/contracts";
2import { OrchestrationEventSchema } from "@t3tools/contracts";
3import * as SqlClient from "effect/unstable/sql/SqlClient";
4import * as SqlSchema from "effect/unstable/sql/SqlSchema";
5import { Effect, Layer, Schema, Stream } from "effect";
6
7import {
8 toPersistenceDecodeCauseError,
9 toPersistenceDecodeError,
10 toPersistenceSerializationError,
11 toPersistenceSqlError,
12 type OrchestrationEventStoreError,
13} from "../Errors.ts";
14import {
15 OrchestrationEventStore,
16 type OrchestrationEventStoreShape,
17} from "../Services/OrchestrationEventStore.ts";
18
19const decodeEvent = Schema.decodeUnknownEffect(OrchestrationEventSchema);
20
21const EventRowSchema = Schema.Struct({
22 sequence: Schema.Number,
23 eventId: Schema.String,
24 type: Schema.String,
25 aggregateType: Schema.String,
26 aggregateId: Schema.String,
27 occurredAt: Schema.String,
28 commandId: Schema.NullOr(Schema.String),
29 payloadJson: Schema.String,
30});
31
32const AppendEventRequestSchema = Schema.Struct({
33 eventId: Schema.String,
34 type: Schema.String,
35 aggregateType: Schema.String,
36 aggregateId: Schema.String,
37 occurredAt: Schema.String,
38 commandId: Schema.NullOr(Schema.String),
39 payloadJson: Schema.String,
40});
41
42const ReadFromSequenceRequestSchema = Schema.Struct({
43 sequenceExclusive: Schema.Number,
44 limit: Schema.Number,
45});
46const DEFAULT_READ_FROM_SEQUENCE_LIMIT = 1_000;
47const READ_PAGE_SIZE = 500;
48
49function eventRowToOrchestrationEvent(
50 row: Schema.Schema.Type<typeof EventRowSchema>,
51 operation: string,
52): Effect.Effect<OrchestrationEvent, OrchestrationEventStoreError> {
53 return Effect.try({
54 try: () => JSON.parse(row.payloadJson) as unknown,
55 catch: toPersistenceDecodeCauseError(`${operation}:parsePayloadJson`),
56 }).pipe(
57 Effect.flatMap((payload) =>
58 decodeEvent({
59 sequence: row.sequence,
60 eventId: row.eventId,
61 type: row.type,
62 aggregateType: row.aggregateType,
63 aggregateId: row.aggregateId,
64 occurredAt: row.occurredAt,
65 commandId: row.commandId,
66 payload,
67 }).pipe(Effect.mapError(toPersistenceDecodeError(`${operation}:decodeEvent`))),
68 ),
69 );
70}
71
72const makeEventStore = Effect.gen(function* () {
73 const sql = yield* SqlClient.SqlClient;
74
75 const appendEventRow = SqlSchema.findOne({
76 Request: AppendEventRequestSchema,
77 Result: EventRowSchema,
78 execute: (request) =>
79 sql`
80 INSERT INTO orchestration_events (
81 event_id,
82 event_type,
83 aggregate_type,
84 aggregate_id,
85 occurred_at,
86 command_id,
87 payload_json
88 )
89 VALUES (
90 ${request.eventId},
91 ${request.type},
92 ${request.aggregateType},
93 ${request.aggregateId},
94 ${request.occurredAt},
95 ${request.commandId},
96 ${request.payloadJson}
97 )
98 RETURNING
99 sequence,
100 event_id AS "eventId",
101 event_type AS "type",
102 aggregate_type AS "aggregateType",
103 aggregate_id AS "aggregateId",
104 occurred_at AS "occurredAt",
105 command_id AS "commandId",
106 payload_json AS "payloadJson"
107 `,
108 });
109
110 const readEventRowsFromSequence = SqlSchema.findAll({
111 Request: ReadFromSequenceRequestSchema,
112 Result: EventRowSchema,
113 execute: (request) =>
114 sql`
115 SELECT
116 sequence,
117 event_id AS "eventId",
118 event_type AS "type",
119 aggregate_type AS "aggregateType",
120 aggregate_id AS "aggregateId",
121 occurred_at AS "occurredAt",
122 command_id AS "commandId",
123 payload_json AS "payloadJson"
124 FROM orchestration_events
125 WHERE sequence > ${request.sequenceExclusive}
126 ORDER BY sequence ASC
127 LIMIT ${request.limit}
128 `,
129 });
130
131 const append: OrchestrationEventStoreShape["append"] = (event) =>
132 Effect.try({
133 try: () => JSON.stringify(event.payload),
134 catch: toPersistenceSerializationError("OrchestrationEventStore.append:encodePayload"),
135 }).pipe(
136 Effect.flatMap((payloadJson) =>
137 appendEventRow({
138 eventId: event.eventId,
139 type: event.type,
140 aggregateType: event.aggregateType,
141 aggregateId: event.aggregateId,
142 occurredAt: event.occurredAt,
143 commandId: event.commandId,
144 payloadJson,
145 }).pipe(
146 Effect.mapError(toPersistenceSqlError("OrchestrationEventStore.append:insert")),
147 Effect.flatMap((row) =>
148 eventRowToOrchestrationEvent(row, "OrchestrationEventStore.append:rowToEvent"),
149 ),
150 ),
151 ),
152 );
153
154 const readFromSequence: OrchestrationEventStoreShape["readFromSequence"] = (
155 sequenceExclusive,
156 limit = DEFAULT_READ_FROM_SEQUENCE_LIMIT,
157 ) => {
158 const normalizedLimit = Math.max(0, Math.floor(limit));
159 if (normalizedLimit === 0) {
160 return Stream.empty;
161 }
162 const readPage = (cursor: number, remaining: number): Stream.Stream<
163 OrchestrationEvent,
164 OrchestrationEventStoreError
165 > =>
166 Stream.fromEffect(
167 readEventRowsFromSequence({
168 sequenceExclusive: cursor,
169 limit: Math.min(remaining, READ_PAGE_SIZE),
170 }).pipe(
171 Effect.mapError(toPersistenceSqlError("OrchestrationEventStore.readFromSequence:query")),
172 Effect.flatMap((rows) =>
173 Effect.forEach(rows, (row) =>
174 eventRowToOrchestrationEvent(
175 row,
176 "OrchestrationEventStore.readFromSequence:rowToEvent",
177 ),
178 ),
179 ),
180 ),
181 ).pipe(
182 Stream.flatMap((events) => {
183 if (events.length === 0) {
184 return Stream.empty;
185 }
186 const nextRemaining = remaining - events.length;
187 if (nextRemaining <= 0) {
188 return Stream.fromIterable(events);
189 }
190 return Stream.concat(
191 Stream.fromIterable(events),
192 readPage(events[events.length - 1]!.sequence, nextRemaining),
193 );
194 }),
195 );
196
197 return readPage(sequenceExclusive, normalizedLimit);
198 };
199
200 return {
201 append,
202 readFromSequence,
203 readAll: () => readFromSequence(0, Number.MAX_SAFE_INTEGER),
204 } satisfies OrchestrationEventStoreShape;
205});
206
207export const OrchestrationEventStoreLive = Layer.effect(
208 OrchestrationEventStore,
209 makeEventStore,
210);
apps/server/src/persistence/Layers/OrchestrationEvents.test.ts deleted-37
...@@ -1,37 +0,0 @@
1import { Effect, Layer } from "effect";
2import { assert, it } from "@effect/vitest";
3
4import { OrchestrationEventRepository } from "../Services/OrchestrationEvents.ts";
5import { SqlitePersistenceMemory } from "./Sqlite.ts";
6import { OrchestrationEventRepositoryLive } from "./OrchestrationEvents.ts";
7import { NodeServices } from "@effect/platform-node";
8
9const layer = it.layer(
10 Layer.mergeAll(
11 Layer.provide(OrchestrationEventRepositoryLive, SqlitePersistenceMemory),
12 NodeServices.layer,
13 ),
14);
15
16layer("OrchestrationEventRepository", (it) => {
17 it.effect("stores and replays events", () =>
18 Effect.gen(function* () {
19 const createdAt = new Date().toISOString();
20
21 const repository = yield* OrchestrationEventRepository;
22 const saved = yield* repository.append({
23 eventId: "event-1",
24 type: "thread.created",
25 aggregateType: "thread",
26 aggregateId: "thread-1",
27 occurredAt: createdAt,
28 commandId: "cmd-1",
29 payload: { id: "thread-1", projectId: "project-1", title: "demo" },
30 });
31 assert.equal(saved.sequence, 1);
32
33 const replayed = yield* repository.readFromSequence(0);
34 assert.deepEqual(replayed, [saved]);
35 }),
36 );
37});
apps/server/src/persistence/Layers/OrchestrationEvents.ts deleted-178
...@@ -1,178 +0,0 @@
1import type { OrchestrationEvent } from "@t3tools/contracts";
2import { OrchestrationEventSchema } from "@t3tools/contracts";
3import * as SqlClient from "effect/unstable/sql/SqlClient";
4import * as SqlSchema from "effect/unstable/sql/SqlSchema";
5import { Effect, Layer, Schema } from "effect";
6
7import {
8 toPersistenceDecodeCauseError,
9 toPersistenceDecodeError,
10 toPersistenceSerializationError,
11 toPersistenceSqlError,
12 type OrchestrationEventRepositoryError,
13} from "../Errors.ts";
14import {
15 OrchestrationEventRepository,
16 type OrchestrationEventRepositoryShape,
17} from "../Services/OrchestrationEvents.ts";
18
19const decodeEvent = Schema.decodeUnknownEffect(OrchestrationEventSchema);
20
21const EventRowSchema = Schema.Struct({
22 sequence: Schema.Number,
23 eventId: Schema.String,
24 type: Schema.String,
25 aggregateType: Schema.String,
26 aggregateId: Schema.String,
27 occurredAt: Schema.String,
28 commandId: Schema.NullOr(Schema.String),
29 payloadJson: Schema.String,
30});
31
32const AppendEventRequestSchema = Schema.Struct({
33 eventId: Schema.String,
34 type: Schema.String,
35 aggregateType: Schema.String,
36 aggregateId: Schema.String,
37 occurredAt: Schema.String,
38 commandId: Schema.NullOr(Schema.String),
39 payloadJson: Schema.String,
40});
41
42const ReadFromSequenceRequestSchema = Schema.Struct({
43 sequenceExclusive: Schema.Number,
44 limit: Schema.Number,
45});
46
47function eventRowToOrchestrationEvent(
48 row: Schema.Schema.Type<typeof EventRowSchema>,
49 operation: string,
50): Effect.Effect<OrchestrationEvent, OrchestrationEventRepositoryError> {
51 return Effect.try({
52 try: () => JSON.parse(row.payloadJson) as unknown,
53 catch: toPersistenceDecodeCauseError(`${operation}:parsePayloadJson`),
54 }).pipe(
55 Effect.flatMap((payload) =>
56 decodeEvent({
57 sequence: row.sequence,
58 eventId: row.eventId,
59 type: row.type,
60 aggregateType: row.aggregateType,
61 aggregateId: row.aggregateId,
62 occurredAt: row.occurredAt,
63 commandId: row.commandId,
64 payload,
65 }).pipe(Effect.mapError(toPersistenceDecodeError(`${operation}:decodeEvent`))),
66 ),
67 );
68}
69
70const makeRepository = Effect.gen(function* () {
71 const sql = yield* SqlClient.SqlClient;
72
73 const appendEventRow = SqlSchema.findOne({
74 Request: AppendEventRequestSchema,
75 Result: EventRowSchema,
76 execute: (request) =>
77 sql`
78 INSERT INTO orchestration_events (
79 event_id,
80 event_type,
81 aggregate_type,
82 aggregate_id,
83 occurred_at,
84 command_id,
85 payload_json
86 )
87 VALUES (
88 ${request.eventId},
89 ${request.type},
90 ${request.aggregateType},
91 ${request.aggregateId},
92 ${request.occurredAt},
93 ${request.commandId},
94 ${request.payloadJson}
95 )
96 RETURNING
97 sequence,
98 event_id AS "eventId",
99 event_type AS "type",
100 aggregate_type AS "aggregateType",
101 aggregate_id AS "aggregateId",
102 occurred_at AS "occurredAt",
103 command_id AS "commandId",
104 payload_json AS "payloadJson"
105 `,
106 });
107
108 const readEventRowsFromSequence = SqlSchema.findAll({
109 Request: ReadFromSequenceRequestSchema,
110 Result: EventRowSchema,
111 execute: (request) =>
112 sql`
113 SELECT
114 sequence,
115 event_id AS "eventId",
116 event_type AS "type",
117 aggregate_type AS "aggregateType",
118 aggregate_id AS "aggregateId",
119 occurred_at AS "occurredAt",
120 command_id AS "commandId",
121 payload_json AS "payloadJson"
122 FROM orchestration_events
123 WHERE sequence > ${request.sequenceExclusive}
124 ORDER BY sequence ASC
125 LIMIT ${request.limit}
126 `,
127 });
128
129 const append: OrchestrationEventRepositoryShape["append"] = (event) =>
130 Effect.try({
131 try: () => JSON.stringify(event.payload),
132 catch: toPersistenceSerializationError("OrchestrationEventRepository.append:encodePayload"),
133 }).pipe(
134 Effect.flatMap((payloadJson) =>
135 appendEventRow({
136 eventId: event.eventId,
137 type: event.type,
138 aggregateType: event.aggregateType,
139 aggregateId: event.aggregateId,
140 occurredAt: event.occurredAt,
141 commandId: event.commandId,
142 payloadJson,
143 }).pipe(
144 Effect.mapError(toPersistenceSqlError("OrchestrationEventRepository.append:insert")),
145 Effect.flatMap((row) =>
146 eventRowToOrchestrationEvent(row, "OrchestrationEventRepository.append:rowToEvent"),
147 ),
148 ),
149 ),
150 );
151
152 const readFromSequence: OrchestrationEventRepositoryShape["readFromSequence"] = (
153 sequenceExclusive,
154 limit = 1_000,
155 ) =>
156 readEventRowsFromSequence({ sequenceExclusive, limit }).pipe(
157 Effect.mapError(toPersistenceSqlError("OrchestrationEventRepository.readFromSequence:query")),
158 Effect.flatMap((rows) =>
159 Effect.forEach(rows, (row) =>
160 eventRowToOrchestrationEvent(
161 row,
162 "OrchestrationEventRepository.readFromSequence:rowToEvent",
163 ),
164 ),
165 ),
166 );
167
168 return {
169 append,
170 readFromSequence,
171 readAll: () => readFromSequence(0, Number.MAX_SAFE_INTEGER),
172 } satisfies OrchestrationEventRepositoryShape;
173});
174
175export const OrchestrationEventRepositoryLive = Layer.effect(
176 OrchestrationEventRepository,
177 makeRepository,
178);
apps/server/src/persistence/Services/OrchestrationEventStore.ts created+63
...@@ -0,0 +1,63 @@
1/**
2 * OrchestrationEventStore - Event store interface for orchestration events.
3 *
4 * Owns durable append/replay access to the orchestration event stream. It does
5 * not reduce events into read models or apply command validation rules.
6 *
7 * Uses Effect `ServiceMap.Service` for dependency injection and exposes typed
8 * persistence/decode errors for event append and replay operations.
9 *
10 * @module OrchestrationEventStore
11 */
12import type { OrchestrationEvent } from "@t3tools/contracts";
13import { ServiceMap } from "effect";
14import type { Effect, Stream } from "effect";
15
16import type { OrchestrationEventStoreError } from "../Errors.ts";
17
18export interface OrchestrationEventStoreShape {
19 /**
20 * Persist a new orchestration event.
21 *
22 * @param event - Event payload without sequence (assigned by storage).
23 * @returns Effect containing the stored event with assigned sequence.
24 */
25 readonly append: (
26 event: Omit<OrchestrationEvent, "sequence">,
27 ) => Effect.Effect<OrchestrationEvent, OrchestrationEventStoreError>;
28
29 /**
30 * Replay events after the provided sequence.
31 *
32 * @param sequenceExclusive - Sequence cursor (exclusive).
33 * @param limit - Maximum number of events to emit.
34 * @returns Stream containing ordered events.
35 */
36 readonly readFromSequence: (
37 sequenceExclusive: number,
38 limit?: number,
39 ) => Stream.Stream<OrchestrationEvent, OrchestrationEventStoreError>;
40
41 /**
42 * Read all events from the beginning of the stream.
43 *
44 * @returns Stream containing all stored events.
45 */
46 readonly readAll: () => Stream.Stream<OrchestrationEvent, OrchestrationEventStoreError>;
47}
48
49/**
50 * OrchestrationEventStore - Service tag for orchestration event persistence.
51 *
52 * @example
53 * ```ts
54 * const program = Effect.gen(function* () {
55 * const events = yield* OrchestrationEventStore
56 * return yield* Stream.runCollect(events.readAll())
57 * })
58 * ```
59 */
60export class OrchestrationEventStore extends ServiceMap.Service<
61 OrchestrationEventStore,
62 OrchestrationEventStoreShape
63>()("orchestration/EventStore") {}
apps/server/src/persistence/Services/OrchestrationEvents.ts deleted-63
...@@ -1,63 +0,0 @@
1/**
2 * OrchestrationEventRepository - Repository interface for orchestration events.
3 *
4 * Owns durable append/replay access to the orchestration event stream. It does
5 * not reduce events into read models or apply command validation rules.
6 *
7 * Uses Effect `ServiceMap.Service` for dependency injection and exposes typed
8 * persistence/decode errors for event append and replay operations.
9 *
10 * @module OrchestrationEventRepository
11 */
12import type { OrchestrationEvent } from "@t3tools/contracts";
13import { ServiceMap } from "effect";
14import type { Effect } from "effect";
15
16import type { OrchestrationEventRepositoryError } from "../Errors.ts";
17
18export interface OrchestrationEventRepositoryShape {
19 /**
20 * Persist a new orchestration event.
21 *
22 * @param event - Event payload without sequence (assigned by storage).
23 * @returns Effect containing the stored event with assigned sequence.
24 */
25 readonly append: (
26 event: Omit<OrchestrationEvent, "sequence">,
27 ) => Effect.Effect<OrchestrationEvent, OrchestrationEventRepositoryError>;
28
29 /**
30 * Replay events after the provided sequence.
31 *
32 * @param sequenceExclusive - Sequence cursor (exclusive).
33 * @param limit - Maximum number of events to return.
34 * @returns Effect containing ordered events.
35 */
36 readonly readFromSequence: (
37 sequenceExclusive: number,
38 limit?: number,
39 ) => Effect.Effect<OrchestrationEvent[], OrchestrationEventRepositoryError>;
40
41 /**
42 * Read all events from the beginning of the stream.
43 *
44 * @returns Effect containing all stored events.
45 */
46 readonly readAll: () => Effect.Effect<OrchestrationEvent[], OrchestrationEventRepositoryError>;
47}
48
49/**
50 * OrchestrationEventRepository - Service tag for orchestration event persistence.
51 *
52 * @example
53 * ```ts
54 * const program = Effect.gen(function* () {
55 * const events = yield* OrchestrationEventRepository
56 * return yield* events.readAll()
57 * })
58 * ```
59 */
60export class OrchestrationEventRepository extends ServiceMap.Service<
61 OrchestrationEventRepository,
62 OrchestrationEventRepositoryShape
63>()("orchestration/EventRepository") {}
apps/server/src/wsServer.ts+23-18
...@@ -36,13 +36,16 @@ import {...@@ -36,13 +36,16 @@ import {
36import { TerminalManager } from "./terminalManager";36import { TerminalManager } from "./terminalManager";
37import { loadResolvedKeybindingsConfig, upsertKeybindingRule } from "./keybindings";37import { loadResolvedKeybindingsConfig, upsertKeybindingRule } from "./keybindings";
38import { searchWorkspaceEntries } from "./workspaceEntries";38import { searchWorkspaceEntries } from "./workspaceEntries";
39import { OrchestrationEngineLive } from "./orchestration/Layer";39import { OrchestrationEngineLive } from "./orchestration/Layers/OrchestrationEngine";
40import { OrchestrationEngineService, type OrchestrationEngineShape } from "./orchestration/Service";40import {
41 OrchestrationEngineService,
42 type OrchestrationEngineShape,
43} from "./orchestration/Services/OrchestrationEngine";
41import { ProjectRepositoryLive } from "./persistence/Layers/Projects";44import { ProjectRepositoryLive } from "./persistence/Layers/Projects";
42import { makeSqlitePersistenceLive } from "./persistence/Layers/Sqlite";45import { makeSqlitePersistenceLive } from "./persistence/Layers/Sqlite";
43import { ProjectRepository, type ProjectRepositoryShape } from "./persistence/Services/Projects";46import { ProjectRepository, type ProjectRepositoryShape } from "./persistence/Services/Projects";
44import assert from "node:assert";47import assert from "node:assert";
45import { OrchestrationEventRepositoryLive } from "./persistence/Layers/OrchestrationEvents";48import { OrchestrationEventStoreLive } from "./persistence/Layers/OrchestrationEventStore";
46import { makeCodexAdapterLive } from "./provider/Layers/CodexAdapter";49import { makeCodexAdapterLive } from "./provider/Layers/CodexAdapter";
47import { ProviderAdapterRegistryLive } from "./provider/Layers/ProviderAdapterRegistry";50import { ProviderAdapterRegistryLive } from "./provider/Layers/ProviderAdapterRegistry";
48import { makeProviderServiceLive } from "./provider/Layers/ProviderService";51import { makeProviderServiceLive } from "./provider/Layers/ProviderService";
...@@ -358,8 +361,8 @@ export function createServer(options: ServerOptions) {...@@ -358,8 +361,8 @@ export function createServer(options: ServerOptions) {
358361
359 const processEvent = (event: ProviderRuntimeEvent) =>362 const processEvent = (event: ProviderRuntimeEvent) =>
360 Effect.gen(function* () {363 Effect.gen(function* () {
361 const snapshot = yield* orchestrationEngine.getSnapshot();364 const readModel = yield* orchestrationEngine.getReadModel();
362 const thread = snapshot.threads.find(365 const thread = readModel.threads.find(
363 (entry) => entry.session?.sessionId === event.sessionId,366 (entry) => entry.session?.sessionId === event.sessionId,
364 );367 );
365 if (!thread) return;368 if (!thread) return;
...@@ -735,8 +738,8 @@ export function createServer(options: ServerOptions) {...@@ -735,8 +738,8 @@ export function createServer(options: ServerOptions) {
735 const params = request.params as { sessionId?: string };738 const params = request.params as { sessionId?: string };
736 const sessionId = params.sessionId;739 const sessionId = params.sessionId;
737 if (typeof sessionId === "string") {740 if (typeof sessionId === "string") {
738 const snapshot = await effectRuntime.runPromise(orchestrationEngine.getSnapshot());741 const readModel = await effectRuntime.runPromise(orchestrationEngine.getReadModel());
739 const thread = snapshot.threads.find((entry) => entry.session?.sessionId === sessionId);742 const thread = readModel.threads.find((entry) => entry.session?.sessionId === sessionId);
740 if (thread) {743 if (thread) {
741 const now = new Date().toISOString();744 const now = new Date().toISOString();
742 await effectRuntime.runPromise(745 await effectRuntime.runPromise(
...@@ -909,7 +912,7 @@ export function createServer(options: ServerOptions) {...@@ -909,7 +912,7 @@ export function createServer(options: ServerOptions) {
909 };912 };
910913
911 case ORCHESTRATION_WS_METHODS.getSnapshot:914 case ORCHESTRATION_WS_METHODS.getSnapshot:
912 return effectRuntime.runPromise(orchestrationEngine.getSnapshot());915 return effectRuntime.runPromise(orchestrationEngine.getReadModel());
913916
914 case ORCHESTRATION_WS_METHODS.dispatchCommand:917 case ORCHESTRATION_WS_METHODS.dispatchCommand:
915 orchestrationCommandLogger?.write({918 orchestrationCommandLogger?.write({
...@@ -917,21 +920,23 @@ export function createServer(options: ServerOptions) {...@@ -917,21 +920,23 @@ export function createServer(options: ServerOptions) {
917 requestId: request.id,920 requestId: request.id,
918 command: request.params,921 command: request.params,
919 });922 });
920 return effectRuntime.runPromise(orchestrationEngine.dispatchUnknown(request.params));923 return effectRuntime.runPromise(orchestrationEngine.dispatchUnknownCommand(request.params));
921924
922 case ORCHESTRATION_WS_METHODS.replayEvents:925 case ORCHESTRATION_WS_METHODS.replayEvents:
923 return effectRuntime.runPromise(926 return effectRuntime.runPromise(
924 orchestrationEngine.replayEvents(927 Stream.runCollect(
925 Math.max(928 orchestrationEngine.readEvents(
926 0,929 Math.max(
927 Math.floor(930 0,
928 Number(931 Math.floor(
929 (request.params as { fromSequenceExclusive?: number } | undefined)932 Number(
930 ?.fromSequenceExclusive ?? 0,933 (request.params as { fromSequenceExclusive?: number } | undefined)
934 ?.fromSequenceExclusive ?? 0,
935 ),
931 ),936 ),
932 ),937 ),
933 ),938 ),
934 ),939 ).pipe(Effect.map((events) => Array.from(events))),
935 );940 );
936941
937 default:942 default:
...@@ -952,7 +957,7 @@ export function createServer(options: ServerOptions) {...@@ -952,7 +957,7 @@ export function createServer(options: ServerOptions) {
952 const persistenceLayer = customPersistenceLayer ?? makeSqlitePersistenceLive(dbPath);957 const persistenceLayer = customPersistenceLayer ?? makeSqlitePersistenceLive(dbPath);
953 const orchestrationLayer = Layer.provide(958 const orchestrationLayer = Layer.provide(
954 OrchestrationEngineLive,959 OrchestrationEngineLive,
955 OrchestrationEventRepositoryLive,960 OrchestrationEventStoreLive,
956 );961 );
957 const codexAdapterLayer = makeCodexAdapterLive({962 const codexAdapterLayer = makeCodexAdapterLive({
958 nativeEventLogPath: path.join(providerLogsDir, "provider-native.ndjson"),963 nativeEventLogPath: path.join(providerLogsDir, "provider-native.ndjson"),