jb/paseo
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Keep oversized tool output out of chat timelines (#2020)

Browse Source 
* 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
This commit is contained in:
Mohamed Boudra
2026-07-12 14:24:18 +00:00
committed by GitHub
parent e18cfb7639
commit cf6c014b6b
5 changed files with 224 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/server/agent/agent-manager-stream-coalescing.test.ts
+117
View File
@@ -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();
packages/server/src/server/agent/agent-manager.test.ts
+29
View File
@@ -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({
packages/server/src/server/agent/agent-manager.ts
+27 -7
View File
@@ -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<void> {
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<boolean> {
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(
{
packages/server/src/server/agent/agent-timeline-content.ts
+46
View File
@@ -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),
},
};
}