From 8018231a00a6b45ed6cda1856f954ba14b4824c0 Mon Sep 17 00:00:00 2001 From: "paseo-bot[bot]" <266920839+paseo-bot[bot]@users.noreply.github.com> Date: Mon, 2 Nov 2026 14:53:40 +0000 Subject: [PATCH] fix(pi): release archived chat runtimes after history replay (#8554) * fix(pi): close history-only runtimes after replay Archived timeline reads resume a Pi RPC process to reconstruct the saved conversation. The history-only session remained open after replay, so browsing or synchronizing archived agents left one idle Pi process per agent. Mark history-purpose resumes for closure and close the runtime in streamHistory's finally path. Interactive sessions remain resident, and idempotent close keeps later manager cleanup safe. * fix(pi): preserve history cleanup across current resume behavior Keep the original session identity and missing-folder cleanup when releasing history-only runtimes. Use an existing directory in the regression so the newer resume fallback does not try to copy a nonexistent session fixture. Cover interrupted and failed replay, interactive residency, and automatic cleanup of a missing-folder history copy. Co-authored-by: Ryan Winograd * test(pi): separate stopped and failed history replay cases --------- Co-authored-by: Ryan Winograd Co-authored-by: paseo-bot[bot] <266920839+paseo-bot[bot]@users.noreply.github.com> --- .../server/agent/providers/pi/agent.test.ts | 112 +++++++++++++++++- .../src/server/agent/providers/pi/agent.ts | 28 +++-- .../agent/providers/pi/test-utils/fake-pi.ts | 7 +- 3 files changed, 134 insertions(+), 13 deletions(-) diff --git a/packages/server/src/server/agent/providers/pi/agent.test.ts b/packages/server/src/server/agent/providers/pi/agent.test.ts index 7dc3fd184c..008a3a4fac 100644 --- a/packages/server/src/server/agent/providers/pi/agent.test.ts +++ b/packages/server/src/server/agent/providers/pi/agent.test.ts @@ -1600,6 +1600,111 @@ describe("PiRpcAgentSession", () => { ]); }); + test("closes a history-only Pi runtime after replaying its timeline", async () => { + const pi = new FakePi(); + pi.queueSessionSetup((runtimeSession) => { + runtimeSession.messages = [ + { role: "user", content: "Old question" }, + { role: "assistant", content: [{ type: "text", text: "Old answer" }] }, + ]; + }); + + const client: import("../../agent-sdk-types.js").AgentClient = createClient(pi); + const session = await client.resumeSession( + { + provider: "pi", + sessionId: "pi-session-1", + nativeHandle: "/tmp/native-pi-session", + metadata: { cwd: tmpdir() }, + }, + {}, + undefined, + { purpose: "history" }, + ); + + onTestFinished(() => session.close()); + const history: AgentStreamEvent[] = []; + for await (const event of session.streamHistory()) history.push(event); + + expect(history).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + type: "timeline", + provider: "pi", + item: expect.objectContaining({ type: "assistant_message", text: "Old answer" }), + }), + ]), + ); + expect(pi.latestSession().closeCalls).toBe(1); + + await session.close(); + expect(pi.latestSession().closeCalls).toBe(1); + }); + + test("closes a history-only Pi runtime when replay is stopped", async () => { + const pi = new FakePi(); + pi.queueSessionSetup((runtimeSession) => { + runtimeSession.messages = [ + { role: "user", content: "Old question" }, + { role: "assistant", content: [{ type: "text", text: "Old answer" }] }, + ]; + }); + const session = await createClient(pi).resumeSession( + { + provider: "pi", + sessionId: "pi-session-1", + nativeHandle: "/tmp/native-pi-session", + metadata: { cwd: tmpdir() }, + }, + {}, + undefined, + { purpose: "history" }, + ); + onTestFinished(() => session.close()); + const history = session.streamHistory(); + expect((await history.next()).done).toBe(false); + await history.return(); + expect(pi.latestSession().closeCalls).toBe(1); + }); + + test("closes a history-only Pi runtime when replay fails", async () => { + const pi = new FakePi(); + pi.queueSessionSetup((runtimeSession) => { + runtimeSession.getMessagesError = new Error("History unavailable"); + }); + const session = await createClient(pi).resumeSession( + { + provider: "pi", + sessionId: "pi-session-1", + nativeHandle: "/tmp/native-pi-session", + metadata: { cwd: tmpdir() }, + }, + {}, + undefined, + { purpose: "history" }, + ); + onTestFinished(() => session.close()); + const history = session.streamHistory(); + await expect(history.next()).rejects.toThrow("History unavailable"); + expect(pi.latestSession().closeCalls).toBe(1); + }); + + test("keeps an interactive Pi runtime open after replaying history", async () => { + const pi = new FakePi(); + const session = await createClient(pi).resumeSession({ + provider: "pi", + sessionId: "pi-session-1", + nativeHandle: "/tmp/native-pi-session", + metadata: { cwd: tmpdir() }, + }); + onTestFinished(() => session.close()); + for await (const _event of session.streamHistory()) { + // Hydrate the session before its next interactive prompt. + } + expect(pi.latestSession().closeCalls).toBe(0); + await expect(session.getRuntimeInfo()).resolves.toMatchObject({ provider: "pi" }); + }); + test("resumes a session whose model was removed on the model Pi falls back to", async () => { const pi = new FakePi(); pi.removeModel("9router/deepseek-v4-flash"); @@ -3740,14 +3845,17 @@ readline.createInterface({ input: process.stdin }).on("line", (line) => { }, ); - test("a history resume of an agent whose folder is gone removes its session copy on close and leaves the original unchanged", async () => { + test("a history resume of an agent whose folder is gone removes its session copy after replay and leaves the original unchanged", async () => { const archived = writeArchivedSession(createDeletedFolder()); const pi = new FakePi(); const session = await resumeArchived(createClient(pi), archived, "history"); const sessionCopy = pi.recordedLaunches[0]!.session!; expect(existsSync(sessionCopy)).toBe(true); - await session.close(); + for await (const _event of session.streamHistory()) { + // Completion releases both the runtime and its temporary session copy. + } + expect(pi.latestSession().closeCalls).toBe(1); expect(existsSync(path.dirname(sessionCopy))).toBe(false); expect(readFileSync(archived.sessionFile)).toEqual(archived.originalBytes); diff --git a/packages/server/src/server/agent/providers/pi/agent.ts b/packages/server/src/server/agent/providers/pi/agent.ts index 8051c57fa9..368c7405a7 100644 --- a/packages/server/src/server/agent/providers/pi/agent.ts +++ b/packages/server/src/server/agent/providers/pi/agent.ts @@ -253,6 +253,7 @@ interface PiRpcAgentSessionOptions { currentModeId?: string | null; persistedSessionFile?: string; cleanup?: () => void; + closeAfterHistory?: boolean; extensionTimeoutMs?: number; logger: Logger; usagePollScheduler?: PiUsagePollScheduler; @@ -1255,6 +1256,7 @@ export class PiRpcAgentSession implements AgentSession { this.currentModeId = options.currentModeId ?? null; this.persistedSessionFile = options.persistedSessionFile; this.cleanup = options.cleanup; + this.closeAfterHistory = options.closeAfterHistory ?? false; this.extensionTimeoutMs = options.extensionTimeoutMs ?? DEFAULT_PI_EXTENSION_RESULT_TIMEOUT_MS; this.logger = options.logger; this.extensionHost = createPiExtensionHost(this.logger); @@ -1286,6 +1288,7 @@ export class PiRpcAgentSession implements AgentSession { private readonly config: AgentSessionConfig; private readonly persistedSessionFile?: string; private readonly cleanup?: () => void; + private readonly closeAfterHistory: boolean; private readonly extensionTimeoutMs: number; get id(): string | null { @@ -1454,16 +1457,20 @@ export class PiRpcAgentSession implements AgentSession { } async *streamHistory(): AsyncGenerator { - await this.requestEntryCapture("history"); - yield* streamPiHistory( - this.provider, - await this.runtimeSession.getMessages(), - this.contextUserEntries, - {}, - // At most eight 2 MiB child files per replay; later cards retain their summaries. - createPiExtensionHost(this.logger, undefined, 16 * 1024 * 1024), - this.closeController.signal, - ); + try { + await this.requestEntryCapture("history"); + yield* streamPiHistory( + this.provider, + await this.runtimeSession.getMessages(), + this.contextUserEntries, + {}, + // At most eight 2 MiB child files per replay; later cards retain their summaries. + createPiExtensionHost(this.logger, undefined, 16 * 1024 * 1024), + this.closeController.signal, + ); + } finally { + if (this.closeAfterHistory) await this.close(); + } } async getRuntimeInfo(): Promise { @@ -2738,6 +2745,7 @@ export class PiRpcAgentClient implements AgentClient { capabilities: capabilitiesForSession(mcp !== null), persistedSessionFile: sessionCopy ? sessionFile : undefined, cleanup: cleanupTempFiles, + closeAfterHistory: options?.purpose === "history", extensionTimeoutMs: providerOptions.extensionTimeoutMs, logger: this.logger, usagePollScheduler: this.usagePollScheduler, diff --git a/packages/server/src/server/agent/providers/pi/test-utils/fake-pi.ts b/packages/server/src/server/agent/providers/pi/test-utils/fake-pi.ts index 90a80e4c66..69aab7b3b8 100644 --- a/packages/server/src/server/agent/providers/pi/test-utils/fake-pi.ts +++ b/packages/server/src/server/agent/providers/pi/test-utils/fake-pi.ts @@ -153,6 +153,7 @@ export class FakePiSession implements PiRuntimeSession { // The user entries on the current branch that getMessages() replays. contextUserEntries: FakePiUserEntry[] = []; abortRequested = false; + closeCalls = 0; readonly canceledExtensionUiRequests: string[] = []; readonly extensionUiResponses: Array<{ id: string; @@ -161,6 +162,7 @@ export class FakePiSession implements PiRuntimeSession { setModelResult: PiModel | null = null; models: PiModel[] = []; messages: PiAgentMessage[] = []; + getMessagesError: Error | null = null; stats: PiSessionStats = { tokens: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, cost: 0, @@ -303,6 +305,7 @@ export class FakePiSession implements PiRuntimeSession { } async getMessages(): Promise { + if (this.getMessagesError) throw this.getMessagesError; return this.messages; } @@ -422,7 +425,9 @@ export class FakePiSession implements PiRuntimeSession { this.respondToExtensionUiRequest(id, { cancelled: true }); } - async close(): Promise {} + async close(): Promise { + this.closeCalls += 1; + } emit(event: PiRuntimeEvent): void { for (const subscriber of this.subscribers) {