mirror of
https://github.com/getpaseo/paseo.git
synced 2026-10-14 13:22:37 +00:00
Stop the provider when archiving an ACP agent that never answers close (#6213)
* fix(acp): terminate the provider when close requests never return Archive and delete wait on ACPAgentSession.close(). cancel() and unstable_closeSession() had no timeout. The ACP SDK does not reject a pending request when the provider never answers, so close() never reached terminateProcess. The agent was already marked archived and the provider process stayed alive. Both calls now use the same 2s withTimeout as probe close. The existing catch blocks still continue into terminateProcess. Example: - Input: archive an ACP session whose cancel and closeSession never settle - Before: close() waits forever and the provider process stays alive - After: both calls time out and terminateProcess runs * test(acp): archive close terminates a real provider that never answers session/close Spawn a real ACP provider process over stdio that advertises session close and never replies to it, then close the session the way archive does. Without the request bound, close() never returns and the provider keeps running. Co-authored-by: Ouyang Xingyuan <galaxyroaming@qq.com> * test(acp): move the close wait and provider cleanup into helpers Clear the race timer once close settles, and dispose the provider process and temp directory in afterEach. --------- Co-authored-by: Ouyang Xingyuan <galaxyroaming@qq.com> Co-authored-by: paseo-bot[bot] <266920839+paseo-bot[bot]@users.noreply.github.com>
This commit is contained in:
co-authored by
Ouyang Xingyuan
parent
c301d6e8aa
commit
a700d7875c
@@ -1,8 +1,10 @@
|
||||
import { type ChildProcess, type ChildProcessWithoutNullStreams } from "node:child_process";
|
||||
import { EventEmitter } from "node:events";
|
||||
import { mkdtemp, rm } from "node:fs/promises";
|
||||
import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
|
||||
import { createRequire } from "node:module";
|
||||
import { tmpdir } from "node:os";
|
||||
import path from "node:path";
|
||||
import { pathToFileURL } from "node:url";
|
||||
import { afterEach, describe, expect, test, vi } from "vitest";
|
||||
import {
|
||||
AgentSideConnection,
|
||||
@@ -3436,6 +3438,8 @@ interface ACPCloseInternals {
|
||||
child: ChildProcess | null;
|
||||
connection: unknown;
|
||||
sessionId: string | null;
|
||||
activeForegroundTurnId: string | null;
|
||||
agentCapabilities: { sessionCapabilities?: { close?: unknown } } | null;
|
||||
}
|
||||
|
||||
async function startTerminal(
|
||||
@@ -3502,6 +3506,38 @@ describe("ACPAgentSession close() tree-kill", () => {
|
||||
expect(child.kill).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
test("close() terminates the provider when cancel and closeSession never settle", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const terminator = new FakeTerminator();
|
||||
const session = createSession({ terminateProcess: terminator.terminate });
|
||||
const child = createTerminalChildStub();
|
||||
const cancel = vi.fn(() => new Promise<void>(() => undefined));
|
||||
const unstableCloseSession = vi.fn(() => new Promise<void>(() => undefined));
|
||||
const internals = asInternals<ACPCloseInternals>(session);
|
||||
internals.child = child;
|
||||
internals.sessionId = "session-1";
|
||||
internals.activeForegroundTurnId = "turn-1";
|
||||
internals.agentCapabilities = { sessionCapabilities: { close: {} } };
|
||||
internals.connection = { cancel, unstable_closeSession: unstableCloseSession };
|
||||
|
||||
let settled = false;
|
||||
const closing = (async () => {
|
||||
await session.close();
|
||||
settled = true;
|
||||
})();
|
||||
await vi.advanceTimersByTimeAsync(5_000);
|
||||
|
||||
expect(settled).toBe(true);
|
||||
expect(cancel).toHaveBeenCalledWith({ sessionId: "session-1" });
|
||||
expect(unstableCloseSession).toHaveBeenCalledWith({ sessionId: "session-1" });
|
||||
expect(terminator.terminated).toContain(child);
|
||||
await closing;
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
test("close() terminates running terminal child processes", async () => {
|
||||
const terminator = new FakeTerminator();
|
||||
const session = createSession({ terminateProcess: terminator.terminate });
|
||||
@@ -4220,3 +4256,117 @@ describe("ACP session/load invariant — cwd and mcpServers always passed", () =
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
const SILENT_CLOSE_ACP_AGENT = `
|
||||
import { readFileSync, writeFileSync } from "node:fs";
|
||||
import { Readable, Writable } from "node:stream";
|
||||
const { AgentSideConnection, PROTOCOL_VERSION, ndJsonStream } = await import(process.env.ACP_SDK_URL);
|
||||
writeFileSync(process.env.ACP_PID_FILE, String(process.pid));
|
||||
new AgentSideConnection(
|
||||
() => ({
|
||||
async initialize() {
|
||||
return {
|
||||
protocolVersion: PROTOCOL_VERSION,
|
||||
agentCapabilities: { sessionCapabilities: { close: {} } },
|
||||
};
|
||||
},
|
||||
async newSession() {
|
||||
return { sessionId: "silent-close-session" };
|
||||
},
|
||||
async authenticate() {},
|
||||
async cancel() {},
|
||||
unstable_closeSession() {
|
||||
return new Promise(() => {});
|
||||
},
|
||||
}),
|
||||
ndJsonStream(Writable.toWeb(process.stdout), Readable.toWeb(process.stdin)),
|
||||
);
|
||||
`;
|
||||
|
||||
function isProcessAlive(pid: number): boolean {
|
||||
try {
|
||||
process.kill(pid, 0);
|
||||
return true;
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
async function settlesWithin(promise: Promise<unknown>, timeoutMs: number): Promise<boolean> {
|
||||
let timer: NodeJS.Timeout | undefined;
|
||||
const timedOut = new Promise<false>((resolve) => {
|
||||
timer = setTimeout(() => resolve(false), timeoutMs);
|
||||
});
|
||||
try {
|
||||
return await Promise.race([promise.then(() => true), timedOut]);
|
||||
} finally {
|
||||
clearTimeout(timer);
|
||||
}
|
||||
}
|
||||
|
||||
interface SilentCloseProvider {
|
||||
session: ACPAgentSession;
|
||||
readPid(): Promise<number>;
|
||||
dispose(): Promise<void>;
|
||||
}
|
||||
|
||||
async function startSilentCloseProvider(): Promise<SilentCloseProvider> {
|
||||
const dir = await mkdtemp(path.join(tmpdir(), "paseo-acp-silent-close-"));
|
||||
const agentScript = path.join(dir, "agent.mjs");
|
||||
const pidFile = path.join(dir, "agent.pid");
|
||||
await writeFile(agentScript, SILENT_CLOSE_ACP_AGENT);
|
||||
const session = new ACPAgentSession(
|
||||
{ provider: "silent-close-acp", cwd: dir },
|
||||
{
|
||||
provider: "silent-close-acp",
|
||||
logger: createTestLogger(),
|
||||
defaultCommand: [process.execPath, agentScript],
|
||||
defaultModes: [],
|
||||
capabilities: {
|
||||
supportsStreaming: true,
|
||||
supportsSessionPersistence: true,
|
||||
supportsDynamicModes: true,
|
||||
supportsMcpServers: true,
|
||||
supportsReasoningStream: true,
|
||||
supportsToolInvocations: true,
|
||||
},
|
||||
launchEnv: {
|
||||
ACP_SDK_URL: pathToFileURL(
|
||||
createRequire(import.meta.url).resolve("@agentclientprotocol/sdk"),
|
||||
).href,
|
||||
ACP_PID_FILE: pidFile,
|
||||
},
|
||||
},
|
||||
);
|
||||
const readPid = async () => Number(await readFile(pidFile, "utf8"));
|
||||
return {
|
||||
session,
|
||||
readPid,
|
||||
async dispose() {
|
||||
const pid = await readPid().catch(() => null);
|
||||
if (pid !== null && isProcessAlive(pid)) {
|
||||
process.kill(pid, "SIGKILL");
|
||||
}
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
describe("ACPAgentSession close() with an unresponsive provider", () => {
|
||||
let provider: SilentCloseProvider | null = null;
|
||||
|
||||
afterEach(async () => {
|
||||
await provider?.dispose();
|
||||
provider = null;
|
||||
});
|
||||
|
||||
test("terminates a provider that never answers session/close", async () => {
|
||||
provider = await startSilentCloseProvider();
|
||||
await provider.session.initializeNewSession();
|
||||
const pid = await provider.readPid();
|
||||
expect(isProcessAlive(pid)).toBe(true);
|
||||
|
||||
expect(await settlesWithin(provider.session.close(), 8_000)).toBe(true);
|
||||
expect(isProcessAlive(pid)).toBe(false);
|
||||
}, 15_000);
|
||||
});
|
||||
|
||||
@@ -301,6 +301,10 @@ export function buildACPClientCapabilities(
|
||||
const PROBE_ENV: Record<string, string> = { NO_BROWSER: "true" };
|
||||
const ACP_DIAGNOSTIC_PHASE_TIMEOUT_MS = 20_000;
|
||||
const ACP_PROBE_CLOSE_TIMEOUT_MS = 2_000;
|
||||
// Archive and delete await close(). The ACP SDK does not reject a pending
|
||||
// request when the provider never answers, so these calls must be bounded
|
||||
// or the provider process is never terminated.
|
||||
const ACP_CLOSE_REQUEST_TIMEOUT_MS = 2_000;
|
||||
const ACP_IMPORT_HISTORY_LOAD_TIMEOUT_MS = 30_000;
|
||||
const ACP_IMPORT_HISTORY_BUDGET_MS = 60_000;
|
||||
|
||||
@@ -2476,13 +2480,21 @@ export class ACPAgentSession implements AgentSession, ACPClient {
|
||||
if (this.connection && this.sessionId) {
|
||||
try {
|
||||
if (this.activeForegroundTurnId) {
|
||||
await this.connection.cancel({ sessionId: this.sessionId });
|
||||
await withTimeout(
|
||||
this.connection.cancel({ sessionId: this.sessionId }),
|
||||
ACP_CLOSE_REQUEST_TIMEOUT_MS,
|
||||
`ACP cancel timed out after ${ACP_CLOSE_REQUEST_TIMEOUT_MS}ms`,
|
||||
);
|
||||
}
|
||||
} catch {}
|
||||
|
||||
try {
|
||||
if (this.agentCapabilities?.sessionCapabilities?.close) {
|
||||
await this.connection.unstable_closeSession({ sessionId: this.sessionId });
|
||||
await withTimeout(
|
||||
this.connection.unstable_closeSession({ sessionId: this.sessionId }),
|
||||
ACP_CLOSE_REQUEST_TIMEOUT_MS,
|
||||
`ACP closeSession timed out after ${ACP_CLOSE_REQUEST_TIMEOUT_MS}ms`,
|
||||
);
|
||||
}
|
||||
} catch (error) {
|
||||
this.logger.debug({ err: error }, "ACP closeSession failed during shutdown");
|
||||
|
||||
Reference in New Issue
Block a user