mirror of
https://github.com/getpaseo/paseo.git
synced 2026-10-23 17:52:49 +00:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
0c05f69402 | ||
|
|
8fdc41f06c | ||
|
|
928c124356 | ||
|
|
0a398833d8 |
@@ -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.
|
> 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
|
## Presence is not delivery
|
||||||
|
|
||||||
Client heartbeat reports presence:
|
Client heartbeat reports presence:
|
||||||
|
|||||||
@@ -30,6 +30,7 @@ import type {
|
|||||||
|
|
||||||
const COALESCE_WINDOW_MS = AGENT_STREAM_COALESCE_DEFAULT_WINDOW_MS;
|
const COALESCE_WINDOW_MS = AGENT_STREAM_COALESCE_DEFAULT_WINDOW_MS;
|
||||||
const BEFORE_COALESCE_WINDOW_MS = Math.max(COALESCE_WINDOW_MS - 1, 0);
|
const BEFORE_COALESCE_WINDOW_MS = Math.max(COALESCE_WINDOW_MS - 1, 0);
|
||||||
|
const TOOL_CALL_CONTENT_MAX_BYTES = 64 * 1024;
|
||||||
|
|
||||||
const TEST_CAPABILITIES: AgentCapabilityFlags = {
|
const TEST_CAPABILITIES: AgentCapabilityFlags = {
|
||||||
supportsStreaming: false,
|
supportsStreaming: false,
|
||||||
@@ -377,6 +378,148 @@ afterEach(() => {
|
|||||||
});
|
});
|
||||||
|
|
||||||
describe("target coalesced behavior", () => {
|
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 () => {
|
test(`coalesces a same-tick assistant burst after the ${COALESCE_WINDOW_MS}ms window`, async () => {
|
||||||
vi.useFakeTimers();
|
vi.useFakeTimers();
|
||||||
const harness = createHarness();
|
const harness = createHarness();
|
||||||
|
|||||||
@@ -2314,6 +2314,22 @@ test("importProviderSession imports the selected session without listing and pub
|
|||||||
item: { type: "assistant_message" as const, text: "Done" },
|
item: { type: "assistant_message" as const, text: "Done" },
|
||||||
timestamp: "2026-01-02T00:00:01.000Z",
|
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([
|
expect(manager.getTimeline(imported.id)).toEqual([
|
||||||
{ type: "user_message", text: "Trace provider imports" },
|
{ type: "user_message", text: "Trace provider imports" },
|
||||||
{ type: "assistant_message", text: "Done" },
|
{ 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).toHaveLength(1);
|
||||||
expect(events[0]).toMatchObject({
|
expect(events[0]).toMatchObject({
|
||||||
|
|||||||
@@ -58,6 +58,7 @@ import {
|
|||||||
AGENT_STREAM_COALESCE_DEFAULT_WINDOW_MS,
|
AGENT_STREAM_COALESCE_DEFAULT_WINDOW_MS,
|
||||||
AgentStreamCoalescer,
|
AgentStreamCoalescer,
|
||||||
} from "./agent-stream-coalescer.js";
|
} from "./agent-stream-coalescer.js";
|
||||||
|
import { limitAgentTimelineItemContent } from "./agent-timeline-content.js";
|
||||||
import { ForegroundRunState, type ForegroundTurnWaiter } from "./foreground-run-state.js";
|
import { ForegroundRunState, type ForegroundTurnWaiter } from "./foreground-run-state.js";
|
||||||
import { getAgentProviderDefinition } from "@getpaseo/protocol/provider-manifest";
|
import { getAgentProviderDefinition } from "@getpaseo/protocol/provider-manifest";
|
||||||
import { invokeRewindCapability, type RewindMode } from "./rewind/rewind.js";
|
import { invokeRewindCapability, type RewindMode } from "./rewind/rewind.js";
|
||||||
@@ -493,7 +494,7 @@ function buildImportedTimelineRows(entries: readonly ImportedTimelineEntry[]): A
|
|||||||
rows.push({
|
rows.push({
|
||||||
seq: rows.length + 1,
|
seq: rows.length + 1,
|
||||||
timestamp: entry.timestamp ?? new Date().toISOString(),
|
timestamp: entry.timestamp ?? new Date().toISOString(),
|
||||||
item: entry.item,
|
item: limitAgentTimelineItemContent(entry.item),
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
return rows;
|
return rows;
|
||||||
@@ -1752,6 +1753,7 @@ export class AgentManager {
|
|||||||
|
|
||||||
async appendTimelineItem(agentId: string, item: AgentTimelineItem): Promise<void> {
|
async appendTimelineItem(agentId: string, item: AgentTimelineItem): Promise<void> {
|
||||||
const agent = this.requireAgent(agentId);
|
const agent = this.requireAgent(agentId);
|
||||||
|
item = limitAgentTimelineItemContent(item);
|
||||||
this.touchUpdatedAt(agent);
|
this.touchUpdatedAt(agent);
|
||||||
const row = this.recordTimeline(agentId, item);
|
const row = this.recordTimeline(agentId, item);
|
||||||
this.dispatchStream(
|
this.dispatchStream(
|
||||||
@@ -2990,17 +2992,22 @@ export class AgentManager {
|
|||||||
agent.historyPrimed = true;
|
agent.historyPrimed = true;
|
||||||
|
|
||||||
for (const event of historyEvents) {
|
for (const event of historyEvents) {
|
||||||
|
const item = limitAgentTimelineItemContent(event.item);
|
||||||
const row = this.recordTimeline(
|
const row = this.recordTimeline(
|
||||||
agent.id,
|
agent.id,
|
||||||
event.item,
|
item,
|
||||||
event.timestamp ? { timestamp: event.timestamp } : undefined,
|
event.timestamp ? { timestamp: event.timestamp } : undefined,
|
||||||
);
|
);
|
||||||
if (options?.broadcast) {
|
if (options?.broadcast) {
|
||||||
this.dispatchStream(agent.id, event, {
|
this.dispatchStream(
|
||||||
seq: row.seq,
|
agent.id,
|
||||||
epoch: this.timelineStore.getEpoch(agent.id),
|
{ ...event, item },
|
||||||
timestamp: row.timestamp,
|
{
|
||||||
});
|
seq: row.seq,
|
||||||
|
epoch: this.timelineStore.getEpoch(agent.id),
|
||||||
|
timestamp: row.timestamp,
|
||||||
|
},
|
||||||
|
);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
this.touchUpdatedAt(agent);
|
this.touchUpdatedAt(agent);
|
||||||
@@ -3057,6 +3064,12 @@ export class AgentManager {
|
|||||||
event: AgentStreamEvent,
|
event: AgentStreamEvent,
|
||||||
options?: HandleStreamEventOptions,
|
options?: HandleStreamEventOptions,
|
||||||
): Promise<boolean> {
|
): Promise<boolean> {
|
||||||
|
if (event.type === "timeline") {
|
||||||
|
event = {
|
||||||
|
...event,
|
||||||
|
item: limitAgentTimelineItemContent(event.item),
|
||||||
|
};
|
||||||
|
}
|
||||||
const eventTurnId = getAgentStreamEventTurnId(event);
|
const eventTurnId = getAgentStreamEventTurnId(event);
|
||||||
const isForegroundEvent = Boolean(eventTurnId && agent.activeForegroundTurnId === eventTurnId);
|
const isForegroundEvent = Boolean(eventTurnId && agent.activeForegroundTurnId === eventTurnId);
|
||||||
this.traceHandleStreamEventStart(agent, event, eventTurnId, isForegroundEvent);
|
this.traceHandleStreamEventStart(agent, event, eventTurnId, isForegroundEvent);
|
||||||
@@ -3553,6 +3566,7 @@ export class AgentManager {
|
|||||||
item: AgentTimelineItem,
|
item: AgentTimelineItem,
|
||||||
options?: { timestamp?: string },
|
options?: { timestamp?: string },
|
||||||
): AgentTimelineRow {
|
): AgentTimelineRow {
|
||||||
|
item = limitAgentTimelineItemContent(item);
|
||||||
const row = this.timelineStore.append(agentId, item, options);
|
const row = this.timelineStore.append(agentId, item, options);
|
||||||
this.enqueueDurableTimelineAppend(agentId, row);
|
this.enqueueDurableTimelineAppend(agentId, row);
|
||||||
return row;
|
return row;
|
||||||
@@ -3742,6 +3756,12 @@ export class AgentManager {
|
|||||||
event: AgentStreamEvent,
|
event: AgentStreamEvent,
|
||||||
metadata?: { seq?: number; epoch?: string; timestamp?: string },
|
metadata?: { seq?: number; epoch?: string; timestamp?: string },
|
||||||
): void {
|
): void {
|
||||||
|
if (event.type === "timeline") {
|
||||||
|
event = {
|
||||||
|
...event,
|
||||||
|
item: limitAgentTimelineItemContent(event.item),
|
||||||
|
};
|
||||||
|
}
|
||||||
const agent = this.agents.get(agentId);
|
const agent = this.agents.get(agentId);
|
||||||
this.logger.trace(
|
this.logger.trace(
|
||||||
{
|
{
|
||||||
|
|||||||
@@ -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,
|
||||||
|
},
|
||||||
|
};
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user