1import { CommandId, EventId, ProviderSessionId, ThreadId } from "@t3tools/contracts";
2import { assert, it } from "@effect/vitest";
3import * as Arr from "effect/Array";
4import * as DateTime from "effect/DateTime";
5import * as Effect from "effect/Effect";
6import * as Layer from "effect/Layer";
7import * as Schema from "effect/Schema";
8import * as Stream from "effect/Stream";
9import * as SqlClient from "effect/sql/SqlClient";
10
11import * as OrchestrationEventStore from "./OrchestrationEventStore.ts";
12import * as SqlitePersistence from "./Sqlite.ts";
13
14const layerEventStore = OrchestrationEventStore.layer.pipe(
15 Layer.provideMerge(SqlitePersistence.layerMemory),
16);
17const occurredAt = "2026-09-03T00:00:00.000Z";
18const encodeJson = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown));
19
20function eventRow(
21 ordinal: number,
22 aggregateKind: "project" | "thread",
23 version: number,
24 streamId = `${aggregateKind}:${ordinal}`,
25 commandId: string | null = null,
26) {
27 return {
28 event_id: `event:${ordinal}`,
29 aggregate_kind: aggregateKind,
30 stream_id: streamId,
31 stream_version: ordinal,
32 event_type: aggregateKind === "project" ? "project.created" : "provider-session.detached",
33 occurred_at: occurredAt,
34 command_id: commandId,
35 causation_event_id: null,
36 correlation_id: null,
37 actor_kind: "server",
38 payload_json: encodeJson(
39 aggregateKind === "project"
40 ? {
41 projectId: streamId,
42 title: "Sequence fixture",
43 workspaceRoot: "/tmp/sequence-fixture",
44 defaultModelSelection: null,
45 scripts: [],
46 createdAt: occurredAt,
47 updatedAt: occurredAt,
48 }
49 : { providerSessionId: `session:${ordinal}`, detachedAt: occurredAt },
50 ),
51 metadata_json: "{}",
52 application_event_version: version,
53 };
54}
55
56const seedEvents = Effect.fn("test.seedSequenceEvents")(function* (
57 rows: Arr.NonEmptyReadonlyArray<ReturnType<typeof eventRow>>,
58) {
59 const sql = yield* SqlClient.SqlClient;
60 return yield* sql<{
61 readonly sequence: number;
62 readonly aggregate_kind: string;
63 readonly application_event_version: number;
64 readonly stream_id: string;
65 readonly command_id: string | null;
66 }>`
67 INSERT INTO orchestration_events ${sql.insert(rows)}
68 RETURNING sequence, aggregate_kind, application_event_version, stream_id, command_id
69 `;
70});
71
72it.effect("keeps application and scoped agent high-water marks separate from legacy history", () =>
73 Effect.gen(function* () {
74 const store = yield* OrchestrationEventStore.OrchestrationEventStore;
75 const sql = yield* SqlClient.SqlClient;
76 const target = ThreadId.make("target");
77 assert.equal(yield* store.latestApplicationSequence, 0);
78 assert.equal(yield* store.latestAgentSequence(), 0);
79 assert.equal(yield* store.latestAgentSequence(target), 0);
80
81 yield* seedEvents([eventRow(1, "thread", 1, "legacy-only")]);
82 assert.equal(yield* store.latestApplicationSequence, 0);
83 assert.equal(yield* store.latestAgentSequence(ThreadId.make("legacy-only")), 0);
84
85 const [legacyProject, targetEvent, otherEvent] = yield* seedEvents([
86 eventRow(2, "project", 1),
87 eventRow(3, "thread", 2, target),
88 eventRow(4, "thread", 2, "other"),
89 ]);
90 assert.ok(legacyProject);
91 assert.ok(targetEvent);
92 assert.ok(otherEvent);
93 assert.equal(yield* store.latestApplicationSequence, otherEvent.sequence);
94 assert.equal(yield* store.latestAgentSequence(), otherEvent.sequence);
95 assert.equal(yield* store.latestAgentSequence(target), targetEvent.sequence);
96
97 const [projectEvent] = yield* seedEvents([
98 eventRow(5, "project", 2),
99 eventRow(6, "thread", 1, target),
100 eventRow(7, "thread", 3, target),
101 ]);
102 assert.ok(projectEvent);
103 assert.equal(yield* store.latestApplicationSequence, projectEvent.sequence);
104 assert.equal(yield* store.latestAgentSequence(), otherEvent.sequence);
105 assert.equal(yield* store.latestAgentSequence(target), targetEvent.sequence);
106 assert.equal(yield* store.latestAgentSequence(ThreadId.make("missing")), 0);
107
108 yield* sql`DELETE FROM orchestration_events WHERE sequence > ${legacyProject.sequence}`;
109 assert.equal(yield* store.latestApplicationSequence, legacyProject.sequence);
110 assert.equal(yield* store.latestAgentSequence(), 0);
111 assert.equal(yield* store.latestAgentSequence(target), 0);
112 }).pipe(Effect.provide(Layer.fresh(layerEventStore))),
113);
114
115it.effect(
116 "preserves mixed application paging, scoped replay, and the catch-up to live boundary",
117 () =>
118 Effect.gen(function* () {
119 const store = yield* OrchestrationEventStore.OrchestrationEventStore;
120 const rows = yield* seedEvents(
121 Arr.makeBy(1_560, (index) => {
122 const kind = index % 3 === 1 ? "project" : "thread";
123 return eventRow(
124 index,
125 kind,
126 index % 3 === 0 ? 1 : 2,
127 kind === "project" ? `project:${index}` : index % 2 === 0 ? "target" : "other",
128 index % 5 === 0 ? "selected-command" : "other-command",
129 );
130 }),
131 );
132 const afterSequence = rows[25]!.sequence;
133 const throughSequence = rows[1_525]!.sequence;
134 const retained = rows.filter(
135 (row) => row.aggregate_kind === "project" || row.application_event_version === 2,
136 );
137 const replay = yield* store
138 .readApplicationEvents({ afterSequence, throughSequence })
139 .pipe(Stream.runCollect);
140 assert.deepEqual(
141 replay.map((event) => event.sequence),
142 retained
143 .filter((row) => row.sequence > afterSequence && row.sequence <= throughSequence)
144 .map((row) => row.sequence),
145 );
146
147 const scoped = yield* store
148 .readAgentEvents({
149 afterSequence,
150 throughSequence,
151 threadId: ThreadId.make("target"),
152 commandId: CommandId.make("selected-command"),
153 limit: 17,
154 })
155 .pipe(Stream.runCollect);
156 assert.deepEqual(
157 scoped.map((event) => event.sequence),
158 retained
159 .filter(
160 (row) =>
161 row.sequence > afterSequence &&
162 row.sequence <= throughSequence &&
163 row.aggregate_kind === "thread" &&
164 row.stream_id === "target" &&
165 row.command_id === "selected-command",
166 )
167 .slice(0, 17)
168 .map((row) => row.sequence),
169 );
170
171 const catchUp = retained.filter((row) => row.sequence > afterSequence);
172 const highWater = catchUp.at(-1)!.sequence;
173 let liveSequence = 0;
174 const streamed = yield* store.streamApplicationEvents({ afterSequence }).pipe(
175 Stream.tap((event) =>
176 event.sequence !== highWater
177 ? Effect.void
178 : Effect.gen(function* () {
179 const [stored] = yield* store.appendAgentEvents({
180 events: [
181 {
182 id: EventId.make("live-after-catch-up"),
183 type: "provider-session.detached",
184 threadId: ThreadId.make("target"),
185 occurredAt: DateTime.makeUnsafe(occurredAt),
186 payload: {
187 providerSessionId: ProviderSessionId.make("live-session"),
188 detachedAt: DateTime.makeUnsafe(occurredAt),
189 },
190 },
191 ],
192 });
193 assert.ok(stored);
194 liveSequence = stored.sequence;
195 yield* store.publishCommitted([stored]);
196 }),
197 ),
198 Stream.take(catchUp.length + 1),
199 Stream.runCollect,
200 );
201 assert.deepEqual(
202 streamed.map((event) => event.sequence),
203 [...catchUp.map((row) => row.sequence), liveSequence],
204 );
205 assert.isAbove(liveSequence, highWater);
206 }).pipe(Effect.provide(Layer.fresh(layerEventStore))),
207);