fix(remote): stop CI teardown reporting itself as a broken connection

src/remote/api-client.test.ts

@@ -0,0 +1,98 @@
+import { afterEach, describe, expect, it } from "vitest";
+import { newWebSocketRpcSession, RpcTarget } from "capnweb";
+import { WebSocketServer } from "ws";
+import type { Remote } from "./remote.ts";
+import { ApiClient } from "./api-client.ts";
+
+/**
+ * Stands in for the Site relay: on each connection it opens a capnweb session
+ * whose main object answers `authenticate` with a `ServerApi`-shaped stub. That
+ * lets `ApiClient.connect` run its real handshake, ping wiring, and dispose path
+ * over a genuine WebSocket transport — the only place capnweb fires
+ * `onRpcBroken` with "shut down by disposing the main stub".
+ */
+class FakeServerApi extends RpcTarget {
+  async ping() {
+    return true;
+  }
+}
+
+class FakeRelay extends RpcTarget {
+  async authenticate(_token: string, _client: unknown, _mode: unknown) {
+    return new FakeServerApi();
+  }
+}
+
+async function startRelay() {
+  const wss = new WebSocketServer({ port: 0 });
+  await new Promise<void>((resolve) => wss.on("listening", () => resolve()));
+  const sockets: import("ws").WebSocket[] = [];
+  const closed: Promise<void>[] = [];
+  wss.on("connection", (socket) => {
+    sockets.push(socket);
+    closed.push(new Promise((resolve) => socket.on("close", () => resolve())));
+    newWebSocketRpcSession(socket as unknown as WebSocket, new FakeRelay());
+  });
+  const { port } = wss.address() as { port: number };
+  return {
+    endpoint: `http://127.0.0.1:${port}`,
+    dropConnection: () => sockets.forEach((s) => s.terminate()),
+    // Resolves once every opened connection has fully torn down, so capnweb has
+    // had its chance to fire onRpcBroken before we assert.
+    waitForTeardown: async () => {
+      await Promise.all(closed);
+      await new Promise((resolve) => setTimeout(resolve, 50));
+    },
+    close: () =>
+      new Promise<void>((resolve) => {
+        wss.close(() => resolve());
+      }),
+  };
+}
+
+describe("ApiClient.connect (integration)", () => {
+  let relay: Awaited<ReturnType<typeof startRelay>> | undefined;
+
+  afterEach(async () => {
+    await relay?.close();
+    relay = undefined;
+  });
+
+  it("does not report a broken connection when the caller disposes", async () => {
+    relay = await startRelay();
+    let reported: unknown = "not-called";
+    const { dispose } = await ApiClient.connect(
+      relay.endpoint,
+      "token",
+      { kind: "ci", branch: "main", sha: "" },
+      {} as unknown as Remote,
+      (err) => {
+        reported = err;
+      },
+    );
+
+    dispose();
+    await relay.waitForTeardown();
+
+    expect(reported).toBe("not-called");
+  });
+
+  it("reports a broken connection when the transport drops mid-run", async () => {
+    relay = await startRelay();
+    let reported: unknown = "not-called";
+    await ApiClient.connect(
+      relay.endpoint,
+      "token",
+      { kind: "ci", branch: "main", sha: "" },
+      {} as unknown as Remote,
+      (err) => {
+        reported = err;
+      },
+    );
+
+    relay.dropConnection();
+    await relay.waitForTeardown();
+
+    expect(reported).toBeInstanceOf(Error);
+  });
+});

src/remote/api-client.ts

@@ -108,6 +108,11 @@ export class ApiClient extends RpcTarget implements ClientApi {
     const dispose = () => {
       if (disposed) return;
       disposed = true;
+      // Mark the connection broken before disposing the stub. Disposing makes
+      // capnweb fire onRpcBroken with "RPC session was shut down by disposing
+      // the main stub"; without this guard triggerBroken would report our own
+      // intentional teardown as a broken connection.
+      broken = true;
       stopPing();
       try { api[Symbol.dispose](); } catch { /* already gone */ }
       try { (unauthenticated as unknown as Disposable)[Symbol.dispose]?.(); } catch { /* already gone */ }