1import type { ProviderDriverKind } from "@t3tools/contracts";
2
3import type {
4 HttpTransferMeasurement,
5 WebSocketTransferTotals,
6} from "./NetworkTransferMeasurement.integration.ts";
7import {
8 TRANSFER_HISTORY_MCP_RESULT_BYTES,
9 TRANSFER_HISTORY_TOOLS_PER_TURN,
10 TRANSFER_HISTORY_TURN_COUNT,
11 TRANSFER_MEASURED_MCP_RESULT_BYTES,
12 TRANSFER_MEASURED_TOOLS,
13} from "./fixtures/transferBudget.ts";
14
15/** Catch-up delivered to a resubscribing client, and which path the server chose. */
16export interface WebSocketCatchUpMeasurement extends WebSocketTransferTotals {
17 readonly mode: "replay" | "snapshot";
18}
19
20export interface TransferBudgetRun {
21 readonly provider: ProviderDriverKind;
22 readonly threadSnapshot: HttpTransferMeasurement;
23 /** One socket holding only the thread subscription. This is the capped measurement. */
24 readonly measuredTurnWebSocket: WebSocketTransferTotals;
25 readonly shellSnapshot: HttpTransferMeasurement;
26 /** One socket holding only the shell (sidebar) subscription during the same turn. */
27 readonly measuredTurnShellWebSocket: WebSocketTransferTotals;
28 /** A second client: one socket holding both the thread and shell subscriptions. */
29 readonly measuredTurnSecondClientWebSocket: WebSocketTransferTotals;
30 /** The second client resubscribes after the turn from the cursor it held before it. */
31 readonly reconnectThread: WebSocketCatchUpMeasurement;
32 readonly reconnectShell: WebSocketCatchUpMeasurement;
33 /** SQL statement spans opened server-wide during the measured turn. */
34 readonly measuredTurnSqlStatements: number;
35 /** SQL statement spans opened while serving both reconnect catch-ups. */
36 readonly reconnectSqlStatements: number;
37}
38
39interface ProviderTransferBudget {
40 readonly totalWireBytes: number;
41 readonly threadSnapshotWireBytes: number;
42 readonly measuredTurnWebSocketWireBytes: number;
43 readonly measuredTurnWebSocketDecodedBytes: number;
44 readonly measuredTurnWebSocketMessages: number;
45}
46
47// These caps leave at least 30% headroom above the client projection of the
48// deterministic 9 MB retained-result fixture. Full MCP results stay in
49// persistence, so accidentally shipping them again exceeds these caps by
50// orders of magnitude. The CI report preserves exact values for review.
51const TRANSFER_BUDGET = {
52 totalWireBytes: 7_000,
53 threadSnapshotWireBytes: 5_000,
54 measuredTurnWebSocketWireBytes: 2_000,
55 measuredTurnWebSocketDecodedBytes: 30_000,
56 measuredTurnWebSocketMessages: 8,
57} satisfies ProviderTransferBudget;
58
59const TRANSFER_BUDGETS: Readonly<Record<string, ProviderTransferBudget>> = {
60 codex: TRANSFER_BUDGET,
61 claudeAgent: TRANSFER_BUDGET,
62};
63
64function totalWireBytes(run: TransferBudgetRun): number {
65 return run.threadSnapshot.wireBytes + run.measuredTurnWebSocket.wireBytes;
66}
67
68/** Bytes the server wrote to every measured socket during the turn. */
69function serverEgressWireBytes(run: TransferBudgetRun): number {
70 return (
71 run.measuredTurnWebSocket.wireBytes +
72 run.measuredTurnShellWebSocket.wireBytes +
73 run.measuredTurnSecondClientWebSocket.wireBytes
74 );
75}
76
77function observedTransfer(run: TransferBudgetRun) {
78 return {
79 totalWireBytes: totalWireBytes(run),
80 threadSnapshotWireBytes: run.threadSnapshot.wireBytes,
81 threadSnapshotDecodedBytes: run.threadSnapshot.decodedBodyBytes,
82 measuredTurnWebSocketWireBytes: run.measuredTurnWebSocket.wireBytes,
83 measuredTurnWebSocketDecodedBytes: run.measuredTurnWebSocket.decodedBytes,
84 measuredTurnWebSocketMessages: run.measuredTurnWebSocket.messages,
85 };
86}
87
88/** Machine-readable input for the trusted PR comment publisher. */
89export function formatTransferBudgetResult(runs: ReadonlyArray<TransferBudgetRun>): string {
90 const providers = Object.fromEntries(
91 runs.flatMap((run) => {
92 const ceiling = TRANSFER_BUDGETS[run.provider];
93 return ceiling ? [[run.provider, { observed: observedTransfer(run), ceiling }]] : [];
94 }),
95 );
96
97 return `${JSON.stringify(
98 {
99 schemaVersion: 1,
100 scenario: {
101 id: "thread-transfer-v2",
102 historyTurns: TRANSFER_HISTORY_TURN_COUNT,
103 historyCommandToolsPerTurn: TRANSFER_HISTORY_TOOLS_PER_TURN,
104 historyMcpResultBytes: TRANSFER_HISTORY_MCP_RESULT_BYTES,
105 measuredCommandTools: TRANSFER_MEASURED_TOOLS,
106 measuredMcpResultBytes: TRANSFER_MEASURED_MCP_RESULT_BYTES,
107 },
108 providers,
109 },
110 null,
111 2,
112 )}\n`;
113}
114
115function formatBytes(bytes: number): string {
116 if (bytes < 1_024) return `${bytes} B`;
117 if (bytes >= 1_024 * 1_024) {
118 return `${(bytes / 1_024 / 1_024).toFixed(2)} MiB (${bytes.toLocaleString("en-US")} B)`;
119 }
120 return `${(bytes / 1_024).toFixed(1)} KiB (${bytes.toLocaleString("en-US")} B)`;
121}
122
123function row(
124 provider: ProviderDriverKind,
125 phase: string,
126 metric: string,
127 observed: number,
128 maximum: number,
129 format: (value: number) => string = formatBytes,
130): string {
131 const status = observed <= maximum ? "PASS" : "FAIL";
132 return `| ${provider} | ${phase} | ${metric} | ${format(observed)} | ${format(maximum)} | ${status} |`;
133}
134
135// Shell, second-client, reconnect, and SQL rows are reported without a cap.
136// Shell delivery coalesces on a 50 ms window, so message counts and bytes move
137// with scheduler timing between runs, and the reconnect and SQL figures follow
138// the same batches. The rows exist so CI shows the numbers next to the capped
139// thread measurement.
140function infoRow(
141 provider: ProviderDriverKind,
142 phase: string,
143 metric: string,
144 observed: number,
145 format: (value: number) => string = formatBytes,
146): string {
147 return `| ${provider} | ${phase} | ${metric} | ${format(observed)} | none | INFO |`;
148}
149
150function webSocketRows(
151 provider: ProviderDriverKind,
152 phase: string,
153 totals: WebSocketTransferTotals,
154): string[] {
155 return [
156 infoRow(provider, phase, "WebSocket wire", totals.wireBytes),
157 infoRow(provider, phase, "WebSocket decoded", totals.decodedBytes),
158 infoRow(provider, phase, "WebSocket messages", totals.messages, String),
159 ];
160}
161
162export function transferBudgetViolations(runs: ReadonlyArray<TransferBudgetRun>): string[] {
163 const violations: string[] = [];
164 for (const run of runs) {
165 const budget = TRANSFER_BUDGETS[run.provider];
166 if (!budget) {
167 violations.push(`${run.provider}: no transfer budget is configured`);
168 continue;
169 }
170 const checks = [
171 ["total thread wire bytes", totalWireBytes(run), budget.totalWireBytes],
172 ["thread snapshot wire bytes", run.threadSnapshot.wireBytes, budget.threadSnapshotWireBytes],
173 [
174 "measured-turn WebSocket wire bytes",
175 run.measuredTurnWebSocket.wireBytes,
176 budget.measuredTurnWebSocketWireBytes,
177 ],
178 [
179 "measured-turn WebSocket decoded bytes",
180 run.measuredTurnWebSocket.decodedBytes,
181 budget.measuredTurnWebSocketDecodedBytes,
182 ],
183 [
184 "measured-turn WebSocket messages",
185 run.measuredTurnWebSocket.messages,
186 budget.measuredTurnWebSocketMessages,
187 ],
188 ] as const;
189 for (const [metric, observed, maximum] of checks) {
190 if (observed > maximum) {
191 violations.push(`${run.provider}: ${metric} was ${observed}, maximum ${maximum}`);
192 }
193 }
194 }
195 return violations;
196}
197
198export function formatTransferBudgetReport(runs: ReadonlyArray<TransferBudgetRun>): string {
199 const lines = [
200 "# T3 Code thread transfer budget",
201 "",
202 "Wire values are thread data bytes read from local HTTP and WebSocket sockets. HTTP includes response headers; WebSocket measurement starts after the resumed thread subscription synchronizes. TCP/IP, TLS framing, and the WebSocket upgrade are excluded. WebSocket permessage-deflate is negotiated.",
203 "The measured turn is observed on three sockets at once: one with only the thread subscription (the capped rows), one with only the shell subscription, and a second client holding both. Server egress is the sum of the three. After the turn the second client disconnects and new sockets resubscribe from the cursor it held before the turn, which is the cursor a backgrounded phone would hold. SQL statements are the SQL client's per-statement spans, counted across V2 persistence and the HTTP/WebSocket handlers.",
204 `Scenario: ${TRANSFER_HISTORY_TURN_COUNT} historical turns with ${TRANSFER_HISTORY_TOOLS_PER_TURN} command tools and one retained ${formatBytes(TRANSFER_HISTORY_MCP_RESULT_BYTES)} MCP result each, followed by one measured turn with ${TRANSFER_MEASURED_TOOLS} command tools and a retained ${formatBytes(TRANSFER_MEASURED_MCP_RESULT_BYTES)} MCP result. Synthetic V2 domain events exercise persistence, wire projection, and production HTTP/subscription handlers; this does not measure provider adapter ingestion. Payloads contain no user data.`,
205 "",
206 "| Provider | Total thread wire | Budget | Result |",
207 "| --- | ---: | ---: | --- |",
208 ...runs.flatMap((run) => {
209 const budget = TRANSFER_BUDGETS[run.provider];
210 if (!budget) return [];
211 const observed = observedTransfer(run).totalWireBytes;
212 return [
213 `| ${run.provider} | ${formatBytes(observed)} | ${formatBytes(budget.totalWireBytes)} | ${observed <= budget.totalWireBytes ? "PASS" : "FAIL"} |`,
214 ];
215 }),
216 "",
217 "## Detailed measurements",
218 "",
219 "| Provider | Phase | Metric | Observed | Budget | Result |",
220 "| --- | --- | --- | ---: | ---: | --- |",
221 ];
222
223 for (const run of runs) {
224 const budget = TRANSFER_BUDGETS[run.provider];
225 if (!budget) continue;
226 lines.push(
227 row(
228 run.provider,
229 "thread snapshot",
230 "HTTP wire",
231 run.threadSnapshot.wireBytes,
232 budget.threadSnapshotWireBytes,
233 ),
234 row(
235 run.provider,
236 "measured turn",
237 "WebSocket wire",
238 run.measuredTurnWebSocket.wireBytes,
239 budget.measuredTurnWebSocketWireBytes,
240 ),
241 row(
242 run.provider,
243 "measured turn",
244 "WebSocket decoded",
245 run.measuredTurnWebSocket.decodedBytes,
246 budget.measuredTurnWebSocketDecodedBytes,
247 ),
248 row(
249 run.provider,
250 "measured turn",
251 "WebSocket messages",
252 run.measuredTurnWebSocket.messages,
253 budget.measuredTurnWebSocketMessages,
254 String,
255 ),
256 infoRow(run.provider, "shell snapshot", "HTTP wire", run.shellSnapshot.wireBytes),
257 ...webSocketRows(run.provider, "measured turn, shell", run.measuredTurnShellWebSocket),
258 ...webSocketRows(
259 run.provider,
260 "measured turn, second client",
261 run.measuredTurnSecondClientWebSocket,
262 ),
263 infoRow(run.provider, "measured turn", "server egress wire", serverEgressWireBytes(run)),
264 infoRow(
265 run.provider,
266 "measured turn",
267 "SQL statements",
268 run.measuredTurnSqlStatements,
269 String,
270 ),
271 ...webSocketRows(
272 run.provider,
273 `reconnect, thread (${run.reconnectThread.mode})`,
274 run.reconnectThread,
275 ),
276 ...webSocketRows(
277 run.provider,
278 `reconnect, shell (${run.reconnectShell.mode})`,
279 run.reconnectShell,
280 ),
281 infoRow(run.provider, "reconnect", "SQL statements", run.reconnectSqlStatements, String),
282 );
283 }
284
285 lines.push("", "## Compression diagnostics", "");
286 for (const run of runs) {
287 lines.push(
288 `- ${run.provider}: thread snapshot ${formatBytes(run.threadSnapshot.decodedBodyBytes)} decoded to ${formatBytes(run.threadSnapshot.encodedBodyBytes)} gzip.`,
289 `- ${run.provider}: shell snapshot ${formatBytes(run.shellSnapshot.decodedBodyBytes)} decoded to ${formatBytes(run.shellSnapshot.encodedBodyBytes)} gzip.`,
290 );
291 }
292
293 return `${lines.join("\n")}\n`;
294}