1import {
2 CommandId,
3 EventId,
4 NodeId,
5 ProjectId,
6 ProviderDriverKind,
7 ProviderInstanceId,
8 ThreadId,
9 TurnItemId,
10 type OrchestrationV2DomainEvent,
11} from "@t3tools/contracts";
12import { assert, it } from "@effect/vitest";
13import * as DateTime from "effect/DateTime";
14import * as Deferred from "effect/Deferred";
15import * as Fiber from "effect/Fiber";
16import * as Effect from "effect/Effect";
17import * as Layer from "effect/Layer";
18import * as Schema from "effect/Schema";
19import * as Stream from "effect/Stream";
20import * as SqlClient from "effect/sql/SqlClient";
21import * as Statement from "effect/sql/Statement";
22
23import {
24 LIVE_STREAM_MAX_ITEMS,
25 LiveStreamBufferError,
26} from "../orchestration-v2/LiveStreamBudget.ts";
27import { PersistenceDecodeError } from "./Errors.ts";
28import { toShellApplicationEvent } from "../orchestration-v2/ShellStream.ts";
29import * as OrchestrationEventStore from "./OrchestrationEventStore.ts";
30import * as SqlitePersistence from "./Sqlite.ts";
31const isPersistenceDecodeError = Schema.is(PersistenceDecodeError);
32const isLiveStreamBufferError = Schema.is(LiveStreamBufferError);
33
34const layerTest = OrchestrationEventStore.layer.pipe(
35 Layer.provideMerge(SqlitePersistence.layerMemory),
36);
37const layer = it.layer(layerTest);
38
39layer("OrchestrationEventStore", (it) => {
40 it.effect("retains only shell metadata from oversized replay and live application events", () =>
41 Effect.gen(function* () {
42 const store = yield* OrchestrationEventStore.OrchestrationEventStore;
43 const afterSequence = yield* store.latestApplicationSequence;
44 const now = yield* DateTime.now;
45 const threadId = ThreadId.make("thread:large-shell-body");
46 const input = "x".repeat(9 * 1024 * 1024);
47 const event: OrchestrationV2DomainEvent = {
48 id: EventId.make("event:large-shell-body:replay"),
49 type: "turn-item.updated",
50 threadId,
51 occurredAt: now,
52 payload: {
53 id: TurnItemId.make("tool:large-shell-body"),
54 type: "command_execution",
55 threadId,
56 runId: null,
57 nodeId: null,
58 providerThreadId: null,
59 providerTurnId: null,
60 nativeItemRef: null,
61 parentItemId: null,
62 ordinal: 1,
63 status: "running",
64 title: "Large command",
65 input,
66 startedAt: now,
67 completedAt: null,
68 updatedAt: now,
69 },
70 };
71 const [replayed] = yield* store.appendAgentEvents({ events: [event] });
72 const pull = yield* Stream.toPull(
73 store.streamProjectedApplicationEvents({
74 afterSequence,
75 project: toShellApplicationEvent,
76 }),
77 );
78 assert.deepEqual(yield* pull, [{ sequence: replayed!.sequence, event: { threadId } }]);
79 const [live] = yield* store.appendAgentEvents({
80 events: [
81 {
82 ...event,
83 id: EventId.make("event:large-shell-body:live"),
84 payload: { ...event.payload, status: "completed", completedAt: now },
85 },
86 ],
87 });
88 yield* store.publishCommitted([live!]);
89 assert.deepEqual(yield* pull, [{ sequence: live!.sequence, event: { threadId } }]);
90 const durable = yield* store
91 .readAgentEvents({ threadId, afterSequence })
92 .pipe(Stream.runCollect);
93 assert.lengthOf(durable, 2);
94 for (const stored of durable) {
95 assert.equal(stored.event.type, "turn-item.updated");
96 if (stored.event.type !== "turn-item.updated") return;
97 assert.equal(stored.event.payload.type, "command_execution");
98 if (stored.event.payload.type !== "command_execution") return;
99 assert.equal(stored.event.payload.input, input);
100 }
101 }),
102 );
103
104 it.effect("stores an unchanged subagent prompt once and replays it on every update", () =>
105 Effect.gen(function* () {
106 const store = yield* OrchestrationEventStore.OrchestrationEventStore;
107 const sql = yield* SqlClient.SqlClient;
108 const afterSequence = yield* store.latestApplicationSequence;
109 const now = yield* DateTime.now;
110 const threadId = ThreadId.make("thread:subagent-prompt");
111 const id = NodeId.make("node:subagent-prompt");
112 const driver = ProviderDriverKind.make("claudeAgent");
113 const providerInstanceId = ProviderInstanceId.make("claudeAgent");
114 const prompt = "Audit the repository. ".repeat(200);
115 const subagent = (index: number, progress: string, nextPrompt = prompt) =>
116 ({
117 id: EventId.make(`event:subagent-prompt:subagent:${index}`),
118 type: "subagent.updated",
119 threadId,
120 occurredAt: now,
121 payload: {
122 id,
123 threadId,
124 runId: null,
125 parentNodeId: NodeId.make("node:subagent-prompt-parent"),
126 origin: "provider_native",
127 createdBy: "agent",
128 driver,
129 providerInstanceId,
130 providerThreadId: null,
131 childThreadId: null,
132 nativeTaskRef: { driver, nativeId: "subagent-prompt", strength: "strong" },
133 prompt: nextPrompt,
134 progress,
135 title: "Audit",
136 model: null,
137 status: "running",
138 result: null,
139 startedAt: now,
140 completedAt: null,
141 updatedAt: now,
142 },
143 }) satisfies OrchestrationV2DomainEvent;
144 const item = (index: number, progress: string, nextPrompt = prompt) =>
145 ({
146 id: EventId.make(`event:subagent-prompt:item:${index}`),
147 type: "turn-item.updated",
148 threadId,
149 occurredAt: now,
150 payload: {
151 id: TurnItemId.make("item:subagent-prompt"),
152 type: "subagent",
153 threadId,
154 runId: null,
155 nodeId: null,
156 providerThreadId: null,
157 providerTurnId: null,
158 nativeItemRef: null,
159 parentItemId: null,
160 ordinal: 1,
161 status: "running",
162 title: "Audit",
163 subagentId: id,
164 origin: "provider_native",
165 driver,
166 providerInstanceId,
167 childThreadId: null,
168 prompt: nextPrompt,
169 progress,
170 result: null,
171 startedAt: now,
172 completedAt: null,
173 updatedAt: now,
174 },
175 }) satisfies OrchestrationV2DomainEvent;
176 const events = [
177 subagent(0, "Reading files"),
178 item(0, "Reading files"),
179 subagent(1, "Running tests"),
180 item(1, "Running tests"),
181 subagent(2, "Writing report"),
182 item(2, "Writing report"),
183 subagent(3, "Retrying", "Audit again."),
184 item(3, "Retrying", "Audit again."),
185 ];
186 yield* store.appendAgentEvents({ events });
187
188 const stored = yield* sql<{ readonly payload_json: string }>`
189 SELECT payload_json FROM orchestration_events
190 WHERE stream_id = ${threadId} ORDER BY sequence ASC
191 `;
192 assert.deepEqual(
193 stored.map((row) => row.payload_json.includes(prompt)),
194 [true, true, false, false, false, false, false, false],
195 );
196 const replayed = yield* store
197 .readAgentEvents({ threadId, afterSequence })
198 .pipe(Stream.runCollect);
199 assert.deepEqual(
200 replayed.map((entry) => entry.event),
201 events,
202 );
203 const stats = yield* store.getAgentReplayStats({
204 threadId,
205 afterSequence,
206 throughSequence: yield* store.latestApplicationSequence,
207 maxEvents: events.length,
208 });
209 assert.isAbove(stats.rawPayloadBytes, prompt.length * 6);
210 const applicationStats = yield* store.getReplayStats({
211 afterSequence,
212 throughSequence: yield* store.latestApplicationSequence,
213 });
214 assert.isAbove(applicationStats.rawPayloadBytes, prompt.length * 6);
215 }),
216 );
217
218 it.effect("stores the full subagent prompt when its hint row was rolled back", () =>
219 Effect.gen(function* () {
220 const store = yield* OrchestrationEventStore.OrchestrationEventStore;
221 const sql = yield* SqlClient.SqlClient;
222 const afterSequence = yield* store.latestApplicationSequence;
223 const now = yield* DateTime.now;
224 const threadId = ThreadId.make("thread:subagent-rollback");
225 const event = {
226 id: EventId.make("event:subagent-rollback"),
227 type: "subagent.updated",
228 threadId,
229 occurredAt: now,
230 payload: {
231 id: NodeId.make("node:subagent-rollback"),
232 threadId,
233 runId: null,
234 parentNodeId: NodeId.make("node:subagent-rollback-parent"),
235 origin: "provider_native",
236 createdBy: "agent",
237 driver: ProviderDriverKind.make("claudeAgent"),
238 providerInstanceId: ProviderInstanceId.make("claudeAgent"),
239 providerThreadId: null,
240 childThreadId: null,
241 nativeTaskRef: null,
242 prompt: "Audit the repository. ".repeat(200),
243 progress: "Reading files",
244 title: "Audit",
245 model: null,
246 status: "running",
247 result: null,
248 startedAt: now,
249 completedAt: null,
250 updatedAt: now,
251 },
252 } satisfies OrchestrationV2DomainEvent;
253 const rolledBack = yield* sql
254 .withTransaction(
255 store
256 .appendAgentEvents({ events: [event] })
257 .pipe(Effect.andThen(Effect.fail("rollback"))),
258 )
259 .pipe(Effect.flip);
260 assert.equal(rolledBack, "rollback");
261 const rowsAfterRollback = yield* sql`
262 SELECT sequence FROM orchestration_events WHERE stream_id = ${threadId}
263 `;
264 assert.deepEqual(rowsAfterRollback, []);
265
266 const events = [event, { ...event, id: EventId.make("event:subagent-after-rollback") }];
267 const appended = yield* store.appendAgentEvents({ events });
268 assert.deepEqual(
269 appended.map((entry) => entry.event),
270 events,
271 );
272 const stored = yield* sql<{
273 readonly prompt: string | null;
274 readonly prompt_sequence: number | null;
275 }>`
276 SELECT json_extract(payload_json, '$.prompt') AS prompt,
277 json_extract(payload_json, '$.promptSequence') AS prompt_sequence
278 FROM orchestration_events WHERE stream_id = ${threadId} ORDER BY sequence
279 `;
280 assert.deepEqual(stored, [
281 { prompt: event.payload.prompt, prompt_sequence: null },
282 { prompt: "", prompt_sequence: appended[0]!.sequence },
283 ]);
284 const replayed = yield* store.readAgentEvents({ threadId }).pipe(Stream.runCollect);
285 assert.deepEqual(replayed, appended);
286 const applicationEvents = yield* store
287 .readApplicationEvents({
288 afterSequence,
289 throughSequence: appended[1]!.sequence,
290 })
291 .pipe(Stream.runCollect);
292 assert.deepEqual(applicationEvents, [...appended]);
293 }),
294 );
295
296 it.effect("orders project and V2 agent events in the retained application event source", () =>
297 Effect.gen(function* () {
298 const eventStore = yield* OrchestrationEventStore.OrchestrationEventStore;
299 const projectId = ProjectId.make("project-shared-stream");
300 const threadId = ThreadId.make("thread-shared-stream");
301 const providerInstanceId = ProviderInstanceId.make("codex");
302 const occurredAt = DateTime.makeUnsafe("2026-01-02T00:00:00.000Z");
303 const now = DateTime.formatIso(occurredAt);
304 const baselineSequence = yield* eventStore.latestApplicationSequence;
305
306 const projectEvent = yield* eventStore.appendProjectEvent({
307 type: "project.created",
308 eventId: EventId.make("event-project-shared-stream"),
309 aggregateKind: "project",
310 aggregateId: projectId,
311 occurredAt: now,
312 commandId: CommandId.make("command-project-shared-stream"),
313 causationEventId: null,
314 correlationId: CommandId.make("command-project-shared-stream"),
315 metadata: {},
316 payload: {
317 projectId,
318 title: "Shared stream",
319 workspaceRoot: "/tmp/shared-stream",
320 defaultModelSelection: null,
321 scripts: [],
322 createdAt: now,
323 updatedAt: now,
324 },
325 });
326 const [threadEvent] = yield* eventStore.appendAgentEvents({
327 commandId: CommandId.make("command-thread-shared-stream"),
328 events: [
329 {
330 id: EventId.make("event-thread-shared-stream"),
331 type: "thread.created",
332 threadId,
333 providerInstanceId,
334 occurredAt,
335 payload: {
336 id: threadId,
337 projectId,
338 title: "Thread",
339 providerInstanceId,
340 modelSelection: { instanceId: providerInstanceId, model: "gpt-5.4" },
341 runtimeMode: "full-access",
342 interactionMode: "default",
343 branch: null,
344 worktreePath: null,
345 activeProviderThreadId: null,
346 lineage: {
347 rootThreadId: threadId,
348 parentThreadId: null,
349 relationshipToParent: null,
350 },
351 forkedFrom: null,
352 createdBy: "user",
353 creationSource: "web",
354 createdAt: occurredAt,
355 updatedAt: occurredAt,
356 archivedAt: null,
357 settledOverride: null,
358 settledAt: null,
359 lastVisitedAt: null,
360 deletedAt: null,
361 },
362 },
363 ],
364 });
365
366 const applicationEvents = yield* eventStore
367 .streamApplicationEvents({ afterSequence: baselineSequence })
368 .pipe(
369 Stream.take(2),
370 Stream.runCollect,
371 Effect.map((chunk) => Array.from(chunk)),
372 );
373 assert.deepEqual(
374 applicationEvents.map((event) => event.sequence),
375 [projectEvent.sequence, threadEvent!.sequence],
376 );
377 assert.isTrue("aggregateKind" in applicationEvents[0]!);
378 assert.isTrue("event" in applicationEvents[1]!);
379
380 const finiteReplay = yield* eventStore
381 .readApplicationEvents({
382 afterSequence: baselineSequence,
383 throughSequence: threadEvent!.sequence,
384 })
385 .pipe(
386 Stream.runCollect,
387 Effect.map((chunk) => Array.from(chunk)),
388 );
389 assert.deepEqual(
390 finiteReplay.map((event) => event.sequence),
391 [projectEvent.sequence, threadEvent!.sequence],
392 );
393 }),
394 );
395 it.effect("measures only a bounded thread replay before decoding its payloads", () =>
396 Effect.gen(function* () {
397 const store = yield* OrchestrationEventStore.OrchestrationEventStore;
398 const sql = yield* SqlClient.SqlClient;
399 const threadId = ThreadId.make("measured-replay-thread");
400 const baseline = yield* store.latestApplicationSequence;
401 const unicodePayload = '{"text":"🦊"}';
402 const oversizedPayload = "{" + "x".repeat(1_048_576);
403 let streamVersion = 0;
404 const insert = Effect.fn(function* (input: {
405 readonly id: string;
406 readonly payload: string;
407 readonly type?: string;
408 readonly threadId?: string;
409 readonly version?: number;
410 }) {
411 const rows = yield* sql<{ readonly sequence: number }>`
412 INSERT INTO orchestration_events (
413 event_id, aggregate_kind, stream_id, stream_version, event_type,
414 occurred_at, actor_kind, payload_json, metadata_json, application_event_version
415 ) VALUES (
416 ${input.id}, 'thread', ${input.threadId ?? threadId},
417 ${streamVersion++}, ${input.type ?? "thread.metadata-updated"},
418 '2026-09-04T00:00:00.000Z', 'server', ${input.payload}, '{}', ${input.version ?? 2}
419 ) RETURNING sequence
420 `;
421 return rows[0]!.sequence;
422 });
423 const first = yield* insert({
424 id: "measured-replay-1",
425 payload: unicodePayload,
426 type: "thread.created",
427 });
428 yield* insert({
429 id: "measured-unrelated",
430 threadId: "other-replay-thread",
431 payload: oversizedPayload,
432 });
433 yield* insert({ id: "measured-legacy", payload: oversizedPayload, version: 1 });
434 const head = yield* insert({ id: "measured-replay-02", payload: oversizedPayload });
435 const later = yield* insert({ id: "measured-replay-003", payload: "{}" });
436 const range = { threadId, afterSequence: baseline, throughSequence: head, maxEvents: 128 };
437 assert.deepEqual(yield* store.getAgentReplayStats(range), {
438 eventCount: 2,
439 rawPayloadBytes: Buffer.byteLength(unicodePayload) + Buffer.byteLength(oversizedPayload),
440 hasCreateEvent: true,
441 });
442 assert.deepEqual(yield* store.getAgentReplayStats({ ...range, afterSequence: first }), {
443 eventCount: 1,
444 rawPayloadBytes: Buffer.byteLength(oversizedPayload),
445 hasCreateEvent: false,
446 });
447 assert.equal(
448 (yield* store.getAgentReplayStats({ ...range, throughSequence: later, maxEvents: 1 }))
449 .eventCount,
450 2,
451 );
452 assert.deepEqual(yield* store.getAgentReplayStats({ ...range, afterSequence: later }), {
453 eventCount: 0,
454 rawPayloadBytes: 0,
455 hasCreateEvent: false,
456 });
457 assert.deepEqual(
458 yield* store.getReplayStats({ afterSequence: baseline, throughSequence: head }),
459 {
460 eventCount: 3,
461 rawPayloadBytes:
462 Buffer.byteLength(unicodePayload) + 2 * Buffer.byteLength(oversizedPayload),
463 },
464 );
465 }),
466 );
467
468 it.effect.each([
469 ["agent", "thread"],
470 ["application", "thread"],
471 ["application", "project"],
472 ] as const)(
473 "%s replay skips unknown %s types but rejects malformed known payloads",
474 ([reader, aggregateKind]) =>
475 Effect.gen(function* () {
476 const store = yield* OrchestrationEventStore.OrchestrationEventStore;
477 const sql = yield* SqlClient.SqlClient;
478 const streamId = ThreadId.make(`skip-unknown:${reader}:${aggregateKind}`);
479 const now = "2026-01-05T00:00:00.000Z";
480 const insertRow = Effect.fnUntraced(function* (
481 version: number,
482 eventType: string,
483 payload: string,
484 ) {
485 const [row] = yield* sql<{ sequence: number }>`
486 INSERT INTO orchestration_events (
487 event_id, aggregate_kind, stream_id, stream_version, event_type,
488 occurred_at, actor_kind, payload_json, metadata_json, application_event_version
489 ) VALUES (
490 ${`${streamId}:${version}`}, ${aggregateKind}, ${streamId}, ${version}, ${eventType},
491 ${now}, 'server', ${payload}, '{}', 2
492 ) RETURNING sequence
493 `;
494 return row!.sequence;
495 });
496 const read = (
497 afterSequence: number,
498 throughSequence: number,
499 skipUnknownEventTypes = false,
500 limit?: number,
501 ) => {
502 const input = { afterSequence, throughSequence, skipUnknownEventTypes };
503 return reader === "agent"
504 ? store
505 .readAgentEvents({
506 ...input,
507 threadId: streamId,
508 ...(limit === undefined ? {} : { limit }),
509 })
510 .pipe(
511 Stream.map((event) => event.sequence),
512 Stream.runCollect,
513 )
514 : store.readApplicationEvents(input).pipe(
515 Stream.map((event) => event.sequence),
516 Stream.runCollect,
517 );
518 };
519 const knownType =
520 aggregateKind === "thread" ? "provider-session.detached" : "project.meta-updated";
521 const knownPayload = JSON.stringify(
522 aggregateKind === "thread"
523 ? { providerSessionId: "session:known", detachedAt: now }
524 : { projectId: streamId, updatedAt: now },
525 );
526 const baseline = yield* store.latestApplicationSequence;
527 const before = yield* insertRow(0, knownType, knownPayload);
528 yield* Effect.forEach(
529 Array.from({ length: 1_000 }, (_, i) => i + 1),
530 (version) => insertRow(version, "future.event", "not JSON"),
531 { discard: true },
532 );
533 yield* insertRow(
534 1_001,
535 aggregateKind === "thread" ? "project.meta-updated" : "provider-session.detached",
536 "{}",
537 );
538 const after = yield* insertRow(1_002, knownType, knownPayload);
539 const highWater = yield* insertRow(1_003, "future.trailing", "{}");
540 yield* insertRow(1_004, knownType, knownPayload);
541 assert.deepEqual(yield* read(baseline, highWater, true), [before, after]);
542 assert.isTrue(isPersistenceDecodeError(yield* read(baseline, highWater).pipe(Effect.flip)));
543 if (reader === "agent") {
544 assert.deepEqual(yield* read(baseline, highWater, true, 2), [before]);
545 }
546 const beforeBroken = yield* store.latestApplicationSequence;
547 const broken = yield* insertRow(
548 1_005,
549 aggregateKind === "thread" ? "thread.metadata-updated" : "project.created",
550 "{}",
551 );
552 assert.isTrue(
553 isPersistenceDecodeError(yield* read(beforeBroken, broken, true).pipe(Effect.flip)),
554 );
555 }),
556 );
557});
558
559it.effect.each(["high-water", "replay"] as const)(
560 "bounds application live events while the %s query is blocked",
561 (phase) =>
562 Effect.scoped(
563 Effect.gen(function* () {
564 const store = yield* OrchestrationEventStore.OrchestrationEventStore;
565 const now = "2026-01-03T00:00:00.000Z";
566 const projectId = ProjectId.make(`project:blocked-${phase}`);
567 const projectEvent = yield* store.appendProjectEvent({
568 type: "project.created",
569 eventId: EventId.make(`event:blocked-${phase}:created`),
570 aggregateKind: "project",
571 aggregateId: projectId,
572 occurredAt: now,
573 commandId: CommandId.make(`command:blocked-${phase}`),
574 causationEventId: null,
575 correlationId: null,
576 metadata: {},
577 payload: {
578 projectId,
579 title: "Blocked stream",
580 workspaceRoot: "/tmp/blocked-stream",
581 defaultModelSelection: null,
582 scripts: [],
583 createdAt: now,
584 updatedAt: now,
585 },
586 });
587 if (projectEvent.type !== "project.created")
588 return yield* Effect.die("Expected project.created");
589 const readStarted = yield* Deferred.make<void>();
590 const readClosed = yield* Deferred.make<void>();
591 const blockRead: Statement.Transformer = (statement) => {
592 const [query] = statement.compile();
593 if (
594 query.includes("FROM orchestration_events") &&
595 query.includes("MAX(sequence)") === (phase === "high-water")
596 ) {
597 return Deferred.succeed(readStarted, undefined).pipe(
598 Effect.andThen(Effect.never),
599 Effect.ensuring(Deferred.succeed(readClosed, undefined)),
600 Effect.as(statement),
601 );
602 }
603 return Effect.succeed(statement);
604 };
605 const reader = yield* store
606 .streamApplicationEvents({ afterSequence: projectEvent.sequence })
607 .pipe(
608 Stream.provideService(Statement.CurrentTransformer, blockRead),
609 Stream.runDrain,
610 Effect.result,
611 Effect.forkScoped,
612 );
613 yield* Deferred.await(readStarted);
614 yield* store.publishCommitted(
615 Array.from({ length: LIVE_STREAM_MAX_ITEMS + 1 }, (_, index) => ({
616 ...projectEvent,
617 eventId: EventId.make(`event:blocked-${phase}:${index}`),
618 sequence: projectEvent.sequence + index + 1,
619 })),
620 );
621 yield* Deferred.await(readClosed);
622 const result = yield* Fiber.join(reader);
623 assert.equal(result._tag, "Failure");
624 if (result._tag === "Failure") {
625 assert.equal(result.failure._tag, "PersistenceSqlError");
626 assert.isTrue(isLiveStreamBufferError(result.failure.cause));
627 }
628 }),
629 ).pipe(Effect.provide(Layer.fresh(layerTest))),
630);