1import { assert, it } from "@effect/vitest";
2import {
3 EventId,
4 NodeId,
5 PlanId,
6 type OrchestrationV2AppThread,
7 type OrchestrationV2DomainEvent,
8 type OrchestrationV2TurnItem,
9 ProjectId,
10 ProviderDriverKind,
11 ProviderInstanceId,
12 ThreadId,
13 TurnItemId,
14} from "@t3tools/contracts";
15import * as DateTime from "effect/DateTime";
16import * as Deferred from "effect/Deferred";
17import * as Effect from "effect/Effect";
18import * as Fiber from "effect/Fiber";
19import * as Layer from "effect/Layer";
20import * as Stream from "effect/Stream";
21import * as TestClock from "effect/testing/TestClock";
22import * as SqlClient from "effect/sql/SqlClient";
23import * as Statement from "effect/sql/Statement";
24
25import * as SqlitePersistence from "../persistence/Sqlite.ts";
26import * as EventSink from "./EventSink.ts";
27import * as EventStore from "./EventStore.ts";
28import * as ProjectionMaintenance from "./ProjectionMaintenance.ts";
29import * as ProjectionStore from "./ProjectionStore.ts";
30import * as OrchestrationEventStore from "../persistence/OrchestrationEventStore.ts";
31
32const database = SqlitePersistence.layerMemory;
33const eventStoreProvided = EventStore.layer.pipe(Layer.provideMerge(database));
34const projectionStoreProvided = ProjectionStore.layer.pipe(Layer.provideMerge(database));
35const stores = Layer.mergeAll(database, eventStoreProvided, projectionStoreProvided);
36const layerTest = Layer.mergeAll(
37 stores,
38 OrchestrationEventStore.layer.pipe(Layer.provide(database)),
39 EventSink.layer.pipe(Layer.provide(stores)),
40 ProjectionMaintenance.layer.pipe(Layer.provide(stores)),
41);
42
43const providerInstanceId = ProviderInstanceId.make("codex");
44const providerDriver = ProviderDriverKind.make("codex");
45const modelSelection = { instanceId: providerInstanceId, model: "gpt-5.4" };
46
47function makeThread(threadId: ThreadId, now: DateTime.Utc): OrchestrationV2AppThread {
48 return {
49 id: threadId,
50 projectId: ProjectId.make(`project:${threadId}`),
51 title: "Turn item compaction",
52 providerInstanceId,
53 modelSelection,
54 runtimeMode: "full-access",
55 interactionMode: "default",
56 branch: null,
57 worktreePath: null,
58 activeProviderThreadId: null,
59 lineage: { rootThreadId: threadId, parentThreadId: null, relationshipToParent: null },
60 forkedFrom: null,
61 createdBy: "user",
62 creationSource: "web",
63 createdAt: now,
64 updatedAt: now,
65 archivedAt: null,
66 settledOverride: null,
67 settledAt: null,
68 lastVisitedAt: null,
69 deletedAt: null,
70 };
71}
72
73function threadCreated(thread: OrchestrationV2AppThread, now: DateTime.Utc) {
74 return {
75 id: EventId.make("event:turn-item-compaction:thread"),
76 type: "thread.created",
77 threadId: thread.id,
78 providerInstanceId,
79 occurredAt: now,
80 payload: thread,
81 } satisfies OrchestrationV2DomainEvent;
82}
83
84const itemSpecs = [
85 { key: "command", type: "command_execution", ordinal: 10 },
86 { key: "reasoning", type: "reasoning", ordinal: 20 },
87 { key: "todo", type: "todo_list", ordinal: 30 },
88 { key: "commanded", type: "command_execution", ordinal: 40 },
89] as const;
90
91function itemEvent(
92 threadId: ThreadId,
93 now: DateTime.Utc,
94 spec: (typeof itemSpecs)[number],
95 revision: number,
96 eventId = `event:turn-item-compaction:${spec.key}:${revision}`,
97): OrchestrationV2DomainEvent {
98 const id = TurnItemId.make(`turn-item:compaction:${spec.key}`);
99 const growing = `${spec.key}:${revision}:` + "x".repeat((revision + 1) * 192);
100 const completed = revision === 69;
101 const base = {
102 id,
103 threadId,
104 runId: null,
105 nodeId: null,
106 providerThreadId: null,
107 providerTurnId: null,
108 nativeItemRef: null,
109 parentItemId: null,
110 ordinal: spec.ordinal,
111 status: completed ? ("completed" as const) : ("running" as const),
112 title: spec.key,
113 startedAt: now,
114 completedAt: completed ? now : null,
115 updatedAt: now,
116 };
117 let payload: OrchestrationV2TurnItem;
118 switch (spec.type) {
119 case "command_execution":
120 payload = {
121 ...base,
122 type: "command_execution",
123 input: `run ${spec.key}`,
124 output: growing,
125 };
126 break;
127 case "reasoning":
128 payload = { ...base, type: "reasoning", text: growing, streaming: !completed };
129 break;
130 case "todo_list":
131 payload = {
132 ...base,
133 type: "todo_list",
134 planId: PlanId.make("plan:turn-item-compaction"),
135 steps: [{ id: "step", text: growing, status: completed ? "completed" : "running" }],
136 explanation: growing,
137 };
138 break;
139 }
140 return {
141 id: EventId.make(eventId),
142 type: "turn-item.updated",
143 threadId,
144 occurredAt: now,
145 payload,
146 };
147}
148
149function subagentEvents(
150 threadId: ThreadId,
151 now: DateTime.Utc,
152 revision: number,
153 eventSuffix = String(revision),
154): ReadonlyArray<OrchestrationV2DomainEvent> {
155 const id = NodeId.make("subagent:turn-item-compaction");
156 const prompt = revision === 0 ? "Review prompt A. ".repeat(40) : "Review prompt B. ".repeat(40);
157 const progress = `Review progress ${revision}`;
158 return [
159 {
160 id: EventId.make(`event:turn-item-compaction:subagent:${eventSuffix}`),
161 type: "subagent.updated",
162 threadId,
163 occurredAt: now,
164 payload: {
165 id,
166 threadId,
167 runId: null,
168 parentNodeId: NodeId.make("node:turn-item-compaction-parent"),
169 origin: "provider_native",
170 createdBy: "agent",
171 driver: providerDriver,
172 providerInstanceId,
173 providerThreadId: null,
174 childThreadId: null,
175 nativeTaskRef: {
176 driver: providerDriver,
177 nativeId: "compaction-review",
178 strength: "strong",
179 },
180 prompt,
181 progress,
182 title: "Review",
183 model: null,
184 status: "running",
185 result: null,
186 startedAt: now,
187 completedAt: null,
188 updatedAt: now,
189 },
190 },
191 {
192 id: EventId.make(`event:turn-item-compaction:subagent-item:${eventSuffix}`),
193 type: "turn-item.updated",
194 threadId,
195 occurredAt: now,
196 payload: {
197 id: TurnItemId.make("turn-item:compaction:subagent"),
198 threadId,
199 runId: null,
200 nodeId: null,
201 providerThreadId: null,
202 providerTurnId: null,
203 nativeItemRef: null,
204 parentItemId: null,
205 ordinal: 50,
206 status: "running",
207 title: "Review",
208 startedAt: now,
209 completedAt: null,
210 updatedAt: now,
211 type: "subagent",
212 subagentId: id,
213 origin: "provider_native",
214 driver: providerDriver,
215 providerInstanceId,
216 childThreadId: null,
217 prompt,
218 progress,
219 result: null,
220 },
221 },
222 ] satisfies ReadonlyArray<OrchestrationV2DomainEvent>;
223}
224
225const insertRunningRunUpdate = (threadId: ThreadId) =>
226 Effect.flatMap(
227 SqlClient.SqlClient,
228 (sql) => sql<{ readonly sequence: number }>`
229 INSERT INTO orchestration_events (
230 event_id, aggregate_kind, stream_id, stream_version, event_type,
231 occurred_at, actor_kind, payload_json, metadata_json, application_event_version
232 ) VALUES (
233 'event:prune:run-running', 'thread', ${threadId},
234 (SELECT MAX(stream_version) + 1 FROM orchestration_events WHERE stream_id = ${threadId}),
235 'run.updated', '2026-11-01T00:00:00.000Z', 'server', '{"status":"running"}', '{}', 2
236 ) RETURNING sequence
237 `,
238 );
239
240const remainingEventIds = (threadId: ThreadId) =>
241 Effect.flatMap(
242 SqlClient.SqlClient,
243 (sql) => sql<{ readonly event_id: string }>`
244 SELECT event_id FROM orchestration_events WHERE stream_id = ${threadId} ORDER BY sequence
245 `,
246 ).pipe(Effect.map((rows) => rows.map((row) => row.event_id)));
247
248it.effect("prunes events the projections cover once the retained day has passed", () =>
249 Effect.gen(function* () {
250 const sink = yield* EventSink.EventSinkV2;
251 const eventStore = yield* EventStore.EventStoreV2;
252 const events = yield* OrchestrationEventStore.OrchestrationEventStore;
253 const projections = yield* ProjectionStore.ProjectionStoreV2;
254 const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2;
255 const now = yield* DateTime.now;
256 const threadId = ThreadId.make("thread:prune");
257 yield* sink.write({ events: [threadCreated(makeThread(threadId, now), now)] });
258 for (const revision of [0, 1, 2]) {
259 yield* sink.write({ events: [itemEvent(threadId, now, itemSpecs[0], revision)] });
260 }
261 yield* sink.write({ events: subagentEvents(threadId, now, 0) });
262 yield* insertRunningRunUpdate(threadId);
263 yield* sink.write({ events: [itemEvent(threadId, now, itemSpecs[1], 0)] });
264
265 // The first pass only marks the head; nothing is a day old yet.
266 assert.equal((yield* maintenance.pruneEventStore()).deletedEventCount, 0);
267 const markedThrough = yield* eventStore.latestSequence();
268
269 // Repeating prompt A borrows it from the anchors written before the mark.
270 yield* sink.write({ events: subagentEvents(threadId, now, 0, "repeat") });
271 yield* sink.write({ events: [itemEvent(threadId, now, itemSpecs[0], 3)] });
272 const highWater = yield* eventStore.latestSequence();
273 const projection = yield* projections.getThreadProjection(threadId);
274
275 yield* TestClock.adjust("23 hours");
276 assert.equal((yield* maintenance.pruneEventStore()).deletedEventCount, 0);
277 yield* TestClock.adjust("1 hour");
278 assert.equal((yield* maintenance.pruneEventStore()).deletedEventCount, 4);
279
280 assert.deepEqual(yield* remainingEventIds(threadId), [
281 "event:turn-item-compaction:thread",
282 "event:turn-item-compaction:subagent:0",
283 "event:turn-item-compaction:subagent-item:0",
284 "event:prune:run-running",
285 "event:turn-item-compaction:subagent:repeat",
286 "event:turn-item-compaction:subagent-item:repeat",
287 "event:turn-item-compaction:command:3",
288 ]);
289 assert.equal(yield* events.prunedThroughSequence, markedThrough);
290 assert.equal(yield* eventStore.latestSequence(), highWater);
291 assert.equal(yield* eventStore.latestSequence({ threadId }), highWater);
292 assert.deepEqual(yield* projections.getThreadProjection(threadId), projection);
293 const repeated = yield* eventStore
294 .read({ threadId, eventType: "subagent.updated", afterSequence: markedThrough })
295 .pipe(Stream.runCollect);
296 assert.deepEqual(
297 Array.from(repeated, (stored) =>
298 stored.event.type === "subagent.updated" ? stored.event.payload.prompt : null,
299 ),
300 ["Review prompt A. ".repeat(40)],
301 );
302 }).pipe(Effect.provide(layerTest)),
303);
304
305it.effect("keeps the anchor a retained stream head borrows from", () =>
306 Effect.gen(function* () {
307 const sink = yield* EventSink.EventSinkV2;
308 const eventStore = yield* EventStore.EventStoreV2;
309 const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2;
310 const now = yield* DateTime.now;
311 const threadId = ThreadId.make("thread:prune-head");
312 yield* sink.write({ events: [threadCreated(makeThread(threadId, now), now)] });
313 yield* sink.write({ events: subagentEvents(threadId, now, 0) });
314 // The stream's newest row borrows prompt A, so it is kept below the boundary.
315 yield* sink.write({ events: subagentEvents(threadId, now, 0, "repeat") });
316 yield* maintenance.pruneEventStore();
317 yield* TestClock.adjust("1 day");
318 assert.isAbove((yield* maintenance.pruneEventStore()).deletedEventCount, 0);
319
320 assert.include(
321 yield* remainingEventIds(threadId),
322 "event:turn-item-compaction:subagent-item:0",
323 );
324 const stored = yield* eventStore.read({ threadId }).pipe(Stream.runCollect);
325 const head = Array.from(stored).at(-1)?.event;
326 assert.deepEqual(
327 head?.type === "turn-item.updated" && head.payload.type === "subagent"
328 ? head.payload.prompt
329 : null,
330 "Review prompt A. ".repeat(40),
331 );
332 }).pipe(Effect.provide(layerTest)),
333);
334
335it.effect("keeps an anchor that a row appended during the pass borrows from", () =>
336 Effect.scoped(
337 Effect.gen(function* () {
338 const sink = yield* EventSink.EventSinkV2;
339 const eventStore = yield* EventStore.EventStoreV2;
340 const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2;
341 const now = yield* DateTime.now;
342 const threadId = ThreadId.make("thread:prune-race");
343 yield* sink.write({ events: [threadCreated(makeThread(threadId, now), now)] });
344 yield* sink.write({ events: subagentEvents(threadId, now, 0) });
345 yield* maintenance.pruneEventStore();
346 yield* sink.write({ events: [itemEvent(threadId, now, itemSpecs[0], 0)] });
347 yield* TestClock.adjust("1 day");
348
349 const scanStarted = yield* Deferred.make<void>();
350 const releaseScan = yield* Deferred.make<void>();
351 let blocked = false;
352 const blockFirstScanPage: Statement.Transformer = (statement) => {
353 const [query] = statement.compile();
354 if (!blocked && query.includes("END AS prunable")) {
355 blocked = true;
356 return Deferred.succeed(scanStarted, undefined).pipe(
357 Effect.andThen(Deferred.await(releaseScan)),
358 Effect.as(statement),
359 );
360 }
361 return Effect.succeed(statement);
362 };
363 const pass = yield* maintenance
364 .pruneEventStore()
365 .pipe(
366 Effect.provideService(Statement.CurrentTransformer, blockFirstScanPage),
367 Effect.forkScoped,
368 );
369 yield* Deferred.await(scanStarted);
370 yield* sink.write({ events: subagentEvents(threadId, now, 0, "late") });
371 yield* Deferred.succeed(releaseScan, undefined);
372 assert.equal((yield* Fiber.join(pass)).deletedEventCount, 0);
373
374 const late = yield* eventStore
375 .read({ threadId, eventType: "subagent.updated" })
376 .pipe(Stream.runCollect);
377 assert.deepEqual(
378 Array.from(late, (stored) =>
379 stored.event.type === "subagent.updated" ? stored.event.payload.prompt : null,
380 ),
381 ["Review prompt A. ".repeat(40), "Review prompt A. ".repeat(40)],
382 );
383 }),
384 ).pipe(Effect.provide(layerTest)),
385);