Compare commits

...

4 Commits

Author SHA1 Message Date
Mohamed Boudra
0c05f69402 fix(server): cap shell timeline text by UTF-8 bytes 2026-07-12 14:28:33 +00:00
Mohamed Boudra
8fdc41f06c fix(server): cover imported and failed shell output 2026-07-12 16:08:15 +02:00
Mohamed Boudra
928c124356 refactor(server): slice oversized tool output directly 2026-07-12 16:01:11 +02:00
Mohamed Boudra
0a398833d8 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.
2026-07-12 15:55:13 +02:00
5 changed files with 271 additions and 7 deletions

View File

@@ -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 and failed
shell error text are capped at 64 KiB of UTF-8 data, 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:

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_BYTES = 64 * 1024;
const TEST_CAPABILITIES: AgentCapabilityFlags = {
supportsStreaming: false,
@@ -377,6 +378,148 @@ 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_BYTES),
});
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 tool output by UTF-8 bytes", async () => {
const harness = createHarness();
try {
const { agentId, session } = await createManagedSession(harness);
const output = "漢".repeat(TOOL_CALL_CONTENT_MAX_BYTES);
const expectedItem = toolCall({
status: "completed",
output: "漢".repeat(Math.floor(TOOL_CALL_CONTENT_MAX_BYTES / 3)),
});
session.pushEvent(timelineEvent(toolCall({ status: "completed", output })));
await waitForSessionEventQueue();
expect(getTimelineItems(await harness.manager.getTimelineRows(agentId))).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_BYTES),
});
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_BYTES),
});
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_BYTES),
});
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 error content and message", 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_BYTES),
message: "a".repeat(TOOL_CALL_CONTENT_MAX_BYTES),
},
});
session.pushEvent(
timelineEvent(toolCall({ status: "failed", error: { content, message: 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();

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({

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(
{

View File

@@ -0,0 +1,67 @@
import { StringDecoder } from "node:string_decoder";
import type { AgentTimelineItem } from "./agent-sdk-types.js";
const TOOL_CALL_CONTENT_MAX_BYTES = 64 * 1024;
function limitTextContent(value: string): string {
if (Buffer.byteLength(value, "utf8") <= TOOL_CALL_CONTENT_MAX_BYTES) {
return value;
}
const bytes = Buffer.from(value, "utf8").subarray(0, TOOL_CALL_CONTENT_MAX_BYTES);
return new StringDecoder("utf8").write(bytes);
}
function limitFailedShellError(item: AgentTimelineItem): AgentTimelineItem {
if (
item.type !== "tool_call" ||
item.detail.type !== "shell" ||
item.status !== "failed" ||
typeof item.error !== "object" ||
item.error === null
) {
return item;
}
const error: Record<string, unknown> = { ...item.error };
let changed = false;
for (const key of ["content", "message"] as const) {
const value = error[key];
if (typeof value !== "string") {
continue;
}
const limitedValue = limitTextContent(value);
if (limitedValue !== value) {
error[key] = limitedValue;
changed = true;
}
}
if (!changed) {
return item;
}
return {
...item,
error,
};
}
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;
}
const output = limitTextContent(item.detail.output);
if (output === item.detail.output) {
return item;
}
return {
...item,
detail: {
...item.detail,
output,
},
};
}