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 <ryan@thewinograds.com>

* test(pi): separate stopped and failed history replay cases

---------

Co-authored-by: Ryan Winograd <ryan@thewinograds.com>
Co-authored-by: paseo-bot[bot] <266920839+paseo-bot[bot]@users.noreply.github.com>
This commit is contained in:
paseo-bot[bot]
2026-11-02 14:53:40 +00:00
committed by GitHub
co-authored by Ryan Winograd
parent 555818f5f7
commit 8018231a00
3 changed files with 134 additions and 13 deletions
@@ -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);
@@ -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<AgentStreamEvent> {
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<AgentRuntimeInfo> {
@@ -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,
@@ -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<PiAgentMessage[]> {
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<void> {}
async close(): Promise<void> {
this.closeCalls += 1;
}
emit(event: PiRuntimeEvent): void {
for (const subscriber of this.subscribers) {