1import {
2 EventId,
3 OrchestrationV2RpcSchemas,
4 ProjectId,
5 ProviderInstanceId,
6 ThreadId,
7 TurnItemId,
8 type OrchestrationV2ProjectedTurnItem,
9 type OrchestrationV2DomainEvent,
10 type OrchestrationV2ThreadProjection,
11 type OrchestrationV2ThreadStreamItem,
12 type OrchestrationV2TurnItem,
13} from "@t3tools/contracts";
14import * as DateTime from "effect/DateTime";
15import * as Schema from "effect/Schema";
16import type * as RpcMessage from "effect/rpc/RpcMessage";
17import { describe, expect, it } from "vite-plus/test";
18
19import { buildBoundedThreadStreamSnapshot } from "./ThreadStream.ts";
20import { buildBoundedThreadProjection } from "./threadHistoryPaging.ts";
21import { projectDomainEventForWire } from "./WireProjection.ts";
22
23const SNAPSHOT_SEQUENCE = 600;
24const ROW_COUNT = 600;
25const OUTPUT_BYTES_PER_ROW = 8_192;
26const NOW = DateTime.makeUnsafe("2026-06-20T00:00:00.000Z");
27const THREAD_ID = ThreadId.make("thread-perf");
28const PROVIDER_INSTANCE_ID = ProviderInstanceId.make("codex");
29
30// Recorded before bounded WebSocket fallback at 91b193653ec. Keep this number
31// here so later transport changes retain a direct, reproducible comparison.
32const HISTORICAL_FULL_SNAPSHOT_APPLICATION_BYTES = 10_375_079;
33// Contract encoding plus the Effect RPC Chunk envelope adds 42 bytes to both
34// snapshot variants. WebSocket framing and compression are intentionally out
35// of scope for this deterministic pre-compression measurement.
36const FULL_SNAPSHOT_RPC_JSON_BYTES = 10_375_121;
37const PRE_OMISSION_BOUNDED_SNAPSHOT_RPC_JSON_BYTES = 1_038_647;
38// The first payload-omitting projection measured 67,412 bytes.
39const MAX_PROJECTED_BOUNDED_SNAPSHOT_RPC_JSON_BYTES = 131_072;
40const PRE_OMISSION_COMMAND_EVENT_RPC_JSON_BYTES = 8_790;
41// The first payload-omitting projection measured 586 bytes.
42const MAX_PROJECTED_COMMAND_EVENT_RPC_JSON_BYTES = 1_024;
43
44const encodeThreadStreamItem = Schema.encodeSync(
45 Schema.toCodecJson(OrchestrationV2RpcSchemas.subscribeThread.output),
46);
47
48function encodedBytes(value: unknown): number {
49 return Buffer.byteLength(JSON.stringify(value), "utf8");
50}
51
52function encodedThreadStreamChunkBytes(value: OrchestrationV2ThreadStreamItem): number {
53 const response = {
54 _tag: "Chunk",
55 requestId: 0,
56 values: [encodeThreadStreamItem(value)],
57 } satisfies RpcMessage.ResponseChunkEncoded;
58 return encodedBytes(response);
59}
60
61function makeCommandItem(index: number): OrchestrationV2TurnItem {
62 return {
63 id: TurnItemId.make(`item-${index}`),
64 type: "command_execution",
65 threadId: THREAD_ID,
66 runId: null,
67 nodeId: null,
68 providerThreadId: null,
69 providerTurnId: null,
70 nativeItemRef: null,
71 parentItemId: null,
72 ordinal: index + 1,
73 status: "completed",
74 title: `Command ${index}`,
75 input: `cmd-${index}`,
76 output: "x".repeat(OUTPUT_BYTES_PER_ROW),
77 exitCode: 0,
78 startedAt: NOW,
79 completedAt: NOW,
80 updatedAt: NOW,
81 };
82}
83
84function makeProjection(): OrchestrationV2ThreadProjection {
85 const visibleTurnItems: OrchestrationV2ProjectedTurnItem[] = Array.from(
86 { length: ROW_COUNT },
87 (_, index) => {
88 const item = makeCommandItem(index);
89 return {
90 position: index,
91 visibility: "local",
92 sourceThreadId: THREAD_ID,
93 sourceItemId: item.id,
94 item,
95 };
96 },
97 );
98
99 return {
100 thread: {
101 id: THREAD_ID,
102 projectId: ProjectId.make("project-perf"),
103 title: "Thread",
104 providerInstanceId: PROVIDER_INSTANCE_ID,
105 modelSelection: { instanceId: PROVIDER_INSTANCE_ID, model: "gpt-5" },
106 runtimeMode: "full-access",
107 interactionMode: "default",
108 branch: null,
109 worktreePath: null,
110 activeProviderThreadId: null,
111 lineage: {
112 parentThreadId: null,
113 relationshipToParent: null,
114 rootThreadId: THREAD_ID,
115 },
116 forkedFrom: null,
117 createdBy: "user",
118 creationSource: "web",
119 createdAt: NOW,
120 updatedAt: NOW,
121 archivedAt: null,
122 deletedAt: null,
123 settledOverride: null,
124 settledAt: null,
125 lastVisitedAt: null,
126 },
127 runs: [],
128 attempts: [],
129 nodes: [],
130 subagents: [],
131 providerSessions: [],
132 providerThreads: [],
133 providerTurns: [],
134 runtimeRequests: [],
135 messages: [],
136 plans: [],
137 turnItems: visibleTurnItems.map((row) => row.item),
138 checkpointScopes: [],
139 checkpoints: [],
140 contextHandoffs: [],
141 contextTransfers: [],
142 visibleTurnItems,
143 updatedAt: NOW,
144 };
145}
146
147function makeTurnItemEvent(item: OrchestrationV2TurnItem): OrchestrationV2DomainEvent {
148 return {
149 id: EventId.make(`event-${item.id}`),
150 type: "turn-item.updated",
151 threadId: THREAD_ID,
152 occurredAt: NOW,
153 payload: item,
154 };
155}
156
157describe("thread transport payload budget", () => {
158 it("records full, bounded, and payload-omitting snapshot improvements", () => {
159 const projection = makeProjection();
160 const historicalFullSnapshot: OrchestrationV2ThreadStreamItem = {
161 kind: "snapshot",
162 snapshotSequence: SNAPSHOT_SEQUENCE,
163 projection,
164 };
165 const preOmissionBounded = buildBoundedThreadProjection({
166 projection,
167 snapshotSequence: SNAPSHOT_SEQUENCE,
168 });
169 const preOmissionBoundedSnapshot: OrchestrationV2ThreadStreamItem = {
170 kind: "snapshot",
171 snapshotSequence: SNAPSHOT_SEQUENCE,
172 projection: preOmissionBounded.projection,
173 historyCursor: preOmissionBounded.historyCursor,
174 hasMoreHistory: preOmissionBounded.hasMoreHistory,
175 latestLocalTurnOrdinal: preOmissionBounded.latestLocalTurnOrdinal,
176 payloadBudgetExceeded: preOmissionBounded.payloadBudgetExceeded,
177 };
178 const projectedBoundedSnapshot = buildBoundedThreadStreamSnapshot({
179 projection,
180 snapshotSequence: SNAPSHOT_SEQUENCE,
181 });
182
183 const historicalApplicationBytes = encodedBytes(historicalFullSnapshot);
184 const fullRpcJsonBytes = encodedThreadStreamChunkBytes(historicalFullSnapshot);
185 const preOmissionBoundedRpcJsonBytes = encodedThreadStreamChunkBytes(
186 preOmissionBoundedSnapshot,
187 );
188 const projectedBoundedRpcJsonBytes = encodedThreadStreamChunkBytes(projectedBoundedSnapshot);
189
190 expect(historicalApplicationBytes).toBe(HISTORICAL_FULL_SNAPSHOT_APPLICATION_BYTES);
191 expect(fullRpcJsonBytes).toBe(FULL_SNAPSHOT_RPC_JSON_BYTES);
192 expect(preOmissionBoundedRpcJsonBytes).toBe(PRE_OMISSION_BOUNDED_SNAPSHOT_RPC_JSON_BYTES);
193 expect(projectedBoundedRpcJsonBytes).toBeLessThanOrEqual(
194 MAX_PROJECTED_BOUNDED_SNAPSHOT_RPC_JSON_BYTES,
195 );
196 expect(projectedBoundedRpcJsonBytes / fullRpcJsonBytes).toBeLessThanOrEqual(0.02);
197 expect(preOmissionBoundedSnapshot.projection.visibleTurnItems).toHaveLength(60);
198 expect(projectedBoundedSnapshot.projection.visibleTurnItems).toHaveLength(75);
199 expect(projectedBoundedSnapshot.hasMoreHistory).toBe(true);
200 expect(projectedBoundedSnapshot.payloadBudgetExceeded).toBe(false);
201 });
202
203 it("omits command output from a live event without changing persisted source data", () => {
204 const sourceItem = makeCommandItem(0);
205 const sourceEvent = makeTurnItemEvent(sourceItem);
206 const projectedEvent = projectDomainEventForWire(sourceEvent);
207 const preOmissionStreamItem: OrchestrationV2ThreadStreamItem = {
208 kind: "event",
209 sequence: 1,
210 event: sourceEvent,
211 };
212 const projectedStreamItem: OrchestrationV2ThreadStreamItem = {
213 kind: "event",
214 sequence: 1,
215 event: projectedEvent,
216 };
217
218 const preOmissionRpcJsonBytes = encodedThreadStreamChunkBytes(preOmissionStreamItem);
219 const projectedRpcJsonBytes = encodedThreadStreamChunkBytes(projectedStreamItem);
220
221 expect(preOmissionRpcJsonBytes).toBe(PRE_OMISSION_COMMAND_EVENT_RPC_JSON_BYTES);
222 expect(projectedRpcJsonBytes).toBeLessThanOrEqual(MAX_PROJECTED_COMMAND_EVENT_RPC_JSON_BYTES);
223 expect(projectedRpcJsonBytes / preOmissionRpcJsonBytes).toBeLessThanOrEqual(0.1);
224 expect(sourceItem).toHaveProperty("output", "x".repeat(OUTPUT_BYTES_PER_ROW));
225 expect(projectedEvent.payload).not.toHaveProperty("output");
226 });
227
228 it("omits ordinary dynamic tool output even when it is small", () => {
229 const sourceItem: OrchestrationV2TurnItem = {
230 id: TurnItemId.make("item-dynamic"),
231 type: "dynamic_tool",
232 threadId: THREAD_ID,
233 runId: null,
234 nodeId: null,
235 providerThreadId: null,
236 providerTurnId: null,
237 nativeItemRef: null,
238 parentItemId: null,
239 ordinal: 1,
240 status: "completed",
241 title: "Dynamic tool",
242 toolName: "lookup",
243 input: { query: "status" },
244 output: { content: "small raw result" },
245 startedAt: NOW,
246 completedAt: NOW,
247 updatedAt: NOW,
248 };
249 const projectedEvent = projectDomainEventForWire(makeTurnItemEvent(sourceItem));
250
251 expect(sourceItem).toHaveProperty("output", { content: "small raw result" });
252 expect(projectedEvent.payload).not.toHaveProperty("output");
253 });
254});