diff --git a/docs/timeline-sync.md b/docs/timeline-sync.md index 812718be58..f205d78924 100644 --- a/docs/timeline-sync.md +++ b/docs/timeline-sync.md @@ -10,6 +10,11 @@ then replaces the previous tool state or merges into the current text item. Inte are never retained for history or catch-up. Provider history is the durable transcript authority and rebuilds the projection when an agent resumes. +Pagination bounds each read, not total resident history. Active and idle agents retain their +projected transcript; archive releases it and later history reads rebuild it from the provider. +Streaming overhead must remain proportional to the retained transcript. Cumulative text snapshots +must be released after their deltas have been consumed. + The invariants are: > A continuously subscribed client applies every committed row in order. Opening or resuming an diff --git a/packages/server/src/server/agent/plugin-provider.test.ts b/packages/server/src/server/agent/plugin-provider.test.ts index 6a61573b6b..101a19a39f 100644 --- a/packages/server/src/server/agent/plugin-provider.test.ts +++ b/packages/server/src/server/agent/plugin-provider.test.ts @@ -4,6 +4,8 @@ import { join } from "node:path"; import { AgentStorage } from "./agent-storage.js"; import { createIdleRuntimeCleanup } from "./idle-runtime-cleanup/index.js"; import { relative } from "node:path"; +import { execFileSync } from "node:child_process"; +import { fileURLToPath } from "node:url"; import type { ProviderConnection, ProviderLaunch, @@ -25,6 +27,21 @@ import { StaleProviderSessionError, } from "./stale-provider-session-error.js"; +test("long concurrent plugin and child streams retain text-sized history and release it on archive", () => { + const fixture = fileURLToPath(new URL("./test-utils/plugin-stream-memory.ts", import.meta.url)); + const output = execFileSync( + process.execPath, + ["--expose-gc", "--max-old-space-size=512", "--import", "tsx", fixture], + { + encoding: "utf8", + timeout: 20_000, + }, + ); + const memory = JSON.parse(output) as { retained: number; closed: number }; + expect(memory.retained).toBeLessThan(24 * 1024 * 1024); + expect(memory.closed).toBeLessThan(6 * 1024 * 1024); +}); + const CAPABILITIES = [ "prompt.message", "session.configure", @@ -292,6 +309,99 @@ function expectNestedChildren(events: AgentStreamEvent[]) { } describe("PluginAgentClientRegistry", () => { + test("history keeps the tool's first position when completion follows later text", async () => { + const harness = createProviderHarness(); + const registry = new PluginAgentClientRegistry(createTestLogger()); + registry.replace([harness.registration]); + const client = registry.clients()[harness.registration.id]!; + const session = await client.createSession({ provider: client.provider, cwd: "/tmp" }); + const opened = harness.inputs.find((input) => input.type === "session.open"); + if (opened?.type !== "session.open") throw new Error("missing session open"); + const tool = { + type: "tool_call" as const, + id: "tool-position", + callId: "call-position", + name: "read", + detail: { type: "unknown" as const, input: null, output: null }, + error: null, + }; + try { + harness.emit({ + type: "timeline.item", + sessionId: opened.sessionId, + item: { ...tool, status: "running" }, + }); + harness.emit({ + type: "timeline.item", + sessionId: opened.sessionId, + item: { type: "assistant_message", id: "later-text", text: "Working" }, + }); + harness.emit({ + type: "timeline.item", + sessionId: opened.sessionId, + item: { ...tool, status: "completed" }, + }); + const items = []; + for await (const event of session.streamHistory()) + if (event.type === "timeline") items.push(event.item); + expect(items).toEqual([ + { + type: "tool_call", + callId: "call-position", + name: "read", + detail: tool.detail, + error: null, + status: "completed", + }, + { type: "assistant_message", messageId: "later-text", text: "Working" }, + ]); + } finally { + await session.close(); + await registry.shutdown(); + } + }); + + test("detached text deltas preserve split surrogates and history replays one complete message", async () => { + const harness = createProviderHarness(); + const registry = new PluginAgentClientRegistry(createTestLogger()); + registry.replace([harness.registration]); + const client = registry.clients()[harness.registration.id]!; + const session = await client.createSession({ provider: client.provider, cwd: "/tmp" }); + const opened = harness.inputs.find((input) => input.type === "session.open"); + if (opened?.type !== "session.open") throw new Error("missing session open"); + const events: AgentStreamEvent[] = []; + session.subscribe((event) => events.push(event)); + const prefix = "long prefix ".repeat(10); + const first = prefix + "a\uD83D"; + const complete = first + "\uDE00b"; + try { + for (const text of [first, complete, complete]) { + harness.emit({ + type: "timeline.item", + sessionId: opened.sessionId, + item: { type: "assistant_message", id: "split-message", text }, + }); + } + const deltas = events.flatMap((event) => + event.type === "timeline" && event.item.type === "assistant_message" + ? [event.item.text] + : [], + ); + expect(deltas).toEqual([first, "\uDE00b"]); + expect(deltas.join("")).toBe(prefix + "a😀b"); + const history: AgentStreamEvent[] = []; + for await (const event of session.streamHistory()) history.push(event); + expect( + history.flatMap((event) => + event.type === "timeline" && event.item.type === "assistant_message" ? [event.item] : [], + ), + ).toEqual([{ type: "assistant_message", messageId: "split-message", text: complete }]); + } finally { + await session.close(); + await registry.shutdown(); + } + }); + test("stores only agent options while the plugin receives merged defaults", async () => { const logger = createTestLogger(); const harness = createProviderHarness(); diff --git a/packages/server/src/server/agent/plugin-provider.ts b/packages/server/src/server/agent/plugin-provider.ts index 3970715a55..f09f6349b0 100644 --- a/packages/server/src/server/agent/plugin-provider.ts +++ b/packages/server/src/server/agent/plugin-provider.ts @@ -1,4 +1,5 @@ import { randomUUID } from "node:crypto"; +import { PluginSessionHistory } from "./plugin-session-history.js"; import { resolve as resolvePath } from "node:path"; import type { Logger } from "pino"; import type { JsonValue } from "@getpaseo/protocol/agent-types"; @@ -592,6 +593,7 @@ class ProviderRuntime { class ProviderRuntimeSession { readonly history: ProviderEvent[] = []; + private historyClaimed = false; private readonly listeners = new Set<(event: ProviderEvent) => void>(); private capabilities: readonly string[] = []; private openRequest: { requestId: string; deferred: Deferred } | null = null; @@ -622,6 +624,10 @@ class ProviderRuntimeSession { } onEvent(listener: (event: ProviderEvent) => void): () => void { + // This buffer bridges session.open and adapter attachment. Once replayed, + // the adapter owns history; retaining live cumulative snapshots is quadratic. + this.historyClaimed = true; + this.history.length = 0; this.listeners.add(listener); return () => this.listeners.delete(listener); } @@ -713,6 +719,7 @@ class ProviderRuntimeSession { } async close(): Promise { + this.history.length = 0; if (this.restoration === "parent") { this.runtime.removeSession(this.id, this.providerSessionId); return; @@ -826,7 +833,7 @@ class ProviderRuntimeSession { if (event.type === "session.persistence" && this.restoration === "core") { this.persistence = event.persistence; } - this.history.push(event); + if (!this.historyClaimed) this.history.push(event); for (const listener of this.listeners) listener(event); } } @@ -1128,17 +1135,30 @@ class PluginAgentClient implements AgentClient { } } +interface TimelineTextSnapshot { + type: "assistant_message" | "reasoning" | "user_message"; + text: string; +} + +function isTextSnapshotItem( + item: ProviderTimelineItem, +): item is Extract { + return ( + item.type === "assistant_message" || item.type === "reasoning" || item.type === "user_message" + ); +} + class PluginAgentSession implements AgentSession { readonly initialTimeline: ImportedProviderSession["timeline"]; private readonly listeners = new Set<(event: AgentStreamEvent) => void>(); - private readonly history: AgentStreamEvent[] = []; + private readonly history = new PluginSessionHistory(); private readonly pendingPermissions = new Map(); private readonly permissionResponses = new Map(); private readonly revertTokens = new Map(); - private readonly timelineSnapshots = new Map(); + private readonly timelineSnapshots = new Map(); private readonly subagentIdsBySession = new Map(); private readonly childUnsubscribes = new Map void>(); - private readonly childSnapshots = new Map>(); + private readonly childSnapshots = new Map>(); private unsubscribe: (() => void) | null = null; private currentTurnId: string | null = null; private closed = false; @@ -1236,7 +1256,7 @@ class PluginAgentSession implements AgentSession { } async *streamHistory(): AsyncGenerator { - for (const event of this.history) yield event; + for (const event of this.history.read(this.provider)) yield event; } get canStopRuntime(): (() => Promise) | undefined { @@ -1304,6 +1324,12 @@ class PluginAgentSession implements AgentSession { for (const unsubscribe of this.childUnsubscribes.values()) unsubscribe(); this.childUnsubscribes.clear(); this.subagentIdsBySession.clear(); + this.history.clear(); + this.timelineSnapshots.clear(); + this.childSnapshots.clear(); + this.revertTokens.clear(); + this.pendingPermissions.clear(); + this.permissionResponses.clear(); this.listeners.clear(); this.onClose(); await this.bridge.close(); @@ -1344,16 +1370,20 @@ class PluginAgentSession implements AgentSession { } timelineHistory(): ImportedProviderSession["timeline"] { - return this.history.flatMap((event) => - event.type === "timeline" ? [{ item: event.item, timestamp: event.timestamp }] : [], - ); + return this.history + .read(this.provider) + .flatMap((event) => + event.type === "timeline" ? [{ item: event.item, timestamp: event.timestamp }] : [], + ); } subagentHistory(): NonNullable { - return this.history.filter( - (event): event is Extract => - event.type === "provider_subagent", - ); + return this.history + .read(this.provider) + .filter( + (event): event is Extract => + event.type === "provider_subagent", + ); } attachChild( @@ -1382,7 +1412,7 @@ class PluginAgentSession implements AgentSession { cwd: opened.cwd, }, }); - const snapshots = new Map(); + const snapshots = new Map(); this.childSnapshots.set(childId, snapshots); for (const event of child.history) this.acceptChildEvent(childId, event, snapshots); this.childUnsubscribes.set( @@ -1568,7 +1598,7 @@ class PluginAgentSession implements AgentSession { private acceptChildEvent( childId: string, event: ProviderEvent, - snapshots: Map, + snapshots: Map, ): void { if (event.type === "timeline.item") { const item = mapTimelineItem(event.item, snapshots); @@ -1721,24 +1751,24 @@ function mapPromptContent(prompt: AgentPromptInput): ProviderContent[] { function mapTimelineItem( item: ProviderTimelineItem, - snapshots: Map, + snapshots: Map, revertTokens?: Map, ): AgentTimelineItem | null { const previous = snapshots.get(item.id); - snapshots.set(item.id, item); - const { id, revertToken, ...withoutIdentity } = item; - if (revertToken !== undefined && revertTokens) { - revertTokens.set(id, revertToken); - if (item.type === "user_message" || item.type === "assistant_message") { - revertTokens.set(item.messageId ?? id, revertToken); - } + if (isTextSnapshotItem(item)) { + snapshots.set(item.id, { type: item.type, text: item.text }); } + const { id, revertToken: _revertToken, ...withoutIdentity } = item; + rememberRevertToken(item, revertTokens); if (item.type === "assistant_message" || item.type === "reasoning") { const previousText = previous?.type === item.type ? previous.text : ""; - const text = item.text.startsWith(previousText) + const suffix = item.text.startsWith(previousText) ? item.text.slice(previousText.length) : item.text; + // Match OpenCode's detachedSuffix: a sliced delta otherwise keeps its entire + // cumulative IPC snapshot alive in the projected timeline. Preserve split surrogates. + const text = Buffer.from(suffix, "utf16le").toString("utf16le"); if (text.length === 0) return null; return item.type === "assistant_message" ? { type: "assistant_message", text, messageId: item.messageId ?? id } @@ -1766,6 +1796,17 @@ function mapTimelineItem( return withoutIdentity as AgentTimelineItem; } +function rememberRevertToken( + item: ProviderTimelineItem, + tokens?: Map, +): void { + if (item.revertToken === undefined || !tokens) return; + tokens.set(item.id, item.revertToken); + if (item.type === "user_message" || item.type === "assistant_message") { + tokens.set(item.messageId ?? item.id, item.revertToken); + } +} + function persistenceHandle( provider: string, persistence: ProviderPersistence, diff --git a/packages/server/src/server/agent/plugin-session-history.ts b/packages/server/src/server/agent/plugin-session-history.ts new file mode 100644 index 0000000000..f5efd446d7 --- /dev/null +++ b/packages/server/src/server/agent/plugin-session-history.ts @@ -0,0 +1,66 @@ +import { TimelineProjection } from "./timeline-projection.js"; +import type { AgentStreamEvent } from "./agent-sdk-types.js"; + +// Replay canonical items, rather than retaining every streamed delta or tool state. +// Sequence numbers here only preserve adapter replay order, not wire cursors. +export class PluginSessionHistory { + private readonly timelines = new Map< + string, + { projection: TimelineProjection; seq: number; positions: Map } + >(); + private readonly events: Array<{ seq: number; event: AgentStreamEvent }> = []; + private seq = 0; + + push(event: AgentStreamEvent): void { + const seq = ++this.seq; + const child = + event.type === "provider_subagent" && event.event.type === "timeline" ? event.event : null; + if (event.type !== "timeline" && !child) { + this.events.push({ seq, event }); + return; + } + const key = child?.id ?? ""; + let projection = this.timelines.get(key); + if (!projection) { + projection = { projection: new TimelineProjection(), seq: 0, positions: new Map() }; + this.timelines.set(key, projection); + } + const timeline = event.type === "timeline" ? event : child!; + const row = projection.projection.append({ + seq: ++projection.seq, + item: timeline.item, + timestamp: timeline.timestamp ?? "", + ...(event.type === "timeline" && event.turnId ? { turnId: event.turnId } : {}), + }); + if (!projection.positions.has(row.seqStart)) projection.positions.set(row.seqStart, seq); + } + + read(provider: string): AgentStreamEvent[] { + const events = [...this.events]; + for (const [id, projection] of this.timelines) { + for (const row of projection.projection.getRows()) { + const timestamp = row.timestamp ? { timestamp: row.timestamp } : {}; + const event: AgentStreamEvent = id + ? { + type: "provider_subagent", + provider, + event: { type: "timeline", id, item: row.item, ...timestamp }, + } + : { + type: "timeline", + provider, + item: row.item, + ...timestamp, + ...(row.turnId ? { turnId: row.turnId } : {}), + }; + events.push({ seq: projection.positions.get(row.seqStart)!, event }); + } + } + return events.sort((a, b) => a.seq - b.seq).map(({ event }) => event); + } + + clear(): void { + this.timelines.clear(); + this.events.length = 0; + } +} diff --git a/packages/server/src/server/agent/test-utils/plugin-stream-memory.ts b/packages/server/src/server/agent/test-utils/plugin-stream-memory.ts new file mode 100644 index 0000000000..948202bae2 --- /dev/null +++ b/packages/server/src/server/agent/test-utils/plugin-stream-memory.ts @@ -0,0 +1,176 @@ +import assert from "node:assert/strict"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { setTimeout as delay } from "node:timers/promises"; +import type { ProviderEvent, ProviderRegistration } from "@getpaseo/plugin/server/provider"; +import pino from "pino"; +import { PluginAgentClientRegistry } from "../plugin-provider.js"; +import { InMemoryAgentTimelineStore } from "../agent-timeline-store.js"; +import { AgentManager } from "../agent-manager.js"; +import { AgentStorage } from "../agent-storage.js"; +import { buildProviderRegistry } from "../provider-registry.js"; + +if (!global.gc) throw new Error("Run with --expose-gc"); +const logger = pino({ level: "silent" }); +const plugins = new PluginAgentClientRegistry(logger); +const root = await mkdtemp(join(tmpdir(), "plugin-stream-memory-")); +const emitters: Array<(event: ProviderEvent) => void> = []; +const ids: string[] = []; +const registrations: ProviderRegistration[] = Array.from({ length: 3 }, (_, index) => ({ + id: `memory-${index}`, + label: "Memory fixture", + async connect() { + let listener: (event: ProviderEvent) => void = () => {}; + const capabilities = ["prompt.message", "session.persistence"] as const; + return { + version: 1, + capabilities, + onEvent(callback) { + listener = callback; + // Plugin IPC delivers independent JSON strings, rather than shared ropes. + emitters[index] = (event) => listener(JSON.parse(JSON.stringify(event))); + return () => { + listener = () => {}; + }; + }, + async send(input) { + if (input.type === "session.open") { + ids[index] = input.sessionId; + listener({ + type: "session.opened", + requestId: input.requestId, + sessionId: input.sessionId, + cwd: "/tmp", + capabilities, + restoration: "core", + persistence: { version: 1, data: { index } }, + }); + listener({ + type: "session.ready", + requestId: input.requestId, + sessionId: input.sessionId, + }); + } else if (input.type === "catalog") { + listener({ + type: "catalog", + requestId: input.requestId, + catalog: { models: [], modes: [], thinkingOptions: [] }, + }); + } else if (input.type === "session.close") { + listener({ type: "session.closed", sessionId: input.sessionId }); + listener({ type: "request.completed", requestId: input.requestId }); + } + }, + async close() {}, + }; + }, +})); +plugins.replace(registrations); +const store = new InMemoryAgentTimelineStore(); +const manager = new AgentManager({ + clients: plugins.clients(), + providerDefinitions: buildProviderRegistry(logger, { pluginProviders: plugins.definitions() }), + registry: new AgentStorage(root, logger), + timelineStore: store, + logger, +}); +const agents = await Promise.all( + Object.keys(plugins.clients()).map((provider) => + manager.createAgent({ provider, cwd: root }, undefined, {}), + ), +); +const sessions = agents.map((agent) => agent.session!); +agents.forEach((_agent, index) => { + emitters[index]!({ + type: "session.opened", + sessionId: `child-${index}`, + parentSessionId: ids[index]!, + capabilities: [], + restoration: "parent", + cwd: root, + }); +}); +global.gc(); +const before = process.memoryUsage().heapUsed; +const steps = 4; +const chunks = 200; +const expected = new Map(); +for (let step = 0; step < steps; step++) { + for (let index = 0; index < sessions.length; index++) { + const text = `${index}:${step}:` + "推理".repeat(15000) + "a\uD83D\uDE00b"; + expected.set(`${index}:${step}`, text); + for (let chunk = 1; chunk <= chunks; chunk++) { + for (const sessionId of [ids[index]!, `child-${index}`]) { + emitters[index]!({ + type: "timeline.item", + sessionId, + item: { + type: "reasoning", + id: `reason-${step}`, + text: text.slice(0, Math.ceil((chunk * text.length) / chunks)), + }, + }); + } + } + emitters[index]!({ + type: "timeline.item", + sessionId: ids[index]!, + item: { + type: "tool_call", + id: `tool-${step}`, + callId: `tool-${step}`, + name: "read", + status: "completed", + error: null, + detail: { type: "unknown", input: null, output: "输出".repeat(16000) }, + }, + }); + emitters[index]!({ + type: "timeline.item", + sessionId: `child-${index}`, + item: { + type: "assistant_message", + id: `answer-${step}`, + text: "Child step complete.", + }, + }); + } + // Drain the manager's event queue; terminal tool states flush its real coalescer. + await delay(80); +} +global.gc(); +const retained = process.memoryUsage().heapUsed - before; +assert(retained < 24 * 1024 * 1024, `Retained heap exceeded 24 MiB: ${retained} bytes`); +for (let index = 0; index < sessions.length; index++) { + const items = store.getItems(agents[index]!.id); + assert.equal(items.length, steps * 2); + for (let step = 0; step < steps; step++) { + const item = items[step * 2]!; + assert.equal(item.type === "reasoning" && item.text, expected.get(`${index}:${step}`)); + } + const children = manager.listProviderSubagents(agents[index]!.id); + assert.equal(children.length, 1); + let rootItems = 0; + let childItems = 0; + for await (const event of sessions[index]!.streamHistory()) { + if (event.type === "timeline") rootItems++; + if (event.type === "provider_subagent" && event.event.type === "timeline") childItems++; + } + assert.equal(rootItems, steps * 2); + assert.equal(childItems, steps * 2); + await manager.archiveAgent(agents[index]!.id); + assert.equal(store.has(agents[index]!.id), false); + assert.equal( + manager.listAgents().some((agent) => agent.id === agents[index]!.id), + false, + ); +} +await plugins.shutdown(); +await manager.flushForShutdown(); +await rm(root, { recursive: true, force: true }); +global.gc(); +const closed = process.memoryUsage().heapUsed - before; +// Keep closed session references alive, as callers can retain them after archive. +assert.equal(sessions.length, 3); +process.stdout.write(JSON.stringify({ retained, closed })); diff --git a/packages/server/src/server/agent/timeline-projection.ts b/packages/server/src/server/agent/timeline-projection.ts index 8b05432545..bd5357e581 100644 --- a/packages/server/src/server/agent/timeline-projection.ts +++ b/packages/server/src/server/agent/timeline-projection.ts @@ -323,7 +323,7 @@ export class TimelineProjection { this.sourceRanges.sort((a, b) => a.startSeq - b.startSeq); } - append(row: AgentTimelineRow): void { + append(row: AgentTimelineRow): ProjectedTimelineRow { const rowIndex = this.appendProjected(row); const last = this.sourceRanges.at(-1); if (last && last.rowIndex === rowIndex && last.endSeq + 1 === row.seq) { @@ -331,6 +331,7 @@ export class TimelineProjection { } else { this.sourceRanges.push({ startSeq: row.seq, endSeq: row.seq, rowIndex }); } + return { ...this.rows[rowIndex] }; } private appendProjected(row: AgentTimelineRow): number {