authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-08-07 02:17:49-04:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-08-06 23:17:49-07:00
logddfe45c66eccd93c0adf61db92a16cbbbcde23e1
tree977180d7b01a014a630aafff27f26bcfc14062a3
parent9547cf24634ccafe5d381964449b722d33f812b8
signature Signed by PGP key B5690EEEBB952194

test(server): catch client transfer regressions in CI (#5350)

12 files changed, 1973 insertions(+), 42 deletions(-)

.github/scripts/thread-transfer-report.cjs created+429
...@@ -0,0 +1,429 @@
1const fs = require("node:fs");
2const path = require("node:path");
3
4const ARTIFACT_NAME = "thread-transfer-results";
5const RESULT_FILE = "thread-transfer-result.json";
6const COMMENT_MARKER = "<!-- t3-thread-transfer-report -->";
7const PROVIDERS = ["codex", "claudeAgent"];
8const OBSERVED_KEYS = [
9 "totalWireBytes",
10 "threadSnapshotWireBytes",
11 "threadSnapshotDecodedBytes",
12 "measuredTurnWebSocketWireBytes",
13 "measuredTurnWebSocketDecodedBytes",
14 "measuredTurnWebSocketMessages",
15];
16const CEILING_KEYS = [
17 "totalWireBytes",
18 "threadSnapshotWireBytes",
19 "measuredTurnWebSocketWireBytes",
20 "measuredTurnWebSocketDecodedBytes",
21 "measuredTurnWebSocketMessages",
22];
23const SCENARIO_KEYS = [
24 "id",
25 "historyTurns",
26 "historyCommandToolsPerTurn",
27 "historyMcpResultBytes",
28 "measuredCommandTools",
29 "measuredMcpResultBytes",
30];
31
32function resultShaMarker(sha) {
33 return `<!-- t3-thread-transfer-result-sha:${sha} -->`;
34}
35
36function assertObject(value, label) {
37 if (value === null || typeof value !== "object" || Array.isArray(value)) {
38 throw new Error(`${label} must be an object`);
39 }
40}
41
42function assertExactKeys(value, expected, label) {
43 assertObject(value, label);
44 const actual = Object.keys(value).sort();
45 const wanted = [...expected].sort();
46 if (actual.length !== wanted.length || actual.some((key, index) => key !== wanted[index])) {
47 throw new Error(`${label} has unexpected fields`);
48 }
49}
50
51function assertMetric(value, label) {
52 if (!Number.isSafeInteger(value) || value < 0 || value > 1_000_000_000) {
53 throw new Error(`${label} must be a non-negative safe integer below 1,000,000,000`);
54 }
55}
56
57function validateResult(value) {
58 assertExactKeys(value, ["schemaVersion", "scenario", "providers"], "result");
59 if (value.schemaVersion !== 1) {
60 throw new Error("result.schemaVersion must be 1");
61 }
62
63 assertExactKeys(value.scenario, SCENARIO_KEYS, "result.scenario");
64 if (value.scenario.id !== "thread-transfer-v1") {
65 throw new Error("result.scenario.id is not supported");
66 }
67 for (const key of SCENARIO_KEYS.slice(1)) {
68 assertMetric(value.scenario[key], `result.scenario.${key}`);
69 }
70
71 assertExactKeys(value.providers, PROVIDERS, "result.providers");
72 for (const provider of PROVIDERS) {
73 const entry = value.providers[provider];
74 assertExactKeys(entry, ["observed", "ceiling"], `result.providers.${provider}`);
75 assertExactKeys(entry.observed, OBSERVED_KEYS, `result.providers.${provider}.observed`);
76 assertExactKeys(entry.ceiling, CEILING_KEYS, `result.providers.${provider}.ceiling`);
77 for (const key of OBSERVED_KEYS) {
78 assertMetric(entry.observed[key], `result.providers.${provider}.observed.${key}`);
79 }
80 for (const key of CEILING_KEYS) {
81 assertMetric(entry.ceiling[key], `result.providers.${provider}.ceiling.${key}`);
82 }
83 }
84
85 return value;
86}
87
88function readResult(directory) {
89 if (!directory) return undefined;
90 const file = path.join(directory, RESULT_FILE);
91 if (!fs.existsSync(file)) return undefined;
92 const stat = fs.lstatSync(file);
93 if (!stat.isFile() || stat.size > 64 * 1_024) {
94 throw new Error("thread transfer result must be a regular file smaller than 64 KiB");
95 }
96 return validateResult(JSON.parse(fs.readFileSync(file, "utf8")));
97}
98
99function formatBytes(bytes) {
100 if (bytes < 1_024) return `${bytes} B`;
101 if (bytes >= 1_024 * 1_024) return `${(bytes / 1_024 / 1_024).toFixed(2)} MiB`;
102 return `${(bytes / 1_024).toFixed(1)} KiB`;
103}
104
105function formatValue(value, kind) {
106 return kind === "messages" ? value.toLocaleString("en-US") : formatBytes(value);
107}
108
109function formatImpact(current, baseline, kind) {
110 if (baseline === undefined) return "—";
111 const delta = current - baseline;
112 const prefix = delta > 0 ? "+" : delta < 0 ? "−" : "";
113 const magnitude = formatValue(Math.abs(delta), kind);
114 const percent =
115 baseline === 0 ? "" : ` (${prefix}${Math.abs((delta / baseline) * 100).toFixed(1)}%)`;
116 return `${prefix}${magnitude}${percent}`;
117}
118
119function sameScenario(left, right) {
120 return SCENARIO_KEYS.every((key) => left[key] === right[key]);
121}
122
123const METRICS = [
124 { key: "totalWireBytes", label: "Total thread wire", kind: "bytes" },
125 { key: "threadSnapshotWireBytes", label: "Thread snapshot wire", kind: "bytes" },
126 {
127 key: "measuredTurnWebSocketWireBytes",
128 label: "Live turn WebSocket wire",
129 kind: "bytes",
130 },
131 {
132 key: "measuredTurnWebSocketDecodedBytes",
133 label: "Live turn WebSocket decoded",
134 kind: "bytes",
135 },
136 { key: "measuredTurnWebSocketMessages", label: "Live turn messages", kind: "messages" },
137];
138
139function renderComment(input) {
140 const current = input.current;
141 const baseline = input.baseline;
142 const comparable = baseline !== undefined && sameScenario(current.scenario, baseline.scenario);
143 const rows = [];
144 const ceilingChanges = [];
145 let failed = false;
146
147 for (const provider of PROVIDERS) {
148 for (const metric of METRICS) {
149 const observed = current.providers[provider].observed[metric.key];
150 const ceiling = current.providers[provider].ceiling[metric.key];
151 const baselineObserved = comparable
152 ? baseline.providers[provider].observed[metric.key]
153 : undefined;
154 const pass = observed <= ceiling;
155 failed ||= !pass;
156 rows.push(
157 `| ${provider === "codex" ? "Codex" : "Claude"} | ${metric.label} | ${baselineObserved === undefined ? "—" : formatValue(baselineObserved, metric.kind)} | ${formatValue(observed, metric.kind)} | ${formatImpact(observed, baselineObserved, metric.kind)} | ${formatValue(ceiling, metric.kind)} | ${pass ? "✅" : "❌"} |`,
158 );
159
160 if (baseline && baseline.providers[provider].ceiling[metric.key] !== ceiling) {
161 ceilingChanges.push(
162 `- ${provider === "codex" ? "Codex" : "Claude"} ${metric.label}: ${formatValue(baseline.providers[provider].ceiling[metric.key], metric.kind)} → ${formatValue(ceiling, metric.kind)}`,
163 );
164 }
165 }
166 }
167
168 const baselineLink = input.baselineRun
169 ? `[\`${input.baselineRun.sha.slice(0, 7)}\`](${input.baselineRun.url})`
170 : "unavailable";
171 const currentLink = `[\`${input.currentRun.sha.slice(0, 7)}\`](${input.currentRun.url})`;
172 const notices = [];
173 if (!baseline) {
174 notices.push(
175 "> ℹ️ No successful `main` baseline artifact is available yet. This run establishes the initial measurement.",
176 );
177 } else if (!comparable) {
178 notices.push(
179 "> ⚠️ The thread fixture changed, so impact percentages are not directly comparable to the `main` baseline.",
180 );
181 } else if (!input.baselineRun.matchesBase) {
182 notices.push(
183 "> ℹ️ The exact PR base did not have a successful artifact. Baseline uses the latest successful `main` measurement shown below.",
184 );
185 }
186 if (ceilingChanges.length > 0) {
187 notices.push(
188 `> ⚠️ **This PR changes transfer ceilings:**\n>\n${ceilingChanges.map((line) => `> ${line}`).join("\n")}`,
189 );
190 }
191
192 return [
193 COMMENT_MARKER,
194 resultShaMarker(input.currentRun.sha),
195 "## Thread transfer impact",
196 "",
197 failed
198 ? "❌ One or more thread transfer ceilings were exceeded."
199 : "✅ Thread transfer remains within every enforced ceiling.",
200 ...(notices.length > 0 ? ["", ...notices] : []),
201 "",
202 "| Provider | Metric | Main baseline | This PR | Impact | PR ceiling | |",
203 "| --- | --- | ---: | ---: | ---: | ---: | --- |",
204 ...rows,
205 "",
206 `Baseline: ${baselineLink} · PR result: ${currentLink} · Source CI: ${input.currentRun.conclusion}`,
207 "",
208 "<details>",
209 "<summary>Scenario and decoded snapshot size</summary>",
210 "",
211 `${current.scenario.historyTurns} historical turns, ${current.scenario.historyCommandToolsPerTurn} command tools per turn, ${formatBytes(current.scenario.historyMcpResultBytes)} retained MCP result per historical turn, and a ${formatBytes(current.scenario.measuredMcpResultBytes)} retained result in the measured turn.`,
212 "",
213 ...PROVIDERS.map(
214 (provider) =>
215 `- ${provider === "codex" ? "Codex" : "Claude"} decoded thread snapshot: ${formatBytes(current.providers[provider].observed.threadSnapshotDecodedBytes)}`,
216 ),
217 "",
218 "</details>",
219 "",
220 "_Updated in place by a trusted workflow. PR artifacts are strictly validated and never executed._",
221 ].join("\n");
222}
223
224async function artifactsForRun(github, owner, repo, runId) {
225 return github.paginate(github.rest.actions.listWorkflowRunArtifacts, {
226 owner,
227 repo,
228 run_id: runId,
229 per_page: 100,
230 });
231}
232
233function findResultArtifact(artifacts) {
234 return artifacts.find((artifact) => artifact.name === ARTIFACT_NAME && !artifact.expired);
235}
236
237async function resolve({ github, context, core }) {
238 const source = context.payload.workflow_run;
239 const { owner, repo } = context.repo;
240 if (source.event !== "pull_request") {
241 core.setOutput("publish", "false");
242 return;
243 }
244
245 let pullNumber = source.pull_requests?.[0]?.number;
246 if (!pullNumber) {
247 const associated = await github.paginate(
248 github.rest.repos.listPullRequestsAssociatedWithCommit,
249 { owner, repo, commit_sha: source.head_sha, per_page: 100 },
250 );
251 const matchingPulls = associated.filter(
252 (pull) =>
253 pull.state === "open" &&
254 pull.head.sha === source.head_sha &&
255 pull.head.ref === source.head_branch,
256 );
257 if (matchingPulls.length !== 1) {
258 core.info(
259 `Expected one open pull request for ${source.head_repository?.full_name ?? "unknown repository"}:${source.head_branch ?? "unknown branch"} at ${source.head_sha}; found ${matchingPulls.length}.`,
260 );
261 core.setOutput("publish", "false");
262 return;
263 }
264 pullNumber = matchingPulls[0].number;
265 }
266 if (!pullNumber) {
267 core.info("No open pull request is associated with the completed CI run.");
268 core.setOutput("publish", "false");
269 return;
270 }
271
272 const { data: pull } = await github.rest.pulls.get({ owner, repo, pull_number: pullNumber });
273 if (pull.head.sha !== source.head_sha) {
274 core.info(`Skipping stale CI result ${source.head_sha}; PR head is ${pull.head.sha}.`);
275 core.setOutput("publish", "false");
276 return;
277 }
278
279 const sourceArtifacts = await artifactsForRun(github, owner, repo, source.id);
280 const sourceArtifact = findResultArtifact(sourceArtifacts);
281 const workflowRuns = await github.paginate(github.rest.actions.listWorkflowRuns, {
282 owner,
283 repo,
284 workflow_id: source.workflow_id,
285 branch: pull.base.ref,
286 event: "push",
287 status: "success",
288 per_page: 100,
289 });
290 const orderedRuns = [
291 ...workflowRuns.filter((run) => run.head_sha === pull.base.sha),
292 ...workflowRuns.filter((run) => run.head_sha !== pull.base.sha),
293 ].slice(0, 20);
294
295 let baselineRun;
296 for (const run of orderedRuns) {
297 const artifacts = await artifactsForRun(github, owner, repo, run.id);
298 if (findResultArtifact(artifacts)) {
299 baselineRun = run;
300 break;
301 }
302 }
303
304 core.setOutput("publish", "true");
305 core.setOutput("pull_number", String(pullNumber));
306 core.setOutput("pr_artifact", sourceArtifact ? "true" : "false");
307 core.setOutput("pr_run_id", String(source.id));
308 core.setOutput("pr_sha", source.head_sha);
309 core.setOutput("pr_conclusion", source.conclusion ?? "unknown");
310 core.setOutput("baseline_artifact", baselineRun ? "true" : "false");
311 core.setOutput("baseline_run_id", baselineRun ? String(baselineRun.id) : "");
312 core.setOutput("baseline_sha", baselineRun?.head_sha ?? "");
313 core.setOutput(
314 "baseline_matches_base",
315 baselineRun?.head_sha === pull.base.sha ? "true" : "false",
316 );
317}
318
319async function upsertComment(github, context, pullNumber, body, options = {}) {
320 const { owner, repo } = context.repo;
321 const comments = await github.paginate(github.rest.issues.listComments, {
322 owner,
323 repo,
324 issue_number: pullNumber,
325 per_page: 100,
326 });
327 const existing = comments.find(
328 (comment) =>
329 comment.user?.login === "github-actions[bot]" && comment.body?.includes(COMMENT_MARKER),
330 );
331 if (
332 options.preserveResultSha &&
333 existing?.body?.includes(resultShaMarker(options.preserveResultSha))
334 ) {
335 return;
336 }
337 if (existing) {
338 await github.rest.issues.updateComment({ owner, repo, comment_id: existing.id, body });
339 } else {
340 await github.rest.issues.createComment({ owner, repo, issue_number: pullNumber, body });
341 }
342}
343
344async function upsertCommentForCurrentHead(
345 github,
346 context,
347 core,
348 pullNumber,
349 expectedSha,
350 body,
351 options,
352) {
353 const { owner, repo } = context.repo;
354 const { data: pull } = await github.rest.pulls.get({
355 owner,
356 repo,
357 pull_number: pullNumber,
358 });
359 if (pull.head.sha !== expectedSha) {
360 core.info(`Skipping stale CI result ${expectedSha}; PR head is ${pull.head.sha}.`);
361 return false;
362 }
363
364 await upsertComment(github, context, pullNumber, body, options);
365 return true;
366}
367
368async function publish({ github, context, core }) {
369 const pullNumber = Number(process.env.PR_NUMBER);
370 if (!Number.isSafeInteger(pullNumber) || pullNumber <= 0) {
371 throw new Error("PR_NUMBER is invalid");
372 }
373
374 const current = readResult(process.env.PR_RESULT_DIR);
375 const currentRun = {
376 sha: process.env.PR_SHA,
377 conclusion: process.env.PR_CONCLUSION,
378 url: `https://github.com/${context.repo.owner}/${context.repo.repo}/actions/runs/${process.env.PR_RUN_ID}`,
379 };
380 if (!current) {
381 await upsertCommentForCurrentHead(
382 github,
383 context,
384 core,
385 pullNumber,
386 currentRun.sha,
387 [
388 COMMENT_MARKER,
389 "## Thread transfer impact",
390 "",
391 `⚠️ The latest [CI run](${currentRun.url}) did not produce a thread transfer result for \`${currentRun.sha.slice(0, 7)}\`.`,
392 "",
393 "_This comment will update automatically after the next completed run._",
394 ].join("\n"),
395 { preserveResultSha: currentRun.sha },
396 );
397 return;
398 }
399
400 const baseline = readResult(process.env.BASELINE_RESULT_DIR);
401 const baselineRun = baseline
402 ? {
403 sha: process.env.BASELINE_SHA,
404 matchesBase: process.env.BASELINE_MATCHES_BASE === "true",
405 url: `https://github.com/${context.repo.owner}/${context.repo.repo}/actions/runs/${process.env.BASELINE_RUN_ID}`,
406 }
407 : undefined;
408 const body = renderComment({ current, baseline, currentRun, baselineRun });
409 const published = await upsertCommentForCurrentHead(
410 github,
411 context,
412 core,
413 pullNumber,
414 currentRun.sha,
415 body,
416 );
417 if (published) {
418 core.info(`Updated thread transfer report on PR #${pullNumber}.`);
419 }
420}
421
422module.exports = {
423 publish,
424 readResult,
425 renderComment,
426 resolve,
427 upsertCommentForCurrentHead,
428 validateResult,
429};
.github/scripts/thread-transfer-report.test.cjs created+292
...@@ -0,0 +1,292 @@
1const assert = require("node:assert/strict");
2const test = require("node:test");
3
4const {
5 renderComment,
6 resolve,
7 upsertCommentForCurrentHead,
8 validateResult,
9} = require("./thread-transfer-report.cjs");
10
11function result(overrides = {}) {
12 const observed = {
13 totalWireBytes: 2_200_000,
14 threadSnapshotWireBytes: 1_950_000,
15 threadSnapshotDecodedBytes: 9_100_000,
16 measuredTurnWebSocketWireBytes: 250_000,
17 measuredTurnWebSocketDecodedBytes: 1_150_000,
18 measuredTurnWebSocketMessages: 15,
19 };
20 const ceiling = {
21 totalWireBytes: 2_900_000,
22 threadSnapshotWireBytes: 2_600_000,
23 measuredTurnWebSocketWireBytes: 320_000,
24 measuredTurnWebSocketDecodedBytes: 1_550_000,
25 measuredTurnWebSocketMessages: 20,
26 };
27 return {
28 schemaVersion: 1,
29 scenario: {
30 id: "thread-transfer-v1",
31 historyTurns: 10,
32 historyCommandToolsPerTurn: 5,
33 historyMcpResultBytes: 900_000,
34 measuredCommandTools: 20,
35 measuredMcpResultBytes: 1_100_000,
36 },
37 providers: {
38 codex: { observed: { ...observed, ...overrides }, ceiling },
39 claudeAgent: { observed, ceiling },
40 },
41 };
42}
43
44test("validates the fixed artifact schema", () => {
45 assert.equal(validateResult(result()).schemaVersion, 1);
46 assert.throws(
47 () => validateResult({ ...result(), injectedMarkdown: "@everyone" }),
48 /unexpected fields/,
49 );
50 assert.throws(
51 () => validateResult(result({ totalWireBytes: "lots" })),
52 /non-negative safe integer/,
53 );
54});
55
56test("renders baseline, impact, ceiling, and ceiling changes", () => {
57 const baseline = result();
58 const current = result({ measuredTurnWebSocketWireBytes: 260_000 });
59 current.providers.codex.ceiling = {
60 ...current.providers.codex.ceiling,
61 measuredTurnWebSocketWireBytes: 330_000,
62 };
63 const comment = renderComment({
64 current,
65 baseline,
66 currentRun: {
67 sha: "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
68 conclusion: "success",
69 url: "https://github.com/pingdotgg/t3code/actions/runs/2",
70 },
71 baselineRun: {
72 sha: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
73 matchesBase: true,
74 url: "https://github.com/pingdotgg/t3code/actions/runs/1",
75 },
76 });
77
78 assert.match(comment, /Main baseline \| This PR \| Impact \| PR ceiling/);
79 assert.match(comment, /\+9\.8 KiB \(\+4\.0%\)/);
80 assert.match(comment, /This PR changes transfer ceilings/);
81 assert.match(comment, /312\.5 KiB → 322\.3 KiB/);
82 assert.match(comment, /<!-- t3-thread-transfer-report -->/);
83 assert.match(
84 comment,
85 /<!-- t3-thread-transfer-result-sha:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb -->/,
86 );
87});
88
89test("resolves a fallback PR with a redacted head repo and exact main baseline", async () => {
90 const outputs = {};
91 const listWorkflowRunArtifacts = () => {};
92 const listWorkflowRuns = () => {};
93 const listPullRequestsAssociatedWithCommit = () => {};
94 const github = {
95 paginate: async (method, input) => {
96 if (method === listPullRequestsAssociatedWithCommit) {
97 return [
98 {
99 number: 5350,
100 state: "open",
101 head: { sha: "head-sha", ref: "feature-branch", repo: null },
102 },
103 ];
104 }
105 if (method === listWorkflowRunArtifacts) {
106 return [
107 {
108 name: "thread-transfer-results",
109 expired: false,
110 runId: input.run_id,
111 },
112 ];
113 }
114 if (method === listWorkflowRuns) {
115 return [{ id: 1, head_sha: "base-sha" }];
116 }
117 throw new Error("unexpected pagination call");
118 },
119 rest: {
120 actions: { listWorkflowRunArtifacts, listWorkflowRuns },
121 pulls: {
122 get: async () => ({
123 data: {
124 head: { sha: "head-sha" },
125 base: { sha: "base-sha", ref: "main" },
126 },
127 }),
128 },
129 repos: { listPullRequestsAssociatedWithCommit },
130 },
131 };
132 await resolve({
133 github,
134 context: {
135 repo: { owner: "pingdotgg", repo: "t3code" },
136 payload: {
137 workflow_run: {
138 id: 2,
139 event: "pull_request",
140 workflow_id: 3,
141 head_sha: "head-sha",
142 head_branch: "feature-branch",
143 head_repository: { full_name: "pingdotgg/t3code" },
144 conclusion: "success",
145 pull_requests: [],
146 },
147 },
148 },
149 core: {
150 info: () => {},
151 setOutput: (key, value) => {
152 outputs[key] = value;
153 },
154 },
155 });
156
157 assert.equal(outputs.publish, "true");
158 assert.equal(outputs.pull_number, "5350");
159 assert.equal(outputs.pr_artifact, "true");
160 assert.equal(outputs.baseline_run_id, "1");
161 assert.equal(outputs.baseline_matches_base, "true");
162});
163
164test("does not guess when a fallback commit belongs to multiple PRs", async () => {
165 const outputs = {};
166 const listPullRequestsAssociatedWithCommit = () => {};
167 let fetchedPull = false;
168 await resolve({
169 github: {
170 paginate: async (method) => {
171 assert.equal(method, listPullRequestsAssociatedWithCommit);
172 return [5350, 5351].map((number) => ({
173 number,
174 state: "open",
175 head: {
176 sha: "head-sha",
177 ref: "feature-branch",
178 repo: { full_name: "pingdotgg/t3code" },
179 },
180 }));
181 },
182 rest: {
183 actions: {},
184 pulls: {
185 get: async () => {
186 fetchedPull = true;
187 },
188 },
189 repos: { listPullRequestsAssociatedWithCommit },
190 },
191 },
192 context: {
193 repo: { owner: "pingdotgg", repo: "t3code" },
194 payload: {
195 workflow_run: {
196 id: 2,
197 event: "pull_request",
198 workflow_id: 3,
199 head_sha: "head-sha",
200 head_branch: "feature-branch",
201 head_repository: { full_name: "pingdotgg/t3code" },
202 conclusion: "success",
203 pull_requests: [],
204 },
205 },
206 },
207 core: {
208 info: () => {},
209 setOutput: (key, value) => {
210 outputs[key] = value;
211 },
212 },
213 });
214
215 assert.equal(outputs.publish, "false");
216 assert.equal(fetchedPull, false);
217});
218
219test("does not publish a stale result after the PR head advances", async () => {
220 let listedComments = false;
221 const info = [];
222 const published = await upsertCommentForCurrentHead(
223 {
224 paginate: async () => {
225 listedComments = true;
226 return [];
227 },
228 rest: {
229 issues: {
230 listComments: () => {},
231 createComment: () => {
232 throw new Error("must not create a stale comment");
233 },
234 updateComment: () => {
235 throw new Error("must not update a stale comment");
236 },
237 },
238 pulls: {
239 get: async () => ({ data: { head: { sha: "new-head-sha" } } }),
240 },
241 },
242 },
243 { repo: { owner: "pingdotgg", repo: "t3code" } },
244 { info: (message) => info.push(message) },
245 5350,
246 "old-head-sha",
247 "stale body",
248 );
249
250 assert.equal(published, false);
251 assert.equal(listedComments, false);
252 assert.deepEqual(info, ["Skipping stale CI result old-head-sha; PR head is new-head-sha."]);
253});
254
255test("preserves a successful result when a same-SHA rerun has no artifact", async () => {
256 let updatedComment = false;
257 const sha = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
258 const published = await upsertCommentForCurrentHead(
259 {
260 paginate: async () => [
261 {
262 id: 1,
263 user: { login: "github-actions[bot]" },
264 body: `<!-- t3-thread-transfer-report -->\n<!-- t3-thread-transfer-result-sha:${sha} -->`,
265 },
266 ],
267 rest: {
268 issues: {
269 listComments: () => {},
270 createComment: () => {
271 updatedComment = true;
272 },
273 updateComment: () => {
274 updatedComment = true;
275 },
276 },
277 pulls: {
278 get: async () => ({ data: { head: { sha } } }),
279 },
280 },
281 },
282 { repo: { owner: "pingdotgg", repo: "t3code" } },
283 { info: () => {} },
284 5350,
285 sha,
286 "missing artifact warning",
287 { preserveResultSha: sha },
288 );
289
290 assert.equal(published, true);
291 assert.equal(updatedComment, false);
292});
.github/workflows/ci.yml+21
...@@ -84,8 +84,29 @@ jobs:...@@ -84,8 +84,29 @@ jobs:
84 run: vp run --filter @t3tools/desktop ensure:electron84 run: vp run --filter @t3tools/desktop ensure:electron
8585
86 - name: Test86 - name: Test
87 env:
88 T3CODE_TRANSFER_BUDGET_REPORT_PATH: ${{ runner.temp }}/t3code-transfer-budget.md
89 T3CODE_TRANSFER_BUDGET_RESULT_PATH: ${{ runner.temp }}/thread-transfer-result.json
87 run: vp run test90 run: vp run test
8891
92 - name: Publish transfer budget report
93 if: always()
94 run: |
95 if test -f "${{ runner.temp }}/t3code-transfer-budget.md"; then
96 tee -a "$GITHUB_STEP_SUMMARY" < "${{ runner.temp }}/t3code-transfer-budget.md"
97 else
98 echo "Transfer budget report was not produced." >> "$GITHUB_STEP_SUMMARY"
99 fi
100
101 - name: Upload thread transfer result
102 if: always()
103 uses: actions/upload-artifact@v7
104 with:
105 name: thread-transfer-results
106 path: ${{ runner.temp }}/thread-transfer-result.json
107 if-no-files-found: ignore
108 retention-days: 30
109
89 - name: Test resource monitor110 - name: Test resource monitor
90 run: cargo test --locked --manifest-path native/resource-monitor/Cargo.toml111 run: cargo test --locked --manifest-path native/resource-monitor/Cargo.toml
91112
.github/workflows/thread-transfer-report.yml created+75
...@@ -0,0 +1,75 @@
1name: Thread Transfer Report
2
3on:
4 workflow_run:
5 workflows: [CI]
6 types: [completed]
7
8permissions:
9 actions: read
10 contents: read
11 pull-requests: write
12
13jobs:
14 publish:
15 name: Publish PR comment
16 if: github.event.workflow_run.event == 'pull_request'
17 runs-on: ubuntu-24.04
18 concurrency:
19 group: thread-transfer-report-${{ github.event.workflow_run.pull_requests[0].number || github.event.workflow_run.id }}
20 cancel-in-progress: true
21 steps:
22 # workflow_run has a write-capable token even for fork PRs. Only load the
23 # publisher from the trusted default branch and never execute PR code.
24 - name: Checkout trusted publisher
25 uses: actions/checkout@v6
26 with:
27 ref: ${{ github.event.repository.default_branch }}
28 sparse-checkout: .github/scripts
29
30 - name: Test trusted publisher
31 run: node --test .github/scripts/thread-transfer-report.test.cjs
32
33 - id: resolve
34 name: Resolve PR and baseline artifacts
35 uses: actions/github-script@v8
36 with:
37 script: |
38 const reporter = require("./.github/scripts/thread-transfer-report.cjs");
39 await reporter.resolve({ github, context, core });
40
41 - name: Download PR result
42 if: steps.resolve.outputs.publish == 'true' && steps.resolve.outputs.pr_artifact == 'true'
43 uses: actions/download-artifact@v8
44 with:
45 name: thread-transfer-results
46 path: ${{ runner.temp }}/thread-transfer/pr
47 github-token: ${{ secrets.GITHUB_TOKEN }}
48 run-id: ${{ steps.resolve.outputs.pr_run_id }}
49
50 - name: Download main baseline
51 if: steps.resolve.outputs.publish == 'true' && steps.resolve.outputs.baseline_artifact == 'true'
52 uses: actions/download-artifact@v8
53 with:
54 name: thread-transfer-results
55 path: ${{ runner.temp }}/thread-transfer/main
56 github-token: ${{ secrets.GITHUB_TOKEN }}
57 run-id: ${{ steps.resolve.outputs.baseline_run_id }}
58
59 - name: Update thread transfer comment
60 if: steps.resolve.outputs.publish == 'true'
61 uses: actions/github-script@v8
62 env:
63 PR_NUMBER: ${{ steps.resolve.outputs.pull_number }}
64 PR_SHA: ${{ steps.resolve.outputs.pr_sha }}
65 PR_CONCLUSION: ${{ steps.resolve.outputs.pr_conclusion }}
66 PR_RUN_ID: ${{ steps.resolve.outputs.pr_run_id }}
67 PR_RESULT_DIR: ${{ runner.temp }}/thread-transfer/pr
68 BASELINE_SHA: ${{ steps.resolve.outputs.baseline_sha }}
69 BASELINE_MATCHES_BASE: ${{ steps.resolve.outputs.baseline_matches_base }}
70 BASELINE_RUN_ID: ${{ steps.resolve.outputs.baseline_run_id }}
71 BASELINE_RESULT_DIR: ${{ runner.temp }}/thread-transfer/main
72 with:
73 script: |
74 const reporter = require("./.github/scripts/thread-transfer-report.cjs");
75 await reporter.publish({ github, context, core });
apps/server/integration/NetworkTransferMeasurement.integration.ts created+177
...@@ -0,0 +1,177 @@
1// @effect-diagnostics nodeBuiltinImport:off - Measures the real Node HTTP and WebSocket transports.
2import * as NodeHttp from "node:http";
3import * as NodeZlib from "node:zlib";
4
5import * as NodeSocket from "@effect/platform-node/NodeSocket";
6import { WsRpcGroup } from "@t3tools/contracts";
7import * as Effect from "effect/Effect";
8import * as Layer from "effect/Layer";
9import * as Schema from "effect/Schema";
10import { RpcClient, RpcSerialization } from "effect/unstable/rpc";
11import * as Socket from "effect/unstable/socket/Socket";
12
13export class TransferHttpRequestError extends Schema.TaggedErrorClass<TransferHttpRequestError>()(
14 "TransferHttpRequestError",
15 {
16 url: Schema.String,
17 cause: Schema.Defect(),
18 },
19) {}
20
21export interface HttpTransferMeasurement {
22 readonly status: number;
23 readonly contentEncoding: string | null;
24 readonly encodedBody: Uint8Array;
25 readonly encodedBodyBytes: number;
26 readonly decodedBody: Uint8Array;
27 readonly decodedBodyBytes: number;
28 /** HTTP response bytes read from the socket, including status line and headers. */
29 readonly wireBytes: number;
30}
31
32export const measureHttpGet = Effect.fn("TransferBudget.measureHttpGet")(function* (input: {
33 readonly url: string;
34 readonly headers?: Readonly<Record<string, string>>;
35}) {
36 return yield* Effect.tryPromise({
37 try: () =>
38 new Promise<HttpTransferMeasurement>((resolve, reject) => {
39 let socketBytesBeforeResponse = 0;
40 const request = NodeHttp.get(
41 input.url,
42 {
43 agent: false,
44 headers: {
45 "accept-encoding": "gzip",
46 connection: "close",
47 ...input.headers,
48 },
49 },
50 (response) => {
51 const chunks: Buffer[] = [];
52 response.on("data", (chunk: Buffer) => chunks.push(chunk));
53 response.once("error", reject);
54 response.once("end", () => {
55 try {
56 const encodedBody = Buffer.concat(chunks);
57 const header = response.headers["content-encoding"];
58 const contentEncoding = Array.isArray(header)
59 ? (header[0] ?? null)
60 : (header ?? null);
61 const decodedBody =
62 contentEncoding === "gzip" ? NodeZlib.gunzipSync(encodedBody) : encodedBody;
63 resolve({
64 status: response.statusCode ?? 0,
65 contentEncoding,
66 encodedBody,
67 encodedBodyBytes: encodedBody.byteLength,
68 decodedBody,
69 decodedBodyBytes: decodedBody.byteLength,
70 wireBytes: Math.max(0, response.socket.bytesRead - socketBytesBeforeResponse),
71 });
72 } catch (cause) {
73 reject(cause);
74 }
75 });
76 },
77 );
78 request.once("socket", (socket) => {
79 socketBytesBeforeResponse = socket.bytesRead;
80 });
81 request.once("error", reject);
82 request.setTimeout(10_000, () => {
83 request.destroy(new Error(`Timed out reading ${input.url}`));
84 });
85 }),
86 catch: (cause) => new TransferHttpRequestError({ url: input.url, cause }),
87 });
88});
89
90export interface WebSocketTransferTotals {
91 readonly wireBytes: number;
92 readonly decodedBytes: number;
93 readonly messages: number;
94}
95
96export interface WebSocketTransferRecorder {
97 readonly connect: (
98 url: string,
99 protocols: string | string[] | undefined,
100 cookie: string,
101 ) => globalThis.WebSocket;
102 readonly totals: () => WebSocketTransferTotals;
103 readonly negotiatedExtensions: () => string;
104}
105
106interface NodeWebSocketWithTransport extends NodeSocket.NodeWS.WebSocket {
107 readonly _socket?: {
108 readonly bytesRead: number;
109 };
110}
111
112function rawDataBytes(data: NodeSocket.NodeWS.RawData): number {
113 if (Array.isArray(data)) {
114 return data.reduce((total, chunk) => total + chunk.byteLength, 0);
115 }
116 return data.byteLength;
117}
118
119export function makeWebSocketTransferRecorder(): WebSocketTransferRecorder {
120 let socket: NodeWebSocketWithTransport | null = null;
121 let decodedBytes = 0;
122 let messages = 0;
123
124 return {
125 connect: (url, protocols, cookie) => {
126 const nextSocket = new NodeSocket.NodeWS.WebSocket(url, protocols, {
127 headers: { cookie },
128 perMessageDeflate: true,
129 }) as NodeWebSocketWithTransport;
130 socket = nextSocket;
131 nextSocket.on("message", (data) => {
132 const bytes = rawDataBytes(data);
133 decodedBytes += bytes;
134 messages += 1;
135 });
136 return nextSocket as unknown as globalThis.WebSocket;
137 },
138 totals: () => ({
139 wireBytes: socket?._socket?.bytesRead ?? 0,
140 decodedBytes,
141 messages,
142 }),
143 negotiatedExtensions: () => socket?.extensions ?? "",
144 };
145}
146
147export function transferDelta(
148 start: WebSocketTransferTotals,
149 end: WebSocketTransferTotals,
150): WebSocketTransferTotals {
151 return {
152 wireBytes: Math.max(0, end.wireBytes - start.wireBytes),
153 decodedBytes: Math.max(0, end.decodedBytes - start.decodedBytes),
154 messages: Math.max(0, end.messages - start.messages),
155 };
156}
157
158export function countingWsRpcProtocolLayer(input: {
159 readonly url: string;
160 readonly cookie: string;
161 readonly recorder: WebSocketTransferRecorder;
162}) {
163 const webSocketConstructorLayer = Layer.succeed(Socket.WebSocketConstructor, (url, protocols) =>
164 input.recorder.connect(url, protocols, input.cookie),
165 );
166 return RpcClient.layerProtocolSocket().pipe(
167 Layer.provide(
168 Socket.layerWebSocket(input.url, { openTimeout: "10 seconds" }).pipe(
169 Layer.provide(webSocketConstructorLayer),
170 ),
171 ),
172 Layer.provide(RpcSerialization.layerJson),
173 );
174}
175
176export const makeCountingWsRpcClient = RpcClient.make(WsRpcGroup);
177export type CountingWsRpcClient = Effect.Success<typeof makeCountingWsRpcClient>;
apps/server/integration/OrchestrationEngineHarness.integration.ts+13
...@@ -55,6 +55,8 @@ import { RuntimeReceiptBusTest } from "../src/orchestration/Layers/RuntimeReceip...@@ -55,6 +55,8 @@ import { RuntimeReceiptBusTest } from "../src/orchestration/Layers/RuntimeReceip
55import { OrchestrationReactorLive } from "../src/orchestration/Layers/OrchestrationReactor.ts";55import { OrchestrationReactorLive } from "../src/orchestration/Layers/OrchestrationReactor.ts";
56import { ProviderCommandReactorLive } from "../src/orchestration/Layers/ProviderCommandReactor.ts";56import { ProviderCommandReactorLive } from "../src/orchestration/Layers/ProviderCommandReactor.ts";
57import { ProviderRuntimeIngestionLive } from "../src/orchestration/Layers/ProviderRuntimeIngestion.ts";57import { ProviderRuntimeIngestionLive } from "../src/orchestration/Layers/ProviderRuntimeIngestion.ts";
58import { CheckpointReactor } from "../src/orchestration/Services/CheckpointReactor.ts";
59import { ProviderRuntimeIngestionService } from "../src/orchestration/Services/ProviderRuntimeIngestion.ts";
58import {60import {
59 OrchestrationEngineService,61 OrchestrationEngineService,
60 type OrchestrationEngineShape,62 type OrchestrationEngineShape,
...@@ -218,6 +220,8 @@ export interface OrchestrationIntegrationHarness {...@@ -218,6 +220,8 @@ export interface OrchestrationIntegrationHarness {
218 timeoutMs?: number,220 timeoutMs?: number,
219 ): Effect.Effect<Receipt, never>;221 ): Effect.Effect<Receipt, never>;
220 };222 };
223 readonly drainProviderRuntime: Effect.Effect<void>;
224 readonly drainCheckpointReactor: Effect.Effect<void>;
221 readonly dispose: Effect.Effect<void, never>;225 readonly dispose: Effect.Effect<void, never>;
222}226}
223227
...@@ -392,6 +396,13 @@ export const makeOrchestrationIntegrationHarness = (...@@ -392,6 +396,13 @@ export const makeOrchestrationIntegrationHarness = (
392 const reactor = yield* tryRuntimePromise("load OrchestrationReactor service", () =>396 const reactor = yield* tryRuntimePromise("load OrchestrationReactor service", () =>
393 runtime.runPromise(Effect.service(OrchestrationReactor)),397 runtime.runPromise(Effect.service(OrchestrationReactor)),
394 ).pipe(Effect.orDie);398 ).pipe(Effect.orDie);
399 const providerRuntimeIngestion = yield* tryRuntimePromise(
400 "load ProviderRuntimeIngestion service",
401 () => runtime.runPromise(Effect.service(ProviderRuntimeIngestionService)),
402 ).pipe(Effect.orDie);
403 const checkpointReactor = yield* tryRuntimePromise("load CheckpointReactor service", () =>
404 runtime.runPromise(Effect.service(CheckpointReactor)),
405 ).pipe(Effect.orDie);
395 const snapshotQuery = yield* tryRuntimePromise("load ProjectionSnapshotQuery service", () =>406 const snapshotQuery = yield* tryRuntimePromise("load ProjectionSnapshotQuery service", () =>
396 runtime.runPromise(Effect.service(ProjectionSnapshotQuery)),407 runtime.runPromise(Effect.service(ProjectionSnapshotQuery)),
397 ).pipe(Effect.orDie);408 ).pipe(Effect.orDie);
...@@ -556,6 +567,8 @@ export const makeOrchestrationIntegrationHarness = (...@@ -556,6 +567,8 @@ export const makeOrchestrationIntegrationHarness = (
556 waitForDomainEvent,567 waitForDomainEvent,
557 waitForPendingApproval,568 waitForPendingApproval,
558 waitForReceipt,569 waitForReceipt,
570 drainProviderRuntime: providerRuntimeIngestion.drain,
571 drainCheckpointReactor: checkpointReactor.drain,
559 dispose,572 dispose,
560 } satisfies OrchestrationIntegrationHarness;573 } satisfies OrchestrationIntegrationHarness;
561 });574 });
apps/server/integration/TestProviderAdapter.integration.ts+7-16
...@@ -11,7 +11,6 @@ import {...@@ -11,7 +11,6 @@ import {
11 ProviderDriverKind,11 ProviderDriverKind,
12} from "@t3tools/contracts";12} from "@t3tools/contracts";
13import * as Effect from "effect/Effect";13import * as Effect from "effect/Effect";
14import * as Crypto from "effect/Crypto";
15import * as Queue from "effect/Queue";14import * as Queue from "effect/Queue";
16import * as Stream from "effect/Stream";15import * as Stream from "effect/Stream";
1716
...@@ -226,9 +225,9 @@ function missingSessionEffect(...@@ -226,9 +225,9 @@ function missingSessionEffect(
226export const makeTestProviderAdapterHarness = (options?: MakeTestProviderAdapterHarnessOptions) =>225export const makeTestProviderAdapterHarness = (options?: MakeTestProviderAdapterHarnessOptions) =>
227 Effect.gen(function* () {226 Effect.gen(function* () {
228 const provider = options?.provider ?? ProviderDriverKind.make("codex");227 const provider = options?.provider ?? ProviderDriverKind.make("codex");
229 const crypto = yield* Crypto.Crypto;
230 const runtimeEvents = yield* Queue.unbounded<ProviderRuntimeEvent>();228 const runtimeEvents = yield* Queue.unbounded<ProviderRuntimeEvent>();
231 let sessionCount = 0;229 let sessionCount = 0;
230 let eventCount = 0;
232 const sessions = new Map<ThreadId, SessionState>();231 const sessions = new Map<ThreadId, SessionState>();
233 const queuedResponsesForNextSession: TestTurnResponse[] = [];232 const queuedResponsesForNextSession: TestTurnResponse[] = [];
234 const interruptCallsBySession = new Map<ThreadId, Array<TurnId | undefined>>();233 const interruptCallsBySession = new Map<ThreadId, Array<TurnId | undefined>>();
...@@ -242,18 +241,10 @@ export const makeTestProviderAdapterHarness = (options?: MakeTestProviderAdapter...@@ -242,18 +241,10 @@ export const makeTestProviderAdapterHarness = (options?: MakeTestProviderAdapter
242 >();241 >();
243242
244 const emit = (event: ProviderRuntimeEvent) => Queue.offer(runtimeEvents, event);243 const emit = (event: ProviderRuntimeEvent) => Queue.offer(runtimeEvents, event);
245 const randomUUIDv4 = (threadId: ThreadId) =>244 const nextEventId = (threadId: ThreadId) => {
246 crypto.randomUUIDv4.pipe(245 eventCount += 1;
247 Effect.mapError(246 return EventId.make(`test-provider:${provider}:${threadId}:${eventCount}`);
248 (cause) =>247 };
249 new ProviderAdapterValidationError({
250 provider,
251 operation: "crypto/randomUUIDv4",
252 issue: `Failed to generate test runtime identifier for thread '${threadId}'.`,
253 cause,
254 }),
255 ),
256 );
257248
258 const startSession: ProviderAdapterShape<ProviderAdapterError>["startSession"] = (input) =>249 const startSession: ProviderAdapterShape<ProviderAdapterError>["startSession"] = (input) =>
259 Effect.gen(function* () {250 Effect.gen(function* () {
...@@ -322,7 +313,7 @@ export const makeTestProviderAdapterHarness = (options?: MakeTestProviderAdapter...@@ -322,7 +313,7 @@ export const makeTestProviderAdapterHarness = (options?: MakeTestProviderAdapter
322 for (const fixtureEvent of response.events) {313 for (const fixtureEvent of response.events) {
323 const rawEvent: Record<string, unknown> = {314 const rawEvent: Record<string, unknown> = {
324 ...(fixtureEvent as Record<string, unknown>),315 ...(fixtureEvent as Record<string, unknown>),
325 eventId: yield* randomUUIDv4(input.threadId),316 eventId: nextEventId(input.threadId),
326 provider,317 provider,
327 sessionId: RuntimeSessionId.make(String(input.threadId)),318 sessionId: RuntimeSessionId.make(String(input.threadId)),
328 };319 };
...@@ -379,7 +370,7 @@ export const makeTestProviderAdapterHarness = (options?: MakeTestProviderAdapter...@@ -379,7 +370,7 @@ export const makeTestProviderAdapterHarness = (options?: MakeTestProviderAdapter
379 if (deferredTurnCompletedEvents.length === 0) {370 if (deferredTurnCompletedEvents.length === 0) {
380 yield* emit({371 yield* emit({
381 type: "turn.completed",372 type: "turn.completed",
382 eventId: EventId.make(yield* randomUUIDv4(input.threadId)),373 eventId: nextEventId(input.threadId),
383 provider,374 provider,
384 createdAt: nowIso(),375 createdAt: nowIso(),
385 threadId: state.snapshot.threadId,376 threadId: state.snapshot.threadId,
apps/server/integration/TransferBudgetReport.integration.ts created+212
...@@ -0,0 +1,212 @@
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
15export interface TransferBudgetRun {
16 readonly provider: ProviderDriverKind;
17 readonly threadSnapshot: HttpTransferMeasurement;
18 readonly measuredTurnWebSocket: WebSocketTransferTotals;
19}
20
21interface ProviderTransferBudget {
22 readonly totalWireBytes: number;
23 readonly threadSnapshotWireBytes: number;
24 readonly measuredTurnWebSocketWireBytes: number;
25 readonly measuredTurnWebSocketDecodedBytes: number;
26 readonly measuredTurnWebSocketMessages: number;
27}
28
29// These caps leave roughly 30% headroom above the client projection of the
30// deterministic 9 MB retained-result fixture. Full MCP results stay in
31// persistence, so accidentally shipping them again exceeds these caps by
32// orders of magnitude. The CI report preserves exact values for review.
33const TRANSFER_BUDGET = {
34 totalWireBytes: 15_500,
35 threadSnapshotWireBytes: 7_500,
36 measuredTurnWebSocketWireBytes: 8_000,
37 measuredTurnWebSocketDecodedBytes: 68_000,
38 measuredTurnWebSocketMessages: 21,
39} satisfies ProviderTransferBudget;
40
41export const TRANSFER_BUDGETS: Readonly<Record<string, ProviderTransferBudget>> = {
42 codex: TRANSFER_BUDGET,
43 claudeAgent: TRANSFER_BUDGET,
44};
45
46function totalWireBytes(run: TransferBudgetRun): number {
47 return run.threadSnapshot.wireBytes + run.measuredTurnWebSocket.wireBytes;
48}
49
50function observedTransfer(run: TransferBudgetRun) {
51 return {
52 totalWireBytes: totalWireBytes(run),
53 threadSnapshotWireBytes: run.threadSnapshot.wireBytes,
54 threadSnapshotDecodedBytes: run.threadSnapshot.decodedBodyBytes,
55 measuredTurnWebSocketWireBytes: run.measuredTurnWebSocket.wireBytes,
56 measuredTurnWebSocketDecodedBytes: run.measuredTurnWebSocket.decodedBytes,
57 measuredTurnWebSocketMessages: run.measuredTurnWebSocket.messages,
58 };
59}
60
61/** Machine-readable input for the trusted PR comment publisher. */
62export function formatTransferBudgetResult(runs: ReadonlyArray<TransferBudgetRun>): string {
63 const providers = Object.fromEntries(
64 runs.flatMap((run) => {
65 const ceiling = TRANSFER_BUDGETS[run.provider];
66 return ceiling ? [[run.provider, { observed: observedTransfer(run), ceiling }]] : [];
67 }),
68 );
69
70 return `${JSON.stringify(
71 {
72 schemaVersion: 1,
73 scenario: {
74 id: "thread-transfer-v1",
75 historyTurns: TRANSFER_HISTORY_TURN_COUNT,
76 historyCommandToolsPerTurn: TRANSFER_HISTORY_TOOLS_PER_TURN,
77 historyMcpResultBytes: TRANSFER_HISTORY_MCP_RESULT_BYTES,
78 measuredCommandTools: TRANSFER_MEASURED_TOOLS,
79 measuredMcpResultBytes: TRANSFER_MEASURED_MCP_RESULT_BYTES,
80 },
81 providers,
82 },
83 null,
84 2,
85 )}\n`;
86}
87
88function formatBytes(bytes: number): string {
89 if (bytes < 1_024) return `${bytes} B`;
90 if (bytes >= 1_024 * 1_024) {
91 return `${(bytes / 1_024 / 1_024).toFixed(2)} MiB (${bytes.toLocaleString("en-US")} B)`;
92 }
93 return `${(bytes / 1_024).toFixed(1)} KiB (${bytes.toLocaleString("en-US")} B)`;
94}
95
96function row(
97 provider: ProviderDriverKind,
98 phase: string,
99 metric: string,
100 observed: number,
101 maximum: number,
102 format: (value: number) => string = formatBytes,
103): string {
104 const status = observed <= maximum ? "PASS" : "FAIL";
105 return `| ${provider} | ${phase} | ${metric} | ${format(observed)} | ${format(maximum)} | ${status} |`;
106}
107
108export function transferBudgetViolations(runs: ReadonlyArray<TransferBudgetRun>): string[] {
109 const violations: string[] = [];
110 for (const run of runs) {
111 const budget = TRANSFER_BUDGETS[run.provider];
112 if (!budget) {
113 violations.push(`${run.provider}: no transfer budget is configured`);
114 continue;
115 }
116 const checks = [
117 ["total thread wire bytes", totalWireBytes(run), budget.totalWireBytes],
118 ["thread snapshot wire bytes", run.threadSnapshot.wireBytes, budget.threadSnapshotWireBytes],
119 [
120 "measured-turn WebSocket wire bytes",
121 run.measuredTurnWebSocket.wireBytes,
122 budget.measuredTurnWebSocketWireBytes,
123 ],
124 [
125 "measured-turn WebSocket decoded bytes",
126 run.measuredTurnWebSocket.decodedBytes,
127 budget.measuredTurnWebSocketDecodedBytes,
128 ],
129 [
130 "measured-turn WebSocket messages",
131 run.measuredTurnWebSocket.messages,
132 budget.measuredTurnWebSocketMessages,
133 ],
134 ] as const;
135 for (const [metric, observed, maximum] of checks) {
136 if (observed > maximum) {
137 violations.push(`${run.provider}: ${metric} was ${observed}, maximum ${maximum}`);
138 }
139 }
140 }
141 return violations;
142}
143
144export function formatTransferBudgetReport(runs: ReadonlyArray<TransferBudgetRun>): string {
145 const lines = [
146 "# T3 Code thread transfer budget",
147 "",
148 "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.",
149 `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. Payload sizes are calibrated from heavy local Codex and Claude histories and contain no user data.`,
150 "",
151 "| Provider | Total thread wire | Budget | Result |",
152 "| --- | ---: | ---: | --- |",
153 ...runs.flatMap((run) => {
154 const budget = TRANSFER_BUDGETS[run.provider];
155 if (!budget) return [];
156 const observed = observedTransfer(run).totalWireBytes;
157 return [
158 `| ${run.provider} | ${formatBytes(observed)} | ${formatBytes(budget.totalWireBytes)} | ${observed <= budget.totalWireBytes ? "PASS" : "FAIL"} |`,
159 ];
160 }),
161 "",
162 "## Detailed measurements",
163 "",
164 "| Provider | Phase | Metric | Observed | Budget | Result |",
165 "| --- | --- | --- | ---: | ---: | --- |",
166 ];
167
168 for (const run of runs) {
169 const budget = TRANSFER_BUDGETS[run.provider];
170 if (!budget) continue;
171 lines.push(
172 row(
173 run.provider,
174 "thread snapshot",
175 "HTTP wire",
176 run.threadSnapshot.wireBytes,
177 budget.threadSnapshotWireBytes,
178 ),
179 row(
180 run.provider,
181 "measured turn",
182 "WebSocket wire",
183 run.measuredTurnWebSocket.wireBytes,
184 budget.measuredTurnWebSocketWireBytes,
185 ),
186 row(
187 run.provider,
188 "measured turn",
189 "WebSocket decoded",
190 run.measuredTurnWebSocket.decodedBytes,
191 budget.measuredTurnWebSocketDecodedBytes,
192 ),
193 row(
194 run.provider,
195 "measured turn",
196 "WebSocket messages",
197 run.measuredTurnWebSocket.messages,
198 budget.measuredTurnWebSocketMessages,
199 String,
200 ),
201 );
202 }
203
204 lines.push("", "## Compression diagnostics", "");
205 for (const run of runs) {
206 lines.push(
207 `- ${run.provider}: thread snapshot ${formatBytes(run.threadSnapshot.decodedBodyBytes)} decoded to ${formatBytes(run.threadSnapshot.encodedBodyBytes)} gzip.`,
208 );
209 }
210
211 return `${lines.join("\n")}\n`;
212}
apps/server/integration/TransferBudgetScenario.integration.ts created+128
...@@ -0,0 +1,128 @@
1import {
2 CommandId,
3 defaultInstanceIdForDriver,
4 DEFAULT_MODEL,
5 DEFAULT_MODEL_BY_PROVIDER,
6 DEFAULT_PROVIDER_INTERACTION_MODE,
7 MessageId,
8 ProjectId,
9 ProviderDriverKind,
10 ThreadId,
11} from "@t3tools/contracts";
12import * as Effect from "effect/Effect";
13
14import type { TurnProcessingQuiescedReceipt } from "../src/orchestration/Services/RuntimeReceiptBus.ts";
15import type { OrchestrationIntegrationHarness } from "./OrchestrationEngineHarness.integration.ts";
16import {
17 expectedRecordedAssistantText,
18 makeRecordedTransferTurn,
19 TRANSFER_HISTORY_TURN_COUNT,
20} from "./fixtures/transferBudget.ts";
21
22export const TRANSFER_PROJECT_ID = ProjectId.make("transfer-budget-project");
23export const TRANSFER_THREAD_ID = ThreadId.make("transfer-budget-thread");
24export const TRANSFER_MEASURED_TURN_INDEX = TRANSFER_HISTORY_TURN_COUNT;
25
26export function transferModelSelection(provider: ProviderDriverKind) {
27 return {
28 instanceId: defaultInstanceIdForDriver(provider),
29 model: DEFAULT_MODEL_BY_PROVIDER[provider] ?? DEFAULT_MODEL,
30 };
31}
32
33function turnTimestamp(turnIndex: number): string {
34 return `2026-06-01T00:${String(turnIndex).padStart(2, "0")}:00.000Z`;
35}
36
37export const TRANSFER_MEASURED_TURN_CREATED_AT = turnTimestamp(TRANSFER_MEASURED_TURN_INDEX);
38
39const waitForTurnQuiesced = Effect.fn("TransferBudget.waitForTurnQuiesced")(function* (
40 harness: OrchestrationIntegrationHarness,
41 checkpointTurnCount: number,
42) {
43 const receipt = yield* harness.waitForReceipt(
44 (receipt): receipt is TurnProcessingQuiescedReceipt =>
45 receipt.type === "turn.processing.quiesced" &&
46 receipt.threadId === TRANSFER_THREAD_ID &&
47 receipt.checkpointTurnCount === checkpointTurnCount,
48 );
49 yield* harness.drainProviderRuntime;
50 yield* harness.drainCheckpointReactor;
51 return receipt;
52});
53
54export const seedTransferBudgetHistory = Effect.fn("TransferBudget.seedHistory")(function* (
55 harness: OrchestrationIntegrationHarness,
56 provider: ProviderDriverKind,
57) {
58 if (!harness.adapterHarness) {
59 return yield* Effect.die(new Error("Transfer budget history requires the replay adapter."));
60 }
61
62 const modelSelection = transferModelSelection(provider);
63 yield* harness.engine.dispatch({
64 type: "project.create",
65 commandId: CommandId.make(`transfer:${provider}:project-create`),
66 projectId: TRANSFER_PROJECT_ID,
67 title: "Transfer Budget Project",
68 workspaceRoot: harness.workspaceDir,
69 defaultModelSelection: modelSelection,
70 createdAt: turnTimestamp(0),
71 });
72 yield* harness.engine.dispatch({
73 type: "thread.create",
74 commandId: CommandId.make(`transfer:${provider}:thread-create`),
75 threadId: TRANSFER_THREAD_ID,
76 projectId: TRANSFER_PROJECT_ID,
77 title: `${provider} transfer history`,
78 modelSelection,
79 runtimeMode: "approval-required",
80 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
81 branch: "main",
82 worktreePath: harness.workspaceDir,
83 createdAt: turnTimestamp(0),
84 });
85
86 for (let turnIndex = 0; turnIndex < TRANSFER_HISTORY_TURN_COUNT; turnIndex += 1) {
87 const response = makeRecordedTransferTurn(provider, turnIndex);
88 if (turnIndex === 0) {
89 yield* harness.adapterHarness.queueTurnResponseForNextSession(response);
90 } else {
91 yield* harness.adapterHarness.queueTurnResponse(TRANSFER_THREAD_ID, response);
92 }
93
94 yield* harness.engine.dispatch({
95 type: "thread.turn.start",
96 commandId: CommandId.make(`transfer:${provider}:turn:${turnIndex + 1}`),
97 threadId: TRANSFER_THREAD_ID,
98 message: {
99 messageId: MessageId.make(`transfer-user-${turnIndex + 1}`),
100 role: "user",
101 text: `Inspect transfer behavior for historical turn ${turnIndex + 1}.`,
102 attachments: [],
103 },
104 modelSelection,
105 runtimeMode: "approval-required",
106 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
107 createdAt: turnTimestamp(turnIndex),
108 });
109 yield* waitForTurnQuiesced(harness, turnIndex + 1);
110 }
111});
112
113export const queueMeasuredTransferTurn = Effect.fn("TransferBudget.queueMeasuredTurn")(function* (
114 harness: OrchestrationIntegrationHarness,
115 provider: ProviderDriverKind,
116) {
117 if (!harness.adapterHarness) {
118 return yield* Effect.die(new Error("Transfer budget measurement requires the replay adapter."));
119 }
120 const response = makeRecordedTransferTurn(provider, TRANSFER_MEASURED_TURN_INDEX);
121 yield* harness.adapterHarness.queueTurnResponse(TRANSFER_THREAD_ID, response);
122});
123
124export function expectedMeasuredAssistantText(provider: ProviderDriverKind): string {
125 return expectedRecordedAssistantText(provider, TRANSFER_MEASURED_TURN_INDEX);
126}
127
128export { TRANSFER_HISTORY_TURN_COUNT, waitForTurnQuiesced };
apps/server/integration/fixtures/transferBudget.ts created+372
...@@ -0,0 +1,372 @@
1import { EventId, ProviderDriverKind } from "@t3tools/contracts";
2
3import type {
4 FixtureProviderRuntimeEvent,
5 TestTurnResponse,
6} from "../TestProviderAdapter.integration.ts";
7
8const FIXTURE_THREAD_ID = "transfer-budget-thread";
9const FIXTURE_TURN_ID = "transfer-budget-turn";
10
11export const TRANSFER_HISTORY_TURN_COUNT = 10;
12export const TRANSFER_HISTORY_TOOLS_PER_TURN = 5;
13export const TRANSFER_MEASURED_TOOLS = 20;
14export const TRANSFER_HISTORY_MCP_RESULT_BYTES = 900_000;
15export const TRANSFER_MEASURED_MCP_RESULT_BYTES = 1_100_000;
16
17const sourceModules = [
18 "connection/session.ts",
19 "connection/supervisor.ts",
20 "rpc/client.ts",
21 "rpc/protocol.ts",
22 "state/threads.ts",
23 "state/threadReducer.ts",
24 "state/threadSnapshotHttp.ts",
25 "orchestration/http.ts",
26 "orchestration/Normalizer.ts",
27 "orchestration/ActivityPayloadProjection.ts",
28 "provider/ProviderService.ts",
29 "provider/ProviderRuntimeIngestion.ts",
30 "persistence/ProjectionSnapshotQuery.ts",
31 "persistence/OrchestrationEventStore.ts",
32 "checkpointing/CheckpointStore.ts",
33 "checkpointing/CheckpointDiffQuery.ts",
34 "server.ts",
35] as const;
36
37function fixtureTimestamp(turnIndex: number, eventIndex: number): string {
38 const minute = String(turnIndex).padStart(2, "0");
39 const second = String(Math.floor(eventIndex / 1_000)).padStart(2, "0");
40 const millisecond = String(eventIndex % 1_000).padStart(3, "0");
41 return `2026-06-01T00:${minute}:${second}.${millisecond}Z`;
42}
43
44function mix(value: number): number {
45 let mixed = value | 0;
46 mixed ^= mixed >>> 16;
47 mixed = Math.imul(mixed, 0x7feb352d);
48 mixed ^= mixed >>> 15;
49 mixed = Math.imul(mixed, 0x846ca68b);
50 mixed ^= mixed >>> 16;
51 return mixed >>> 0;
52}
53
54function digest(seed: number): string {
55 return [0, 1, 2, 3]
56 .map((offset) =>
57 mix(seed + offset * 0x9e3779b9)
58 .toString(16)
59 .padStart(8, "0"),
60 )
61 .join("");
62}
63
64/** Produces safe, deterministic output with enough entropy to exercise gzip. */
65function diagnosticOutput(input: {
66 readonly provider: ProviderDriverKind;
67 readonly turnIndex: number;
68 readonly toolIndex: number;
69 readonly targetBytes: number;
70}): string {
71 const chunks: string[] = [];
72 const providerSeed = input.provider === "codex" ? 0x43_4f_44_45 : 0x43_4c_41_55;
73 let length = 0;
74 let lineIndex = 0;
75
76 while (length < input.targetBytes) {
77 const modulePath = sourceModules[(input.toolIndex + lineIndex) % sourceModules.length];
78 const seed =
79 providerSeed + input.turnIndex * 100_003 + input.toolIndex * 10_007 + lineIndex * 101;
80 const line =
81 `${String(lineIndex + 1).padStart(6, "0")} ${modulePath} ` +
82 `operation=project-transfer-${input.turnIndex + 1}-${input.toolIndex + 1} ` +
83 `cursor=${mix(seed)} digest=${digest(seed)} status=completed\n`;
84 chunks.push(line);
85 length += line.length;
86 lineIndex += 1;
87 }
88
89 return chunks.join("").slice(0, input.targetBytes);
90}
91
92function assistantChunks(provider: ProviderDriverKind, turnIndex: number): ReadonlyArray<string> {
93 const providerName = provider === "codex" ? "Codex" : "Claude";
94 const paragraphs: string[] = [
95 `I traced the ${providerName} request through the environment connection and orchestration layers. `,
96 ];
97 let paragraphIndex = 0;
98 while (paragraphs.join("").length < 4_096) {
99 const modulePath = sourceModules[paragraphIndex % sourceModules.length];
100 paragraphs.push(
101 `Pass ${paragraphIndex + 1} reviewed ${modulePath} for turn ${turnIndex + 1}. ` +
102 "The shell cursor stayed monotonic, the thread snapshot remained resumable, and the client received only incremental events. ",
103 );
104 paragraphIndex += 1;
105 }
106 const text = paragraphs.join("").slice(0, 4_096);
107 return Array.from({ length: Math.ceil(text.length / 256) }, (_, index) =>
108 text.slice(index * 256, (index + 1) * 256),
109 );
110}
111
112export function expectedRecordedAssistantText(
113 provider: ProviderDriverKind,
114 turnIndex: number,
115): string {
116 return assistantChunks(provider, turnIndex).join("");
117}
118
119function unifiedDiff(provider: ProviderDriverKind, turnIndex: number): string {
120 const lines = sourceModules
121 .slice(0, 8)
122 .flatMap((modulePath, index) => [
123 `diff --git a/${modulePath} b/${modulePath}`,
124 `--- a/${modulePath}`,
125 `+++ b/${modulePath}`,
126 `@@ -${index + 1},2 +${index + 1},3 @@`,
127 ` const provider = "${provider}";`,
128 `+const transferTurn = ${turnIndex + 1};`,
129 `+const transferSample = ${1_500 + index * 97};`,
130 ]);
131 return lines.join("\n");
132}
133
134function baseEvent(
135 provider: ProviderDriverKind,
136 turnIndex: number,
137 eventIndex: number,
138): Pick<FixtureProviderRuntimeEvent, "eventId" | "provider" | "createdAt" | "threadId"> {
139 return {
140 eventId: EventId.make(`recorded:${provider}:${turnIndex}:${eventIndex}`),
141 provider,
142 createdAt: fixtureTimestamp(turnIndex, eventIndex),
143 threadId: FIXTURE_THREAD_ID,
144 };
145}
146
147/**
148 * Synthetic canonical events calibrated from heavy local Codex and Claude
149 * threads. Ten historical turns produce 9 MB of retained MCP results without
150 * committing user content. Command output is intentionally modest because the
151 * client projection strips it.
152 */
153export function makeRecordedTransferTurn(
154 provider: ProviderDriverKind,
155 turnIndex: number,
156): TestTurnResponse {
157 const measuredTurn = turnIndex >= TRANSFER_HISTORY_TURN_COUNT;
158 const toolCount = measuredTurn ? TRANSFER_MEASURED_TOOLS : TRANSFER_HISTORY_TOOLS_PER_TURN;
159 const turnId = `${FIXTURE_TURN_ID}-${turnIndex + 1}`;
160 const events: FixtureProviderRuntimeEvent[] = [];
161 let eventIndex = 0;
162
163 events.push({
164 type: "turn.started",
165 ...baseEvent(provider, turnIndex, eventIndex++),
166 turnId,
167 payload: {
168 model: provider === "codex" ? "gpt-5.4" : "claude-opus-4-1",
169 effort: provider === "codex" ? "high" : "default",
170 },
171 });
172
173 for (let toolIndex = 0; toolIndex < toolCount; toolIndex += 1) {
174 const itemId = `tool-${turnIndex + 1}-${toolIndex + 1}`;
175 const command =
176 provider === "codex"
177 ? `vp test transfer-budget-${toolIndex + 1}`
178 : `review transfer budget ${toolIndex + 1}`;
179 events.push(
180 {
181 type: "item.started",
182 ...baseEvent(provider, turnIndex, eventIndex++),
183 turnId,
184 itemId,
185 payload: {
186 itemType: "command_execution",
187 status: "inProgress",
188 title: `Inspect transfer path ${toolIndex + 1}`,
189 detail: "Inspecting the HTTP snapshot and WebSocket projection boundaries.",
190 data: {
191 threadId: FIXTURE_THREAD_ID,
192 turnId,
193 startedAtMs: turnIndex * 60_000 + eventIndex,
194 item: {
195 id: itemId,
196 type: "commandExecution",
197 command,
198 cwd: "/workspace/transfer-budget",
199 processId: String(toolIndex + 1),
200 status: "inProgress",
201 commandActions: [],
202 aggregatedOutput: "",
203 },
204 },
205 },
206 },
207 {
208 type: "item.completed",
209 ...baseEvent(provider, turnIndex, eventIndex++),
210 turnId,
211 itemId,
212 payload: {
213 itemType: "command_execution",
214 status: "completed",
215 title: `Inspected transfer path ${toolIndex + 1}`,
216 detail: "Collected a deterministic multi-module transfer diagnostic.",
217 data: {
218 threadId: FIXTURE_THREAD_ID,
219 turnId,
220 completedAtMs: turnIndex * 60_000 + eventIndex,
221 item: {
222 id: itemId,
223 type: "commandExecution",
224 command,
225 cwd: "/workspace/transfer-budget",
226 processId: String(toolIndex + 1),
227 status: "completed",
228 commandActions: [],
229 aggregatedOutput: diagnosticOutput({
230 provider,
231 turnIndex,
232 toolIndex,
233 targetBytes: 1_000,
234 }),
235 exitCode: 0,
236 durationMs: 500 + toolIndex,
237 },
238 },
239 },
240 },
241 );
242 }
243
244 const mcpItemId = `mcp-${turnIndex + 1}`;
245 const mcpResultBytes = measuredTurn
246 ? TRANSFER_MEASURED_MCP_RESULT_BYTES
247 : TRANSFER_HISTORY_MCP_RESULT_BYTES;
248 events.push(
249 {
250 type: "item.started",
251 ...baseEvent(provider, turnIndex, eventIndex++),
252 turnId,
253 itemId: mcpItemId,
254 payload: {
255 itemType: "mcp_tool_call",
256 status: "inProgress",
257 title: "fixture-history · inspect_transfer_log",
258 detail: "Reading a retained diagnostic result from the provider history.",
259 data: {
260 startedAtMs: turnIndex * 60_000 + eventIndex,
261 threadId: FIXTURE_THREAD_ID,
262 turnId,
263 item: {
264 type: "mcpToolCall",
265 id: mcpItemId,
266 server: "fixture-history",
267 tool: "inspect_transfer_log",
268 arguments: { turn: turnIndex + 1 },
269 status: "inProgress",
270 },
271 },
272 },
273 },
274 {
275 type: "item.completed",
276 ...baseEvent(provider, turnIndex, eventIndex++),
277 turnId,
278 itemId: mcpItemId,
279 payload: {
280 itemType: "mcp_tool_call",
281 status: "completed",
282 title: "fixture-history · inspect_transfer_log",
283 detail: "Retained a deterministic diagnostic result in the thread history.",
284 data: {
285 completedAtMs: turnIndex * 60_000 + eventIndex,
286 threadId: FIXTURE_THREAD_ID,
287 turnId,
288 item: {
289 type: "mcpToolCall",
290 id: mcpItemId,
291 server: "fixture-history",
292 tool: "inspect_transfer_log",
293 arguments: { turn: turnIndex + 1 },
294 durationMs: 1_000 + turnIndex,
295 error: null,
296 result: {
297 content: [
298 {
299 type: "text",
300 text: diagnosticOutput({
301 provider,
302 turnIndex,
303 toolIndex: toolCount,
304 targetBytes: mcpResultBytes,
305 }),
306 },
307 ],
308 },
309 status: "completed",
310 },
311 },
312 },
313 },
314 );
315
316 const chunks = assistantChunks(provider, turnIndex);
317 for (const [contentIndex, delta] of chunks.entries()) {
318 events.push({
319 type: "content.delta",
320 ...baseEvent(provider, turnIndex, eventIndex++),
321 turnId,
322 itemId: `assistant-${turnIndex + 1}`,
323 payload: {
324 streamKind: "assistant_text",
325 delta,
326 contentIndex,
327 },
328 });
329 }
330
331 events.push(
332 {
333 type: "thread.token-usage.updated",
334 ...baseEvent(provider, turnIndex, eventIndex++),
335 turnId,
336 payload: {
337 usage: {
338 usedTokens: 18_000 + turnIndex * 1_900,
339 maxTokens: 200_000,
340 inputTokens: 15_000 + turnIndex * 1_700,
341 cachedInputTokens: 9_000 + turnIndex * 1_100,
342 outputTokens: 3_000 + turnIndex * 200,
343 toolUses: toolCount,
344 durationMs: 4_000 + turnIndex * 250,
345 },
346 },
347 },
348 {
349 type: "turn.diff.updated",
350 ...baseEvent(provider, turnIndex, eventIndex++),
351 turnId,
352 payload: {
353 unifiedDiff: unifiedDiff(provider, turnIndex),
354 },
355 },
356 {
357 type: "turn.completed",
358 ...baseEvent(provider, turnIndex, eventIndex),
359 turnId,
360 payload: {
361 state: "completed",
362 stopReason: "end_turn",
363 usage: {
364 inputTokens: 15_000 + turnIndex * 1_700,
365 outputTokens: 3_000 + turnIndex * 200,
366 },
367 },
368 },
369 );
370
371 return { events };
372}
apps/server/src/server.test.ts+246-25
...@@ -2,7 +2,7 @@ import * as NodeHttpServer from "@effect/platform-node/NodeHttpServer";...@@ -2,7 +2,7 @@ import * as NodeHttpServer from "@effect/platform-node/NodeHttpServer";
2import * as NodeSocket from "@effect/platform-node/NodeSocket";2import * as NodeSocket from "@effect/platform-node/NodeSocket";
3import * as NodeServices from "@effect/platform-node/NodeServices";3import * as NodeServices from "@effect/platform-node/NodeServices";
4import * as NodeCrypto from "node:crypto";4import * as NodeCrypto from "node:crypto";
5import { HostProcessPlatform } from "@t3tools/shared/hostProcess";5import { HostProcessEnvironment, HostProcessPlatform } from "@t3tools/shared/hostProcess";
66
7import {7import {
8 AuthAccessTokenType,8 AuthAccessTokenType,
...@@ -16,6 +16,8 @@ import {...@@ -16,6 +16,8 @@ import {
16 KeybindingRule,16 KeybindingRule,
17 MessageId,17 MessageId,
18 ExternalLauncherCommandNotFoundError,18 ExternalLauncherCommandNotFoundError,
19 OrchestrationThreadDetailSnapshot,
20 type OrchestrationThreadStreamItem,
19 type OrchestrationThreadShell,21 type OrchestrationThreadShell,
20 TerminalNotRunningError,22 TerminalNotRunningError,
21 type OrchestrationCommand,23 type OrchestrationCommand,
...@@ -41,6 +43,7 @@ import * as RelayClient from "@t3tools/shared/relayClient";...@@ -41,6 +43,7 @@ import * as RelayClient from "@t3tools/shared/relayClient";
41import { assert, it } from "@effect/vitest";43import { assert, it } from "@effect/vitest";
42import { assertFailure, assertInclude, assertTrue } from "@effect/vitest/utils";44import { assertFailure, assertInclude, assertTrue } from "@effect/vitest/utils";
43import * as Clock from "effect/Clock";45import * as Clock from "effect/Clock";
46import * as Config from "effect/Config";
44import * as Deferred from "effect/Deferred";47import * as Deferred from "effect/Deferred";
45import * as DateTime from "effect/DateTime";48import * as DateTime from "effect/DateTime";
46import * as Duration from "effect/Duration";49import * as Duration from "effect/Duration";
...@@ -52,6 +55,8 @@ import * as ManagedRuntime from "effect/ManagedRuntime";...@@ -52,6 +55,8 @@ import * as ManagedRuntime from "effect/ManagedRuntime";
52import * as Option from "effect/Option";55import * as Option from "effect/Option";
53import * as Path from "effect/Path";56import * as Path from "effect/Path";
54import * as PubSub from "effect/PubSub";57import * as PubSub from "effect/PubSub";
58import * as Queue from "effect/Queue";
59import * as Schema from "effect/Schema";
55import * as Stream from "effect/Stream";60import * as Stream from "effect/Stream";
56import * as TestClock from "effect/testing/TestClock";61import * as TestClock from "effect/testing/TestClock";
57import { ChildProcessSpawner } from "effect/unstable/process";62import { ChildProcessSpawner } from "effect/unstable/process";
...@@ -70,11 +75,34 @@ import * as Socket from "effect/unstable/socket/Socket";...@@ -70,11 +75,34 @@ import * as Socket from "effect/unstable/socket/Socket";
70import { vi } from "vite-plus/test";75import { vi } from "vite-plus/test";
7176
72const TEST_EPOCH = DateTime.makeUnsafe("1970-01-01T00:00:00.000Z");77const TEST_EPOCH = DateTime.makeUnsafe("1970-01-01T00:00:00.000Z");
78const decodeTransferThreadSnapshot = Schema.decodeUnknownEffect(
79 Schema.fromJsonString(OrchestrationThreadDetailSnapshot),
80);
81
82const collectQueueUntil = Effect.fn("TransferBudget.collectQueueUntil")(function* <A>(
83 queue: Queue.Queue<A>,
84 predicate: (value: A) => boolean,
85 waitDescription: string,
86) {
87 return yield* Effect.gen(function* () {
88 const values: A[] = [];
89 while (true) {
90 const value = yield* Queue.take(queue);
91 values.push(value);
92 if (predicate(value)) return values;
93 }
94 }).pipe(
95 Effect.timeoutOrElse({
96 duration: "10 seconds",
97 orElse: () => Effect.die(new Error(`Timed out waiting for ${waitDescription}`)),
98 }),
99 );
100});
73101
74import * as BackgroundPolicy from "./background/BackgroundPolicy.ts";102import * as BackgroundPolicy from "./background/BackgroundPolicy.ts";
75import * as ServerConfig from "./config.ts";103import * as ServerConfig from "./config.ts";
76import { makeRoutesLayer } from "./server.ts";104import { makeRoutesLayer } from "./server.ts";
77import { resolveAvailableEditorsForConfig } from "./ws.ts";105import { isThreadDetailEvent, resolveAvailableEditorsForConfig } from "./ws.ts";
78import * as CheckpointDiffQuery from "./checkpointing/CheckpointDiffQuery.ts";106import * as CheckpointDiffQuery from "./checkpointing/CheckpointDiffQuery.ts";
79import * as GitManager from "./git/GitManager.ts";107import * as GitManager from "./git/GitManager.ts";
80import * as Keybindings from "./keybindings.ts";108import * as Keybindings from "./keybindings.ts";
...@@ -123,6 +151,32 @@ import * as ResourceAttribution from "./resourceTelemetry/ResourceAttribution.ts...@@ -123,6 +151,32 @@ import * as ResourceAttribution from "./resourceTelemetry/ResourceAttribution.ts
123import * as ResourceTelemetry from "./resourceTelemetry/ResourceTelemetry.ts";151import * as ResourceTelemetry from "./resourceTelemetry/ResourceTelemetry.ts";
124import * as Data from "effect/Data";152import * as Data from "effect/Data";
125153
154import { makeOrchestrationIntegrationHarness } from "../integration/OrchestrationEngineHarness.integration.ts";
155import {
156 countingWsRpcProtocolLayer,
157 makeCountingWsRpcClient,
158 makeWebSocketTransferRecorder,
159 measureHttpGet,
160 transferDelta,
161} from "../integration/NetworkTransferMeasurement.integration.ts";
162import {
163 expectedMeasuredAssistantText,
164 queueMeasuredTransferTurn,
165 seedTransferBudgetHistory,
166 TRANSFER_HISTORY_TURN_COUNT,
167 TRANSFER_MEASURED_TURN_CREATED_AT,
168 TRANSFER_MEASURED_TURN_INDEX,
169 TRANSFER_THREAD_ID,
170 transferModelSelection,
171 waitForTurnQuiesced,
172} from "../integration/TransferBudgetScenario.integration.ts";
173import {
174 formatTransferBudgetReport,
175 formatTransferBudgetResult,
176 type TransferBudgetRun,
177 transferBudgetViolations,
178} from "../integration/TransferBudgetReport.integration.ts";
179
126const defaultProjectId = ProjectId.make("project-default");180const defaultProjectId = ProjectId.make("project-default");
127const defaultThreadId = ThreadId.make("thread-default");181const defaultThreadId = ThreadId.make("thread-default");
128const defaultDesktopBootstrapToken = "test-desktop-bootstrap-token";182const defaultDesktopBootstrapToken = "test-desktop-bootstrap-token";
...@@ -549,9 +603,12 @@ const buildAppUnderTest = (options?: {...@@ -549,9 +603,12 @@ const buildAppUnderTest = (options?: {
549 ),603 ),
550 ),604 ),
551 );605 );
606 const serviceLauncherClientLayer = ServiceLauncherClient.layer.pipe(
607 Layer.provide(Layer.succeed(HostProcessEnvironment, {})),
608 );
552609
553 const servedRoutesLayer = HttpRouter.serve(610 const servedRoutesLayer = HttpRouter.serve(
554 makeRoutesLayer.pipe(Layer.provide(ServiceLauncherClient.layer)),611 makeRoutesLayer.pipe(Layer.provide(serviceLauncherClientLayer)),
555 {612 {
556 disableListenLog: true,613 disableListenLog: true,
557 disableLogger: true,614 disableLogger: true,
...@@ -1319,6 +1376,28 @@ const getWsServerUrl = (...@@ -1319,6 +1376,28 @@ const getWsServerUrl = (
1319 );1376 );
1320 });1377 });
13211378
1379// Mirrors NodeHttpServer.layerTest, which does not expose server options,
1380// with the production `websocket: { perMessageDeflate: true }` setting.
1381const NodeHttpServerTestWithWsDeflate = HttpServer.layerTestClient.pipe(
1382 Layer.provide(
1383 Layer.fresh(FetchHttpClient.layer).pipe(
1384 Layer.provide(Layer.succeed(FetchHttpClient.RequestInit)({ keepalive: false })),
1385 ),
1386 ),
1387 Layer.provideMerge(
1388 Layer.unwrap(
1389 Effect.map(
1390 Effect.promise(() => import("node:http")),
1391 (NodeHttp) =>
1392 NodeHttpServer.layer(NodeHttp.createServer, {
1393 port: 0,
1394 websocket: { perMessageDeflate: true },
1395 }),
1396 ),
1397 ),
1398 ),
1399);
1400
1322it.layer(NodeServices.layer)("server router seam", (it) => {1401it.layer(NodeServices.layer)("server router seam", (it) => {
1323 it.effect("parks HTTP ingress until command readiness", () =>1402 it.effect("parks HTTP ingress until command readiness", () =>
1324 Effect.gen(function* () {1403 Effect.gen(function* () {
...@@ -3219,28 +3298,6 @@ it.layer(NodeServices.layer)("server router seam", (it) => {...@@ -3219,28 +3298,6 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
3219 }).pipe(Effect.provide(NodeHttpServer.layerTest)),3298 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
3220 );3299 );
32213300
3222 // Mirrors NodeHttpServer.layerTest, which does not expose server options,
3223 // with the production `websocket: { perMessageDeflate: true }` setting.
3224 const NodeHttpServerTestWithWsDeflate = HttpServer.layerTestClient.pipe(
3225 Layer.provide(
3226 Layer.fresh(FetchHttpClient.layer).pipe(
3227 Layer.provide(Layer.succeed(FetchHttpClient.RequestInit)({ keepalive: false })),
3228 ),
3229 ),
3230 Layer.provideMerge(
3231 Layer.unwrap(
3232 Effect.map(
3233 Effect.promise(() => import("node:http")),
3234 (NodeHttp) =>
3235 NodeHttpServer.layer(NodeHttp.createServer, {
3236 port: 0,
3237 websocket: { perMessageDeflate: true },
3238 }),
3239 ),
3240 ),
3241 ),
3242 );
3243
3244 it.effect("negotiates permessage-deflate with clients that offer it", () =>3301 it.effect("negotiates permessage-deflate with clients that offer it", () =>
3245 Effect.gen(function* () {3302 Effect.gen(function* () {
3246 yield* buildAppUnderTest();3303 yield* buildAppUnderTest();
...@@ -7839,3 +7896,167 @@ it.layer(NodeServices.layer)("server router seam", (it) => {...@@ -7839,3 +7896,167 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
7839 }).pipe(Effect.provide(NodeHttpServer.layerTest)),7896 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
7840 );7897 );
7841});7898});
7899
7900it.live(
7901 "reports thread HTTP and WebSocket transfer budgets",
7902 () =>
7903 Effect.gen(function* () {
7904 const providers = [
7905 ProviderDriverKind.make("codex"),
7906 ProviderDriverKind.make("claudeAgent"),
7907 ] as const;
7908
7909 const runs = yield* Effect.forEach(
7910 providers,
7911 (provider) =>
7912 Effect.acquireUseRelease(
7913 makeOrchestrationIntegrationHarness({ provider }),
7914 (harness) =>
7915 Effect.gen(function* () {
7916 yield* seedTransferBudgetHistory(harness, provider);
7917 yield* buildAppUnderTest({
7918 layers: {
7919 orchestrationEngine: harness.engine,
7920 projectionSnapshotQuery: harness.snapshotQuery,
7921 },
7922 });
7923
7924 const baseUrl = yield* getHttpServerUrl();
7925 const cookie = yield* getAuthenticatedSessionCookieHeader();
7926
7927 const recorder = makeWebSocketTransferRecorder();
7928 const wsUrl = baseUrl.replace(/^http:/, "ws:") + "/ws";
7929 const protocolLayer = countingWsRpcProtocolLayer({
7930 url: wsUrl,
7931 cookie,
7932 recorder,
7933 });
7934
7935 return yield* Effect.scoped(
7936 Effect.gen(function* () {
7937 const client = yield* makeCountingWsRpcClient;
7938
7939 const threadSnapshot = yield* measureHttpGet({
7940 url: `${baseUrl}/api/orchestration/threads/${TRANSFER_THREAD_ID}`,
7941 headers: { cookie },
7942 });
7943 assert.equal(threadSnapshot.status, 200);
7944 assert.equal(threadSnapshot.contentEncoding, "gzip");
7945 const decodedThread = yield* decodeTransferThreadSnapshot(
7946 Buffer.from(threadSnapshot.decodedBody).toString("utf8"),
7947 );
7948 assert.equal(
7949 decodedThread.thread.messages.length,
7950 TRANSFER_HISTORY_TURN_COUNT * 2,
7951 );
7952
7953 const threadItems = yield* Queue.unbounded<OrchestrationThreadStreamItem>();
7954 yield* client[ORCHESTRATION_WS_METHODS.subscribeThread]({
7955 threadId: TRANSFER_THREAD_ID,
7956 afterSequence: decodedThread.snapshotSequence,
7957 requestCompletionMarker: true,
7958 }).pipe(
7959 Stream.runForEach((item) =>
7960 Queue.offer(threadItems, item).pipe(Effect.asVoid),
7961 ),
7962 Effect.forkScoped,
7963 );
7964 const initialThreadItems = yield* collectQueueUntil(
7965 threadItems,
7966 (item) => item.kind === "synchronized",
7967 `${provider} thread subscription to synchronize`,
7968 );
7969 assert.isFalse(initialThreadItems.some((item) => item.kind === "snapshot"));
7970 assert.include(recorder.negotiatedExtensions(), "permessage-deflate");
7971
7972 yield* queueMeasuredTransferTurn(harness, provider);
7973 const turnStartTotals = recorder.totals();
7974 yield* client[ORCHESTRATION_WS_METHODS.dispatchCommand]({
7975 type: "thread.turn.start",
7976 commandId: CommandId.make(`transfer:${provider}:measured-turn`),
7977 threadId: TRANSFER_THREAD_ID,
7978 message: {
7979 messageId: MessageId.make("transfer-user-measured"),
7980 role: "user",
7981 text: "Measure the client-bound transfer for this turn.",
7982 attachments: [],
7983 },
7984 modelSelection: transferModelSelection(provider),
7985 runtimeMode: "approval-required",
7986 interactionMode: "default",
7987 createdAt: TRANSFER_MEASURED_TURN_CREATED_AT,
7988 });
7989 yield* waitForTurnQuiesced(harness, TRANSFER_MEASURED_TURN_INDEX + 1);
7990 const finalThreadSequence = yield* harness.engine
7991 .readEvents(decodedThread.snapshotSequence, 10_000)
7992 .pipe(
7993 Stream.runFold(
7994 () => decodedThread.snapshotSequence,
7995 (sequence, event) =>
7996 event.aggregateId === TRANSFER_THREAD_ID && isThreadDetailEvent(event)
7997 ? Math.max(sequence, event.sequence)
7998 : sequence,
7999 ),
8000 );
8001 assert.isAbove(finalThreadSequence, decodedThread.snapshotSequence);
8002
8003 yield* collectQueueUntil(
8004 threadItems,
8005 (item) =>
8006 item.kind === "event" && item.event.sequence === finalThreadSequence,
8007 `${provider} thread stream to reach sequence ${finalThreadSequence}`,
8008 );
8009 const measuredTurnWebSocket = transferDelta(turnStartTotals, recorder.totals());
8010
8011 const finalThreadSnapshot = yield* harness.snapshotQuery
8012 .getThreadDetailSnapshot(TRANSFER_THREAD_ID)
8013 .pipe(Effect.map(Option.getOrThrow));
8014 const expectedAssistantText = expectedMeasuredAssistantText(provider);
8015 const measuredAssistant = finalThreadSnapshot.thread.messages.find(
8016 (message) =>
8017 message.role === "assistant" && message.text === expectedAssistantText,
8018 );
8019 assert.isDefined(measuredAssistant);
8020 assert.isTrue(
8021 finalThreadSnapshot.thread.messages.length >= TRANSFER_HISTORY_TURN_COUNT * 2,
8022 );
8023 assert.equal(measuredAssistant?.streaming, false);
8024 assert.equal(finalThreadSnapshot.thread.session?.status, "ready");
8025 assert.equal(
8026 finalThreadSnapshot.thread.checkpoints.length,
8027 TRANSFER_HISTORY_TURN_COUNT + 1,
8028 );
8029
8030 return {
8031 provider,
8032 threadSnapshot,
8033 measuredTurnWebSocket,
8034 } satisfies TransferBudgetRun;
8035 }).pipe(Effect.provide(protocolLayer)),
8036 );
8037 }),
8038 (harness) => harness.dispose,
8039 ).pipe(Effect.provide(NodeHttpServerTestWithWsDeflate)),
8040 { concurrency: 1 },
8041 );
8042
8043 const report = formatTransferBudgetReport(runs);
8044 yield* Effect.logInfo(`\n${report}`);
8045 const reportPath = yield* Config.string("T3CODE_TRANSFER_BUDGET_REPORT_PATH").pipe(
8046 Config.option,
8047 );
8048 if (Option.isSome(reportPath)) {
8049 const fileSystem = yield* FileSystem.FileSystem;
8050 yield* fileSystem.writeFileString(reportPath.value, report);
8051 }
8052 const resultPath = yield* Config.string("T3CODE_TRANSFER_BUDGET_RESULT_PATH").pipe(
8053 Config.option,
8054 );
8055 if (Option.isSome(resultPath)) {
8056 const fileSystem = yield* FileSystem.FileSystem;
8057 yield* fileSystem.writeFileString(resultPath.value, formatTransferBudgetResult(runs));
8058 }
8059 assert.deepEqual(transferBudgetViolations(runs), []);
8060 }).pipe(Effect.provide(NodeServices.layer)),
8061 120_000,
8062);
apps/server/src/ws.ts+1-1
...@@ -268,7 +268,7 @@ function projectSetupScriptCompatibilityDetail(...@@ -268,7 +268,7 @@ function projectSetupScriptCompatibilityDetail(
268 }268 }
269}269}
270270
271function isThreadDetailEvent(event: OrchestrationEvent): event is Extract<271export function isThreadDetailEvent(event: OrchestrationEvent): event is Extract<
272 OrchestrationEvent,272 OrchestrationEvent,
273 {273 {
274 type:274 type: