Reject calls matched by invalid RPC responses

src/__tests__/session.test.ts

@@ -1479,44 +1479,60 @@ describe("session error resilience", () => {
   });
 
   describe("Invalid response", () => {
-    it("response that fails type guard is logged with INVALID_RESPONSE code", async () => {
+    it("rejects a matched invalid response before reporting it and keeps the session usable", async () => {
       const [transportA, transportB] = createLinkedTransports();
 
       type EchoService = {
         echo(v: string): Promise<string>;
       };
 
       const errors: unknown[] = [];
+      const diagnosticFailure = new Error("Diagnostic callback failed");
       const sessionA = new RpcSession<EchoService, Record<string, never>>(
         transportA,
         {},
         {
           role: "initiator",
           onError(err) {
             errors.push(err);
+            throw diagnosticFailure;
           },
         },
       );
 
       const sessionB = new RpcSession<Record<string, never>, EchoService>(
         transportB,
-        { echo: (v: string) => Promise.resolve(v) },
+        {
+          echo(value) {
+            if (value === "pending") return new Promise(() => {});
+            return Promise.resolve(value);
+          },
+        },
         { role: "acceptor" },
       );
 
-      // Inject a response-like object that fails the type guard (missing jsonrpc version)
-      transportB.send(
-        JSON.stringify({
-          id: 1,
-          result: "bad",
-        }),
-      );
-
-      await new Promise((resolve) => setTimeout(resolve, 10));
+      const send = vi.spyOn(transportA, "send");
+      const pending = sessionA.remote.echo("pending");
+      const request = JSON.parse(send.mock.calls[0][0]);
 
-      expect(errors.length).toBe(1);
+      // Inject a response-like object that fails the type guard (missing jsonrpc version)
+      expect(() =>
+        transportB.send(
+          JSON.stringify({
+            id: request.id,
+            result: "bad",
+          }),
+        ),
+      ).toThrow(diagnosticFailure);
+
+      await expect(pending).rejects.toMatchObject({
+        code: "INVALID_RESPONSE",
+        message: "Received invalid JSON-RPC response",
+      });
+      expect(errors).toHaveLength(1);
       expect(errors[0]).toBeInstanceOf(RpcProtocolError);
       expect((errors[0] as RpcProtocolError).code).toBe("INVALID_RESPONSE");
+      await expect(sessionA.remote.echo("ok")).resolves.toBe("ok");
 
       sessionA.close();
       sessionB.close();

src/session.ts

@@ -154,9 +154,18 @@ class RpcSessionImpl<TRemote extends object, TLocal extends object> {
 
   private handleIncomingResponse(parsed: unknown): void {
     if (!isJsonRpcResponse(parsed)) {
-      this.onError?.(
-        new RpcProtocolError("INVALID_RESPONSE", "Received invalid JSON-RPC response"),
-      );
+      const error = new RpcProtocolError("INVALID_RESPONSE", "Received invalid JSON-RPC response");
+      if (typeof parsed === "object" && parsed !== null) {
+        const id = (parsed as Record<string, unknown>).id;
+        if (typeof id === "string" || typeof id === "number") {
+          const pending = this.pendingCalls.get(id);
+          if (pending) {
+            this.pendingCalls.delete(id);
+            pending.reject(error);
+          }
+        }
+      }
+      this.onError?.(error);
       return;
     }