authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-02-25 20:40:51-08:00
committergravatar for jmarminge@gmail.comJulius Marminge <jmarminge@gmail.com> 2026-02-26 13:20:36-08:00
log5a08b9dd64e5f13a9ad88c910e9f34cb7598199a
treead9b75bcb5a0ed6f394f5e580471561e548d3801
parent96d610b1f82255acd0fc077917dde7a944033f23

run in transaction

4 files changed, 596 insertions(+), 461 deletions(-)

.plans/16c-pr89-remediation-checklist.md+2-2
......@@ -154,8 +154,8 @@ Counts: active `53` (`valid=34`, `partially-valid=19`), closed-invalid `5`
154154
155155### Phase 2
156156
157- [ ] `C001` Non-atomic event appending can corrupt state on retry. If an error occurs mid-loop (lines 96-102) after some events are persisted but before the receipt is written, the command appears to fail. A retry generates new UUIDs via `crypto.randomUUID()` in the decider, appending duplicate events. Consider wrapping the loop in a transaction or using deterministic event IDs derived from `commandId`. <details> <summary>🚀 Reply "<strong>fix it for me</strong>" or copy this <strong>AI Prompt</strong> for your agent:</summary>
158 - Status: `TODO`
157- [x] `C001` Non-atomic event appending can corrupt state on retry. If an error occurs mid-loop (lines 96-102) after some events are persisted but before the receipt is written, the command appears to fail. A retry generates new UUIDs via `crypto.randomUUID()` in the decider, appending duplicate events. Consider wrapping the loop in a transaction or using deterministic event IDs derived from `commandId`. <details> <summary>🚀 Reply "<strong>fix it for me</strong>" or copy this <strong>AI Prompt</strong> for your agent:</summary>
158 - Status: `DONE`
159159 - Verdict: `valid`
160160 - Severity: `High`
161161 - Area: `Event ordering and state consistency`
apps/server/src/orchestration/Layer.test.ts deleted-432
......@@ -1,432 +0,0 @@
1import {
2 CheckpointRef,
3 CommandId,
4 MessageId,
5 ProjectId,
6 ThreadId,
7 TurnId,
8 type OrchestrationEvent,
9} from "@t3tools/contracts";
10import { Effect, Layer, ManagedRuntime, Queue, Stream } from "effect";
11import { describe, expect, it } from "vitest";
12
13import { PersistenceSqlError } from "../persistence/Errors.ts";
14import { OrchestrationCommandReceiptRepositoryLive } from "../persistence/Layers/OrchestrationCommandReceipts.ts";
15import { OrchestrationEventStoreLive } from "../persistence/Layers/OrchestrationEventStore.ts";
16import { SqlitePersistenceMemory } from "../persistence/Layers/Sqlite.ts";
17import {
18 OrchestrationEventStore,
19 type OrchestrationEventStoreShape,
20} from "../persistence/Services/OrchestrationEventStore.ts";
21import { OrchestrationEngineLive } from "./Layers/OrchestrationEngine.ts";
22import { OrchestrationProjectionPipelineLive } from "./Layers/ProjectionPipeline.ts";
23import { OrchestrationEngineService } from "./Services/OrchestrationEngine.ts";
24
25const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value);
26const asMessageId = (value: string): MessageId => MessageId.makeUnsafe(value);
27const asTurnId = (value: string): TurnId => TurnId.makeUnsafe(value);
28const asCheckpointRef = (value: string): CheckpointRef => CheckpointRef.makeUnsafe(value);
29
30async function createOrchestrationSystem() {
31 const orchestrationLayer = OrchestrationEngineLive.pipe(
32 Layer.provide(OrchestrationProjectionPipelineLive),
33 Layer.provide(OrchestrationEventStoreLive),
34 Layer.provide(OrchestrationCommandReceiptRepositoryLive),
35 Layer.provide(SqlitePersistenceMemory),
36 );
37 const runtime = ManagedRuntime.make(orchestrationLayer);
38 const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService));
39 return {
40 engine,
41 run: <A, E>(effect: Effect.Effect<A, E>) => runtime.runPromise(effect),
42 dispose: () => runtime.dispose(),
43 };
44}
45
46function now() {
47 return new Date().toISOString();
48}
49
50describe("OrchestrationEngine", () => {
51 it("returns deterministic read models for repeated reads", async () => {
52 const createdAt = now();
53 const system = await createOrchestrationSystem();
54 const { engine } = system;
55
56 await system.run(
57 engine.dispatch({
58 type: "project.create",
59 commandId: CommandId.makeUnsafe("cmd-project-1-create"),
60 projectId: asProjectId("project-1"),
61 title: "Project 1",
62 workspaceRoot: "/tmp/project-1",
63 defaultModel: "gpt-5-codex",
64 createdAt,
65 }),
66 );
67 await system.run(
68 engine.dispatch({
69 type: "thread.create",
70 commandId: CommandId.makeUnsafe("cmd-thread-1-create"),
71 threadId: ThreadId.makeUnsafe("thread-1"),
72 projectId: asProjectId("project-1"),
73 title: "Thread",
74 model: "gpt-5-codex",
75 branch: null,
76 worktreePath: null,
77 createdAt,
78 }),
79 );
80 await system.run(
81 engine.dispatch({
82 type: "thread.turn.start",
83 commandId: CommandId.makeUnsafe("cmd-turn-start-1"),
84 threadId: ThreadId.makeUnsafe("thread-1"),
85 message: {
86 messageId: asMessageId("msg-1"),
87 role: "user",
88 text: "hello",
89 attachments: [],
90 },
91 createdAt,
92 }),
93 );
94
95 const readModelA = await system.run(engine.getReadModel());
96 const readModelB = await system.run(engine.getReadModel());
97 expect(readModelB).toEqual(readModelA);
98 await system.dispose();
99 });
100
101 it("replays append-only events from sequence", async () => {
102 const system = await createOrchestrationSystem();
103 const { engine } = system;
104 const createdAt = now();
105
106 await system.run(
107 engine.dispatch({
108 type: "project.create",
109 commandId: CommandId.makeUnsafe("cmd-project-replay-create"),
110 projectId: asProjectId("project-replay"),
111 title: "Replay Project",
112 workspaceRoot: "/tmp/project-replay",
113 defaultModel: "gpt-5-codex",
114 createdAt,
115 }),
116 );
117 await system.run(
118 engine.dispatch({
119 type: "thread.create",
120 commandId: CommandId.makeUnsafe("cmd-thread-replay-create"),
121 threadId: ThreadId.makeUnsafe("thread-replay"),
122 projectId: asProjectId("project-replay"),
123 title: "replay",
124 model: "gpt-5-codex",
125 branch: null,
126 worktreePath: null,
127 createdAt,
128 }),
129 );
130 await system.run(
131 engine.dispatch({
132 type: "thread.delete",
133 commandId: CommandId.makeUnsafe("cmd-thread-replay-delete"),
134 threadId: ThreadId.makeUnsafe("thread-replay"),
135 }),
136 );
137
138 const events = await system.run(
139 Stream.runCollect(engine.readEvents(0)).pipe(
140 Effect.map((chunk): OrchestrationEvent[] => Array.from(chunk)),
141 ),
142 );
143 expect(events.map((event) => event.type)).toEqual([
144 "project.created",
145 "thread.created",
146 "thread.deleted",
147 ]);
148 await system.dispose();
149 });
150
151 it("streams persisted domain events in order", async () => {
152 const system = await createOrchestrationSystem();
153 const { engine } = system;
154 const createdAt = now();
155
156 await system.run(
157 engine.dispatch({
158 type: "project.create",
159 commandId: CommandId.makeUnsafe("cmd-project-stream-create"),
160 projectId: asProjectId("project-stream"),
161 title: "Stream Project",
162 workspaceRoot: "/tmp/project-stream",
163 defaultModel: "gpt-5-codex",
164 createdAt,
165 }),
166 );
167
168 const eventTypes: string[] = [];
169 await system.run(
170 Effect.gen(function* () {
171 const eventQueue = yield* Queue.unbounded<OrchestrationEvent>();
172 yield* Effect.forkScoped(
173 Stream.take(engine.streamDomainEvents, 2).pipe(
174 Stream.runForEach((event) => Queue.offer(eventQueue, event).pipe(Effect.asVoid)),
175 ),
176 );
177 yield* Effect.sleep("10 millis");
178 yield* engine.dispatch({
179 type: "thread.create",
180 commandId: CommandId.makeUnsafe("cmd-stream-thread-create"),
181 threadId: ThreadId.makeUnsafe("thread-stream"),
182 projectId: asProjectId("project-stream"),
183 title: "domain-stream",
184 model: "gpt-5-codex",
185 branch: null,
186 worktreePath: null,
187 createdAt,
188 });
189 yield* engine.dispatch({
190 type: "thread.meta.update",
191 commandId: CommandId.makeUnsafe("cmd-stream-thread-update"),
192 threadId: ThreadId.makeUnsafe("thread-stream"),
193 title: "domain-stream-updated",
194 });
195 eventTypes.push((yield* Queue.take(eventQueue)).type);
196 eventTypes.push((yield* Queue.take(eventQueue)).type);
197 }).pipe(Effect.scoped),
198 );
199
200 expect(eventTypes).toEqual(["thread.created", "thread.meta-updated"]);
201 await system.dispose();
202 });
203
204 it("stores completed checkpoint summaries even when no files changed", async () => {
205 const system = await createOrchestrationSystem();
206 const { engine } = system;
207 const createdAt = now();
208
209 await system.run(
210 engine.dispatch({
211 type: "project.create",
212 commandId: CommandId.makeUnsafe("cmd-project-turn-diff-create"),
213 projectId: asProjectId("project-turn-diff"),
214 title: "Turn Diff Project",
215 workspaceRoot: "/tmp/project-turn-diff",
216 defaultModel: "gpt-5-codex",
217 createdAt,
218 }),
219 );
220 await system.run(
221 engine.dispatch({
222 type: "thread.create",
223 commandId: CommandId.makeUnsafe("cmd-thread-turn-diff-create"),
224 threadId: ThreadId.makeUnsafe("thread-turn-diff"),
225 projectId: asProjectId("project-turn-diff"),
226 title: "Turn diff thread",
227 model: "gpt-5-codex",
228 branch: null,
229 worktreePath: null,
230 createdAt,
231 }),
232 );
233 await system.run(
234 engine.dispatch({
235 type: "thread.turn.diff.complete",
236 commandId: CommandId.makeUnsafe("cmd-turn-diff-complete"),
237 threadId: ThreadId.makeUnsafe("thread-turn-diff"),
238 turnId: asTurnId("turn-1"),
239 completedAt: createdAt,
240 checkpointRef: asCheckpointRef("refs/t3/checkpoints/thread-turn-diff/turn/1"),
241 status: "ready",
242 files: [],
243 checkpointTurnCount: 1,
244 createdAt,
245 }),
246 );
247
248 const thread = (await system.run(engine.getReadModel())).threads.find(
249 (entry) => entry.id === "thread-turn-diff",
250 );
251 expect(thread?.checkpoints).toEqual([
252 {
253 turnId: asTurnId("turn-1"),
254 checkpointTurnCount: 1,
255 checkpointRef: asCheckpointRef("refs/t3/checkpoints/thread-turn-diff/turn/1"),
256 status: "ready",
257 files: [],
258 assistantMessageId: null,
259 completedAt: createdAt,
260 },
261 ]);
262 await system.dispose();
263 });
264
265 it("keeps processing queued commands after a storage failure", async () => {
266 type StoredEvent =
267 ReturnType<OrchestrationEventStoreShape["append"]> extends Effect.Effect<infer A, any, any>
268 ? A
269 : never;
270 const events: StoredEvent[] = [];
271 let nextSequence = 1;
272 let shouldFailFirstAppend = true;
273
274 const flakyStore: OrchestrationEventStoreShape = {
275 append(event) {
276 if (shouldFailFirstAppend && event.commandId === CommandId.makeUnsafe("cmd-flaky-1")) {
277 shouldFailFirstAppend = false;
278 return Effect.fail(
279 new PersistenceSqlError({
280 operation: "test.append",
281 detail: "append failed",
282 }),
283 );
284 }
285 const savedEvent = {
286 ...event,
287 sequence: nextSequence,
288 } as StoredEvent;
289 nextSequence += 1;
290 events.push(savedEvent);
291 return Effect.succeed(savedEvent);
292 },
293 readFromSequence(sequenceExclusive) {
294 return Stream.fromIterable(events.filter((event) => event.sequence > sequenceExclusive));
295 },
296 readAll() {
297 return Stream.fromIterable(events);
298 },
299 };
300
301 const runtime = ManagedRuntime.make(
302 OrchestrationEngineLive.pipe(
303 Layer.provide(OrchestrationProjectionPipelineLive),
304 Layer.provide(Layer.succeed(OrchestrationEventStore, flakyStore)),
305 Layer.provide(OrchestrationCommandReceiptRepositoryLive),
306 Layer.provide(SqlitePersistenceMemory),
307 ),
308 );
309 const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService));
310 const createdAt = now();
311
312 await runtime.runPromise(
313 engine.dispatch({
314 type: "project.create",
315 commandId: CommandId.makeUnsafe("cmd-project-flaky-create"),
316 projectId: asProjectId("project-flaky"),
317 title: "Flaky Project",
318 workspaceRoot: "/tmp/project-flaky",
319 defaultModel: "gpt-5-codex",
320 createdAt,
321 }),
322 );
323
324 await expect(
325 runtime.runPromise(
326 engine.dispatch({
327 type: "thread.create",
328 commandId: CommandId.makeUnsafe("cmd-flaky-1"),
329 threadId: ThreadId.makeUnsafe("thread-flaky-fail"),
330 projectId: asProjectId("project-flaky"),
331 title: "flaky-fail",
332 model: "gpt-5-codex",
333 branch: null,
334 worktreePath: null,
335 createdAt,
336 }),
337 ),
338 ).rejects.toThrow("append failed");
339
340 const result = await runtime.runPromise(
341 engine.dispatch({
342 type: "thread.create",
343 commandId: CommandId.makeUnsafe("cmd-flaky-2"),
344 threadId: ThreadId.makeUnsafe("thread-flaky-ok"),
345 projectId: asProjectId("project-flaky"),
346 title: "flaky-ok",
347 model: "gpt-5-codex",
348 branch: null,
349 worktreePath: null,
350 createdAt,
351 }),
352 );
353
354 expect(result.sequence).toBe(2);
355 expect((await runtime.runPromise(engine.getReadModel())).snapshotSequence).toBe(2);
356 await runtime.dispose();
357 });
358
359 it("fails command dispatch when command invariants are violated", async () => {
360 const system = await createOrchestrationSystem();
361 const { engine } = system;
362
363 await expect(
364 system.run(
365 engine.dispatch({
366 type: "thread.turn.start",
367 commandId: CommandId.makeUnsafe("cmd-invariant-missing-thread"),
368 threadId: ThreadId.makeUnsafe("thread-missing"),
369 message: {
370 messageId: asMessageId("msg-missing"),
371 role: "user",
372 text: "hello",
373 attachments: [],
374 },
375 createdAt: now(),
376 }),
377 ),
378 ).rejects.toThrow("Thread 'thread-missing' does not exist");
379
380 await system.dispose();
381 });
382
383 it("rejects duplicate thread creation", async () => {
384 const system = await createOrchestrationSystem();
385 const { engine } = system;
386 const createdAt = now();
387
388 await system.run(
389 engine.dispatch({
390 type: "project.create",
391 commandId: CommandId.makeUnsafe("cmd-project-duplicate-create"),
392 projectId: asProjectId("project-duplicate"),
393 title: "Duplicate Project",
394 workspaceRoot: "/tmp/project-duplicate",
395 defaultModel: "gpt-5-codex",
396 createdAt,
397 }),
398 );
399
400 await system.run(
401 engine.dispatch({
402 type: "thread.create",
403 commandId: CommandId.makeUnsafe("cmd-thread-duplicate-1"),
404 threadId: ThreadId.makeUnsafe("thread-duplicate"),
405 projectId: asProjectId("project-duplicate"),
406 title: "duplicate",
407 model: "gpt-5-codex",
408 branch: null,
409 worktreePath: null,
410 createdAt,
411 }),
412 );
413
414 await expect(
415 system.run(
416 engine.dispatch({
417 type: "thread.create",
418 commandId: CommandId.makeUnsafe("cmd-thread-duplicate-2"),
419 threadId: ThreadId.makeUnsafe("thread-duplicate"),
420 projectId: asProjectId("project-duplicate"),
421 title: "duplicate",
422 model: "gpt-5-codex",
423 branch: null,
424 worktreePath: null,
425 createdAt,
426 }),
427 ),
428 ).rejects.toThrow("already exists");
429
430 await system.dispose();
431 });
432});
apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts created+543
......@@ -0,0 +1,543 @@
1import {
2 CheckpointRef,
3 CommandId,
4 MessageId,
5 ProjectId,
6 ThreadId,
7 TurnId,
8 type OrchestrationEvent,
9} from "@t3tools/contracts";
10import { Effect, Layer, ManagedRuntime, Queue, Stream } from "effect";
11import { describe, expect, it } from "vitest";
12
13import { PersistenceSqlError } from "../../persistence/Errors.ts";
14import { OrchestrationCommandReceiptRepositoryLive } from "../../persistence/Layers/OrchestrationCommandReceipts.ts";
15import { OrchestrationEventStoreLive } from "../../persistence/Layers/OrchestrationEventStore.ts";
16import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts";
17import {
18 OrchestrationEventStore,
19 type OrchestrationEventStoreShape,
20} from "../../persistence/Services/OrchestrationEventStore.ts";
21import { OrchestrationEngineLive } from "./OrchestrationEngine.ts";
22import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts";
23import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts";
24import {
25 OrchestrationProjectionPipeline,
26 type OrchestrationProjectionPipelineShape,
27} from "../Services/ProjectionPipeline.ts";
28
29const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value);
30const asMessageId = (value: string): MessageId => MessageId.makeUnsafe(value);
31const asTurnId = (value: string): TurnId => TurnId.makeUnsafe(value);
32const asCheckpointRef = (value: string): CheckpointRef => CheckpointRef.makeUnsafe(value);
33
34async function createOrchestrationSystem() {
35 const orchestrationLayer = OrchestrationEngineLive.pipe(
36 Layer.provide(OrchestrationProjectionPipelineLive),
37 Layer.provide(OrchestrationEventStoreLive),
38 Layer.provide(OrchestrationCommandReceiptRepositoryLive),
39 Layer.provide(SqlitePersistenceMemory),
40 );
41 const runtime = ManagedRuntime.make(orchestrationLayer);
42 const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService));
43 return {
44 engine,
45 run: <A, E>(effect: Effect.Effect<A, E>) => runtime.runPromise(effect),
46 dispose: () => runtime.dispose(),
47 };
48}
49
50function now() {
51 return new Date().toISOString();
52}
53
54describe("OrchestrationEngine", () => {
55 it("returns deterministic read models for repeated reads", async () => {
56 const createdAt = now();
57 const system = await createOrchestrationSystem();
58 const { engine } = system;
59
60 await system.run(
61 engine.dispatch({
62 type: "project.create",
63 commandId: CommandId.makeUnsafe("cmd-project-1-create"),
64 projectId: asProjectId("project-1"),
65 title: "Project 1",
66 workspaceRoot: "/tmp/project-1",
67 defaultModel: "gpt-5-codex",
68 createdAt,
69 }),
70 );
71 await system.run(
72 engine.dispatch({
73 type: "thread.create",
74 commandId: CommandId.makeUnsafe("cmd-thread-1-create"),
75 threadId: ThreadId.makeUnsafe("thread-1"),
76 projectId: asProjectId("project-1"),
77 title: "Thread",
78 model: "gpt-5-codex",
79 branch: null,
80 worktreePath: null,
81 createdAt,
82 }),
83 );
84 await system.run(
85 engine.dispatch({
86 type: "thread.turn.start",
87 commandId: CommandId.makeUnsafe("cmd-turn-start-1"),
88 threadId: ThreadId.makeUnsafe("thread-1"),
89 message: {
90 messageId: asMessageId("msg-1"),
91 role: "user",
92 text: "hello",
93 attachments: [],
94 },
95 createdAt,
96 }),
97 );
98
99 const readModelA = await system.run(engine.getReadModel());
100 const readModelB = await system.run(engine.getReadModel());
101 expect(readModelB).toEqual(readModelA);
102 await system.dispose();
103 });
104
105 it("replays append-only events from sequence", async () => {
106 const system = await createOrchestrationSystem();
107 const { engine } = system;
108 const createdAt = now();
109
110 await system.run(
111 engine.dispatch({
112 type: "project.create",
113 commandId: CommandId.makeUnsafe("cmd-project-replay-create"),
114 projectId: asProjectId("project-replay"),
115 title: "Replay Project",
116 workspaceRoot: "/tmp/project-replay",
117 defaultModel: "gpt-5-codex",
118 createdAt,
119 }),
120 );
121 await system.run(
122 engine.dispatch({
123 type: "thread.create",
124 commandId: CommandId.makeUnsafe("cmd-thread-replay-create"),
125 threadId: ThreadId.makeUnsafe("thread-replay"),
126 projectId: asProjectId("project-replay"),
127 title: "replay",
128 model: "gpt-5-codex",
129 branch: null,
130 worktreePath: null,
131 createdAt,
132 }),
133 );
134 await system.run(
135 engine.dispatch({
136 type: "thread.delete",
137 commandId: CommandId.makeUnsafe("cmd-thread-replay-delete"),
138 threadId: ThreadId.makeUnsafe("thread-replay"),
139 }),
140 );
141
142 const events = await system.run(
143 Stream.runCollect(engine.readEvents(0)).pipe(
144 Effect.map((chunk): OrchestrationEvent[] => Array.from(chunk)),
145 ),
146 );
147 expect(events.map((event) => event.type)).toEqual([
148 "project.created",
149 "thread.created",
150 "thread.deleted",
151 ]);
152 await system.dispose();
153 });
154
155 it("streams persisted domain events in order", async () => {
156 const system = await createOrchestrationSystem();
157 const { engine } = system;
158 const createdAt = now();
159
160 await system.run(
161 engine.dispatch({
162 type: "project.create",
163 commandId: CommandId.makeUnsafe("cmd-project-stream-create"),
164 projectId: asProjectId("project-stream"),
165 title: "Stream Project",
166 workspaceRoot: "/tmp/project-stream",
167 defaultModel: "gpt-5-codex",
168 createdAt,
169 }),
170 );
171
172 const eventTypes: string[] = [];
173 await system.run(
174 Effect.gen(function* () {
175 const eventQueue = yield* Queue.unbounded<OrchestrationEvent>();
176 yield* Effect.forkScoped(
177 Stream.take(engine.streamDomainEvents, 2).pipe(
178 Stream.runForEach((event) => Queue.offer(eventQueue, event).pipe(Effect.asVoid)),
179 ),
180 );
181 yield* Effect.sleep("10 millis");
182 yield* engine.dispatch({
183 type: "thread.create",
184 commandId: CommandId.makeUnsafe("cmd-stream-thread-create"),
185 threadId: ThreadId.makeUnsafe("thread-stream"),
186 projectId: asProjectId("project-stream"),
187 title: "domain-stream",
188 model: "gpt-5-codex",
189 branch: null,
190 worktreePath: null,
191 createdAt,
192 });
193 yield* engine.dispatch({
194 type: "thread.meta.update",
195 commandId: CommandId.makeUnsafe("cmd-stream-thread-update"),
196 threadId: ThreadId.makeUnsafe("thread-stream"),
197 title: "domain-stream-updated",
198 });
199 eventTypes.push((yield* Queue.take(eventQueue)).type);
200 eventTypes.push((yield* Queue.take(eventQueue)).type);
201 }).pipe(Effect.scoped),
202 );
203
204 expect(eventTypes).toEqual(["thread.created", "thread.meta-updated"]);
205 await system.dispose();
206 });
207
208 it("stores completed checkpoint summaries even when no files changed", async () => {
209 const system = await createOrchestrationSystem();
210 const { engine } = system;
211 const createdAt = now();
212
213 await system.run(
214 engine.dispatch({
215 type: "project.create",
216 commandId: CommandId.makeUnsafe("cmd-project-turn-diff-create"),
217 projectId: asProjectId("project-turn-diff"),
218 title: "Turn Diff Project",
219 workspaceRoot: "/tmp/project-turn-diff",
220 defaultModel: "gpt-5-codex",
221 createdAt,
222 }),
223 );
224 await system.run(
225 engine.dispatch({
226 type: "thread.create",
227 commandId: CommandId.makeUnsafe("cmd-thread-turn-diff-create"),
228 threadId: ThreadId.makeUnsafe("thread-turn-diff"),
229 projectId: asProjectId("project-turn-diff"),
230 title: "Turn diff thread",
231 model: "gpt-5-codex",
232 branch: null,
233 worktreePath: null,
234 createdAt,
235 }),
236 );
237 await system.run(
238 engine.dispatch({
239 type: "thread.turn.diff.complete",
240 commandId: CommandId.makeUnsafe("cmd-turn-diff-complete"),
241 threadId: ThreadId.makeUnsafe("thread-turn-diff"),
242 turnId: asTurnId("turn-1"),
243 completedAt: createdAt,
244 checkpointRef: asCheckpointRef("refs/t3/checkpoints/thread-turn-diff/turn/1"),
245 status: "ready",
246 files: [],
247 checkpointTurnCount: 1,
248 createdAt,
249 }),
250 );
251
252 const thread = (await system.run(engine.getReadModel())).threads.find(
253 (entry) => entry.id === "thread-turn-diff",
254 );
255 expect(thread?.checkpoints).toEqual([
256 {
257 turnId: asTurnId("turn-1"),
258 checkpointTurnCount: 1,
259 checkpointRef: asCheckpointRef("refs/t3/checkpoints/thread-turn-diff/turn/1"),
260 status: "ready",
261 files: [],
262 assistantMessageId: null,
263 completedAt: createdAt,
264 },
265 ]);
266 await system.dispose();
267 });
268
269 it("keeps processing queued commands after a storage failure", async () => {
270 type StoredEvent =
271 ReturnType<OrchestrationEventStoreShape["append"]> extends Effect.Effect<infer A, any, any>
272 ? A
273 : never;
274 const events: StoredEvent[] = [];
275 let nextSequence = 1;
276 let shouldFailFirstAppend = true;
277
278 const flakyStore: OrchestrationEventStoreShape = {
279 append(event) {
280 if (shouldFailFirstAppend && event.commandId === CommandId.makeUnsafe("cmd-flaky-1")) {
281 shouldFailFirstAppend = false;
282 return Effect.fail(
283 new PersistenceSqlError({
284 operation: "test.append",
285 detail: "append failed",
286 }),
287 );
288 }
289 const savedEvent = {
290 ...event,
291 sequence: nextSequence,
292 } as StoredEvent;
293 nextSequence += 1;
294 events.push(savedEvent);
295 return Effect.succeed(savedEvent);
296 },
297 readFromSequence(sequenceExclusive) {
298 return Stream.fromIterable(events.filter((event) => event.sequence > sequenceExclusive));
299 },
300 readAll() {
301 return Stream.fromIterable(events);
302 },
303 };
304
305 const runtime = ManagedRuntime.make(
306 OrchestrationEngineLive.pipe(
307 Layer.provide(OrchestrationProjectionPipelineLive),
308 Layer.provide(Layer.succeed(OrchestrationEventStore, flakyStore)),
309 Layer.provide(OrchestrationCommandReceiptRepositoryLive),
310 Layer.provide(SqlitePersistenceMemory),
311 ),
312 );
313 const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService));
314 const createdAt = now();
315
316 await runtime.runPromise(
317 engine.dispatch({
318 type: "project.create",
319 commandId: CommandId.makeUnsafe("cmd-project-flaky-create"),
320 projectId: asProjectId("project-flaky"),
321 title: "Flaky Project",
322 workspaceRoot: "/tmp/project-flaky",
323 defaultModel: "gpt-5-codex",
324 createdAt,
325 }),
326 );
327
328 await expect(
329 runtime.runPromise(
330 engine.dispatch({
331 type: "thread.create",
332 commandId: CommandId.makeUnsafe("cmd-flaky-1"),
333 threadId: ThreadId.makeUnsafe("thread-flaky-fail"),
334 projectId: asProjectId("project-flaky"),
335 title: "flaky-fail",
336 model: "gpt-5-codex",
337 branch: null,
338 worktreePath: null,
339 createdAt,
340 }),
341 ),
342 ).rejects.toThrow("append failed");
343
344 const result = await runtime.runPromise(
345 engine.dispatch({
346 type: "thread.create",
347 commandId: CommandId.makeUnsafe("cmd-flaky-2"),
348 threadId: ThreadId.makeUnsafe("thread-flaky-ok"),
349 projectId: asProjectId("project-flaky"),
350 title: "flaky-ok",
351 model: "gpt-5-codex",
352 branch: null,
353 worktreePath: null,
354 createdAt,
355 }),
356 );
357
358 expect(result.sequence).toBe(2);
359 expect((await runtime.runPromise(engine.getReadModel())).snapshotSequence).toBe(2);
360 await runtime.dispose();
361 });
362
363 it("rolls back all events for a multi-event command when projection fails mid-dispatch", async () => {
364 let shouldFailRequestedProjection = true;
365 const flakyProjectionPipeline: OrchestrationProjectionPipelineShape = {
366 bootstrap: Effect.void,
367 projectEvent: (event) => {
368 if (
369 shouldFailRequestedProjection &&
370 event.commandId === CommandId.makeUnsafe("cmd-turn-start-atomic") &&
371 event.type === "thread.turn-start-requested"
372 ) {
373 shouldFailRequestedProjection = false;
374 return Effect.fail(
375 new PersistenceSqlError({
376 operation: "test.projection",
377 detail: "projection failed",
378 }),
379 );
380 }
381 return Effect.void;
382 },
383 };
384
385 const runtime = ManagedRuntime.make(
386 OrchestrationEngineLive.pipe(
387 Layer.provide(Layer.succeed(OrchestrationProjectionPipeline, flakyProjectionPipeline)),
388 Layer.provide(OrchestrationEventStoreLive),
389 Layer.provide(OrchestrationCommandReceiptRepositoryLive),
390 Layer.provide(SqlitePersistenceMemory),
391 ),
392 );
393 const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService));
394 const createdAt = now();
395
396 await runtime.runPromise(
397 engine.dispatch({
398 type: "project.create",
399 commandId: CommandId.makeUnsafe("cmd-project-atomic-create"),
400 projectId: asProjectId("project-atomic"),
401 title: "Atomic Project",
402 workspaceRoot: "/tmp/project-atomic",
403 defaultModel: "gpt-5-codex",
404 createdAt,
405 }),
406 );
407 await runtime.runPromise(
408 engine.dispatch({
409 type: "thread.create",
410 commandId: CommandId.makeUnsafe("cmd-thread-atomic-create"),
411 threadId: ThreadId.makeUnsafe("thread-atomic"),
412 projectId: asProjectId("project-atomic"),
413 title: "atomic",
414 model: "gpt-5-codex",
415 branch: null,
416 worktreePath: null,
417 createdAt,
418 }),
419 );
420
421 const turnStartCommand = {
422 type: "thread.turn.start" as const,
423 commandId: CommandId.makeUnsafe("cmd-turn-start-atomic"),
424 threadId: ThreadId.makeUnsafe("thread-atomic"),
425 message: {
426 messageId: asMessageId("msg-atomic-1"),
427 role: "user" as const,
428 text: "hello",
429 attachments: [],
430 },
431 createdAt,
432 };
433
434 await expect(runtime.runPromise(engine.dispatch(turnStartCommand))).rejects.toThrow(
435 "projection failed",
436 );
437
438 const eventsAfterFailure = await runtime.runPromise(
439 Stream.runCollect(engine.readEvents(0)).pipe(
440 Effect.map((chunk): OrchestrationEvent[] => Array.from(chunk)),
441 ),
442 );
443 expect(eventsAfterFailure.map((event) => event.type)).toEqual([
444 "project.created",
445 "thread.created",
446 ]);
447 expect((await runtime.runPromise(engine.getReadModel())).snapshotSequence).toBe(2);
448
449 const retryResult = await runtime.runPromise(engine.dispatch(turnStartCommand));
450 expect(retryResult.sequence).toBe(4);
451
452 const eventsAfterRetry = await runtime.runPromise(
453 Stream.runCollect(engine.readEvents(0)).pipe(
454 Effect.map((chunk): OrchestrationEvent[] => Array.from(chunk)),
455 ),
456 );
457 expect(eventsAfterRetry.map((event) => event.type)).toEqual([
458 "project.created",
459 "thread.created",
460 "thread.message-sent",
461 "thread.turn-start-requested",
462 ]);
463 expect(
464 eventsAfterRetry.filter((event) => event.commandId === turnStartCommand.commandId),
465 ).toHaveLength(2);
466
467 await runtime.dispose();
468 });
469
470 it("fails command dispatch when command invariants are violated", async () => {
471 const system = await createOrchestrationSystem();
472 const { engine } = system;
473
474 await expect(
475 system.run(
476 engine.dispatch({
477 type: "thread.turn.start",
478 commandId: CommandId.makeUnsafe("cmd-invariant-missing-thread"),
479 threadId: ThreadId.makeUnsafe("thread-missing"),
480 message: {
481 messageId: asMessageId("msg-missing"),
482 role: "user",
483 text: "hello",
484 attachments: [],
485 },
486 createdAt: now(),
487 }),
488 ),
489 ).rejects.toThrow("Thread 'thread-missing' does not exist");
490
491 await system.dispose();
492 });
493
494 it("rejects duplicate thread creation", async () => {
495 const system = await createOrchestrationSystem();
496 const { engine } = system;
497 const createdAt = now();
498
499 await system.run(
500 engine.dispatch({
501 type: "project.create",
502 commandId: CommandId.makeUnsafe("cmd-project-duplicate-create"),
503 projectId: asProjectId("project-duplicate"),
504 title: "Duplicate Project",
505 workspaceRoot: "/tmp/project-duplicate",
506 defaultModel: "gpt-5-codex",
507 createdAt,
508 }),
509 );
510
511 await system.run(
512 engine.dispatch({
513 type: "thread.create",
514 commandId: CommandId.makeUnsafe("cmd-thread-duplicate-1"),
515 threadId: ThreadId.makeUnsafe("thread-duplicate"),
516 projectId: asProjectId("project-duplicate"),
517 title: "duplicate",
518 model: "gpt-5-codex",
519 branch: null,
520 worktreePath: null,
521 createdAt,
522 }),
523 );
524
525 await expect(
526 system.run(
527 engine.dispatch({
528 type: "thread.create",
529 commandId: CommandId.makeUnsafe("cmd-thread-duplicate-2"),
530 threadId: ThreadId.makeUnsafe("thread-duplicate"),
531 projectId: asProjectId("project-duplicate"),
532 title: "duplicate",
533 model: "gpt-5-codex",
534 branch: null,
535 worktreePath: null,
536 createdAt,
537 }),
538 ),
539 ).rejects.toThrow("already exists");
540
541 await system.dispose();
542 });
543});
apps/server/src/orchestration/Layers/OrchestrationEngine.ts+51-27
......@@ -6,7 +6,9 @@ import type {
66} from "@t3tools/contracts";
77import { OrchestrationCommand } from "@t3tools/contracts";
88import { Deferred, Effect, Layer, Option, PubSub, Queue, Stream } from "effect";
9import * as SqlClient from "effect/unstable/sql/SqlClient";
910
11import { toPersistenceSqlError } from "../../persistence/Errors.ts";
1012import { OrchestrationEventStore } from "../../persistence/Services/OrchestrationEventStore.ts";
1113import { OrchestrationCommandReceiptRepository } from "../../persistence/Services/OrchestrationCommandReceipts.ts";
1214import {
......@@ -55,6 +57,7 @@ function formatDispatchError(error: OrchestrationDispatchError): string {
5557}
5658
5759const makeOrchestrationEngine = Effect.gen(function* () {
60 const sql = yield* SqlClient.SqlClient;
5861 const eventStore = yield* OrchestrationEventStore;
5962 const commandReceiptRepository = yield* OrchestrationCommandReceiptRepository;
6063 const projectionPipeline = yield* OrchestrationProjectionPipeline;
......@@ -91,36 +94,57 @@ const makeOrchestrationEngine = Effect.gen(function* () {
9194 readModel,
9295 });
9396 const eventBases = Array.isArray(eventBase) ? eventBase : [eventBase];
94 let lastSavedEvent = null as OrchestrationEvent | null;
95
96 for (const nextEvent of eventBases) {
97 const savedEvent = yield* eventStore.append(nextEvent);
98 yield* projectionPipeline.projectEvent(savedEvent);
99 readModel = yield* projectEvent(readModel, savedEvent);
100 yield* PubSub.publish(eventPubSub, savedEvent);
101 lastSavedEvent = savedEvent;
102 }
103
104 if (lastSavedEvent === null) {
105 return yield* Deferred.fail(
106 envelope.result,
107 new OrchestrationCommandInvariantError({
108 commandType: envelope.command.type,
109 detail: "Command produced no events.",
97 const committedCommand = yield* sql
98 .withTransaction(
99 Effect.gen(function* () {
100 const committedEvents: OrchestrationEvent[] = [];
101 let nextReadModel = readModel;
102
103 for (const nextEvent of eventBases) {
104 const savedEvent = yield* eventStore.append(nextEvent);
105 yield* projectionPipeline.projectEvent(savedEvent);
106 nextReadModel = yield* projectEvent(nextReadModel, savedEvent);
107 committedEvents.push(savedEvent);
108 }
109
110 const lastSavedEvent = committedEvents.at(-1) ?? null;
111 if (lastSavedEvent === null) {
112 return yield* new OrchestrationCommandInvariantError({
113 commandType: envelope.command.type,
114 detail: "Command produced no events.",
115 });
116 }
117
118 yield* commandReceiptRepository.upsert({
119 commandId: envelope.command.commandId,
120 aggregateKind: lastSavedEvent.aggregateKind,
121 aggregateId: lastSavedEvent.aggregateId,
122 acceptedAt: lastSavedEvent.occurredAt,
123 resultSequence: lastSavedEvent.sequence,
124 status: "accepted",
125 error: null,
126 });
127
128 return {
129 committedEvents,
130 lastSequence: lastSavedEvent.sequence,
131 nextReadModel,
132 } as const;
110133 }),
134 )
135 .pipe(
136 Effect.catchTag("SqlError", (sqlError) =>
137 Effect.fail(
138 toPersistenceSqlError("OrchestrationEngine.processEnvelope:transaction")(sqlError),
139 ),
140 ),
111141 );
112 }
113142
114 yield* commandReceiptRepository.upsert({
115 commandId: envelope.command.commandId,
116 aggregateKind: lastSavedEvent.aggregateKind,
117 aggregateId: lastSavedEvent.aggregateId,
118 acceptedAt: lastSavedEvent.occurredAt,
119 resultSequence: lastSavedEvent.sequence,
120 status: "accepted",
121 error: null,
122 });
123 yield* Deferred.succeed(envelope.result, { sequence: lastSavedEvent.sequence });
143 readModel = committedCommand.nextReadModel;
144 for (const event of committedCommand.committedEvents) {
145 yield* PubSub.publish(eventPubSub, event);
146 }
147 yield* Deferred.succeed(envelope.result, { sequence: committedCommand.lastSequence });
124148 }).pipe(
125149 Effect.catch((error) =>
126150 Effect.gen(function* () {