fix(server): bound every timeline item so one fetch cannot close the socket

A timeline page is one WebSocket message. Only three tool-call fields were
capped, so MCP results, file contents, errors and merged assistant rows could
make a 20-entry or even a 1-entry fetch pass the outbound high-water mark.

limitAgentTimelineItemContent is now one budgeted walk over any item: 64 KiB
for tool calls, 256 KiB for other items. Over budget, the largest strings are
cut to a shared length with a visible marker and oversized data URLs become a
marker. Fetches also apply it to merged assistant and reasoning rows.
This commit is contained in:
Mohamed Boudra
2026-10-30 08:03:21 +01:00
parent a8e0effb32
commit 40ba2cee69
10 changed files with 277 additions and 116 deletions
@@ -6,6 +6,7 @@ import { afterEach, describe, expect, test, vi } from "vitest";
import { createTestLogger } from "../../test-utils/test-logger.js";
import { AgentManager, type AgentManagerEvent } from "./agent-manager.js";
import { AGENT_STREAM_COALESCE_DEFAULT_WINDOW_MS } from "./agent-stream-coalescer.js";
import { limitAgentTimelineItemContent } from "./agent-timeline-content.js";
import type { AgentTimelineRow } from "./agent-timeline-store-types.js";
import { projectTimelineRows } from "./timeline-projection.js";
import type {
@@ -31,7 +32,6 @@ import type {
const COALESCE_WINDOW_MS = AGENT_STREAM_COALESCE_DEFAULT_WINDOW_MS;
const BEFORE_COALESCE_WINDOW_MS = Math.max(COALESCE_WINDOW_MS - 1, 0);
const TOOL_CALL_CONTENT_MAX_LENGTH = 64 * 1024;
const TEST_CAPABILITIES: AgentCapabilityFlags = {
supportsStreaming: false,
@@ -400,10 +400,7 @@ describe("target coalesced behavior", () => {
try {
const { agentId, session } = await createManagedSession(harness);
const output = `${"a".repeat(512 * 1024)}${"z".repeat(512 * 1024)}`;
const expectedItem = toolCall({
status: "completed",
output: "a".repeat(TOOL_CALL_CONTENT_MAX_LENGTH),
});
const expectedItem = limitAgentTimelineItemContent(toolCall({ status: "completed", output }));
session.pushEvent(timelineEvent(toolCall({ status: "completed", output })));
await waitForSessionEventQueue();
@@ -425,10 +422,7 @@ describe("target coalesced behavior", () => {
try {
const { agentId } = await createManagedSession(harness);
const output = `${"a".repeat(512 * 1024)}${"z".repeat(512 * 1024)}`;
const expectedItem = toolCall({
status: "completed",
output: "a".repeat(TOOL_CALL_CONTENT_MAX_LENGTH),
});
const expectedItem = limitAgentTimelineItemContent(toolCall({ status: "completed", output }));
await harness.manager.appendTimelineItem(agentId, toolCall({ status: "completed", output }));
@@ -449,10 +443,7 @@ describe("target coalesced behavior", () => {
try {
const { agentId, session } = await createManagedSession(harness);
const output = `${"a".repeat(512 * 1024)}${"z".repeat(512 * 1024)}`;
const expectedItem = toolCall({
status: "completed",
output: "a".repeat(TOOL_CALL_CONTENT_MAX_LENGTH),
});
const expectedItem = limitAgentTimelineItemContent(toolCall({ status: "completed", output }));
session.setHistory([timelineEvent(toolCall({ status: "completed", output }))]);
await harness.manager.hydrateTimelineFromProvider(agentId, { force: true });
@@ -470,10 +461,7 @@ describe("target coalesced behavior", () => {
try {
const { agentId } = await createManagedSession(harness);
const output = `${"a".repeat(512 * 1024)}${"z".repeat(512 * 1024)}`;
const expectedItem = toolCall({
status: "completed",
output: "a".repeat(TOOL_CALL_CONTENT_MAX_LENGTH),
});
const expectedItem = limitAgentTimelineItemContent(toolCall({ status: "completed", output }));
await harness.manager.emitLiveTimelineItem(
agentId,
@@ -495,10 +483,9 @@ describe("target coalesced behavior", () => {
try {
const { agentId, session } = await createManagedSession(harness);
const content = `${"a".repeat(512 * 1024)}${"z".repeat(512 * 1024)}`;
const expectedItem = toolCall({
status: "failed",
error: { content: "a".repeat(TOOL_CALL_CONTENT_MAX_LENGTH) },
});
const expectedItem = limitAgentTimelineItemContent(
toolCall({ status: "failed", error: { content } }),
);
session.pushEvent(timelineEvent(toolCall({ status: "failed", error: { content } })));
await waitForSessionEventQueue();
@@ -21,6 +21,7 @@ import {
} from "./agent-manager.js";
import { AgentStorage } from "./agent-storage.js";
import { InMemoryAgentTimelineStore } from "./agent-timeline-store.js";
import { limitAgentTimelineItemContent } from "./agent-timeline-content.js";
import { toAgentPayload } from "./agent-projections.js";
import { projectTimelineRows } from "./timeline-projection.js";
import { getOpenAgentTabLabel, PARENT_AGENT_ID_LABEL } from "@getpaseo/protocol/agent-labels";
@@ -4592,7 +4593,7 @@ test("importProviderSession imports the selected session without listing and pub
expect(manager.getTimeline(imported.id)).toEqual([
{ type: "user_message", text: "Trace provider imports" },
{ type: "assistant_message", text: "Done" },
{
limitAgentTimelineItemContent({
type: "tool_call",
callId: "large-shell-result",
name: "shell",
@@ -4601,10 +4602,10 @@ test("importProviderSession imports the selected session without listing and pub
detail: {
type: "shell",
command: "print output",
output: "x".repeat(64 * 1024),
output: "x".repeat(1024 * 1024),
exitCode: 0,
},
},
}),
]);
expect(manager.listProviderSubagents(imported.id)).toEqual([
expect.objectContaining({ id: "thread-child", title: "Imported child", status: "completed" }),
@@ -11957,10 +11958,7 @@ test("commits startup notices once on create and after restored history", async
exitCode: 0,
},
};
const limitedToolOutput: AgentTimelineItem = {
...toolOutput,
detail: { type: "shell", command: "print output", output: "x".repeat(64 * 1024), exitCode: 0 },
};
const limitedToolOutput = limitAgentTimelineItemContent(toolOutput);
class NotifyingSession extends TestAgentSession {
readonly initialTimeline = [{ item: notice, timestamp: "2026-09-22T00:00:00.000Z" }];
override async *streamHistory(): AsyncGenerator<AgentStreamEvent> {
@@ -1,12 +1,29 @@
import { describe, expect, test } from "vitest";
import type { AgentTimelineItem } from "./agent-sdk-types.js";
import { limitAgentTimelineItemContent } from "./agent-timeline-content.js";
function serializedBytes(item: AgentTimelineItem): number {
return Buffer.byteLength(JSON.stringify(item), "utf8");
}
describe("agent timeline content", () => {
test("returns an item within budget unchanged", () => {
const item: AgentTimelineItem = {
type: "tool_call",
callId: "small",
name: "Bash",
status: "completed",
error: null,
detail: { type: "shell", command: "ls", output: "x".repeat(60 * 1024) },
};
expect(limitAgentTimelineItemContent(item)).toBe(item);
});
test.each([false, true])(
"preserves the exact UTF-16 prefix for oversized shell output (failed=%s)",
"cuts oversized shell output and keeps short fields whole (failed=%s)",
(failed) => {
const text = "x".repeat(64 * 1024 - 1) + "🙂";
const text = "x".repeat(1024 * 1024) + "🙂";
const item = limitAgentTimelineItemContent({
type: "tool_call",
callId: "large-shell",
@@ -15,45 +32,74 @@ describe("agent timeline content", () => {
detail: { type: "shell", command: "echo large", output: text },
error: failed ? { content: text } : null,
});
expect(item).toEqual({
type: "tool_call",
expect(item).toMatchObject({
callId: "large-shell",
name: "Bash",
status: failed ? "failed" : "completed",
detail: { type: "shell", command: "echo large", output: text.slice(0, 64 * 1024) },
error: failed ? { content: text.slice(0, 64 * 1024) } : null,
detail: { type: "shell", command: "echo large" },
});
if (item.type !== "tool_call" || item.detail.type !== "shell") throw new Error("shape");
expect(item.detail.output).toMatch(/^x+…\[truncated \d{3}\.\d KB\]$/);
expect(serializedBytes(item)).toBeLessThanOrEqual(64 * 1024);
},
);
test("marks truncated plain-text tool content within the content budget", () => {
const oversizedInput = "x".repeat(64 * 1024 + 1);
test("replaces an oversized data URL in an unknown tool result with a marker", () => {
const url = `data:image/jpeg;base64,${"A".repeat(14 * 1024 * 1024)}`;
const item = limitAgentTimelineItemContent({
type: "tool_call",
callId: "terminal-session-4242",
name: "terminal",
callId: "screenshot",
name: "node_repl.js",
status: "completed",
error: null,
detail: {
type: "plain_text",
text: oversizedInput,
icon: "square_terminal",
type: "unknown",
input: { index: 1 },
output: { content: [{ type: "image", url }], note: "captured" },
},
});
expect(limitAgentTimelineItemContent(item)).toEqual(item);
expect(item).toEqual({
type: "tool_call",
callId: "terminal-session-4242",
name: "terminal",
callId: "screenshot",
name: "node_repl.js",
status: "completed",
error: null,
detail: {
type: "plain_text",
text: "x".repeat(64 * 1024 - "\n\n[Content truncated]".length) + "\n\n[Content truncated]",
icon: "square_terminal",
type: "unknown",
input: { index: 1 },
output: { content: [{ type: "image", url: "…[truncated 14.0 MB]" }], note: "captured" },
},
});
});
test("gives non-tool items a 256 KiB budget", () => {
const text = "a".repeat(10 * 1024 * 1024);
const item = limitAgentTimelineItemContent({ type: "assistant_message", text });
if (item.type !== "assistant_message") throw new Error("shape");
expect(item.text).toMatch(/^a+…\[truncated 9\.\d MB\]$/);
expect(item.text.length).toBeGreaterThan(200 * 1024);
expect(serializedBytes(item)).toBeLessThanOrEqual(256 * 1024);
});
test("bounds an item whose structure alone exceeds the budget", () => {
const item = limitAgentTimelineItemContent({
type: "tool_call",
callId: "many",
name: "query",
status: "completed",
error: null,
detail: { type: "unknown", input: null, output: Array.from({ length: 1_000_000 }, () => 1) },
});
if (item.type !== "tool_call" || item.detail.type !== "unknown") throw new Error("shape");
const output = item.detail.output as unknown[];
expect(output.at(-1)).toMatch(/^…\[truncated \d+ items\]$/);
expect(serializedBytes(item)).toBeLessThanOrEqual(64 * 1024);
});
test("is stable on an already-limited item", () => {
const item = limitAgentTimelineItemContent({
type: "reasoning",
text: "r".repeat(1024 * 1024),
});
expect(limitAgentTimelineItemContent(item)).toBe(item);
});
});
@@ -1,9 +1,16 @@
import type { AgentTimelineItem } from "./agent-sdk-types.js";
import type { JsonValue } from "@getpaseo/protocol/agent-types";
const TOOL_CALL_CONTENT_MAX_LENGTH = 64 * 1024;
export const PLUGIN_TIMELINE_DATA_MAX_BYTES = 64 * 1024;
// One timeline fetch sends its entries in one WebSocket message, so every item must stay small.
// Budgets count UTF-16 units plus a fixed cost per node; neither is exact JSON bytes.
const TOOL_CALL_BUDGET = 64 * 1024;
const ITEM_BUDGET = 256 * 1024;
const NODE_COST = 8;
const MARKER_COST = 32;
const MAX_DEPTH = 64;
export function assertPluginTimelineDataSize(data: JsonValue): void {
const serializedBytes = Buffer.byteLength(JSON.stringify(data), "utf8");
if (serializedBytes > PLUGIN_TIMELINE_DATA_MAX_BYTES) {
@@ -17,66 +24,117 @@ function retainTextPrefix(text: string, length: number): string {
return Buffer.from(text.slice(0, length), "utf16le").toString("utf16le");
}
function limitFailedShellError(item: AgentTimelineItem): AgentTimelineItem {
if (
item.type !== "tool_call" ||
item.detail.type !== "shell" ||
item.status !== "failed" ||
typeof item.error !== "object" ||
item.error === null ||
!("content" in item.error) ||
typeof item.error.content !== "string" ||
item.error.content.length <= TOOL_CALL_CONTENT_MAX_LENGTH
) {
return item;
}
return {
...item,
error: {
...item.error,
content: retainTextPrefix(item.error.content, TOOL_CALL_CONTENT_MAX_LENGTH),
},
};
function formatSize(units: number): string {
if (units < 1024) return `${units} B`;
if (units < 1024 * 1024) return `${(units / 1024).toFixed(1)} KB`;
return `${(units / (1024 * 1024)).toFixed(1)} MB`;
}
function limitPlainText(item: AgentTimelineItem): AgentTimelineItem {
if (
item.type !== "tool_call" ||
item.detail.type !== "plain_text" ||
typeof item.detail.text !== "string" ||
item.detail.text.length <= TOOL_CALL_CONTENT_MAX_LENGTH
) {
return item;
}
const notice = "\n\n[Content truncated]";
return {
...item,
detail: {
...item.detail,
text:
retainTextPrefix(item.detail.text, TOOL_CALL_CONTENT_MAX_LENGTH - notice.length) + notice,
},
};
function truncatedMarker(units: number): string {
return `…[truncated ${formatSize(units)}]`;
}
interface ItemSize {
nodeCost: number;
stringLengths: number[];
tooDeep: boolean;
}
function measure(value: unknown, depth: number, size: ItemSize): void {
size.nodeCost += NODE_COST;
if (depth > MAX_DEPTH) {
size.tooDeep = true;
} else if (typeof value === "string") {
size.stringLengths.push(value.length);
} else if (Array.isArray(value)) {
for (const element of value) measure(element, depth + 1, size);
} else if (typeof value === "object" && value !== null) {
for (const [key, child] of Object.entries(value)) {
size.nodeCost += key.length;
measure(child, depth + 1, size);
}
}
}
/** The largest per-string length that keeps all strings, once cut, within `available`. */
function findStringCap(lengths: readonly number[], available: number): number {
let low = 0;
let high = 0;
for (const length of lengths) high = Math.max(high, length);
while (low < high) {
const cap = Math.ceil((low + high) / 2);
let total = 0;
for (const length of lengths) total += Math.min(length, cap);
if (total <= available) low = cap;
else high = cap - 1;
}
return low;
}
interface CutState {
stringCap: number;
remaining: number;
}
function cutString(text: string, cap: number): string {
if (text.length <= cap) return text;
// A partial data URL is useless to every reader, so drop it whole.
if (text.startsWith("data:")) return truncatedMarker(text.length);
return retainTextPrefix(text, cap) + truncatedMarker(text.length - cap);
}
function cut(value: unknown, depth: number, state: CutState): unknown {
state.remaining -= NODE_COST;
if (depth > MAX_DEPTH) return "…[truncated]";
if (typeof value === "string") {
const text = cutString(value, state.stringCap);
state.remaining -= text.length;
return text;
}
if (Array.isArray(value)) {
let changed = false;
const output: unknown[] = [];
for (const [index, element] of value.entries()) {
if (state.remaining <= 0) {
output.push(`…[truncated ${value.length - index} items]`);
changed = true;
break;
}
const next = cut(element, depth + 1, state);
changed ||= next !== element;
output.push(next);
}
return changed ? output : value;
}
if (typeof value === "object" && value !== null) {
let changed = false;
const output: Record<string, unknown> = {};
for (const [key, child] of Object.entries(value)) {
state.remaining -= key.length;
const next = cut(child, depth + 1, state);
changed ||= next !== child;
output[key] = next;
}
return changed ? output : value;
}
return value;
}
/** Returns the item itself when it fits its budget; otherwise a copy with its largest strings cut. */
export function limitAgentTimelineItemContent(item: AgentTimelineItem): AgentTimelineItem {
item = limitFailedShellError(item);
item = limitPlainText(item);
if (
item.type !== "tool_call" ||
item.detail.type !== "shell" ||
typeof item.detail.output !== "string"
) {
return item;
}
if (item.detail.output.length <= TOOL_CALL_CONTENT_MAX_LENGTH) {
return item;
}
return {
...item,
detail: {
...item.detail,
output: retainTextPrefix(item.detail.output, TOOL_CALL_CONTENT_MAX_LENGTH),
},
};
const budget = item.type === "tool_call" ? TOOL_CALL_BUDGET : ITEM_BUDGET;
const size: ItemSize = { nodeCost: 0, stringLengths: [], tooDeep: false };
measure(item, 0, size);
let stringTotal = 0;
for (const length of size.stringLengths) stringTotal += length;
if (!size.tooDeep && size.nodeCost + stringTotal <= budget) return item;
// Strings get what structure leaves, but at least half the budget, less a margin for JSON
// escaping. Arrays stop once the copy has spent the budget, which bounds items whose
// structure alone is too big.
const markerCost =
MARKER_COST * size.stringLengths.filter((length) => length > MARKER_COST).length;
const available = (budget * 15) / 16 - Math.min(size.nodeCost, budget / 2) - markerCost;
const stringCap = Math.max(MARKER_COST, findStringCap(size.stringLengths, available));
return cut(item, 0, { stringCap, remaining: budget }) as AgentTimelineItem;
}
@@ -181,6 +181,20 @@ describe("projected timeline retention", () => {
},
]);
});
it("serves a merged assistant message within the item budget", () => {
const store = new InMemoryAgentTimelineStore();
store.initialize("agent", { epoch: "e" });
const chunk = "a".repeat(200 * 1024);
for (let index = 0; index < 10; index++) {
store.append("agent", { type: "assistant_message", messageId: "m", text: chunk });
}
const [row] = store.fetch("agent", { limit: 1 }).rows;
expect(row).toMatchObject({ seqStart: 1, seqEnd: 10 });
expect(row.item).toMatchObject({ type: "assistant_message", messageId: "m" });
expect(Buffer.byteLength(JSON.stringify(row.item))).toBeLessThanOrEqual(256 * 1024);
expect(store.getLastAssistantMessage("agent")).toHaveLength(10 * chunk.length);
});
});
describe("projected sequence ownership", () => {
@@ -1,5 +1,6 @@
import { randomUUID } from "node:crypto";
import { TimelineProjection, type ProjectedTimelineRow } from "./timeline-projection.js";
import { limitAgentTimelineItemContent } from "./agent-timeline-content.js";
import type { AgentTimelineItem } from "./agent-sdk-types.js";
import type {
AgentTimelineFetchOptions,
@@ -120,7 +121,12 @@ export class InMemoryAgentTimelineStore {
hasNewer: page.hasNewer,
startSeq: page.startSeq,
endSeq: page.endSeq,
rows: page.entries.map((entry) => Object.assign({ seq: entry.seqEnd }, entry)),
// Stored rows are already within budget; this bounds assistant and reasoning merges.
rows: page.entries.map((entry) =>
Object.assign({ seq: entry.seqEnd }, entry, {
item: limitAgentTimelineItemContent(entry.item),
}),
),
};
}
@@ -80,7 +80,7 @@ describe("ProviderSubagentStore", () => {
const [row] = subagents.fetchTimeline("parent-a", "child-1").rows;
expect(row?.item).toMatchObject({
type: "tool_call",
detail: { type: "shell", output: "x".repeat(64 * 1024) },
detail: { type: "shell", output: expect.stringMatching(/^x+…\[truncated \d+\.\d KB\]$/) },
});
});
@@ -737,7 +737,7 @@ describe("ClaudeAgentSession persisted subagent replay", () => {
);
}, 120_000);
test("limits retained child shell output to the timeline's existing content limit", async () => {
test("limits retained child shell output to the timeline item budget", async () => {
const output = "x".repeat(128 * 1024);
writeSession({
parentLines: [taskToolUse(), taskToolResult()],
@@ -782,7 +782,10 @@ describe("ClaudeAgentSession persisted subagent replay", () => {
type: "timeline",
id: TOOL_USE_ID,
item: expect.objectContaining({
detail: expect.objectContaining({ type: "shell", output: output.slice(0, 64 * 1024) }),
detail: expect.objectContaining({
type: "shell",
output: expect.stringMatching(/^x+…\[truncated \d+\.\d KB\]$/),
}),
}),
}),
]);
@@ -0,0 +1,46 @@
import { mkdtempSync, rmSync } from "node:fs";
import { tmpdir } from "node:os";
import path from "node:path";
import { expect, test } from "vitest";
import { createDaemonTestContext } from "../test-utils/index.js";
// #2220: a timeline page is one WebSocket message, so a few huge entries used to close the socket.
test.each([40, 0])(
"fetching image-heavy history with limit %i keeps the direct connection usable",
async (limit) => {
const ctx = await createDaemonTestContext();
const cwd = mkdtempSync(path.join(tmpdir(), "oversized-timeline-"));
try {
const agent = await ctx.client.createAgent({ provider: "codex", cwd });
const imageUrl = `data:image/jpeg;base64,${"A".repeat(14 * 1024 * 1024)}`;
for (let index = 0; index < 5; index += 1) {
await ctx.daemon.daemon.agentManager.appendTimelineItem(agent.id, {
type: "tool_call",
callId: `screenshot-${index}`,
name: "node_repl.js",
status: "completed",
error: null,
detail: {
type: "unknown",
input: { index },
output: { _meta: { "codex/toolSurface": { screenshot: { url: imageUrl } } } },
},
});
}
const page = await ctx.client.fetchAgentTimeline(agent.id, { direction: "tail", limit });
const screenshots = page.entries.filter((entry) => entry.item.type === "tool_call");
expect(screenshots).toHaveLength(5);
for (const entry of screenshots) {
expect(Buffer.byteLength(JSON.stringify(entry.item))).toBeLessThanOrEqual(64 * 1024);
}
const next = await ctx.client.fetchAgentTimeline(agent.id, { direction: "tail", limit: 1 });
expect(next.entries).toHaveLength(1);
} finally {
await ctx.cleanup();
rmSync(cwd, { recursive: true, force: true });
}
},
60_000,
);