diff --git a/.github/scripts/thread-transfer-report.cjs b/.github/scripts/thread-transfer-report.cjs
new file mode 100644
index 00000000000..94a02b7806d
--- /dev/null
+++ b/.github/scripts/thread-transfer-report.cjs
@@ -0,0 +1,429 @@
+const fs = require("node:fs");
+const path = require("node:path");
+
+const ARTIFACT_NAME = "thread-transfer-results";
+const RESULT_FILE = "thread-transfer-result.json";
+const COMMENT_MARKER = "";
+const PROVIDERS = ["codex", "claudeAgent"];
+const OBSERVED_KEYS = [
+ "totalWireBytes",
+ "threadSnapshotWireBytes",
+ "threadSnapshotDecodedBytes",
+ "measuredTurnWebSocketWireBytes",
+ "measuredTurnWebSocketDecodedBytes",
+ "measuredTurnWebSocketMessages",
+];
+const CEILING_KEYS = [
+ "totalWireBytes",
+ "threadSnapshotWireBytes",
+ "measuredTurnWebSocketWireBytes",
+ "measuredTurnWebSocketDecodedBytes",
+ "measuredTurnWebSocketMessages",
+];
+const SCENARIO_KEYS = [
+ "id",
+ "historyTurns",
+ "historyCommandToolsPerTurn",
+ "historyMcpResultBytes",
+ "measuredCommandTools",
+ "measuredMcpResultBytes",
+];
+
+function resultShaMarker(sha) {
+ return ``;
+}
+
+function assertObject(value, label) {
+ if (value === null || typeof value !== "object" || Array.isArray(value)) {
+ throw new Error(`${label} must be an object`);
+ }
+}
+
+function assertExactKeys(value, expected, label) {
+ assertObject(value, label);
+ const actual = Object.keys(value).sort();
+ const wanted = [...expected].sort();
+ if (actual.length !== wanted.length || actual.some((key, index) => key !== wanted[index])) {
+ throw new Error(`${label} has unexpected fields`);
+ }
+}
+
+function assertMetric(value, label) {
+ if (!Number.isSafeInteger(value) || value < 0 || value > 1_000_000_000) {
+ throw new Error(`${label} must be a non-negative safe integer below 1,000,000,000`);
+ }
+}
+
+function validateResult(value) {
+ assertExactKeys(value, ["schemaVersion", "scenario", "providers"], "result");
+ if (value.schemaVersion !== 1) {
+ throw new Error("result.schemaVersion must be 1");
+ }
+
+ assertExactKeys(value.scenario, SCENARIO_KEYS, "result.scenario");
+ if (value.scenario.id !== "thread-transfer-v1") {
+ throw new Error("result.scenario.id is not supported");
+ }
+ for (const key of SCENARIO_KEYS.slice(1)) {
+ assertMetric(value.scenario[key], `result.scenario.${key}`);
+ }
+
+ assertExactKeys(value.providers, PROVIDERS, "result.providers");
+ for (const provider of PROVIDERS) {
+ const entry = value.providers[provider];
+ assertExactKeys(entry, ["observed", "ceiling"], `result.providers.${provider}`);
+ assertExactKeys(entry.observed, OBSERVED_KEYS, `result.providers.${provider}.observed`);
+ assertExactKeys(entry.ceiling, CEILING_KEYS, `result.providers.${provider}.ceiling`);
+ for (const key of OBSERVED_KEYS) {
+ assertMetric(entry.observed[key], `result.providers.${provider}.observed.${key}`);
+ }
+ for (const key of CEILING_KEYS) {
+ assertMetric(entry.ceiling[key], `result.providers.${provider}.ceiling.${key}`);
+ }
+ }
+
+ return value;
+}
+
+function readResult(directory) {
+ if (!directory) return undefined;
+ const file = path.join(directory, RESULT_FILE);
+ if (!fs.existsSync(file)) return undefined;
+ const stat = fs.lstatSync(file);
+ if (!stat.isFile() || stat.size > 64 * 1_024) {
+ throw new Error("thread transfer result must be a regular file smaller than 64 KiB");
+ }
+ return validateResult(JSON.parse(fs.readFileSync(file, "utf8")));
+}
+
+function formatBytes(bytes) {
+ if (bytes < 1_024) return `${bytes} B`;
+ if (bytes >= 1_024 * 1_024) return `${(bytes / 1_024 / 1_024).toFixed(2)} MiB`;
+ return `${(bytes / 1_024).toFixed(1)} KiB`;
+}
+
+function formatValue(value, kind) {
+ return kind === "messages" ? value.toLocaleString("en-US") : formatBytes(value);
+}
+
+function formatImpact(current, baseline, kind) {
+ if (baseline === undefined) return "—";
+ const delta = current - baseline;
+ const prefix = delta > 0 ? "+" : delta < 0 ? "−" : "";
+ const magnitude = formatValue(Math.abs(delta), kind);
+ const percent =
+ baseline === 0 ? "" : ` (${prefix}${Math.abs((delta / baseline) * 100).toFixed(1)}%)`;
+ return `${prefix}${magnitude}${percent}`;
+}
+
+function sameScenario(left, right) {
+ return SCENARIO_KEYS.every((key) => left[key] === right[key]);
+}
+
+const METRICS = [
+ { key: "totalWireBytes", label: "Total thread wire", kind: "bytes" },
+ { key: "threadSnapshotWireBytes", label: "Thread snapshot wire", kind: "bytes" },
+ {
+ key: "measuredTurnWebSocketWireBytes",
+ label: "Live turn WebSocket wire",
+ kind: "bytes",
+ },
+ {
+ key: "measuredTurnWebSocketDecodedBytes",
+ label: "Live turn WebSocket decoded",
+ kind: "bytes",
+ },
+ { key: "measuredTurnWebSocketMessages", label: "Live turn messages", kind: "messages" },
+];
+
+function renderComment(input) {
+ const current = input.current;
+ const baseline = input.baseline;
+ const comparable = baseline !== undefined && sameScenario(current.scenario, baseline.scenario);
+ const rows = [];
+ const ceilingChanges = [];
+ let failed = false;
+
+ for (const provider of PROVIDERS) {
+ for (const metric of METRICS) {
+ const observed = current.providers[provider].observed[metric.key];
+ const ceiling = current.providers[provider].ceiling[metric.key];
+ const baselineObserved = comparable
+ ? baseline.providers[provider].observed[metric.key]
+ : undefined;
+ const pass = observed <= ceiling;
+ failed ||= !pass;
+ rows.push(
+ `| ${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 ? "✅" : "❌"} |`,
+ );
+
+ if (baseline && baseline.providers[provider].ceiling[metric.key] !== ceiling) {
+ ceilingChanges.push(
+ `- ${provider === "codex" ? "Codex" : "Claude"} ${metric.label}: ${formatValue(baseline.providers[provider].ceiling[metric.key], metric.kind)} → ${formatValue(ceiling, metric.kind)}`,
+ );
+ }
+ }
+ }
+
+ const baselineLink = input.baselineRun
+ ? `[\`${input.baselineRun.sha.slice(0, 7)}\`](${input.baselineRun.url})`
+ : "unavailable";
+ const currentLink = `[\`${input.currentRun.sha.slice(0, 7)}\`](${input.currentRun.url})`;
+ const notices = [];
+ if (!baseline) {
+ notices.push(
+ "> ℹ️ No successful `main` baseline artifact is available yet. This run establishes the initial measurement.",
+ );
+ } else if (!comparable) {
+ notices.push(
+ "> ⚠️ The thread fixture changed, so impact percentages are not directly comparable to the `main` baseline.",
+ );
+ } else if (!input.baselineRun.matchesBase) {
+ notices.push(
+ "> ℹ️ The exact PR base did not have a successful artifact. Baseline uses the latest successful `main` measurement shown below.",
+ );
+ }
+ if (ceilingChanges.length > 0) {
+ notices.push(
+ `> ⚠️ **This PR changes transfer ceilings:**\n>\n${ceilingChanges.map((line) => `> ${line}`).join("\n")}`,
+ );
+ }
+
+ return [
+ COMMENT_MARKER,
+ resultShaMarker(input.currentRun.sha),
+ "## Thread transfer impact",
+ "",
+ failed
+ ? "❌ One or more thread transfer ceilings were exceeded."
+ : "✅ Thread transfer remains within every enforced ceiling.",
+ ...(notices.length > 0 ? ["", ...notices] : []),
+ "",
+ "| Provider | Metric | Main baseline | This PR | Impact | PR ceiling | |",
+ "| --- | --- | ---: | ---: | ---: | ---: | --- |",
+ ...rows,
+ "",
+ `Baseline: ${baselineLink} · PR result: ${currentLink} · Source CI: ${input.currentRun.conclusion}`,
+ "",
+ "",
+ "Scenario and decoded snapshot size
",
+ "",
+ `${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.`,
+ "",
+ ...PROVIDERS.map(
+ (provider) =>
+ `- ${provider === "codex" ? "Codex" : "Claude"} decoded thread snapshot: ${formatBytes(current.providers[provider].observed.threadSnapshotDecodedBytes)}`,
+ ),
+ "",
+ " ",
+ "",
+ "_Updated in place by a trusted workflow. PR artifacts are strictly validated and never executed._",
+ ].join("\n");
+}
+
+async function artifactsForRun(github, owner, repo, runId) {
+ return github.paginate(github.rest.actions.listWorkflowRunArtifacts, {
+ owner,
+ repo,
+ run_id: runId,
+ per_page: 100,
+ });
+}
+
+function findResultArtifact(artifacts) {
+ return artifacts.find((artifact) => artifact.name === ARTIFACT_NAME && !artifact.expired);
+}
+
+async function resolve({ github, context, core }) {
+ const source = context.payload.workflow_run;
+ const { owner, repo } = context.repo;
+ if (source.event !== "pull_request") {
+ core.setOutput("publish", "false");
+ return;
+ }
+
+ let pullNumber = source.pull_requests?.[0]?.number;
+ if (!pullNumber) {
+ const associated = await github.paginate(
+ github.rest.repos.listPullRequestsAssociatedWithCommit,
+ { owner, repo, commit_sha: source.head_sha, per_page: 100 },
+ );
+ const matchingPulls = associated.filter(
+ (pull) =>
+ pull.state === "open" &&
+ pull.head.sha === source.head_sha &&
+ pull.head.ref === source.head_branch,
+ );
+ if (matchingPulls.length !== 1) {
+ core.info(
+ `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}.`,
+ );
+ core.setOutput("publish", "false");
+ return;
+ }
+ pullNumber = matchingPulls[0].number;
+ }
+ if (!pullNumber) {
+ core.info("No open pull request is associated with the completed CI run.");
+ core.setOutput("publish", "false");
+ return;
+ }
+
+ const { data: pull } = await github.rest.pulls.get({ owner, repo, pull_number: pullNumber });
+ if (pull.head.sha !== source.head_sha) {
+ core.info(`Skipping stale CI result ${source.head_sha}; PR head is ${pull.head.sha}.`);
+ core.setOutput("publish", "false");
+ return;
+ }
+
+ const sourceArtifacts = await artifactsForRun(github, owner, repo, source.id);
+ const sourceArtifact = findResultArtifact(sourceArtifacts);
+ const workflowRuns = await github.paginate(github.rest.actions.listWorkflowRuns, {
+ owner,
+ repo,
+ workflow_id: source.workflow_id,
+ branch: pull.base.ref,
+ event: "push",
+ status: "success",
+ per_page: 100,
+ });
+ const orderedRuns = [
+ ...workflowRuns.filter((run) => run.head_sha === pull.base.sha),
+ ...workflowRuns.filter((run) => run.head_sha !== pull.base.sha),
+ ].slice(0, 20);
+
+ let baselineRun;
+ for (const run of orderedRuns) {
+ const artifacts = await artifactsForRun(github, owner, repo, run.id);
+ if (findResultArtifact(artifacts)) {
+ baselineRun = run;
+ break;
+ }
+ }
+
+ core.setOutput("publish", "true");
+ core.setOutput("pull_number", String(pullNumber));
+ core.setOutput("pr_artifact", sourceArtifact ? "true" : "false");
+ core.setOutput("pr_run_id", String(source.id));
+ core.setOutput("pr_sha", source.head_sha);
+ core.setOutput("pr_conclusion", source.conclusion ?? "unknown");
+ core.setOutput("baseline_artifact", baselineRun ? "true" : "false");
+ core.setOutput("baseline_run_id", baselineRun ? String(baselineRun.id) : "");
+ core.setOutput("baseline_sha", baselineRun?.head_sha ?? "");
+ core.setOutput(
+ "baseline_matches_base",
+ baselineRun?.head_sha === pull.base.sha ? "true" : "false",
+ );
+}
+
+async function upsertComment(github, context, pullNumber, body, options = {}) {
+ const { owner, repo } = context.repo;
+ const comments = await github.paginate(github.rest.issues.listComments, {
+ owner,
+ repo,
+ issue_number: pullNumber,
+ per_page: 100,
+ });
+ const existing = comments.find(
+ (comment) =>
+ comment.user?.login === "github-actions[bot]" && comment.body?.includes(COMMENT_MARKER),
+ );
+ if (
+ options.preserveResultSha &&
+ existing?.body?.includes(resultShaMarker(options.preserveResultSha))
+ ) {
+ return;
+ }
+ if (existing) {
+ await github.rest.issues.updateComment({ owner, repo, comment_id: existing.id, body });
+ } else {
+ await github.rest.issues.createComment({ owner, repo, issue_number: pullNumber, body });
+ }
+}
+
+async function upsertCommentForCurrentHead(
+ github,
+ context,
+ core,
+ pullNumber,
+ expectedSha,
+ body,
+ options,
+) {
+ const { owner, repo } = context.repo;
+ const { data: pull } = await github.rest.pulls.get({
+ owner,
+ repo,
+ pull_number: pullNumber,
+ });
+ if (pull.head.sha !== expectedSha) {
+ core.info(`Skipping stale CI result ${expectedSha}; PR head is ${pull.head.sha}.`);
+ return false;
+ }
+
+ await upsertComment(github, context, pullNumber, body, options);
+ return true;
+}
+
+async function publish({ github, context, core }) {
+ const pullNumber = Number(process.env.PR_NUMBER);
+ if (!Number.isSafeInteger(pullNumber) || pullNumber <= 0) {
+ throw new Error("PR_NUMBER is invalid");
+ }
+
+ const current = readResult(process.env.PR_RESULT_DIR);
+ const currentRun = {
+ sha: process.env.PR_SHA,
+ conclusion: process.env.PR_CONCLUSION,
+ url: `https://github.com/${context.repo.owner}/${context.repo.repo}/actions/runs/${process.env.PR_RUN_ID}`,
+ };
+ if (!current) {
+ await upsertCommentForCurrentHead(
+ github,
+ context,
+ core,
+ pullNumber,
+ currentRun.sha,
+ [
+ COMMENT_MARKER,
+ "## Thread transfer impact",
+ "",
+ `⚠️ The latest [CI run](${currentRun.url}) did not produce a thread transfer result for \`${currentRun.sha.slice(0, 7)}\`.`,
+ "",
+ "_This comment will update automatically after the next completed run._",
+ ].join("\n"),
+ { preserveResultSha: currentRun.sha },
+ );
+ return;
+ }
+
+ const baseline = readResult(process.env.BASELINE_RESULT_DIR);
+ const baselineRun = baseline
+ ? {
+ sha: process.env.BASELINE_SHA,
+ matchesBase: process.env.BASELINE_MATCHES_BASE === "true",
+ url: `https://github.com/${context.repo.owner}/${context.repo.repo}/actions/runs/${process.env.BASELINE_RUN_ID}`,
+ }
+ : undefined;
+ const body = renderComment({ current, baseline, currentRun, baselineRun });
+ const published = await upsertCommentForCurrentHead(
+ github,
+ context,
+ core,
+ pullNumber,
+ currentRun.sha,
+ body,
+ );
+ if (published) {
+ core.info(`Updated thread transfer report on PR #${pullNumber}.`);
+ }
+}
+
+module.exports = {
+ publish,
+ readResult,
+ renderComment,
+ resolve,
+ upsertCommentForCurrentHead,
+ validateResult,
+};
diff --git a/.github/scripts/thread-transfer-report.test.cjs b/.github/scripts/thread-transfer-report.test.cjs
new file mode 100644
index 00000000000..4935864e46f
--- /dev/null
+++ b/.github/scripts/thread-transfer-report.test.cjs
@@ -0,0 +1,292 @@
+const assert = require("node:assert/strict");
+const test = require("node:test");
+
+const {
+ renderComment,
+ resolve,
+ upsertCommentForCurrentHead,
+ validateResult,
+} = require("./thread-transfer-report.cjs");
+
+function result(overrides = {}) {
+ const observed = {
+ totalWireBytes: 2_200_000,
+ threadSnapshotWireBytes: 1_950_000,
+ threadSnapshotDecodedBytes: 9_100_000,
+ measuredTurnWebSocketWireBytes: 250_000,
+ measuredTurnWebSocketDecodedBytes: 1_150_000,
+ measuredTurnWebSocketMessages: 15,
+ };
+ const ceiling = {
+ totalWireBytes: 2_900_000,
+ threadSnapshotWireBytes: 2_600_000,
+ measuredTurnWebSocketWireBytes: 320_000,
+ measuredTurnWebSocketDecodedBytes: 1_550_000,
+ measuredTurnWebSocketMessages: 20,
+ };
+ return {
+ schemaVersion: 1,
+ scenario: {
+ id: "thread-transfer-v1",
+ historyTurns: 10,
+ historyCommandToolsPerTurn: 5,
+ historyMcpResultBytes: 900_000,
+ measuredCommandTools: 20,
+ measuredMcpResultBytes: 1_100_000,
+ },
+ providers: {
+ codex: { observed: { ...observed, ...overrides }, ceiling },
+ claudeAgent: { observed, ceiling },
+ },
+ };
+}
+
+test("validates the fixed artifact schema", () => {
+ assert.equal(validateResult(result()).schemaVersion, 1);
+ assert.throws(
+ () => validateResult({ ...result(), injectedMarkdown: "@everyone" }),
+ /unexpected fields/,
+ );
+ assert.throws(
+ () => validateResult(result({ totalWireBytes: "lots" })),
+ /non-negative safe integer/,
+ );
+});
+
+test("renders baseline, impact, ceiling, and ceiling changes", () => {
+ const baseline = result();
+ const current = result({ measuredTurnWebSocketWireBytes: 260_000 });
+ current.providers.codex.ceiling = {
+ ...current.providers.codex.ceiling,
+ measuredTurnWebSocketWireBytes: 330_000,
+ };
+ const comment = renderComment({
+ current,
+ baseline,
+ currentRun: {
+ sha: "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
+ conclusion: "success",
+ url: "https://github.com/pingdotgg/t3code/actions/runs/2",
+ },
+ baselineRun: {
+ sha: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
+ matchesBase: true,
+ url: "https://github.com/pingdotgg/t3code/actions/runs/1",
+ },
+ });
+
+ assert.match(comment, /Main baseline \| This PR \| Impact \| PR ceiling/);
+ assert.match(comment, /\+9\.8 KiB \(\+4\.0%\)/);
+ assert.match(comment, /This PR changes transfer ceilings/);
+ assert.match(comment, /312\.5 KiB → 322\.3 KiB/);
+ assert.match(comment, //);
+ assert.match(
+ comment,
+ //,
+ );
+});
+
+test("resolves a fallback PR with a redacted head repo and exact main baseline", async () => {
+ const outputs = {};
+ const listWorkflowRunArtifacts = () => {};
+ const listWorkflowRuns = () => {};
+ const listPullRequestsAssociatedWithCommit = () => {};
+ const github = {
+ paginate: async (method, input) => {
+ if (method === listPullRequestsAssociatedWithCommit) {
+ return [
+ {
+ number: 5350,
+ state: "open",
+ head: { sha: "head-sha", ref: "feature-branch", repo: null },
+ },
+ ];
+ }
+ if (method === listWorkflowRunArtifacts) {
+ return [
+ {
+ name: "thread-transfer-results",
+ expired: false,
+ runId: input.run_id,
+ },
+ ];
+ }
+ if (method === listWorkflowRuns) {
+ return [{ id: 1, head_sha: "base-sha" }];
+ }
+ throw new Error("unexpected pagination call");
+ },
+ rest: {
+ actions: { listWorkflowRunArtifacts, listWorkflowRuns },
+ pulls: {
+ get: async () => ({
+ data: {
+ head: { sha: "head-sha" },
+ base: { sha: "base-sha", ref: "main" },
+ },
+ }),
+ },
+ repos: { listPullRequestsAssociatedWithCommit },
+ },
+ };
+ await resolve({
+ github,
+ context: {
+ repo: { owner: "pingdotgg", repo: "t3code" },
+ payload: {
+ workflow_run: {
+ id: 2,
+ event: "pull_request",
+ workflow_id: 3,
+ head_sha: "head-sha",
+ head_branch: "feature-branch",
+ head_repository: { full_name: "pingdotgg/t3code" },
+ conclusion: "success",
+ pull_requests: [],
+ },
+ },
+ },
+ core: {
+ info: () => {},
+ setOutput: (key, value) => {
+ outputs[key] = value;
+ },
+ },
+ });
+
+ assert.equal(outputs.publish, "true");
+ assert.equal(outputs.pull_number, "5350");
+ assert.equal(outputs.pr_artifact, "true");
+ assert.equal(outputs.baseline_run_id, "1");
+ assert.equal(outputs.baseline_matches_base, "true");
+});
+
+test("does not guess when a fallback commit belongs to multiple PRs", async () => {
+ const outputs = {};
+ const listPullRequestsAssociatedWithCommit = () => {};
+ let fetchedPull = false;
+ await resolve({
+ github: {
+ paginate: async (method) => {
+ assert.equal(method, listPullRequestsAssociatedWithCommit);
+ return [5350, 5351].map((number) => ({
+ number,
+ state: "open",
+ head: {
+ sha: "head-sha",
+ ref: "feature-branch",
+ repo: { full_name: "pingdotgg/t3code" },
+ },
+ }));
+ },
+ rest: {
+ actions: {},
+ pulls: {
+ get: async () => {
+ fetchedPull = true;
+ },
+ },
+ repos: { listPullRequestsAssociatedWithCommit },
+ },
+ },
+ context: {
+ repo: { owner: "pingdotgg", repo: "t3code" },
+ payload: {
+ workflow_run: {
+ id: 2,
+ event: "pull_request",
+ workflow_id: 3,
+ head_sha: "head-sha",
+ head_branch: "feature-branch",
+ head_repository: { full_name: "pingdotgg/t3code" },
+ conclusion: "success",
+ pull_requests: [],
+ },
+ },
+ },
+ core: {
+ info: () => {},
+ setOutput: (key, value) => {
+ outputs[key] = value;
+ },
+ },
+ });
+
+ assert.equal(outputs.publish, "false");
+ assert.equal(fetchedPull, false);
+});
+
+test("does not publish a stale result after the PR head advances", async () => {
+ let listedComments = false;
+ const info = [];
+ const published = await upsertCommentForCurrentHead(
+ {
+ paginate: async () => {
+ listedComments = true;
+ return [];
+ },
+ rest: {
+ issues: {
+ listComments: () => {},
+ createComment: () => {
+ throw new Error("must not create a stale comment");
+ },
+ updateComment: () => {
+ throw new Error("must not update a stale comment");
+ },
+ },
+ pulls: {
+ get: async () => ({ data: { head: { sha: "new-head-sha" } } }),
+ },
+ },
+ },
+ { repo: { owner: "pingdotgg", repo: "t3code" } },
+ { info: (message) => info.push(message) },
+ 5350,
+ "old-head-sha",
+ "stale body",
+ );
+
+ assert.equal(published, false);
+ assert.equal(listedComments, false);
+ assert.deepEqual(info, ["Skipping stale CI result old-head-sha; PR head is new-head-sha."]);
+});
+
+test("preserves a successful result when a same-SHA rerun has no artifact", async () => {
+ let updatedComment = false;
+ const sha = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
+ const published = await upsertCommentForCurrentHead(
+ {
+ paginate: async () => [
+ {
+ id: 1,
+ user: { login: "github-actions[bot]" },
+ body: `\n`,
+ },
+ ],
+ rest: {
+ issues: {
+ listComments: () => {},
+ createComment: () => {
+ updatedComment = true;
+ },
+ updateComment: () => {
+ updatedComment = true;
+ },
+ },
+ pulls: {
+ get: async () => ({ data: { head: { sha } } }),
+ },
+ },
+ },
+ { repo: { owner: "pingdotgg", repo: "t3code" } },
+ { info: () => {} },
+ 5350,
+ sha,
+ "missing artifact warning",
+ { preserveResultSha: sha },
+ );
+
+ assert.equal(published, true);
+ assert.equal(updatedComment, false);
+});
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 1e51867cbe7..052a8c20cf7 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -84,8 +84,29 @@ jobs:
run: vp run --filter @t3tools/desktop ensure:electron
- name: Test
+ env:
+ T3CODE_TRANSFER_BUDGET_REPORT_PATH: ${{ runner.temp }}/t3code-transfer-budget.md
+ T3CODE_TRANSFER_BUDGET_RESULT_PATH: ${{ runner.temp }}/thread-transfer-result.json
run: vp run test
+ - name: Publish transfer budget report
+ if: always()
+ run: |
+ if test -f "${{ runner.temp }}/t3code-transfer-budget.md"; then
+ tee -a "$GITHUB_STEP_SUMMARY" < "${{ runner.temp }}/t3code-transfer-budget.md"
+ else
+ echo "Transfer budget report was not produced." >> "$GITHUB_STEP_SUMMARY"
+ fi
+
+ - name: Upload thread transfer result
+ if: always()
+ uses: actions/upload-artifact@v7
+ with:
+ name: thread-transfer-results
+ path: ${{ runner.temp }}/thread-transfer-result.json
+ if-no-files-found: ignore
+ retention-days: 30
+
- name: Test resource monitor
run: cargo test --locked --manifest-path native/resource-monitor/Cargo.toml
diff --git a/.github/workflows/thread-transfer-report.yml b/.github/workflows/thread-transfer-report.yml
new file mode 100644
index 00000000000..23eec72923b
--- /dev/null
+++ b/.github/workflows/thread-transfer-report.yml
@@ -0,0 +1,75 @@
+name: Thread Transfer Report
+
+on:
+ workflow_run:
+ workflows: [CI]
+ types: [completed]
+
+permissions:
+ actions: read
+ contents: read
+ pull-requests: write
+
+jobs:
+ publish:
+ name: Publish PR comment
+ if: github.event.workflow_run.event == 'pull_request'
+ runs-on: ubuntu-24.04
+ concurrency:
+ group: thread-transfer-report-${{ github.event.workflow_run.pull_requests[0].number || github.event.workflow_run.id }}
+ cancel-in-progress: true
+ steps:
+ # workflow_run has a write-capable token even for fork PRs. Only load the
+ # publisher from the trusted default branch and never execute PR code.
+ - name: Checkout trusted publisher
+ uses: actions/checkout@v6
+ with:
+ ref: ${{ github.event.repository.default_branch }}
+ sparse-checkout: .github/scripts
+
+ - name: Test trusted publisher
+ run: node --test .github/scripts/thread-transfer-report.test.cjs
+
+ - id: resolve
+ name: Resolve PR and baseline artifacts
+ uses: actions/github-script@v8
+ with:
+ script: |
+ const reporter = require("./.github/scripts/thread-transfer-report.cjs");
+ await reporter.resolve({ github, context, core });
+
+ - name: Download PR result
+ if: steps.resolve.outputs.publish == 'true' && steps.resolve.outputs.pr_artifact == 'true'
+ uses: actions/download-artifact@v8
+ with:
+ name: thread-transfer-results
+ path: ${{ runner.temp }}/thread-transfer/pr
+ github-token: ${{ secrets.GITHUB_TOKEN }}
+ run-id: ${{ steps.resolve.outputs.pr_run_id }}
+
+ - name: Download main baseline
+ if: steps.resolve.outputs.publish == 'true' && steps.resolve.outputs.baseline_artifact == 'true'
+ uses: actions/download-artifact@v8
+ with:
+ name: thread-transfer-results
+ path: ${{ runner.temp }}/thread-transfer/main
+ github-token: ${{ secrets.GITHUB_TOKEN }}
+ run-id: ${{ steps.resolve.outputs.baseline_run_id }}
+
+ - name: Update thread transfer comment
+ if: steps.resolve.outputs.publish == 'true'
+ uses: actions/github-script@v8
+ env:
+ PR_NUMBER: ${{ steps.resolve.outputs.pull_number }}
+ PR_SHA: ${{ steps.resolve.outputs.pr_sha }}
+ PR_CONCLUSION: ${{ steps.resolve.outputs.pr_conclusion }}
+ PR_RUN_ID: ${{ steps.resolve.outputs.pr_run_id }}
+ PR_RESULT_DIR: ${{ runner.temp }}/thread-transfer/pr
+ BASELINE_SHA: ${{ steps.resolve.outputs.baseline_sha }}
+ BASELINE_MATCHES_BASE: ${{ steps.resolve.outputs.baseline_matches_base }}
+ BASELINE_RUN_ID: ${{ steps.resolve.outputs.baseline_run_id }}
+ BASELINE_RESULT_DIR: ${{ runner.temp }}/thread-transfer/main
+ with:
+ script: |
+ const reporter = require("./.github/scripts/thread-transfer-report.cjs");
+ await reporter.publish({ github, context, core });
diff --git a/apps/server/integration/NetworkTransferMeasurement.integration.ts b/apps/server/integration/NetworkTransferMeasurement.integration.ts
new file mode 100644
index 00000000000..75714d1519e
--- /dev/null
+++ b/apps/server/integration/NetworkTransferMeasurement.integration.ts
@@ -0,0 +1,177 @@
+// @effect-diagnostics nodeBuiltinImport:off - Measures the real Node HTTP and WebSocket transports.
+import * as NodeHttp from "node:http";
+import * as NodeZlib from "node:zlib";
+
+import * as NodeSocket from "@effect/platform-node/NodeSocket";
+import { WsRpcGroup } from "@t3tools/contracts";
+import * as Effect from "effect/Effect";
+import * as Layer from "effect/Layer";
+import * as Schema from "effect/Schema";
+import { RpcClient, RpcSerialization } from "effect/unstable/rpc";
+import * as Socket from "effect/unstable/socket/Socket";
+
+export class TransferHttpRequestError extends Schema.TaggedErrorClass()(
+ "TransferHttpRequestError",
+ {
+ url: Schema.String,
+ cause: Schema.Defect(),
+ },
+) {}
+
+export interface HttpTransferMeasurement {
+ readonly status: number;
+ readonly contentEncoding: string | null;
+ readonly encodedBody: Uint8Array;
+ readonly encodedBodyBytes: number;
+ readonly decodedBody: Uint8Array;
+ readonly decodedBodyBytes: number;
+ /** HTTP response bytes read from the socket, including status line and headers. */
+ readonly wireBytes: number;
+}
+
+export const measureHttpGet = Effect.fn("TransferBudget.measureHttpGet")(function* (input: {
+ readonly url: string;
+ readonly headers?: Readonly>;
+}) {
+ return yield* Effect.tryPromise({
+ try: () =>
+ new Promise((resolve, reject) => {
+ let socketBytesBeforeResponse = 0;
+ const request = NodeHttp.get(
+ input.url,
+ {
+ agent: false,
+ headers: {
+ "accept-encoding": "gzip",
+ connection: "close",
+ ...input.headers,
+ },
+ },
+ (response) => {
+ const chunks: Buffer[] = [];
+ response.on("data", (chunk: Buffer) => chunks.push(chunk));
+ response.once("error", reject);
+ response.once("end", () => {
+ try {
+ const encodedBody = Buffer.concat(chunks);
+ const header = response.headers["content-encoding"];
+ const contentEncoding = Array.isArray(header)
+ ? (header[0] ?? null)
+ : (header ?? null);
+ const decodedBody =
+ contentEncoding === "gzip" ? NodeZlib.gunzipSync(encodedBody) : encodedBody;
+ resolve({
+ status: response.statusCode ?? 0,
+ contentEncoding,
+ encodedBody,
+ encodedBodyBytes: encodedBody.byteLength,
+ decodedBody,
+ decodedBodyBytes: decodedBody.byteLength,
+ wireBytes: Math.max(0, response.socket.bytesRead - socketBytesBeforeResponse),
+ });
+ } catch (cause) {
+ reject(cause);
+ }
+ });
+ },
+ );
+ request.once("socket", (socket) => {
+ socketBytesBeforeResponse = socket.bytesRead;
+ });
+ request.once("error", reject);
+ request.setTimeout(10_000, () => {
+ request.destroy(new Error(`Timed out reading ${input.url}`));
+ });
+ }),
+ catch: (cause) => new TransferHttpRequestError({ url: input.url, cause }),
+ });
+});
+
+export interface WebSocketTransferTotals {
+ readonly wireBytes: number;
+ readonly decodedBytes: number;
+ readonly messages: number;
+}
+
+export interface WebSocketTransferRecorder {
+ readonly connect: (
+ url: string,
+ protocols: string | string[] | undefined,
+ cookie: string,
+ ) => globalThis.WebSocket;
+ readonly totals: () => WebSocketTransferTotals;
+ readonly negotiatedExtensions: () => string;
+}
+
+interface NodeWebSocketWithTransport extends NodeSocket.NodeWS.WebSocket {
+ readonly _socket?: {
+ readonly bytesRead: number;
+ };
+}
+
+function rawDataBytes(data: NodeSocket.NodeWS.RawData): number {
+ if (Array.isArray(data)) {
+ return data.reduce((total, chunk) => total + chunk.byteLength, 0);
+ }
+ return data.byteLength;
+}
+
+export function makeWebSocketTransferRecorder(): WebSocketTransferRecorder {
+ let socket: NodeWebSocketWithTransport | null = null;
+ let decodedBytes = 0;
+ let messages = 0;
+
+ return {
+ connect: (url, protocols, cookie) => {
+ const nextSocket = new NodeSocket.NodeWS.WebSocket(url, protocols, {
+ headers: { cookie },
+ perMessageDeflate: true,
+ }) as NodeWebSocketWithTransport;
+ socket = nextSocket;
+ nextSocket.on("message", (data) => {
+ const bytes = rawDataBytes(data);
+ decodedBytes += bytes;
+ messages += 1;
+ });
+ return nextSocket as unknown as globalThis.WebSocket;
+ },
+ totals: () => ({
+ wireBytes: socket?._socket?.bytesRead ?? 0,
+ decodedBytes,
+ messages,
+ }),
+ negotiatedExtensions: () => socket?.extensions ?? "",
+ };
+}
+
+export function transferDelta(
+ start: WebSocketTransferTotals,
+ end: WebSocketTransferTotals,
+): WebSocketTransferTotals {
+ return {
+ wireBytes: Math.max(0, end.wireBytes - start.wireBytes),
+ decodedBytes: Math.max(0, end.decodedBytes - start.decodedBytes),
+ messages: Math.max(0, end.messages - start.messages),
+ };
+}
+
+export function countingWsRpcProtocolLayer(input: {
+ readonly url: string;
+ readonly cookie: string;
+ readonly recorder: WebSocketTransferRecorder;
+}) {
+ const webSocketConstructorLayer = Layer.succeed(Socket.WebSocketConstructor, (url, protocols) =>
+ input.recorder.connect(url, protocols, input.cookie),
+ );
+ return RpcClient.layerProtocolSocket().pipe(
+ Layer.provide(
+ Socket.layerWebSocket(input.url, { openTimeout: "10 seconds" }).pipe(
+ Layer.provide(webSocketConstructorLayer),
+ ),
+ ),
+ Layer.provide(RpcSerialization.layerJson),
+ );
+}
+
+export const makeCountingWsRpcClient = RpcClient.make(WsRpcGroup);
+export type CountingWsRpcClient = Effect.Success;
diff --git a/apps/server/integration/OrchestrationEngineHarness.integration.ts b/apps/server/integration/OrchestrationEngineHarness.integration.ts
index c3f77d677b1..d192cbeac8e 100644
--- a/apps/server/integration/OrchestrationEngineHarness.integration.ts
+++ b/apps/server/integration/OrchestrationEngineHarness.integration.ts
@@ -55,6 +55,8 @@ import { RuntimeReceiptBusTest } from "../src/orchestration/Layers/RuntimeReceip
import { OrchestrationReactorLive } from "../src/orchestration/Layers/OrchestrationReactor.ts";
import { ProviderCommandReactorLive } from "../src/orchestration/Layers/ProviderCommandReactor.ts";
import { ProviderRuntimeIngestionLive } from "../src/orchestration/Layers/ProviderRuntimeIngestion.ts";
+import { CheckpointReactor } from "../src/orchestration/Services/CheckpointReactor.ts";
+import { ProviderRuntimeIngestionService } from "../src/orchestration/Services/ProviderRuntimeIngestion.ts";
import {
OrchestrationEngineService,
type OrchestrationEngineShape,
@@ -218,6 +220,8 @@ export interface OrchestrationIntegrationHarness {
timeoutMs?: number,
): Effect.Effect;
};
+ readonly drainProviderRuntime: Effect.Effect;
+ readonly drainCheckpointReactor: Effect.Effect;
readonly dispose: Effect.Effect;
}
@@ -392,6 +396,13 @@ export const makeOrchestrationIntegrationHarness = (
const reactor = yield* tryRuntimePromise("load OrchestrationReactor service", () =>
runtime.runPromise(Effect.service(OrchestrationReactor)),
).pipe(Effect.orDie);
+ const providerRuntimeIngestion = yield* tryRuntimePromise(
+ "load ProviderRuntimeIngestion service",
+ () => runtime.runPromise(Effect.service(ProviderRuntimeIngestionService)),
+ ).pipe(Effect.orDie);
+ const checkpointReactor = yield* tryRuntimePromise("load CheckpointReactor service", () =>
+ runtime.runPromise(Effect.service(CheckpointReactor)),
+ ).pipe(Effect.orDie);
const snapshotQuery = yield* tryRuntimePromise("load ProjectionSnapshotQuery service", () =>
runtime.runPromise(Effect.service(ProjectionSnapshotQuery)),
).pipe(Effect.orDie);
@@ -556,6 +567,8 @@ export const makeOrchestrationIntegrationHarness = (
waitForDomainEvent,
waitForPendingApproval,
waitForReceipt,
+ drainProviderRuntime: providerRuntimeIngestion.drain,
+ drainCheckpointReactor: checkpointReactor.drain,
dispose,
} satisfies OrchestrationIntegrationHarness;
});
diff --git a/apps/server/integration/TestProviderAdapter.integration.ts b/apps/server/integration/TestProviderAdapter.integration.ts
index 0e64699de97..095cca4e5e7 100644
--- a/apps/server/integration/TestProviderAdapter.integration.ts
+++ b/apps/server/integration/TestProviderAdapter.integration.ts
@@ -11,7 +11,6 @@ import {
ProviderDriverKind,
} from "@t3tools/contracts";
import * as Effect from "effect/Effect";
-import * as Crypto from "effect/Crypto";
import * as Queue from "effect/Queue";
import * as Stream from "effect/Stream";
@@ -226,9 +225,9 @@ function missingSessionEffect(
export const makeTestProviderAdapterHarness = (options?: MakeTestProviderAdapterHarnessOptions) =>
Effect.gen(function* () {
const provider = options?.provider ?? ProviderDriverKind.make("codex");
- const crypto = yield* Crypto.Crypto;
const runtimeEvents = yield* Queue.unbounded();
let sessionCount = 0;
+ let eventCount = 0;
const sessions = new Map();
const queuedResponsesForNextSession: TestTurnResponse[] = [];
const interruptCallsBySession = new Map>();
@@ -242,18 +241,10 @@ export const makeTestProviderAdapterHarness = (options?: MakeTestProviderAdapter
>();
const emit = (event: ProviderRuntimeEvent) => Queue.offer(runtimeEvents, event);
- const randomUUIDv4 = (threadId: ThreadId) =>
- crypto.randomUUIDv4.pipe(
- Effect.mapError(
- (cause) =>
- new ProviderAdapterValidationError({
- provider,
- operation: "crypto/randomUUIDv4",
- issue: `Failed to generate test runtime identifier for thread '${threadId}'.`,
- cause,
- }),
- ),
- );
+ const nextEventId = (threadId: ThreadId) => {
+ eventCount += 1;
+ return EventId.make(`test-provider:${provider}:${threadId}:${eventCount}`);
+ };
const startSession: ProviderAdapterShape["startSession"] = (input) =>
Effect.gen(function* () {
@@ -322,7 +313,7 @@ export const makeTestProviderAdapterHarness = (options?: MakeTestProviderAdapter
for (const fixtureEvent of response.events) {
const rawEvent: Record = {
...(fixtureEvent as Record),
- eventId: yield* randomUUIDv4(input.threadId),
+ eventId: nextEventId(input.threadId),
provider,
sessionId: RuntimeSessionId.make(String(input.threadId)),
};
@@ -379,7 +370,7 @@ export const makeTestProviderAdapterHarness = (options?: MakeTestProviderAdapter
if (deferredTurnCompletedEvents.length === 0) {
yield* emit({
type: "turn.completed",
- eventId: EventId.make(yield* randomUUIDv4(input.threadId)),
+ eventId: nextEventId(input.threadId),
provider,
createdAt: nowIso(),
threadId: state.snapshot.threadId,
diff --git a/apps/server/integration/TransferBudgetReport.integration.ts b/apps/server/integration/TransferBudgetReport.integration.ts
new file mode 100644
index 00000000000..f773b5b8b84
--- /dev/null
+++ b/apps/server/integration/TransferBudgetReport.integration.ts
@@ -0,0 +1,212 @@
+import type { ProviderDriverKind } from "@t3tools/contracts";
+
+import type {
+ HttpTransferMeasurement,
+ WebSocketTransferTotals,
+} from "./NetworkTransferMeasurement.integration.ts";
+import {
+ TRANSFER_HISTORY_MCP_RESULT_BYTES,
+ TRANSFER_HISTORY_TOOLS_PER_TURN,
+ TRANSFER_HISTORY_TURN_COUNT,
+ TRANSFER_MEASURED_MCP_RESULT_BYTES,
+ TRANSFER_MEASURED_TOOLS,
+} from "./fixtures/transferBudget.ts";
+
+export interface TransferBudgetRun {
+ readonly provider: ProviderDriverKind;
+ readonly threadSnapshot: HttpTransferMeasurement;
+ readonly measuredTurnWebSocket: WebSocketTransferTotals;
+}
+
+interface ProviderTransferBudget {
+ readonly totalWireBytes: number;
+ readonly threadSnapshotWireBytes: number;
+ readonly measuredTurnWebSocketWireBytes: number;
+ readonly measuredTurnWebSocketDecodedBytes: number;
+ readonly measuredTurnWebSocketMessages: number;
+}
+
+// These caps leave roughly 30% headroom above the client projection of the
+// deterministic 9 MB retained-result fixture. Full MCP results stay in
+// persistence, so accidentally shipping them again exceeds these caps by
+// orders of magnitude. The CI report preserves exact values for review.
+const TRANSFER_BUDGET = {
+ totalWireBytes: 15_500,
+ threadSnapshotWireBytes: 7_500,
+ measuredTurnWebSocketWireBytes: 8_000,
+ measuredTurnWebSocketDecodedBytes: 68_000,
+ measuredTurnWebSocketMessages: 21,
+} satisfies ProviderTransferBudget;
+
+export const TRANSFER_BUDGETS: Readonly> = {
+ codex: TRANSFER_BUDGET,
+ claudeAgent: TRANSFER_BUDGET,
+};
+
+function totalWireBytes(run: TransferBudgetRun): number {
+ return run.threadSnapshot.wireBytes + run.measuredTurnWebSocket.wireBytes;
+}
+
+function observedTransfer(run: TransferBudgetRun) {
+ return {
+ totalWireBytes: totalWireBytes(run),
+ threadSnapshotWireBytes: run.threadSnapshot.wireBytes,
+ threadSnapshotDecodedBytes: run.threadSnapshot.decodedBodyBytes,
+ measuredTurnWebSocketWireBytes: run.measuredTurnWebSocket.wireBytes,
+ measuredTurnWebSocketDecodedBytes: run.measuredTurnWebSocket.decodedBytes,
+ measuredTurnWebSocketMessages: run.measuredTurnWebSocket.messages,
+ };
+}
+
+/** Machine-readable input for the trusted PR comment publisher. */
+export function formatTransferBudgetResult(runs: ReadonlyArray): string {
+ const providers = Object.fromEntries(
+ runs.flatMap((run) => {
+ const ceiling = TRANSFER_BUDGETS[run.provider];
+ return ceiling ? [[run.provider, { observed: observedTransfer(run), ceiling }]] : [];
+ }),
+ );
+
+ return `${JSON.stringify(
+ {
+ schemaVersion: 1,
+ scenario: {
+ id: "thread-transfer-v1",
+ historyTurns: TRANSFER_HISTORY_TURN_COUNT,
+ historyCommandToolsPerTurn: TRANSFER_HISTORY_TOOLS_PER_TURN,
+ historyMcpResultBytes: TRANSFER_HISTORY_MCP_RESULT_BYTES,
+ measuredCommandTools: TRANSFER_MEASURED_TOOLS,
+ measuredMcpResultBytes: TRANSFER_MEASURED_MCP_RESULT_BYTES,
+ },
+ providers,
+ },
+ null,
+ 2,
+ )}\n`;
+}
+
+function formatBytes(bytes: number): string {
+ if (bytes < 1_024) return `${bytes} B`;
+ if (bytes >= 1_024 * 1_024) {
+ return `${(bytes / 1_024 / 1_024).toFixed(2)} MiB (${bytes.toLocaleString("en-US")} B)`;
+ }
+ return `${(bytes / 1_024).toFixed(1)} KiB (${bytes.toLocaleString("en-US")} B)`;
+}
+
+function row(
+ provider: ProviderDriverKind,
+ phase: string,
+ metric: string,
+ observed: number,
+ maximum: number,
+ format: (value: number) => string = formatBytes,
+): string {
+ const status = observed <= maximum ? "PASS" : "FAIL";
+ return `| ${provider} | ${phase} | ${metric} | ${format(observed)} | ${format(maximum)} | ${status} |`;
+}
+
+export function transferBudgetViolations(runs: ReadonlyArray): string[] {
+ const violations: string[] = [];
+ for (const run of runs) {
+ const budget = TRANSFER_BUDGETS[run.provider];
+ if (!budget) {
+ violations.push(`${run.provider}: no transfer budget is configured`);
+ continue;
+ }
+ const checks = [
+ ["total thread wire bytes", totalWireBytes(run), budget.totalWireBytes],
+ ["thread snapshot wire bytes", run.threadSnapshot.wireBytes, budget.threadSnapshotWireBytes],
+ [
+ "measured-turn WebSocket wire bytes",
+ run.measuredTurnWebSocket.wireBytes,
+ budget.measuredTurnWebSocketWireBytes,
+ ],
+ [
+ "measured-turn WebSocket decoded bytes",
+ run.measuredTurnWebSocket.decodedBytes,
+ budget.measuredTurnWebSocketDecodedBytes,
+ ],
+ [
+ "measured-turn WebSocket messages",
+ run.measuredTurnWebSocket.messages,
+ budget.measuredTurnWebSocketMessages,
+ ],
+ ] as const;
+ for (const [metric, observed, maximum] of checks) {
+ if (observed > maximum) {
+ violations.push(`${run.provider}: ${metric} was ${observed}, maximum ${maximum}`);
+ }
+ }
+ }
+ return violations;
+}
+
+export function formatTransferBudgetReport(runs: ReadonlyArray): string {
+ const lines = [
+ "# T3 Code thread transfer budget",
+ "",
+ "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.",
+ `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.`,
+ "",
+ "| Provider | Total thread wire | Budget | Result |",
+ "| --- | ---: | ---: | --- |",
+ ...runs.flatMap((run) => {
+ const budget = TRANSFER_BUDGETS[run.provider];
+ if (!budget) return [];
+ const observed = observedTransfer(run).totalWireBytes;
+ return [
+ `| ${run.provider} | ${formatBytes(observed)} | ${formatBytes(budget.totalWireBytes)} | ${observed <= budget.totalWireBytes ? "PASS" : "FAIL"} |`,
+ ];
+ }),
+ "",
+ "## Detailed measurements",
+ "",
+ "| Provider | Phase | Metric | Observed | Budget | Result |",
+ "| --- | --- | --- | ---: | ---: | --- |",
+ ];
+
+ for (const run of runs) {
+ const budget = TRANSFER_BUDGETS[run.provider];
+ if (!budget) continue;
+ lines.push(
+ row(
+ run.provider,
+ "thread snapshot",
+ "HTTP wire",
+ run.threadSnapshot.wireBytes,
+ budget.threadSnapshotWireBytes,
+ ),
+ row(
+ run.provider,
+ "measured turn",
+ "WebSocket wire",
+ run.measuredTurnWebSocket.wireBytes,
+ budget.measuredTurnWebSocketWireBytes,
+ ),
+ row(
+ run.provider,
+ "measured turn",
+ "WebSocket decoded",
+ run.measuredTurnWebSocket.decodedBytes,
+ budget.measuredTurnWebSocketDecodedBytes,
+ ),
+ row(
+ run.provider,
+ "measured turn",
+ "WebSocket messages",
+ run.measuredTurnWebSocket.messages,
+ budget.measuredTurnWebSocketMessages,
+ String,
+ ),
+ );
+ }
+
+ lines.push("", "## Compression diagnostics", "");
+ for (const run of runs) {
+ lines.push(
+ `- ${run.provider}: thread snapshot ${formatBytes(run.threadSnapshot.decodedBodyBytes)} decoded to ${formatBytes(run.threadSnapshot.encodedBodyBytes)} gzip.`,
+ );
+ }
+
+ return `${lines.join("\n")}\n`;
+}
diff --git a/apps/server/integration/TransferBudgetScenario.integration.ts b/apps/server/integration/TransferBudgetScenario.integration.ts
new file mode 100644
index 00000000000..77dfbc1dd7f
--- /dev/null
+++ b/apps/server/integration/TransferBudgetScenario.integration.ts
@@ -0,0 +1,128 @@
+import {
+ CommandId,
+ defaultInstanceIdForDriver,
+ DEFAULT_MODEL,
+ DEFAULT_MODEL_BY_PROVIDER,
+ DEFAULT_PROVIDER_INTERACTION_MODE,
+ MessageId,
+ ProjectId,
+ ProviderDriverKind,
+ ThreadId,
+} from "@t3tools/contracts";
+import * as Effect from "effect/Effect";
+
+import type { TurnProcessingQuiescedReceipt } from "../src/orchestration/Services/RuntimeReceiptBus.ts";
+import type { OrchestrationIntegrationHarness } from "./OrchestrationEngineHarness.integration.ts";
+import {
+ expectedRecordedAssistantText,
+ makeRecordedTransferTurn,
+ TRANSFER_HISTORY_TURN_COUNT,
+} from "./fixtures/transferBudget.ts";
+
+export const TRANSFER_PROJECT_ID = ProjectId.make("transfer-budget-project");
+export const TRANSFER_THREAD_ID = ThreadId.make("transfer-budget-thread");
+export const TRANSFER_MEASURED_TURN_INDEX = TRANSFER_HISTORY_TURN_COUNT;
+
+export function transferModelSelection(provider: ProviderDriverKind) {
+ return {
+ instanceId: defaultInstanceIdForDriver(provider),
+ model: DEFAULT_MODEL_BY_PROVIDER[provider] ?? DEFAULT_MODEL,
+ };
+}
+
+function turnTimestamp(turnIndex: number): string {
+ return `2026-06-01T00:${String(turnIndex).padStart(2, "0")}:00.000Z`;
+}
+
+export const TRANSFER_MEASURED_TURN_CREATED_AT = turnTimestamp(TRANSFER_MEASURED_TURN_INDEX);
+
+const waitForTurnQuiesced = Effect.fn("TransferBudget.waitForTurnQuiesced")(function* (
+ harness: OrchestrationIntegrationHarness,
+ checkpointTurnCount: number,
+) {
+ const receipt = yield* harness.waitForReceipt(
+ (receipt): receipt is TurnProcessingQuiescedReceipt =>
+ receipt.type === "turn.processing.quiesced" &&
+ receipt.threadId === TRANSFER_THREAD_ID &&
+ receipt.checkpointTurnCount === checkpointTurnCount,
+ );
+ yield* harness.drainProviderRuntime;
+ yield* harness.drainCheckpointReactor;
+ return receipt;
+});
+
+export const seedTransferBudgetHistory = Effect.fn("TransferBudget.seedHistory")(function* (
+ harness: OrchestrationIntegrationHarness,
+ provider: ProviderDriverKind,
+) {
+ if (!harness.adapterHarness) {
+ return yield* Effect.die(new Error("Transfer budget history requires the replay adapter."));
+ }
+
+ const modelSelection = transferModelSelection(provider);
+ yield* harness.engine.dispatch({
+ type: "project.create",
+ commandId: CommandId.make(`transfer:${provider}:project-create`),
+ projectId: TRANSFER_PROJECT_ID,
+ title: "Transfer Budget Project",
+ workspaceRoot: harness.workspaceDir,
+ defaultModelSelection: modelSelection,
+ createdAt: turnTimestamp(0),
+ });
+ yield* harness.engine.dispatch({
+ type: "thread.create",
+ commandId: CommandId.make(`transfer:${provider}:thread-create`),
+ threadId: TRANSFER_THREAD_ID,
+ projectId: TRANSFER_PROJECT_ID,
+ title: `${provider} transfer history`,
+ modelSelection,
+ runtimeMode: "approval-required",
+ interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
+ branch: "main",
+ worktreePath: harness.workspaceDir,
+ createdAt: turnTimestamp(0),
+ });
+
+ for (let turnIndex = 0; turnIndex < TRANSFER_HISTORY_TURN_COUNT; turnIndex += 1) {
+ const response = makeRecordedTransferTurn(provider, turnIndex);
+ if (turnIndex === 0) {
+ yield* harness.adapterHarness.queueTurnResponseForNextSession(response);
+ } else {
+ yield* harness.adapterHarness.queueTurnResponse(TRANSFER_THREAD_ID, response);
+ }
+
+ yield* harness.engine.dispatch({
+ type: "thread.turn.start",
+ commandId: CommandId.make(`transfer:${provider}:turn:${turnIndex + 1}`),
+ threadId: TRANSFER_THREAD_ID,
+ message: {
+ messageId: MessageId.make(`transfer-user-${turnIndex + 1}`),
+ role: "user",
+ text: `Inspect transfer behavior for historical turn ${turnIndex + 1}.`,
+ attachments: [],
+ },
+ modelSelection,
+ runtimeMode: "approval-required",
+ interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
+ createdAt: turnTimestamp(turnIndex),
+ });
+ yield* waitForTurnQuiesced(harness, turnIndex + 1);
+ }
+});
+
+export const queueMeasuredTransferTurn = Effect.fn("TransferBudget.queueMeasuredTurn")(function* (
+ harness: OrchestrationIntegrationHarness,
+ provider: ProviderDriverKind,
+) {
+ if (!harness.adapterHarness) {
+ return yield* Effect.die(new Error("Transfer budget measurement requires the replay adapter."));
+ }
+ const response = makeRecordedTransferTurn(provider, TRANSFER_MEASURED_TURN_INDEX);
+ yield* harness.adapterHarness.queueTurnResponse(TRANSFER_THREAD_ID, response);
+});
+
+export function expectedMeasuredAssistantText(provider: ProviderDriverKind): string {
+ return expectedRecordedAssistantText(provider, TRANSFER_MEASURED_TURN_INDEX);
+}
+
+export { TRANSFER_HISTORY_TURN_COUNT, waitForTurnQuiesced };
diff --git a/apps/server/integration/fixtures/transferBudget.ts b/apps/server/integration/fixtures/transferBudget.ts
new file mode 100644
index 00000000000..d3567d386b9
--- /dev/null
+++ b/apps/server/integration/fixtures/transferBudget.ts
@@ -0,0 +1,372 @@
+import { EventId, ProviderDriverKind } from "@t3tools/contracts";
+
+import type {
+ FixtureProviderRuntimeEvent,
+ TestTurnResponse,
+} from "../TestProviderAdapter.integration.ts";
+
+const FIXTURE_THREAD_ID = "transfer-budget-thread";
+const FIXTURE_TURN_ID = "transfer-budget-turn";
+
+export const TRANSFER_HISTORY_TURN_COUNT = 10;
+export const TRANSFER_HISTORY_TOOLS_PER_TURN = 5;
+export const TRANSFER_MEASURED_TOOLS = 20;
+export const TRANSFER_HISTORY_MCP_RESULT_BYTES = 900_000;
+export const TRANSFER_MEASURED_MCP_RESULT_BYTES = 1_100_000;
+
+const sourceModules = [
+ "connection/session.ts",
+ "connection/supervisor.ts",
+ "rpc/client.ts",
+ "rpc/protocol.ts",
+ "state/threads.ts",
+ "state/threadReducer.ts",
+ "state/threadSnapshotHttp.ts",
+ "orchestration/http.ts",
+ "orchestration/Normalizer.ts",
+ "orchestration/ActivityPayloadProjection.ts",
+ "provider/ProviderService.ts",
+ "provider/ProviderRuntimeIngestion.ts",
+ "persistence/ProjectionSnapshotQuery.ts",
+ "persistence/OrchestrationEventStore.ts",
+ "checkpointing/CheckpointStore.ts",
+ "checkpointing/CheckpointDiffQuery.ts",
+ "server.ts",
+] as const;
+
+function fixtureTimestamp(turnIndex: number, eventIndex: number): string {
+ const minute = String(turnIndex).padStart(2, "0");
+ const second = String(Math.floor(eventIndex / 1_000)).padStart(2, "0");
+ const millisecond = String(eventIndex % 1_000).padStart(3, "0");
+ return `2026-06-01T00:${minute}:${second}.${millisecond}Z`;
+}
+
+function mix(value: number): number {
+ let mixed = value | 0;
+ mixed ^= mixed >>> 16;
+ mixed = Math.imul(mixed, 0x7feb352d);
+ mixed ^= mixed >>> 15;
+ mixed = Math.imul(mixed, 0x846ca68b);
+ mixed ^= mixed >>> 16;
+ return mixed >>> 0;
+}
+
+function digest(seed: number): string {
+ return [0, 1, 2, 3]
+ .map((offset) =>
+ mix(seed + offset * 0x9e3779b9)
+ .toString(16)
+ .padStart(8, "0"),
+ )
+ .join("");
+}
+
+/** Produces safe, deterministic output with enough entropy to exercise gzip. */
+function diagnosticOutput(input: {
+ readonly provider: ProviderDriverKind;
+ readonly turnIndex: number;
+ readonly toolIndex: number;
+ readonly targetBytes: number;
+}): string {
+ const chunks: string[] = [];
+ const providerSeed = input.provider === "codex" ? 0x43_4f_44_45 : 0x43_4c_41_55;
+ let length = 0;
+ let lineIndex = 0;
+
+ while (length < input.targetBytes) {
+ const modulePath = sourceModules[(input.toolIndex + lineIndex) % sourceModules.length];
+ const seed =
+ providerSeed + input.turnIndex * 100_003 + input.toolIndex * 10_007 + lineIndex * 101;
+ const line =
+ `${String(lineIndex + 1).padStart(6, "0")} ${modulePath} ` +
+ `operation=project-transfer-${input.turnIndex + 1}-${input.toolIndex + 1} ` +
+ `cursor=${mix(seed)} digest=${digest(seed)} status=completed\n`;
+ chunks.push(line);
+ length += line.length;
+ lineIndex += 1;
+ }
+
+ return chunks.join("").slice(0, input.targetBytes);
+}
+
+function assistantChunks(provider: ProviderDriverKind, turnIndex: number): ReadonlyArray {
+ const providerName = provider === "codex" ? "Codex" : "Claude";
+ const paragraphs: string[] = [
+ `I traced the ${providerName} request through the environment connection and orchestration layers. `,
+ ];
+ let paragraphIndex = 0;
+ while (paragraphs.join("").length < 4_096) {
+ const modulePath = sourceModules[paragraphIndex % sourceModules.length];
+ paragraphs.push(
+ `Pass ${paragraphIndex + 1} reviewed ${modulePath} for turn ${turnIndex + 1}. ` +
+ "The shell cursor stayed monotonic, the thread snapshot remained resumable, and the client received only incremental events. ",
+ );
+ paragraphIndex += 1;
+ }
+ const text = paragraphs.join("").slice(0, 4_096);
+ return Array.from({ length: Math.ceil(text.length / 256) }, (_, index) =>
+ text.slice(index * 256, (index + 1) * 256),
+ );
+}
+
+export function expectedRecordedAssistantText(
+ provider: ProviderDriverKind,
+ turnIndex: number,
+): string {
+ return assistantChunks(provider, turnIndex).join("");
+}
+
+function unifiedDiff(provider: ProviderDriverKind, turnIndex: number): string {
+ const lines = sourceModules
+ .slice(0, 8)
+ .flatMap((modulePath, index) => [
+ `diff --git a/${modulePath} b/${modulePath}`,
+ `--- a/${modulePath}`,
+ `+++ b/${modulePath}`,
+ `@@ -${index + 1},2 +${index + 1},3 @@`,
+ ` const provider = "${provider}";`,
+ `+const transferTurn = ${turnIndex + 1};`,
+ `+const transferSample = ${1_500 + index * 97};`,
+ ]);
+ return lines.join("\n");
+}
+
+function baseEvent(
+ provider: ProviderDriverKind,
+ turnIndex: number,
+ eventIndex: number,
+): Pick {
+ return {
+ eventId: EventId.make(`recorded:${provider}:${turnIndex}:${eventIndex}`),
+ provider,
+ createdAt: fixtureTimestamp(turnIndex, eventIndex),
+ threadId: FIXTURE_THREAD_ID,
+ };
+}
+
+/**
+ * Synthetic canonical events calibrated from heavy local Codex and Claude
+ * threads. Ten historical turns produce 9 MB of retained MCP results without
+ * committing user content. Command output is intentionally modest because the
+ * client projection strips it.
+ */
+export function makeRecordedTransferTurn(
+ provider: ProviderDriverKind,
+ turnIndex: number,
+): TestTurnResponse {
+ const measuredTurn = turnIndex >= TRANSFER_HISTORY_TURN_COUNT;
+ const toolCount = measuredTurn ? TRANSFER_MEASURED_TOOLS : TRANSFER_HISTORY_TOOLS_PER_TURN;
+ const turnId = `${FIXTURE_TURN_ID}-${turnIndex + 1}`;
+ const events: FixtureProviderRuntimeEvent[] = [];
+ let eventIndex = 0;
+
+ events.push({
+ type: "turn.started",
+ ...baseEvent(provider, turnIndex, eventIndex++),
+ turnId,
+ payload: {
+ model: provider === "codex" ? "gpt-5.4" : "claude-opus-4-1",
+ effort: provider === "codex" ? "high" : "default",
+ },
+ });
+
+ for (let toolIndex = 0; toolIndex < toolCount; toolIndex += 1) {
+ const itemId = `tool-${turnIndex + 1}-${toolIndex + 1}`;
+ const command =
+ provider === "codex"
+ ? `vp test transfer-budget-${toolIndex + 1}`
+ : `review transfer budget ${toolIndex + 1}`;
+ events.push(
+ {
+ type: "item.started",
+ ...baseEvent(provider, turnIndex, eventIndex++),
+ turnId,
+ itemId,
+ payload: {
+ itemType: "command_execution",
+ status: "inProgress",
+ title: `Inspect transfer path ${toolIndex + 1}`,
+ detail: "Inspecting the HTTP snapshot and WebSocket projection boundaries.",
+ data: {
+ threadId: FIXTURE_THREAD_ID,
+ turnId,
+ startedAtMs: turnIndex * 60_000 + eventIndex,
+ item: {
+ id: itemId,
+ type: "commandExecution",
+ command,
+ cwd: "/workspace/transfer-budget",
+ processId: String(toolIndex + 1),
+ status: "inProgress",
+ commandActions: [],
+ aggregatedOutput: "",
+ },
+ },
+ },
+ },
+ {
+ type: "item.completed",
+ ...baseEvent(provider, turnIndex, eventIndex++),
+ turnId,
+ itemId,
+ payload: {
+ itemType: "command_execution",
+ status: "completed",
+ title: `Inspected transfer path ${toolIndex + 1}`,
+ detail: "Collected a deterministic multi-module transfer diagnostic.",
+ data: {
+ threadId: FIXTURE_THREAD_ID,
+ turnId,
+ completedAtMs: turnIndex * 60_000 + eventIndex,
+ item: {
+ id: itemId,
+ type: "commandExecution",
+ command,
+ cwd: "/workspace/transfer-budget",
+ processId: String(toolIndex + 1),
+ status: "completed",
+ commandActions: [],
+ aggregatedOutput: diagnosticOutput({
+ provider,
+ turnIndex,
+ toolIndex,
+ targetBytes: 1_000,
+ }),
+ exitCode: 0,
+ durationMs: 500 + toolIndex,
+ },
+ },
+ },
+ },
+ );
+ }
+
+ const mcpItemId = `mcp-${turnIndex + 1}`;
+ const mcpResultBytes = measuredTurn
+ ? TRANSFER_MEASURED_MCP_RESULT_BYTES
+ : TRANSFER_HISTORY_MCP_RESULT_BYTES;
+ events.push(
+ {
+ type: "item.started",
+ ...baseEvent(provider, turnIndex, eventIndex++),
+ turnId,
+ itemId: mcpItemId,
+ payload: {
+ itemType: "mcp_tool_call",
+ status: "inProgress",
+ title: "fixture-history · inspect_transfer_log",
+ detail: "Reading a retained diagnostic result from the provider history.",
+ data: {
+ startedAtMs: turnIndex * 60_000 + eventIndex,
+ threadId: FIXTURE_THREAD_ID,
+ turnId,
+ item: {
+ type: "mcpToolCall",
+ id: mcpItemId,
+ server: "fixture-history",
+ tool: "inspect_transfer_log",
+ arguments: { turn: turnIndex + 1 },
+ status: "inProgress",
+ },
+ },
+ },
+ },
+ {
+ type: "item.completed",
+ ...baseEvent(provider, turnIndex, eventIndex++),
+ turnId,
+ itemId: mcpItemId,
+ payload: {
+ itemType: "mcp_tool_call",
+ status: "completed",
+ title: "fixture-history · inspect_transfer_log",
+ detail: "Retained a deterministic diagnostic result in the thread history.",
+ data: {
+ completedAtMs: turnIndex * 60_000 + eventIndex,
+ threadId: FIXTURE_THREAD_ID,
+ turnId,
+ item: {
+ type: "mcpToolCall",
+ id: mcpItemId,
+ server: "fixture-history",
+ tool: "inspect_transfer_log",
+ arguments: { turn: turnIndex + 1 },
+ durationMs: 1_000 + turnIndex,
+ error: null,
+ result: {
+ content: [
+ {
+ type: "text",
+ text: diagnosticOutput({
+ provider,
+ turnIndex,
+ toolIndex: toolCount,
+ targetBytes: mcpResultBytes,
+ }),
+ },
+ ],
+ },
+ status: "completed",
+ },
+ },
+ },
+ },
+ );
+
+ const chunks = assistantChunks(provider, turnIndex);
+ for (const [contentIndex, delta] of chunks.entries()) {
+ events.push({
+ type: "content.delta",
+ ...baseEvent(provider, turnIndex, eventIndex++),
+ turnId,
+ itemId: `assistant-${turnIndex + 1}`,
+ payload: {
+ streamKind: "assistant_text",
+ delta,
+ contentIndex,
+ },
+ });
+ }
+
+ events.push(
+ {
+ type: "thread.token-usage.updated",
+ ...baseEvent(provider, turnIndex, eventIndex++),
+ turnId,
+ payload: {
+ usage: {
+ usedTokens: 18_000 + turnIndex * 1_900,
+ maxTokens: 200_000,
+ inputTokens: 15_000 + turnIndex * 1_700,
+ cachedInputTokens: 9_000 + turnIndex * 1_100,
+ outputTokens: 3_000 + turnIndex * 200,
+ toolUses: toolCount,
+ durationMs: 4_000 + turnIndex * 250,
+ },
+ },
+ },
+ {
+ type: "turn.diff.updated",
+ ...baseEvent(provider, turnIndex, eventIndex++),
+ turnId,
+ payload: {
+ unifiedDiff: unifiedDiff(provider, turnIndex),
+ },
+ },
+ {
+ type: "turn.completed",
+ ...baseEvent(provider, turnIndex, eventIndex),
+ turnId,
+ payload: {
+ state: "completed",
+ stopReason: "end_turn",
+ usage: {
+ inputTokens: 15_000 + turnIndex * 1_700,
+ outputTokens: 3_000 + turnIndex * 200,
+ },
+ },
+ },
+ );
+
+ return { events };
+}
diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts
index a403e228b06..4ddb01e09dd 100644
--- a/apps/server/src/server.test.ts
+++ b/apps/server/src/server.test.ts
@@ -2,7 +2,7 @@ import * as NodeHttpServer from "@effect/platform-node/NodeHttpServer";
import * as NodeSocket from "@effect/platform-node/NodeSocket";
import * as NodeServices from "@effect/platform-node/NodeServices";
import * as NodeCrypto from "node:crypto";
-import { HostProcessPlatform } from "@t3tools/shared/hostProcess";
+import { HostProcessEnvironment, HostProcessPlatform } from "@t3tools/shared/hostProcess";
import {
AuthAccessTokenType,
@@ -16,6 +16,8 @@ import {
KeybindingRule,
MessageId,
ExternalLauncherCommandNotFoundError,
+ OrchestrationThreadDetailSnapshot,
+ type OrchestrationThreadStreamItem,
type OrchestrationThreadShell,
TerminalNotRunningError,
type OrchestrationCommand,
@@ -41,6 +43,7 @@ import * as RelayClient from "@t3tools/shared/relayClient";
import { assert, it } from "@effect/vitest";
import { assertFailure, assertInclude, assertTrue } from "@effect/vitest/utils";
import * as Clock from "effect/Clock";
+import * as Config from "effect/Config";
import * as Deferred from "effect/Deferred";
import * as DateTime from "effect/DateTime";
import * as Duration from "effect/Duration";
@@ -52,6 +55,8 @@ import * as ManagedRuntime from "effect/ManagedRuntime";
import * as Option from "effect/Option";
import * as Path from "effect/Path";
import * as PubSub from "effect/PubSub";
+import * as Queue from "effect/Queue";
+import * as Schema from "effect/Schema";
import * as Stream from "effect/Stream";
import * as TestClock from "effect/testing/TestClock";
import { ChildProcessSpawner } from "effect/unstable/process";
@@ -70,11 +75,34 @@ import * as Socket from "effect/unstable/socket/Socket";
import { vi } from "vite-plus/test";
const TEST_EPOCH = DateTime.makeUnsafe("1970-01-01T00:00:00.000Z");
+const decodeTransferThreadSnapshot = Schema.decodeUnknownEffect(
+ Schema.fromJsonString(OrchestrationThreadDetailSnapshot),
+);
+
+const collectQueueUntil = Effect.fn("TransferBudget.collectQueueUntil")(function* (
+ queue: Queue.Queue,
+ predicate: (value: A) => boolean,
+ waitDescription: string,
+) {
+ return yield* Effect.gen(function* () {
+ const values: A[] = [];
+ while (true) {
+ const value = yield* Queue.take(queue);
+ values.push(value);
+ if (predicate(value)) return values;
+ }
+ }).pipe(
+ Effect.timeoutOrElse({
+ duration: "10 seconds",
+ orElse: () => Effect.die(new Error(`Timed out waiting for ${waitDescription}`)),
+ }),
+ );
+});
import * as BackgroundPolicy from "./background/BackgroundPolicy.ts";
import * as ServerConfig from "./config.ts";
import { makeRoutesLayer } from "./server.ts";
-import { resolveAvailableEditorsForConfig } from "./ws.ts";
+import { isThreadDetailEvent, resolveAvailableEditorsForConfig } from "./ws.ts";
import * as CheckpointDiffQuery from "./checkpointing/CheckpointDiffQuery.ts";
import * as GitManager from "./git/GitManager.ts";
import * as Keybindings from "./keybindings.ts";
@@ -123,6 +151,32 @@ import * as ResourceAttribution from "./resourceTelemetry/ResourceAttribution.ts
import * as ResourceTelemetry from "./resourceTelemetry/ResourceTelemetry.ts";
import * as Data from "effect/Data";
+import { makeOrchestrationIntegrationHarness } from "../integration/OrchestrationEngineHarness.integration.ts";
+import {
+ countingWsRpcProtocolLayer,
+ makeCountingWsRpcClient,
+ makeWebSocketTransferRecorder,
+ measureHttpGet,
+ transferDelta,
+} from "../integration/NetworkTransferMeasurement.integration.ts";
+import {
+ expectedMeasuredAssistantText,
+ queueMeasuredTransferTurn,
+ seedTransferBudgetHistory,
+ TRANSFER_HISTORY_TURN_COUNT,
+ TRANSFER_MEASURED_TURN_CREATED_AT,
+ TRANSFER_MEASURED_TURN_INDEX,
+ TRANSFER_THREAD_ID,
+ transferModelSelection,
+ waitForTurnQuiesced,
+} from "../integration/TransferBudgetScenario.integration.ts";
+import {
+ formatTransferBudgetReport,
+ formatTransferBudgetResult,
+ type TransferBudgetRun,
+ transferBudgetViolations,
+} from "../integration/TransferBudgetReport.integration.ts";
+
const defaultProjectId = ProjectId.make("project-default");
const defaultThreadId = ThreadId.make("thread-default");
const defaultDesktopBootstrapToken = "test-desktop-bootstrap-token";
@@ -549,9 +603,12 @@ const buildAppUnderTest = (options?: {
),
),
);
+ const serviceLauncherClientLayer = ServiceLauncherClient.layer.pipe(
+ Layer.provide(Layer.succeed(HostProcessEnvironment, {})),
+ );
const servedRoutesLayer = HttpRouter.serve(
- makeRoutesLayer.pipe(Layer.provide(ServiceLauncherClient.layer)),
+ makeRoutesLayer.pipe(Layer.provide(serviceLauncherClientLayer)),
{
disableListenLog: true,
disableLogger: true,
@@ -1319,6 +1376,28 @@ const getWsServerUrl = (
);
});
+// Mirrors NodeHttpServer.layerTest, which does not expose server options,
+// with the production `websocket: { perMessageDeflate: true }` setting.
+const NodeHttpServerTestWithWsDeflate = HttpServer.layerTestClient.pipe(
+ Layer.provide(
+ Layer.fresh(FetchHttpClient.layer).pipe(
+ Layer.provide(Layer.succeed(FetchHttpClient.RequestInit)({ keepalive: false })),
+ ),
+ ),
+ Layer.provideMerge(
+ Layer.unwrap(
+ Effect.map(
+ Effect.promise(() => import("node:http")),
+ (NodeHttp) =>
+ NodeHttpServer.layer(NodeHttp.createServer, {
+ port: 0,
+ websocket: { perMessageDeflate: true },
+ }),
+ ),
+ ),
+ ),
+);
+
it.layer(NodeServices.layer)("server router seam", (it) => {
it.effect("parks HTTP ingress until command readiness", () =>
Effect.gen(function* () {
@@ -3219,28 +3298,6 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
}).pipe(Effect.provide(NodeHttpServer.layerTest)),
);
- // Mirrors NodeHttpServer.layerTest, which does not expose server options,
- // with the production `websocket: { perMessageDeflate: true }` setting.
- const NodeHttpServerTestWithWsDeflate = HttpServer.layerTestClient.pipe(
- Layer.provide(
- Layer.fresh(FetchHttpClient.layer).pipe(
- Layer.provide(Layer.succeed(FetchHttpClient.RequestInit)({ keepalive: false })),
- ),
- ),
- Layer.provideMerge(
- Layer.unwrap(
- Effect.map(
- Effect.promise(() => import("node:http")),
- (NodeHttp) =>
- NodeHttpServer.layer(NodeHttp.createServer, {
- port: 0,
- websocket: { perMessageDeflate: true },
- }),
- ),
- ),
- ),
- );
-
it.effect("negotiates permessage-deflate with clients that offer it", () =>
Effect.gen(function* () {
yield* buildAppUnderTest();
@@ -7839,3 +7896,167 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
}).pipe(Effect.provide(NodeHttpServer.layerTest)),
);
});
+
+it.live(
+ "reports thread HTTP and WebSocket transfer budgets",
+ () =>
+ Effect.gen(function* () {
+ const providers = [
+ ProviderDriverKind.make("codex"),
+ ProviderDriverKind.make("claudeAgent"),
+ ] as const;
+
+ const runs = yield* Effect.forEach(
+ providers,
+ (provider) =>
+ Effect.acquireUseRelease(
+ makeOrchestrationIntegrationHarness({ provider }),
+ (harness) =>
+ Effect.gen(function* () {
+ yield* seedTransferBudgetHistory(harness, provider);
+ yield* buildAppUnderTest({
+ layers: {
+ orchestrationEngine: harness.engine,
+ projectionSnapshotQuery: harness.snapshotQuery,
+ },
+ });
+
+ const baseUrl = yield* getHttpServerUrl();
+ const cookie = yield* getAuthenticatedSessionCookieHeader();
+
+ const recorder = makeWebSocketTransferRecorder();
+ const wsUrl = baseUrl.replace(/^http:/, "ws:") + "/ws";
+ const protocolLayer = countingWsRpcProtocolLayer({
+ url: wsUrl,
+ cookie,
+ recorder,
+ });
+
+ return yield* Effect.scoped(
+ Effect.gen(function* () {
+ const client = yield* makeCountingWsRpcClient;
+
+ const threadSnapshot = yield* measureHttpGet({
+ url: `${baseUrl}/api/orchestration/threads/${TRANSFER_THREAD_ID}`,
+ headers: { cookie },
+ });
+ assert.equal(threadSnapshot.status, 200);
+ assert.equal(threadSnapshot.contentEncoding, "gzip");
+ const decodedThread = yield* decodeTransferThreadSnapshot(
+ Buffer.from(threadSnapshot.decodedBody).toString("utf8"),
+ );
+ assert.equal(
+ decodedThread.thread.messages.length,
+ TRANSFER_HISTORY_TURN_COUNT * 2,
+ );
+
+ const threadItems = yield* Queue.unbounded();
+ yield* client[ORCHESTRATION_WS_METHODS.subscribeThread]({
+ threadId: TRANSFER_THREAD_ID,
+ afterSequence: decodedThread.snapshotSequence,
+ requestCompletionMarker: true,
+ }).pipe(
+ Stream.runForEach((item) =>
+ Queue.offer(threadItems, item).pipe(Effect.asVoid),
+ ),
+ Effect.forkScoped,
+ );
+ const initialThreadItems = yield* collectQueueUntil(
+ threadItems,
+ (item) => item.kind === "synchronized",
+ `${provider} thread subscription to synchronize`,
+ );
+ assert.isFalse(initialThreadItems.some((item) => item.kind === "snapshot"));
+ assert.include(recorder.negotiatedExtensions(), "permessage-deflate");
+
+ yield* queueMeasuredTransferTurn(harness, provider);
+ const turnStartTotals = recorder.totals();
+ yield* client[ORCHESTRATION_WS_METHODS.dispatchCommand]({
+ type: "thread.turn.start",
+ commandId: CommandId.make(`transfer:${provider}:measured-turn`),
+ threadId: TRANSFER_THREAD_ID,
+ message: {
+ messageId: MessageId.make("transfer-user-measured"),
+ role: "user",
+ text: "Measure the client-bound transfer for this turn.",
+ attachments: [],
+ },
+ modelSelection: transferModelSelection(provider),
+ runtimeMode: "approval-required",
+ interactionMode: "default",
+ createdAt: TRANSFER_MEASURED_TURN_CREATED_AT,
+ });
+ yield* waitForTurnQuiesced(harness, TRANSFER_MEASURED_TURN_INDEX + 1);
+ const finalThreadSequence = yield* harness.engine
+ .readEvents(decodedThread.snapshotSequence, 10_000)
+ .pipe(
+ Stream.runFold(
+ () => decodedThread.snapshotSequence,
+ (sequence, event) =>
+ event.aggregateId === TRANSFER_THREAD_ID && isThreadDetailEvent(event)
+ ? Math.max(sequence, event.sequence)
+ : sequence,
+ ),
+ );
+ assert.isAbove(finalThreadSequence, decodedThread.snapshotSequence);
+
+ yield* collectQueueUntil(
+ threadItems,
+ (item) =>
+ item.kind === "event" && item.event.sequence === finalThreadSequence,
+ `${provider} thread stream to reach sequence ${finalThreadSequence}`,
+ );
+ const measuredTurnWebSocket = transferDelta(turnStartTotals, recorder.totals());
+
+ const finalThreadSnapshot = yield* harness.snapshotQuery
+ .getThreadDetailSnapshot(TRANSFER_THREAD_ID)
+ .pipe(Effect.map(Option.getOrThrow));
+ const expectedAssistantText = expectedMeasuredAssistantText(provider);
+ const measuredAssistant = finalThreadSnapshot.thread.messages.find(
+ (message) =>
+ message.role === "assistant" && message.text === expectedAssistantText,
+ );
+ assert.isDefined(measuredAssistant);
+ assert.isTrue(
+ finalThreadSnapshot.thread.messages.length >= TRANSFER_HISTORY_TURN_COUNT * 2,
+ );
+ assert.equal(measuredAssistant?.streaming, false);
+ assert.equal(finalThreadSnapshot.thread.session?.status, "ready");
+ assert.equal(
+ finalThreadSnapshot.thread.checkpoints.length,
+ TRANSFER_HISTORY_TURN_COUNT + 1,
+ );
+
+ return {
+ provider,
+ threadSnapshot,
+ measuredTurnWebSocket,
+ } satisfies TransferBudgetRun;
+ }).pipe(Effect.provide(protocolLayer)),
+ );
+ }),
+ (harness) => harness.dispose,
+ ).pipe(Effect.provide(NodeHttpServerTestWithWsDeflate)),
+ { concurrency: 1 },
+ );
+
+ const report = formatTransferBudgetReport(runs);
+ yield* Effect.logInfo(`\n${report}`);
+ const reportPath = yield* Config.string("T3CODE_TRANSFER_BUDGET_REPORT_PATH").pipe(
+ Config.option,
+ );
+ if (Option.isSome(reportPath)) {
+ const fileSystem = yield* FileSystem.FileSystem;
+ yield* fileSystem.writeFileString(reportPath.value, report);
+ }
+ const resultPath = yield* Config.string("T3CODE_TRANSFER_BUDGET_RESULT_PATH").pipe(
+ Config.option,
+ );
+ if (Option.isSome(resultPath)) {
+ const fileSystem = yield* FileSystem.FileSystem;
+ yield* fileSystem.writeFileString(resultPath.value, formatTransferBudgetResult(runs));
+ }
+ assert.deepEqual(transferBudgetViolations(runs), []);
+ }).pipe(Effect.provide(NodeServices.layer)),
+ 120_000,
+);
diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts
index 6bafb9ec3ba..a6b155c296f 100644
--- a/apps/server/src/ws.ts
+++ b/apps/server/src/ws.ts
@@ -268,7 +268,7 @@ function projectSetupScriptCompatibilityDetail(
}
}
-function isThreadDetailEvent(event: OrchestrationEvent): event is Extract<
+export function isThreadDetailEvent(event: OrchestrationEvent): event is Extract<
OrchestrationEvent,
{
type: