mirror of
https://github.com/getpaseo/paseo.git
synced 2026-10-14 13:22:37 +00:00
fix(server): release ACP provider process after history resume (#6766)
* fix(server): release ACP provider process after history resume Resuming a persisted agent with purpose "history" only replays the session into the canonical timeline, but ACP sessions kept the spawned CLI process resident afterwards. Every client that fetches an archived agent's timeline (the app reconnects fetching every visible agent) therefore re-materialized a provider process per archived agent and nothing ever released it — measured ~280 MB resident per dsh agent. After the replay and configured overrides finish, a history-purpose resume now settles the commands gate, drops the connection, and terminates the child. The session object stays registered with the replayed history so read paths keep working; the existing `this.child !== child` exit guard already ignores the released process, and interactive use goes through reloadAgentSession which spawns a fresh process. This matches the runtime-only contract the Codex provider already implements. * fix(server): retain child refs when history-runtime release times out Addresses Greptile review: if terminateProcess returns kill-timeout the child stays registered so close() can retry instead of leaking an untracked process. The test now asserts the process is already dead when resumeSession resolves, which pins the documented release-before-return contract instead of a post-hoc poll. Greptile iteration 1.
This commit is contained in:
@@ -1750,6 +1750,7 @@ export class ACPAgentSession implements AgentSession, ACPClient {
|
||||
private fallbackAssistantMessageId: string | null = null;
|
||||
private closed = false;
|
||||
private historyPending = false;
|
||||
private historyRuntimeReleased = false;
|
||||
private replayingHistory = false;
|
||||
private bootstrapThreadEventPending = false;
|
||||
private readonly terminateProcess: ProcessTerminator;
|
||||
@@ -1867,11 +1868,38 @@ export class ACPAgentSession implements AgentSession, ACPClient {
|
||||
}
|
||||
|
||||
await this.applyConfiguredOverrides();
|
||||
|
||||
if (this.resumePurpose === "history") {
|
||||
await this.releaseHistoryRuntime();
|
||||
}
|
||||
} catch (error) {
|
||||
await this.closeAfterInitializationFailure(error);
|
||||
}
|
||||
}
|
||||
|
||||
// History resumes exist to replay persisted events into the canonical
|
||||
// timeline; the provider process is released once replay finishes so
|
||||
// archived agents hold no resident runtime. Interactive work goes through
|
||||
// reloadAgentSession, which spawns a fresh process.
|
||||
private async releaseHistoryRuntime(): Promise<void> {
|
||||
const child = this.child;
|
||||
if (child) {
|
||||
const result = await this.terminateProcess(child, {
|
||||
gracefulTimeoutMs: 2_000,
|
||||
forceTimeoutMs: 2_000,
|
||||
});
|
||||
if (result === "kill-timeout") {
|
||||
// Keep the child registered so close() can retry the shutdown.
|
||||
this.logger.warn("ACP history resume could not release the provider process");
|
||||
return;
|
||||
}
|
||||
}
|
||||
this.historyRuntimeReleased = true;
|
||||
this.connection = null;
|
||||
this.child = null;
|
||||
this.settleCommandsReady();
|
||||
}
|
||||
|
||||
private async closeAfterInitializationFailure(error: unknown): Promise<never> {
|
||||
try {
|
||||
await this.close();
|
||||
@@ -2841,7 +2869,7 @@ export class ACPAgentSession implements AgentSession, ACPClient {
|
||||
});
|
||||
const spawnError = rejectOnSpawnError(child, stderrChunks);
|
||||
child.once("exit", (code, signal) => {
|
||||
if (this.closed) {
|
||||
if (this.closed || this.historyRuntimeReleased) {
|
||||
return;
|
||||
}
|
||||
if (this.activeForegroundTurnId) {
|
||||
|
||||
+61
-2
@@ -1,4 +1,4 @@
|
||||
import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
|
||||
import { mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import path from "node:path";
|
||||
|
||||
@@ -68,8 +68,60 @@ describe("GenericACPAgentClient archived history", () => {
|
||||
).rejects.toThrow("ENOENT");
|
||||
});
|
||||
});
|
||||
|
||||
test("releases the provider process once an archived session's history is replayed", async () => {
|
||||
await withFakeACPAgent(async ({ command, removedCwd, pidPath }) => {
|
||||
const client = new GenericACPAgentClient({ logger: createTestLogger(), command });
|
||||
|
||||
const session = await client.resumeSession(
|
||||
{ provider: "acp", sessionId: "archived-session", metadata: { cwd: removedCwd } },
|
||||
undefined,
|
||||
undefined,
|
||||
{ purpose: "history" },
|
||||
);
|
||||
try {
|
||||
const pid = Number(await readFile(pidPath, "utf8"));
|
||||
expect(isProcessAlive(pid)).toBe(false);
|
||||
expect(await collectTimelineText(session.streamHistory())).toEqual([
|
||||
"user_message:What is in the README?",
|
||||
"assistant_message:It describes the project.",
|
||||
]);
|
||||
} finally {
|
||||
await session.close();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
test("keeps the provider process for an interactive resume", async () => {
|
||||
await withFakeACPAgent(async ({ command, removedCwd, pidPath }) => {
|
||||
await mkdir(removedCwd, { recursive: true });
|
||||
const client = new GenericACPAgentClient({ logger: createTestLogger(), command });
|
||||
|
||||
const session = await client.resumeSession(
|
||||
{ provider: "acp", sessionId: "archived-session", metadata: { cwd: removedCwd } },
|
||||
undefined,
|
||||
undefined,
|
||||
{ purpose: "interactive" },
|
||||
);
|
||||
try {
|
||||
const pid = Number(await readFile(pidPath, "utf8"));
|
||||
expect(isProcessAlive(pid)).toBe(true);
|
||||
} finally {
|
||||
await session.close();
|
||||
}
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
function isProcessAlive(pid: number): boolean {
|
||||
try {
|
||||
process.kill(pid, 0);
|
||||
return true;
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
async function collectTimelineText(events: AsyncGenerator<AgentStreamEvent>): Promise<string[]> {
|
||||
const text: string[] = [];
|
||||
for await (const event of events) {
|
||||
@@ -88,17 +140,20 @@ async function withFakeACPAgent(
|
||||
command: [string, ...string[]];
|
||||
removedCwd: string;
|
||||
loadRequestsPath: string;
|
||||
pidPath: string;
|
||||
}) => Promise<void>,
|
||||
): Promise<void> {
|
||||
const testDir = await mkdtemp(path.join(tmpdir(), "paseo-acp-archived-history-"));
|
||||
try {
|
||||
const scriptPath = path.join(testDir, "fake-acp-agent.cjs");
|
||||
const loadRequestsPath = path.join(testDir, "load-requests.json");
|
||||
const pidPath = path.join(testDir, "agent.pid");
|
||||
await writeFile(scriptPath, fakeACPAgentScript, "utf8");
|
||||
await run({
|
||||
command: [process.execPath, scriptPath, loadRequestsPath],
|
||||
command: [process.execPath, scriptPath, loadRequestsPath, pidPath],
|
||||
removedCwd: path.join(testDir, "removed-worktree"),
|
||||
loadRequestsPath,
|
||||
pidPath,
|
||||
});
|
||||
} finally {
|
||||
await rm(testDir, { recursive: true, force: true });
|
||||
@@ -110,6 +165,10 @@ const fs = require("node:fs");
|
||||
const readline = require("node:readline");
|
||||
|
||||
const loadRequestsPath = process.argv[2];
|
||||
const pidPath = process.argv[3];
|
||||
if (pidPath) {
|
||||
fs.writeFileSync(pidPath, String(process.pid));
|
||||
}
|
||||
const loadRequests = [];
|
||||
const rl = readline.createInterface({ input: process.stdin });
|
||||
|
||||
|
||||
Reference in New Issue
Block a user