1import {
2 EventId,
3 OrchestrationV2ThreadStreamItem,
4 ThreadId,
5 TurnItemId,
6 type OrchestrationV2ProjectedTurnItem,
7 type OrchestrationV2ThreadProjection,
8 type OrchestrationV2TurnItem,
9} from "@t3tools/contracts";
10import { describe, expect, it } from "vite-plus/test";
11import * as DateTime from "effect/DateTime";
12import * as Schema from "effect/Schema";
13import { boundedSnapshotProjection } from "@t3tools/shared/orchestrationV2BoundedSnapshot";
14
15import {
16 buildBoundedThreadStreamSnapshot,
17 decideThreadResume,
18 isThreadReplayRawPayloadSafe,
19 threadReplayEncodedBytes,
20 THREAD_RESUME_MAX_RAW_PAYLOAD_BYTES,
21 THREAD_RESUME_MAX_REPLAY_ENCODED_BYTES,
22 THREAD_RESUME_MAX_REPLAY_EVENTS,
23} from "./ThreadStream.ts";
24import { projectDomainEventForWire } from "./WireProjection.ts";
25
26const NOW = DateTime.makeUnsafe("2026-09-08T00:00:00.000Z");
27const THREAD_ID = ThreadId.make("thread-stream-test");
28const LARGE_PROJECTABLE_OUTPUT_BYTES = 10 * 1_048_576;
29const LARGE_PROJECTABLE_RAW_PAYLOAD_BYTES = 10_486_214;
30// The first projected replay measured 821 bytes; leave room for small envelope additions.
31const MAX_PROJECTED_DYNAMIC_REPLAY_BYTES = 2_048;
32
33function timelineProjection(itemCount: number): OrchestrationV2ThreadProjection {
34 const visibleTurnItems: OrchestrationV2ProjectedTurnItem[] = Array.from(
35 { length: itemCount },
36 (_, index) => {
37 const id = TurnItemId.make(`item-${index}`);
38 const item = {
39 id,
40 type: "command_execution",
41 threadId: THREAD_ID,
42 runId: null,
43 nodeId: null,
44 providerThreadId: null,
45 providerTurnId: null,
46 nativeItemRef: null,
47 parentItemId: null,
48 ordinal: index + 1,
49 status: "completed",
50 title: `Command ${index}`,
51 input: `command-${index}`,
52 output: `output-${index}`,
53 exitCode: 0,
54 startedAt: NOW,
55 completedAt: NOW,
56 updatedAt: NOW,
57 } satisfies OrchestrationV2TurnItem;
58 return {
59 position: index,
60 visibility: "local",
61 sourceThreadId: THREAD_ID,
62 sourceItemId: id,
63 item,
64 };
65 },
66 );
67 return {
68 thread: {
69 id: THREAD_ID,
70 projectId: "project-stream-test",
71 title: "Thread stream test",
72 providerInstanceId: "codex",
73 modelSelection: null,
74 runtimeMode: "full-access",
75 interactionMode: "default",
76 branch: null,
77 worktreePath: null,
78 activeProviderThreadId: null,
79 lineage: {
80 parentThreadId: null,
81 relationshipToParent: null,
82 rootThreadId: THREAD_ID,
83 },
84 forkedFrom: null,
85 createdBy: "user",
86 creationSource: "web",
87 createdAt: NOW,
88 updatedAt: NOW,
89 archivedAt: null,
90 settledOverride: null,
91 settledAt: null,
92 deletedAt: null,
93 },
94 runs: [],
95 attempts: [],
96 nodes: [],
97 subagents: [],
98 providerSessions: [],
99 providerThreads: [],
100 providerTurns: [],
101 runtimeRequests: [],
102 messages: [],
103 plans: [],
104 turnItems: visibleTurnItems.map((row) => row.item),
105 checkpointScopes: [],
106 checkpoints: [],
107 contextHandoffs: [],
108 contextTransfers: [],
109 visibleTurnItems,
110 updatedAt: NOW,
111 } as unknown as OrchestrationV2ThreadProjection;
112}
113
114describe("decideThreadResume", () => {
115 it("replays when the gap is zero", () => {
116 expect(
117 decideThreadResume({
118 afterSequence: 10,
119 highWater: 10,
120 replayEventCount: 0,
121 replayEncodedBytes: 0,
122 }),
123 ).toEqual({ mode: "replay", afterSequence: 10, throughSequence: 10 });
124 });
125
126 it("replays when the event count is within the bound", () => {
127 expect(
128 decideThreadResume({
129 afterSequence: 10,
130 highWater: 20_000,
131 replayEventCount: THREAD_RESUME_MAX_REPLAY_EVENTS,
132 replayEncodedBytes: THREAD_RESUME_MAX_REPLAY_ENCODED_BYTES,
133 }),
134 ).toEqual({
135 mode: "replay",
136 afterSequence: 10,
137 throughSequence: 20_000,
138 });
139 });
140
141 it("falls back to a snapshot when the event count exceeds the bound", () => {
142 expect(
143 decideThreadResume({
144 afterSequence: 10,
145 highWater: 20_000,
146 replayEventCount: THREAD_RESUME_MAX_REPLAY_EVENTS + 1,
147 replayEncodedBytes: 1,
148 }),
149 ).toEqual({ mode: "snapshot" });
150 });
151
152 it("falls back to a snapshot when encoded replay bytes exceed the bound", () => {
153 expect(
154 decideThreadResume({
155 afterSequence: 10,
156 highWater: 11,
157 replayEventCount: 1,
158 replayEncodedBytes: THREAD_RESUME_MAX_REPLAY_ENCODED_BYTES + 1,
159 }),
160 ).toEqual({ mode: "snapshot" });
161 });
162
163 it("rejects pathological raw payloads before decoding them", () => {
164 expect(isThreadReplayRawPayloadSafe(THREAD_RESUME_MAX_RAW_PAYLOAD_BYTES)).toBe(true);
165 expect(isThreadReplayRawPayloadSafe(THREAD_RESUME_MAX_RAW_PAYLOAD_BYTES + 1)).toBe(false);
166
167 // Raw safety and projected transport size are deliberately separate. A
168 // tiny projected replay is transport-safe even when its raw source is not.
169 expect(
170 decideThreadResume({
171 afterSequence: 9,
172 highWater: 10,
173 replayEventCount: 1,
174 replayEncodedBytes: 1,
175 }),
176 ).toEqual({ mode: "replay", afterSequence: 9, throughSequence: 10 });
177 });
178
179 it("falls back to a snapshot when the client cursor is ahead of the store", () => {
180 expect(
181 decideThreadResume({
182 afterSequence: 50,
183 highWater: 40,
184 replayEventCount: 0,
185 replayEncodedBytes: 0,
186 }),
187 ).toEqual({ mode: "snapshot" });
188 });
189
190 it("counts UTF-8 bytes across projected stream items", () => {
191 expect(threadReplayEncodedBytes([{ value: "a" }, { value: "🦊" }])).toBe(
192 Buffer.byteLength('{"value":"a"}', "utf8") + Buffer.byteLength('{"value":"🦊"}', "utf8"),
193 );
194 });
195
196 it("builds socket fallback snapshots with a bounded timeline and history cursor", () => {
197 const snapshot = buildBoundedThreadStreamSnapshot({
198 snapshotSequence: 44,
199 projection: timelineProjection(80),
200 });
201
202 expect(snapshot.kind).toBe("snapshot");
203 expect(snapshot.snapshotSequence).toBe(44);
204 expect(snapshot.projection.visibleTurnItems).toHaveLength(75);
205 expect(snapshot.projection.turnItems).toHaveLength(75);
206 expect(snapshot.historyCursor).not.toBeNull();
207 expect(snapshot.hasMoreHistory).toBe(true);
208 expect(snapshot.latestLocalTurnOrdinal).toBe(80);
209 expect(snapshot.payloadBudgetExceeded).toBe(false);
210 });
211
212 it("rejects a 10 MiB raw replay even when its projected form fits the wire budget", () => {
213 const item = {
214 id: TurnItemId.make("large-dynamic-tool"),
215 type: "dynamic_tool",
216 threadId: THREAD_ID,
217 runId: null,
218 nodeId: null,
219 providerThreadId: null,
220 providerTurnId: null,
221 nativeItemRef: null,
222 parentItemId: null,
223 ordinal: 1,
224 status: "completed",
225 title: "Large result",
226 toolName: "mcp__test__large_result",
227 input: { query: "small" },
228 output: { text: "x".repeat(LARGE_PROJECTABLE_OUTPUT_BYTES) },
229 startedAt: NOW,
230 completedAt: NOW,
231 updatedAt: NOW,
232 } satisfies OrchestrationV2TurnItem;
233 const event = {
234 id: EventId.make("large-dynamic-tool-event"),
235 type: "turn-item.updated" as const,
236 threadId: THREAD_ID,
237 occurredAt: NOW,
238 payload: item,
239 };
240 const rawReplay = [{ kind: "event" as const, sequence: 10, event }];
241 const projectedReplay = [
242 {
243 kind: "event" as const,
244 sequence: 10,
245 event: projectDomainEventForWire(event),
246 },
247 ];
248
249 const rawPayloadBytes = Buffer.byteLength(JSON.stringify(event.payload), "utf8");
250 expect(rawPayloadBytes).toBe(LARGE_PROJECTABLE_RAW_PAYLOAD_BYTES);
251 expect(rawPayloadBytes).toBeGreaterThan(THREAD_RESUME_MAX_RAW_PAYLOAD_BYTES);
252 expect(isThreadReplayRawPayloadSafe(rawPayloadBytes)).toBe(false);
253 expect(threadReplayEncodedBytes(rawReplay)).toBeGreaterThan(
254 THREAD_RESUME_MAX_REPLAY_ENCODED_BYTES,
255 );
256 const projectedBytes = threadReplayEncodedBytes(projectedReplay);
257 expect(projectedBytes).toBeLessThanOrEqual(MAX_PROJECTED_DYNAMIC_REPLAY_BYTES);
258 expect(
259 decideThreadResume({
260 afterSequence: 9,
261 highWater: 10,
262 replayEventCount: projectedReplay.length,
263 replayEncodedBytes: projectedBytes,
264 }),
265 ).toEqual({ mode: "replay", afterSequence: 9, throughSequence: 10 });
266 });
267});
268
269describe("compact bounded snapshots", () => {
270 const FORK_PARENT = ThreadId.make("thread-stream-parent");
271 const decodeStreamItem = Schema.decodeUnknownSync(
272 Schema.toCodecJson(OrchestrationV2ThreadStreamItem),
273 );
274 const encodeStreamItem = Schema.encodeSync(Schema.toCodecJson(OrchestrationV2ThreadStreamItem));
275
276 /** Fork child: inherited parent rows and marker, then local rows, plus an older interrupt request. */
277 function forkProjection(): OrchestrationV2ThreadProjection {
278 const base = timelineProjection(6);
279 const local = base.visibleTurnItems.slice(3).map((row, index) => ({ ...row, position: index }));
280 const inherited = base.visibleTurnItems.slice(0, 3).map((row, index) => {
281 const item = { ...row.item, id: TurnItemId.make(`parent-${index}`), threadId: FORK_PARENT };
282 return {
283 position: index,
284 visibility: "inherited" as const,
285 sourceThreadId: FORK_PARENT,
286 sourceItemId: item.id,
287 item,
288 };
289 });
290 const request = {
291 ...base.turnItems[0]!,
292 id: TurnItemId.make("interrupt-request"),
293 type: "run_interrupt_request",
294 ordinal: 0,
295 message: "Stop",
296 } as unknown as OrchestrationV2TurnItem;
297 return {
298 ...base,
299 turnItems: [request, ...local.map((row) => row.item)],
300 visibleTurnItems: [...inherited, ...local].map((row, position) => ({ ...row, position })),
301 };
302 }
303
304 it.each([
305 ["linear", () => timelineProjection(80)],
306 ["fork with retained interrupt request", forkProjection],
307 ])("restores the exact opted-out snapshot for %s threads", (_, makeProjection) => {
308 const base = makeProjection();
309 // The shared fixture omits a model; the wire codec requires one.
310 const projection = {
311 ...base,
312 thread: { ...base.thread, modelSelection: { instanceId: "codex", model: "gpt-5" } },
313 } as OrchestrationV2ThreadProjection;
314 const full = buildBoundedThreadStreamSnapshot({ snapshotSequence: 44, projection });
315 const compact = buildBoundedThreadStreamSnapshot({
316 snapshotSequence: 44,
317 projection,
318 compactTurnItems: true,
319 });
320
321 // Opted-out clients get exactly the previous representation.
322 expect("turnItemsOmitLocalVisible" in full).toBe(false);
323 expect(compact.turnItemsOmitLocalVisible).toBe(true);
324 expect(compact.projection.turnItems.length).toBeLessThan(full.projection.turnItems.length);
325 const { turnItemsOmitLocalVisible: _marker, ...compactFields } = compact;
326 expect({ ...compactFields, projection: full.projection }).toEqual(full);
327
328 // Decode the wire form like a client, then restore before use.
329 const decodedFull = decodeStreamItem(JSON.parse(JSON.stringify(encodeStreamItem(full))));
330 const decodedCompact = decodeStreamItem(JSON.parse(JSON.stringify(encodeStreamItem(compact))));
331 if (decodedFull.kind !== "snapshot" || decodedCompact.kind !== "snapshot") {
332 throw new Error("Expected snapshots");
333 }
334 expect(boundedSnapshotProjection(decodedCompact)).toEqual(decodedFull.projection);
335 });
336
337 it("keeps the retained interrupt request outside the visible window", () => {
338 const compact = buildBoundedThreadStreamSnapshot({
339 snapshotSequence: 1,
340 projection: forkProjection(),
341 compactTurnItems: true,
342 });
343 expect(compact.projection.turnItems.map((item) => item.id)).toEqual(["interrupt-request"]);
344 expect(boundedSnapshotProjection(compact).turnItems.map((item) => String(item.id))).toEqual([
345 "item-3",
346 "item-4",
347 "item-5",
348 "interrupt-request",
349 ]);
350 });
351
352 it("omits the marker when the window has no local rows", () => {
353 const fork = forkProjection();
354 const inheritedOnly = {
355 ...fork,
356 turnItems: fork.turnItems.slice(0, 1),
357 visibleTurnItems: fork.visibleTurnItems.slice(0, 3),
358 };
359 const compact = buildBoundedThreadStreamSnapshot({
360 snapshotSequence: 1,
361 projection: inheritedOnly,
362 compactTurnItems: true,
363 });
364 expect("turnItemsOmitLocalVisible" in compact).toBe(false);
365 expect(compact.projection.turnItems.map((item) => item.id)).toEqual(["interrupt-request"]);
366 });
367});