fix(rpc): reject pending calls when target disconnects

packages/opencode/src/util/rpc.ts

@@ -19,16 +19,33 @@ export function emit(event: string, data: unknown) {
 export function client<T extends Definition>(target: {
   postMessage: (data: string) => void | null
   onmessage: ((this: Worker, ev: MessageEvent<any>) => any) | null
+  addEventListener?: (event: string, handler: (...args: any[]) => void) => void
 }) {
-  const pending = new Map<number, (result: any) => void>()
+  const pending = new Map<number, { resolve: (result: any) => void; reject: (error: unknown) => void }>()
   const listeners = new Map<string, Set<(data: any) => void>>()
   let id = 0
+  let closed = false
+
+  const rejectAll = () => {
+    closed = true
+    const err = new Error("RPC target disconnected")
+    for (const entry of pending.values()) {
+      entry.reject(err)
+    }
+    pending.clear()
+  }
+
+  if (typeof target.addEventListener === "function") {
+    target.addEventListener("error", rejectAll)
+    target.addEventListener("messageerror", rejectAll)
+  }
+
   target.onmessage = async (evt) => {
     const parsed = JSON.parse(evt.data)
     if (parsed.type === "rpc.result") {
-      const resolve = pending.get(parsed.id)
-      if (resolve) {
-        resolve(parsed.result)
+      const entry = pending.get(parsed.id)
+      if (entry) {
+        entry.resolve(parsed.result)
         pending.delete(parsed.id)
       }
     }
@@ -44,9 +61,15 @@ export function client<T extends Definition>(target: {
   return {
     call<Method extends keyof T>(method: Method, input: Parameters<T[Method]>[0]): Promise<ReturnType<T[Method]>> {
       const requestId = id++
-      return new Promise((resolve) => {
-        pending.set(requestId, resolve)
-        target.postMessage(JSON.stringify({ type: "rpc.request", method, input, id: requestId }))
+      if (closed) return Promise.reject(new Error("RPC target disconnected"))
+      return new Promise((resolve, reject) => {
+        pending.set(requestId, { resolve, reject })
+        try {
+          target.postMessage(JSON.stringify({ type: "rpc.request", method, input, id: requestId }))
+        } catch (err) {
+          pending.delete(requestId)
+          reject(err)
+        }
       })
     },
     on<Data>(event: string, handler: (data: Data) => void) {

packages/opencode/test/fixture/crashing-worker.ts

@@ -0,0 +1,3 @@
+// Worker fixture: crashes immediately with an uncaught exception
+// Used by rpc-hang.test.ts to test real Worker death scenarios
+throw new Error("Worker crashed on purpose")

packages/opencode/test/fixture/responding-worker.ts

@@ -0,0 +1,10 @@
+// Worker fixture: responds to one RPC call then throws
+import { Rpc } from "../../src/util/rpc"
+
+const rpc = {
+  async ping() {
+    return "pong"
+  },
+}
+
+Rpc.listen(rpc)

packages/opencode/test/rpc-hang.test.ts

@@ -0,0 +1,113 @@
+import { Rpc } from "../src/util/rpc"
+import { describe, it, expect, mock } from "bun:test"
+
+describe("Rpc.client", () => {
+  it("rejects pending calls when target disconnects (error event)", async () => {
+    const listeners = new Map<string, Set<(...args: any[]) => void>>()
+    const target = {
+      postMessage: mock(() => {}),
+      onmessage: null as any,
+      addEventListener(event: string, handler: (...args: any[]) => void) {
+        let set = listeners.get(event)
+        if (!set) {
+          set = new Set()
+          listeners.set(event, set)
+        }
+        set.add(handler)
+      },
+    }
+
+    const client = Rpc.client<{ ping: (input: {}) => string }>(target as any)
+    const call = client.call("ping", {})
+
+    const errorHandler = listeners.get("error")?.values().next().value
+    errorHandler?.(new ErrorEvent("error", { message: "worker crashed" }))
+
+    await expect(call).rejects.toThrow("RPC target disconnected")
+  })
+
+  it("rejects pending calls on messageerror", async () => {
+    const listeners = new Map<string, Set<(...args: any[]) => void>>()
+    const target = {
+      postMessage: mock(() => {}),
+      onmessage: null as any,
+      addEventListener(event: string, handler: (...args: any[]) => void) {
+        let set = listeners.get(event)
+        if (!set) {
+          set = new Set()
+          listeners.set(event, set)
+        }
+        set.add(handler)
+      },
+    }
+
+    const client = Rpc.client<{ ping: (input: {}) => string }>(target as any)
+    const call = client.call("ping", {})
+
+    const errorHandler = listeners.get("messageerror")?.values().next().value
+    errorHandler?.(new MessageEvent("messageerror"))
+
+    await expect(call).rejects.toThrow("RPC target disconnected")
+  })
+
+  it("rejects calls made after target already disconnected", async () => {
+    const listeners = new Map<string, Set<(...args: any[]) => void>>()
+    const target = {
+      postMessage: mock(() => {}),
+      onmessage: null as any,
+      addEventListener(event: string, handler: (...args: any[]) => void) {
+        let set = listeners.get(event)
+        if (!set) {
+          set = new Set()
+          listeners.set(event, set)
+        }
+        set.add(handler)
+      },
+    }
+
+    const client = Rpc.client<{ ping: (input: {}) => string }>(target as any)
+
+    // First call: worker dies
+    const call1 = client.call("ping", {})
+    const errorHandler = listeners.get("error")?.values().next().value
+    errorHandler?.(new ErrorEvent("error", { message: "worker crashed" }))
+    await expect(call1).rejects.toThrow("RPC target disconnected")
+
+    // Second call: made after already disconnected
+    const call2 = client.call("ping", {})
+    await expect(call2).rejects.toThrow("RPC target disconnected")
+  })
+
+  it("resolves normally when target responds", async () => {
+    const target = {
+      postMessage: mock((data: string) => {
+        const parsed = JSON.parse(data)
+        if (parsed.type === "rpc.request") {
+          setImmediate(() => {
+            target.onmessage?.({ data: JSON.stringify({ type: "rpc.result", result: "pong", id: parsed.id }) })
+          })
+        }
+      }),
+      onmessage: null as any,
+    }
+
+    const client = Rpc.client<{ ping: (input: {}) => string }>(target as any)
+    const result = await client.call("ping", {})
+
+    expect(result).toBe("pong")
+    expect(target.postMessage).toHaveBeenCalledTimes(1)
+  })
+
+  it("rejects when a real Bun Worker throws an uncaught error", async () => {
+    const file = new URL("./fixture/crashing-worker.ts", import.meta.url)
+    const worker = new Worker(file)
+    const client = Rpc.client<{ ping: (input: {}) => string }>(worker)
+
+    // Give the worker time to boot and throw
+    await new Promise((resolve) => setTimeout(resolve, 200))
+
+    const call = client.call("ping", {})
+    await expect(call).rejects.toThrow("RPC target disconnected")
+    worker.terminate()
+  })
+})