🪲 BUG-#1: Fix RpcClient pending requests leak

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<string, (message: RpcMessage) => void>();
+  private dead = false;
 
-  constructor(command: string, args: string[], cwd: string, env?: Record<string, string>) {
+  constructor(
+    command: string,
+    args: string[],
+    cwd: string,
+    env?: Record<string, string>,
+    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,14 +74,21 @@ export class RpcClient extends EventEmitter {
 
   request(method: string, params: Record<string, unknown> = {}): Promise<RpcMessage> {
     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 }));
     });
   }
 
   dispose(): void {
-    this.pending.clear();
+    this.rejectPending("Disposed");
     this.process.kill();
   }
 }

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");
+});