fix(web): settle RPCs parked on reconnect readiness

apps/web/src/__tests__/connection.test.ts

@@ -2,6 +2,7 @@ import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
 import { createWsTransport } from "../transport/ws-transport";
 
 class MockWebSocket {
+  static readonly OPEN = 1;
   onopen: (() => void) | null = null;
   onclose: ((event: { code: number; reason: string }) => void) | null = null;
   onerror: (() => void) | null = null;
@@ -232,3 +233,199 @@ describe("4001 auth failure handling", () => {
     vi.useRealTimers();
   });
 });
+
+/**
+ * Browser-faithful socket: send() throws while CONNECTING and silently
+ * discards while CLOSING/CLOSED, matching the WebSocket spec. The silent
+ * drop is what orphans an RPC that resumes onto a dead socket.
+ */
+class LifecycleSocket {
+  static instances: LifecycleSocket[] = [];
+  static readonly OPEN = 1;
+  onopen: (() => void) | null = null;
+  onclose: ((event: { code: number; reason: string }) => void) | null = null;
+  onerror: (() => void) | null = null;
+  onmessage: ((event: { data: string }) => void) | null = null;
+  readyState = 0;
+  binaryType = "";
+
+  constructor() {
+    LifecycleSocket.instances.push(this);
+  }
+
+  send(raw: string | ArrayBuffer): void {
+    if (this.readyState === 0) throw new Error("InvalidStateError: socket is CONNECTING");
+    if (this.readyState !== LifecycleSocket.OPEN) return;
+    if (typeof raw !== "string") return;
+    const request = JSON.parse(raw) as { id: string; method?: string };
+    const result = request.method === "workspace.list" ? [{ id: "fixture" }] : null;
+    queueMicrotask(() =>
+      this.onmessage?.({ data: JSON.stringify({ id: request.id, result }) }),
+    );
+  }
+
+  open(): void {
+    this.readyState = 1;
+    this.onopen?.();
+  }
+
+  fail(): void {
+    this.readyState = 3;
+    this.onclose?.({ code: 1006, reason: "fixture failure" });
+  }
+
+  close(): void {
+    this.readyState = 3;
+    this.onclose?.({ code: 1000, reason: "" });
+  }
+}
+
+describe("RPC readiness across reconnects (issue #1733)", () => {
+  let transport: ReturnType<typeof createWsTransport>;
+
+  function settleSpy(promise: Promise<unknown>): { state: string } {
+    const spy = { state: "pending" };
+    void promise.then(
+      () => { spy.state = "resolved"; },
+      () => { spy.state = "rejected"; },
+    );
+    return spy;
+  }
+
+  beforeEach(() => {
+    vi.useFakeTimers();
+    LifecycleSocket.instances = [];
+    vi.stubGlobal("WebSocket", LifecycleSocket);
+    transport = createWsTransport("ws://fixture.invalid");
+  });
+
+  afterEach(() => {
+    transport.close();
+    vi.useRealTimers();
+    vi.unstubAllGlobals();
+  });
+
+  it("settles an RPC waiting during a failed connecting attempt after the next socket opens", async () => {
+    LifecycleSocket.instances[0]!.open();
+    await vi.advanceTimersByTimeAsync(0);
+    await expect(transport.listWorkspaces()).resolves.toEqual([{ id: "fixture" }]);
+
+    LifecycleSocket.instances[0]!.fail();
+    await vi.advanceTimersByTimeAsync(1100);
+    expect(LifecycleSocket.instances).toHaveLength(2);
+
+    const original = settleSpy(transport.listWorkspaces());
+    LifecycleSocket.instances[1]!.fail();
+    await vi.advanceTimersByTimeAsync(2100);
+    expect(LifecycleSocket.instances).toHaveLength(3);
+    LifecycleSocket.instances[2]!.open();
+    await vi.advanceTimersByTimeAsync(0);
+
+    await expect(transport.listWorkspaces()).resolves.toEqual([{ id: "fixture" }]);
+    await vi.advanceTimersByTimeAsync(60_000);
+    expect(original.state).not.toBe("pending");
+  });
+
+  it("keeps a waiting RPC alive across consecutive failed reconnect attempts", async () => {
+    LifecycleSocket.instances[0]!.open();
+    await vi.advanceTimersByTimeAsync(0);
+    LifecycleSocket.instances[0]!.fail();
+    await vi.advanceTimersByTimeAsync(1100);
+    expect(LifecycleSocket.instances).toHaveLength(2);
+
+    const waiting = settleSpy(transport.listWorkspaces());
+    LifecycleSocket.instances[1]!.fail();
+    await vi.advanceTimersByTimeAsync(2100);
+    LifecycleSocket.instances[2]!.fail();
+    await vi.advanceTimersByTimeAsync(4100);
+    expect(LifecycleSocket.instances).toHaveLength(4);
+
+    LifecycleSocket.instances[3]!.open();
+    await vi.advanceTimersByTimeAsync(60_000);
+    expect(waiting.state).toBe("resolved");
+  });
+
+  it("rejects an RPC waiting for readiness when the transport is closed", async () => {
+    LifecycleSocket.instances[0]!.open();
+    await vi.advanceTimersByTimeAsync(0);
+    LifecycleSocket.instances[0]!.fail();
+    await vi.advanceTimersByTimeAsync(1100);
+    expect(LifecycleSocket.instances).toHaveLength(2);
+
+    const waiting = settleSpy(transport.listWorkspaces());
+    transport.close();
+    await vi.advanceTimersByTimeAsync(60_000);
+    expect(waiting.state).toBe("rejected");
+  });
+
+  it("rejects RPCs issued after close instead of queueing them forever", async () => {
+    LifecycleSocket.instances[0]!.open();
+    await vi.advanceTimersByTimeAsync(0);
+    transport.close();
+
+    const late = settleSpy(transport.listWorkspaces());
+    await vi.advanceTimersByTimeAsync(60_000);
+    expect(late.state).toBe("rejected");
+  });
+
+  it("settles a binary RPC waiting during a failed connecting attempt after the next socket opens", async () => {
+    LifecycleSocket.instances[0]!.open();
+    await vi.advanceTimersByTimeAsync(0);
+    LifecycleSocket.instances[0]!.fail();
+    await vi.advanceTimersByTimeAsync(1100);
+    expect(LifecycleSocket.instances).toHaveLength(2);
+
+    const waiting = settleSpy(
+      transport.saveClipboardFile(new ArrayBuffer(4), "image/png", "clip.png"),
+    );
+    LifecycleSocket.instances[1]!.fail();
+    await vi.advanceTimersByTimeAsync(2100);
+    LifecycleSocket.instances[2]!.open();
+    await vi.advanceTimersByTimeAsync(60_000);
+    expect(waiting.state).toBe("resolved");
+  });
+
+  it("does not open a new socket or re-arm readiness when close() lands mid-reconnect", async () => {
+    transport.close();
+    let releaseDiscovery: (() => void) | null = null;
+    transport = createWsTransport("ws://fixture.invalid", {
+      discoverServerUrl: () =>
+        new Promise<string>((resolve) => {
+          releaseDiscovery = () => resolve("ws://fixture.invalid");
+        }),
+    });
+
+    // instances[0] belongs to the beforeEach transport closed above.
+    LifecycleSocket.instances[1]!.open();
+    await vi.advanceTimersByTimeAsync(0);
+    LifecycleSocket.instances[1]!.fail();
+    // Reconnect callback fires and suspends inside discoverServerUrl.
+    await vi.advanceTimersByTimeAsync(1100);
+
+    transport.close();
+    releaseDiscovery!();
+    await vi.advanceTimersByTimeAsync(60_000);
+
+    // No post-close connect() may run: no new socket, and `ready` stays
+    // rejected so late RPCs fail fast instead of parking forever.
+    expect(LifecycleSocket.instances).toHaveLength(2);
+    const late = settleSpy(transport.listWorkspaces());
+    await vi.advanceTimersByTimeAsync(0);
+    expect(late.state).toBe("rejected");
+  });
+
+  it("rejects a binary RPC waiting for readiness when the transport is closed", async () => {
+    LifecycleSocket.instances[0]!.open();
+    await vi.advanceTimersByTimeAsync(0);
+    LifecycleSocket.instances[0]!.fail();
+    await vi.advanceTimersByTimeAsync(1100);
+
+    const waiting = settleSpy(
+      transport.saveClipboardFile(new ArrayBuffer(4), "image/png", "clip.png"),
+    );
+    transport.close();
+    await vi.advanceTimersByTimeAsync(60_000);
+    expect(waiting.state).toBe("rejected");
+  });
+});
+

apps/web/src/__tests__/thread-reconnect-refresh.test.ts

@@ -16,6 +16,7 @@ vi.mock("@/transport", async () => ({
 }));
 
 class ReconnectWebSocket {
+  static readonly OPEN = 1;
   onopen: (() => void) | null = null;
   onclose: ((event: { code: number; reason: string }) => void) | null = null;
   onerror: (() => void) | null = null;

apps/web/src/transport/__tests__/agent-commands.test.ts

@@ -4,6 +4,7 @@ import { useSettingsStore } from "@/stores/settingsStore";
 import { createWsTransport } from "../ws-transport";
 
 class MockWebSocket {
+  static readonly OPEN = 1;
   onopen: (() => void) | null = null;
   onclose: ((event: { code: number; reason: string }) => void) | null = null;
   onerror: (() => void) | null = null;

apps/web/src/transport/ws-transport.ts

@@ -340,19 +340,27 @@ export function createWsTransport(
   /** Resolves when the current WebSocket connection is open. */
   let ready: Promise<void>;
   let resolveReady: () => void = () => undefined;
+  let rejectReady: (err: Error) => void = () => undefined;
   let readyPending = false;
 
   function resetReady() {
     // A still-pending promise may have rpc() calls parked on it; replacing it
     // would strand them forever.
     if (readyPending) return;
     readyPending = true;
-    ready = new Promise<void>((resolve) => {
+    ready = new Promise<void>((resolve, reject) => {
       resolveReady = () => {
         readyPending = false;
         resolve();
       };
+      rejectReady = (err) => {
+        readyPending = false;
+        reject(err);
+      };
     });
+    // Rejection is consumed by rpc() awaiters; this swallows it when close()
+    // settles `ready` with no one parked on it.
+    ready.catch(() => {});
   }
 
   function emitTerminalDataFrame(view: Uint8Array): void {
@@ -564,6 +572,9 @@ export function createWsTransport(
           // Discovery failed, retry with current URL
         }
       }
+      // close() cannot cancel a callback already awaiting discovery; bail so a
+      // new socket (and a fresh `ready` promise) is not created after dispose.
+      if (closed) return;
       connect();
     }, delay);
   }
@@ -572,6 +583,13 @@ export function createWsTransport(
   async function rpc<T>(method: string, params: Record<string, unknown>): Promise<T> {
     await ready;
     return new Promise<T>((resolve, reject) => {
+      // `ready` may have resolved on a socket that closed before this
+      // continuation ran; send() on a dead socket is silently dropped and the
+      // request would sit in `pending` forever.
+      if (ws.readyState !== WebSocket.OPEN) {
+        reject(new Error("WebSocket disconnected"));
+        return;
+      }
       const id = `req_${++idCounter}`;
       pending.set(id, {
         resolve: resolve as (v: unknown) => void,
@@ -635,6 +653,10 @@ export function createWsTransport(
   ): Promise<T> {
     await ready;
     return new Promise<T>((resolve, reject) => {
+      if (ws.readyState !== WebSocket.OPEN) {
+        reject(new Error("WebSocket disconnected"));
+        return;
+      }
       const id = `req_${++idCounter}`;
       pending.set(id, {
         resolve: resolve as (v: unknown) => void,
@@ -663,10 +685,16 @@ export function createWsTransport(
         reject(new Error(`Could not connect to server at ${displayUrl}`));
       }, timeoutMs);
 
-      ready.then(() => {
-        clearTimeout(timer);
-        resolve();
-      });
+      ready.then(
+        () => {
+          clearTimeout(timer);
+          resolve();
+        },
+        (err) => {
+          clearTimeout(timer);
+          reject(err);
+        },
+      );
     });
   }
 
@@ -1258,7 +1286,7 @@ export function createWsTransport(
         clearTimeout(reconnectTimer);
         reconnectTimer = null;
       }
-      resolveReady();
+      rejectReady(new Error("Transport closed"));
       rejectPending("Transport closed");
       ws.close();
     },