1import { assert, it } from "@effect/vitest";
2import { MessageId, ProviderDriverKind, ProviderTurnId, ThreadId } from "@t3tools/contracts";
3import * as KeyedLock from "@t3tools/shared/KeyedLock";
4import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
5import * as Cause from "effect/Cause";
6import * as DateTime from "effect/DateTime";
7import * as Deferred from "effect/Deferred";
8import * as Effect from "effect/Effect";
9import * as Exit from "effect/Exit";
10import * as Fiber from "effect/Fiber";
11import * as Queue from "effect/Queue";
12import * as Stream from "effect/Stream";
13import * as TestClock from "effect/testing/TestClock";
14import * as ProviderCommitBatch from "./ProviderCommitBatch.ts";
15
16const driver = ProviderDriverKind.make("codex");
17function event(index: number, text = `delta:${index}`): ProviderAdapter.ProviderAdapterV2Event {
18 const threadId = ThreadId.make("single-thread");
19 const now = DateTime.makeUnsafe("2026-11-03T00:00:00Z");
20 return {
21 type: "message.updated",
22 driver,
23 message: {
24 id: MessageId.make("single-message"),
25 threadId,
26 runId: null,
27 nodeId: null,
28 createdBy: "agent",
29 creationSource: "provider",
30 role: "assistant",
31 text,
32 attachments: [],
33 streaming: true,
34 createdAt: now,
35 updatedAt: now,
36 },
37 };
38}
39const control: ProviderAdapter.ProviderAdapterV2Event = {
40 type: "background_work.updated",
41 driver,
42 providerTurnId: ProviderTurnId.make("turn"),
43 pending: false,
44};
45
46const recordBatches = Effect.fn("test.recordProviderBatches")(function* (
47 beforeCommit: Effect.Effect<void> = Effect.void,
48) {
49 const output = yield* Queue.unbounded<ReadonlyArray<ProviderAdapter.ProviderAdapterV2Event>>();
50 let current: ProviderAdapter.ProviderAdapterV2Event[] = [];
51 const coordinator = yield* ProviderCommitBatch.make(
52 (work) =>
53 Effect.gen(function* () {
54 yield* beforeCommit;
55 current = [];
56 yield* work;
57 if (current.length > 0) yield* Queue.offer(output, current);
58 }),
59 true,
60 );
61 const process = (raw: ProviderAdapter.ProviderAdapterV2Event) =>
62 raw.type === "background_work.updated"
63 ? Queue.offer(output, [raw]).pipe(Effect.asVoid)
64 : Effect.sync(() => {
65 current.push(raw);
66 });
67 return { ...coordinator, process, output };
68});
69
70it.effect("publishes preceding snapshots before a control", () =>
71 Effect.gen(function* () {
72 const recorder = yield* recordBatches();
73 const input = yield* Queue.unbounded<ProviderAdapter.ProviderAdapterV2Event>();
74 const consumer = yield* recorder
75 .consume(Stream.fromQueue(input), recorder.process, Effect.succeed(true))
76 .pipe(Effect.forkScoped);
77 yield* Queue.offer(input, event(1));
78 assert.deepEqual(yield* Queue.take(recorder.output), [event(1)]);
79 yield* Queue.offer(input, event(2));
80 yield* Queue.offer(input, control);
81 assert.deepEqual(yield* Queue.take(recorder.output), [event(2)]);
82 assert.deepEqual(yield* Queue.take(recorder.output), [control]);
83 yield* Fiber.join(consumer);
84 }).pipe(Effect.scoped),
85);
86
87it.effect("publishes isolated snapshots without advancing the clock", () =>
88 Effect.gen(function* () {
89 const recorder = yield* recordBatches();
90 const input = yield* Queue.unbounded<ProviderAdapter.ProviderAdapterV2Event>();
91 const pulled = yield* Queue.unbounded<ProviderAdapter.ProviderAdapterV2Event>();
92 yield* recorder
93 .consume(
94 Stream.fromQueue(input).pipe(Stream.tap((raw) => Queue.offer(pulled, raw))),
95 recorder.process,
96 Effect.succeed(false),
97 )
98 .pipe(Effect.forkScoped);
99 for (const raw of [event(1), event(2)]) {
100 yield* Queue.offer(input, raw);
101 yield* Queue.take(pulled);
102 yield* TestClock.adjust(0);
103 const published = yield* Queue.poll(recorder.output);
104 assert.equal(published._tag, "Some");
105 if (published._tag === "Some") assert.deepEqual(published.value, [raw]);
106 }
107 }).pipe(Effect.scoped),
108);
109
110it.effect("bounds transactions by count and bytes, including the final partial group", () =>
111 Effect.gen(function* () {
112 const workloads = [
113 { events: Array.from({ length: 129 }, (_, index) => event(index)) },
114 { events: [event(0), event(1, "x".repeat(1024 * 1024)), event(2)] },
115 {
116 events: [event(0, "x".repeat(600_000)), event(1, "y".repeat(600_000)), event(2)],
117 },
118 ];
119 for (const { events } of workloads) {
120 const recorder = yield* recordBatches();
121 const pulled = yield* Deferred.make<void>();
122 let arrivals = 0;
123 const consumer = yield* recorder
124 .consume(
125 Stream.fromArray(events).pipe(
126 Stream.tap(() =>
127 ++arrivals === events.length ? Deferred.succeed(pulled, undefined) : Effect.void,
128 ),
129 ),
130 recorder.process,
131 Effect.succeed(false),
132 )
133 .pipe(Effect.forkScoped({ startImmediately: true }));
134 yield* Deferred.await(pulled);
135 yield* TestClock.adjust(0);
136 yield* recorder.barrier(Effect.void);
137 yield* Fiber.join(consumer);
138 const batches = yield* Queue.takeAll(recorder.output);
139 assert.isTrue(batches.every((batch) => batch.length <= 64));
140 if (events.length === 129) assert.isTrue(batches.some((batch) => batch.length > 1));
141 assert.isTrue(
142 batches.every(
143 (batch) =>
144 batch.length === 1 ||
145 batch.reduce((bytes, raw) => bytes + Buffer.byteLength(JSON.stringify(raw)), 0) <=
146 1024 * 1024,
147 ),
148 );
149 assert.deepEqual(batches.flat(), events);
150 }
151 }).pipe(Effect.scoped),
152);
153
154it.effect("keeps updates immediate and ordered when batching is disabled", () =>
155 Effect.gen(function* () {
156 const seen: ProviderAdapter.ProviderAdapterV2Event[] = [];
157 const coordinator = yield* ProviderCommitBatch.make(() => Effect.die("must not batch"), false);
158 yield* coordinator.consume(
159 Stream.make(event(1), control, event(2)),
160 (raw) =>
161 Effect.sync(() => {
162 seen.push(raw);
163 }),
164 Effect.succeed(false),
165 );
166 assert.deepEqual(seen, [event(1), control, event(2)]);
167 }).pipe(Effect.scoped),
168);
169
170it.effect.each([
171 { streams: 1, updates: 10 },
172 { streams: 10, updates: 1 },
173 { streams: 10, updates: 10 },
174])(
175 "shares bounded transactions across $streams streams with $updates updates each",
176 ({ streams, updates }) =>
177 Effect.gen(function* () {
178 const coordinator = yield* recordBatches();
179 const inputs = yield* Effect.forEach(Array.from({ length: streams }), () =>
180 Queue.unbounded<ProviderAdapter.ProviderAdapterV2Event>(),
181 );
182 const pulled = yield* Deferred.make<void>();
183 let arrivals = 0;
184 const consumers = yield* Effect.forEach(inputs, (input) =>
185 coordinator
186 .consume(
187 Stream.fromQueue(input).pipe(
188 Stream.tap(() => {
189 arrivals++;
190 return arrivals === streams * updates
191 ? Deferred.succeed(pulled, undefined)
192 : Effect.void;
193 }),
194 ),
195 coordinator.process,
196 Effect.succeed(true),
197 )
198 .pipe(Effect.forkScoped({ startImmediately: true })),
199 );
200 for (let n = 0; n < updates; n++) {
201 for (let stream = 0; stream < streams; stream++)
202 yield* Queue.offer(inputs[stream]!, event(stream * updates + n));
203 }
204 yield* Deferred.await(pulled);
205 yield* TestClock.adjust(0);
206
207 yield* coordinator.barrier(Effect.void);
208 const batches: Array<ReadonlyArray<ProviderAdapter.ProviderAdapterV2Event>> = [];
209 let count = 0;
210 while (count < streams * updates) {
211 const batch = yield* Queue.take(coordinator.output);
212 batches.push(batch);
213 count += batch.length;
214 }
215 assert.deepEqual(
216 batches.map((batch) => batch.length),
217 streams * updates > 64 ? [64, 36] : [10],
218 );
219 assert.equal(
220 new Set(
221 batches
222 .flat()
223 .flatMap((raw) => (raw.type === "message.updated" ? [raw.message.text] : [])),
224 ).size,
225 streams * updates,
226 );
227 for (const input of inputs) yield* Queue.offer(input, control);
228 yield* Effect.forEach(consumers, Fiber.join);
229 }).pipe(Effect.scoped),
230);
231
232it.effect("does not hold the shared worker while one consumer checks background work", () =>
233 Effect.gen(function* () {
234 const checking = yield* Deferred.make<void>();
235 const release = yield* Deferred.make<void>();
236 const coordinator = yield* ProviderCommitBatch.make((work) => work, true);
237 const consumer = yield* coordinator
238 .consume(
239 Stream.make(control),
240 () => Effect.void,
241 Deferred.succeed(checking, undefined).pipe(
242 Effect.andThen(Deferred.await(release)),
243 Effect.as(true),
244 ),
245 )
246 .pipe(Effect.forkScoped({ startImmediately: true }));
247 yield* Deferred.await(checking);
248 assert.equal(
249 yield* coordinator.barrier(Effect.succeed("another run can commit")),
250 "another run can commit",
251 );
252 yield* Deferred.succeed(release, undefined);
253 yield* Fiber.join(consumer);
254 }).pipe(Effect.scoped),
255);
256
257it.effect.each(["job", "commit"] as const)(
258 "isolates a %s failure at its transaction boundary",
259 (mode) =>
260 Effect.gen(function* () {
261 const coordinator = yield* ProviderCommitBatch.make(
262 (work) =>
263 mode === "commit"
264 ? work.pipe(
265 Effect.andThen(
266 Effect.fail(new ProviderCommitBatch.CommitBatchError({ cause: "COMMIT failed" })),
267 ),
268 )
269 : work,
270 true,
271 );
272 const inputs = yield* Effect.forEach([0, 1], () =>
273 Queue.unbounded<ProviderAdapter.ProviderAdapterV2Event>(),
274 );
275 const consumers = yield* Effect.forEach(inputs, (input, index) =>
276 coordinator
277 .consume(
278 Stream.fromQueue(input),
279 () => (mode === "job" && index === 1 ? Effect.fail("invalid update") : Effect.void),
280 Effect.succeed(true),
281 )
282 .pipe(Effect.exit, Effect.forkScoped({ startImmediately: true })),
283 );
284 for (const input of inputs) yield* Queue.offer(input, event(0));
285 yield* TestClock.adjust(0);
286 yield* coordinator.barrier(Effect.void);
287 assert.isTrue(Exit.isFailure(yield* Fiber.join(consumers[1]!)));
288 if (mode === "commit") assert.isTrue(Exit.isFailure(yield* Fiber.join(consumers[0]!)));
289 else {
290 assert.isUndefined(consumers[0]!.pollUnsafe());
291 yield* Queue.offer(inputs[0]!, control);
292 assert.isTrue(Exit.isSuccess(yield* Fiber.join(consumers[0]!)));
293 }
294 assert.equal(yield* coordinator.barrier(Effect.succeed("recovered")), "recovered");
295 assert.deepEqual(
296 yield* coordinator.barrier(Effect.fail("original error")).pipe(Effect.exit),
297 Exit.fail("original error"),
298 );
299 }).pipe(Effect.scoped),
300);
301
302it.effect("keeps a transaction failure when rolling back one failed snapshot", () =>
303 Effect.gen(function* () {
304 const transactionError = new ProviderCommitBatch.CommitBatchError({ cause: "ROLLBACK failed" });
305 const coordinator = yield* ProviderCommitBatch.make(
306 (work) => work.pipe(Effect.mapError(() => transactionError)),
307 true,
308 );
309 const exit = yield* coordinator
310 .consume(
311 Stream.concat(Stream.make(event(1)), Stream.never),
312 () => Effect.fail("invalid snapshot"),
313 Effect.succeed(false),
314 )
315 .pipe(Effect.exit);
316 assert.isTrue(Exit.isFailure(exit));
317 if (Exit.isFailure(exit)) assert.strictEqual(Cause.squash(exit.cause), transactionError);
318 }).pipe(Effect.scoped),
319);
320
321it.effect(
322 "allows a command holding the thread lock to commit while a provider control waits for it",
323 () =>
324 Effect.gen(function* () {
325 const coordinator = yield* ProviderCommitBatch.make((work) => work, true);
326 const locks = yield* KeyedLock.make<ThreadId>();
327 const threadId = ThreadId.make("locked-thread");
328 const held = yield* Deferred.make<void>();
329 const waiting = yield* Deferred.make<void>();
330 const committed = yield* Deferred.make<void>();
331 const command = yield* locks
332 .withLock(
333 threadId,
334 Deferred.succeed(held, undefined).pipe(
335 Effect.andThen(Deferred.await(waiting)),
336 Effect.andThen(coordinator.barrier(Deferred.succeed(committed, undefined))),
337 ),
338 )
339 .pipe(Effect.forkScoped({ startImmediately: true }));
340 yield* Deferred.await(held);
341 const consumer = yield* coordinator
342 .consume(
343 Stream.make(control),
344 () =>
345 Deferred.succeed(waiting, undefined).pipe(
346 Effect.andThen(locks.withLock(threadId, Effect.void)),
347 ),
348 Effect.succeed(true),
349 )
350 .pipe(Effect.forkScoped({ startImmediately: true }));
351 yield* Deferred.await(waiting);
352 yield* TestClock.adjust(0);
353 assert.isTrue(yield* Deferred.isDone(committed));
354 yield* Fiber.join(command);
355 yield* Fiber.join(consumer);
356 }).pipe(Effect.scoped),
357);
358
359it.effect.each([false, true])(
360 "batches work accumulated behind a busy writer with cancellation=%s",
361 (cancel) =>
362 Effect.gen(function* () {
363 const started = yield* Deferred.make<void>();
364 const release = yield* Deferred.make<void>();
365 let first = true;
366 const recorder = yield* recordBatches(
367 Effect.gen(function* () {
368 if (!first) return;
369 first = false;
370 yield* Deferred.succeed(started, undefined);
371 yield* Deferred.await(release);
372 }),
373 );
374 const leading = yield* recorder
375 .consume(Stream.make(event(-1)), recorder.process, Effect.succeed(false))
376 .pipe(Effect.forkScoped({ startImmediately: true }));
377 yield* Deferred.await(started);
378 const pulled = yield* Deferred.make<void>();
379 let arrivals = 0;
380 const consumers = yield* Effect.forEach(Array.from({ length: 10 }), (_, index) =>
381 recorder
382 .consume(
383 Stream.make(event(index)).pipe(
384 Stream.tap(() =>
385 ++arrivals === 10 ? Deferred.succeed(pulled, undefined) : Effect.void,
386 ),
387 ),
388 recorder.process,
389 Effect.succeed(false),
390 )
391 .pipe(Effect.forkScoped({ startImmediately: true })),
392 );
393 yield* Deferred.await(pulled);
394 yield* TestClock.adjust(0);
395 if (cancel) yield* Fiber.interrupt(consumers[0]!);
396 assert.equal(yield* Queue.size(recorder.output), 0);
397 yield* Deferred.succeed(release, undefined);
398 yield* Fiber.join(leading);
399 yield* Effect.forEach(consumers.slice(cancel ? 1 : 0), Fiber.join);
400 assert.deepEqual(yield* Queue.takeAll(recorder.output), [
401 [event(-1)],
402 Array.from({ length: cancel ? 9 : 10 }, (_, index) => event(index + (cancel ? 1 : 0))),
403 ]);
404 }).pipe(Effect.scoped),
405);
406
407it.effect.each(["failure", "stop"] as const)(
408 "flushes the prefix and respects a source %s barrier",
409 (mode) =>
410 Effect.gen(function* () {
411 const seen: ProviderAdapter.ProviderAdapterV2Event[] = [];
412 const coordinator = yield* ProviderCommitBatch.make((work) => work, true);
413 const source =
414 mode === "failure"
415 ? Stream.concat(Stream.make(event(1), event(2)), Stream.fail("provider failed"))
416 : Stream.make(event(1), control, event(2));
417 const exit = yield* coordinator
418 .consume(
419 source,
420 (raw) =>
421 Effect.sync(() => {
422 seen.push(raw);
423 }),
424 Effect.succeed(mode === "stop"),
425 )
426 .pipe(Effect.exit);
427 assert.deepEqual(seen, mode === "failure" ? [event(1), event(2)] : [event(1), control]);
428 assert.equal(Exit.isFailure(exit), mode === "failure");
429 assert.equal(
430 yield* coordinator.barrier(Effect.succeed("still available")),
431 "still available",
432 );
433 }).pipe(Effect.scoped),
434);
435
436it.effect("does not force a separate commit for every consumer that ends", () =>
437 Effect.gen(function* () {
438 const pulled = yield* Deferred.make<void>();
439 let arrivals = 0;
440 const coordinator = yield* recordBatches();
441 for (let index = 0; index < 10; index++) {
442 yield* coordinator
443 .consume(
444 Stream.make(event(index)).pipe(
445 Stream.tap(() =>
446 ++arrivals === 10 ? Deferred.succeed(pulled, undefined) : Effect.void,
447 ),
448 ),
449 coordinator.process,
450 Effect.succeed(false),
451 )
452 .pipe(Effect.forkScoped({ startImmediately: true }));
453 }
454 yield* Deferred.await(pulled);
455 yield* TestClock.adjust(0);
456 yield* coordinator.barrier(Effect.void);
457 assert.deepEqual(
458 (yield* Queue.takeAll(coordinator.output)).map((batch) => batch.length),
459 [10],
460 );
461 }).pipe(Effect.scoped),
462);
463
464it.effect("settles calls made after the coordinator's scope closes", () =>
465 Effect.gen(function* () {
466 const coordinator = yield* ProviderCommitBatch.make((work) => work, true).pipe(Effect.scoped);
467 const exit = yield* coordinator
468 .barrier(Effect.die("must not execute after shutdown"))
469 .pipe(Effect.exit);
470 assert.isTrue(Exit.isFailure(exit) && Cause.hasInterruptsOnly(exit.cause));
471 }),
472);