1import { CommandId, EventId, ProviderSessionId, ThreadId } from "@t3tools/contracts";
2import { assert, it } from "@effect/vitest";
3import * as DateTime from "effect/DateTime";
4import * as Effect from "effect/Effect";
5import * as Layer from "effect/Layer";
6import * as Schema from "effect/Schema";
7import * as Stream from "effect/Stream";
8import * as Tracer from "effect/Tracer";
9import * as SqlClient from "effect/sql/SqlClient";
10
11import { runMigrations } from "./Migrations.ts";
12import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
13import * as OrchestrationEventStore from "./OrchestrationEventStore.ts";
14import * as SqlitePersistence from "./Sqlite.ts";
15
16const layerEventStore = OrchestrationEventStore.layer.pipe(
17 Layer.provideMerge(SqlitePersistence.layerMemory),
18);
19const occurredAt = "2026-09-03T00:00:00.000Z";
20const encodeJson = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown));
21
22function eventRow(
23 ordinal: number,
24 aggregateKind: "project" | "thread",
25 version: number,
26 streamId = `${aggregateKind}:${ordinal}`,
27 commandId: string | null = null,
28) {
29 return {
30 event_id: `event:${ordinal}`,
31 aggregate_kind: aggregateKind,
32 stream_id: streamId,
33 stream_version: ordinal,
34 event_type: aggregateKind === "project" ? "project.created" : "provider-session.detached",
35 occurred_at: occurredAt,
36 command_id: commandId,
37 causation_event_id: null,
38 correlation_id: null,
39 actor_kind: "server",
40 payload_json: encodeJson(
41 aggregateKind === "project"
42 ? {
43 projectId: streamId,
44 title: "Sequence fixture",
45 workspaceRoot: "/tmp/sequence-fixture",
46 defaultModelSelection: null,
47 scripts: [],
48 createdAt: occurredAt,
49 updatedAt: occurredAt,
50 }
51 : { providerSessionId: `session:${ordinal}`, detachedAt: occurredAt },
52 ),
53 metadata_json: "{}",
54 application_event_version: version,
55 };
56}
57
58const seedEvents = Effect.fn("test.seedSequenceEvents")(function* (
59 rows: ReadonlyArray<ReturnType<typeof eventRow>>,
60) {
61 const sql = yield* SqlClient.SqlClient;
62 return yield* sql<{
63 readonly sequence: number;
64 readonly aggregate_kind: string;
65 readonly application_event_version: number;
66 readonly stream_id: string;
67 readonly command_id: string | null;
68 }>`
69 INSERT INTO orchestration_events ${sql.insert(rows)}
70 RETURNING sequence, aggregate_kind, application_event_version, stream_id, command_id
71 `;
72});
73
74it.effect("keeps application and scoped agent high-water marks separate from legacy history", () =>
75 Effect.gen(function* () {
76 const store = yield* OrchestrationEventStore.OrchestrationEventStore;
77 const sql = yield* SqlClient.SqlClient;
78 const target = ThreadId.make("target");
79 assert.equal(yield* store.latestApplicationSequence, 0);
80 assert.equal(yield* store.latestAgentSequence(), 0);
81 assert.equal(yield* store.latestAgentSequence(target), 0);
82
83 yield* seedEvents([eventRow(1, "thread", 1, "legacy-only")]);
84 assert.equal(yield* store.latestApplicationSequence, 0);
85 assert.equal(yield* store.latestAgentSequence(ThreadId.make("legacy-only")), 0);
86
87 const [legacyProject, targetEvent, otherEvent] = yield* seedEvents([
88 eventRow(2, "project", 1),
89 eventRow(3, "thread", 2, target),
90 eventRow(4, "thread", 2, "other"),
91 ]);
92 assert.ok(legacyProject);
93 assert.ok(targetEvent);
94 assert.ok(otherEvent);
95 assert.equal(yield* store.latestApplicationSequence, otherEvent.sequence);
96 assert.equal(yield* store.latestAgentSequence(), otherEvent.sequence);
97 assert.equal(yield* store.latestAgentSequence(target), targetEvent.sequence);
98
99 const [projectEvent] = yield* seedEvents([
100 eventRow(5, "project", 2),
101 eventRow(6, "thread", 1, target),
102 eventRow(7, "thread", 3, target),
103 ]);
104 assert.ok(projectEvent);
105 assert.equal(yield* store.latestApplicationSequence, projectEvent.sequence);
106 assert.equal(yield* store.latestAgentSequence(), otherEvent.sequence);
107 assert.equal(yield* store.latestAgentSequence(target), targetEvent.sequence);
108 assert.equal(yield* store.latestAgentSequence(ThreadId.make("missing")), 0);
109
110 yield* sql`DELETE FROM orchestration_events WHERE sequence > ${legacyProject.sequence}`;
111 assert.equal(yield* store.latestApplicationSequence, legacyProject.sequence);
112 assert.equal(yield* store.latestAgentSequence(), 0);
113 assert.equal(yield* store.latestAgentSequence(target), 0);
114 }).pipe(Effect.provide(Layer.fresh(layerEventStore))),
115);
116
117it.effect(
118 "preserves mixed application paging, scoped replay, and the catch-up to live boundary",
119 () =>
120 Effect.gen(function* () {
121 const store = yield* OrchestrationEventStore.OrchestrationEventStore;
122 const rows = yield* seedEvents(
123 Array.from({ length: 1_560 }, (_, index) => {
124 const kind = index % 3 === 1 ? "project" : "thread";
125 return eventRow(
126 index,
127 kind,
128 index % 3 === 0 ? 1 : 2,
129 kind === "project" ? `project:${index}` : index % 2 === 0 ? "target" : "other",
130 index % 5 === 0 ? "selected-command" : "other-command",
131 );
132 }),
133 );
134 const afterSequence = rows[25]!.sequence;
135 const throughSequence = rows[1_525]!.sequence;
136 const retained = rows.filter(
137 (row) => row.aggregate_kind === "project" || row.application_event_version === 2,
138 );
139 const replay = yield* store
140 .readApplicationEvents({ afterSequence, throughSequence })
141 .pipe(Stream.runCollect);
142 assert.deepEqual(
143 replay.map((event) => event.sequence),
144 retained
145 .filter((row) => row.sequence > afterSequence && row.sequence <= throughSequence)
146 .map((row) => row.sequence),
147 );
148
149 const scoped = yield* store
150 .readAgentEvents({
151 afterSequence,
152 throughSequence,
153 threadId: ThreadId.make("target"),
154 commandId: CommandId.make("selected-command"),
155 limit: 17,
156 })
157 .pipe(Stream.runCollect);
158 assert.deepEqual(
159 scoped.map((event) => event.sequence),
160 retained
161 .filter(
162 (row) =>
163 row.sequence > afterSequence &&
164 row.sequence <= throughSequence &&
165 row.aggregate_kind === "thread" &&
166 row.stream_id === "target" &&
167 row.command_id === "selected-command",
168 )
169 .slice(0, 17)
170 .map((row) => row.sequence),
171 );
172
173 const catchUp = retained.filter((row) => row.sequence > afterSequence);
174 const highWater = catchUp.at(-1)!.sequence;
175 let liveSequence = 0;
176 const streamed = yield* store.streamApplicationEvents({ afterSequence }).pipe(
177 Stream.tap((event) =>
178 event.sequence !== highWater
179 ? Effect.void
180 : Effect.gen(function* () {
181 const [stored] = yield* store.appendAgentEvents({
182 events: [
183 {
184 id: EventId.make("live-after-catch-up"),
185 type: "provider-session.detached",
186 threadId: ThreadId.make("target"),
187 occurredAt: DateTime.makeUnsafe(occurredAt),
188 payload: {
189 providerSessionId: ProviderSessionId.make("live-session"),
190 detachedAt: DateTime.makeUnsafe(occurredAt),
191 },
192 },
193 ],
194 });
195 assert.ok(stored);
196 liveSequence = stored.sequence;
197 yield* store.publishCommitted([stored]);
198 }),
199 ),
200 Stream.take(catchUp.length + 1),
201 Stream.runCollect,
202 );
203 assert.deepEqual(
204 streamed.map((event) => event.sequence),
205 [...catchUp.map((row) => row.sequence), liveSequence],
206 );
207 assert.isAbove(liveSequence, highWater);
208 }).pipe(Effect.provide(Layer.fresh(layerEventStore))),
209);
210
211it.effect("uses indexed high-water lookups for populated history without OR scans", () =>
212 Effect.gen(function* () {
213 const store = yield* OrchestrationEventStore.OrchestrationEventStore;
214 const sql = yield* SqlClient.SqlClient;
215 yield* runMigrations();
216 yield* sql`
217 WITH RECURSIVE history(n) AS (
218 SELECT 1 UNION ALL SELECT n + 1 FROM history WHERE n < 25000
219 )
220 INSERT INTO orchestration_events (
221 event_id, aggregate_kind, stream_id, stream_version, event_type,
222 occurred_at, actor_kind, payload_json, metadata_json, application_event_version
223 )
224 SELECT 'history:' || n, 'thread',
225 CASE WHEN n = 1 THEN 'target' ELSE 'other' END,
226 n, 'provider-session.detached', ${occurredAt}, 'server', '{}', '{}',
227 CASE WHEN n % 3 = 0 THEN 1 ELSE 2 END
228 FROM history
229 `;
230 const statements: Array<string> = [];
231 const tracer = Tracer.make({
232 span(options) {
233 const span = new Tracer.NativeSpan(options);
234 const end = span.end.bind(span);
235 span.end = (endTime, exit) => {
236 end(endTime, exit);
237 const query = span.attributes.get("db.query.text");
238 if (typeof query === "string") statements.push(query);
239 };
240 return span;
241 },
242 });
243 assert.equal(yield* store.latestApplicationSequence.pipe(Effect.withTracer(tracer)), 25_000);
244 assert.equal(
245 yield* store.latestAgentSequence(ThreadId.make("target")).pipe(Effect.withTracer(tracer)),
246 1,
247 );
248 assert.equal(statements.length, 2);
249 const applicationPlan = yield* sql.unsafe<{ readonly detail: string }>(
250 `EXPLAIN QUERY PLAN ${statements[0]}`,
251 );
252 const agentPlan = yield* sql.unsafe<{ readonly detail: string }>(
253 `EXPLAIN QUERY PLAN ${statements[1]}`,
254 ["target"],
255 );
256 assert.match(
257 applicationPlan.map((row) => row.detail).join("\n"),
258 /SEARCH orchestration_events USING INDEX idx_orchestration_events_application_high_water/,
259 );
260 assert.match(
261 agentPlan.map((row) => row.detail).join("\n"),
262 /USING COVERING INDEX idx_orchestration_events_agent_stream_sequence \(stream_id=\?\)/,
263 );
264 for (const row of [...applicationPlan, ...agentPlan]) {
265 assert.notMatch(row.detail, /MULTI-INDEX OR|TEMP B-TREE/);
266 }
267 }).pipe(
268 Effect.provide(
269 OrchestrationEventStore.layer.pipe(
270 Layer.provideMerge(NodeSqliteClient.layer({ filename: ":memory:" })),
271 ),
272 ),
273 ),
274);
275
276it.effect("replays a command's events through the command index, not the sequence range", () =>
277 Effect.gen(function* () {
278 const store = yield* OrchestrationEventStore.OrchestrationEventStore;
279 const sql = yield* SqlClient.SqlClient;
280 yield* runMigrations();
281 yield* sql`
282 WITH RECURSIVE history(n) AS (
283 SELECT 1 UNION ALL SELECT n + 1 FROM history WHERE n < 25000
284 )
285 INSERT INTO orchestration_events (
286 event_id, aggregate_kind, stream_id, stream_version, event_type, occurred_at,
287 command_id, actor_kind, payload_json, metadata_json, application_event_version
288 )
289 SELECT 'history:' || n, 'thread', 'thread', n, 'provider-session.detached', ${occurredAt},
290 CASE WHEN n IN (12, 24990) THEN 'retried-command' ELSE 'command:' || n END,
291 'server',
292 '{"providerSessionId":"session","detachedAt":"' || ${occurredAt} || '"}',
293 '{}', 2
294 FROM history
295 `;
296 const statements: Array<string> = [];
297 const tracer = Tracer.make({
298 span(options) {
299 const span = new Tracer.NativeSpan(options);
300 const end = span.end.bind(span);
301 span.end = (endTime, exit) => {
302 end(endTime, exit);
303 const query = span.attributes.get("db.query.text");
304 if (typeof query === "string") statements.push(query);
305 };
306 return span;
307 },
308 });
309 const replayed = yield* store
310 .readAgentEvents({ commandId: CommandId.make("retried-command") })
311 .pipe(Stream.runCollect, Effect.withTracer(tracer));
312 assert.deepEqual(
313 replayed.map((event) => event.sequence),
314 [12, 24990],
315 );
316 assert.equal(statements.length, 1);
317 const plan = yield* sql.unsafe<{ readonly detail: string }>(
318 `EXPLAIN QUERY PLAN ${statements[0]}`,
319 [0, Number.MAX_SAFE_INTEGER, "retried-command", 500],
320 );
321 assert.match(
322 plan.map((row) => row.detail).join("\n"),
323 /SEARCH orchestration_events USING INDEX idx_orch_events_command_id \(command_id=\?/,
324 );
325 }).pipe(
326 Effect.provide(
327 OrchestrationEventStore.layer.pipe(
328 Layer.provideMerge(NodeSqliteClient.layer({ filename: ":memory:" })),
329 ),
330 ),
331 ),
332);
333
334it.effect(
335 "dates a settlement from its first event, keeps duplicate settlements and imports stable, and resets on activation",
336 () =>
337 Effect.gen(function* () {
338 const store = yield* OrchestrationEventStore.OrchestrationEventStore;
339 const threadId = ThreadId.make("weekly-settlement");
340 const settled = { settledOverride: "settled", settledAt: "2026-08-01T00:00:00.000Z" };
341 const first = "2026-09-04T17:00:00.000Z";
342 const second = "2026-09-06T17:00:00.000Z";
343 const event = (ordinal: number, eventType: string, at: string, payload: unknown) => ({
344 ...eventRow(ordinal, "thread", 2, threadId),
345 event_type: eventType,
346 occurred_at: at,
347 payload_json: encodeJson(payload),
348 });
349 assert.equal(yield* store.getThreadSettlementTime(threadId), null);
350 yield* seedEvents([
351 event(1001, "thread.created", occurredAt, { settledOverride: null, settledAt: null }),
352 event(1002, "thread.settled", first, settled),
353 event(1003, "thread.settled", second, settled),
354 ]);
355 assert.deepStrictEqual(
356 yield* store.getThreadSettlementTime(threadId),
357 DateTime.makeUnsafe(first),
358 );
359 yield* seedEvents([
360 event(1004, "thread.unsettled", second, { settledOverride: "active", settledAt: null }),
361 event(1005, "thread.settled", second, settled),
362 ]);
363 assert.deepStrictEqual(
364 yield* store.getThreadSettlementTime(threadId),
365 DateTime.makeUnsafe(second),
366 );
367 const imported = ThreadId.make("legacy-settlement");
368 yield* seedEvents([
369 {
370 ...eventRow(1006, "thread", 2, imported),
371 event_type: "thread.created",
372 payload_json: encodeJson(settled),
373 },
374 ]);
375 assert.deepStrictEqual(
376 yield* store.getThreadSettlementTime(imported),
377 DateTime.makeUnsafe(settled.settledAt),
378 );
379 }).pipe(Effect.provide(layerEventStore)),
380);