1import {
2 EventId,
3 MessageId,
4 ThreadId,
5 RunId,
6 TurnItemId,
7 type OrchestrationV2ThreadStreamItem,
8} from "@t3tools/contracts";
9import { it } from "@effect/vitest";
10import * as Clock from "effect/Clock";
11import * as DateTime from "effect/DateTime";
12import * as Effect from "effect/Effect";
13import * as Stream from "effect/Stream";
14import * as TestClock from "effect/testing/TestClock";
15import { describe, expect } from "vite-plus/test";
16import {
17 coalesceLiveToolUpdatedEvents,
18 coalesceThreadLiveStream,
19 makeThreadLiveEventCoalescer,
20} from "./ThreadLiveEventCoalescer.ts";
21
22type ThreadEvent = Extract<OrchestrationV2ThreadStreamItem, { readonly kind: "event" }>;
23const threadId = ThreadId.make("thread-coalescer-test");
24const runId = RunId.make("run-coalescer-test");
25const now = DateTime.makeUnsafe("2026-01-01T00:00:01.000Z");
26const encodeEvent = JSON.stringify;
27
28function makeToolActivity(
29 sequence: number,
30 options: {
31 readonly kind?: "tool.updated" | "tool.completed";
32 readonly toolCallId?: string;
33 readonly turnId?: RunId;
34 } = {},
35): ThreadEvent {
36 const itemRunId = options.turnId ?? runId;
37 const status = options.kind === "tool.completed" ? "completed" : "running";
38 return {
39 kind: "event",
40 sequence,
41 event: {
42 id: EventId.make(`event-${sequence}`),
43 type: "turn-item.updated",
44 threadId,
45 runId: itemRunId,
46 occurredAt: now,
47 payload: {
48 id: TurnItemId.make(options.toolCallId ?? "call-edit"),
49 type: "command_execution",
50 threadId,
51 runId: itemRunId,
52 nodeId: null,
53 providerThreadId: null,
54 providerTurnId: null,
55 nativeItemRef: null,
56 parentItemId: null,
57 ordinal: 0,
58 status,
59 title: "Editing app.ts",
60 startedAt: now,
61 completedAt: status === "completed" ? now : null,
62 updatedAt: now,
63 input: "echo app.ts",
64 output: `output-${sequence}`,
65 },
66 },
67 };
68}
69
70function makeMessage(sequence: number, text = "Still working"): ThreadEvent {
71 return {
72 kind: "event",
73 sequence,
74 event: {
75 id: EventId.make(`event-${sequence}`),
76 type: "message.updated",
77 threadId,
78 runId,
79 occurredAt: now,
80 payload: {
81 id: MessageId.make(`message-${sequence}`),
82 threadId,
83 runId,
84 nodeId: null,
85 createdBy: "agent",
86 creationSource: "provider",
87 role: "assistant",
88 text,
89 attachments: [],
90 streaming: false,
91 createdAt: now,
92 updatedAt: now,
93 },
94 },
95 };
96}
97
98describe("ThreadLiveEventCoalescer", () => {
99 it("coalesces only calls with a stable toolCallId", () => {
100 const events = [
101 makeToolActivity(1, { toolCallId: "call-a" }),
102 makeToolActivity(2, { toolCallId: "call-b" }),
103 makeToolActivity(3, { toolCallId: "call-a" }),
104 ];
105
106 expect(coalesceLiveToolUpdatedEvents(events).map((event) => event.sequence)).toEqual([2, 3]);
107 });
108
109 it("does not coalesce stable tool calls across turns", () => {
110 const events = [
111 makeToolActivity(1, { turnId: RunId.make("turn-old") }),
112 makeToolActivity(2, { turnId: RunId.make("turn-new") }),
113 ];
114
115 expect(coalesceLiveToolUpdatedEvents(events).map((event) => event.sequence)).toEqual([1, 2]);
116 });
117
118 it("flushes a stable update run before a completion boundary", () => {
119 const events = [
120 makeToolActivity(1),
121 makeToolActivity(2),
122 makeToolActivity(3, { kind: "tool.completed" }),
123 makeToolActivity(4),
124 ];
125
126 expect(coalesceLiveToolUpdatedEvents(events).map((event) => event.sequence)).toEqual([2, 3, 4]);
127 });
128
129 it.effect("flushes pending tool updates as soon as an unrelated event arrives", () =>
130 Effect.scoped(
131 Effect.gen(function* () {
132 const coalescer = yield* makeThreadLiveEventCoalescer({ coalesceWindow: "500 millis" });
133 const startedAt = yield* Clock.currentTimeMillis;
134 yield* Effect.forEach(
135 Array.from({ length: 10 }, (_, index) => index + 2),
136 (sequence) => coalescer.offer(makeToolActivity(sequence)),
137 { discard: true },
138 );
139 yield* coalescer.offer(makeMessage(12));
140
141 expect(yield* Clock.currentTimeMillis).toBe(startedAt);
142 expect(
143 (yield* coalescer.stream.pipe(Stream.take(2), Stream.runCollect)).map((item) =>
144 item.kind === "event" ? item.sequence : item.kind,
145 ),
146 ).toEqual([11, 12]);
147 }),
148 ).pipe(Effect.provide(TestClock.layer())),
149 );
150
151 it.effect("flushes pending tool updates as soon as a synchronization marker arrives", () =>
152 Effect.scoped(
153 Effect.gen(function* () {
154 const coalescer = yield* makeThreadLiveEventCoalescer({ coalesceWindow: "500 millis" });
155 const startedAt = yield* Clock.currentTimeMillis;
156 yield* coalescer.offer(makeToolActivity(2));
157 yield* coalescer.offer(makeToolActivity(3));
158 yield* coalescer.offer({ kind: "synchronized" });
159
160 expect(yield* Clock.currentTimeMillis).toBe(startedAt);
161 expect(
162 (yield* coalescer.stream.pipe(Stream.take(2), Stream.runCollect)).map((item) =>
163 item.kind === "event" ? item.sequence : item.kind,
164 ),
165 ).toEqual([3, "synchronized"]);
166 }),
167 ).pipe(Effect.provide(TestClock.layer())),
168 );
169
170 it.effect("fails and clears pending updates when their serialized payload fills the budget", () =>
171 Effect.scoped(
172 Effect.gen(function* () {
173 const first = makeToolActivity(1);
174 const coalescer = yield* makeThreadLiveEventCoalescer({
175 coalesceWindow: "500 millis",
176 maxSerializedBytes: Buffer.byteLength(encodeEvent(first)),
177 });
178 yield* coalescer.offer(first);
179 expect(yield* coalescer.usage).toEqual({
180 retainedItems: 1,
181 retainedSerializedBytes: Buffer.byteLength(encodeEvent(first)),
182 });
183
184 const overflow = yield* coalescer.offer(makeToolActivity(2)).pipe(Effect.result);
185 expect(overflow._tag).toBe("Failure");
186 yield* coalescer.closed;
187 expect(yield* coalescer.usage).toEqual({ retainedItems: 0, retainedSerializedBytes: 0 });
188 const marker = yield* coalescer.offer({ kind: "synchronized" }).pipe(Effect.result);
189 expect(marker._tag).toBe("Failure");
190 const delivered = yield* coalescer.stream.pipe(Stream.runCollect, Effect.result);
191 expect(delivered._tag).toBe("Failure");
192 }),
193 ),
194 );
195
196 it.effect("keeps the flush timer alive when an offer's shorter scope closes", () =>
197 Effect.scoped(
198 Effect.gen(function* () {
199 const coalescer = yield* makeThreadLiveEventCoalescer({ coalesceWindow: "50 millis" });
200 yield* Effect.scoped(coalescer.offer(makeToolActivity(1)));
201 yield* TestClock.adjust("50 millis");
202 const items = yield* coalescer.stream.pipe(Stream.take(1), Stream.runCollect);
203 expect(items.map((item) => (item.kind === "event" ? item.sequence : item.kind))).toEqual([
204 1,
205 ]);
206 }),
207 ),
208 );
209
210 it.effect("keeps an unacknowledged batch charged and clears later events on overflow", () =>
211 Effect.scoped(
212 Effect.gen(function* () {
213 const coalescer = yield* makeThreadLiveEventCoalescer({ maxItems: 3 });
214 const first = makeMessage(1, "é".repeat(1_024));
215 yield* coalescer.offer(first);
216
217 yield* Effect.scoped(
218 Effect.gen(function* () {
219 const pull = yield* Stream.toPull(coalescer.stream);
220 const batch = yield* pull;
221 expect(batch.map((item) => (item.kind === "event" ? item.sequence : null))).toEqual([
222 1,
223 ]);
224 yield* coalescer.offer(makeMessage(2));
225 yield* coalescer.offer(makeToolActivity(3));
226 expect((yield* coalescer.usage).retainedItems).toBe(3);
227
228 const overflow = yield* coalescer
229 .offer(makeToolActivity(4, { kind: "tool.completed" }))
230 .pipe(Effect.result);
231 expect(overflow._tag).toBe("Failure");
232 // Do not pull or acknowledge the batch. Cleanup must still finish.
233 yield* coalescer.closed;
234 expect(yield* coalescer.usage).toEqual({
235 retainedItems: 1,
236 retainedSerializedBytes: Buffer.byteLength(encodeEvent(first)),
237 });
238 }),
239 );
240 expect(yield* coalescer.usage).toEqual({ retainedItems: 0, retainedSerializedBytes: 0 });
241 }),
242 ),
243 );
244});
245
246it.effect("flushes the last update before a finite source ends", () =>
247 Effect.gen(function* () {
248 const result = yield* Stream.make(makeToolActivity(1), makeToolActivity(2)).pipe(
249 coalesceThreadLiveStream,
250 Stream.runCollect,
251 );
252 expect(result.map((item) => (item.kind === "event" ? item.sequence : null))).toEqual([2]);
253 }),
254);
255
256it.effect("propagates an upstream failure instead of leaving the subscriber waiting", () =>
257 Effect.gen(function* () {
258 const result = yield* Stream.fail("source stopped").pipe(
259 coalesceThreadLiveStream,
260 Stream.runCollect,
261 Effect.result,
262 );
263 expect(result._tag).toBe("Failure");
264 if (result._tag === "Failure") expect(result.failure).toBe("source stopped");
265 }),
266);