mirror of
https://github.com/getpaseo/paseo.git
synced 2026-07-29 12:01:31 +00:00
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.
This commit is contained in:
@@ -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 keeps
|
||||
the head and tail within a 64 KiB UTF-8 budget, 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:
|
||||
|
||||
@@ -30,6 +30,8 @@ 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 TOOL_CALL_CONTENT_TRUNCATION_MARKER = "\n...<tool output truncated in the middle>...\n";
|
||||
|
||||
const TEST_CAPABILITIES: AgentCapabilityFlags = {
|
||||
supportsStreaming: false,
|
||||
@@ -83,6 +85,15 @@ function toolCall(options?: {
|
||||
};
|
||||
}
|
||||
|
||||
function boundedToolCallOutput(head: string, tail: string): string {
|
||||
const availableBytes =
|
||||
TOOL_CALL_CONTENT_MAX_BYTES - Buffer.byteLength(TOOL_CALL_CONTENT_TRUNCATION_MARKER);
|
||||
const headBytes = Math.floor(availableBytes / 2);
|
||||
return `${head.repeat(headBytes)}${TOOL_CALL_CONTENT_TRUNCATION_MARKER}${tail.repeat(
|
||||
availableBytes - headBytes,
|
||||
)}`;
|
||||
}
|
||||
|
||||
class TestAgentSession implements AgentSession {
|
||||
readonly capabilities = TEST_CAPABILITIES;
|
||||
readonly id: string;
|
||||
@@ -377,6 +388,101 @@ 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: boundedToolCallOutput("a", "z"),
|
||||
});
|
||||
|
||||
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: boundedToolCallOutput("a", "z"),
|
||||
});
|
||||
|
||||
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: boundedToolCallOutput("a", "z"),
|
||||
});
|
||||
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: boundedToolCallOutput("a", "z"),
|
||||
});
|
||||
|
||||
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(`coalesces a same-tick assistant burst after the ${COALESCE_WINDOW_MS}ms window`, async () => {
|
||||
vi.useFakeTimers();
|
||||
const harness = createHarness();
|
||||
|
||||
@@ -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";
|
||||
@@ -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(
|
||||
{
|
||||
|
||||
75
packages/server/src/server/agent/agent-timeline-content.ts
Normal file
75
packages/server/src/server/agent/agent-timeline-content.ts
Normal file
@@ -0,0 +1,75 @@
|
||||
import type { AgentTimelineItem } from "./agent-sdk-types.js";
|
||||
|
||||
const TOOL_CALL_CONTENT_MAX_BYTES = 64 * 1024;
|
||||
const TOOL_CALL_CONTENT_TRUNCATION_MARKER = "\n...<tool output truncated in the middle>...\n";
|
||||
|
||||
function utf8ByteLength(text: string): number {
|
||||
return Buffer.byteLength(text, "utf8");
|
||||
}
|
||||
|
||||
function takeFirstUtf8Bytes(text: string, maxBytes: number): string {
|
||||
let low = 0;
|
||||
let high = text.length;
|
||||
while (low < high) {
|
||||
const midpoint = Math.ceil((low + high) / 2);
|
||||
if (utf8ByteLength(text.slice(0, midpoint)) <= maxBytes) {
|
||||
low = midpoint;
|
||||
} else {
|
||||
high = midpoint - 1;
|
||||
}
|
||||
}
|
||||
if (low > 0 && low < text.length && /[\uD800-\uDBFF]/.test(text[low - 1] ?? "")) {
|
||||
low -= 1;
|
||||
}
|
||||
return text.slice(0, low);
|
||||
}
|
||||
|
||||
function takeLastUtf8Bytes(text: string, maxBytes: number): string {
|
||||
let low = 0;
|
||||
let high = text.length;
|
||||
while (low < high) {
|
||||
const length = Math.ceil((low + high) / 2);
|
||||
if (utf8ByteLength(text.slice(text.length - length)) <= maxBytes) {
|
||||
low = length;
|
||||
} else {
|
||||
high = length - 1;
|
||||
}
|
||||
}
|
||||
let start = text.length - low;
|
||||
if (start > 0 && start < text.length && /[\uDC00-\uDFFF]/.test(text[start] ?? "")) {
|
||||
start += 1;
|
||||
}
|
||||
return text.slice(start);
|
||||
}
|
||||
|
||||
function limitToolCallText(text: string): string {
|
||||
if (utf8ByteLength(text) <= TOOL_CALL_CONTENT_MAX_BYTES) {
|
||||
return text;
|
||||
}
|
||||
const availableBytes =
|
||||
TOOL_CALL_CONTENT_MAX_BYTES - utf8ByteLength(TOOL_CALL_CONTENT_TRUNCATION_MARKER);
|
||||
const headBytes = Math.floor(availableBytes / 2);
|
||||
const tailBytes = availableBytes - headBytes;
|
||||
return `${takeFirstUtf8Bytes(text, headBytes)}${TOOL_CALL_CONTENT_TRUNCATION_MARKER}${takeLastUtf8Bytes(text, tailBytes)}`;
|
||||
}
|
||||
|
||||
export function limitAgentTimelineItemContent(item: AgentTimelineItem): AgentTimelineItem {
|
||||
if (
|
||||
item.type !== "tool_call" ||
|
||||
item.detail.type !== "shell" ||
|
||||
typeof item.detail.output !== "string"
|
||||
) {
|
||||
return item;
|
||||
}
|
||||
const output = limitToolCallText(item.detail.output);
|
||||
if (output === item.detail.output) {
|
||||
return item;
|
||||
}
|
||||
return {
|
||||
...item,
|
||||
detail: {
|
||||
...item.detail,
|
||||
output,
|
||||
},
|
||||
};
|
||||
}
|
||||
Reference in New Issue
Block a user