1import { describe, expect, it } from "vite-plus/test";
2import { MessageId, NodeId, ProviderDriverKind, ThreadId, TurnItemId } from "@t3tools/contracts";
3import * as DateTime from "effect/DateTime";
4import { makeAssistantStreamingFilter, splitBufferedAssistantText } from "./assistantStreaming.ts";
5import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
6
7const message = (text: string, streaming = true): ProviderAdapter.ProviderAdapterV2Event => ({
8 type: "message.updated",
9 driver: ProviderDriverKind.make("codex"),
10 message: {
11 id: MessageId.make("message"),
12 threadId: ThreadId.make("thread"),
13 runId: null,
14 nodeId: null,
15 createdBy: "agent",
16 creationSource: "provider",
17 updatedAt: DateTime.makeUnsafe("2026-09-14T00:00:00Z"),
18 role: "assistant",
19 text,
20 attachments: [],
21 createdAt: DateTime.makeUnsafe("2026-09-14T00:00:00Z"),
22 streaming,
23 },
24});
25
26const turnItem = (
27 text: string,
28 streaming = true,
29 type: "assistant_message" | "reasoning" = "assistant_message",
30): ProviderAdapter.ProviderAdapterV2Event => ({
31 type: "turn_item.updated",
32 driver: ProviderDriverKind.make("codex"),
33 turnItem: {
34 id: TurnItemId.make("item"),
35 threadId: ThreadId.make("thread"),
36 runId: null,
37 nodeId: null,
38 providerThreadId: null,
39 providerTurnId: null,
40 nativeItemRef: null,
41 parentItemId: null,
42 ordinal: 0,
43 status: streaming ? "running" : "completed",
44 title: null,
45 startedAt: null,
46 completedAt: null,
47 updatedAt: DateTime.makeUnsafe("2026-09-14T00:00:00Z"),
48 ...(type === "assistant_message" ? { type, messageId: MessageId.make("message") } : { type }),
49 text,
50 streaming,
51 },
52});
53
54describe("V2 assistant streaming", () => {
55 it("delivers completed paragraphs, coalesces rapid updates, and flushes final text", () => {
56 const filter = makeAssistantStreamingFilter("paragraph");
57 expect(filter(message("First"), 0)).toBeNull();
58 expect(filter(message("First\n\nSec"), 10)).toMatchObject({ message: { text: "First\n\n" } });
59 expect(filter(message("First\n\nSecond\n\nThi"), 100)).toBeNull();
60 expect(filter(message("First\n\nSecond\n\nThird"), 410)).toMatchObject({
61 message: { text: "First\n\nSecond\n\n" },
62 });
63 const final = message("First\n\nSecond\n\nThird", false);
64 expect(filter(final, 420)).toBe(final);
65 });
66 it("keeps code fences intact", () => {
67 expect(splitBufferedAssistantText("Intro\n\n```ts\nx()\n\n")).toEqual({
68 ready: "Intro\n\n",
69 rest: "```ts\nx()\n\n",
70 });
71 expect(splitBufferedAssistantText("```ts\nx()\n```\nrest")).toEqual({
72 ready: "```ts\nx()\n```\n",
73 rest: "rest",
74 });
75 });
76 it("holds streaming text until the full response completes", () => {
77 const running = message("partial");
78 const final = message("complete", false);
79 const buffered = makeAssistantStreamingFilter("turn");
80 expect(buffered(running, 0)).toBeNull();
81 expect(buffered(message("First\n\nSecond\n\n"), 500)).toBeNull();
82 expect(buffered(final, 501)).toBe(final);
83 });
84
85 it.each(["assistant_message", "reasoning"] as const)(
86 "buffers %s at paragraph boundaries and flushes the final item",
87 (type) => {
88 const item = (text: string, streaming = true) => turnItem(text, streaming, type);
89 const filter = makeAssistantStreamingFilter("paragraph");
90 expect(filter(item("First"), 0)).toBeNull();
91 expect(filter(item("First\n\nSec"), 10)).toMatchObject({
92 turnItem: { text: "First\n\n", streaming: true, type },
93 });
94 expect(filter(item("First\n\nSecond\n\nThi"), 100)).toBeNull();
95 expect(filter(item("First\n\nSecond\n\nThi"), 410)).toMatchObject({
96 turnItem: { text: "First\n\nSecond\n\n", streaming: true },
97 });
98 const final = item("First\n\nSecond\n\nThird", false);
99 expect(filter(final, 420)).toBe(final);
100
101 const buffered = makeAssistantStreamingFilter("turn");
102 expect(buffered(item("First\n\nSecond"), 0)).toBeNull();
103 expect(buffered(final, 1)).toBe(final);
104 },
105 );
106
107 it.each(["turn", "paragraph"] as const)(
108 "suppresses running assistant nodes in %s mode while delivering tool and completed nodes",
109 (mode) => {
110 const filter = makeAssistantStreamingFilter(mode);
111 const running: Extract<ProviderAdapter.ProviderAdapterV2Event, { type: "node.updated" }> = {
112 type: "node.updated",
113 driver: ProviderDriverKind.make("codex"),
114 node: {
115 id: NodeId.make("node"),
116 threadId: ThreadId.make("thread"),
117 runId: null,
118 parentNodeId: null,
119 rootNodeId: NodeId.make("root"),
120 kind: "assistant_message",
121 status: "running",
122 countsForRun: false,
123 providerThreadId: null,
124 providerTurnId: null,
125 nativeItemRef: null,
126 runtimeRequestId: null,
127 checkpointScopeId: null,
128 startedAt: null,
129 completedAt: null,
130 },
131 };
132 expect(filter(running, 0)).toBeNull();
133 const completed = { ...running, node: { ...running.node, status: "completed" as const } };
134 expect(filter(completed, 1)).toBe(completed);
135 const tool = { ...running, node: { ...running.node, kind: "tool_call" as const } };
136 expect(filter(tool, 2)).toBe(tool);
137 },
138 );
139});
140
141it("holds a streamed section heading until its content has a boundary", () => {
142 expect(splitBufferedAssistantText("Intro\n\n## Results\n\n")).toEqual({
143 ready: "Intro\n\n",
144 rest: "## Results\n\n",
145 });
146 expect(splitBufferedAssistantText("**Results**\n\nBody\n\nNext")).toEqual({
147 ready: "**Results**\n\nBody\n\n",
148 rest: "Next",
149 });
150});