fix(server): release retained plugin streaming snapshots (#7675)

Release cumulative plugin snapshots after adapter attachment and retain canonical replay items. Preserve Unicode text, root/child ordering and current tool state; release history on close.

Co-authored-by: suntianc <suntc8985@gmail.com>

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();

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<void> } | 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<void> {
+    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<ProviderTimelineItem, { type: TimelineTextSnapshot["type"] }> {
+  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<string, AgentPermissionRequest>();
   private readonly permissionResponses = new Map<string, AgentPermissionResponse>();
   private readonly revertTokens = new Map<string, ProviderTimelineItem["revertToken"]>();
-  private readonly timelineSnapshots = new Map<string, ProviderTimelineItem>();
+  private readonly timelineSnapshots = new Map<string, TimelineTextSnapshot>();
   private readonly subagentIdsBySession = new Map<string, string | null>();
   private readonly childUnsubscribes = new Map<string, () => void>();
-  private readonly childSnapshots = new Map<string, Map<string, ProviderTimelineItem>>();
+  private readonly childSnapshots = new Map<string, Map<string, TimelineTextSnapshot>>();
   private unsubscribe: (() => void) | null = null;
   private currentTurnId: string | null = null;
   private closed = false;
@@ -1236,7 +1256,7 @@ class PluginAgentSession implements AgentSession {
   }
 
   async *streamHistory(): AsyncGenerator<AgentStreamEvent> {
-    for (const event of this.history) yield event;
+    for (const event of this.history.read(this.provider)) yield event;
   }
 
   get canStopRuntime(): (() => Promise<boolean>) | 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<ImportedProviderSession["providerSubagentEvents"]> {
-    return this.history.filter(
-      (event): event is Extract<AgentStreamEvent, { type: "provider_subagent" }> =>
-        event.type === "provider_subagent",
-    );
+    return this.history
+      .read(this.provider)
+      .filter(
+        (event): event is Extract<AgentStreamEvent, { type: "provider_subagent" }> =>
+          event.type === "provider_subagent",
+      );
   }
 
   attachChild(
@@ -1382,7 +1412,7 @@ class PluginAgentSession implements AgentSession {
         cwd: opened.cwd,
       },
     });
-    const snapshots = new Map<string, ProviderTimelineItem>();
+    const snapshots = new Map<string, TimelineTextSnapshot>();
     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<string, ProviderTimelineItem>,
+    snapshots: Map<string, TimelineTextSnapshot>,
   ): 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<string, ProviderTimelineItem>,
+  snapshots: Map<string, TimelineTextSnapshot>,
   revertTokens?: Map<string, ProviderTimelineItem["revertToken"]>,
 ): 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<string, ProviderTimelineItem["revertToken"]>,
+): 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,

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<number, number> }
+  >();
+  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;
+  }
+}

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<string, string>();
+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 }));

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 {