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.scoped(
163 Effect.gen(function* () {
164 const updates = Array.from({ length: 1_100 }, (_, index) => makeToolActivity(index + 1));
165 const latest = updates.at(-1)!;
166 const coalescer = yield* makeThreadLiveEventCoalescer(
167 limit === "items"
168 ? { maxItems: 1 }
169 : { maxSerializedBytes: Buffer.byteLength(encodeEvent(latest)) },
170 );
171 yield* coalescer.offerAll(updates);
172 expect(yield* coalescer.usage).toEqual({
173 retainedItems: 1,
174 retainedSerializedBytes: Buffer.byteLength(encodeEvent(latest)),
175 });
176 yield* coalescer.end;
177 expect(yield* coalescer.stream.pipe(Stream.runCollect)).toEqual([latest]);
178 expect(yield* coalescer.usage).toEqual({ retainedItems: 0, retainedSerializedBytes: 0 });
179 }),
180 ),
181 );
182
183 it.effect("handles a tool output burst larger than the default 8 MiB buffer", () =>
184 Effect.scoped(
185 Effect.gen(function* () {
186 const coalescer = yield* makeThreadLiveEventCoalescer();
187 const updates = Array.from({ length: 150 }, (_, index) =>
188 makeToolActivity(index + 1, { output: "x".repeat(64 * 1024) }),
189 );
190 yield* coalescer.offerAll(updates);
191 yield* coalescer.end;
192 expect(yield* coalescer.stream.pipe(Stream.runCollect)).toEqual([updates.at(-1)]);
193 }),
194 ),
195 );
196
197 it.effect("still fails when a single replacement exceeds the byte budget", () =>
198 Effect.scoped(
199 Effect.gen(function* () {
200 const first = makeToolActivity(1);
201 const coalescer = yield* makeThreadLiveEventCoalescer({
202 maxSerializedBytes: Buffer.byteLength(encodeEvent(first)),
203 });
204 yield* coalescer.offer(first);
205 const overflow = yield* coalescer
206 .offer(makeToolActivity(2, { output: "x".repeat(2_000) }))
207 .pipe(Effect.result);
208 expect(overflow._tag).toBe("Failure");
209 yield* coalescer.closed;
210 expect(yield* coalescer.usage).toEqual({ retainedItems: 0, retainedSerializedBytes: 0 });
211 }),
212 ),
213 );
214
215 it("preserves tool-call identity switches as insertion-order boundaries", () => {
216 const events = [
217 makeToolActivity(1, { toolCallId: "call-a" }),
218 makeToolActivity(2, { toolCallId: "call-b" }),
219 makeToolActivity(3, { toolCallId: "call-a" }),
220 ];
221
222 expect(coalesceLiveUpdatedEvents(events).map((event) => event.sequence)).toEqual([1, 2, 3]);
223 });
224
225 it("does not coalesce stable tool calls across turns", () => {
226 const events = [
227 makeToolActivity(1, { turnId: RunId.make("turn-old") }),
228 makeToolActivity(2, { turnId: RunId.make("turn-new") }),
229 ];
230
231 expect(coalesceLiveUpdatedEvents(events).map((event) => event.sequence)).toEqual([1, 2]);
232 });
233
234 it("flushes a stable update run before a completion boundary", () => {
235 const events = [
236 makeToolActivity(1),
237 makeToolActivity(2),
238 makeToolActivity(3, { kind: "tool.completed" }),
239 makeToolActivity(4),
240 ];
241
242 expect(coalesceLiveUpdatedEvents(events).map((event) => event.sequence)).toEqual([2, 3, 4]);
243 });
244
245 it("keeps the latest full text in sequence order and preserves state after every survivor cursor", () => {
246 const events = [
247 makeStreamingMessage(1, { id: "shared", text: "H" }),
248 makeTextItem(2, { id: "shared", text: "H" }),
249 makeTextItem(3, { id: "reasoning", type: "reasoning", text: "Think" }),
250 makeToolActivity(4),
251 makeTextItem(5, { id: "unique", text: "Keep me" }),
252 makeStreamingMessage(6, { id: "shared", text: "Hello" }),
253 makeTextItem(7, { id: "shared", text: "Hello" }),
254 makeTextItem(8, { id: "reasoning", type: "reasoning", text: "Thinking done" }),
255 makeToolActivity(9),
256 makeStreamingMessage(10, { id: "shared", streaming: false, text: "Hello!" }),
257 makeTextItem(11, { id: "shared", streaming: false, status: "completed", text: "Hello!" }),
258 ];
259 const survivors = coalesceLiveUpdatedEvents(events);
260 expect(survivors).toEqual(events);
261
262 const latestStates = (input: ReadonlyArray<ThreadEvent>) =>
263 Object.fromEntries(
264 input.map(({ event }) => [
265 `${event.type}:${"id" in event.payload ? event.payload.id : ""}`,
266 event.payload,
267 ]),
268 );
269 for (let index = 0; index < survivors.length; index += 1) {
270 const received = survivors.slice(0, index + 1);
271 const cursor = received.at(-1)!.sequence;
272 const resumed = coalesceLiveUpdatedEvents(events.filter((event) => event.sequence > cursor));
273 expect(latestStates([...received, ...resumed])).toEqual(latestStates(events));
274 }
275 });
276
277 it("preserves assistant-message identity changes as insertion-order boundaries", () => {
278 const events = [
279 makeStreamingMessage(1, { id: "message-a" }),
280 makeTextItem(2, { id: "item-a" }),
281 makeStreamingMessage(3, { id: "message-b" }),
282 makeTextItem(4, { id: "item-b" }),
283 makeStreamingMessage(5, { id: "message-a" }),
284 makeTextItem(6, { id: "item-a" }),
285 ];
286 expect(coalesceLiveUpdatedEvents(events)).toEqual(events);
287 });
288
289 it.effect(
290 "preserves turn-item insertion and partial timeline state after every survivor cursor",
291 () =>
292 Effect.gen(function* () {
293 const item = (
294 sequence: number,
295 id: string,
296 text: string,
297 completed = false,
298 ): ThreadEvent => {
299 const input = makeTextItem(sequence, {
300 id,
301 text,
302 type: "reasoning",
303 status: completed ? "completed" : "running",
304 streaming: !completed,
305 });
306 if (input.event.type !== "turn-item.updated") throw new Error("Expected turn item");
307 return {
308 ...input,
309 event: {
310 ...input.event,
311 runId: undefined,
312 payload: {
313 ...input.event.payload,
314 runId: null,
315 ordinal: id === "a" ? 1 : 2,
316 completedAt: completed ? now : null,
317 },
318 },
319 };
320 };
321 const events = [
322 item(1, "a", "A"),
323 item(2, "b", "B"),
324 item(3, "a", "AA"),
325 item(4, "a", "Final A", true),
326 item(5, "b", "Final B", true),
327 ];
328 const emptyProjection = {
329 thread: {
330 id: threadId,
331 projectId: ProjectId.make("project-reducer"),
332 title: "Reducer",
333 providerInstanceId: ProviderInstanceId.make("codex"),
334 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" },
335 runtimeMode: "full-access",
336 interactionMode: "default",
337 branch: null,
338 worktreePath: null,
339 activeProviderThreadId: null,
340 lineage: { rootThreadId: threadId, parentThreadId: null, relationshipToParent: null },
341 forkedFrom: null,
342 createdBy: "user",
343 creationSource: "web",
344 createdAt: now,
345 updatedAt: now,
346 archivedAt: null,
347 settledOverride: null,
348 settledAt: null,
349 lastVisitedAt: null,
350 deletedAt: null,
351 },
352 runs: [],
353 attempts: [],
354 nodes: [],
355 subagents: [],
356 providerSessions: [],
357 providerThreads: [],
358 providerTurns: [],
359 runtimeRequests: [],
360 messages: [],
361 plans: [],
362 turnItems: [],
363 checkpointScopes: [],
364 checkpoints: [],
365 contextHandoffs: [],
366 contextTransfers: [],
367 visibleTurnItems: [],
368 updatedAt: now,
369 } satisfies OrchestrationV2ThreadProjection;
370 const apply = (inputs: ReadonlyArray<ThreadEvent>, partialTimeline: boolean) => {
371 let projection: OrchestrationV2ThreadProjection = emptyProjection;
372 let latestLocalTurnOrdinal = 0;
373 for (const { event } of inputs) {
374 const next = applyOrchestrationV2ProjectionEvent(projection, event, {
375 partialTimeline,
376 latestLocalTurnOrdinal,
377 });
378 if (next === null || next === projection) continue;
379 projection = next;
380 if (partialTimeline && event.type === "turn-item.updated") {
381 latestLocalTurnOrdinal = Math.max(latestLocalTurnOrdinal, event.payload.ordinal);
382 }
383 }
384 return projection;
385 };
386 const survivors = yield* Stream.fromIterable(events).pipe(
387 coalesceThreadLiveStream,
388 Stream.filter((input): input is ThreadEvent => input.kind === "event"),
389 Stream.runCollect,
390 );
391 for (const partialTimeline of [true, false]) {
392 const expected = apply(events, partialTimeline);
393 expect(
394 expected.turnItems.map((item) => [item.id, "text" in item ? item.text : null]),
395 ).toEqual([
396 ["a", "Final A"],
397 ["b", "Final B"],
398 ]);
399 expect(apply(survivors, partialTimeline)).toEqual(expected);
400 for (const [index, survivor] of survivors.entries()) {
401 const received = survivors.slice(0, index + 1);
402 const replay = yield* Stream.fromIterable(
403 events.filter((event) => event.sequence > survivor.sequence),
404 ).pipe(
405 coalesceThreadLiveStream,
406 Stream.filter((input): input is ThreadEvent => input.kind === "event"),
407 Stream.runCollect,
408 );
409 expect(apply([...received, ...replay], partialTimeline)).toEqual(expected);
410 }
411 }
412 }),
413 );
414
415 it.effect("flushes text at synchronization and completion boundaries", () =>
416 Effect.scoped(
417 Effect.gen(function* () {
418 const coalescer = yield* makeThreadLiveEventCoalescer();
419 const events = [
420 makeStreamingMessage(1),
421 makeTextItem(2, { type: "reasoning" }),
422 makeStreamingMessage(3),
423 makeTextItem(4, { type: "reasoning" }),
424 { kind: "synchronized" } as const,
425 makeStreamingMessage(5),
426 makeTextItem(6, { type: "reasoning" }),
427 makeStreamingMessage(7, { streaming: false, text: "Final answer" }),
428 makeTextItem(8, {
429 type: "reasoning",
430 streaming: false,
431 status: "completed",
432 text: "Final answer",
433 }),
434 makeStreamingMessage(9),
435 makeTextItem(10, { type: "reasoning" }),
436 ];
437 yield* coalescer.offerAll(events);
438 yield* coalescer.end;
439 expect(yield* Stream.runCollect(coalescer.stream)).toEqual(events.slice(2));
440 expect(yield* coalescer.usage).toEqual({ retainedItems: 0, retainedSerializedBytes: 0 });
441 }),
442 ),
443 );
444
445 it.effect(
446 "flushes growing text on the fixed window and charges only unacknowledged survivors",
447 () =>
448 Effect.scoped(
449 Effect.gen(function* () {
450 const coalescer = yield* makeThreadLiveEventCoalescer();
451 const first = [makeStreamingMessage(1, { text: "é" }), makeTextItem(2, { text: "é" })];
452 const latest = [makeStreamingMessage(3, { text: "éé" }), makeTextItem(4, { text: "éé" })];
453 const serializedBytes = (events: ReadonlyArray<ThreadEvent>) =>
454 events.reduce((sum, event) => sum + Buffer.byteLength(encodeEvent(event)), 0);
455 yield* coalescer.offerAll(first);
456 yield* TestClock.adjust("40 millis");
457 yield* coalescer.offerAll(latest);
458 expect(yield* coalescer.usage).toEqual({
459 retainedItems: 2,
460 retainedSerializedBytes: serializedBytes(latest),
461 });
462 yield* TestClock.adjust("10 millis");
463 const expectedUsage = {
464 retainedItems: 2,
465 retainedSerializedBytes: serializedBytes(latest),
466 };
467 expect(yield* coalescer.usage).toEqual(expectedUsage);
468 const pull = yield* Stream.toPull(coalescer.stream);
469 expect(yield* pull).toEqual(latest);
470 expect(yield* coalescer.usage).toEqual(expectedUsage);
471
472 const next = [makeStreamingMessage(5), makeTextItem(6)];
473 yield* coalescer.offerAll(next);
474 yield* TestClock.adjust("50 millis");
475 expect(yield* pull).toEqual(next);
476 yield* coalescer.end;
477 yield* pull.pipe(Effect.result);
478 expect(yield* coalescer.usage).toEqual({ retainedItems: 0, retainedSerializedBytes: 0 });
479 }),
480 ),
481 );
482
483 it.effect("keeps the flush timer alive when an offer's shorter scope closes", () =>
484 Effect.scoped(
485 Effect.gen(function* () {
486 const coalescer = yield* makeThreadLiveEventCoalescer();
487 yield* Effect.scoped(coalescer.offer(makeToolActivity(1)));
488 yield* TestClock.adjust("50 millis");
489 const items = yield* coalescer.stream.pipe(Stream.take(1), Stream.runCollect);
490 expect(items.map((item) => (item.kind === "event" ? item.sequence : item.kind))).toEqual([
491 1,
492 ]);
493 }),
494 ),
495 );
496
497 it.effect("keeps an unacknowledged batch charged and clears later events on overflow", () =>
498 Effect.scoped(
499 Effect.gen(function* () {
500 const coalescer = yield* makeThreadLiveEventCoalescer({ maxItems: 3 });
501 const first = makeMessage(1, "é".repeat(1_024));
502 yield* coalescer.offer(first);
503
504 yield* Effect.scoped(
505 Effect.gen(function* () {
506 const pull = yield* Stream.toPull(coalescer.stream);
507 const batch = yield* pull;
508 expect(batch.map((item) => (item.kind === "event" ? item.sequence : null))).toEqual([
509 1,
510 ]);
511 yield* coalescer.offer(makeMessage(2));
512 yield* coalescer.offer(makeToolActivity(3));
513 expect((yield* coalescer.usage).retainedItems).toBe(3);
514
515 const overflow = yield* coalescer
516 .offer(makeToolActivity(4, { kind: "tool.completed" }))
517 .pipe(Effect.result);
518 expect(overflow._tag).toBe("Failure");
519 yield* coalescer.closed;
520 expect(yield* coalescer.usage).toEqual({
521 retainedItems: 1,
522 retainedSerializedBytes: Buffer.byteLength(encodeEvent(first)),
523 });
524 }),
525 );
526 expect(yield* coalescer.usage).toEqual({ retainedItems: 0, retainedSerializedBytes: 0 });
527 }),
528 ),
529 );
530});
531
532it.effect("flushes the last update before a finite source ends", () =>
533 Effect.gen(function* () {
534 const result = yield* Stream.make(makeToolActivity(1), makeToolActivity(2)).pipe(
535 coalesceThreadLiveStream,
536 Stream.runCollect,
537 );
538 expect(result.map((item) => (item.kind === "event" ? item.sequence : null))).toEqual([2]);
539 }),
540);
541
542it.effect("propagates an upstream failure instead of leaving the subscriber waiting", () =>
543 Effect.gen(function* () {
544 const result = yield* Stream.fail("source stopped").pipe(
545 coalesceThreadLiveStream,
546 Stream.runCollect,
547 Effect.result,
548 );
549 expect(result._tag).toBe("Failure");
550 if (result._tag === "Failure") expect(result.failure).toBe("source stopped");
551 }),
552);