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