authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-02-22 01:27:18-08:00
committergravatar for jmarminge@gmail.comJulius Marminge <jmarminge@gmail.com> 2026-02-26 13:20:36-08:00
loge219efa22a4a8a6cc1424af54b04c58be9a8b68d
treedf54987927a3bd242b63e9890e740ca909b1a307
parent9c070efface28ef266dc9736f9c1df4e9d61066d

simplify layer composition

8 files changed, 959 insertions(+), 890 deletions(-)

apps/server/src/orchestration/Layer.test.ts created+439
......@@ -0,0 +1,439 @@
1import fs from "node:fs";
2import os from "node:os";
3import path from "node:path";
4
5import type { OrchestrationEvent } from "@t3tools/contracts";
6import { Effect } from "effect";
7import { afterEach, describe, expect, it } from "vitest";
8
9import {
10 OrchestrationEventRepository,
11 type OrchestrationEventRepositoryShape,
12} from "../persistence/Services/OrchestrationEvents";
13
14import { Layer, ManagedRuntime } from "effect";
15
16import { makeSqlitePersistenceLive } from "../persistence/Layers/Sqlite";
17
18import { OrchestrationEngineLive } from "./Layer";
19import { OrchestrationEngineService } from "./Service";
20import { OrchestrationEventRepositoryLive } from "../persistence/Layers/OrchestrationEvents";
21
22const tempDirs: string[] = [];
23
24function makeTempDir(prefix: string): string {
25 const dir = fs.mkdtempSync(path.join(os.tmpdir(), prefix));
26 tempDirs.push(dir);
27 return dir;
28}
29
30afterEach(() => {
31 for (const dir of tempDirs.splice(0, tempDirs.length)) {
32 fs.rmSync(dir, { recursive: true, force: true });
33 }
34});
35
36export async function createOrchestrationSystem(stateDir: string) {
37 const dbPath = path.join(stateDir, "orchestration.sqlite");
38 const orchestrationLayer = OrchestrationEngineLive.pipe(
39 Layer.provide(OrchestrationEventRepositoryLive),
40 Layer.provide(makeSqlitePersistenceLive(dbPath)),
41 );
42 const runtime = ManagedRuntime.make(orchestrationLayer);
43 const engine = await runtime.runPromise(OrchestrationEngineService);
44 return {
45 engine,
46 run: <A, E>(effect: Effect.Effect<A, E>) => runtime.runPromise(effect),
47 dispose: () => runtime.dispose(),
48 };
49}
50
51describe("OrchestrationEngine", () => {
52 it("replays to the same deterministic snapshot", async () => {
53 const stateDir = makeTempDir("t3code-orchestration-");
54 const createdAt = new Date().toISOString();
55 const projectId = "project-1";
56 const threadId = "thread-1";
57
58 const firstSystem = await createOrchestrationSystem(stateDir);
59 const engineA = firstSystem.engine;
60 await firstSystem.run(
61 engineA.dispatch({
62 type: "thread.create",
63 commandId: "cmd-1",
64 threadId,
65 projectId,
66 title: "Thread",
67 model: "gpt-5-codex",
68 branch: null,
69 worktreePath: null,
70 createdAt,
71 }),
72 );
73 await firstSystem.run(
74 engineA.dispatch({
75 type: "message.send",
76 commandId: "cmd-3",
77 threadId,
78 messageId: "msg-1",
79 role: "user",
80 text: "hello",
81 streaming: false,
82 createdAt,
83 }),
84 );
85 const snapshotA = await firstSystem.run(engineA.getSnapshot());
86 await firstSystem.dispose();
87
88 const secondSystem = await createOrchestrationSystem(stateDir);
89 const engineB = secondSystem.engine;
90 const snapshotB = await secondSystem.run(engineB.getSnapshot());
91 expect(snapshotB).toEqual(snapshotA);
92 await secondSystem.dispose();
93 });
94
95 it("fans out read-model updates to subscribers", async () => {
96 const stateDir = makeTempDir("t3code-orchestration-fanout-");
97 const system = await createOrchestrationSystem(stateDir);
98 const engine = system.engine;
99 const updates: number[] = [];
100 const unsubscribe = await system.run(
101 engine.subscribeToReadModel((snapshot) => {
102 updates.push(snapshot.sequence);
103 }),
104 );
105 await system.run(
106 engine.dispatch({
107 type: "thread.create",
108 commandId: "cmd-thread",
109 threadId: "thread-2",
110 projectId: "project-2",
111 title: "fanout",
112 model: "gpt-5-codex",
113 branch: null,
114 worktreePath: null,
115 createdAt: new Date().toISOString(),
116 }),
117 );
118 unsubscribe();
119 expect(updates.length).toBeGreaterThan(0);
120 await system.dispose();
121 });
122
123 it("replays append-only events from sequence", async () => {
124 const stateDir = makeTempDir("t3code-orchestration-replay-");
125 const system = await createOrchestrationSystem(stateDir);
126 const engine = system.engine;
127 await system.run(
128 engine.dispatch({
129 type: "thread.create",
130 commandId: "cmd-a",
131 threadId: "thread-replay",
132 projectId: "project-replay",
133 title: "replay",
134 model: "gpt-5-codex",
135 branch: null,
136 worktreePath: null,
137 createdAt: new Date().toISOString(),
138 }),
139 );
140 await system.run(
141 engine.dispatch({
142 type: "thread.delete",
143 commandId: "cmd-b",
144 threadId: "thread-replay",
145 createdAt: new Date().toISOString(),
146 }),
147 );
148 const events = await system.run(engine.replayEvents(0));
149 expect(events.length).toBe(2);
150 expect(events[0]?.type).toBe("thread.created");
151 expect(events[1]?.type).toBe("thread.deleted");
152 await system.dispose();
153 });
154
155 it("stores completed turn summaries even when no files changed", async () => {
156 const stateDir = makeTempDir("t3code-orchestration-turn-diff-");
157 const firstSystem = await createOrchestrationSystem(stateDir);
158 const engine = firstSystem.engine;
159 const completedAt = new Date().toISOString();
160 await firstSystem.run(
161 engine.dispatch({
162 type: "thread.create",
163 commandId: "cmd-thread-turn-diff",
164 threadId: "thread-turn-diff",
165 projectId: "project-turn-diff",
166 title: "Turn diff thread",
167 model: "gpt-5-codex",
168 branch: null,
169 worktreePath: null,
170 createdAt: completedAt,
171 }),
172 );
173 await firstSystem.run(
174 engine.dispatch({
175 type: "thread.turnDiff.complete",
176 commandId: "cmd-turn-diff-complete",
177 threadId: "thread-turn-diff",
178 turnId: "turn-1",
179 completedAt,
180 status: "completed",
181 files: [],
182 checkpointTurnCount: 1,
183 createdAt: completedAt,
184 }),
185 );
186
187 const firstThread = (await firstSystem.run(engine.getSnapshot())).threads.find(
188 (thread) => thread.id === "thread-turn-diff",
189 );
190 expect(firstThread?.turnDiffSummaries).toEqual([
191 {
192 turnId: "turn-1",
193 completedAt,
194 status: "completed",
195 files: [],
196 checkpointTurnCount: 1,
197 },
198 ]);
199 await firstSystem.dispose();
200
201 const secondSystem = await createOrchestrationSystem(stateDir);
202 const restartedEngine = secondSystem.engine;
203 const restartedThread = (await secondSystem.run(restartedEngine.getSnapshot())).threads.find(
204 (thread) => thread.id === "thread-turn-diff",
205 );
206 expect(restartedThread?.turnDiffSummaries).toEqual([
207 {
208 turnId: "turn-1",
209 completedAt,
210 status: "completed",
211 files: [],
212 checkpointTurnCount: 1,
213 },
214 ]);
215 await secondSystem.dispose();
216 });
217
218 it("reverts thread messages and turn summaries to a checkpoint", async () => {
219 const stateDir = makeTempDir("t3code-orchestration-revert-");
220 const firstSystem = await createOrchestrationSystem(stateDir);
221 const engine = firstSystem.engine;
222 const createdAt = new Date().toISOString();
223 await firstSystem.run(
224 engine.dispatch({
225 type: "thread.create",
226 commandId: "cmd-thread-revert",
227 threadId: "thread-revert",
228 projectId: "project-revert",
229 title: "Revert thread",
230 model: "gpt-5-codex",
231 branch: null,
232 worktreePath: null,
233 createdAt,
234 }),
235 );
236 await firstSystem.run(
237 engine.dispatch({
238 type: "message.send",
239 commandId: "cmd-msg-1-user",
240 threadId: "thread-revert",
241 messageId: "user-1",
242 role: "user",
243 text: "first",
244 streaming: false,
245 createdAt,
246 }),
247 );
248 await firstSystem.run(
249 engine.dispatch({
250 type: "message.send",
251 commandId: "cmd-msg-1-assistant",
252 threadId: "thread-revert",
253 messageId: "assistant:turn-1",
254 role: "assistant",
255 text: "first-response",
256 streaming: false,
257 createdAt,
258 }),
259 );
260 await firstSystem.run(
261 engine.dispatch({
262 type: "thread.turnDiff.complete",
263 commandId: "cmd-turn-1-complete",
264 threadId: "thread-revert",
265 turnId: "turn-1",
266 completedAt: createdAt,
267 status: "completed",
268 files: [],
269 assistantMessageId: "assistant:turn-1",
270 checkpointTurnCount: 1,
271 createdAt,
272 }),
273 );
274
275 const createdAtSecond = new Date(Date.now() + 1_000).toISOString();
276 await firstSystem.run(
277 engine.dispatch({
278 type: "message.send",
279 commandId: "cmd-msg-2-user",
280 threadId: "thread-revert",
281 messageId: "user-2",
282 role: "user",
283 text: "second",
284 streaming: false,
285 createdAt: createdAtSecond,
286 }),
287 );
288 await firstSystem.run(
289 engine.dispatch({
290 type: "message.send",
291 commandId: "cmd-msg-2-assistant",
292 threadId: "thread-revert",
293 messageId: "assistant:turn-2",
294 role: "assistant",
295 text: "second-response",
296 streaming: false,
297 createdAt: createdAtSecond,
298 }),
299 );
300 await firstSystem.run(
301 engine.dispatch({
302 type: "thread.turnDiff.complete",
303 commandId: "cmd-turn-2-complete",
304 threadId: "thread-revert",
305 turnId: "turn-2",
306 completedAt: createdAtSecond,
307 status: "completed",
308 files: [],
309 assistantMessageId: "assistant:turn-2",
310 checkpointTurnCount: 2,
311 createdAt: createdAtSecond,
312 }),
313 );
314
315 await firstSystem.run(
316 engine.dispatch({
317 type: "thread.revert",
318 commandId: "cmd-thread-revert-apply",
319 threadId: "thread-revert",
320 turnCount: 1,
321 messageCount: 2,
322 createdAt: createdAtSecond,
323 }),
324 );
325
326 const thread = (await firstSystem.run(engine.getSnapshot())).threads.find(
327 (entry) => entry.id === "thread-revert",
328 );
329 expect(thread?.messages.map((message) => message.id)).toEqual(["user-1", "assistant:turn-1"]);
330 expect(thread?.turnDiffSummaries.map((summary) => summary.turnId)).toEqual(["turn-1"]);
331 expect(thread?.latestTurnId).toBe("turn-1");
332 await firstSystem.dispose();
333
334 const secondSystem = await createOrchestrationSystem(stateDir);
335 const restarted = secondSystem.engine;
336 const restartedThread = (await secondSystem.run(restarted.getSnapshot())).threads.find(
337 (entry) => entry.id === "thread-revert",
338 );
339 expect(restartedThread?.messages.map((message) => message.id)).toEqual([
340 "user-1",
341 "assistant:turn-1",
342 ]);
343 expect(restartedThread?.turnDiffSummaries.map((summary) => summary.turnId)).toEqual(["turn-1"]);
344 await secondSystem.dispose();
345 });
346
347 it("allows stop to be called multiple times", async () => {
348 const inMemoryStore: OrchestrationEventRepositoryShape = {
349 append(event) {
350 return Effect.succeed({ ...event, sequence: 1 });
351 },
352 readFromSequence() {
353 return Effect.succeed([]);
354 },
355 readAll() {
356 return Effect.succeed([]);
357 },
358 };
359 const runtime = ManagedRuntime.make(
360 OrchestrationEngineLive.pipe(
361 Layer.provide(Layer.succeed(OrchestrationEventRepository, inMemoryStore)),
362 ),
363 );
364 await runtime.runPromise(OrchestrationEngineService);
365 await runtime.dispose();
366 await expect(runtime.dispose()).resolves.toBeUndefined();
367 });
368
369 it("keeps processing queued commands after a storage failure", async () => {
370 const events: OrchestrationEvent[] = [];
371 let nextSequence = 1;
372 let shouldFailFirstAppend = true;
373
374 const flakyStore: OrchestrationEventRepositoryShape = {
375 append(event) {
376 if (shouldFailFirstAppend) {
377 shouldFailFirstAppend = false;
378 return Effect.die(new Error("append failed"));
379 }
380 const savedEvent = {
381 ...event,
382 sequence: nextSequence,
383 } satisfies OrchestrationEvent;
384 nextSequence += 1;
385 events.push(savedEvent);
386 return Effect.succeed(savedEvent);
387 },
388 readFromSequence(sequenceExclusive) {
389 return Effect.succeed(events.filter((event) => event.sequence > sequenceExclusive));
390 },
391 readAll() {
392 return Effect.succeed(events);
393 },
394 };
395
396 const runtime = ManagedRuntime.make(
397 OrchestrationEngineLive.pipe(
398 Layer.provide(Layer.succeed(OrchestrationEventRepository, flakyStore)),
399 ),
400 );
401 const engine = await runtime.runPromise(OrchestrationEngineService);
402 const createdAt = new Date().toISOString();
403
404 await expect(
405 runtime.runPromise(
406 engine.dispatch({
407 type: "thread.create",
408 commandId: "cmd-flaky-1",
409 threadId: "thread-flaky-fail",
410 projectId: "project-flaky",
411 title: "flaky-fail",
412 model: "gpt-5-codex",
413 branch: null,
414 worktreePath: null,
415 createdAt,
416 }),
417 ),
418 ).rejects.toThrow("append failed");
419
420 const result = await runtime.runPromise(
421 engine.dispatch({
422 type: "thread.create",
423 commandId: "cmd-flaky-2",
424 threadId: "thread-flaky-ok",
425 projectId: "project-flaky",
426 title: "flaky-ok",
427 model: "gpt-5-codex",
428 branch: null,
429 worktreePath: null,
430 createdAt,
431 }),
432 );
433
434 expect(result.sequence).toBe(1);
435 expect((await runtime.runPromise(engine.getSnapshot())).sequence).toBe(1);
436
437 await runtime.dispose();
438 });
439});
apps/server/src/orchestration/Layer.ts created+297
......@@ -0,0 +1,297 @@
1import type {
2 OrchestrationCommand,
3 OrchestrationEvent,
4 OrchestrationReadModel,
5} from "@t3tools/contracts";
6import { OrchestrationCommandSchema } from "@t3tools/contracts";
7import {
8 Cause,
9 Deferred,
10 Effect,
11 Either,
12 Fiber,
13 Layer,
14 PubSub,
15 Queue,
16 Runtime,
17 Schema,
18} from "effect";
19
20import { createLogger } from "../logger";
21import { OrchestrationEventRepository } from "../persistence/Services/OrchestrationEvents";
22import { createEmptyReadModel, reduceEvent } from "./reducer";
23import { OrchestrationEngineService, type OrchestrationEngineShape } from "./Service";
24
25type CommandEnvelope = {
26 command: OrchestrationCommand;
27 result: Deferred.Deferred<{ sequence: number }, Error>;
28};
29
30function asError(error: unknown, fallbackMessage: string): Error {
31 return error instanceof Error ? error : new Error(fallbackMessage);
32}
33
34function mapCommandToEvent(command: OrchestrationCommand): Omit<OrchestrationEvent, "sequence"> {
35 const eventId = crypto.randomUUID();
36 switch (command.type) {
37 case "thread.create":
38 return {
39 eventId,
40 type: "thread.created",
41 aggregateType: "thread",
42 aggregateId: command.threadId,
43 occurredAt: command.createdAt,
44 commandId: command.commandId,
45 payload: {
46 id: command.threadId,
47 projectId: command.projectId,
48 title: command.title,
49 model: command.model,
50 branch: command.branch,
51 worktreePath: command.worktreePath,
52 createdAt: command.createdAt,
53 updatedAt: command.createdAt,
54 },
55 };
56 case "thread.delete":
57 return {
58 eventId,
59 type: "thread.deleted",
60 aggregateType: "thread",
61 aggregateId: command.threadId,
62 occurredAt: command.createdAt,
63 commandId: command.commandId,
64 payload: {
65 id: command.threadId,
66 deletedAt: command.createdAt,
67 },
68 };
69 case "thread.meta.update":
70 return {
71 eventId,
72 type: "thread.meta-updated",
73 aggregateType: "thread",
74 aggregateId: command.threadId,
75 occurredAt: command.createdAt,
76 commandId: command.commandId,
77 payload: {
78 threadId: command.threadId,
79 ...(command.title !== undefined ? { title: command.title } : {}),
80 ...(command.model !== undefined ? { model: command.model } : {}),
81 ...(command.branch !== undefined ? { branch: command.branch } : {}),
82 ...(command.worktreePath !== undefined ? { worktreePath: command.worktreePath } : {}),
83 updatedAt: command.createdAt,
84 },
85 };
86 case "message.send":
87 return {
88 eventId,
89 type: "message.sent",
90 aggregateType: "thread",
91 aggregateId: command.threadId,
92 occurredAt: command.createdAt,
93 commandId: command.commandId,
94 payload: {
95 id: command.messageId,
96 role: command.role,
97 text: command.text,
98 threadId: command.threadId,
99 createdAt: command.createdAt,
100 streaming: command.streaming === true,
101 },
102 };
103 case "thread.session":
104 return {
105 eventId,
106 type: "thread.session-set",
107 aggregateType: "thread",
108 aggregateId: command.threadId,
109 occurredAt: command.createdAt,
110 commandId: command.commandId,
111 payload: {
112 threadId: command.threadId,
113 session: command.session,
114 },
115 };
116 case "git.readModel.upsert":
117 return {
118 eventId,
119 type: "git.read-model-upsert",
120 aggregateType: "project",
121 aggregateId: command.projectId,
122 occurredAt: command.createdAt,
123 commandId: command.commandId,
124 payload: {
125 projectId: command.projectId,
126 branch: command.branch,
127 hasWorkingTreeChanges: command.hasWorkingTreeChanges,
128 aheadCount: command.aheadCount,
129 behindCount: command.behindCount,
130 updatedAt: command.createdAt,
131 },
132 };
133 case "thread.turnDiff.complete":
134 return {
135 eventId,
136 type: "thread.turn-diff-completed",
137 aggregateType: "thread",
138 aggregateId: command.threadId,
139 occurredAt: command.createdAt,
140 commandId: command.commandId,
141 payload: {
142 threadId: command.threadId,
143 turnId: command.turnId,
144 completedAt: command.completedAt,
145 ...(command.status !== undefined ? { status: command.status } : {}),
146 files: command.files,
147 ...(command.assistantMessageId !== undefined
148 ? { assistantMessageId: command.assistantMessageId }
149 : {}),
150 ...(command.checkpointTurnCount !== undefined
151 ? { checkpointTurnCount: command.checkpointTurnCount }
152 : {}),
153 },
154 };
155 case "thread.revert":
156 return {
157 eventId,
158 type: "thread.reverted",
159 aggregateType: "thread",
160 aggregateId: command.threadId,
161 occurredAt: command.createdAt,
162 commandId: command.commandId,
163 payload: {
164 threadId: command.threadId,
165 turnCount: command.turnCount,
166 messageCount: command.messageCount,
167 },
168 };
169 }
170}
171
172const decodeUnknownCommand = Schema.decodeUnknownEither(OrchestrationCommandSchema);
173
174export const makeOrchestrationEngine = Effect.gen(function* () {
175 const logger = createLogger("orchestration");
176 const eventStore = yield* OrchestrationEventRepository;
177 const runtime = Runtime.defaultRuntime;
178
179 let readModel = createEmptyReadModel(new Date().toISOString());
180
181 const commandQueue = yield* Queue.unbounded<CommandEnvelope>();
182 const readModelPubSub = yield* PubSub.unbounded<OrchestrationReadModel>();
183 const eventPubSub = yield* PubSub.unbounded<OrchestrationEvent>();
184
185 const readModelListeners = new Set<(snapshot: OrchestrationReadModel) => void>();
186 const domainEventListeners = new Set<(event: OrchestrationEvent) => void>();
187
188 const processEnvelope = (envelope: CommandEnvelope): Effect.Effect<void> =>
189 Effect.gen(function* () {
190 const eventBase = mapCommandToEvent(envelope.command);
191 const savedEvent = yield* eventStore.append(eventBase);
192 readModel = yield* reduceEvent(readModel, savedEvent);
193
194 const snapshot = readModel;
195 yield* Effect.all([
196 PubSub.publish(eventPubSub, savedEvent),
197 PubSub.publish(readModelPubSub, snapshot),
198 ]);
199
200 for (const listener of domainEventListeners) {
201 listener(savedEvent);
202 }
203 for (const listener of readModelListeners) {
204 listener(snapshot);
205 }
206
207 yield* Deferred.succeed(envelope.result, { sequence: savedEvent.sequence });
208 }).pipe(
209 Effect.catchAllCause((cause) =>
210 Deferred.fail(
211 envelope.result,
212 asError(Cause.squash(cause), "Unknown command processing error"),
213 ).pipe(Effect.asVoid),
214 ),
215 );
216
217 const bootstrapReadModel: Effect.Effect<void, unknown> = Effect.gen(function* () {
218 const existingEvents = yield* eventStore.readAll();
219 for (const event of existingEvents) {
220 readModel = yield* reduceEvent(readModel, event);
221 }
222 });
223
224 yield* bootstrapReadModel;
225
226 const worker = Effect.forever(Queue.take(commandQueue).pipe(Effect.flatMap(processEnvelope)));
227 const workerFiber: Fiber.RuntimeFiber<void, unknown> = Runtime.runFork(runtime)(worker);
228 logger.info("orchestration engine started", {
229 sequence: readModel.sequence,
230 });
231
232 yield* Effect.addFinalizer(() => Fiber.interrupt(workerFiber).pipe(Effect.orDie));
233
234 const getSnapshot: OrchestrationEngineShape["getSnapshot"] = () =>
235 Effect.sync((): OrchestrationReadModel => readModel);
236
237 const replayEvents: OrchestrationEngineShape["replayEvents"] = (fromSequenceExclusive) =>
238 eventStore.readFromSequence(fromSequenceExclusive);
239
240 const dispatch: OrchestrationEngineShape["dispatch"] = (command) =>
241 Effect.gen(function* () {
242 const result = yield* Deferred.make<{ sequence: number }, Error>();
243 yield* Queue.offer(commandQueue, { command, result });
244 return yield* Deferred.await(result);
245 }).pipe(Effect.mapError((error) => asError(error, "Queue offer failed")));
246
247 const dispatchUnknown: OrchestrationEngineShape["dispatchUnknown"] = (command) =>
248 Effect.gen(function* () {
249 let payload: unknown = command;
250 if (typeof command === "string") {
251 try {
252 payload = JSON.parse(command) as unknown;
253 } catch {
254 return yield* Effect.fail(
255 new Error("Invalid orchestration command: payload is not valid JSON"),
256 );
257 }
258 }
259 const decoded = decodeUnknownCommand(payload);
260 if (Either.isLeft(decoded)) {
261 return yield* Effect.fail(
262 new Error(`Invalid orchestration command: ${decoded.left.toString()}`),
263 );
264 }
265 return yield* dispatch(decoded.right);
266 });
267
268 const subscribeToReadModel: OrchestrationEngineShape["subscribeToReadModel"] = (callback) =>
269 Effect.sync(() => {
270 readModelListeners.add(callback);
271 return () => {
272 readModelListeners.delete(callback);
273 };
274 });
275
276 const subscribeToDomainEvents: OrchestrationEngineShape["subscribeToDomainEvents"] = (callback) =>
277 Effect.sync(() => {
278 domainEventListeners.add(callback);
279 return () => {
280 domainEventListeners.delete(callback);
281 };
282 });
283
284 return {
285 getSnapshot,
286 replayEvents,
287 dispatchUnknown,
288 dispatch,
289 subscribeToReadModel,
290 subscribeToDomainEvents,
291 } satisfies OrchestrationEngineShape;
292});
293
294export const OrchestrationEngineLive = Layer.scoped(
295 OrchestrationEngineService,
296 makeOrchestrationEngine,
297);
apps/server/src/orchestration/Service.ts created+25
......@@ -0,0 +1,25 @@
1import type {
2 OrchestrationCommand,
3 OrchestrationEvent,
4 OrchestrationReadModel,
5} from "@t3tools/contracts";
6import { Context } from "effect";
7import type { Effect } from "effect";
8
9export interface OrchestrationEngineShape {
10 readonly getSnapshot: () => Effect.Effect<OrchestrationReadModel>;
11 readonly replayEvents: (fromSequenceExclusive: number) => Effect.Effect<OrchestrationEvent[]>;
12 readonly dispatchUnknown: (command: unknown) => Effect.Effect<{ sequence: number }, Error>;
13 readonly dispatch: (command: OrchestrationCommand) => Effect.Effect<{ sequence: number }, Error>;
14 readonly subscribeToReadModel: (
15 callback: (snapshot: OrchestrationReadModel) => void,
16 ) => Effect.Effect<() => void>;
17 readonly subscribeToDomainEvents: (
18 callback: (event: OrchestrationEvent) => void,
19 ) => Effect.Effect<() => void>;
20}
21
22export class OrchestrationEngineService extends Context.Tag("orchestration/Engine")<
23 OrchestrationEngineService,
24 OrchestrationEngineShape
25>() {}
apps/server/src/orchestration/engine.test.ts deleted-388
......@@ -1,388 +0,0 @@
1import fs from "node:fs";
2import os from "node:os";
3import path from "node:path";
4
5import type { OrchestrationEvent } from "@t3tools/contracts";
6import { Effect } from "effect";
7import { afterEach, describe, expect, it } from "vitest";
8
9import type { OrchestrationEventRepositoryShape } from "../persistence/Services/OrchestrationEvents";
10import { OrchestrationEngine } from "./engine";
11
12import { Layer, ManagedRuntime } from "effect";
13
14import { makeSqlitePersistenceLive } from "../persistence/Layers/Sqlite";
15
16import { OrchestrationLive } from "./layers";
17import { OrchestrationEngineService } from "./services";
18
19const tempDirs: string[] = [];
20
21function makeTempDir(prefix: string): string {
22 const dir = fs.mkdtempSync(path.join(os.tmpdir(), prefix));
23 tempDirs.push(dir);
24 return dir;
25}
26
27afterEach(() => {
28 for (const dir of tempDirs.splice(0, tempDirs.length)) {
29 fs.rmSync(dir, { recursive: true, force: true });
30 }
31});
32
33export async function createOrchestrationSystem(stateDir: string) {
34 const dbPath = path.join(stateDir, "orchestration.sqlite");
35 const orchestrationLayer = OrchestrationLive.pipe(
36 Layer.provide(makeSqlitePersistenceLive(dbPath)),
37 );
38 const runtime = ManagedRuntime.make(orchestrationLayer);
39 const engine = await runtime.runPromise(OrchestrationEngineService);
40 return {
41 engine,
42 dispose: () => runtime.dispose(),
43 };
44}
45
46describe("OrchestrationEngine", () => {
47 it("replays to the same deterministic snapshot", async () => {
48 const stateDir = makeTempDir("t3code-orchestration-");
49 const createdAt = new Date().toISOString();
50 const projectId = "project-1";
51 const threadId = "thread-1";
52
53 const firstSystem = await createOrchestrationSystem(stateDir);
54 const engineA = firstSystem.engine;
55 await engineA.dispatch({
56 type: "thread.create",
57 commandId: "cmd-1",
58 threadId,
59 projectId,
60 title: "Thread",
61 model: "gpt-5-codex",
62 branch: null,
63 worktreePath: null,
64 createdAt,
65 });
66 await engineA.dispatch({
67 type: "message.send",
68 commandId: "cmd-3",
69 threadId,
70 messageId: "msg-1",
71 role: "user",
72 text: "hello",
73 streaming: false,
74 createdAt,
75 });
76 const snapshotA = engineA.getSnapshot();
77 await firstSystem.dispose();
78
79 const secondSystem = await createOrchestrationSystem(stateDir);
80 const engineB = secondSystem.engine;
81 const snapshotB = engineB.getSnapshot();
82 expect(snapshotB).toEqual(snapshotA);
83 await secondSystem.dispose();
84 });
85
86 it("fans out read-model updates to subscribers", async () => {
87 const stateDir = makeTempDir("t3code-orchestration-fanout-");
88 const system = await createOrchestrationSystem(stateDir);
89 const engine = system.engine;
90 const updates: number[] = [];
91 const unsubscribe = engine.subscribeToReadModel((snapshot) => {
92 updates.push(snapshot.sequence);
93 });
94 await engine.dispatch({
95 type: "thread.create",
96 commandId: "cmd-thread",
97 threadId: "thread-2",
98 projectId: "project-2",
99 title: "fanout",
100 model: "gpt-5-codex",
101 branch: null,
102 worktreePath: null,
103 createdAt: new Date().toISOString(),
104 });
105 unsubscribe();
106 expect(updates.length).toBeGreaterThan(0);
107 await system.dispose();
108 });
109
110 it("replays append-only events from sequence", async () => {
111 const stateDir = makeTempDir("t3code-orchestration-replay-");
112 const system = await createOrchestrationSystem(stateDir);
113 const engine = system.engine;
114 await engine.dispatch({
115 type: "thread.create",
116 commandId: "cmd-a",
117 threadId: "thread-replay",
118 projectId: "project-replay",
119 title: "replay",
120 model: "gpt-5-codex",
121 branch: null,
122 worktreePath: null,
123 createdAt: new Date().toISOString(),
124 });
125 await engine.dispatch({
126 type: "thread.delete",
127 commandId: "cmd-b",
128 threadId: "thread-replay",
129 createdAt: new Date().toISOString(),
130 });
131 const events = await engine.replayEvents(0);
132 expect(events.length).toBe(2);
133 expect(events[0]?.type).toBe("thread.created");
134 expect(events[1]?.type).toBe("thread.deleted");
135 await system.dispose();
136 });
137
138 it("stores completed turn summaries even when no files changed", async () => {
139 const stateDir = makeTempDir("t3code-orchestration-turn-diff-");
140 const firstSystem = await createOrchestrationSystem(stateDir);
141 const engine = firstSystem.engine;
142 const completedAt = new Date().toISOString();
143 await engine.dispatch({
144 type: "thread.create",
145 commandId: "cmd-thread-turn-diff",
146 threadId: "thread-turn-diff",
147 projectId: "project-turn-diff",
148 title: "Turn diff thread",
149 model: "gpt-5-codex",
150 branch: null,
151 worktreePath: null,
152 createdAt: completedAt,
153 });
154 await engine.dispatch({
155 type: "thread.turnDiff.complete",
156 commandId: "cmd-turn-diff-complete",
157 threadId: "thread-turn-diff",
158 turnId: "turn-1",
159 completedAt,
160 status: "completed",
161 files: [],
162 checkpointTurnCount: 1,
163 createdAt: completedAt,
164 });
165
166 const firstThread = engine
167 .getSnapshot()
168 .threads.find((thread) => thread.id === "thread-turn-diff");
169 expect(firstThread?.turnDiffSummaries).toEqual([
170 {
171 turnId: "turn-1",
172 completedAt,
173 status: "completed",
174 files: [],
175 checkpointTurnCount: 1,
176 },
177 ]);
178 await firstSystem.dispose();
179
180 const secondSystem = await createOrchestrationSystem(stateDir);
181 const restartedEngine = secondSystem.engine;
182 const restartedThread = restartedEngine
183 .getSnapshot()
184 .threads.find((thread) => thread.id === "thread-turn-diff");
185 expect(restartedThread?.turnDiffSummaries).toEqual([
186 {
187 turnId: "turn-1",
188 completedAt,
189 status: "completed",
190 files: [],
191 checkpointTurnCount: 1,
192 },
193 ]);
194 await secondSystem.dispose();
195 });
196
197 it("reverts thread messages and turn summaries to a checkpoint", async () => {
198 const stateDir = makeTempDir("t3code-orchestration-revert-");
199 const firstSystem = await createOrchestrationSystem(stateDir);
200 const engine = firstSystem.engine;
201 const createdAt = new Date().toISOString();
202 await engine.dispatch({
203 type: "thread.create",
204 commandId: "cmd-thread-revert",
205 threadId: "thread-revert",
206 projectId: "project-revert",
207 title: "Revert thread",
208 model: "gpt-5-codex",
209 branch: null,
210 worktreePath: null,
211 createdAt,
212 });
213 await engine.dispatch({
214 type: "message.send",
215 commandId: "cmd-msg-1-user",
216 threadId: "thread-revert",
217 messageId: "user-1",
218 role: "user",
219 text: "first",
220 streaming: false,
221 createdAt,
222 });
223 await engine.dispatch({
224 type: "message.send",
225 commandId: "cmd-msg-1-assistant",
226 threadId: "thread-revert",
227 messageId: "assistant:turn-1",
228 role: "assistant",
229 text: "first-response",
230 streaming: false,
231 createdAt,
232 });
233 await engine.dispatch({
234 type: "thread.turnDiff.complete",
235 commandId: "cmd-turn-1-complete",
236 threadId: "thread-revert",
237 turnId: "turn-1",
238 completedAt: createdAt,
239 status: "completed",
240 files: [],
241 assistantMessageId: "assistant:turn-1",
242 checkpointTurnCount: 1,
243 createdAt,
244 });
245
246 const createdAtSecond = new Date(Date.now() + 1_000).toISOString();
247 await engine.dispatch({
248 type: "message.send",
249 commandId: "cmd-msg-2-user",
250 threadId: "thread-revert",
251 messageId: "user-2",
252 role: "user",
253 text: "second",
254 streaming: false,
255 createdAt: createdAtSecond,
256 });
257 await engine.dispatch({
258 type: "message.send",
259 commandId: "cmd-msg-2-assistant",
260 threadId: "thread-revert",
261 messageId: "assistant:turn-2",
262 role: "assistant",
263 text: "second-response",
264 streaming: false,
265 createdAt: createdAtSecond,
266 });
267 await engine.dispatch({
268 type: "thread.turnDiff.complete",
269 commandId: "cmd-turn-2-complete",
270 threadId: "thread-revert",
271 turnId: "turn-2",
272 completedAt: createdAtSecond,
273 status: "completed",
274 files: [],
275 assistantMessageId: "assistant:turn-2",
276 checkpointTurnCount: 2,
277 createdAt: createdAtSecond,
278 });
279
280 await engine.dispatch({
281 type: "thread.revert",
282 commandId: "cmd-thread-revert-apply",
283 threadId: "thread-revert",
284 turnCount: 1,
285 messageCount: 2,
286 createdAt: createdAtSecond,
287 });
288
289 const thread = engine.getSnapshot().threads.find((entry) => entry.id === "thread-revert");
290 expect(thread?.messages.map((message) => message.id)).toEqual(["user-1", "assistant:turn-1"]);
291 expect(thread?.turnDiffSummaries.map((summary) => summary.turnId)).toEqual(["turn-1"]);
292 expect(thread?.latestTurnId).toBe("turn-1");
293 await firstSystem.dispose();
294
295 const secondSystem = await createOrchestrationSystem(stateDir);
296 const restarted = secondSystem.engine;
297 const restartedThread = restarted
298 .getSnapshot()
299 .threads.find((entry) => entry.id === "thread-revert");
300 expect(restartedThread?.messages.map((message) => message.id)).toEqual([
301 "user-1",
302 "assistant:turn-1",
303 ]);
304 expect(restartedThread?.turnDiffSummaries.map((summary) => summary.turnId)).toEqual(["turn-1"]);
305 await secondSystem.dispose();
306 });
307
308 it("allows stop to be called multiple times", async () => {
309 const inMemoryStore: OrchestrationEventRepositoryShape = {
310 append(event) {
311 return Effect.succeed({ ...event, sequence: 1 });
312 },
313 readFromSequence() {
314 return Effect.succeed([]);
315 },
316 readAll() {
317 return Effect.succeed([]);
318 },
319 };
320 const engine = new OrchestrationEngine(inMemoryStore);
321 await engine.start();
322 await engine.stop();
323 await expect(engine.stop()).resolves.toBeUndefined();
324 });
325
326 it("keeps processing queued commands after a storage failure", async () => {
327 const events: OrchestrationEvent[] = [];
328 let nextSequence = 1;
329 let shouldFailFirstAppend = true;
330
331 const flakyStore: OrchestrationEventRepositoryShape = {
332 append(event) {
333 if (shouldFailFirstAppend) {
334 shouldFailFirstAppend = false;
335 return Effect.die(new Error("append failed"));
336 }
337 const savedEvent = {
338 ...event,
339 sequence: nextSequence,
340 } satisfies OrchestrationEvent;
341 nextSequence += 1;
342 events.push(savedEvent);
343 return Effect.succeed(savedEvent);
344 },
345 readFromSequence(sequenceExclusive) {
346 return Effect.succeed(events.filter((event) => event.sequence > sequenceExclusive));
347 },
348 readAll() {
349 return Effect.succeed(events);
350 },
351 };
352
353 const engine = new OrchestrationEngine(flakyStore);
354 const createdAt = new Date().toISOString();
355 await engine.start();
356
357 await expect(
358 engine.dispatch({
359 type: "thread.create",
360 commandId: "cmd-flaky-1",
361 threadId: "thread-flaky-fail",
362 projectId: "project-flaky",
363 title: "flaky-fail",
364 model: "gpt-5-codex",
365 branch: null,
366 worktreePath: null,
367 createdAt,
368 }),
369 ).rejects.toThrow("append failed");
370
371 const result = await engine.dispatch({
372 type: "thread.create",
373 commandId: "cmd-flaky-2",
374 threadId: "thread-flaky-ok",
375 projectId: "project-flaky",
376 title: "flaky-ok",
377 model: "gpt-5-codex",
378 branch: null,
379 worktreePath: null,
380 createdAt,
381 });
382
383 expect(result.sequence).toBe(1);
384 expect(engine.getSnapshot().sequence).toBe(1);
385
386 await engine.stop();
387 });
388});
apps/server/src/orchestration/engine.ts deleted-309
......@@ -1,309 +0,0 @@
1import type {
2 OrchestrationCommand,
3 OrchestrationEvent,
4 OrchestrationReadModel,
5} from "@t3tools/contracts";
6import { OrchestrationCommandSchema } from "@t3tools/contracts";
7import {
8 PubSub,
9 Queue,
10 Schema,
11 Stream,
12 Effect,
13 Fiber,
14 Runtime,
15 Either,
16 Deferred,
17 Cause,
18} from "effect";
19
20import { createLogger } from "../logger";
21import type { OrchestrationEventRepositoryShape } from "../persistence/Services/OrchestrationEvents";
22import { createEmptyReadModel, reduceEvent } from "./reducer";
23
24type CommandEnvelope = {
25 command: OrchestrationCommand;
26 result: Deferred.Deferred<{ sequence: number }, Error>;
27};
28
29function asError(error: unknown, fallbackMessage: string): Error {
30 return error instanceof Error ? error : new Error(fallbackMessage);
31}
32
33function mapCommandToEvent(command: OrchestrationCommand): Omit<OrchestrationEvent, "sequence"> {
34 const eventId = crypto.randomUUID();
35 switch (command.type) {
36 case "thread.create":
37 return {
38 eventId,
39 type: "thread.created",
40 aggregateType: "thread",
41 aggregateId: command.threadId,
42 occurredAt: command.createdAt,
43 commandId: command.commandId,
44 payload: {
45 id: command.threadId,
46 projectId: command.projectId,
47 title: command.title,
48 model: command.model,
49 branch: command.branch,
50 worktreePath: command.worktreePath,
51 createdAt: command.createdAt,
52 updatedAt: command.createdAt,
53 },
54 };
55 case "thread.delete":
56 return {
57 eventId,
58 type: "thread.deleted",
59 aggregateType: "thread",
60 aggregateId: command.threadId,
61 occurredAt: command.createdAt,
62 commandId: command.commandId,
63 payload: {
64 id: command.threadId,
65 deletedAt: command.createdAt,
66 },
67 };
68 case "thread.meta.update":
69 return {
70 eventId,
71 type: "thread.meta-updated",
72 aggregateType: "thread",
73 aggregateId: command.threadId,
74 occurredAt: command.createdAt,
75 commandId: command.commandId,
76 payload: {
77 threadId: command.threadId,
78 ...(command.title !== undefined ? { title: command.title } : {}),
79 ...(command.model !== undefined ? { model: command.model } : {}),
80 ...(command.branch !== undefined ? { branch: command.branch } : {}),
81 ...(command.worktreePath !== undefined ? { worktreePath: command.worktreePath } : {}),
82 updatedAt: command.createdAt,
83 },
84 };
85 case "message.send":
86 return {
87 eventId,
88 type: "message.sent",
89 aggregateType: "thread",
90 aggregateId: command.threadId,
91 occurredAt: command.createdAt,
92 commandId: command.commandId,
93 payload: {
94 id: command.messageId,
95 role: command.role,
96 text: command.text,
97 threadId: command.threadId,
98 createdAt: command.createdAt,
99 streaming: command.streaming === true,
100 },
101 };
102 case "thread.session":
103 return {
104 eventId,
105 type: "thread.session-set",
106 aggregateType: "thread",
107 aggregateId: command.threadId,
108 occurredAt: command.createdAt,
109 commandId: command.commandId,
110 payload: {
111 threadId: command.threadId,
112 session: command.session,
113 },
114 };
115 case "git.readModel.upsert":
116 return {
117 eventId,
118 type: "git.read-model-upsert",
119 aggregateType: "project",
120 aggregateId: command.projectId,
121 occurredAt: command.createdAt,
122 commandId: command.commandId,
123 payload: {
124 projectId: command.projectId,
125 branch: command.branch,
126 hasWorkingTreeChanges: command.hasWorkingTreeChanges,
127 aheadCount: command.aheadCount,
128 behindCount: command.behindCount,
129 updatedAt: command.createdAt,
130 },
131 };
132 case "thread.turnDiff.complete":
133 return {
134 eventId,
135 type: "thread.turn-diff-completed",
136 aggregateType: "thread",
137 aggregateId: command.threadId,
138 occurredAt: command.createdAt,
139 commandId: command.commandId,
140 payload: {
141 threadId: command.threadId,
142 turnId: command.turnId,
143 completedAt: command.completedAt,
144 ...(command.status !== undefined ? { status: command.status } : {}),
145 files: command.files,
146 ...(command.assistantMessageId !== undefined
147 ? { assistantMessageId: command.assistantMessageId }
148 : {}),
149 ...(command.checkpointTurnCount !== undefined
150 ? { checkpointTurnCount: command.checkpointTurnCount }
151 : {}),
152 },
153 };
154 case "thread.revert":
155 return {
156 eventId,
157 type: "thread.reverted",
158 aggregateType: "thread",
159 aggregateId: command.threadId,
160 occurredAt: command.createdAt,
161 commandId: command.commandId,
162 payload: {
163 threadId: command.threadId,
164 turnCount: command.turnCount,
165 messageCount: command.messageCount,
166 },
167 };
168 }
169}
170
171export class OrchestrationEngine {
172 private readonly logger = createLogger("orchestration");
173 private readonly runtime = Runtime.defaultRuntime;
174 private readonly eventStore: OrchestrationEventRepositoryShape;
175
176 private readModel: OrchestrationReadModel;
177 private commandQueue: Queue.Queue<CommandEnvelope>;
178 private readModelPubSub: PubSub.PubSub<OrchestrationReadModel>;
179 private eventPubSub: PubSub.PubSub<OrchestrationEvent>;
180 private workerFiber: Fiber.RuntimeFiber<void, unknown> | null = null;
181 private readonly readModelListeners = new Set<(snapshot: OrchestrationReadModel) => void>();
182 private readonly domainEventListeners = new Set<(event: OrchestrationEvent) => void>();
183
184 constructor(eventStore: OrchestrationEventRepositoryShape) {
185 this.eventStore = eventStore;
186 this.readModel = createEmptyReadModel(new Date().toISOString());
187 this.commandQueue = Runtime.runSync(this.runtime)(Queue.unbounded<CommandEnvelope>());
188 this.readModelPubSub = Runtime.runSync(this.runtime)(
189 PubSub.unbounded<OrchestrationReadModel>(),
190 );
191 this.eventPubSub = Runtime.runSync(this.runtime)(PubSub.unbounded<OrchestrationEvent>());
192 }
193
194 private processEnvelope(envelope: CommandEnvelope): Effect.Effect<void> {
195 return Effect.gen(this, function* () {
196 const eventBase = mapCommandToEvent(envelope.command);
197 const savedEvent = yield* this.eventStore.append(eventBase);
198 this.readModel = yield* reduceEvent(this.readModel, savedEvent);
199
200 const snapshot = this.readModel;
201 yield* Effect.all([
202 PubSub.publish(this.eventPubSub, savedEvent),
203 PubSub.publish(this.readModelPubSub, snapshot),
204 ]);
205
206 for (const listener of this.domainEventListeners) {
207 listener(savedEvent);
208 }
209 for (const listener of this.readModelListeners) {
210 listener(snapshot);
211 }
212
213 yield* Deferred.succeed(envelope.result, { sequence: savedEvent.sequence });
214 }).pipe(
215 Effect.catchAllCause((cause) =>
216 Deferred.fail(
217 envelope.result,
218 asError(Cause.squash(cause), "Unknown command processing error"),
219 ).pipe(Effect.asVoid),
220 ),
221 );
222 }
223
224 private bootstrapReadModel(): Effect.Effect<void, unknown> {
225 return Effect.gen(this, function* () {
226 const existingEvents = yield* this.eventStore.readAll();
227 for (const event of existingEvents) {
228 this.readModel = yield* reduceEvent(this.readModel, event);
229 }
230 });
231 }
232
233 async start(): Promise<void> {
234 if (this.workerFiber) {
235 return;
236 }
237
238 await Runtime.runPromise(this.runtime)(this.bootstrapReadModel());
239
240 const worker = Stream.fromQueue(this.commandQueue).pipe(
241 Stream.runForEach((envelope) => this.processEnvelope(envelope)),
242 );
243
244 this.workerFiber = Runtime.runFork(this.runtime)(worker);
245 this.logger.info("orchestration engine started", {
246 sequence: this.readModel.sequence,
247 });
248 }
249
250 async stop(): Promise<void> {
251 if (this.workerFiber) {
252 await Runtime.runPromise(this.runtime)(Fiber.interrupt(this.workerFiber));
253 this.workerFiber = null;
254 }
255 }
256
257 getSnapshot(): OrchestrationReadModel {
258 return this.readModel;
259 }
260
261 async replayEvents(fromSequenceExclusive: number): Promise<OrchestrationEvent[]> {
262 return Runtime.runPromise(this.runtime)(
263 this.eventStore.readFromSequence(fromSequenceExclusive),
264 );
265 }
266
267 async dispatchUnknown(command: unknown): Promise<{ sequence: number }> {
268 const decode = Schema.decodeUnknownEither(OrchestrationCommandSchema);
269 let payload: unknown = command;
270 if (typeof command === "string") {
271 try {
272 payload = JSON.parse(command) as unknown;
273 } catch {
274 throw new Error("Invalid orchestration command: payload is not valid JSON");
275 }
276 }
277 const decoded = decode(payload);
278 if (Either.isLeft(decoded)) {
279 const issues = decoded.left.toString();
280 throw new Error(`Invalid orchestration command: ${issues}`);
281 }
282 return this.dispatch(decoded.right);
283 }
284
285 async dispatch(command: OrchestrationCommand): Promise<{ sequence: number }> {
286 const program = Effect.gen(this, function* () {
287 const result = yield* Deferred.make<{ sequence: number }, Error>();
288 yield* Queue.offer(this.commandQueue, { command, result });
289 return yield* Deferred.await(result);
290 });
291 return Runtime.runPromise(this.runtime)(program).catch((error) => {
292 throw asError(error, "Queue offer failed");
293 });
294 }
295
296 subscribeToReadModel(callback: (snapshot: OrchestrationReadModel) => void): () => void {
297 this.readModelListeners.add(callback);
298 return () => {
299 this.readModelListeners.delete(callback);
300 };
301 }
302
303 subscribeToDomainEvents(callback: (event: OrchestrationEvent) => void): () => void {
304 this.domainEventListeners.add(callback);
305 return () => {
306 this.domainEventListeners.delete(callback);
307 };
308 }
309}
apps/server/src/orchestration/layers.ts deleted-23
......@@ -1,23 +0,0 @@
1import { Effect, Layer } from "effect";
2
3import { OrchestrationEventRepository } from "../persistence/Services/OrchestrationEvents";
4import { OrchestrationEventRepositoryLive } from "../persistence/Layers/OrchestrationEvents";
5import { OrchestrationEngine } from "./engine";
6import { OrchestrationEngineService } from "./services";
7
8export const OrchestrationEngineLive = Layer.scoped(
9 OrchestrationEngineService,
10 Effect.acquireRelease(
11 Effect.gen(function* () {
12 const eventRepository = yield* OrchestrationEventRepository;
13 const engine = new OrchestrationEngine(eventRepository);
14 yield* Effect.promise(() => engine.start());
15 return engine;
16 }),
17 (engine) => Effect.promise(() => engine.stop()),
18 ),
19);
20
21export const OrchestrationLive = OrchestrationEngineLive.pipe(
22 Layer.provide(OrchestrationEventRepositoryLive),
23);
apps/server/src/orchestration/services.ts deleted-8
......@@ -1,8 +0,0 @@
1import { Context } from "effect";
2
3import { OrchestrationEngine } from "./engine";
4
5export class OrchestrationEngineService extends Context.Tag("orchestration/Engine")<
6 OrchestrationEngineService,
7 OrchestrationEngine
8>() {}
apps/server/src/wsServer.ts+198-162
......@@ -36,13 +36,13 @@ import {
3636import { TerminalManager } from "./terminalManager";
3737import { loadResolvedKeybindingsConfig, upsertKeybindingRule } from "./keybindings";
3838import { searchWorkspaceEntries } from "./workspaceEntries";
39import type { OrchestrationEngine } from "./orchestration/engine";
40import { OrchestrationLive } from "./orchestration/layers";
41import { OrchestrationEngineService } from "./orchestration/services";
39import { OrchestrationEngineLive } from "./orchestration/Layer";
40import { OrchestrationEngineService, type OrchestrationEngineShape } from "./orchestration/Service";
4241import { ProjectRepositoryLive } from "./persistence/Layers/Projects";
4342import { makeSqlitePersistenceLive } from "./persistence/Layers/Sqlite";
4443import { ProjectRepository, type ProjectRepositoryShape } from "./persistence/Services/Projects";
4544import assert from "node:assert";
45import { OrchestrationEventRepositoryLive } from "./persistence/Layers/OrchestrationEvents";
4646
4747const MIME_TYPES: Record<string, string> = {
4848 ".html": "text/html; charset=utf-8",
......@@ -136,7 +136,7 @@ export function createServer(options: ServerOptions) {
136136 unknown
137137 > | null = null;
138138 let projectRepository: ProjectRepositoryShape | null = null;
139 let orchestrationEngine: OrchestrationEngine | null = null;
139 let orchestrationEngine: OrchestrationEngineShape | null = null;
140140 const clients = new Set<WebSocket>();
141141 const logger = createLogger("ws");
142142 const logWebSocketEvents =
......@@ -169,141 +169,157 @@ export function createServer(options: ServerOptions) {
169169 return effectRuntime;
170170 };
171171
172 function attachOrchestrationSubscriptions() {
173 const orchestrationEngine = getOrchestrationEngine();
174
175 unsubscribeReadModel = orchestrationEngine.subscribeToReadModel((snapshot) => {
176 const push: WsPush = {
177 type: "push",
178 channel: ORCHESTRATION_WS_CHANNELS.readModel,
179 data: {
180 sequence: snapshot.sequence,
181 snapshot,
182 } satisfies OrchestrationReadModelPush,
183 };
184 const message = JSON.stringify(push);
185 let recipients = 0;
186 for (const client of clients) {
187 if (client.readyState === client.OPEN) {
188 client.send(message);
189 recipients += 1;
172 async function attachOrchestrationSubscriptions(
173 runtime: ManagedRuntime.ManagedRuntime<ProjectRepository | OrchestrationEngineService, unknown>,
174 orchestrationEngine: OrchestrationEngineShape,
175 ) {
176 unsubscribeReadModel = await runtime.runPromise(
177 orchestrationEngine.subscribeToReadModel((snapshot) => {
178 const push: WsPush = {
179 type: "push",
180 channel: ORCHESTRATION_WS_CHANNELS.readModel,
181 data: {
182 sequence: snapshot.sequence,
183 snapshot,
184 } satisfies OrchestrationReadModelPush,
185 };
186 const message = JSON.stringify(push);
187 let recipients = 0;
188 for (const client of clients) {
189 if (client.readyState === client.OPEN) {
190 client.send(message);
191 recipients += 1;
192 }
190193 }
191 }
192 logOutgoingPush(push, recipients);
193 });
194 logOutgoingPush(push, recipients);
195 }),
196 );
194197
195 unsubscribeDomainEvents = orchestrationEngine.subscribeToDomainEvents((event) => {
196 const push: WsPush = {
197 type: "push",
198 channel: ORCHESTRATION_WS_CHANNELS.domainEvent,
199 data: event,
200 };
201 const message = JSON.stringify(push);
202 let recipients = 0;
203 for (const client of clients) {
204 if (client.readyState === client.OPEN) {
205 client.send(message);
206 recipients += 1;
198 unsubscribeDomainEvents = await runtime.runPromise(
199 orchestrationEngine.subscribeToDomainEvents((event) => {
200 const push: WsPush = {
201 type: "push",
202 channel: ORCHESTRATION_WS_CHANNELS.domainEvent,
203 data: event,
204 };
205 const message = JSON.stringify(push);
206 let recipients = 0;
207 for (const client of clients) {
208 if (client.readyState === client.OPEN) {
209 client.send(message);
210 recipients += 1;
211 }
207212 }
208 }
209 logOutgoingPush(push, recipients);
210 });
213 logOutgoingPush(push, recipients);
214 }),
215 );
211216 }
212217
213218 // Forward provider events to all connected WebSocket clients
214219 providerManager.on("event", (event) => {
215220 void (async () => {
216221 const liveOrchestrationEngine = orchestrationEngine;
217 if (!liveOrchestrationEngine) {
222 const runtime = effectRuntime;
223 if (!liveOrchestrationEngine || !runtime) {
218224 return;
219225 }
220 const snapshot = liveOrchestrationEngine.getSnapshot();
226 const snapshot = await runtime.runPromise(liveOrchestrationEngine.getSnapshot());
221227 const thread = snapshot.threads.find((entry) => entry.session?.sessionId === event.sessionId);
222228 if (!thread) return;
223229 const now = event.createdAt;
224230 if (event.method === "turn/started" || event.method === "turn/completed") {
225 await liveOrchestrationEngine.dispatch({
226 type: "thread.session",
227 commandId: crypto.randomUUID(),
228 threadId: thread.id,
229 session: {
230 sessionId: event.sessionId,
231 provider: event.provider,
232 status: event.method === "turn/started" ? "running" : "ready",
231 await runtime.runPromise(
232 liveOrchestrationEngine.dispatch({
233 type: "thread.session",
234 commandId: crypto.randomUUID(),
233235 threadId: thread.id,
234 activeTurnId: event.method === "turn/started" ? (event.turnId ?? null) : null,
235 createdAt: thread.session?.createdAt ?? now,
236 updatedAt: now,
237 lastError: null,
238 },
239 createdAt: now,
240 });
236 session: {
237 sessionId: event.sessionId,
238 provider: event.provider,
239 status: event.method === "turn/started" ? "running" : "ready",
240 threadId: thread.id,
241 activeTurnId: event.method === "turn/started" ? (event.turnId ?? null) : null,
242 createdAt: thread.session?.createdAt ?? now,
243 updatedAt: now,
244 lastError: null,
245 },
246 createdAt: now,
247 }),
248 );
241249 }
242250 if (event.textDelta && event.textDelta.length > 0) {
243251 const assistantMessageId = `assistant:${event.turnId ?? event.itemId ?? event.sessionId}`;
244 await liveOrchestrationEngine.dispatch({
245 type: "message.send",
246 commandId: crypto.randomUUID(),
247 threadId: thread.id,
248 messageId: assistantMessageId,
249 role: "assistant",
250 text: event.textDelta,
251 streaming: true,
252 createdAt: now,
253 });
252 await runtime.runPromise(
253 liveOrchestrationEngine.dispatch({
254 type: "message.send",
255 commandId: crypto.randomUUID(),
256 threadId: thread.id,
257 messageId: assistantMessageId,
258 role: "assistant",
259 text: event.textDelta,
260 streaming: true,
261 createdAt: now,
262 }),
263 );
254264 }
255265 if (event.method === "turn/completed") {
256266 const assistantMessageId = `assistant:${event.turnId ?? event.itemId ?? event.sessionId}`;
257 await liveOrchestrationEngine.dispatch({
258 type: "message.send",
259 commandId: crypto.randomUUID(),
260 threadId: thread.id,
261 messageId: assistantMessageId,
262 role: "assistant",
263 text: "",
264 streaming: false,
265 createdAt: now,
266 });
267 await runtime.runPromise(
268 liveOrchestrationEngine.dispatch({
269 type: "message.send",
270 commandId: crypto.randomUUID(),
271 threadId: thread.id,
272 messageId: assistantMessageId,
273 role: "assistant",
274 text: "",
275 streaming: false,
276 createdAt: now,
277 }),
278 );
267279 }
268280 if (event.method === "checkpoint/captured") {
269281 const payload = asObject(event.payload);
270282 const turnId = event.turnId ?? asString(payload?.turnId);
271283 if (turnId) {
272 await liveOrchestrationEngine.dispatch({
273 type: "thread.turnDiff.complete",
274 commandId: crypto.randomUUID(),
275 threadId: thread.id,
276 turnId,
277 completedAt: now,
278 ...(asString(payload?.status) !== null
279 ? { status: asString(payload?.status) ?? undefined }
280 : {}),
281 files: [],
282 assistantMessageId: `assistant:${turnId}`,
283 ...(asNumber(payload?.turnCount) !== null
284 ? { checkpointTurnCount: asNumber(payload?.turnCount) ?? undefined }
285 : {}),
286 createdAt: now,
287 });
284 await runtime.runPromise(
285 liveOrchestrationEngine.dispatch({
286 type: "thread.turnDiff.complete",
287 commandId: crypto.randomUUID(),
288 threadId: thread.id,
289 turnId,
290 completedAt: now,
291 ...(asString(payload?.status) !== null
292 ? { status: asString(payload?.status) ?? undefined }
293 : {}),
294 files: [],
295 assistantMessageId: `assistant:${turnId}`,
296 ...(asNumber(payload?.turnCount) !== null
297 ? { checkpointTurnCount: asNumber(payload?.turnCount) ?? undefined }
298 : {}),
299 createdAt: now,
300 }),
301 );
288302 }
289303 }
290304 if (event.kind === "error" && event.message) {
291 await liveOrchestrationEngine.dispatch({
292 type: "thread.session",
293 commandId: crypto.randomUUID(),
294 threadId: thread.id,
295 session: {
296 sessionId: event.sessionId,
297 provider: event.provider,
298 status: "error",
305 await runtime.runPromise(
306 liveOrchestrationEngine.dispatch({
307 type: "thread.session",
308 commandId: crypto.randomUUID(),
299309 threadId: thread.id,
300 activeTurnId: event.turnId ?? null,
301 createdAt: thread.session?.createdAt ?? now,
302 updatedAt: now,
303 lastError: event.message,
304 },
305 createdAt: now,
306 });
310 session: {
311 sessionId: event.sessionId,
312 provider: event.provider,
313 status: "error",
314 threadId: thread.id,
315 activeTurnId: event.turnId ?? null,
316 createdAt: thread.session?.createdAt ?? now,
317 updatedAt: now,
318 lastError: event.message,
319 },
320 createdAt: now,
321 }),
322 );
307323 }
308324 })().catch(() => undefined);
309325 });
......@@ -410,6 +426,7 @@ export function createServer(options: ServerOptions) {
410426 wss.on("connection", (ws) => {
411427 clients.add(ws);
412428 const orchestrationEngine = getOrchestrationEngine();
429 const runtime = getEffectRuntime();
413430
414431 // Send welcome message with project info
415432 const segments = cwd.split(/[/\\]/).filter(Boolean);
......@@ -423,16 +440,24 @@ export function createServer(options: ServerOptions) {
423440 logOutgoingPush(welcome, 1);
424441 ws.send(JSON.stringify(welcome));
425442
426 const snapshotPush: WsPush = {
427 type: "push",
428 channel: ORCHESTRATION_WS_CHANNELS.readModel,
429 data: {
430 sequence: orchestrationEngine.getSnapshot().sequence,
431 snapshot: orchestrationEngine.getSnapshot(),
432 } satisfies OrchestrationReadModelPush,
433 };
434 logOutgoingPush(snapshotPush, 1);
435 ws.send(JSON.stringify(snapshotPush));
443 void runtime
444 .runPromise(orchestrationEngine.getSnapshot())
445 .then((snapshot) => {
446 if (ws.readyState !== ws.OPEN) {
447 return;
448 }
449 const snapshotPush: WsPush = {
450 type: "push",
451 channel: ORCHESTRATION_WS_CHANNELS.readModel,
452 data: {
453 sequence: snapshot.sequence,
454 snapshot,
455 } satisfies OrchestrationReadModelPush,
456 };
457 logOutgoingPush(snapshotPush, 1);
458 ws.send(JSON.stringify(snapshotPush));
459 })
460 .catch(() => undefined);
436461
437462 ws.on("message", (raw) => {
438463 void handleMessage(ws, raw);
......@@ -509,32 +534,36 @@ export function createServer(options: ServerOptions) {
509534 const params = request.params as { sessionId?: string };
510535 const sessionId = params.sessionId;
511536 if (typeof sessionId === "string") {
512 const snapshot = orchestrationEngine.getSnapshot();
537 const snapshot = await effectRuntime.runPromise(orchestrationEngine.getSnapshot());
513538 const thread = snapshot.threads.find((entry) => entry.session?.sessionId === sessionId);
514539 if (thread) {
515540 const now = new Date().toISOString();
516 await orchestrationEngine.dispatch({
517 type: "thread.revert",
518 commandId: crypto.randomUUID(),
519 threadId: thread.id,
520 turnCount: result.turnCount,
521 messageCount: result.messageCount,
522 createdAt: now,
523 });
524 if (thread.session) {
525 await orchestrationEngine.dispatch({
526 type: "thread.session",
541 await effectRuntime.runPromise(
542 orchestrationEngine.dispatch({
543 type: "thread.revert",
527544 commandId: crypto.randomUUID(),
528545 threadId: thread.id,
529 session: {
530 ...thread.session,
531 status: "ready",
532 activeTurnId: null,
533 updatedAt: now,
534 lastError: null,
535 },
546 turnCount: result.turnCount,
547 messageCount: result.messageCount,
536548 createdAt: now,
537 });
549 }),
550 );
551 if (thread.session) {
552 await effectRuntime.runPromise(
553 orchestrationEngine.dispatch({
554 type: "thread.session",
555 commandId: crypto.randomUUID(),
556 threadId: thread.id,
557 session: {
558 ...thread.session,
559 status: "ready",
560 activeTurnId: null,
561 updatedAt: now,
562 lastError: null,
563 },
564 createdAt: now,
565 }),
566 );
538567 }
539568 }
540569 }
......@@ -605,16 +634,18 @@ export function createServer(options: ServerOptions) {
605634 const projects = await effectRuntime.runPromise(projectRepository.list());
606635 const project = projects.find((entry) => entry.cwd === params.cwd);
607636 if (project) {
608 await orchestrationEngine.dispatch({
609 type: "git.readModel.upsert",
610 commandId: crypto.randomUUID(),
611 projectId: project.id,
612 branch: status.branch,
613 hasWorkingTreeChanges: status.hasWorkingTreeChanges,
614 aheadCount: status.aheadCount,
615 behindCount: status.behindCount,
616 createdAt: new Date().toISOString(),
617 });
637 await effectRuntime.runPromise(
638 orchestrationEngine.dispatch({
639 type: "git.readModel.upsert",
640 commandId: crypto.randomUUID(),
641 projectId: project.id,
642 branch: status.branch,
643 hasWorkingTreeChanges: status.hasWorkingTreeChanges,
644 aheadCount: status.aheadCount,
645 behindCount: status.behindCount,
646 createdAt: new Date().toISOString(),
647 }),
648 );
618649 }
619650 return status;
620651 }
......@@ -677,19 +708,21 @@ export function createServer(options: ServerOptions) {
677708 };
678709
679710 case ORCHESTRATION_WS_METHODS.getSnapshot:
680 return orchestrationEngine.getSnapshot();
711 return effectRuntime.runPromise(orchestrationEngine.getSnapshot());
681712
682713 case ORCHESTRATION_WS_METHODS.dispatchCommand:
683 return orchestrationEngine.dispatchUnknown(request.params);
714 return effectRuntime.runPromise(orchestrationEngine.dispatchUnknown(request.params));
684715
685716 case ORCHESTRATION_WS_METHODS.replayEvents:
686 return orchestrationEngine.replayEvents(
687 Math.max(
688 0,
689 Math.floor(
690 Number(
691 (request.params as { fromSequenceExclusive?: number } | undefined)
692 ?.fromSequenceExclusive ?? 0,
717 return effectRuntime.runPromise(
718 orchestrationEngine.replayEvents(
719 Math.max(
720 0,
721 Math.floor(
722 Number(
723 (request.params as { fromSequenceExclusive?: number } | undefined)
724 ?.fromSequenceExclusive ?? 0,
725 ),
693726 ),
694727 ),
695728 ),
......@@ -703,11 +736,14 @@ export function createServer(options: ServerOptions) {
703736 async function createEffectRuntime() {
704737 const dbPath = path.join(resolvedStateDir, "orchestration.sqlite");
705738 const persistenceLayer = makeSqlitePersistenceLive(dbPath);
706 const runtime = ManagedRuntime.make(
707 Layer.mergeAll(OrchestrationLive, ProjectRepositoryLive).pipe(
708 Layer.provide(persistenceLayer),
709 ),
739 const orchestrationLayer = Layer.provide(
740 OrchestrationEngineLive,
741 OrchestrationEventRepositoryLive,
742 );
743 const layer = Layer.mergeAll(orchestrationLayer, ProjectRepositoryLive).pipe(
744 Layer.provide(persistenceLayer),
710745 );
746 const runtime = ManagedRuntime.make(layer);
711747
712748 try {
713749 const [nextOrchestrationEngine, repository] = await Promise.all([
......@@ -717,7 +753,7 @@ export function createServer(options: ServerOptions) {
717753 orchestrationEngine = nextOrchestrationEngine;
718754 projectRepository = repository;
719755 await runtime.runPromise(repository.pruneMissing());
720 attachOrchestrationSubscriptions();
756 await attachOrchestrationSubscriptions(runtime, nextOrchestrationEngine);
721757 } catch (error) {
722758 await runtime.dispose().catch(() => undefined);
723759 throw error;