diff --git a/src/core-client/core-client.ts b/src/core-client/core-client.ts index f7c1e23..d630aa1 100644 --- a/src/core-client/core-client.ts +++ b/src/core-client/core-client.ts @@ -4,14 +4,29 @@ import { RpcMessage, decodeRpcMessage, encodeRpcMessage } from "./protocol"; export { RpcMessage } from "./protocol"; +/** What RpcClient needs from a child process — satisfied by CoreProcess, + * and by a lightweight fake in tests so they don't have to spawn a real + * process (which keeps the test runner alive until it fully exits). */ +export interface ProcessHandle extends EventEmitter { + write(data: string): void; + kill(): void; +} + export class RpcClient extends EventEmitter { - private process: CoreProcess; + private process: ProcessHandle; private nextId = 1; private pending = new Map void>(); + private dead = false; - constructor(command: string, args: string[], cwd: string, env?: Record) { + constructor( + command: string, + args: string[], + cwd: string, + env?: Record, + process: ProcessHandle = new CoreProcess(command, args, cwd, env) + ) { super(); - this.process = new CoreProcess(command, args, cwd, env); + this.process = process; this.process.on("line", (line: string) => { const message = decodeRpcMessage(line); @@ -20,10 +35,23 @@ export class RpcClient extends EventEmitter { } }); this.process.on("stderr", (text: string) => this.emit("stderr", text)); - this.process.on("exit", (info: { code: number | null; signal: string | null }) => - this.emit("exit", info) - ); - this.process.on("spawnError", (error: NodeJS.ErrnoException) => this.emit("spawnError", error)); + this.process.on("exit", (info: { code: number | null; signal: string | null }) => { + this.rejectPending("pycodeloop serve exited"); + this.emit("exit", info); + }); + this.process.on("spawnError", (error: NodeJS.ErrnoException) => { + this.rejectPending(error.message); + this.emit("spawnError", error); + }); + } + + private rejectPending(message: string): void { + this.dead = true; + const snapshot = new Map(this.pending); + this.pending.clear(); + for (const [id, resolve] of snapshot) { + resolve({ jsonrpc: "2.0", id, error: { code: -32000, message } }); + } } private dispatch(message: RpcMessage): void { @@ -46,6 +74,13 @@ export class RpcClient extends EventEmitter { request(method: string, params: Record = {}): Promise { const id = String(this.nextId++); + if (this.dead) { + return Promise.resolve({ + jsonrpc: "2.0", + id, + error: { code: -32000, message: "RpcClient is disposed" }, + }); + } return new Promise((resolve) => { this.pending.set(id, resolve); this.process.write(encodeRpcMessage({ jsonrpc: "2.0", id, method, params })); @@ -53,7 +88,7 @@ export class RpcClient extends EventEmitter { } dispose(): void { - this.pending.clear(); + this.rejectPending("Disposed"); this.process.kill(); } } diff --git a/test/core-client.test.ts b/test/core-client.test.ts new file mode 100644 index 0000000..ad40997 --- /dev/null +++ b/test/core-client.test.ts @@ -0,0 +1,86 @@ +import assert from "node:assert/strict"; +import { EventEmitter } from "node:events"; +import { test } from "node:test"; +import { ProcessHandle, RpcClient } from "../src/core-client/core-client"; + +class FakeProcess extends EventEmitter implements ProcessHandle { + killed = false; + + write(_data: string): void {} + + kill(): void { + this.killed = true; + } +} + +test("pending requests reject when the process exits mid-flight", async () => { + const fake = new FakeProcess(); + const client = new RpcClient("fake", [], process.cwd(), undefined, fake); + + const pending = client.request("chat/send", { prompt: "hi" }); + fake.emit("exit", { code: 1, signal: null }); + + const response = await pending; + assert.equal(response.error?.code, -32000); +}); + +test("pending requests reject when the process fails to spawn", async () => { + const fake = new FakeProcess(); + const client = new RpcClient("fake", [], process.cwd(), undefined, fake); + + const pending = client.request("chat/send", { prompt: "hi" }); + fake.emit("spawnError", new Error("ENOENT")); + + const response = await pending; + assert.equal(response.error?.code, -32000); + assert.equal(response.error?.message, "ENOENT"); +}); + +test("dispose rejects any still-pending request and kills the process", async () => { + const fake = new FakeProcess(); + const client = new RpcClient("fake", [], process.cwd(), undefined, fake); + + const pending = client.request("chat/send", { prompt: "hi" }); + client.dispose(); + + const response = await pending; + assert.equal(response.error?.code, -32000); + assert.equal(fake.killed, true); +}); + +test("a real RPC response still resolves normally, not through rejectPending", async () => { + const fake = new FakeProcess(); + const client = new RpcClient("fake", [], process.cwd(), undefined, fake); + + const pending = client.request("chat/send", { prompt: "hi" }); + fake.emit("line", JSON.stringify({ jsonrpc: "2.0", id: "1", result: { text: "ok" } })); + + const response = await pending; + assert.deepEqual(response.result, { text: "ok" }); +}); + +test("a re-entrant request made from inside a rejected .then() is not silently dropped", async () => { + const fake = new FakeProcess(); + const client = new RpcClient("fake", [], process.cwd(), undefined, fake); + + const retryResponse = client.request("chat/send", { prompt: "first" }).then((first) => { + assert.equal(first.error?.code, -32000); + return client.request("chat/send", { prompt: "retry" }); + }); + fake.emit("exit", { code: 1, signal: null }); + + const response = await retryResponse; + assert.equal(response.error?.code, -32000); + assert.equal(response.error?.message, "RpcClient is disposed"); +}); + +test("requests made after the process has already exited reject immediately instead of hanging", async () => { + const fake = new FakeProcess(); + const client = new RpcClient("fake", [], process.cwd(), undefined, fake); + + fake.emit("exit", { code: 1, signal: null }); + const response = await client.request("diagnostics/status"); + + assert.equal(response.error?.code, -32000); + assert.equal(response.error?.message, "RpcClient is disposed"); +});