1import type {
2 CheckpointId,
3 MessageId,
4 NodeId,
5 OrchestrationV2Checkpoint,
6 OrchestrationV2DomainEvent,
7 OrchestrationV2ContextHandoff,
8 OrchestrationV2StoredEvent,
9 OrchestrationV2Subagent,
10 OrchestrationV2ThreadProjection,
11 OrchestrationV2ThreadStreamItem,
12 OrchestrationV2TurnItem,
13} from "@t3tools/contracts";
14import {
15 dedupeSubagentBodies,
16 dedupeTurnItemBodies,
17} from "@t3tools/shared/orchestrationV2DedupedBodies";
18import { MCP_APP_OUTPUT_KEY, readMcpAppReference } from "@t3tools/shared/mcpApp";
19import {
20 compactDynamicToolOutput,
21 omitToolOutputImageData,
22 toolOutputImages,
23 toolOutputIndicatesFailure,
24} from "@t3tools/shared/toolOutput";
25
26// A checkpoint over vendored or generated trees can list 15k+ files; the card needs a prefix.
27const MAX_WIRE_CHECKPOINT_FILES = 200;
28const MAX_DETAIL_STRING_BYTES = 32_768;
29const MAX_DYNAMIC_VALUE_BYTES = 16_384;
30const MAX_ON_DEMAND_BYTES = 256 * 1024;
31const TRUNCATION_MARKER = "\n… output truncated for transport";
32const TRUNCATION_MARKER_BYTES = Buffer.byteLength(TRUNCATION_MARKER, "utf8");
33
34function truncateDetail(
35 value: string | undefined,
36 maxBytes = MAX_DETAIL_STRING_BYTES,
37): string | undefined {
38 if (
39 value === undefined ||
40 (value.length <= maxBytes && Buffer.byteLength(value, "utf8") <= maxBytes)
41 ) {
42 return value;
43 }
44 // The marker counts toward the cap, so projecting truncated text again leaves it unchanged.
45 const budget = maxBytes - TRUNCATION_MARKER_BYTES;
46 // UTF-8 needs at least one byte per UTF-16 unit; encode only a potentially fitting prefix.
47 const prefix = Buffer.from(value.slice(0, budget), "utf8")
48 .subarray(0, budget)
49 .toString("utf8")
50 .replace(/\uFFFD$/u, "");
51 return `${prefix}${TRUNCATION_MARKER}`;
52}
53
54/** Bound stored diffs to the client wire limit to avoid retaining invisible data in SQLite and heap. */
55export function boundFileChangeForStorage(item: OrchestrationV2TurnItem): OrchestrationV2TurnItem {
56 if (item.type !== "file_change") return item;
57 const { diffStr, oldStr, newStr } = item;
58 const bounded = {
59 ...item,
60 ...(diffStr === undefined ? {} : { diffStr: truncateDetail(diffStr) }),
61 ...(oldStr === undefined ? {} : { oldStr: truncateDetail(oldStr) }),
62 ...(newStr === undefined ? {} : { newStr: truncateDetail(newStr) }),
63 };
64 return bounded.diffStr === diffStr && bounded.oldStr === oldStr && bounded.newStr === newStr
65 ? item
66 : bounded;
67}
68
69function summarizeDynamicValue(value: unknown): unknown {
70 let serialized: string;
71 try {
72 if (typeof value === "string" && value.length > MAX_DYNAMIC_VALUE_BYTES) {
73 serialized = value;
74 } else {
75 const json = JSON.stringify(value) ?? String(value);
76 if (Buffer.byteLength(json, "utf8") <= MAX_DYNAMIC_VALUE_BYTES) {
77 return value;
78 }
79 serialized = typeof value === "string" ? value : json;
80 }
81 } catch {
82 serialized = "Unserializable tool output";
83 }
84
85 // Splitting and normalizing every line can allocate far more than the input.
86 const start = /\S/u.exec(serialized)?.index;
87 let firstLine = start === undefined ? "Large tool output" : "";
88 let pendingSpace = false;
89 for (let index = start ?? serialized.length; index < serialized.length; index += 1) {
90 const character = serialized[index]!;
91 if (character === "\n") break;
92 if (/\s/u.test(character)) {
93 pendingSpace = true;
94 continue;
95 }
96 if (pendingSpace) firstLine += " ";
97 firstLine += character;
98 pendingSpace = false;
99 if (firstLine.length > 160) break;
100 }
101 const last = firstLine.charCodeAt(158);
102 const end = last >= 0xd800 && last <= 0xdbff ? 158 : 159;
103 return {
104 summary: firstLine.length <= 160 ? firstLine : `${firstLine.slice(0, end).trimEnd()}…`,
105 truncated: true,
106 };
107}
108
109type CheckpointFiles = Pick<OrchestrationV2Checkpoint, "files" | "filesTotal">;
110
111// Same input, same output object: snapshot compaction relies on shared item identity.
112const cappedCheckpointFiles = new WeakMap<object, object>();
113
114function capCheckpointFiles<T extends CheckpointFiles>(value: T): T {
115 if (value.files.length <= MAX_WIRE_CHECKPOINT_FILES || value.filesTotal !== undefined) {
116 return value;
117 }
118 const cached = cappedCheckpointFiles.get(value);
119 if (cached !== undefined) return cached as T;
120 let additions = 0;
121 let deletions = 0;
122 for (const file of value.files) {
123 additions += file.additions;
124 deletions += file.deletions;
125 }
126 const capped = {
127 ...value,
128 files: value.files.slice(0, MAX_WIRE_CHECKPOINT_FILES),
129 filesTotal: { count: value.files.length, additions, deletions },
130 };
131 cappedCheckpointFiles.set(value, capped);
132 return capped;
133}
134
135/** Idempotent; bounded snapshots that skip `projectThreadProjectionForWire` still need it. */
136export function capProjectionCheckpointFilesForWire(
137 projection: OrchestrationV2ThreadProjection,
138): OrchestrationV2ThreadProjection {
139 const checkpoints = projection.checkpoints.map(capCheckpointFiles);
140 const turnItems = projection.turnItems.map(capCheckpointItemFiles);
141 const visibleTurnItems = projection.visibleTurnItems.map((row) => {
142 const item = capCheckpointItemFiles(row.item);
143 return item === row.item ? row : { ...row, item };
144 });
145 const unchanged = (next: ReadonlyArray<unknown>, previous: ReadonlyArray<unknown>) =>
146 next.every((value, index) => value === previous[index]);
147 return unchanged(checkpoints, projection.checkpoints) &&
148 unchanged(turnItems, projection.turnItems) &&
149 unchanged(visibleTurnItems, projection.visibleTurnItems)
150 ? projection
151 : { ...projection, checkpoints, turnItems, visibleTurnItems };
152}
153
154function capCheckpointItemFiles(item: OrchestrationV2TurnItem): OrchestrationV2TurnItem {
155 return item.type === "checkpoint" ? capCheckpointFiles(item) : item;
156}
157
158export function projectTurnItemForWire(item: OrchestrationV2TurnItem): OrchestrationV2TurnItem {
159 switch (item.type) {
160 case "checkpoint":
161 return capCheckpointFiles(item);
162 case "handoff": {
163 const { summary: _summary, ...projected } = item;
164 return projected;
165 }
166 case "command_execution": {
167 const { output, ...projected } = item;
168 // A successful search can print "ENOENT" or "exit code 2", so output text
169 // decides only when the provider reported no exit code.
170 const failed =
171 item.exitCode === undefined
172 ? item.outputIndicatesFailure === true ||
173 (output !== undefined &&
174 toolOutputIndicatesFailure(output.slice(0, MAX_DETAIL_STRING_BYTES)))
175 : item.exitCode !== 0;
176 return {
177 ...projected,
178 ...(failed ? { outputIndicatesFailure: true } : {}),
179 ...(output?.trim() ? { outputOmitted: true } : {}),
180 };
181 }
182 case "file_change": {
183 const { diffStr, oldStr: _old, newStr: _new, ...projected } = item;
184 return item.status === "failed" && diffStr?.trim()
185 ? { ...projected, diffStr: truncateDetail(diffStr) }
186 : projected;
187 }
188 case "subagent":
189 return {
190 ...item,
191 prompt: truncateDetail(item.prompt) ?? "",
192 progress: truncateDetail(item.progress),
193 result: item.result === null ? null : (truncateDetail(item.result) ?? null),
194 };
195 case "dynamic_tool": {
196 const { output: rawOutput, ...projected } = item;
197 const output = compactDynamicToolOutput(rawOutput);
198 return {
199 ...projected,
200 input: summarizeDynamicValue(item.input),
201 ...(output === undefined ? {} : { output }),
202 ...(hasDynamicValue(rawOutput) ? { outputOmitted: true } : {}),
203 };
204 }
205 default:
206 return item;
207 }
208}
209
210function hasDynamicValue(value: unknown): boolean {
211 if (value === undefined || value === null) return false;
212 if (typeof value === "string") return value.trim().length > 0;
213 if (Array.isArray(value)) return value.length > 0;
214 return typeof value !== "object" || Object.keys(value).length > 0;
215}
216
217function boundDynamicValue(value: unknown): unknown {
218 if (value === undefined) return value;
219 if (typeof value === "string") return truncateDetail(value, MAX_ON_DEMAND_BYTES);
220 let json: string;
221 try {
222 json = JSON.stringify(value) ?? String(value);
223 } catch {
224 return "Unserializable tool value";
225 }
226 return Buffer.byteLength(json, "utf8") <= MAX_ON_DEMAND_BYTES
227 ? value
228 : truncateDetail(json, MAX_ON_DEMAND_BYTES);
229}
230
231/** Preserve ordered image markers in truncated output so clients can load images by index. */
232function boundToolResult(output: unknown): unknown {
233 const omitted = omitToolOutputImageData(output);
234 const bounded = boundDynamicValue(omitted);
235 if (typeof bounded !== "string" || typeof omitted === "string") return bounded;
236 const images = toolOutputImages(omitted);
237 return images.length === 0
238 ? bounded
239 : [
240 { type: "text", text: bounded },
241 ...images.map((image) => ({ type: "image", mimeType: image.mimeType })),
242 ];
243}
244
245function boundToolOutput(output: unknown): unknown {
246 if (typeof output !== "object" || output === null || !("result" in output)) {
247 return boundToolResult(output);
248 }
249 const app =
250 MCP_APP_OUTPUT_KEY in output ? readMcpAppReference(output[MCP_APP_OUTPUT_KEY]) : undefined;
251 if (app === undefined) return boundToolResult(output);
252 // The app receives this result, so its image data stays unless it is too large to send.
253 const fits = boundDynamicValue(output.result) === output.result;
254 return {
255 [MCP_APP_OUTPUT_KEY]: app,
256 result: fits ? output.result : boundToolResult(output.result),
257 };
258}
259
260export function projectTurnItemForDetail(item: OrchestrationV2TurnItem): OrchestrationV2TurnItem {
261 switch (item.type) {
262 case "command_execution":
263 return {
264 ...item,
265 input: truncateDetail(item.input, MAX_ON_DEMAND_BYTES) ?? "",
266 output: truncateDetail(item.output, MAX_ON_DEMAND_BYTES),
267 };
268 case "dynamic_tool":
269 return {
270 ...item,
271 input: boundDynamicValue(item.input),
272 output: boundToolOutput(item.output),
273 };
274 case "subagent":
275 return {
276 ...item,
277 prompt: truncateDetail(item.prompt, MAX_ON_DEMAND_BYTES) ?? "",
278 progress: truncateDetail(item.progress, MAX_ON_DEMAND_BYTES),
279 result:
280 item.result === null ? null : (truncateDetail(item.result, MAX_ON_DEMAND_BYTES) ?? null),
281 };
282 case "file_change":
283 return {
284 ...projectTurnItemForWire(item),
285 ...(item.diffStr === undefined
286 ? {}
287 : { diffStr: truncateDetail(item.diffStr, MAX_ON_DEMAND_BYTES) }),
288 ...(!item.diffStr?.trim() && item.oldStr !== undefined
289 ? { oldStr: truncateDetail(item.oldStr, MAX_ON_DEMAND_BYTES) }
290 : {}),
291 ...(!item.diffStr?.trim() && item.newStr !== undefined
292 ? { newStr: truncateDetail(item.newStr, MAX_ON_DEMAND_BYTES) }
293 : {}),
294 };
295 case "handoff":
296 return projectTurnItemForWire(item);
297 default:
298 return item;
299 }
300}
301
302export function projectContextHandoffForWire(
303 handoff: OrchestrationV2ContextHandoff,
304): OrchestrationV2ContextHandoff {
305 const { history: _history, delivery: _delivery, ...projected } = handoff;
306 return { ...projected, summaryText: "" };
307}
308
309export function projectThreadProjectionForWire(
310 projection: OrchestrationV2ThreadProjection,
311): OrchestrationV2ThreadProjection {
312 const projectedById = new Map<string, OrchestrationV2TurnItem>();
313 const project = (item: OrchestrationV2TurnItem) => {
314 const key = `${item.threadId}:${item.id}`;
315 const existing = projectedById.get(key);
316 if (existing !== undefined) return existing;
317 const projected = projectTurnItemForWire(item);
318 projectedById.set(key, projected);
319 return projected;
320 };
321 return {
322 ...projection,
323 checkpoints: projection.checkpoints.map(capCheckpointFiles),
324 contextHandoffs: projection.contextHandoffs.map(projectContextHandoffForWire),
325 turnItems: projection.turnItems.map(project),
326 visibleTurnItems: projection.visibleTurnItems.map((row) => ({
327 ...row,
328 item: project(row.item),
329 })),
330 };
331}
332
333export function projectDomainEventForWire(
334 event: OrchestrationV2DomainEvent,
335): OrchestrationV2DomainEvent {
336 return event.type === "turn-item.updated"
337 ? { ...event, payload: projectTurnItemForWire(event.payload) }
338 : event.type === "context-handoff.updated"
339 ? { ...event, payload: projectContextHandoffForWire(event.payload) }
340 : event.type === "checkpoint.captured"
341 ? { ...event, payload: capCheckpointFiles(event.payload) }
342 : event;
343}
344
345// Keyed on the stored event PubSub hands every subscriber, so all of them share one projection
346// and LiveStreamBudget sizes it once. Consumers must not mutate the shared result.
347const projectedStoredEvents = new WeakMap<OrchestrationV2StoredEvent, OrchestrationV2StoredEvent>();
348
349export function projectStoredEventForWire(
350 stored: OrchestrationV2StoredEvent,
351): OrchestrationV2StoredEvent {
352 const cached = projectedStoredEvents.get(stored);
353 if (cached !== undefined) return cached;
354 const event = projectDomainEventForWire(stored.event);
355 if (event === stored.event) return stored;
356 const projected = { ...stored, event };
357 projectedStoredEvents.set(stored, projected);
358 return projected;
359}
360
361type ThreadStreamEvent = Extract<OrchestrationV2ThreadStreamItem, { readonly kind: "event" }>;
362
363const threadStreamEvents = new WeakMap<OrchestrationV2StoredEvent, ThreadStreamEvent>();
364
365/** For an event already projected for the wire; subscribers share the item and its cached size. */
366export function threadStreamEventItem(stored: OrchestrationV2StoredEvent): ThreadStreamEvent {
367 const cached = threadStreamEvents.get(stored);
368 if (cached !== undefined) return cached;
369 const item = { kind: "event" as const, sequence: stored.sequence, event: stored.event };
370 threadStreamEvents.set(stored, item);
371 return item;
372}
373
374const LIVE_DEDUPE_MEMORY = 32;
375
376function remember<K, V>(memory: Map<K, V>, key: K, value: V): void {
377 memory.delete(key);
378 memory.set(key, value);
379 if (memory.size > LIVE_DEDUPE_MEMORY) memory.delete(memory.keys().next().value as K);
380}
381
382/**
383 * One per subscription, after coalescing, so it remembers only records the client received. A
384 * record event comes before the turn item that repeats its body, so the item (or a repeated
385 * subagent record) ships blank and marked. The other order, or a forgotten record, only costs
386 * bytes: the repeat ships whole.
387 */
388export function makeLiveBodyDeduper(): (
389 item: OrchestrationV2ThreadStreamItem,
390) => OrchestrationV2ThreadStreamItem {
391 const messageText = new Map<MessageId, string>();
392 const checkpointFiles = new Map<CheckpointId, OrchestrationV2Checkpoint["files"]>();
393 const subagents = new Map<NodeId, Pick<OrchestrationV2Subagent, "prompt" | "result">>();
394 const source = {
395 messageText: (id: MessageId) => messageText.get(id),
396 checkpointFiles: (id: CheckpointId) => checkpointFiles.get(id),
397 subagent: (id: NodeId) => subagents.get(id),
398 };
399 const rememberSubagent = ({ id, prompt, result }: OrchestrationV2Subagent) =>
400 remember(subagents, id, { prompt, result });
401 return (item) => {
402 if (item.kind === "snapshot") {
403 // The client now holds exactly these records; their bodies are never blanked.
404 const { messages, checkpoints, subagents: records } = item.projection;
405 for (const message of messages) remember(messageText, message.id, message.text);
406 for (const checkpoint of checkpoints)
407 remember(checkpointFiles, checkpoint.id, checkpoint.files);
408 for (const subagent of records) rememberSubagent(subagent);
409 return item;
410 }
411 if (item.kind !== "event") return item;
412 const { event } = item;
413 switch (event.type) {
414 case "message.updated":
415 remember(messageText, event.payload.id, event.payload.text);
416 return item;
417 case "checkpoint.captured":
418 remember(checkpointFiles, event.payload.id, event.payload.files);
419 return item;
420 case "subagent.updated": {
421 const payload = dedupeSubagentBodies(event.payload, subagents.get(event.payload.id));
422 rememberSubagent(event.payload);
423 return payload === event.payload ? item : { ...item, event: { ...event, payload } };
424 }
425 case "turn-item.updated": {
426 const payload = dedupeTurnItemBodies(event.payload, source);
427 return payload === event.payload ? item : { ...item, event: { ...event, payload } };
428 }
429 default:
430 return item;
431 }
432 };
433}