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