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