authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-07-27 02:56:34-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-07-27 02:56:34-07:00
logd60f6e971f8f982836f26507275d651e2e7af4a5
tree833d3f3948e7dfe0d55b1cbaba135d9f019a9907
parent89c5a192f4d36bcf4201e8b490ab5ad37a4adac7
signature Signed by PGP key B5690EEEBB952194

fix: cut websocket throughput in half by pruning activity payloads (#4622)

4 files changed, 454 insertions(+), 4 deletions(-)

apps/server/src/orchestration/ActivityPayloadProjection.ts created+227
......@@ -0,0 +1,227 @@
1import type {
2 OrchestrationEvent,
3 OrchestrationThreadActivity,
4 OrchestrationThreadDetailSnapshot,
5} from "@t3tools/contracts";
6
7function asRecord(value: unknown): Record<string, unknown> | null {
8 return value !== null && typeof value === "object" && !Array.isArray(value)
9 ? (value as Record<string, unknown>)
10 : null;
11}
12
13function asTrimmedString(value: unknown): string | null {
14 if (typeof value !== "string") {
15 return null;
16 }
17 const trimmed = value.trim();
18 return trimmed.length > 0 ? trimmed : null;
19}
20
21function pushChangedFile(target: string[], seen: Set<string>, value: unknown): void {
22 const normalized = asTrimmedString(value);
23 if (!normalized || seen.has(normalized)) {
24 return;
25 }
26 seen.add(normalized);
27 target.push(normalized);
28}
29
30function collectChangedFiles(
31 value: unknown,
32 target: string[],
33 seen: Set<string>,
34 depth: number,
35): void {
36 if (depth > 4 || target.length >= 12) {
37 return;
38 }
39 if (Array.isArray(value)) {
40 for (const entry of value) {
41 collectChangedFiles(entry, target, seen, depth + 1);
42 if (target.length >= 12) {
43 return;
44 }
45 }
46 return;
47 }
48
49 const record = asRecord(value);
50 if (!record) {
51 return;
52 }
53
54 pushChangedFile(target, seen, record.path);
55 pushChangedFile(target, seen, record.filePath);
56 pushChangedFile(target, seen, record.relativePath);
57 pushChangedFile(target, seen, record.filename);
58 pushChangedFile(target, seen, record.newPath);
59 pushChangedFile(target, seen, record.oldPath);
60
61 for (const nestedKey of [
62 "item",
63 "result",
64 "input",
65 "data",
66 "changes",
67 "files",
68 "edits",
69 "patch",
70 "patches",
71 "operations",
72 ]) {
73 if (!(nestedKey in record)) {
74 continue;
75 }
76 collectChangedFiles(record[nestedKey], target, seen, depth + 1);
77 if (target.length >= 12) {
78 return;
79 }
80 }
81}
82
83function projectCommandData(data: Record<string, unknown>): Record<string, unknown> | undefined {
84 const item = asRecord(data.item);
85 if (!item) {
86 return undefined;
87 }
88
89 const projectedItem: Record<string, unknown> = {};
90 if ("command" in item) {
91 projectedItem.command = item.command;
92 }
93
94 const input = asRecord(item.input);
95 if (input && "command" in input) {
96 projectedItem.input = { command: input.command };
97 }
98
99 const result = asRecord(item.result);
100 if (result && "command" in result) {
101 projectedItem.result = { command: result.command };
102 }
103
104 return Object.keys(projectedItem).length > 0 ? projectedItem : undefined;
105}
106
107function summarizeToolTextOutput(value: string): string | null {
108 const lines: string[] = [];
109 for (const rawLine of value.split(/\r?\n/u)) {
110 const line = rawLine.replace(/\s+/g, " ").trim();
111 if (line.length > 0) {
112 lines.push(line);
113 }
114 }
115
116 const firstLine = lines.find((line) => line !== "```");
117 if (firstLine) {
118 return firstLine.length <= 84 ? firstLine : `${firstLine.slice(0, 83).trimEnd()}…`;
119 }
120 if (lines.length > 1) {
121 return `${lines.length.toLocaleString()} lines`;
122 }
123 return null;
124}
125
126function projectRawOutput(value: unknown): Record<string, unknown> | undefined {
127 const rawOutput = asRecord(value);
128 if (!rawOutput) {
129 return undefined;
130 }
131
132 if (typeof rawOutput.totalFiles === "number" && Number.isFinite(rawOutput.totalFiles)) {
133 return {
134 totalFiles: rawOutput.totalFiles,
135 ...(rawOutput.truncated === true ? { truncated: true } : {}),
136 };
137 }
138
139 const content = asTrimmedString(rawOutput.content);
140 if (content) {
141 const summary = summarizeToolTextOutput(content);
142 return summary ? { content: summary } : undefined;
143 }
144
145 const stdout = asTrimmedString(rawOutput.stdout);
146 if (stdout) {
147 const summary = summarizeToolTextOutput(stdout);
148 return summary ? { content: summary } : undefined;
149 }
150
151 return undefined;
152}
153
154/**
155 * Removes activity payload fields that no current client reads while retaining
156 * the full payload in persistence and the event store.
157 */
158export function projectActivityPayload(
159 activity: OrchestrationThreadActivity,
160): OrchestrationThreadActivity {
161 const payload = asRecord(activity.payload);
162 const data = asRecord(payload?.data);
163 if (!payload || !data || payload.itemType === "mcp_tool_call") {
164 return activity;
165 }
166
167 const projectedData: Record<string, unknown> = {};
168 const item = projectCommandData(data);
169 if (item) {
170 projectedData.item = item;
171 }
172 if ("command" in data) {
173 projectedData.command = data.command;
174 }
175
176 const changedFiles: string[] = [];
177 collectChangedFiles(data, changedFiles, new Set<string>(), 0);
178 if (changedFiles.length > 0) {
179 // Both clients discover file names by walking objects with path-like keys.
180 projectedData.files = changedFiles.map((path) => ({ path }));
181 }
182
183 if ("toolCallId" in data) {
184 projectedData.toolCallId = data.toolCallId;
185 }
186 if ("kind" in data) {
187 projectedData.kind = data.kind;
188 }
189
190 const rawOutput = projectRawOutput(data.rawOutput);
191 if (rawOutput) {
192 projectedData.rawOutput = rawOutput;
193 }
194
195 return {
196 ...activity,
197 payload: {
198 ...payload,
199 data: projectedData,
200 },
201 };
202}
203
204export function projectThreadDetailSnapshot(
205 snapshot: OrchestrationThreadDetailSnapshot,
206): OrchestrationThreadDetailSnapshot {
207 return {
208 ...snapshot,
209 thread: {
210 ...snapshot.thread,
211 activities: snapshot.thread.activities.map(projectActivityPayload),
212 },
213 };
214}
215
216export function projectActivityEvent(event: OrchestrationEvent): OrchestrationEvent {
217 if (event.type !== "thread.activity-appended") {
218 return event;
219 }
220 return {
221 ...event,
222 payload: {
223 ...event.payload,
224 activity: projectActivityPayload(event.payload.activity),
225 },
226 };
227}
apps/server/src/orchestration/http.ts+2-1
......@@ -7,6 +7,7 @@ import * as Effect from "effect/Effect";
77import * as Option from "effect/Option";
88import * as HttpApiBuilder from "effect/unstable/httpapi/HttpApiBuilder";
99
10import { projectThreadDetailSnapshot } from "./ActivityPayloadProjection.ts";
1011import { normalizeDispatchCommand } from "./Normalizer.ts";
1112import {
1213 annotateEnvironmentRequest,
......@@ -69,7 +70,7 @@ export const orchestrationHttpApiLayer = HttpApiBuilder.group(
6970 if (Option.isNone(snapshot)) {
7071 return yield* failEnvironmentNotFound("thread_not_found");
7172 }
72 return snapshot.value;
73 return projectThreadDetailSnapshot(snapshot.value);
7374 }),
7475 )
7576 .handle(
apps/server/src/ws.ts+11-3
......@@ -68,6 +68,10 @@ import * as CheckpointDiffQuery from "./checkpointing/CheckpointDiffQuery.ts";
6868import * as ServerConfig from "./config.ts";
6969import * as Keybindings from "./keybindings.ts";
7070import * as ExternalLauncher from "./process/externalLauncher.ts";
71import {
72 projectActivityEvent,
73 projectThreadDetailSnapshot,
74} from "./orchestration/ActivityPayloadProjection.ts";
7175import { normalizeDispatchCommand } from "./orchestration/Normalizer.ts";
7276import * as OrchestrationEngine from "./orchestration/Services/OrchestrationEngine.ts";
7377import * as ProjectionSnapshotQuery from "./orchestration/Services/ProjectionSnapshotQuery.ts";
......@@ -1213,6 +1217,7 @@ const makeWsRpcLayer = (
12131217 ).pipe(
12141218 Effect.map((events) => Array.from(events)),
12151219 Effect.flatMap(enrichOrchestrationEvents),
1220 Effect.map((events) => events.map(projectActivityEvent)),
12161221 Effect.mapError(
12171222 (cause) =>
12181223 new OrchestrationReplayEventsError({
......@@ -1360,7 +1365,7 @@ const makeWsRpcLayer = (
13601365 Stream.filter(isThisThreadDetailEvent),
13611366 Stream.map((event) => ({
13621367 kind: "event" as const,
1363 event,
1368 event: projectActivityEvent(event),
13641369 })),
13651370 );
13661371
......@@ -1395,7 +1400,10 @@ const makeWsRpcLayer = (
13951400 .readEvents(afterSequence, Number.MAX_SAFE_INTEGER)
13961401 .pipe(
13971402 Stream.filter(isThisThreadDetailEvent),
1398 Stream.map((event) => ({ kind: "event" as const, event })),
1403 Stream.map((event) => ({
1404 kind: "event" as const,
1405 event: projectActivityEvent(event),
1406 })),
13991407 Stream.mapError(
14001408 (cause) =>
14011409 new OrchestrationGetSnapshotError({
......@@ -1447,7 +1455,7 @@ const makeWsRpcLayer = (
14471455 return Stream.concat(
14481456 Stream.make({
14491457 kind: "snapshot" as const,
1450 snapshot: snapshot.value,
1458 snapshot: projectThreadDetailSnapshot(snapshot.value),
14511459 }),
14521460 afterSnapshot,
14531461 );
apps/server/test/ActivityPayloadProjection.test.ts created+214
......@@ -0,0 +1,214 @@
1import {
2 EventId,
3 ProjectId,
4 ProviderInstanceId,
5 ThreadId,
6 TurnId,
7 type OrchestrationEvent,
8 type OrchestrationThread,
9 type OrchestrationThreadActivity,
10} from "@t3tools/contracts";
11import { describe, expect, it } from "vite-plus/test";
12
13import { buildThreadFeed } from "../../mobile/src/lib/threadActivity.ts";
14import { deriveWorkLogEntries } from "../../web/src/session-logic.ts";
15import {
16 projectActivityEvent,
17 projectActivityPayload,
18 projectThreadDetailSnapshot,
19} from "../src/orchestration/ActivityPayloadProjection.ts";
20
21function makeActivity(
22 id: string,
23 itemType: string,
24 data: Record<string, unknown>,
25): OrchestrationThreadActivity {
26 return {
27 id: EventId.make(id),
28 tone: "tool",
29 kind: "tool.completed",
30 summary: `Completed ${itemType}`,
31 payload: {
32 itemType,
33 title: itemType,
34 detail: `${itemType} detail`,
35 status: "completed",
36 requestKind: "command",
37 data,
38 },
39 turnId: TurnId.make(`turn-${id}`),
40 createdAt: "2026-07-27T00:00:00.000Z",
41 };
42}
43
44function makeThread(activities: ReadonlyArray<OrchestrationThreadActivity>): OrchestrationThread {
45 return {
46 id: ThreadId.make("thread-projection"),
47 projectId: ProjectId.make("project-projection"),
48 title: "Activity projection",
49 modelSelection: {
50 instanceId: ProviderInstanceId.make("codex"),
51 model: "gpt-5.4",
52 },
53 runtimeMode: "full-access",
54 interactionMode: "default",
55 branch: null,
56 worktreePath: null,
57 latestTurn: null,
58 createdAt: "2026-07-27T00:00:00.000Z",
59 updatedAt: "2026-07-27T00:00:00.000Z",
60 archivedAt: null,
61 settledOverride: null,
62 settledAt: null,
63 deletedAt: null,
64 messages: [],
65 proposedPlans: [],
66 activities,
67 checkpoints: [],
68 session: null,
69 };
70}
71
72const fixtures = [
73 makeActivity("command", "command_execution", {
74 item: {
75 command: ["bash", "-lc", "pnpm test"],
76 input: { command: "fallback input", ignored: "input bulk" },
77 result: { command: "fallback result", aggregatedOutput: "x".repeat(10_000) },
78 commandActions: [{ type: "unknown", output: "y".repeat(5_000) }],
79 },
80 command: "fallback data",
81 kind: "execute",
82 toolCallId: "tool-command",
83 rawOutput: {
84 content: "\n```\nfirst useful line\nsecond line",
85 stdout: "unused stdout",
86 ignored: "raw bulk",
87 },
88 ignored: "top-level bulk",
89 }),
90 makeActivity("file-change", "file_change", {
91 item: {
92 changes: [
93 { oldPath: "src/old.ts", newPath: "src/new.ts", patch: "large patch".repeat(1_000) },
94 { filePath: "src/second.ts" },
95 ],
96 },
97 ignored: "top-level bulk",
98 }),
99 makeActivity("dynamic", "dynamic_tool_call", {
100 toolCallId: "tool-dynamic",
101 rawOutput: {
102 stdout: "dynamic summary\nlong output".repeat(1_000),
103 },
104 ignored: "top-level bulk",
105 }),
106 makeActivity("collab", "collab_agent_tool_call", {
107 kind: "delegate",
108 rawOutput: {
109 content: "``` \n```",
110 stdout: "must not be used when content is present",
111 },
112 ignored: "top-level bulk",
113 }),
114 makeActivity("mcp", "mcp_tool_call", {
115 item: {
116 server: "repository",
117 tool: "search",
118 arguments: { query: "activity projection" },
119 aggregatedOutput: "mcp payload remains available",
120 },
121 ignored: "MCP data is rendered verbatim",
122 }),
123 makeActivity("search", "web_search", {
124 rawOutput: {
125 totalFiles: 42,
126 truncated: true,
127 content: "ignored because totalFiles wins",
128 },
129 ignored: "top-level bulk",
130 }),
131 makeActivity("image", "image_view", {
132 ignored: "top-level bulk",
133 }),
134] satisfies ReadonlyArray<OrchestrationThreadActivity>;
135
136describe("projectActivityPayload", () => {
137 it("drops unread bulk while retaining command, file, tool, and summary inputs", () => {
138 const projected = projectActivityPayload(fixtures[0]!);
139 expect(projected.payload).toEqual({
140 itemType: "command_execution",
141 title: "command_execution",
142 detail: "command_execution detail",
143 status: "completed",
144 requestKind: "command",
145 data: {
146 item: {
147 command: ["bash", "-lc", "pnpm test"],
148 input: { command: "fallback input" },
149 result: { command: "fallback result" },
150 },
151 command: "fallback data",
152 toolCallId: "tool-command",
153 kind: "execute",
154 rawOutput: { content: "first useful line" },
155 },
156 });
157
158 expect(projectActivityPayload(fixtures[1]!).payload).toMatchObject({
159 data: {
160 files: [{ path: "src/new.ts" }, { path: "src/old.ts" }, { path: "src/second.ts" }],
161 },
162 });
163 });
164
165 it("passes MCP tool data through unchanged", () => {
166 expect(projectActivityPayload(fixtures[4]!)).toBe(fixtures[4]);
167 });
168
169 it("keeps current web and mobile derived output identical for every tool item type", () => {
170 for (const activity of fixtures) {
171 const projected = projectActivityPayload(activity);
172 expect(deriveWorkLogEntries([projected])).toEqual(deriveWorkLogEntries([activity]));
173 expect(buildThreadFeed(makeThread([projected]))).toEqual(
174 buildThreadFeed(makeThread([activity])),
175 );
176 }
177 });
178
179 it("projects snapshot and event transports without mutating their sources", () => {
180 const activity = fixtures[0]!;
181 const thread = makeThread([activity]);
182 const snapshot = { snapshotSequence: 7, thread };
183 const projectedSnapshot = projectThreadDetailSnapshot(snapshot);
184
185 expect(projectedSnapshot.thread.activities[0]).not.toBe(activity);
186 expect(snapshot.thread.activities[0]).toBe(activity);
187
188 const event = {
189 sequence: 8,
190 eventId: EventId.make("event-activity"),
191 aggregateKind: "thread",
192 aggregateId: thread.id,
193 occurredAt: "2026-07-27T00:00:01.000Z",
194 commandId: null,
195 causationEventId: null,
196 correlationId: null,
197 metadata: {},
198 type: "thread.activity-appended",
199 payload: {
200 threadId: thread.id,
201 activity,
202 },
203 } satisfies Extract<OrchestrationEvent, { type: "thread.activity-appended" }>;
204
205 const projectedEvent = projectActivityEvent(event);
206 expect(projectedEvent).not.toBe(event);
207 expect(
208 projectedEvent.type === "thread.activity-appended"
209 ? projectedEvent.payload.activity
210 : undefined,
211 ).toEqual(projectActivityPayload(activity));
212 expect(event.payload.activity).toBe(activity);
213 });
214});