1import * as NodeV8 from "node:v8";
2
3import {
4 CommandId,
5 EventId,
6 NodeId,
7 ProjectId,
8 ProviderDriverKind,
9 ProviderInstanceId,
10 ThreadId,
11 TurnItemId,
12 type OrchestrationV2DomainEvent,
13} from "@t3tools/contracts";
14import { assert, it } from "@effect/vitest";
15import * as DateTime from "effect/DateTime";
16import * as Deferred from "effect/Deferred";
17import * as Fiber from "effect/Fiber";
18import * as Effect from "effect/Effect";
19import * as Layer from "effect/Layer";
20import * as Schema from "effect/Schema";
21import * as Stream from "effect/Stream";
22import * as SqlClient from "effect/sql/SqlClient";
23import * as Statement from "effect/sql/Statement";
24
25import {
26 LIVE_STREAM_MAX_ITEMS,
27 LiveStreamBufferError,
28} from "../orchestration-v2/LiveStreamBudget.ts";
29import { PersistenceDecodeError } from "./Errors.ts";
30import { toShellApplicationEvent } from "../orchestration-v2/ShellStream.ts";
31import * as OrchestrationEventStore from "./OrchestrationEventStore.ts";
32import * as SqlitePersistence from "./Sqlite.ts";
33const isPersistenceDecodeError = Schema.is(PersistenceDecodeError);
34const isLiveStreamBufferError = Schema.is(LiveStreamBufferError);
35
36const layerTest = OrchestrationEventStore.layer.pipe(
37 Layer.provideMerge(SqlitePersistence.layerMemory),
38);
39const layer = it.layer(layerTest);
40
41layer("OrchestrationEventStore", (it) => {
42 it.effect("retains only shell metadata from oversized replay and live application events", () =>
43 Effect.scoped(
44 Effect.gen(function* () {
45 const store = yield* OrchestrationEventStore.OrchestrationEventStore;
46 const afterSequence = yield* store.latestApplicationSequence;
47 const now = yield* DateTime.now;
48 const threadId = ThreadId.make("thread:large-shell-body");
49 const input = "x".repeat(9 * 1024 * 1024);
50 const event: OrchestrationV2DomainEvent = {
51 id: EventId.make("event:large-shell-body:replay"),
52 type: "turn-item.updated",
53 threadId,
54 occurredAt: now,
55 payload: {
56 id: TurnItemId.make("tool:large-shell-body"),
57 type: "command_execution",
58 threadId,
59 runId: null,
60 nodeId: null,
61 providerThreadId: null,
62 providerTurnId: null,
63 nativeItemRef: null,
64 parentItemId: null,
65 ordinal: 1,
66 status: "running",
67 title: "Large command",
68 input,
69 startedAt: now,
70 completedAt: null,
71 updatedAt: now,
72 },
73 };
74 const [replayed] = yield* store.appendAgentEvents({ events: [event] });
75 const pull = yield* Stream.toPull(
76 store.streamProjectedApplicationEvents({
77 afterSequence,
78 project: toShellApplicationEvent,
79 }),
80 );
81 assert.deepEqual(yield* pull, [{ sequence: replayed!.sequence, event: { threadId } }]);
82 const [live] = yield* store.appendAgentEvents({
83 events: [
84 {
85 ...event,
86 id: EventId.make("event:large-shell-body:live"),
87 payload: { ...event.payload, status: "completed", completedAt: now },
88 },
89 ],
90 });
91 yield* store.publishCommitted([live!]);
92 assert.deepEqual(yield* pull, [{ sequence: live!.sequence, event: { threadId } }]);
93 const durable = yield* store
94 .readAgentEvents({ threadId, afterSequence })
95 .pipe(Stream.runCollect);
96 assert.lengthOf(durable, 2);
97 for (const stored of durable) {
98 assert.equal(stored.event.type, "turn-item.updated");
99 if (stored.event.type !== "turn-item.updated") return;
100 assert.equal(stored.event.payload.type, "command_execution");
101 if (stored.event.payload.type !== "command_execution") return;
102 assert.equal(stored.event.payload.input, input);
103 }
104 }),
105 ),
106 );
107
108 it.effect("stores an unchanged subagent prompt once and replays it on every update", () =>
109 Effect.gen(function* () {
110 const store = yield* OrchestrationEventStore.OrchestrationEventStore;
111 const sql = yield* SqlClient.SqlClient;
112 const afterSequence = yield* store.latestApplicationSequence;
113 const now = yield* DateTime.now;
114 const threadId = ThreadId.make("thread:subagent-prompt");
115 const id = NodeId.make("node:subagent-prompt");
116 const driver = ProviderDriverKind.make("claudeAgent");
117 const providerInstanceId = ProviderInstanceId.make("claudeAgent");
118 const prompt = "Audit the repository. ".repeat(200);
119 const subagent = (index: number, progress: string, nextPrompt = prompt) =>
120 ({
121 id: EventId.make(`event:subagent-prompt:subagent:${index}`),
122 type: "subagent.updated",
123 threadId,
124 occurredAt: now,
125 payload: {
126 id,
127 threadId,
128 runId: null,
129 parentNodeId: NodeId.make("node:subagent-prompt-parent"),
130 origin: "provider_native",
131 createdBy: "agent",
132 driver,
133 providerInstanceId,
134 providerThreadId: null,
135 childThreadId: null,
136 nativeTaskRef: { driver, nativeId: "subagent-prompt", strength: "strong" },
137 prompt: nextPrompt,
138 progress,
139 title: "Audit",
140 model: null,
141 status: "running",
142 result: null,
143 startedAt: now,
144 completedAt: null,
145 updatedAt: now,
146 },
147 }) satisfies OrchestrationV2DomainEvent;
148 const item = (index: number, progress: string, nextPrompt = prompt) =>
149 ({
150 id: EventId.make(`event:subagent-prompt:item:${index}`),
151 type: "turn-item.updated",
152 threadId,
153 occurredAt: now,
154 payload: {
155 id: TurnItemId.make("item:subagent-prompt"),
156 type: "subagent",
157 threadId,
158 runId: null,
159 nodeId: null,
160 providerThreadId: null,
161 providerTurnId: null,
162 nativeItemRef: null,
163 parentItemId: null,
164 ordinal: 1,
165 status: "running",
166 title: "Audit",
167 subagentId: id,
168 origin: "provider_native",
169 driver,
170 providerInstanceId,
171 childThreadId: null,
172 prompt: nextPrompt,
173 progress,
174 result: null,
175 startedAt: now,
176 completedAt: null,
177 updatedAt: now,
178 },
179 }) satisfies OrchestrationV2DomainEvent;
180 const events = [
181 subagent(0, "Reading files"),
182 item(0, "Reading files"),
183 subagent(1, "Running tests"),
184 item(1, "Running tests"),
185 subagent(2, "Writing report"),
186 item(2, "Writing report"),
187 subagent(3, "Retrying", "Audit again."),
188 item(3, "Retrying", "Audit again."),
189 ];
190 yield* store.appendAgentEvents({ events });
191
192 const stored = yield* sql<{ readonly payload_json: string }>`
193 SELECT payload_json FROM orchestration_events
194 WHERE stream_id = ${threadId} ORDER BY sequence ASC
195 `;
196 assert.deepEqual(
197 stored.map((row) => row.payload_json.includes(prompt)),
198 [true, true, false, false, false, false, false, false],
199 );
200 const replayed = yield* store
201 .readAgentEvents({ threadId, afterSequence })
202 .pipe(Stream.runCollect);
203 assert.deepEqual(
204 replayed.map((entry) => entry.event),
205 events,
206 );
207 // Replay budgets measure the payloads replay decodes, not the slimmer stored rows.
208 const stats = yield* store.getAgentReplayStats({
209 threadId,
210 afterSequence,
211 throughSequence: yield* store.latestApplicationSequence,
212 maxEvents: events.length,
213 });
214 assert.isAbove(stats.rawPayloadBytes, prompt.length * 6);
215 const applicationStats = yield* store.getReplayStats({
216 afterSequence,
217 throughSequence: yield* store.latestApplicationSequence,
218 });
219 assert.isAbove(applicationStats.rawPayloadBytes, prompt.length * 6);
220 }),
221 );
222
223 it.effect("stores the full subagent prompt when its hint row was rolled back", () =>
224 Effect.gen(function* () {
225 const store = yield* OrchestrationEventStore.OrchestrationEventStore;
226 const sql = yield* SqlClient.SqlClient;
227 const afterSequence = yield* store.latestApplicationSequence;
228 const now = yield* DateTime.now;
229 const threadId = ThreadId.make("thread:subagent-rollback");
230 const event = {
231 id: EventId.make("event:subagent-rollback"),
232 type: "subagent.updated",
233 threadId,
234 occurredAt: now,
235 payload: {
236 id: NodeId.make("node:subagent-rollback"),
237 threadId,
238 runId: null,
239 parentNodeId: NodeId.make("node:subagent-rollback-parent"),
240 origin: "provider_native",
241 createdBy: "agent",
242 driver: ProviderDriverKind.make("claudeAgent"),
243 providerInstanceId: ProviderInstanceId.make("claudeAgent"),
244 providerThreadId: null,
245 childThreadId: null,
246 nativeTaskRef: null,
247 prompt: "Audit the repository. ".repeat(200),
248 progress: "Reading files",
249 title: "Audit",
250 model: null,
251 status: "running",
252 result: null,
253 startedAt: now,
254 completedAt: null,
255 updatedAt: now,
256 },
257 } satisfies OrchestrationV2DomainEvent;
258 const rolledBack = yield* sql
259 .withTransaction(
260 store
261 .appendAgentEvents({ events: [event] })
262 .pipe(Effect.andThen(Effect.fail("rollback"))),
263 )
264 .pipe(Effect.flip);
265 assert.equal(rolledBack, "rollback");
266 const rowsAfterRollback = yield* sql`
267 SELECT sequence FROM orchestration_events WHERE stream_id = ${threadId}
268 `;
269 assert.deepEqual(rowsAfterRollback, []);
270
271 const events = [event, { ...event, id: EventId.make("event:subagent-after-rollback") }];
272 const appended = yield* store.appendAgentEvents({ events });
273 assert.deepEqual(
274 appended.map((entry) => entry.event),
275 events,
276 );
277 const stored = yield* sql<{
278 readonly prompt: string | null;
279 readonly prompt_sequence: number | null;
280 }>`
281 SELECT json_extract(payload_json, '$.prompt') AS prompt,
282 json_extract(payload_json, '$.promptSequence') AS prompt_sequence
283 FROM orchestration_events WHERE stream_id = ${threadId} ORDER BY sequence
284 `;
285 assert.deepEqual(stored, [
286 { prompt: event.payload.prompt, prompt_sequence: null },
287 { prompt: "", prompt_sequence: appended[0]!.sequence },
288 ]);
289 const replayed = yield* store.readAgentEvents({ threadId }).pipe(Stream.runCollect);
290 assert.deepEqual(replayed, appended);
291 const applicationEvents = yield* store
292 .readApplicationEvents({
293 afterSequence,
294 throughSequence: appended[1]!.sequence,
295 })
296 .pipe(Stream.runCollect);
297 assert.deepEqual(applicationEvents, [...appended]);
298 }),
299 );
300
301 it.effect("stores json columns as strings and replays CLI-origin events", () =>
302 Effect.gen(function* () {
303 const eventStore = yield* OrchestrationEventStore.OrchestrationEventStore;
304 const sql = yield* SqlClient.SqlClient;
305 const now = "2026-01-01T00:00:00.000Z";
306
307 const appended = yield* eventStore.appendProjectEvent({
308 type: "project.created",
309 eventId: EventId.make("evt-store-roundtrip"),
310 aggregateKind: "project",
311 aggregateId: ProjectId.make("project-roundtrip"),
312 occurredAt: now,
313 commandId: CommandId.make("cmd-store-roundtrip"),
314 causationEventId: null,
315 correlationId: CommandId.make("cmd-store-roundtrip"),
316 metadata: {
317 adapterKey: "codex",
318 origin: {
319 surface: "cli",
320 },
321 },
322 payload: {
323 projectId: ProjectId.make("project-roundtrip"),
324 title: "Roundtrip Project",
325 workspaceRoot: "/tmp/project-roundtrip",
326 defaultModelSelection: null,
327 scripts: [],
328 createdAt: now,
329 updatedAt: now,
330 },
331 });
332
333 const storedRows = yield* sql<{
334 readonly payloadJson: string;
335 readonly metadataJson: string;
336 }>`
337 SELECT
338 payload_json AS "payloadJson",
339 metadata_json AS "metadataJson"
340 FROM orchestration_events
341 WHERE event_id = ${appended.eventId}
342 `;
343 assert.equal(storedRows.length, 1);
344 assert.equal(typeof storedRows[0]?.payloadJson, "string");
345 assert.equal(typeof storedRows[0]?.metadataJson, "string");
346
347 const replayed = yield* eventStore
348 .readApplicationEvents({
349 afterSequence: appended.sequence - 1,
350 throughSequence: appended.sequence,
351 })
352 .pipe(
353 Stream.runCollect,
354 Effect.map((chunk) => Array.from(chunk)),
355 );
356 assert.equal(replayed.length, 1);
357 const [project] = replayed;
358 if (project === undefined || !("aggregateKind" in project)) {
359 return assert.fail("Expected a project event");
360 }
361 assert.equal(project.type, "project.created");
362 assert.equal(project.metadata.adapterKey, "codex");
363 assert.deepEqual(project.metadata.origin, { surface: "cli" });
364 }),
365 );
366
367 it.effect("fails with PersistenceDecodeError when stored json is invalid", () =>
368 Effect.gen(function* () {
369 const eventStore = yield* OrchestrationEventStore.OrchestrationEventStore;
370 const sql = yield* SqlClient.SqlClient;
371 const now = "2026-01-01T00:00:00.000Z";
372
373 yield* sql`
374 INSERT INTO orchestration_events (
375 event_id,
376 aggregate_kind,
377 stream_id,
378 stream_version,
379 event_type,
380 occurred_at,
381 command_id,
382 causation_event_id,
383 correlation_id,
384 actor_kind,
385 payload_json,
386 metadata_json
387 )
388 VALUES (
389 ${EventId.make("evt-store-invalid-json")},
390 ${"project"},
391 ${ProjectId.make("project-invalid-json")},
392 ${0},
393 ${"project.created"},
394 ${now},
395 ${CommandId.make("cmd-store-invalid-json")},
396 ${null},
397 ${null},
398 ${"server"},
399 ${"{"},
400 ${"{}"}
401 )
402 `;
403
404 const replayResult = yield* Effect.result(
405 Stream.runCollect(
406 eventStore.readApplicationEvents({
407 afterSequence: 0,
408 throughSequence: Number.MAX_SAFE_INTEGER,
409 }),
410 ),
411 );
412 assert.equal(replayResult._tag, "Failure");
413 if (replayResult._tag === "Failure") {
414 assert.ok(isPersistenceDecodeError(replayResult.failure));
415 }
416 }),
417 );
418
419 it.effect("orders project and V2 agent events in the retained application event source", () =>
420 Effect.gen(function* () {
421 const eventStore = yield* OrchestrationEventStore.OrchestrationEventStore;
422 const projectId = ProjectId.make("project-shared-stream");
423 const threadId = ThreadId.make("thread-shared-stream");
424 const providerInstanceId = ProviderInstanceId.make("codex");
425 const occurredAt = DateTime.makeUnsafe("2026-01-02T00:00:00.000Z");
426 const now = DateTime.formatIso(occurredAt);
427 const baselineSequence = yield* eventStore.latestApplicationSequence;
428
429 const projectEvent = yield* eventStore.appendProjectEvent({
430 type: "project.created",
431 eventId: EventId.make("event-project-shared-stream"),
432 aggregateKind: "project",
433 aggregateId: projectId,
434 occurredAt: now,
435 commandId: CommandId.make("command-project-shared-stream"),
436 causationEventId: null,
437 correlationId: CommandId.make("command-project-shared-stream"),
438 metadata: {},
439 payload: {
440 projectId,
441 title: "Shared stream",
442 workspaceRoot: "/tmp/shared-stream",
443 defaultModelSelection: null,
444 scripts: [],
445 createdAt: now,
446 updatedAt: now,
447 },
448 });
449 const [threadEvent] = yield* eventStore.appendAgentEvents({
450 commandId: CommandId.make("command-thread-shared-stream"),
451 events: [
452 {
453 id: EventId.make("event-thread-shared-stream"),
454 type: "thread.created",
455 threadId,
456 providerInstanceId,
457 occurredAt,
458 payload: {
459 id: threadId,
460 projectId,
461 title: "Thread",
462 providerInstanceId,
463 modelSelection: { instanceId: providerInstanceId, model: "gpt-5.4" },
464 runtimeMode: "full-access",
465 interactionMode: "default",
466 branch: null,
467 worktreePath: null,
468 activeProviderThreadId: null,
469 lineage: {
470 rootThreadId: threadId,
471 parentThreadId: null,
472 relationshipToParent: null,
473 },
474 forkedFrom: null,
475 createdBy: "user",
476 creationSource: "web",
477 createdAt: occurredAt,
478 updatedAt: occurredAt,
479 archivedAt: null,
480 settledOverride: null,
481 settledAt: null,
482 lastVisitedAt: null,
483 deletedAt: null,
484 },
485 },
486 ],
487 });
488
489 const applicationEvents = yield* eventStore
490 .streamApplicationEvents({ afterSequence: baselineSequence })
491 .pipe(
492 Stream.take(2),
493 Stream.runCollect,
494 Effect.map((chunk) => Array.from(chunk)),
495 );
496 assert.deepEqual(
497 applicationEvents.map((event) => event.sequence),
498 [projectEvent.sequence, threadEvent!.sequence],
499 );
500 assert.isTrue("aggregateKind" in applicationEvents[0]!);
501 assert.isTrue("event" in applicationEvents[1]!);
502
503 const finiteReplay = yield* eventStore
504 .readApplicationEvents({
505 afterSequence: baselineSequence,
506 throughSequence: threadEvent!.sequence,
507 })
508 .pipe(
509 Stream.runCollect,
510 Effect.map((chunk) => Array.from(chunk)),
511 );
512 assert.deepEqual(
513 finiteReplay.map((event) => event.sequence),
514 [projectEvent.sequence, threadEvent!.sequence],
515 );
516 }),
517 );
518 it.effect("measures only a bounded thread replay before decoding its payloads", () =>
519 Effect.gen(function* () {
520 const store = yield* OrchestrationEventStore.OrchestrationEventStore;
521 const sql = yield* SqlClient.SqlClient;
522 const threadId = ThreadId.make("measured-replay-thread");
523 const baseline = yield* store.latestApplicationSequence;
524 const unicodePayload = '{"text":"🦊"}';
525 const oversizedPayload = "{" + "x".repeat(1_048_576);
526 let streamVersion = 0;
527 const insert = Effect.fn(function* (input: {
528 readonly id: string;
529 readonly payload: string;
530 readonly type?: string;
531 readonly threadId?: string;
532 readonly version?: number;
533 }) {
534 const rows = yield* sql<{ readonly sequence: number }>`
535 INSERT INTO orchestration_events (
536 event_id, aggregate_kind, stream_id, stream_version, event_type,
537 occurred_at, actor_kind, payload_json, metadata_json, application_event_version
538 ) VALUES (
539 ${input.id}, 'thread', ${input.threadId ?? threadId},
540 ${streamVersion++}, ${input.type ?? "thread.metadata-updated"},
541 '2026-09-04T00:00:00.000Z', 'server', ${input.payload}, '{}', ${input.version ?? 2}
542 ) RETURNING sequence
543 `;
544 return rows[0]!.sequence;
545 });
546 const first = yield* insert({
547 id: "measured-replay-1",
548 payload: unicodePayload,
549 type: "thread.created",
550 });
551 yield* insert({
552 id: "measured-unrelated",
553 threadId: "other-replay-thread",
554 payload: oversizedPayload,
555 });
556 yield* insert({ id: "measured-legacy", payload: oversizedPayload, version: 1 });
557 const head = yield* insert({ id: "measured-replay-02", payload: oversizedPayload });
558 const later = yield* insert({ id: "measured-replay-003", payload: "{}" });
559 const range = { threadId, afterSequence: baseline, throughSequence: head, maxEvents: 128 };
560 assert.deepEqual(yield* store.getAgentReplayStats(range), {
561 eventCount: 2,
562 rawPayloadBytes: Buffer.byteLength(unicodePayload) + Buffer.byteLength(oversizedPayload),
563 hasCreateEvent: true,
564 });
565 assert.deepEqual(yield* store.getAgentReplayStats({ ...range, afterSequence: first }), {
566 eventCount: 1,
567 rawPayloadBytes: Buffer.byteLength(oversizedPayload),
568 hasCreateEvent: false,
569 });
570 assert.equal(
571 (yield* store.getAgentReplayStats({ ...range, throughSequence: later, maxEvents: 1 }))
572 .eventCount,
573 2,
574 );
575 assert.deepEqual(yield* store.getAgentReplayStats({ ...range, afterSequence: later }), {
576 eventCount: 0,
577 rawPayloadBytes: 0,
578 hasCreateEvent: false,
579 });
580 // The application range skips V1 rows, which a shell resume never replays.
581 assert.deepEqual(
582 yield* store.getReplayStats({ afterSequence: baseline, throughSequence: head }),
583 {
584 eventCount: 3,
585 rawPayloadBytes:
586 Buffer.byteLength(unicodePayload) + 2 * Buffer.byteLength(oversizedPayload),
587 },
588 );
589 }),
590 );
591
592 it.effect.each([
593 ["agent", "thread"],
594 ["application", "thread"],
595 ["application", "project"],
596 ] as const)(
597 "%s replay skips unknown %s types but rejects malformed known payloads",
598 ([reader, aggregateKind]) =>
599 Effect.gen(function* () {
600 const store = yield* OrchestrationEventStore.OrchestrationEventStore;
601 const sql = yield* SqlClient.SqlClient;
602 const streamId = ThreadId.make(`skip-unknown:${reader}:${aggregateKind}`);
603 const now = "2026-01-05T00:00:00.000Z";
604 // Direct SQL models rows written by a newer build, which append cannot accept.
605 const insertRow = Effect.fnUntraced(function* (
606 version: number,
607 eventType: string,
608 payload: string,
609 ) {
610 const [row] = yield* sql<{ sequence: number }>`
611 INSERT INTO orchestration_events (
612 event_id, aggregate_kind, stream_id, stream_version, event_type,
613 occurred_at, actor_kind, payload_json, metadata_json, application_event_version
614 ) VALUES (
615 ${`${streamId}:${version}`}, ${aggregateKind}, ${streamId}, ${version}, ${eventType},
616 ${now}, 'server', ${payload}, '{}', 2
617 ) RETURNING sequence
618 `;
619 return row!.sequence;
620 });
621 const read = (
622 afterSequence: number,
623 throughSequence: number,
624 skipUnknownEventTypes = false,
625 limit?: number,
626 ) => {
627 const input = { afterSequence, throughSequence, skipUnknownEventTypes };
628 return reader === "agent"
629 ? store
630 .readAgentEvents({
631 ...input,
632 threadId: streamId,
633 ...(limit === undefined ? {} : { limit }),
634 })
635 .pipe(
636 Stream.map((event) => event.sequence),
637 Stream.runCollect,
638 )
639 : store.readApplicationEvents(input).pipe(
640 Stream.map((event) => event.sequence),
641 Stream.runCollect,
642 );
643 };
644 const knownType =
645 aggregateKind === "thread" ? "provider-session.detached" : "project.meta-updated";
646 const knownPayload = JSON.stringify(
647 aggregateKind === "thread"
648 ? { providerSessionId: "session:known", detachedAt: now }
649 : { projectId: streamId, updatedAt: now },
650 );
651 const baseline = yield* store.latestApplicationSequence;
652 const before = yield* insertRow(0, knownType, knownPayload);
653 // More than one raw page ensures replay crosses an entirely filtered page.
654 yield* Effect.forEach(
655 Array.from({ length: 1_000 }, (_, i) => i + 1),
656 (version) => insertRow(version, "future.event", "not JSON"),
657 { discard: true },
658 );
659 // A type known for the other aggregate kind is unknown for this one.
660 yield* insertRow(
661 1_001,
662 aggregateKind === "thread" ? "project.meta-updated" : "provider-session.detached",
663 "{}",
664 );
665 const after = yield* insertRow(1_002, knownType, knownPayload);
666 const highWater = yield* insertRow(1_003, "future.trailing", "{}");
667 yield* insertRow(1_004, knownType, knownPayload);
668 assert.deepEqual(yield* read(baseline, highWater, true), [before, after]);
669 assert.isTrue(isPersistenceDecodeError(yield* read(baseline, highWater).pipe(Effect.flip)));
670 if (reader === "agent") {
671 // The work limit includes skipped rows, rather than fetching extra pages to fill it.
672 assert.deepEqual(yield* read(baseline, highWater, true, 2), [before]);
673 }
674 const beforeBroken = yield* store.latestApplicationSequence;
675 // Exercise a grouped V2 literal as well as the project type discriminator.
676 const broken = yield* insertRow(
677 1_005,
678 aggregateKind === "thread" ? "thread.metadata-updated" : "project.created",
679 "{}",
680 );
681 assert.isTrue(
682 isPersistenceDecodeError(yield* read(beforeBroken, broken, true).pipe(Effect.flip)),
683 );
684 }),
685 );
686});
687
688it.effect.each(["high-water", "replay"] as const)(
689 "bounds application live events while the %s query is blocked",
690 (phase) =>
691 Effect.scoped(
692 Effect.gen(function* () {
693 const store = yield* OrchestrationEventStore.OrchestrationEventStore;
694 const now = "2026-01-03T00:00:00.000Z";
695 const projectId = ProjectId.make(`project:blocked-${phase}`);
696 const projectEvent = yield* store.appendProjectEvent({
697 type: "project.created",
698 eventId: EventId.make(`event:blocked-${phase}:created`),
699 aggregateKind: "project",
700 aggregateId: projectId,
701 occurredAt: now,
702 commandId: CommandId.make(`command:blocked-${phase}`),
703 causationEventId: null,
704 correlationId: null,
705 metadata: {},
706 payload: {
707 projectId,
708 title: "Blocked stream",
709 workspaceRoot: "/tmp/blocked-stream",
710 defaultModelSelection: null,
711 scripts: [],
712 createdAt: now,
713 updatedAt: now,
714 },
715 });
716 if (projectEvent.type !== "project.created")
717 return yield* Effect.die("Expected project.created");
718 const readStarted = yield* Deferred.make<void>();
719 const readClosed = yield* Deferred.make<void>();
720 const blockRead: Statement.Transformer = (statement) => {
721 const [query] = statement.compile();
722 if (
723 query.includes("FROM orchestration_events") &&
724 query.includes("MAX(sequence)") === (phase === "high-water")
725 ) {
726 return Deferred.succeed(readStarted, undefined).pipe(
727 Effect.andThen(Effect.never),
728 Effect.ensuring(Deferred.succeed(readClosed, undefined)),
729 Effect.as(statement),
730 );
731 }
732 return Effect.succeed(statement);
733 };
734 const reader = yield* store
735 .streamApplicationEvents({ afterSequence: projectEvent.sequence })
736 .pipe(
737 Stream.provideService(Statement.CurrentTransformer, blockRead),
738 Stream.runDrain,
739 Effect.result,
740 Effect.forkScoped,
741 );
742 yield* Deferred.await(readStarted);
743 yield* store.publishCommitted(
744 Array.from({ length: LIVE_STREAM_MAX_ITEMS + 1 }, (_, index) => ({
745 ...projectEvent,
746 eventId: EventId.make(`event:blocked-${phase}:${index}`),
747 sequence: projectEvent.sequence + index + 1,
748 })),
749 );
750 yield* Deferred.await(readClosed);
751 const result = yield* Fiber.join(reader);
752 assert.equal(result._tag, "Failure");
753 if (result._tag === "Failure") {
754 assert.equal(result.failure._tag, "PersistenceSqlError");
755 assert.isTrue(isLiveStreamBufferError(result.failure.cause));
756 }
757 }),
758 ).pipe(Effect.provide(Layer.fresh(layerTest))),
759);
760
761it.effect("releases consumed application replay pages", () =>
762 Effect.gen(function* () {
763 const store = yield* OrchestrationEventStore.OrchestrationEventStore;
764 const afterSequence = yield* store.latestApplicationSequence;
765 yield* Effect.forEach(
766 Array.from({ length: 1501 }, (_, index) => index),
767 (index) =>
768 store.appendProjectEvent({
769 type: "project.created",
770 eventId: EventId.make(`retention-${index}`),
771 aggregateKind: "project",
772 aggregateId: ProjectId.make(`retention-${index}`),
773 occurredAt: "2026-09-08T00:00:00Z",
774 commandId: null,
775 causationEventId: null,
776 correlationId: null,
777 metadata: {},
778 payload: {
779 projectId: ProjectId.make(`retention-${index}`),
780 title: "Replay",
781 workspaceRoot: "/tmp/replay",
782 defaultModelSelection: null,
783 scripts: [],
784 createdAt: "2026-09-08T00:00:00Z",
785 updatedAt: "2026-09-08T00:00:00Z",
786 },
787 }),
788 { discard: true },
789 );
790 // oxlint-disable-next-line typescript/no-extraneous-class -- Counts retained page markers after full GC.
791 class ReplayPage {}
792 let count = 0;
793 yield* store
794 .readApplicationEvents({
795 afterSequence,
796 throughSequence: yield* store.latestApplicationSequence,
797 })
798 .pipe(
799 Stream.runForEach((event) =>
800 Effect.sync(() => {
801 if (count % 500 === 0) {
802 Object.assign(event, { replayPage: new ReplayPage() });
803 assert.isAtMost(NodeV8.queryObjects(ReplayPage, { format: "count" }), 1);
804 }
805 count++;
806 }),
807 ),
808 );
809 assert.equal(count, 1501);
810 }).pipe(Effect.provide(layerTest)),
811);