1import {
2 EventId,
3 MessageId,
4 ProjectId,
5 ProviderInstanceId,
6 type OrchestrationV2ThreadProjection,
7 ThreadId,
8 RunId,
9 TurnItemId,
10 type OrchestrationV2ThreadStreamItem,
11 type OrchestrationV2TurnItem,
12} from "@t3tools/contracts";
13import { it } from "@effect/vitest";
14import * as DateTime from "effect/DateTime";
15import * as Effect from "effect/Effect";
16import * as Stream from "effect/Stream";
17import * as TestClock from "effect/testing/TestClock";
18import { describe, expect } from "vite-plus/test";
19import {
20 coalesceLiveUpdatedEvents,
21 coalesceThreadLiveStream,
22 makeThreadLiveEventCoalescer,
23} from "./ThreadLiveEventCoalescer.ts";
24
25import { applyOrchestrationV2ProjectionEvent } from "../../../../packages/client-runtime/src/state/orchestrationV2Projection.ts";
26
27type ThreadEvent = Extract<OrchestrationV2ThreadStreamItem, { readonly kind: "event" }>;
28const threadId = ThreadId.make("thread-coalescer-test");
29const runId = RunId.make("run-coalescer-test");
30const now = DateTime.makeUnsafe("2026-01-01T00:00:01.000Z");
31const encodeEvent = JSON.stringify;
32
33function makeToolActivity(
34 sequence: number,
35 options: {
36 readonly kind?: "tool.updated" | "tool.completed";
37 readonly toolCallId?: string;
38 readonly turnId?: RunId;
39 readonly output?: string;
40 } = {},
41): ThreadEvent {
42 const itemRunId = options.turnId ?? runId;
43 const status = options.kind === "tool.completed" ? "completed" : "running";
44 return {
45 kind: "event",
46 sequence,
47 event: {
48 id: EventId.make(`event-${sequence}`),
49 type: "turn-item.updated",
50 threadId,
51 runId: itemRunId,
52 occurredAt: now,
53 payload: {
54 id: TurnItemId.make(options.toolCallId ?? "call-edit"),
55 type: "command_execution",
56 threadId,
57 runId: itemRunId,
58 nodeId: null,
59 providerThreadId: null,
60 providerTurnId: null,
61 nativeItemRef: null,
62 parentItemId: null,
63 ordinal: 0,
64 status,
65 title: "Editing app.ts",
66 startedAt: now,
67 completedAt: status === "completed" ? now : null,
68 updatedAt: now,
69 input: "echo app.ts",
70 output: options.output ?? `output-${sequence}`,
71 },
72 },
73 };
74}
75
76function makeMessage(sequence: number, text = "Still working"): ThreadEvent {
77 return {
78 kind: "event",
79 sequence,
80 event: {
81 id: EventId.make(`event-${sequence}`),
82 type: "message.updated",
83 threadId,
84 runId,
85 occurredAt: now,
86 payload: {
87 id: MessageId.make(`message-${sequence}`),
88 threadId,
89 runId,
90 nodeId: null,
91 createdBy: "agent",
92 creationSource: "provider",
93 role: "assistant",
94 text,
95 attachments: [],
96 streaming: false,
97 createdAt: now,
98 updatedAt: now,
99 },
100 },
101 };
102}
103
104function makeStreamingMessage(
105 sequence: number,
106 options: {
107 readonly id?: string;
108 readonly text?: string;
109 readonly streaming?: boolean;
110 } = {},
111): ThreadEvent {
112 const message = makeMessage(sequence, options.text ?? `text-${sequence}`);
113 if (message.event.type !== "message.updated") throw new Error("Expected message");
114 return {
115 ...message,
116 event: {
117 ...message.event,
118 payload: {
119 ...message.event.payload,
120 id: MessageId.make(options.id ?? "message"),
121 streaming: options.streaming ?? true,
122 },
123 },
124 };
125}
126
127function makeTextItem(
128 sequence: number,
129 options: {
130 readonly id?: string;
131 readonly text?: string;
132 readonly type?: "assistant_message" | "reasoning";
133 readonly status?: OrchestrationV2TurnItem["status"];
134 readonly streaming?: boolean;
135 } = {},
136): ThreadEvent {
137 const tool = makeToolActivity(sequence);
138 if (tool.event.type !== "turn-item.updated" || tool.event.payload.type !== "command_execution")
139 throw new Error("Expected command item");
140 const { input: _input, output: _output, ...base } = tool.event.payload;
141 return {
142 ...tool,
143 event: {
144 ...tool.event,
145 payload: {
146 ...base,
147 id: TurnItemId.make(options.id ?? "item"),
148 type: options.type ?? "assistant_message",
149 messageId: MessageId.make("message"),
150 text: options.text ?? `text-${sequence}`,
151 status: options.status ?? "running",
152 streaming: options.streaming ?? true,
153 },
154 },
155 };
156}
157
158describe("ThreadLiveEventCoalescer", () => {
159 it.effect.each(["items", "bytes"] as const)(
160 "replaces superseded tool output before charging the %s budget",
161 (limit) =>
162 Effect.gen(function* () {
163 const updates = Array.from({ length: 1_100 }, (_, index) => makeToolActivity(index + 1));
164 const latest = updates.at(-1)!;
165 const coalescer = yield* makeThreadLiveEventCoalescer(
166 limit === "items"
167 ? { maxItems: 1 }
168 : { maxSerializedBytes: Buffer.byteLength(encodeEvent(latest)) },
169 );
170 yield* coalescer.offerAll(updates);
171 expect(yield* coalescer.usage).toEqual({
172 retainedItems: 1,
173 retainedSerializedBytes: Buffer.byteLength(encodeEvent(latest)),
174 });
175 yield* coalescer.end;
176 expect(yield* coalescer.stream.pipe(Stream.runCollect)).toEqual([latest]);
177 expect(yield* coalescer.usage).toEqual({ retainedItems: 0, retainedSerializedBytes: 0 });
178 }),
179 );
180
181 it.effect("handles a tool output burst larger than the default 8 MiB buffer", () =>
182 Effect.gen(function* () {
183 const coalescer = yield* makeThreadLiveEventCoalescer();
184 const updates = Array.from({ length: 150 }, (_, index) =>
185 makeToolActivity(index + 1, { output: "x".repeat(64 * 1024) }),
186 );
187 yield* coalescer.offerAll(updates);
188 yield* coalescer.end;
189 expect(yield* coalescer.stream.pipe(Stream.runCollect)).toEqual([updates.at(-1)]);
190 }),
191 );
192
193 it.effect("still fails when a single replacement exceeds the byte budget", () =>
194 Effect.gen(function* () {
195 const first = makeToolActivity(1);
196 const coalescer = yield* makeThreadLiveEventCoalescer({
197 maxSerializedBytes: Buffer.byteLength(encodeEvent(first)),
198 });
199 yield* coalescer.offer(first);
200 const overflow = yield* coalescer
201 .offer(makeToolActivity(2, { output: "x".repeat(2_000) }))
202 .pipe(Effect.result);
203 expect(overflow._tag).toBe("Failure");
204 yield* coalescer.closed;
205 expect(yield* coalescer.usage).toEqual({ retainedItems: 0, retainedSerializedBytes: 0 });
206 }),
207 );
208
209 it("preserves tool-call identity switches as insertion-order boundaries", () => {
210 const events = [
211 makeToolActivity(1, { toolCallId: "call-a" }),
212 makeToolActivity(2, { toolCallId: "call-b" }),
213 makeToolActivity(3, { toolCallId: "call-a" }),
214 ];
215
216 expect(coalesceLiveUpdatedEvents(events).map((event) => event.sequence)).toEqual([1, 2, 3]);
217 });
218
219 it("does not coalesce stable tool calls across turns", () => {
220 const events = [
221 makeToolActivity(1, { turnId: RunId.make("turn-old") }),
222 makeToolActivity(2, { turnId: RunId.make("turn-new") }),
223 ];
224
225 expect(coalesceLiveUpdatedEvents(events).map((event) => event.sequence)).toEqual([1, 2]);
226 });
227
228 it("flushes a stable update run before a completion boundary", () => {
229 const events = [
230 makeToolActivity(1),
231 makeToolActivity(2),
232 makeToolActivity(3, { kind: "tool.completed" }),
233 makeToolActivity(4),
234 ];
235
236 expect(coalesceLiveUpdatedEvents(events).map((event) => event.sequence)).toEqual([2, 3, 4]);
237 });
238
239 it("keeps the latest full text in sequence order and preserves state after every survivor cursor", () => {
240 const events = [
241 makeStreamingMessage(1, { id: "shared", text: "H" }),
242 makeTextItem(2, { id: "shared", text: "H" }),
243 makeTextItem(3, { id: "reasoning", type: "reasoning", text: "Think" }),
244 makeToolActivity(4),
245 makeTextItem(5, { id: "unique", text: "Keep me" }),
246 makeStreamingMessage(6, { id: "shared", text: "Hello" }),
247 makeTextItem(7, { id: "shared", text: "Hello" }),
248 makeTextItem(8, { id: "reasoning", type: "reasoning", text: "Thinking done" }),
249 makeToolActivity(9),
250 makeStreamingMessage(10, { id: "shared", streaming: false, text: "Hello!" }),
251 makeTextItem(11, { id: "shared", streaming: false, status: "completed", text: "Hello!" }),
252 ];
253 const survivors = coalesceLiveUpdatedEvents(events);
254 expect(survivors).toEqual(events);
255
256 const latestStates = (input: ReadonlyArray<ThreadEvent>) =>
257 Object.fromEntries(
258 input.map(({ event }) => [
259 `${event.type}:${"id" in event.payload ? event.payload.id : ""}`,
260 event.payload,
261 ]),
262 );
263 for (let index = 0; index < survivors.length; index += 1) {
264 const received = survivors.slice(0, index + 1);
265 const cursor = received.at(-1)!.sequence;
266 const resumed = coalesceLiveUpdatedEvents(events.filter((event) => event.sequence > cursor));
267 expect(latestStates([...received, ...resumed])).toEqual(latestStates(events));
268 }
269 });
270
271 it("preserves assistant-message identity changes as insertion-order boundaries", () => {
272 const events = [
273 makeStreamingMessage(1, { id: "message-a" }),
274 makeTextItem(2, { id: "item-a" }),
275 makeStreamingMessage(3, { id: "message-b" }),
276 makeTextItem(4, { id: "item-b" }),
277 makeStreamingMessage(5, { id: "message-a" }),
278 makeTextItem(6, { id: "item-a" }),
279 ];
280 expect(coalesceLiveUpdatedEvents(events)).toEqual(events);
281 });
282
283 it.effect(
284 "preserves turn-item insertion and partial timeline state after every survivor cursor",
285 () =>
286 Effect.gen(function* () {
287 const item = (
288 sequence: number,
289 id: string,
290 text: string,
291 completed = false,
292 ): ThreadEvent => {
293 const input = makeTextItem(sequence, {
294 id,
295 text,
296 type: "reasoning",
297 status: completed ? "completed" : "running",
298 streaming: !completed,
299 });
300 if (input.event.type !== "turn-item.updated") throw new Error("Expected turn item");
301 return {
302 ...input,
303 event: {
304 ...input.event,
305 runId: undefined,
306 payload: {
307 ...input.event.payload,
308 runId: null,
309 ordinal: id === "a" ? 1 : 2,
310 completedAt: completed ? now : null,
311 },
312 },
313 };
314 };
315 const events = [
316 item(1, "a", "A"),
317 item(2, "b", "B"),
318 item(3, "a", "AA"),
319 item(4, "a", "Final A", true),
320 item(5, "b", "Final B", true),
321 ];
322 const emptyProjection = {
323 thread: {
324 id: threadId,
325 projectId: ProjectId.make("project-reducer"),
326 title: "Reducer",
327 providerInstanceId: ProviderInstanceId.make("codex"),
328 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" },
329 runtimeMode: "full-access",
330 interactionMode: "default",
331 branch: null,
332 worktreePath: null,
333 activeProviderThreadId: null,
334 lineage: { rootThreadId: threadId, parentThreadId: null, relationshipToParent: null },
335 forkedFrom: null,
336 createdBy: "user",
337 creationSource: "web",
338 createdAt: now,
339 updatedAt: now,
340 archivedAt: null,
341 settledOverride: null,
342 settledAt: null,
343 lastVisitedAt: null,
344 deletedAt: null,
345 },
346 runs: [],
347 attempts: [],
348 nodes: [],
349 subagents: [],
350 providerSessions: [],
351 providerThreads: [],
352 providerTurns: [],
353 runtimeRequests: [],
354 messages: [],
355 plans: [],
356 turnItems: [],
357 checkpointScopes: [],
358 checkpoints: [],
359 contextHandoffs: [],
360 contextTransfers: [],
361 visibleTurnItems: [],
362 updatedAt: now,
363 } satisfies OrchestrationV2ThreadProjection;
364 const apply = (inputs: ReadonlyArray<ThreadEvent>, partialTimeline: boolean) => {
365 let projection: OrchestrationV2ThreadProjection = emptyProjection;
366 let latestLocalTurnOrdinal = 0;
367 for (const { event } of inputs) {
368 const next = applyOrchestrationV2ProjectionEvent(projection, event, {
369 partialTimeline,
370 latestLocalTurnOrdinal,
371 });
372 if (next === null || next === projection) continue;
373 projection = next;
374 if (partialTimeline && event.type === "turn-item.updated") {
375 latestLocalTurnOrdinal = Math.max(latestLocalTurnOrdinal, event.payload.ordinal);
376 }
377 }
378 return projection;
379 };
380 const survivors = yield* Stream.fromIterable(events).pipe(
381 coalesceThreadLiveStream,
382 Stream.filter((input): input is ThreadEvent => input.kind === "event"),
383 Stream.runCollect,
384 );
385 for (const partialTimeline of [true, false]) {
386 const expected = apply(events, partialTimeline);
387 expect(
388 expected.turnItems.map((item) => [item.id, "text" in item ? item.text : null]),
389 ).toEqual([
390 ["a", "Final A"],
391 ["b", "Final B"],
392 ]);
393 expect(apply(survivors, partialTimeline)).toEqual(expected);
394 for (const [index, survivor] of survivors.entries()) {
395 const received = survivors.slice(0, index + 1);
396 const replay = yield* Stream.fromIterable(
397 events.filter((event) => event.sequence > survivor.sequence),
398 ).pipe(
399 coalesceThreadLiveStream,
400 Stream.filter((input): input is ThreadEvent => input.kind === "event"),
401 Stream.runCollect,
402 );
403 expect(apply([...received, ...replay], partialTimeline)).toEqual(expected);
404 }
405 }
406 }),
407 );
408
409 it.effect("flushes text at synchronization and completion boundaries", () =>
410 Effect.gen(function* () {
411 const coalescer = yield* makeThreadLiveEventCoalescer();
412 const events = [
413 makeStreamingMessage(1),
414 makeTextItem(2, { type: "reasoning" }),
415 makeStreamingMessage(3),
416 makeTextItem(4, { type: "reasoning" }),
417 { kind: "synchronized" } as const,
418 makeStreamingMessage(5),
419 makeTextItem(6, { type: "reasoning" }),
420 makeStreamingMessage(7, { streaming: false, text: "Final answer" }),
421 makeTextItem(8, {
422 type: "reasoning",
423 streaming: false,
424 status: "completed",
425 text: "Final answer",
426 }),
427 makeStreamingMessage(9),
428 makeTextItem(10, { type: "reasoning" }),
429 ];
430 yield* coalescer.offerAll(events);
431 yield* coalescer.end;
432 expect(yield* Stream.runCollect(coalescer.stream)).toEqual(events.slice(2));
433 expect(yield* coalescer.usage).toEqual({ retainedItems: 0, retainedSerializedBytes: 0 });
434 }),
435 );
436
437 it.effect(
438 "flushes growing text on the fixed window and charges only unacknowledged survivors",
439 () =>
440 Effect.gen(function* () {
441 const coalescer = yield* makeThreadLiveEventCoalescer();
442 const first = [makeStreamingMessage(1, { text: "é" }), makeTextItem(2, { text: "é" })];
443 const latest = [makeStreamingMessage(3, { text: "éé" }), makeTextItem(4, { text: "éé" })];
444 const serializedBytes = (events: ReadonlyArray<ThreadEvent>) =>
445 events.reduce((sum, event) => sum + Buffer.byteLength(encodeEvent(event)), 0);
446 yield* coalescer.offerAll(first);
447 yield* TestClock.adjust("40 millis");
448 yield* coalescer.offerAll(latest);
449 expect(yield* coalescer.usage).toEqual({
450 retainedItems: 2,
451 retainedSerializedBytes: serializedBytes(latest),
452 });
453 yield* TestClock.adjust("10 millis");
454 const expectedUsage = {
455 retainedItems: 2,
456 retainedSerializedBytes: serializedBytes(latest),
457 };
458 expect(yield* coalescer.usage).toEqual(expectedUsage);
459 const pull = yield* Stream.toPull(coalescer.stream);
460 expect(yield* pull).toEqual(latest);
461 expect(yield* coalescer.usage).toEqual(expectedUsage);
462
463 const next = [makeStreamingMessage(5), makeTextItem(6)];
464 yield* coalescer.offerAll(next);
465 yield* TestClock.adjust("50 millis");
466 expect(yield* pull).toEqual(next);
467 yield* coalescer.end;
468 yield* pull.pipe(Effect.result);
469 expect(yield* coalescer.usage).toEqual({ retainedItems: 0, retainedSerializedBytes: 0 });
470 }),
471 );
472
473 it.effect("keeps the flush timer alive when an offer's shorter scope closes", () =>
474 Effect.gen(function* () {
475 const coalescer = yield* makeThreadLiveEventCoalescer();
476 yield* Effect.scoped(coalescer.offer(makeToolActivity(1)));
477 yield* TestClock.adjust("50 millis");
478 const items = yield* coalescer.stream.pipe(Stream.take(1), Stream.runCollect);
479 expect(items.map((item) => (item.kind === "event" ? item.sequence : item.kind))).toEqual([1]);
480 }),
481 );
482
483 it.effect("keeps an unacknowledged batch charged and clears later events on overflow", () =>
484 Effect.gen(function* () {
485 const coalescer = yield* makeThreadLiveEventCoalescer({ maxItems: 3 });
486 const first = makeMessage(1, "é".repeat(1_024));
487 yield* coalescer.offer(first);
488
489 yield* Effect.scoped(
490 Effect.gen(function* () {
491 const pull = yield* Stream.toPull(coalescer.stream);
492 const batch = yield* pull;
493 expect(batch.map((item) => (item.kind === "event" ? item.sequence : null))).toEqual([1]);
494 yield* coalescer.offer(makeMessage(2));
495 yield* coalescer.offer(makeToolActivity(3));
496 expect((yield* coalescer.usage).retainedItems).toBe(3);
497
498 const overflow = yield* coalescer
499 .offer(makeToolActivity(4, { kind: "tool.completed" }))
500 .pipe(Effect.result);
501 expect(overflow._tag).toBe("Failure");
502 yield* coalescer.closed;
503 expect(yield* coalescer.usage).toEqual({
504 retainedItems: 1,
505 retainedSerializedBytes: Buffer.byteLength(encodeEvent(first)),
506 });
507 }),
508 );
509 expect(yield* coalescer.usage).toEqual({ retainedItems: 0, retainedSerializedBytes: 0 });
510 }),
511 );
512});
513
514it.effect("flushes the last update before a finite source ends", () =>
515 Effect.gen(function* () {
516 const result = yield* Stream.make(makeToolActivity(1), makeToolActivity(2)).pipe(
517 coalesceThreadLiveStream,
518 Stream.runCollect,
519 );
520 expect(result.map((item) => (item.kind === "event" ? item.sequence : null))).toEqual([2]);
521 }),
522);
523
524it.effect("propagates an upstream failure instead of leaving the subscriber waiting", () =>
525 Effect.gen(function* () {
526 const result = yield* Stream.fail("source stopped").pipe(
527 coalesceThreadLiveStream,
528 Stream.runCollect,
529 Effect.result,
530 );
531 expect(result._tag).toBe("Failure");
532 if (result._tag === "Failure") expect(result.failure).toBe("source stopped");
533 }),
534);