Compare commits

...
Author SHA1 Message Date
Mohamed Boudra 5771efaa03 docs(plugins): keep runtime-entry migration focused on 0.8 2026-10-20 08:49:51 +02:00
Mohamed Boudra f44300b523 perf(plugins): keep turn-ended payloads independent of history
Stop materializing and copying the entire conversation through lifecycle hooks. Keep the deprecated timeline field empty and direct plugins to paged timeline reads.
2026-10-20 08:48:07 +02:00
7 changed files with 140 additions and 25 deletions
+4
View File
@@ -50,6 +50,10 @@ export interface PluginLifecycleEvents {
agent: PluginHookAgent;
turnId: string | null;
outcome: PluginTurnOutcome;
/**
* @deprecated Always empty. Fetch history on demand with
* context.paseo.agents.ref(agent.id).timeline.refetch({ direction: "tail", limit: 20 }).
*/
timeline: readonly AgentTimelineItem[];
};
"agent.permission_requested": { agent: PluginHookAgent; request: AgentPermissionRequest };
@@ -3,7 +3,7 @@ import {
formatAgentMessage,
formatSystemNotificationPrompt,
} from "./agent-messages/index.js";
import { expect, test, vi } from "vitest";
import { expect, test, vi, onTestFinished } from "vitest";
import { spawn } from "node:child_process";
import { mkdtempSync, rmSync, writeFileSync } from "node:fs";
import { join } from "node:path";
@@ -11892,3 +11892,102 @@ test.each([false, true])(
}
},
);
test.each([
{
event: {
type: "turn_completed",
provider: "codex",
turnId: "hook-turn",
} satisfies AgentStreamEvent,
outcome: { kind: "completed" },
},
{
event: {
type: "turn_failed",
provider: "codex",
turnId: "hook-turn",
error: "Failed",
code: "test",
} satisfies AgentStreamEvent,
outcome: { kind: "failed", error: { message: "Failed", code: "test" } },
},
{
event: {
type: "turn_canceled",
provider: "codex",
turnId: "hook-turn",
reason: "Stopped",
} satisfies AgentStreamEvent,
outcome: { kind: "canceled", reason: "Stopped" },
},
])(
"lifecycle hooks receive no conversation for $event.type and stream dispatch never reads timeline items",
async ({ event: terminalEvent, outcome }) => {
class CountingTimelineStore extends InMemoryAgentTimelineStore {
itemReads = 0;
override getItems(agentId: string): AgentTimelineItem[] {
this.itemReads += 1;
return super.getItems(agentId);
}
}
const workdir = mkdtempSync(join(tmpdir(), "agent-hook-payload-"));
const client = new SessionRecordingAgentClient();
const timelineStore = new CountingTimelineStore();
const events: Array<{ name: string; event: unknown }> = [];
const manager = new AgentManager({
clients: { codex: client },
registry: new AgentStorage(join(workdir, "agents"), logger),
logger,
timelineStore,
pluginLifecycle: {
before: async (_name, request) => request,
emit: (name, event) => {
events.push({ name, event });
},
},
});
onTestFinished(async () => {
await manager.closeAgent(manager.listAgents()[0]!.id);
await manager.flush();
rmSync(workdir, { recursive: true, force: true });
});
const agent = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, {
workspaceId: undefined,
});
await manager.appendTimelineItem(agent.id, {
type: "assistant_message",
text: "A long conversation".repeat(10_000),
});
expect(manager.getTimeline(agent.id)).toHaveLength(1);
const readsBeforeDispatch = timelineStore.itemReads;
const session = client.sessions[0]!;
session.pushEvent({ type: "turn_started", provider: "codex", turnId: "hook-turn" });
await expect
.poll(() => events.filter(({ name }) => name === "agent.turn_started").length)
.toBe(1);
expect(timelineStore.itemReads).toBe(readsBeforeDispatch);
session.pushEvent(terminalEvent);
await expect
.poll(() => events.filter(({ name }) => name === "agent.turn_ended"))
.toEqual([
{
name: "agent.turn_ended",
event: {
agent: {
id: agent.id,
workspaceId: null,
parentAgentId: null,
provider: "codex",
cwd: workdir,
title: null,
},
turnId: "hook-turn",
outcome,
timeline: [],
},
},
]);
expect(timelineStore.itemReads).toBe(readsBeforeDispatch);
},
);
@@ -326,6 +326,7 @@ export interface AgentManagerOptions {
registry?: AgentStorage;
onAgentAttention?: AgentAttentionCallback;
onWorkspaceStateMayHaveChanged?: (params: { cwd: string }) => void;
timelineStore?: InMemoryAgentTimelineStore;
durableTimelineStore?: AgentTimelineStore;
terminalManager?: TerminalManager | null;
mcpBaseUrl?: string;
@@ -741,7 +742,7 @@ export class AgentManager {
private readonly providerEnabled = new Map<AgentProvider, boolean>();
private readonly providerDefinitions = new Map<AgentProvider, ProviderEnabledFlag>();
private readonly agents = new Map<string, LiveManagedAgent>();
private readonly timelineStore = new InMemoryAgentTimelineStore();
private readonly timelineStore: InMemoryAgentTimelineStore;
private readonly providerSubagents = new ProviderSubagentStore();
private readonly agentsAwaitingInitialSnapshotPersist = new Set<string>();
private readonly sessionEventTails = new Map<string, Promise<void>>();
@@ -778,7 +779,8 @@ export class AgentManager {
constructor(options: AgentManagerOptions) {
this.pluginLifecycle = options.pluginLifecycle;
this.idFactory = options?.idFactory ?? (() => randomUUID());
this.timelineStore = options.timelineStore ?? new InMemoryAgentTimelineStore();
this.idFactory = options.idFactory ?? (() => randomUUID());
this.registry = options?.registry;
this.durableTimelineStore = options?.durableTimelineStore;
this.onAgentAttention = options?.onAgentAttention;
@@ -5164,7 +5166,6 @@ export class AgentManager {
this.pluginLifecycle,
describeHookAgent({ ...agent, title: agent.config.title }),
event,
this.timelineStore.getItems(agentId),
);
}
}
@@ -146,10 +146,12 @@ export default function contribute(server) {
{ hook: "agent.permission_resolved", event: { resolution: { behavior: "deny" } } },
{
hook: "agent.turn_ended",
event: { outcome: { kind: "completed" }, timeline: expect.any(Array) },
event: { outcome: { kind: "completed" }, timeline: [] },
},
{ hook: "agent.turn_started", event: { agent: { id: agent.id } } },
]);
const page = await client.fetchAgentTimeline(agent.id, { direction: "tail", limit: 20 });
expect(page.entries.length).toBeGreaterThan(0);
await client.archiveAgent(agent.id);
await expect
.poll(async () => {
@@ -1,4 +1,4 @@
import type { AgentStreamEvent, AgentTimelineItem } from "../../agent/agent-sdk-types.js";
import type { AgentStreamEvent } from "../../agent/agent-sdk-types.js";
import { z } from "zod";
import { CreateAgentRequestMessageSchema } from "@getpaseo/protocol/messages";
import type {
@@ -94,7 +94,6 @@ export function publishAgentStream(
lifecycle: PluginLifecycle,
agent: PluginHookAgent,
event: AgentStreamEvent,
timeline: readonly AgentTimelineItem[],
): void {
if (event.type === "turn_started") {
lifecycle.emit("agent.turn_started", { agent, turnId: event.turnId ?? null });
@@ -102,21 +101,21 @@ export function publishAgentStream(
lifecycle.emit("agent.turn_ended", {
agent,
turnId: event.turnId ?? null,
timeline,
timeline: [],
outcome: { kind: "completed" },
});
} else if (event.type === "turn_failed") {
lifecycle.emit("agent.turn_ended", {
agent,
turnId: event.turnId ?? null,
timeline,
timeline: [],
outcome: { kind: "failed", error: { message: event.error, code: event.code } },
});
} else if (event.type === "turn_canceled") {
lifecycle.emit("agent.turn_ended", {
agent,
turnId: event.turnId ?? null,
timeline,
timeline: [],
outcome: { kind: "canceled", reason: event.reason },
});
} else if (event.type === "permission_requested") {
@@ -21,9 +21,10 @@ export default function contribute(server: PluginServerContext) {
if (event.outcome.kind === "canceled") {
return;
}
const text = latestOutputText(event.timeline);
const agent = context.paseo.agents.ref(event.agent.id);
const page = await agent.timeline.refetch({ direction: "tail", limit: 20 });
const text = latestOutputText(page.entries.map(({ item }) => item));
if (/out of credits/i.test(text)) {
const agent = context.paseo.agents.ref(event.agent.id);
console.log("Sending a follow-up after out of credits", event.agent.id);
await agent.send("Try again.");
}
+22 -13
View File
@@ -654,9 +654,11 @@ export default function contribute(server: PluginServerContext) {
return;
}
const text = latestOutputText(event.timeline);
const agent = context.paseo.agents.ref(event.agent.id);
const page = await agent.timeline.refetch({ direction: "tail", limit: 20 });
const text = latestOutputText(page.entries.map(({ item }) => item));
if (/out of credits/i.test(text)) {
await context.paseo.agents.ref(event.agent.id).send("Try again.");
await agent.send("Try again.");
}
});
@@ -670,6 +672,9 @@ Turn ends: "out of credits"
→ a new turn starts
```
The handler fetches only the last 20 timeline items. Output before that page is not inspected.
Use `cursor` and `direction` to page further when your plugin needs more history.
This sends a new message with the existing SDK. Persistent matches keep sending follow-ups;
add limits or delays in your plugin when needed. Attachments and tool effects are not replayed.
@@ -721,22 +726,26 @@ plans, and mode changes; requesting permission does not end the turn.
### Events
| Name | Event fields | Trigger |
| ---------------------------- | ---------------------------------------- | -------------------------------------------------- |
| `agent.created` | `agent` | Ordinary creation finishes; excludes import/resume |
| `agent.closed` | `agent` | A live agent runtime closes |
| `agent.turn_started` | `agent`, `turnId` | Live turn starts |
| `agent.turn_ended` | `agent`, `turnId`, `outcome`, `timeline` | Live turn completes, fails, or is canceled |
| `agent.permission_requested` | `agent`, `request` | Permission or question becomes pending |
| `agent.permission_resolved` | `agent`, `requestId`, `resolution` | Pending request is answered or cleared |
| `agent.archived` | `agent`, `archivedAt` | Archive state is saved |
| `workspace.created` | `workspace` | Record created; directory available |
| `workspace.archived` | `workspace` | Archive state is saved |
| Name | Event fields | Trigger |
| ---------------------------- | ------------------------------------------------------- | -------------------------------------------------- |
| `agent.created` | `agent` | Ordinary creation finishes; excludes import/resume |
| `agent.closed` | `agent` | A live agent runtime closes |
| `agent.turn_started` | `agent`, `turnId` | Live turn starts |
| `agent.turn_ended` | `agent`, `turnId`, `outcome`, deprecated `timeline: []` | Live turn completes, fails, or is canceled |
| `agent.permission_requested` | `agent`, `request` | Permission or question becomes pending |
| `agent.permission_resolved` | `agent`, `requestId`, `resolution` | Pending request is answered or cleared |
| `agent.archived` | `agent`, `archivedAt` | Archive state is saved |
| `workspace.created` | `workspace` | Record created; directory available |
| `workspace.archived` | `workspace` | Archive state is saved |
Agent events exclude the daemon's private disposable helpers. Public background workspaces and
their ordinary agents run the same lifecycle hooks as other workspaces. Archive events can precede
runtime/worktree cleanup; observe lifecycle completion separately when cleanup matters.
`agent.turn_ended.timeline` is deprecated and always empty. Fetch history on demand with
`context.paseo.agents.ref(event.agent.id).timeline.refetch({ direction: "tail", limit: 20 })`.
The hook payload size does not grow with conversation history.
**Shared payload shapes** (`@getpaseo/plugin/server`):
```ts