From cf6c014b6b99d62d51ede9d2c17f3c67e45c287b Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Sun, 12 Jul 2026 16:24:18 +0200 Subject: [PATCH] Keep oversized tool output out of chat timelines (#2020) * fix(server): bound shell tool output in timelines Provider tool output could enter timeline persistence and live streams without a size limit. Bound canonical shell output before coalescing, storage, history hydration, and dispatch so every provider follows the same 64 KiB budget. * refactor(server): slice oversized tool output directly * fix(server): cover imported and failed shell output --- docs/timeline-sync.md | 5 + .../agent-manager-stream-coalescing.test.ts | 117 ++++++++++++++++++ .../src/server/agent/agent-manager.test.ts | 29 +++++ .../server/src/server/agent/agent-manager.ts | 34 +++-- .../server/agent/agent-timeline-content.ts | 46 +++++++ 5 files changed, 224 insertions(+), 7 deletions(-) create mode 100644 packages/server/src/server/agent/agent-timeline-content.ts diff --git a/docs/timeline-sync.md b/docs/timeline-sync.md index f2b9bf6b1..86414257a 100644 --- a/docs/timeline-sync.md +++ b/docs/timeline-sync.md @@ -9,6 +9,11 @@ The invariant is: > If the daemon has committed timeline rows for an agent, any connected client that opens or resumes that agent eventually displays every row through the daemon's current tail. +Tool output is bounded before it enters either delivery path. Canonical shell tool output is sliced +to 64 KiB, and the same bounded item is used for durable timeline rows and live stream events. +Provider history hydration applies the same rule so reopening an agent cannot restore an oversized +tool payload. + ## Presence is not delivery Client heartbeat reports presence: diff --git a/packages/server/src/server/agent/agent-manager-stream-coalescing.test.ts b/packages/server/src/server/agent/agent-manager-stream-coalescing.test.ts index b34e9a31b..6e7f813c5 100644 --- a/packages/server/src/server/agent/agent-manager-stream-coalescing.test.ts +++ b/packages/server/src/server/agent/agent-manager-stream-coalescing.test.ts @@ -30,6 +30,7 @@ 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, @@ -377,6 +378,122 @@ afterEach(() => { }); describe("target coalesced behavior", () => { + test("bounds tool output before persisting and streaming it", async () => { + const harness = createHarness(); + 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), + }); + + session.pushEvent(timelineEvent(toolCall({ status: "completed", output }))); + await waitForSessionEventQueue(); + + const rows = await harness.manager.getTimelineRows(agentId); + const events = getTimelineStreamEvents(harness.events, agentId); + + expect(getTimelineItems(rows)).toEqual([expectedItem]); + expect( + events.map((event) => (event.type === "agent_stream" ? event.event.item : null)), + ).toEqual([expectedItem]); + } finally { + harness.cleanup(); + } + }); + + test("bounds appended tool output before persisting and streaming it", async () => { + const harness = createHarness(); + 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), + }); + + await harness.manager.appendTimelineItem(agentId, toolCall({ status: "completed", output })); + + const rows = await harness.manager.getTimelineRows(agentId); + const events = getTimelineStreamEvents(harness.events, agentId); + + expect(getTimelineItems(rows)).toEqual([expectedItem]); + expect( + events.map((event) => (event.type === "agent_stream" ? event.event.item : null)), + ).toEqual([expectedItem]); + } finally { + harness.cleanup(); + } + }); + + test("bounds tool output while hydrating provider history", async () => { + const harness = createHarness(); + 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), + }); + session.setHistory([timelineEvent(toolCall({ status: "completed", output }))]); + + await harness.manager.hydrateTimelineFromProvider(agentId); + + expect(getTimelineItems(await harness.manager.getTimelineRows(agentId))).toEqual([ + expectedItem, + ]); + } finally { + harness.cleanup(); + } + }); + + test("bounds tool output emitted only to the live stream", async () => { + const harness = createHarness(); + 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), + }); + + await harness.manager.emitLiveTimelineItem( + agentId, + toolCall({ status: "completed", output }), + ); + + const events = getTimelineStreamEvents(harness.events, agentId); + expect(await harness.manager.getTimelineRows(agentId)).toEqual([]); + expect( + events.map((event) => (event.type === "agent_stream" ? event.event.item : null)), + ).toEqual([expectedItem]); + } finally { + harness.cleanup(); + } + }); + + test("bounds failed shell output carried in the error", async () => { + const harness = createHarness(); + 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) }, + }); + + session.pushEvent(timelineEvent(toolCall({ status: "failed", error: { content } }))); + await waitForSessionEventQueue(); + + expect(getTimelineItems(await harness.manager.getTimelineRows(agentId))).toEqual([ + expectedItem, + ]); + } finally { + harness.cleanup(); + } + }); + test(`coalesces a same-tick assistant burst after the ${COALESCE_WINDOW_MS}ms window`, async () => { vi.useFakeTimers(); const harness = createHarness(); diff --git a/packages/server/src/server/agent/agent-manager.test.ts b/packages/server/src/server/agent/agent-manager.test.ts index c8d4e89c1..64c4689e7 100644 --- a/packages/server/src/server/agent/agent-manager.test.ts +++ b/packages/server/src/server/agent/agent-manager.test.ts @@ -2314,6 +2314,22 @@ test("importProviderSession imports the selected session without listing and pub item: { type: "assistant_message" as const, text: "Done" }, timestamp: "2026-01-02T00:00:01.000Z", }, + { + item: { + type: "tool_call" as const, + callId: "large-shell-result", + name: "shell", + status: "completed" as const, + error: null, + detail: { + type: "shell" as const, + command: "print output", + output: "x".repeat(1024 * 1024), + exitCode: 0, + }, + }, + timestamp: "2026-01-02T00:00:02.000Z", + }, ], }; } @@ -2343,6 +2359,19 @@ 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" }, + { + type: "tool_call", + callId: "large-shell-result", + name: "shell", + status: "completed", + error: null, + detail: { + type: "shell", + command: "print output", + output: "x".repeat(64 * 1024), + exitCode: 0, + }, + }, ]); expect(events).toHaveLength(1); expect(events[0]).toMatchObject({ diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index 232f5c57a..690cfe2a2 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -58,6 +58,7 @@ import { AGENT_STREAM_COALESCE_DEFAULT_WINDOW_MS, AgentStreamCoalescer, } from "./agent-stream-coalescer.js"; +import { limitAgentTimelineItemContent } from "./agent-timeline-content.js"; import { ForegroundRunState, type ForegroundTurnWaiter } from "./foreground-run-state.js"; import { getAgentProviderDefinition } from "@getpaseo/protocol/provider-manifest"; import { invokeRewindCapability, type RewindMode } from "./rewind/rewind.js"; @@ -493,7 +494,7 @@ function buildImportedTimelineRows(entries: readonly ImportedTimelineEntry[]): A rows.push({ seq: rows.length + 1, timestamp: entry.timestamp ?? new Date().toISOString(), - item: entry.item, + item: limitAgentTimelineItemContent(entry.item), }); } return rows; @@ -1752,6 +1753,7 @@ export class AgentManager { async appendTimelineItem(agentId: string, item: AgentTimelineItem): Promise { const agent = this.requireAgent(agentId); + item = limitAgentTimelineItemContent(item); this.touchUpdatedAt(agent); const row = this.recordTimeline(agentId, item); this.dispatchStream( @@ -2990,17 +2992,22 @@ export class AgentManager { agent.historyPrimed = true; for (const event of historyEvents) { + const item = limitAgentTimelineItemContent(event.item); const row = this.recordTimeline( agent.id, - event.item, + item, event.timestamp ? { timestamp: event.timestamp } : undefined, ); if (options?.broadcast) { - this.dispatchStream(agent.id, event, { - seq: row.seq, - epoch: this.timelineStore.getEpoch(agent.id), - timestamp: row.timestamp, - }); + this.dispatchStream( + agent.id, + { ...event, item }, + { + seq: row.seq, + epoch: this.timelineStore.getEpoch(agent.id), + timestamp: row.timestamp, + }, + ); } } this.touchUpdatedAt(agent); @@ -3057,6 +3064,12 @@ export class AgentManager { event: AgentStreamEvent, options?: HandleStreamEventOptions, ): Promise { + if (event.type === "timeline") { + event = { + ...event, + item: limitAgentTimelineItemContent(event.item), + }; + } const eventTurnId = getAgentStreamEventTurnId(event); const isForegroundEvent = Boolean(eventTurnId && agent.activeForegroundTurnId === eventTurnId); this.traceHandleStreamEventStart(agent, event, eventTurnId, isForegroundEvent); @@ -3553,6 +3566,7 @@ export class AgentManager { item: AgentTimelineItem, options?: { timestamp?: string }, ): AgentTimelineRow { + item = limitAgentTimelineItemContent(item); const row = this.timelineStore.append(agentId, item, options); this.enqueueDurableTimelineAppend(agentId, row); return row; @@ -3742,6 +3756,12 @@ export class AgentManager { event: AgentStreamEvent, metadata?: { seq?: number; epoch?: string; timestamp?: string }, ): void { + if (event.type === "timeline") { + event = { + ...event, + item: limitAgentTimelineItemContent(event.item), + }; + } const agent = this.agents.get(agentId); this.logger.trace( { diff --git a/packages/server/src/server/agent/agent-timeline-content.ts b/packages/server/src/server/agent/agent-timeline-content.ts new file mode 100644 index 000000000..5bdc17e0a --- /dev/null +++ b/packages/server/src/server/agent/agent-timeline-content.ts @@ -0,0 +1,46 @@ +import type { AgentTimelineItem } from "./agent-sdk-types.js"; + +const TOOL_CALL_CONTENT_MAX_LENGTH = 64 * 1024; + +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: item.error.content.slice(0, TOOL_CALL_CONTENT_MAX_LENGTH), + }, + }; +} + +export function limitAgentTimelineItemContent(item: AgentTimelineItem): AgentTimelineItem { + item = limitFailedShellError(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: item.detail.output.slice(0, TOOL_CALL_CONTENT_MAX_LENGTH), + }, + }; +}