From 4bb629322233cb828acca2453ada03f7baf7cd07 Mon Sep 17 00:00:00 2001
From: Taras Mankovski <74687+taras@users.noreply.github.com>
Date: Mon, 6 Jul 2026 17:26:27 -0400
Subject: [PATCH 01/16] =?UTF-8?q?=E2=9C=A8=20Add=20WebSocket=20server=20to?=
 =?UTF-8?q?=20@effectionx/websocket?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Add useWebSocketServer(), the server counterpart of useWebSocket(). It
yields a stream of incoming connections, each a full-duplex
WebSocketResource, so client and server share the same handle type.

The underlying server is supplied via a factory, keeping the package free
of any concrete server dependency and platform-agnostic.

BREAKING CHANGE: WebSocketResource.send() is now an Operation, invoked as
`yield* socket.send(...)` on both client and server. Bumped to 3.0.0.
---
 websocket/README.md         |  64 ++++++++++++-
 websocket/mod.ts            |   1 +
 websocket/package.json      |   4 +-
 websocket/server.test.ts    | 183 ++++++++++++++++++++++++++++++++++++
 websocket/server.ts         | 140 +++++++++++++++++++++++++++
 websocket/websocket.test.ts |   4 +-
 websocket/websocket.ts      |   6 +-
 7 files changed, 395 insertions(+), 7 deletions(-)
 create mode 100644 websocket/server.test.ts
 create mode 100644 websocket/server.ts

diff --git a/websocket/README.md b/websocket/README.md
index e4b6863e..37768b27 100644
--- a/websocket/README.md
+++ b/websocket/README.md
@@ -26,7 +26,7 @@ await main(function* () {
   let socket = yield* useWebSocket("ws://websocket.example.org");
 
   // Send messages to the server
-  socket.send("Hello World");
+  yield* socket.send("Hello World");
 
   // Receive messages using a simple iterator
   for (let message of yield* each(socket)) {
@@ -46,6 +46,68 @@ await main(function* () {
 - **Clean Resource Management**: Connections are properly cleaned up when the
   operation completes
 
+## WebSocket Server
+
+`useWebSocketServer()` is the server counterpart of `useWebSocket()`. It yields a
+stream of incoming connections, where **each connection is the same full-duplex
+`WebSocketResource`** produced by the client — you receive messages by iterating
+it and reply with `yield* connection.send()`.
+
+The underlying server is supplied through a factory, so this package never
+imports a concrete server implementation and stays platform-agnostic. On Node
+this is typically the [`ws`](https://github.com/websockets/ws) `WebSocketServer`.
+
+```typescript
+import { each, main, spawn } from "effection";
+import { WebSocketServer } from "ws";
+import {
+  useWebSocketServer,
+  type WebSocketServerLike,
+} from "@effectionx/websocket";
+
+await main(function* () {
+  let server = yield* useWebSocketServer<string>(
+    () => new WebSocketServer({ port: 3000 }) as unknown as WebSocketServerLike,
+  );
+
+  // A stream is consumed sequentially, so spawn a handler per connection to
+  // serve many clients concurrently.
+  for (let connection of yield* each(server)) {
+    yield* spawn(function* () {
+      for (let message of yield* each(connection)) {
+        yield* connection.send(`echo: ${message.data}`);
+        yield* each.next();
+      }
+    });
+    yield* each.next();
+  }
+});
+```
+
+A client — using `useWebSocket()` from the same package — pairs with it directly.
+Because `send` is an `Operation`, invoke it with `yield*` on both sides:
+
+```typescript
+import { each, main } from "effection";
+import { useWebSocket } from "@effectionx/websocket";
+
+await main(function* () {
+  let socket = yield* useWebSocket<string>("ws://localhost:3000");
+
+  yield* socket.send("hello"); // client -> server
+
+  for (let message of yield* each(socket)) {
+    console.log(message.data); // "echo: hello"  (server -> client)
+    yield* each.next();
+  }
+});
+```
+
+Connections are buffered, so none are dropped between the moment the server
+starts listening and the moment you begin iterating. The server — and every live
+connection it produced — is automatically closed when the resource passes out of
+scope.
+
 ## Advanced Usage
 
 ### Custom WebSocket Implementations
diff --git a/websocket/mod.ts b/websocket/mod.ts
index 5adfd3a9..6b8974fe 100644
--- a/websocket/mod.ts
+++ b/websocket/mod.ts
@@ -1 +1,2 @@
 export * from "./websocket.ts";
+export * from "./server.ts";
diff --git a/websocket/package.json b/websocket/package.json
index 77ed750d..198c6676 100644
--- a/websocket/package.json
+++ b/websocket/package.json
@@ -1,7 +1,7 @@
 {
   "name": "@effectionx/websocket",
-  "description": "WebSocket client with stream-based message handling and automatic cleanup",
-  "version": "2.3.4",
+  "description": "WebSocket client and server with stream-based message handling and automatic cleanup",
+  "version": "3.0.0",
   "keywords": ["io", "streams"],
   "type": "module",
   "main": "./dist/mod.js",
diff --git a/websocket/server.test.ts b/websocket/server.test.ts
new file mode 100644
index 00000000..c22efa5a
--- /dev/null
+++ b/websocket/server.test.ts
@@ -0,0 +1,183 @@
+import { createServer } from "node:http";
+import type { AddressInfo } from "node:net";
+import { describe, it } from "@effectionx/vitest";
+import {
+  type Operation,
+  type Subscription,
+  createQueue,
+  resource,
+  spawn,
+  suspend,
+  withResolvers,
+} from "effection";
+import { expect } from "expect";
+import { WebSocketServer } from "ws";
+
+import {
+  type WebSocketServerLike,
+  type WebSocketServerResource,
+  useWebSocketServer,
+} from "./server.ts";
+import { type WebSocketResource, useWebSocket } from "./websocket.ts";
+
+describe("WebSocketServer", () => {
+  it("yields a connection and receives a message from the client", function* () {
+    let { server, port } = yield* useTestServer();
+    let incoming = yield* server;
+
+    let client = yield* connect(port);
+    let connection = (yield* incoming.next()).value;
+
+    let messages = yield* connection;
+    yield* client.send("hello from client");
+
+    let { value } = yield* messages.next();
+    expect(value).toMatchObject({ data: "hello from client" });
+  });
+
+  it("sends a message from a server connection to the client", function* () {
+    let { server, port } = yield* useTestServer();
+    let incoming = yield* server;
+
+    let client = yield* connect(port);
+    let connection = (yield* incoming.next()).value;
+
+    let clientMessages = yield* client;
+    yield* connection.send("hello from server");
+
+    let { value } = yield* clientMessages.next();
+    expect(value).toMatchObject({ data: "hello from server" });
+  });
+
+  it("completes a connection stream when its client disconnects", function* () {
+    let { server, port } = yield* useTestServer();
+    let incoming = yield* server;
+
+    let raw = new WebSocket(`ws://localhost:${port}`);
+    yield* useWebSocket<string>(() => raw);
+    let connection = (yield* incoming.next()).value;
+    let messages = yield* connection;
+
+    raw.close();
+
+    let event = yield* drain(messages);
+    expect(event.type).toEqual("close");
+    expect(event.wasClean).toEqual(true);
+  });
+
+  it("closes live client connections when the server is torn down", function* () {
+    let { httpServer, port } = yield* useHttp();
+    let accepted = createQueue<void, never>();
+
+    let serverTask = yield* spawn(function* () {
+      let server = yield* useWebSocketServer<string>(
+        () =>
+          new WebSocketServer({
+            server: httpServer,
+          }) as unknown as WebSocketServerLike,
+      );
+      let incoming = yield* server;
+      yield* incoming.next();
+      accepted.add();
+      yield* suspend();
+    });
+
+    let client = yield* connect(port);
+    let messages = yield* client;
+
+    // wait until the server has accepted the connection before tearing it down
+    yield* accepted.next();
+    yield* serverTask.halt();
+
+    let event = yield* drain(messages);
+    expect(event.type).toEqual("close");
+    expect(event.wasClean).toEqual(true);
+  });
+
+  it("surfaces each simultaneous client as a distinct connection", function* () {
+    let { server, port } = yield* useTestServer();
+    let incoming = yield* server;
+
+    // connect two clients, then read two buffered connections back out
+    let clientA = yield* connect(port);
+    let clientB = yield* connect(port);
+
+    let first = (yield* incoming.next()).value;
+    let second = (yield* incoming.next()).value;
+
+    expect(first).not.toBe(second);
+
+    // each connection receives only its own client's message, regardless of order
+    let firstMessages = yield* first;
+    let secondMessages = yield* second;
+
+    yield* clientA.send("A");
+    yield* clientB.send("B");
+
+    let received = [
+      (yield* firstMessages.next()).value?.data,
+      (yield* secondMessages.next()).value?.data,
+    ].sort();
+
+    expect(received).toEqual(["A", "B"]);
+  });
+});
+
+interface TestServer {
+  server: WebSocketServerResource<string>;
+  port: number;
+}
+
+function useTestServer(): Operation<TestServer> {
+  return resource(function* (provide) {
+    let { httpServer, port } = yield* useHttp();
+
+    let server = yield* useWebSocketServer<string>(
+      () =>
+        new WebSocketServer({
+          server: httpServer,
+        }) as unknown as WebSocketServerLike,
+    );
+
+    yield* provide({ server, port });
+  });
+}
+
+function useHttp(): Operation<{
+  httpServer: ReturnType<typeof createServer>;
+  port: number;
+}> {
+  return resource(function* (provide) {
+    let httpServer = createServer();
+
+    let listening = withResolvers<void>();
+    httpServer.listen(0, listening.resolve);
+    yield* listening.operation;
+
+    let port = (httpServer.address() as AddressInfo).port;
+
+    try {
+      yield* provide({ httpServer, port });
+    } finally {
+      let closed = withResolvers<void>();
+      httpServer.close(() => closed.resolve());
+      yield* closed.operation;
+    }
+  });
+}
+
+function* connect(port: number): Operation<WebSocketResource<string>> {
+  return yield* useWebSocket<string>(
+    () => new WebSocket(`ws://localhost:${port}`),
+  );
+}
+
+function* drain<T, TClose>(
+  subscription: Subscription<T, TClose>,
+): Operation<TClose> {
+  let next = yield* subscription.next();
+  while (!next.done) {
+    next = yield* subscription.next();
+  }
+  return next.value;
+}
diff --git a/websocket/server.ts b/websocket/server.ts
new file mode 100644
index 00000000..bbbc0f47
--- /dev/null
+++ b/websocket/server.ts
@@ -0,0 +1,140 @@
+import {
+  createQueue,
+  resource,
+  spawn,
+  useScope,
+  withResolvers,
+} from "effection";
+import type { Operation, Stream } from "effection";
+
+import { type WebSocketResource, useWebSocket } from "./websocket.ts";
+
+/**
+ * The minimal structural surface of a
+ * [`ws`](https://github.com/websockets/ws) `WebSocketServer` (or any
+ * compatible server) that {@link useWebSocketServer} needs.
+ *
+ * This is intentionally narrow so that the package never has to import a
+ * concrete server implementation and stays platform-agnostic. Because the
+ * `connection` event of the `ws` library yields its own `WebSocket` type
+ * rather than the DOM `WebSocket`, you may need to cast when passing a real
+ * server, e.g. `new WebSocketServer({ port }) as unknown as WebSocketServerLike`
+ * — mirroring the `ws as unknown as WebSocket` cast used with the client.
+ */
+export interface WebSocketServerLike {
+  on(event: "connection", listener: (socket: WebSocket) => void): void;
+  on(event: "error", listener: (error: Error) => void): void;
+  off(event: "connection", listener: (socket: WebSocket) => void): void;
+  off(event: "error", listener: (error: Error) => void): void;
+  close(callback?: () => void): void;
+}
+
+/**
+ * Handle to a WebSocket server consumed as an Effection {@link Stream}. Each
+ * value in the stream is a {@link WebSocketResource} representing a single
+ * client connection.
+ *
+ * A `WebSocketServerResource` has no explicit close method. The underlying
+ * server — and every live connection it produced — is automatically closed
+ * when the resource passes out of scope.
+ */
+export interface WebSocketServerResource<T>
+  extends Stream<WebSocketResource<T>, never> {}
+
+/**
+ * Create a WebSocket server resource that yields a {@link Stream} of incoming
+ * client connections. Each connection is a {@link WebSocketResource} — the very
+ * same full-duplex handle produced by {@link useWebSocket} on the client — so
+ * you receive messages by iterating it and reply with `yield* connection.send()`.
+ *
+ * The creation of the underlying server is delegated to a factory function,
+ * keeping this package free of any concrete server dependency. On Node this is
+ * typically the [`ws`](https://github.com/websockets/ws) `WebSocketServer`.
+ *
+ * Connections are buffered, so none are dropped between the moment the server
+ * starts listening and the moment you begin iterating. Since a stream is
+ * consumed sequentially, spawn a handler per connection to serve many clients
+ * concurrently:
+ *
+ * ```ts
+ * import { each, main, spawn } from "effection";
+ * import { WebSocketServer } from "ws";
+ * import { useWebSocketServer, type WebSocketServerLike } from "@effectionx/websocket";
+ *
+ * await main(function* () {
+ *   let server = yield* useWebSocketServer<string>(
+ *     () => new WebSocketServer({ port: 3000 }) as unknown as WebSocketServerLike,
+ *   );
+ *
+ *   for (let connection of yield* each(server)) {
+ *     yield* spawn(function* () {
+ *       for (let message of yield* each(connection)) {
+ *         yield* connection.send(`echo: ${message.data}`);
+ *         yield* each.next();
+ *       }
+ *     });
+ *     yield* each.next();
+ *   }
+ * });
+ * ```
+ *
+ * @param create - a function that constructs the underlying server object that
+ * this resource will manage
+ * @returns an operation yielding a {@link WebSocketServerResource}
+ */
+export function useWebSocketServer<T>(
+  create: () => WebSocketServerLike,
+): Operation<WebSocketServerResource<T>> {
+  return resource(function* (provide) {
+    let server = create();
+
+    let connections = createQueue<WebSocketResource<T>, never>();
+
+    // crash the resource scope if the server itself errors, mirroring the
+    // client's `throw yield* once(socket, "error")` behavior
+    let errored = withResolvers<Error>();
+    let onError = (error: Error) => errored.resolve(error);
+    server.on("error", onError);
+    yield* spawn(function* () {
+      throw yield* errored.operation;
+    });
+
+    let scope = yield* useScope();
+
+    // each connection lives as an independent task in this scope: it wraps the
+    // raw socket with the client resource, publishes it, and stays alive until
+    // the socket closes — at which point its scope (and the wrapped resource) is
+    // torn down instead of leaking a task per past connection.
+    let onConnection = (raw: WebSocket) => {
+      scope.run(function* () {
+        let connection = yield* useWebSocket<T>(() => raw);
+        connections.add(connection);
+
+        let subscription = yield* connection;
+        let next = yield* subscription.next();
+        while (!next.done) {
+          next = yield* subscription.next();
+        }
+      });
+    };
+    server.on("connection", onConnection);
+
+    try {
+      // a queue is itself a subscription; expose it as a stream whose
+      // subscription is the shared connection queue
+      yield* provide({
+        *[Symbol.iterator]() {
+          return connections;
+        },
+      });
+    } finally {
+      server.off("connection", onConnection);
+      server.off("error", onError);
+      // stop accepting new connections; the live connection tasks close their
+      // own sockets as this scope tears down. We don't await the close callback
+      // here because it can depend on those sockets closing, which happens as
+      // part of this same teardown.
+      server.close();
+    }
+  });
+}
diff --git a/websocket/websocket.test.ts b/websocket/websocket.test.ts
index 77debcf2..9ce65401 100644
--- a/websocket/websocket.test.ts
+++ b/websocket/websocket.test.ts
@@ -21,7 +21,7 @@ describe("WebSocket", () => {
 
     let subscription = yield* server.socket;
 
-    client.socket.send("hello from client");
+    yield* client.socket.send("hello from client");
 
     let { value } = yield* subscription.next();
 
@@ -33,7 +33,7 @@ describe("WebSocket", () => {
 
     let subscription = yield* client.socket;
 
-    server.socket.send("hello from server");
+    yield* server.socket.send("hello from server");
 
     let { value } = yield* subscription.next();
 
diff --git a/websocket/websocket.ts b/websocket/websocket.ts
index 76220085..1ceefed7 100644
--- a/websocket/websocket.ts
+++ b/websocket/websocket.ts
@@ -31,7 +31,7 @@ export interface WebSocketResource<T>
   readonly protocol: string;
   readonly readyState: number;
   readonly url: string;
-  send(data: WebSocketData): void;
+  send(data: WebSocketData): Operation<void>;
 }
 
 /**
@@ -167,7 +167,9 @@ export function useWebSocket<T>(
         get url() {
           return socket.url;
         },
-        send: (data) => socket.send(data),
+        *send(data) {
+          socket.send(data);
+        },
         [Symbol.iterator]: messages[Symbol.iterator],
       }),
     ]);

From d87695ec772453ff18ac643b5766f308d782637c Mon Sep 17 00:00:00 2001
From: Taras Mankovski <74687+taras@users.noreply.github.com>
Date: Mon, 6 Jul 2026 17:32:57 -0400
Subject: [PATCH 02/16] =?UTF-8?q?=F0=9F=90=9B=20Fix=20strict=20typecheck?=
 =?UTF-8?q?=20in=20server=20test=20message=20assertion?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 websocket/server.test.ts | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/websocket/server.test.ts b/websocket/server.test.ts
index c22efa5a..5dca7fce 100644
--- a/websocket/server.test.ts
+++ b/websocket/server.test.ts
@@ -115,8 +115,8 @@ describe("WebSocketServer", () => {
     yield* clientB.send("B");
 
     let received = [
-      (yield* firstMessages.next()).value?.data,
-      (yield* secondMessages.next()).value?.data,
+      ((yield* firstMessages.next()).value as MessageEvent<string>).data,
+      ((yield* secondMessages.next()).value as MessageEvent<string>).data,
     ].sort();
 
     expect(received).toEqual(["A", "B"]);

From 2aabdf024e294461b42c5392167da9866617555e Mon Sep 17 00:00:00 2001
From: Taras Mankovski <74687+taras@users.noreply.github.com>
Date: Mon, 6 Jul 2026 17:42:38 -0400
Subject: [PATCH 03/16] =?UTF-8?q?=E2=99=BB=EF=B8=8F=20Use=20on/once=20from?=
 =?UTF-8?q?=20@effectionx/node=20for=20server=20events?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 pnpm-lock.yaml          |  4 +++
 websocket/package.json  |  3 ++
 websocket/server.ts     | 66 ++++++++++++++++++-----------------------
 websocket/tsconfig.json |  3 ++
 4 files changed, 39 insertions(+), 37 deletions(-)

diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml
index bf415dd1..d447c924 100644
--- a/pnpm-lock.yaml
+++ b/pnpm-lock.yaml
@@ -419,6 +419,10 @@ importers:
         version: 4.1.0
 
   websocket:
+    dependencies:
+      '@effectionx/node':
+        specifier: workspace:*
+        version: link:../node
     devDependencies:
       '@effectionx/vitest':
         specifier: workspace:*
diff --git a/websocket/package.json b/websocket/package.json
index 198c6676..89be6ae7 100644
--- a/websocket/package.json
+++ b/websocket/package.json
@@ -15,6 +15,9 @@
     }
   },
   "files": ["dist"],
+  "dependencies": {
+    "@effectionx/node": "workspace:*"
+  },
   "peerDependencies": {
     "effection": "^3 || ^4"
   },
diff --git a/websocket/server.ts b/websocket/server.ts
index bbbc0f47..da8e56e3 100644
--- a/websocket/server.ts
+++ b/websocket/server.ts
@@ -1,18 +1,16 @@
-import {
-  createQueue,
-  resource,
-  spawn,
-  useScope,
-  withResolvers,
-} from "effection";
+import { createQueue, each, resource, spawn, useScope } from "effection";
 import type { Operation, Stream } from "effection";
+import { on, once } from "@effectionx/node";
+import type { EventEmitterLike } from "@effectionx/node";
 
 import { type WebSocketResource, useWebSocket } from "./websocket.ts";
 
 /**
  * The minimal structural surface of a
  * [`ws`](https://github.com/websockets/ws) `WebSocketServer` (or any
- * compatible server) that {@link useWebSocketServer} needs.
+ * compatible server) that {@link useWebSocketServer} needs: an
+ * {@link EventEmitterLike} that emits `connection` and `error` events, plus a
+ * `close` method.
  *
  * This is intentionally narrow so that the package never has to import a
  * concrete server implementation and stays platform-agnostic. Because the
@@ -21,11 +19,7 @@ import { type WebSocketResource, useWebSocket } from "./websocket.ts";
  * server, e.g. `new WebSocketServer({ port }) as unknown as WebSocketServerLike`
  * — mirroring the `ws as unknown as WebSocket` cast used with the client.
  */
-export interface WebSocketServerLike {
-  on(event: "connection", listener: (socket: WebSocket) => void): void;
-  on(event: "error", listener: (error: Error) => void): void;
-  off(event: "connection", listener: (socket: WebSocket) => void): void;
-  off(event: "error", listener: (error: Error) => void): void;
+export interface WebSocketServerLike extends EventEmitterLike {
   close(callback?: () => void): void;
 }
 
@@ -90,34 +84,34 @@ export function useWebSocketServer<T>(
 
     let connections = createQueue<WebSocketResource<T>, never>();
 
+    let scope = yield* useScope();
+
     // crash the resource scope if the server itself errors, mirroring the
     // client's `throw yield* once(socket, "error")` behavior
-    let errored = withResolvers<Error>();
-    let onError = (error: Error) => errored.resolve(error);
-    server.on("error", onError);
     yield* spawn(function* () {
-      throw yield* errored.operation;
+      let [error] = yield* once<[Error]>(server, "error");
+      throw error;
     });
 
-    let scope = yield* useScope();
-
-    // each connection lives as an independent task in this scope: it wraps the
-    // raw socket with the client resource, publishes it, and stays alive until
-    // the socket closes — at which point its scope (and the wrapped resource) is
-    // torn down instead of leaking a task per past connection.
-    let onConnection = (raw: WebSocket) => {
-      scope.run(function* () {
-        let connection = yield* useWebSocket<T>(() => raw);
-        connections.add(connection);
+    // each incoming socket lives as an independent task in this scope: it wraps
+    // the raw socket with the client resource, publishes it, and stays alive
+    // until the socket closes — at which point its scope (and the wrapped
+    // resource) is torn down instead of leaking a task per past connection.
+    yield* spawn(function* () {
+      for (let [raw] of yield* each(on<[WebSocket]>(server, "connection"))) {
+        scope.run(function* () {
+          let connection = yield* useWebSocket<T>(() => raw);
+          connections.add(connection);
 
-        let subscription = yield* connection;
-        let next = yield* subscription.next();
-        while (!next.done) {
-          next = yield* subscription.next();
-        }
-      });
-    };
-    server.on("connection", onConnection);
+          let subscription = yield* connection;
+          let next = yield* subscription.next();
+          while (!next.done) {
+            next = yield* subscription.next();
+          }
+        });
+        yield* each.next();
+      }
+    });
 
     try {
       // a queue is itself a subscription; expose it as a stream whose
@@ -128,8 +122,6 @@ export function useWebSocketServer<T>(
         },
       });
     } finally {
-      server.off("connection", onConnection);
-      server.off("error", onError);
       // stop accepting new connections; the live connection tasks close their
       // own sockets as this scope tears down. We don't await the close callback
       // here because it can depend on those sockets closing, which happens as
diff --git a/websocket/tsconfig.json b/websocket/tsconfig.json
index 49b10377..abcae16e 100644
--- a/websocket/tsconfig.json
+++ b/websocket/tsconfig.json
@@ -7,6 +7,9 @@
   "include": ["**/*.ts"],
   "exclude": ["**/*.test.ts", "dist"],
   "references": [
+    {
+      "path": "../node"
+    },
     {
       "path": "../vitest"
     }

From 3a17c86251ee14680715fe23f5258eb3a3e9df0a Mon Sep 17 00:00:00 2001
From: Taras Mankovski <74687+taras@users.noreply.github.com>
Date: Mon, 6 Jul 2026 17:50:23 -0400
Subject: [PATCH 04/16] =?UTF-8?q?=F0=9F=8F=B7=EF=B8=8F=20Add=20explicit=20?=
 =?UTF-8?q?types=20to=20WebSocketResource.send?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 websocket/websocket.ts | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/websocket/websocket.ts b/websocket/websocket.ts
index 1ceefed7..e27a5611 100644
--- a/websocket/websocket.ts
+++ b/websocket/websocket.ts
@@ -167,7 +167,7 @@ export function useWebSocket<T>(
         get url() {
           return socket.url;
         },
-        *send(data) {
+        *send(data: WebSocketData): Operation<void> {
           socket.send(data);
         },
         [Symbol.iterator]: messages[Symbol.iterator],

From a098d4a7353303f44a9e0de178a063aa46e297f3 Mon Sep 17 00:00:00 2001
From: Taras Mankovski <74687+taras@users.noreply.github.com>
Date: Mon, 6 Jul 2026 18:24:58 -0400
Subject: [PATCH 05/16] =?UTF-8?q?=E2=99=BB=EF=B8=8F=20Drop=20redundant=20p?=
 =?UTF-8?q?er-connection=20task=20in=20websocket=20server?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 websocket/server.ts | 23 ++++++-----------------
 1 file changed, 6 insertions(+), 17 deletions(-)

diff --git a/websocket/server.ts b/websocket/server.ts
index da8e56e3..10bb88c1 100644
--- a/websocket/server.ts
+++ b/websocket/server.ts
@@ -1,4 +1,4 @@
-import { createQueue, each, resource, spawn, useScope } from "effection";
+import { createQueue, each, resource, spawn } from "effection";
 import type { Operation, Stream } from "effection";
 import { on, once } from "@effectionx/node";
 import type { EventEmitterLike } from "@effectionx/node";
@@ -84,8 +84,6 @@ export function useWebSocketServer<T>(
 
     let connections = createQueue<WebSocketResource<T>, never>();
 
-    let scope = yield* useScope();
-
     // crash the resource scope if the server itself errors, mirroring the
     // client's `throw yield* once(socket, "error")` behavior
     yield* spawn(function* () {
@@ -93,22 +91,13 @@ export function useWebSocketServer<T>(
       throw error;
     });
 
-    // each incoming socket lives as an independent task in this scope: it wraps
-    // the raw socket with the client resource, publishes it, and stays alive
-    // until the socket closes — at which point its scope (and the wrapped
-    // resource) is torn down instead of leaking a task per past connection.
+    // accept connections: wrap each raw socket with the client resource and
+    // publish it. `useWebSocket` self-terminates when its socket closes, so no
+    // per-connection task or drain loop is needed — the wrapped connections live
+    // concurrently in this accept task's scope.
     yield* spawn(function* () {
       for (let [raw] of yield* each(on<[WebSocket]>(server, "connection"))) {
-        scope.run(function* () {
-          let connection = yield* useWebSocket<T>(() => raw);
-          connections.add(connection);
-
-          let subscription = yield* connection;
-          let next = yield* subscription.next();
-          while (!next.done) {
-            next = yield* subscription.next();
-          }
-        });
+        connections.add(yield* useWebSocket<T>(() => raw));
         yield* each.next();
       }
     });

From 8f86ee13a132c68026368daefa63debe2d2effdc Mon Sep 17 00:00:00 2001
From: Taras Mankovski <74687+taras@users.noreply.github.com>
Date: Mon, 6 Jul 2026 18:48:41 -0400
Subject: [PATCH 06/16] =?UTF-8?q?=E2=9C=85=20Cover=20close=20code/reason?=
 =?UTF-8?q?=20propagation=20and=20accept-queue=20buffering?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 websocket/server.test.ts | 23 ++++++++++++++++++++++-
 1 file changed, 22 insertions(+), 1 deletion(-)

diff --git a/websocket/server.test.ts b/websocket/server.test.ts
index 5dca7fce..54a89bb8 100644
--- a/websocket/server.test.ts
+++ b/websocket/server.test.ts
@@ -58,11 +58,13 @@ describe("WebSocketServer", () => {
     let connection = (yield* incoming.next()).value;
     let messages = yield* connection;
 
-    raw.close();
+    raw.close(4001, "goodbye");
 
     let event = yield* drain(messages);
     expect(event.type).toEqual("close");
     expect(event.wasClean).toEqual(true);
+    expect(event.code).toEqual(4001);
+    expect(event.reason).toEqual("goodbye");
   });
 
   it("closes live client connections when the server is torn down", function* () {
@@ -92,6 +94,25 @@ describe("WebSocketServer", () => {
     let event = yield* drain(messages);
     expect(event.type).toEqual("close");
     expect(event.wasClean).toEqual(true);
+    expect(event.code).toEqual(1000);
+    expect(event.reason).toEqual("released");
+  });
+
+  it("buffers a connection that arrives before it is consumed", function* () {
+    let { server, port } = yield* useTestServer();
+
+    // connect the client before subscribing to the server stream
+    let client = yield* connect(port);
+    let incoming = yield* server;
+
+    // the connection was buffered while nobody was subscribed
+    let connection = (yield* incoming.next()).value;
+
+    let messages = yield* connection;
+    yield* client.send("buffered hello");
+
+    let { value } = yield* messages.next();
+    expect(value).toMatchObject({ data: "buffered hello" });
   });
 
   it("surfaces each simultaneous client as a distinct connection", function* () {

From c418a21ec24681e79719a1675397eec0c59fda70 Mon Sep 17 00:00:00 2001
From: Taras Mankovski <74687+taras@users.noreply.github.com>
Date: Mon, 6 Jul 2026 19:53:12 -0400
Subject: [PATCH 07/16] =?UTF-8?q?=E2=9C=A8=20Bound=20websocket=20teardown?=
 =?UTF-8?q?=20with=20a=20close=20timeout=20(timebox)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Phase 1: replace the unbounded wait for the peer close handshake in
useWebSocket with timebox(), so a silent peer can no longer hang scope
teardown. Adds @effectionx/timebox.
---
 pnpm-lock.yaml              |  3 +++
 websocket/package.json      |  3 ++-
 websocket/tsconfig.json     |  3 +++
 websocket/websocket.test.ts | 41 +++++++++++++++++++++++++++++++++++++
 websocket/websocket.ts      | 11 +++++++++-
 5 files changed, 59 insertions(+), 2 deletions(-)

diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml
index d447c924..27ed5fbe 100644
--- a/pnpm-lock.yaml
+++ b/pnpm-lock.yaml
@@ -423,6 +423,9 @@ importers:
       '@effectionx/node':
         specifier: workspace:*
         version: link:../node
+      '@effectionx/timebox':
+        specifier: workspace:*
+        version: link:../timebox
     devDependencies:
       '@effectionx/vitest':
         specifier: workspace:*
diff --git a/websocket/package.json b/websocket/package.json
index 89be6ae7..4e2f8cb3 100644
--- a/websocket/package.json
+++ b/websocket/package.json
@@ -16,7 +16,8 @@
   },
   "files": ["dist"],
   "dependencies": {
-    "@effectionx/node": "workspace:*"
+    "@effectionx/node": "workspace:*",
+    "@effectionx/timebox": "workspace:*"
   },
   "peerDependencies": {
     "effection": "^3 || ^4"
diff --git a/websocket/tsconfig.json b/websocket/tsconfig.json
index abcae16e..f74cfd60 100644
--- a/websocket/tsconfig.json
+++ b/websocket/tsconfig.json
@@ -10,6 +10,9 @@
     {
       "path": "../node"
     },
+    {
+      "path": "../timebox"
+    },
     {
       "path": "../vitest"
     }
diff --git a/websocket/websocket.test.ts b/websocket/websocket.test.ts
index 9ce65401..a40f50fb 100644
--- a/websocket/websocket.test.ts
+++ b/websocket/websocket.test.ts
@@ -6,6 +6,7 @@ import {
   createQueue,
   ensure,
   resource,
+  scoped,
   suspend,
   useScope,
   withResolvers,
@@ -63,8 +64,48 @@ describe("WebSocket", () => {
     expect(event.type).toEqual("close");
     expect(event.wasClean).toEqual(true);
   });
+
+  it("does not hang teardown when the peer never sends a close frame", function* () {
+    let socket = makeSilentSocket();
+
+    // `scoped` runs the body in a child scope and tears it down when the body
+    // returns — releasing the connection. Even though the socket never emits
+    // "close", the release must complete (bounded by the close timeout) rather
+    // than hang forever.
+    yield* scoped(function* () {
+      yield* useWebSocket(() => socket as unknown as WebSocket);
+    });
+
+    // reaching here means teardown completed; it also attempted the close
+    expect(socket.closeCalls).toEqual(1);
+  });
 });
 
+/**
+ * A minimal WebSocket stand-in that is already OPEN and never emits an "open",
+ * "close", or "error" event — used to prove teardown does not hang on a silent
+ * peer.
+ */
+function makeSilentSocket() {
+  return {
+    readyState: WebSocket.OPEN,
+    binaryType: "blob" as BinaryType,
+    bufferedAmount: 0,
+    extensions: "",
+    protocol: "",
+    url: "ws://silent.test",
+    closeCalls: 0,
+    // record listeners but never fire any event
+    addEventListener() {},
+    removeEventListener() {},
+    send() {},
+    close() {
+      // deliberately never fires a "close" event
+      this.closeCalls += 1;
+    },
+  };
+}
+
 interface TestSocket {
   close(): void;
   socket: WebSocketResource<unknown>;
diff --git a/websocket/websocket.ts b/websocket/websocket.ts
index e27a5611..3350439c 100644
--- a/websocket/websocket.ts
+++ b/websocket/websocket.ts
@@ -8,6 +8,13 @@ import {
   withResolvers,
 } from "effection";
 import type { Operation, Stream } from "effection";
+import { timebox } from "@effectionx/timebox";
+
+/**
+ * How long to wait for the peer's `close` handshake when a socket is released
+ * before giving up and moving on, so a silent peer can never hang teardown.
+ */
+const CLOSE_TIMEOUT_MS = 1000;
 
 /**
  * Handle to a
@@ -138,7 +145,9 @@ export function useWebSocket<T>(
     // `closed`.
     yield* ensure(function* () {
       socket.close(1000, "released");
-      yield* closed;
+      // Bound the wait for the peer's close handshake so a silent peer can't
+      // hang teardown; on timeout we simply stop waiting (no forced terminate).
+      yield* timebox(CLOSE_TIMEOUT_MS, () => closed);
       socket.removeEventListener("message", messages.send);
       socket.removeEventListener("close", messages.close);
     });

From 5edf1e8aca13c13caea0c669d037a799b6d30617 Mon Sep 17 00:00:00 2001
From: Taras Mankovski <74687+taras@users.noreply.github.com>
Date: Mon, 6 Jul 2026 19:56:25 -0400
Subject: [PATCH 08/16] =?UTF-8?q?=E2=9C=A8=20Add=20composable=20close(code?=
 =?UTF-8?q?,=20reason)=20+=20going-away=20server=20shutdown?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Phase 2: WebSocketResource gains a composable close(code?, reason?)
operation; useWebSocketServer composes a 1001 "server shutting down"
close for live connections on teardown. First-close-wins, so it takes
precedence over the scope-exit 1000.
---
 websocket/server.test.ts | 19 +++++++++++++++++--
 websocket/server.ts      | 15 ++++++++++-----
 websocket/websocket.ts   | 33 +++++++++++++++++++++++++++------
 3 files changed, 54 insertions(+), 13 deletions(-)

diff --git a/websocket/server.test.ts b/websocket/server.test.ts
index 54a89bb8..c868f607 100644
--- a/websocket/server.test.ts
+++ b/websocket/server.test.ts
@@ -94,8 +94,23 @@ describe("WebSocketServer", () => {
     let event = yield* drain(messages);
     expect(event.type).toEqual("close");
     expect(event.wasClean).toEqual(true);
-    expect(event.code).toEqual(1000);
-    expect(event.reason).toEqual("released");
+    expect(event.code).toEqual(1001);
+    expect(event.reason).toEqual("server shutting down");
+  });
+
+  it("closes a connection with an explicit code and reason", function* () {
+    let { server, port } = yield* useTestServer();
+    let incoming = yield* server;
+
+    let client = yield* connect(port);
+    let connection = (yield* incoming.next()).value;
+    let clientMessages = yield* client;
+
+    yield* connection.close(4002, "custom");
+
+    let event = yield* drain(clientMessages);
+    expect(event.code).toEqual(4002);
+    expect(event.reason).toEqual("custom");
   });
 
   it("buffers a connection that arrives before it is consumed", function* () {
diff --git a/websocket/server.ts b/websocket/server.ts
index 10bb88c1..c1ab0281 100644
--- a/websocket/server.ts
+++ b/websocket/server.ts
@@ -83,6 +83,7 @@ export function useWebSocketServer<T>(
     let server = create();
 
     let connections = createQueue<WebSocketResource<T>, never>();
+    let live = new Set<WebSocketResource<T>>();
 
     // crash the resource scope if the server itself errors, mirroring the
     // client's `throw yield* once(socket, "error")` behavior
@@ -97,7 +98,9 @@ export function useWebSocketServer<T>(
     // concurrently in this accept task's scope.
     yield* spawn(function* () {
       for (let [raw] of yield* each(on<[WebSocket]>(server, "connection"))) {
-        connections.add(yield* useWebSocket<T>(() => raw));
+        let connection = yield* useWebSocket<T>(() => raw);
+        live.add(connection);
+        connections.add(connection);
         yield* each.next();
       }
     });
@@ -111,10 +114,12 @@ export function useWebSocketServer<T>(
         },
       });
     } finally {
-      // stop accepting new connections; the live connection tasks close their
-      // own sockets as this scope tears down. We don't await the close callback
-      // here because it can depend on those sockets closing, which happens as
-      // part of this same teardown.
+      // Compose a going-away shutdown: close live connections with 1001 before
+      // releasing. The first close wins, so this takes precedence over each
+      // connection's scope-exit close (1000) as the accept task tears down.
+      for (let connection of live) {
+        yield* connection.close(1001, "server shutting down");
+      }
       server.close();
     }
   });
diff --git a/websocket/websocket.ts b/websocket/websocket.ts
index 3350439c..d42b4ef3 100644
--- a/websocket/websocket.ts
+++ b/websocket/websocket.ts
@@ -24,8 +24,10 @@ const CLOSE_TIMEOUT_MS = 1000;
  * itself is a subscribale stream. When the socket is closed, the stream will
  * complete with a [`CloseEvent`](https://developer.mozilla.org/en-US/docs/Web/API/CloseEvent)
  *
- * A WebSocketResource does not have an explicit close method. Rather, the underlying
- * socket will be automatically closed when the resource passes out of scope.
+ * The underlying socket is automatically closed when the resource passes out of
+ * scope (with code `1000` and reason `"released"`). For a different close code
+ * or reason — e.g. a `1001` "going away" on server shutdown — compose an
+ * explicit {@link WebSocketResource.close} before the resource is released.
  */
 export interface WebSocketResource<T>
   extends Stream<MessageEvent<T>, CloseEvent> {
@@ -39,6 +41,17 @@ export interface WebSocketResource<T>
   readonly readyState: number;
   readonly url: string;
   send(data: WebSocketData): Operation<void>;
+  /**
+   * Close the socket with an explicit code and reason, resolving once the close
+   * handshake completes (bounded by an internal timeout so a silent peer cannot
+   * hang). Because the first close wins, calling this before the resource is
+   * released lets you choose the close code the peer observes; the automatic
+   * scope-exit close then becomes a no-op.
+   *
+   * @param code - a valid WebSocket close code (default `1000`)
+   * @param reason - a close reason string (default `"released"`)
+   */
+  close(code?: number, reason?: string): Operation<void>;
 }
 
 /**
@@ -141,13 +154,18 @@ export function useWebSocket<T>(
       close(next.value);
     });
 
+    // The first close wins, so whoever calls this first picks the code the peer
+    // sees. On timeout we stop waiting rather than forcing a terminate.
+    function* closeSocket(code: number, reason: string): Operation<void> {
+      socket.close(code, reason);
+      yield* timebox(CLOSE_TIMEOUT_MS, () => closed);
+    }
+
     // Don't hoist this above the spawns — teardown would hang waiting on
     // `closed`.
     yield* ensure(function* () {
-      socket.close(1000, "released");
-      // Bound the wait for the peer's close handshake so a silent peer can't
-      // hang teardown; on timeout we simply stop waiting (no forced terminate).
-      yield* timebox(CLOSE_TIMEOUT_MS, () => closed);
+      // A no-op if the caller already closed with an explicit code via `close()`.
+      yield* closeSocket(1000, "released");
       socket.removeEventListener("message", messages.send);
       socket.removeEventListener("close", messages.close);
     });
@@ -179,6 +197,9 @@ export function useWebSocket<T>(
         *send(data: WebSocketData): Operation<void> {
           socket.send(data);
         },
+        *close(code = 1000, reason = "released"): Operation<void> {
+          yield* closeSocket(code, reason);
+        },
         [Symbol.iterator]: messages[Symbol.iterator],
       }),
     ]);

From 84c896f0597d20d9f212ab56eb880d7cedb1ec0f Mon Sep 17 00:00:00 2001
From: Taras Mankovski <74687+taras@users.noreply.github.com>
Date: Mon, 6 Jul 2026 20:00:43 -0400
Subject: [PATCH 09/16] =?UTF-8?q?=E2=9C=A8=20Isolate=20per-connection=20er?=
 =?UTF-8?q?rors=20with=20scoped=20+=20errors=20stream?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Phase 3: each accepted connection now runs inside a scoped() error
boundary, so one socket erroring is contained instead of crashing the
server. Failures are surfaced compositionally on a new server.errors
stream rather than a callback.
---
 websocket/server.test.ts | 37 +++++++++++++++++++++++-
 websocket/server.ts      | 61 ++++++++++++++++++++++++++++++++--------
 2 files changed, 86 insertions(+), 12 deletions(-)

diff --git a/websocket/server.test.ts b/websocket/server.test.ts
index c868f607..254303b6 100644
--- a/websocket/server.test.ts
+++ b/websocket/server.test.ts
@@ -11,7 +11,7 @@ import {
   withResolvers,
 } from "effection";
 import { expect } from "expect";
-import { WebSocketServer } from "ws";
+import { WebSocketServer, type WebSocket as WsWebSocket } from "ws";
 
 import {
   type WebSocketServerLike,
@@ -130,6 +130,41 @@ describe("WebSocketServer", () => {
     expect(value).toMatchObject({ data: "buffered hello" });
   });
 
+  it("isolates a connection error so the server keeps serving other clients", function* () {
+    let { httpServer, port } = yield* useHttp();
+
+    // capture the raw accepted sockets so we can force an error on one
+    let wss = new WebSocketServer({ server: httpServer });
+    let rawSockets = createQueue<WsWebSocket, never>();
+    wss.on("connection", (ws) => rawSockets.add(ws));
+
+    let server = yield* useWebSocketServer<string>(
+      () => wss as unknown as WebSocketServerLike,
+    );
+    let incoming = yield* server;
+    let serverErrors = yield* server.errors;
+
+    // accept one client, then make its underlying socket error
+    yield* connect(port);
+    yield* incoming.next();
+    let raw = (yield* rawSockets.next()).value;
+    raw.emit("error", new Error("boom"));
+
+    // the failure is surfaced on the errors stream, not thrown at the server.
+    // A socket failure surfaces as the DOM `error` event, whose message is the
+    // underlying error's message.
+    let { value: error } = yield* serverErrors.next();
+    expect((error as ErrorEvent).message).toContain("boom");
+
+    // and the server survives, serving a fresh client
+    let client = yield* connect(port);
+    let connection = (yield* incoming.next()).value;
+    let messages = yield* connection;
+    yield* client.send("still alive");
+    let { value } = yield* messages.next();
+    expect(value).toMatchObject({ data: "still alive" });
+  });
+
   it("surfaces each simultaneous client as a distinct connection", function* () {
     let { server, port } = yield* useTestServer();
     let incoming = yield* server;
diff --git a/websocket/server.ts b/websocket/server.ts
index c1ab0281..63e22b8f 100644
--- a/websocket/server.ts
+++ b/websocket/server.ts
@@ -1,4 +1,11 @@
-import { createQueue, each, resource, spawn } from "effection";
+import {
+  createQueue,
+  createSignal,
+  each,
+  resource,
+  scoped,
+  spawn,
+} from "effection";
 import type { Operation, Stream } from "effection";
 import { on, once } from "@effectionx/node";
 import type { EventEmitterLike } from "@effectionx/node";
@@ -33,7 +40,17 @@ export interface WebSocketServerLike extends EventEmitterLike {
  * when the resource passes out of scope.
  */
 export interface WebSocketServerResource<T>
-  extends Stream<WebSocketResource<T>, never> {}
+  extends Stream<WebSocketResource<T>, never> {
+  /**
+   * A stream of errors raised by individual connections. A failing connection
+   * is isolated — it does not crash the server — and whatever it threw (for a
+   * socket failure, the DOM `error` event) is published here so you can observe
+   * per-connection failures by consuming this stream (rather than via a
+   * callback). It is lossy: errors emitted while nobody is subscribed are not
+   * buffered.
+   */
+  errors: Stream<unknown, never>;
+}
 
 /**
  * Create a WebSocket server resource that yields a {@link Stream} of incoming
@@ -83,6 +100,7 @@ export function useWebSocketServer<T>(
     let server = create();
 
     let connections = createQueue<WebSocketResource<T>, never>();
+    let errors = createSignal<unknown, never>();
     let live = new Set<WebSocketResource<T>>();
 
     // crash the resource scope if the server itself errors, mirroring the
@@ -92,15 +110,34 @@ export function useWebSocketServer<T>(
       throw error;
     });
 
-    // accept connections: wrap each raw socket with the client resource and
-    // publish it. `useWebSocket` self-terminates when its socket closes, so no
-    // per-connection task or drain loop is needed — the wrapped connections live
-    // concurrently in this accept task's scope.
+    // accept connections. Each is handled in its own task wrapped in `scoped`,
+    // which is a real error boundary (its trap/delimiter contains a crash) — so
+    // a single socket erroring is isolated to that connection and published on
+    // `errors` instead of taking down the server. The connection is held open
+    // until its socket closes.
     yield* spawn(function* () {
       for (let [raw] of yield* each(on<[WebSocket]>(server, "connection"))) {
-        let connection = yield* useWebSocket<T>(() => raw);
-        live.add(connection);
-        connections.add(connection);
+        yield* spawn(function* () {
+          try {
+            yield* scoped(function* () {
+              let connection = yield* useWebSocket<T>(() => raw);
+              live.add(connection);
+              connections.add(connection);
+              try {
+                // stay alive until the socket closes
+                let subscription = yield* connection;
+                let next = yield* subscription.next();
+                while (!next.done) {
+                  next = yield* subscription.next();
+                }
+              } finally {
+                live.delete(connection);
+              }
+            });
+          } catch (error) {
+            errors.send(error);
+          }
+        });
         yield* each.next();
       }
     });
@@ -112,12 +149,14 @@ export function useWebSocketServer<T>(
         *[Symbol.iterator]() {
           return connections;
         },
+        errors,
       });
     } finally {
       // Compose a going-away shutdown: close live connections with 1001 before
       // releasing. The first close wins, so this takes precedence over each
-      // connection's scope-exit close (1000) as the accept task tears down.
-      for (let connection of live) {
+      // connection's scope-exit close (1000) as its task tears down. Snapshot
+      // the set because connections delete themselves from it as they close.
+      for (let connection of [...live]) {
         yield* connection.close(1001, "server shutting down");
       }
       server.close();

From 1b2da1cb74f4bbc4a6f7e111f4838649441450a1 Mon Sep 17 00:00:00 2001
From: Taras Mankovski <74687+taras@users.noreply.github.com>
Date: Fri, 7 Aug 2026 15:39:09 -0400
Subject: [PATCH 10/16] =?UTF-8?q?=E2=9C=A8=20Make=20WebSocket=20close=20ti?=
 =?UTF-8?q?meout=20configurable?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 websocket/README.md         | 14 +++++++++++++-
 websocket/server.ts         | 14 ++++++++++----
 websocket/websocket.test.ts | 17 +++++++++--------
 websocket/websocket.ts      | 35 +++++++++++++++++++++++++----------
 4 files changed, 57 insertions(+), 23 deletions(-)

diff --git a/websocket/README.md b/websocket/README.md
index 37768b27..4c2aff00 100644
--- a/websocket/README.md
+++ b/websocket/README.md
@@ -36,6 +36,16 @@ await main(function* () {
 });
 ```
 
+By default, teardown waits up to one second for the peer's close handshake.
+Configure that deadline when creating the resource if your environment needs a
+different shutdown policy:
+
+```typescript
+let socket = yield* useWebSocket("ws://websocket.example.org", {
+  closeTimeout: 5_000,
+});
+```
+
 ## Features
 
 - **Ready-to-use Connections**: `useWebSocket()` returns only after the
@@ -68,6 +78,7 @@ import {
 await main(function* () {
   let server = yield* useWebSocketServer<string>(
     () => new WebSocketServer({ port: 3000 }) as unknown as WebSocketServerLike,
+    { closeTimeout: 5_000 },
   );
 
   // A stream is consumed sequentially, so spawn a handler per connection to
@@ -106,7 +117,8 @@ await main(function* () {
 Connections are buffered, so none are dropped between the moment the server
 starts listening and the moment you begin iterating. The server — and every live
 connection it produced — is automatically closed when the resource passes out of
-scope.
+scope. The server's second argument configures the close-handshake timeout for
+every accepted connection.
 
 ## Advanced Usage
 
diff --git a/websocket/server.ts b/websocket/server.ts
index 63e22b8f..ac43bc23 100644
--- a/websocket/server.ts
+++ b/websocket/server.ts
@@ -1,3 +1,5 @@
+import { on, once } from "@effectionx/node";
+import type { EventEmitterLike } from "@effectionx/node";
 import {
   createQueue,
   createSignal,
@@ -7,10 +9,12 @@ import {
   spawn,
 } from "effection";
 import type { Operation, Stream } from "effection";
-import { on, once } from "@effectionx/node";
-import type { EventEmitterLike } from "@effectionx/node";
 
-import { type WebSocketResource, useWebSocket } from "./websocket.ts";
+import {
+  type UseWebSocketOptions,
+  type WebSocketResource,
+  useWebSocket,
+} from "./websocket.ts";
 
 /**
  * The minimal structural surface of a
@@ -91,10 +95,12 @@ export interface WebSocketServerResource<T>
  *
  * @param create - a function that constructs the underlying server object that
  * this resource will manage
+ * @param options - options applied to every accepted WebSocket connection
  * @returns an operation yielding a {@link WebSocketServerResource}
  */
 export function useWebSocketServer<T>(
   create: () => WebSocketServerLike,
+  options: UseWebSocketOptions = {},
 ): Operation<WebSocketServerResource<T>> {
   return resource(function* (provide) {
     let server = create();
@@ -120,7 +126,7 @@ export function useWebSocketServer<T>(
         yield* spawn(function* () {
           try {
             yield* scoped(function* () {
-              let connection = yield* useWebSocket<T>(() => raw);
+              let connection = yield* useWebSocket<T>(() => raw, options);
               live.add(connection);
               connections.add(connection);
               try {
diff --git a/websocket/websocket.test.ts b/websocket/websocket.test.ts
index a40f50fb..2df7df5a 100644
--- a/websocket/websocket.test.ts
+++ b/websocket/websocket.test.ts
@@ -1,4 +1,5 @@
 import { createServer } from "node:http";
+import { timebox } from "@effectionx/timebox";
 import { describe, it } from "@effectionx/vitest";
 import {
   type Operation,
@@ -68,15 +69,15 @@ describe("WebSocket", () => {
   it("does not hang teardown when the peer never sends a close frame", function* () {
     let socket = makeSilentSocket();
 
-    // `scoped` runs the body in a child scope and tears it down when the body
-    // returns — releasing the connection. Even though the socket never emits
-    // "close", the release must complete (bounded by the close timeout) rather
-    // than hang forever.
-    yield* scoped(function* () {
-      yield* useWebSocket(() => socket as unknown as WebSocket);
-    });
+    let outcome = yield* timebox(100, () =>
+      scoped(function* () {
+        yield* useWebSocket(() => socket as unknown as WebSocket, {
+          closeTimeout: 0,
+        });
+      }),
+    );
 
-    // reaching here means teardown completed; it also attempted the close
+    expect(outcome.timeout).toEqual(false);
     expect(socket.closeCalls).toEqual(1);
   });
 });
diff --git a/websocket/websocket.ts b/websocket/websocket.ts
index d42b4ef3..da3e2729 100644
--- a/websocket/websocket.ts
+++ b/websocket/websocket.ts
@@ -1,3 +1,4 @@
+import { timebox } from "@effectionx/timebox";
 import {
   createSignal,
   ensure,
@@ -8,13 +9,14 @@ import {
   withResolvers,
 } from "effection";
 import type { Operation, Stream } from "effection";
-import { timebox } from "@effectionx/timebox";
 
-/**
- * How long to wait for the peer's `close` handshake when a socket is released
- * before giving up and moving on, so a silent peer can never hang teardown.
- */
-const CLOSE_TIMEOUT_MS = 1000;
+export interface UseWebSocketOptions {
+  /**
+   * How many milliseconds to wait for the peer's close handshake before
+   * allowing teardown to continue. Defaults to `1000`.
+   */
+  closeTimeout?: number;
+}
 
 /**
  * Handle to a
@@ -79,13 +81,16 @@ export interface WebSocketResource<T>
  *
  * @param url - The URL of the target WebSocket server to connect to. The URL must use one of the following schemes: ws, wss, http, or https, and cannot include a URL fragment. If a relative URL is provided, it is relative to the base URL of the calling script. For more detail, see https://developer.mozilla.org/en-US/docs/Web/API/WebSocket/WebSocket#url
  *
- * @param prototol - A single string or an array of strings representing the sub-protocol(s) that the client would like to use, in order of preference. If it is omitted, an empty array is used by default, i.e. []. For more details, see
+ * @param protocolsOrOptions - A sub-protocol string, or resource options when
+ * no sub-protocol is needed
+ * @param options - Resource options when a sub-protocol is provided
  *
  * @returns an operation yielding a {@link WebSocketResource}
  */
 export function useWebSocket<T>(
   url: string,
-  protocols?: string,
+  protocolsOrOptions?: string | UseWebSocketOptions,
+  options?: UseWebSocketOptions,
 ): Operation<WebSocketResource<T>>;
 
 /**
@@ -116,10 +121,12 @@ export function useWebSocket<T>(
  *
  * ```
  * @param create - a function that will construct the underlying [`WebSocket`](https://developer.mozilla.org/en-US/docs/Web/API/WebSocket) object that this resource wil use
+ * @param options - Resource options
  * @returns an operation yielding a {@link WebSocketResource}
  */
 export function useWebSocket<T>(
   create: () => WebSocket,
+  options?: UseWebSocketOptions,
 ): Operation<WebSocketResource<T>>;
 
 /**
@@ -127,9 +134,17 @@ export function useWebSocket<T>(
  */
 export function useWebSocket<T>(
   url: string | (() => WebSocket),
-  protocols?: string,
+  protocolsOrOptions?: string | UseWebSocketOptions,
+  additionalOptions: UseWebSocketOptions = {},
 ): Operation<WebSocketResource<T>> {
   return resource(function* (provide) {
+    let protocols =
+      typeof protocolsOrOptions === "string" ? protocolsOrOptions : undefined;
+    let options =
+      typeof protocolsOrOptions === "object"
+        ? protocolsOrOptions
+        : additionalOptions;
+    let { closeTimeout = 1000 } = options;
     let socket =
       typeof url === "string" ? new WebSocket(url, protocols) : url();
 
@@ -158,7 +173,7 @@ export function useWebSocket<T>(
     // sees. On timeout we stop waiting rather than forcing a terminate.
     function* closeSocket(code: number, reason: string): Operation<void> {
       socket.close(code, reason);
-      yield* timebox(CLOSE_TIMEOUT_MS, () => closed);
+      yield* timebox(closeTimeout, () => closed);
     }
 
     // Don't hoist this above the spawns — teardown would hang waiting on

From 743252b366b7c99bb9cd0b3671790a08dc8f4621 Mon Sep 17 00:00:00 2001
From: Taras Mankovski <74687+taras@users.noreply.github.com>
Date: Wed, 12 Aug 2026 00:12:18 -0400
Subject: [PATCH 11/16] =?UTF-8?q?=E2=99=BB=EF=B8=8F=20Move=20server=20tear?=
 =?UTF-8?q?down=20into=20ensure()?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The going-away shutdown yielded inside a `finally`, which the Async
Teardown policy forbids: a halt that unwinds through a yielding `finally`
comes back as `iterator.next()` and the frame leaves return-mode, losing
the halt. Register it with `ensure()` instead, placed after the accept
spawns so it still runs while their connections are alive.

Close the live connections concurrently while here — sequential closes
cost one close timeout per silent peer.
---
 websocket/server.ts | 41 +++++++++++++++++++++++------------------
 1 file changed, 23 insertions(+), 18 deletions(-)

diff --git a/websocket/server.ts b/websocket/server.ts
index ac43bc23..5e083490 100644
--- a/websocket/server.ts
+++ b/websocket/server.ts
@@ -1,9 +1,11 @@
 import { on, once } from "@effectionx/node";
 import type { EventEmitterLike } from "@effectionx/node";
 import {
+  all,
   createQueue,
   createSignal,
   each,
+  ensure,
   resource,
   scoped,
   spawn,
@@ -148,24 +150,27 @@ export function useWebSocketServer<T>(
       }
     });
 
-    try {
-      // a queue is itself a subscription; expose it as a stream whose
-      // subscription is the shared connection queue
-      yield* provide({
-        *[Symbol.iterator]() {
-          return connections;
-        },
-        errors,
-      });
-    } finally {
-      // Compose a going-away shutdown: close live connections with 1001 before
-      // releasing. The first close wins, so this takes precedence over each
-      // connection's scope-exit close (1000) as its task tears down. Snapshot
-      // the set because connections delete themselves from it as they close.
-      for (let connection of [...live]) {
-        yield* connection.close(1001, "server shutting down");
-      }
+    // Registered after the spawns so that it runs before they are halted: the
+    // going-away close has to land while each connection task is still alive.
+    yield* ensure(function* () {
+      // The first close wins, so 1001 takes precedence over the 1000 each
+      // connection sends as its own scope exits. Snapshot the set, because
+      // connections remove themselves from it as they close. Closing
+      // concurrently keeps a silent peer from serializing the whole shutdown
+      // into one close timeout apiece.
+      yield* all(
+        [...live].map((connection) =>
+          connection.close(1001, "server shutting down"),
+        ),
+      );
       server.close();
-    }
+    });
+
+    yield* provide({
+      *[Symbol.iterator]() {
+        return connections;
+      },
+      errors,
+    });
   });
 }

From 7d4add554557fa7b872668c074c5c9c57255540f Mon Sep 17 00:00:00 2001
From: Taras Mankovski <74687+taras@users.noreply.github.com>
Date: Wed, 12 Aug 2026 00:14:52 -0400
Subject: [PATCH 12/16] =?UTF-8?q?=E2=99=BB=EF=B8=8F=20Address=20server=20r?=
 =?UTF-8?q?eview=20feedback?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Hand back connections as a Subscription, not a Stream. A stream is
stateless — subscribing is what allocates state and starts the work.
This server does the opposite: it listens and buffers from the moment
the resource is created, and every subscriber drew from the same shared
queue, so two subscribers silently stole each other's connections.
Typing it as a Subscription says what it actually is. The tests already
read it that way, opening with `yield* server` purely to reach a
subscription; that indirection is gone.

Drop the `as unknown as WebSocketServerLike` cast. It was carried over
from the client's `ws as unknown as WebSocket` cast, but a `ws`
WebSocketServer already satisfies the interface structurally — the cast
was never needed and only made the API look worse than it is.

Name the two collections for their jobs: `accepted` is the delivery
buffer that drains as connections are read, `live` is the roster of open
connections closed on shutdown. A connection is in both until read.

Stop describing `scoped` in terms of trap/delimiter, which are private
concepts, and move the test's HTTP-server teardown out of a yielding
`finally` into `ensure()` per the Async Teardown policy.
---
 websocket/README.md      | 37 ++++++++++----------
 websocket/server.test.ts | 55 +++++++++++------------------
 websocket/server.ts      | 75 ++++++++++++++++++++++------------------
 3 files changed, 81 insertions(+), 86 deletions(-)

diff --git a/websocket/README.md b/websocket/README.md
index 4c2aff00..12d5c847 100644
--- a/websocket/README.md
+++ b/websocket/README.md
@@ -58,10 +58,10 @@ let socket = yield* useWebSocket("ws://websocket.example.org", {
 
 ## WebSocket Server
 
-`useWebSocketServer()` is the server counterpart of `useWebSocket()`. It yields a
-stream of incoming connections, where **each connection is the same full-duplex
-`WebSocketResource`** produced by the client — you receive messages by iterating
-it and reply with `yield* connection.send()`.
+`useWebSocketServer()` is the server counterpart of `useWebSocket()`. It hands
+back a subscription of incoming connections, where **each connection is the same
+full-duplex `WebSocketResource`** produced by the client — you receive messages
+by iterating it and reply with `yield* connection.send()`.
 
 The underlying server is supplied through a factory, so this package never
 imports a concrete server implementation and stays platform-agnostic. On Node
@@ -70,27 +70,24 @@ this is typically the [`ws`](https://github.com/websockets/ws) `WebSocketServer`
 ```typescript
 import { each, main, spawn } from "effection";
 import { WebSocketServer } from "ws";
-import {
-  useWebSocketServer,
-  type WebSocketServerLike,
-} from "@effectionx/websocket";
+import { useWebSocketServer } from "@effectionx/websocket";
 
 await main(function* () {
-  let server = yield* useWebSocketServer<string>(
-    () => new WebSocketServer({ port: 3000 }) as unknown as WebSocketServerLike,
+  let connections = yield* useWebSocketServer<string>(
+    () => new WebSocketServer({ port: 3000 }),
     { closeTimeout: 5_000 },
   );
 
-  // A stream is consumed sequentially, so spawn a handler per connection to
+  // Connections are read one at a time, so spawn a handler per connection to
   // serve many clients concurrently.
-  for (let connection of yield* each(server)) {
+  while (true) {
+    let { value: connection } = yield* connections.next();
     yield* spawn(function* () {
       for (let message of yield* each(connection)) {
         yield* connection.send(`echo: ${message.data}`);
         yield* each.next();
       }
     });
-    yield* each.next();
   }
 });
 ```
@@ -114,11 +111,15 @@ await main(function* () {
 });
 ```
 
-Connections are buffered, so none are dropped between the moment the server
-starts listening and the moment you begin iterating. The server — and every live
-connection it produced — is automatically closed when the resource passes out of
-scope. The server's second argument configures the close-handshake timeout for
-every accepted connection.
+Connections are buffered from the moment the resource is created, so none are
+dropped before you start reading. That is why the server is a subscription rather
+than a stream: reading a connection consumes it, and every consumer draws from
+the same buffer instead of getting an independent replay.
+
+The server — and every live connection it produced — is automatically closed when
+the resource passes out of scope, with close code `1001` ("going away"). The
+server's second argument configures the close-handshake timeout for every
+accepted connection.
 
 ## Advanced Usage
 
diff --git a/websocket/server.test.ts b/websocket/server.test.ts
index 254303b6..5db99f74 100644
--- a/websocket/server.test.ts
+++ b/websocket/server.test.ts
@@ -5,6 +5,7 @@ import {
   type Operation,
   type Subscription,
   createQueue,
+  ensure,
   resource,
   spawn,
   suspend,
@@ -13,20 +14,15 @@ import {
 import { expect } from "expect";
 import { WebSocketServer, type WebSocket as WsWebSocket } from "ws";
 
-import {
-  type WebSocketServerLike,
-  type WebSocketServerResource,
-  useWebSocketServer,
-} from "./server.ts";
+import { type WebSocketServerResource, useWebSocketServer } from "./server.ts";
 import { type WebSocketResource, useWebSocket } from "./websocket.ts";
 
 describe("WebSocketServer", () => {
   it("yields a connection and receives a message from the client", function* () {
     let { server, port } = yield* useTestServer();
-    let incoming = yield* server;
 
     let client = yield* connect(port);
-    let connection = (yield* incoming.next()).value;
+    let connection = (yield* server.next()).value;
 
     let messages = yield* connection;
     yield* client.send("hello from client");
@@ -37,10 +33,9 @@ describe("WebSocketServer", () => {
 
   it("sends a message from a server connection to the client", function* () {
     let { server, port } = yield* useTestServer();
-    let incoming = yield* server;
 
     let client = yield* connect(port);
-    let connection = (yield* incoming.next()).value;
+    let connection = (yield* server.next()).value;
 
     let clientMessages = yield* client;
     yield* connection.send("hello from server");
@@ -51,11 +46,10 @@ describe("WebSocketServer", () => {
 
   it("completes a connection stream when its client disconnects", function* () {
     let { server, port } = yield* useTestServer();
-    let incoming = yield* server;
 
     let raw = new WebSocket(`ws://localhost:${port}`);
     yield* useWebSocket<string>(() => raw);
-    let connection = (yield* incoming.next()).value;
+    let connection = (yield* server.next()).value;
     let messages = yield* connection;
 
     raw.close(4001, "goodbye");
@@ -76,10 +70,9 @@ describe("WebSocketServer", () => {
         () =>
           new WebSocketServer({
             server: httpServer,
-          }) as unknown as WebSocketServerLike,
+          }),
       );
-      let incoming = yield* server;
-      yield* incoming.next();
+      yield* server.next();
       accepted.add();
       yield* suspend();
     });
@@ -100,10 +93,9 @@ describe("WebSocketServer", () => {
 
   it("closes a connection with an explicit code and reason", function* () {
     let { server, port } = yield* useTestServer();
-    let incoming = yield* server;
 
     let client = yield* connect(port);
-    let connection = (yield* incoming.next()).value;
+    let connection = (yield* server.next()).value;
     let clientMessages = yield* client;
 
     yield* connection.close(4002, "custom");
@@ -116,12 +108,11 @@ describe("WebSocketServer", () => {
   it("buffers a connection that arrives before it is consumed", function* () {
     let { server, port } = yield* useTestServer();
 
-    // connect the client before subscribing to the server stream
+    // connect the client before reading any connection
     let client = yield* connect(port);
-    let incoming = yield* server;
 
-    // the connection was buffered while nobody was subscribed
-    let connection = (yield* incoming.next()).value;
+    // the connection was buffered before anybody read it
+    let connection = (yield* server.next()).value;
 
     let messages = yield* connection;
     yield* client.send("buffered hello");
@@ -138,15 +129,12 @@ describe("WebSocketServer", () => {
     let rawSockets = createQueue<WsWebSocket, never>();
     wss.on("connection", (ws) => rawSockets.add(ws));
 
-    let server = yield* useWebSocketServer<string>(
-      () => wss as unknown as WebSocketServerLike,
-    );
-    let incoming = yield* server;
+    let server = yield* useWebSocketServer<string>(() => wss);
     let serverErrors = yield* server.errors;
 
     // accept one client, then make its underlying socket error
     yield* connect(port);
-    yield* incoming.next();
+    yield* server.next();
     let raw = (yield* rawSockets.next()).value;
     raw.emit("error", new Error("boom"));
 
@@ -158,7 +146,7 @@ describe("WebSocketServer", () => {
 
     // and the server survives, serving a fresh client
     let client = yield* connect(port);
-    let connection = (yield* incoming.next()).value;
+    let connection = (yield* server.next()).value;
     let messages = yield* connection;
     yield* client.send("still alive");
     let { value } = yield* messages.next();
@@ -167,14 +155,13 @@ describe("WebSocketServer", () => {
 
   it("surfaces each simultaneous client as a distinct connection", function* () {
     let { server, port } = yield* useTestServer();
-    let incoming = yield* server;
 
     // connect two clients, then read two buffered connections back out
     let clientA = yield* connect(port);
     let clientB = yield* connect(port);
 
-    let first = (yield* incoming.next()).value;
-    let second = (yield* incoming.next()).value;
+    let first = (yield* server.next()).value;
+    let second = (yield* server.next()).value;
 
     expect(first).not.toBe(second);
 
@@ -207,7 +194,7 @@ function useTestServer(): Operation<TestServer> {
       () =>
         new WebSocketServer({
           server: httpServer,
-        }) as unknown as WebSocketServerLike,
+        }),
     );
 
     yield* provide({ server, port });
@@ -227,13 +214,13 @@ function useHttp(): Operation<{
 
     let port = (httpServer.address() as AddressInfo).port;
 
-    try {
-      yield* provide({ httpServer, port });
-    } finally {
+    yield* ensure(function* () {
       let closed = withResolvers<void>();
       httpServer.close(() => closed.resolve());
       yield* closed.operation;
-    }
+    });
+
+    yield* provide({ httpServer, port });
   });
 }
 
diff --git a/websocket/server.ts b/websocket/server.ts
index 5e083490..a24b2c0d 100644
--- a/websocket/server.ts
+++ b/websocket/server.ts
@@ -10,7 +10,7 @@ import {
   scoped,
   spawn,
 } from "effection";
-import type { Operation, Stream } from "effection";
+import type { Operation, Stream, Subscription } from "effection";
 
 import {
   type UseWebSocketOptions,
@@ -26,27 +26,32 @@ import {
  * `close` method.
  *
  * This is intentionally narrow so that the package never has to import a
- * concrete server implementation and stays platform-agnostic. Because the
- * `connection` event of the `ws` library yields its own `WebSocket` type
- * rather than the DOM `WebSocket`, you may need to cast when passing a real
- * server, e.g. `new WebSocketServer({ port }) as unknown as WebSocketServerLike`
- * — mirroring the `ws as unknown as WebSocket` cast used with the client.
+ * concrete server implementation and stays platform-agnostic. A `ws`
+ * `WebSocketServer` satisfies it structurally, so it can be passed directly
+ * with no cast.
  */
 export interface WebSocketServerLike extends EventEmitterLike {
   close(callback?: () => void): void;
 }
 
 /**
- * Handle to a WebSocket server consumed as an Effection {@link Stream}. Each
- * value in the stream is a {@link WebSocketResource} representing a single
- * client connection.
+ * Handle to a WebSocket server, consumed as an Effection
+ * {@link Subscription} of incoming client connections. Each value is a
+ * {@link WebSocketResource} representing a single client.
+ *
+ * This is deliberately a subscription rather than a {@link Stream}. A stream is
+ * stateless — subscribing to it is what allocates state — whereas a server
+ * starts listening and buffering connections the moment the resource is
+ * created, and every consumer draws from that one shared buffer. Handing back a
+ * subscription says so in the type: reading a connection consumes it, and there
+ * is no second independent replay of the connections that already arrived.
  *
  * A `WebSocketServerResource` has no explicit close method. The underlying
  * server — and every live connection it produced — is automatically closed
  * when the resource passes out of scope.
  */
 export interface WebSocketServerResource<T>
-  extends Stream<WebSocketResource<T>, never> {
+  extends Subscription<WebSocketResource<T>, never> {
   /**
    * A stream of errors raised by individual connections. A failing connection
    * is isolated — it does not crash the server — and whatever it threw (for a
@@ -59,38 +64,38 @@ export interface WebSocketServerResource<T>
 }
 
 /**
- * Create a WebSocket server resource that yields a {@link Stream} of incoming
- * client connections. Each connection is a {@link WebSocketResource} — the very
- * same full-duplex handle produced by {@link useWebSocket} on the client — so
- * you receive messages by iterating it and reply with `yield* connection.send()`.
+ * Create a WebSocket server resource that hands back a {@link Subscription} of
+ * incoming client connections. Each connection is a {@link WebSocketResource} —
+ * the very same full-duplex handle produced by {@link useWebSocket} on the
+ * client — so you receive messages by iterating it and reply with
+ * `yield* connection.send()`.
  *
  * The creation of the underlying server is delegated to a factory function,
  * keeping this package free of any concrete server dependency. On Node this is
  * typically the [`ws`](https://github.com/websockets/ws) `WebSocketServer`.
  *
- * Connections are buffered, so none are dropped between the moment the server
- * starts listening and the moment you begin iterating. Since a stream is
- * consumed sequentially, spawn a handler per connection to serve many clients
- * concurrently:
+ * Connections are buffered from the moment the resource is created, so none are
+ * dropped before you start reading. Because connections are read one at a time,
+ * spawn a handler per connection to serve many clients concurrently:
  *
  * ```ts
  * import { each, main, spawn } from "effection";
  * import { WebSocketServer } from "ws";
- * import { useWebSocketServer, type WebSocketServerLike } from "@effectionx/websocket";
+ * import { useWebSocketServer } from "@effectionx/websocket";
  *
  * await main(function* () {
- *   let server = yield* useWebSocketServer<string>(
- *     () => new WebSocketServer({ port: 3000 }) as unknown as WebSocketServerLike,
+ *   let connections = yield* useWebSocketServer<string>(
+ *     () => new WebSocketServer({ port: 3000 }),
  *   );
  *
- *   for (let connection of yield* each(server)) {
+ *   while (true) {
+ *     let { value: connection } = yield* connections.next();
  *     yield* spawn(function* () {
  *       for (let message of yield* each(connection)) {
  *         yield* connection.send(`echo: ${message.data}`);
  *         yield* each.next();
  *       }
  *     });
- *     yield* each.next();
  *   }
  * });
  * ```
@@ -107,9 +112,13 @@ export function useWebSocketServer<T>(
   return resource(function* (provide) {
     let server = create();
 
-    let connections = createQueue<WebSocketResource<T>, never>();
-    let errors = createSignal<unknown, never>();
+    // Two collections with different jobs: `accepted` is the delivery buffer,
+    // drained as the consumer reads, while `live` is the roster of still-open
+    // connections used to close them on shutdown. A connection sits in both
+    // until it is read, and stays in `live` long after it has left the buffer.
+    let accepted = createQueue<WebSocketResource<T>, never>();
     let live = new Set<WebSocketResource<T>>();
+    let errors = createSignal<unknown, never>();
 
     // crash the resource scope if the server itself errors, mirroring the
     // client's `throw yield* once(socket, "error")` behavior
@@ -118,11 +127,10 @@ export function useWebSocketServer<T>(
       throw error;
     });
 
-    // accept connections. Each is handled in its own task wrapped in `scoped`,
-    // which is a real error boundary (its trap/delimiter contains a crash) — so
-    // a single socket erroring is isolated to that connection and published on
-    // `errors` instead of taking down the server. The connection is held open
-    // until its socket closes.
+    // `scoped` contains a crash rather than letting it escalate, so one socket
+    // erroring is isolated to its own connection and reported on `errors`
+    // instead of taking down the server. Each connection is held open until its
+    // socket closes.
     yield* spawn(function* () {
       for (let [raw] of yield* each(on<[WebSocket]>(server, "connection"))) {
         yield* spawn(function* () {
@@ -130,7 +138,7 @@ export function useWebSocketServer<T>(
             yield* scoped(function* () {
               let connection = yield* useWebSocket<T>(() => raw, options);
               live.add(connection);
-              connections.add(connection);
+              accepted.add(connection);
               try {
                 // stay alive until the socket closes
                 let subscription = yield* connection;
@@ -166,10 +174,9 @@ export function useWebSocketServer<T>(
       server.close();
     });
 
+    // A queue is already a subscription, so it is the handle itself.
     yield* provide({
-      *[Symbol.iterator]() {
-        return connections;
-      },
+      next: () => accepted.next(),
       errors,
     });
   });

From 9590f22cff86f8cb048cce26c13f00c847c3f084 Mon Sep 17 00:00:00 2001
From: Taras Mankovski <74687+taras@users.noreply.github.com>
Date: Wed, 12 Aug 2026 00:16:02 -0400
Subject: [PATCH 13/16] =?UTF-8?q?=E2=99=BB=EF=B8=8F=20Import=20event=20hel?=
 =?UTF-8?q?pers=20from=20@effectionx/node/events?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Matches how the rest of the repo consumes the package (see process/) and
narrows the import to the entrypoint actually used, rather than pulling
the Node stream adapter in through the barrel.
---
 websocket/server.ts | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/websocket/server.ts b/websocket/server.ts
index a24b2c0d..33f91b57 100644
--- a/websocket/server.ts
+++ b/websocket/server.ts
@@ -1,5 +1,5 @@
-import { on, once } from "@effectionx/node";
-import type { EventEmitterLike } from "@effectionx/node";
+import { on, once } from "@effectionx/node/events";
+import type { EventEmitterLike } from "@effectionx/node/events";
 import {
   all,
   createQueue,

From 67dcf31f73921b41afca98c9808fac83339f8c7a Mon Sep 17 00:00:00 2001
From: Taras Mankovski <74687+taras@users.noreply.github.com>
Date: Wed, 12 Aug 2026 00:39:14 -0400
Subject: [PATCH 14/16] =?UTF-8?q?=F0=9F=90=9B=20Read=20the=20socket=20erro?=
 =?UTF-8?q?r=20through=20ThrownValueError's=20cause?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The isolation test asserted that the value on `errors` is the DOM `error`
event, which held only up to effection 4.0. Since 4.1, `Err()` boxes a
thrown non-Error in a `ThrownValueError` whose message is
`String(value)` — "[object Object]" for an event — and keeps the original
on `cause`. Rebasing onto main pulled effection 4.1.0 in, so the
assertion started failing on CI.

`useWebSocket` throws the raw event deliberately, for parity with the
client, so unwrap at the reading end instead: prefer `cause`, fall back
to the value. The peer range is `^3 || ^4` and the matrix exercises both
ends, so both shapes have to keep working. Document the same on the
`errors` stream, which promised a shape it no longer delivers.
---
 websocket/server.test.ts | 18 ++++++++++++++----
 websocket/server.ts      | 15 ++++++++++-----
 2 files changed, 24 insertions(+), 9 deletions(-)

diff --git a/websocket/server.test.ts b/websocket/server.test.ts
index 5db99f74..32f3ea99 100644
--- a/websocket/server.test.ts
+++ b/websocket/server.test.ts
@@ -138,11 +138,9 @@ describe("WebSocketServer", () => {
     let raw = (yield* rawSockets.next()).value;
     raw.emit("error", new Error("boom"));
 
-    // the failure is surfaced on the errors stream, not thrown at the server.
-    // A socket failure surfaces as the DOM `error` event, whose message is the
-    // underlying error's message.
+    // the failure is surfaced on the errors stream, not thrown at the server
     let { value: error } = yield* serverErrors.next();
-    expect((error as ErrorEvent).message).toContain("boom");
+    expect(socketErrorEvent(error).message).toContain("boom");
 
     // and the server survives, serving a fresh client
     let client = yield* connect(port);
@@ -224,6 +222,18 @@ function useHttp(): Operation<{
   });
 }
 
+/**
+ * A socket failure throws the DOM `error` event, which is not an `Error`.
+ * Effection 4.1 and later box such a value in a `ThrownValueError` that keeps
+ * the original on `cause`, while earlier versions publish the event itself, and
+ * this package supports both.
+ */
+function socketErrorEvent(error: unknown): ErrorEvent {
+  return (
+    error instanceof Error && error.cause ? error.cause : error
+  ) as ErrorEvent;
+}
+
 function* connect(port: number): Operation<WebSocketResource<string>> {
   return yield* useWebSocket<string>(
     () => new WebSocket(`ws://localhost:${port}`),
diff --git a/websocket/server.ts b/websocket/server.ts
index 33f91b57..802bc07e 100644
--- a/websocket/server.ts
+++ b/websocket/server.ts
@@ -54,11 +54,16 @@ export interface WebSocketServerResource<T>
   extends Subscription<WebSocketResource<T>, never> {
   /**
    * A stream of errors raised by individual connections. A failing connection
-   * is isolated — it does not crash the server — and whatever it threw (for a
-   * socket failure, the DOM `error` event) is published here so you can observe
-   * per-connection failures by consuming this stream (rather than via a
-   * callback). It is lossy: errors emitted while nobody is subscribed are not
-   * buffered.
+   * is isolated — it does not crash the server — and whatever it threw is
+   * published here, so per-connection failures are observed by consuming this
+   * stream rather than through a callback. It is lossy: errors emitted while
+   * nobody is subscribed are not buffered.
+   *
+   * A socket failure throws the DOM `error` event, which is not an `Error`.
+   * Effection 4.1 and later box a thrown non-`Error` in a `ThrownValueError`
+   * whose `message` is the useless `String(event)` but whose `cause` is the
+   * event itself; earlier versions publish the event directly. Read `cause`
+   * first and fall back to the value.
    */
   errors: Stream<unknown, never>;
 }

From 24f195735feb63eeb4d8cd1209996e3d0d269dba Mon Sep 17 00:00:00 2001
From: Taras Mankovski <74687+taras@users.noreply.github.com>
Date: Wed, 12 Aug 2026 05:59:12 -0400
Subject: [PATCH 15/16] =?UTF-8?q?=F0=9F=90=9B=20Await=20the=20server=20clo?=
 =?UTF-8?q?se=20callback=20on=20teardown?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

`server.close()` takes a completion callback that was ignored, so
teardown returned before the listening socket was released and a
resource binding the same port next could lose the race with
EADDRINUSE. The test's own `useHttp` helper already awaited its close
callback, so the two paths disagreed.

Also document what a close code may be. `close()` passes the code
straight to the socket, and the legal set depends on the implementation:
the WHATWG API allows only 1000 and 3000-4999 and throws
InvalidAccessError otherwise, while `ws` takes the full RFC 6455 range.
Shutdown's 1001 works on `ws` but would throw on a WHATWG-conformant
socket, so say so on both the option and `WebSocketServerLike` rather
than quietly changing what the peer observes.
---
 websocket/README.md    | 21 +++++++++++++++++++++
 websocket/server.ts    | 11 ++++++++++-
 websocket/websocket.ts |  8 +++++++-
 3 files changed, 38 insertions(+), 2 deletions(-)

diff --git a/websocket/README.md b/websocket/README.md
index 12d5c847..68170550 100644
--- a/websocket/README.md
+++ b/websocket/README.md
@@ -121,6 +121,27 @@ the resource passes out of scope, with close code `1001` ("going away"). The
 server's second argument configures the close-handshake timeout for every
 accepted connection.
 
+### Observing connection failures
+
+The two kinds of failure are reported differently. An error on the server itself
+crashes the resource's scope, reaching your error boundary like any other
+failure. An error on a single connection is isolated so it cannot take the server
+down, and is published on `server.errors` instead — spawn a task to watch that
+stream if you want to see them:
+
+```typescript
+yield* spawn(function* () {
+  for (let error of yield* each(connections.errors)) {
+    // a socket failure throws the DOM `error` event, which is not an `Error`;
+    // effection 4.1+ boxes it and keeps the original on `cause`
+    console.error("connection failed:", (error as Error)?.cause ?? error);
+    yield* each.next();
+  }
+});
+```
+
+`errors` is lossy: failures raised while nobody is subscribed are not buffered.
+
 ## Advanced Usage
 
 ### Custom WebSocket Implementations
diff --git a/websocket/server.ts b/websocket/server.ts
index 802bc07e..8f0ca6f0 100644
--- a/websocket/server.ts
+++ b/websocket/server.ts
@@ -9,6 +9,7 @@ import {
   resource,
   scoped,
   spawn,
+  withResolvers,
 } from "effection";
 import type { Operation, Stream, Subscription } from "effection";
 
@@ -29,6 +30,10 @@ import {
  * concrete server implementation and stays platform-agnostic. A `ws`
  * `WebSocketServer` satisfies it structurally, so it can be passed directly
  * with no cast.
+ *
+ * Shutdown closes accepted sockets with `1001` ("going away"), which the full
+ * RFC 6455 range allows but the WHATWG API does not, so an implementation whose
+ * accepted sockets enforce the browser restriction needs a different code.
  */
 export interface WebSocketServerLike extends EventEmitterLike {
   close(callback?: () => void): void;
@@ -176,7 +181,11 @@ export function useWebSocketServer<T>(
           connection.close(1001, "server shutting down"),
         ),
       );
-      server.close();
+      // Wait for the listening socket to be released, so a resource that binds
+      // the same port after this one cannot lose the race with EADDRINUSE.
+      let closed = withResolvers<void>();
+      server.close(() => closed.resolve());
+      yield* closed.operation;
     });
 
     // A queue is already a subscription, so it is the handle itself.
diff --git a/websocket/websocket.ts b/websocket/websocket.ts
index da3e2729..4a4260c0 100644
--- a/websocket/websocket.ts
+++ b/websocket/websocket.ts
@@ -50,7 +50,13 @@ export interface WebSocketResource<T>
    * released lets you choose the close code the peer observes; the automatic
    * scope-exit close then becomes a no-op.
    *
-   * @param code - a valid WebSocket close code (default `1000`)
+   * The code is handed to the underlying socket unchanged, so which codes are
+   * legal depends on the implementation. The WHATWG API accepts only `1000` and
+   * `3000`–`4999`, throwing `InvalidAccessError` for anything else, while a
+   * `ws` socket accepts the full RFC 6455 range — `1001` ("going away")
+   * included.
+   *
+   * @param code - a close code the underlying socket accepts (default `1000`)
    * @param reason - a close reason string (default `"released"`)
    */
   close(code?: number, reason?: string): Operation<void>;

From d51c18ec4c81e2a4470fc05fb69ebfeca20c11c5 Mon Sep 17 00:00:00 2001
From: Taras Mankovski <74687+taras@users.noreply.github.com>
Date: Wed, 12 Aug 2026 06:35:20 -0400
Subject: [PATCH 16/16] =?UTF-8?q?=E2=9C=85=20Cover=20multi-connection=20an?=
 =?UTF-8?q?d=20silent-peer=20server=20teardown?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Three termination properties the suite could not see, each checked by
mutating the implementation and confirming the test fails:

- Every live connection gets the going-away close, not just the first.
  The existing teardown test used a single client, so closing only
  `[...live][0]` still passed it.
- Teardown does not complete until the server has finished closing.
- Teardown stays bounded when peers never answer the close handshake,
  and the going-away 1001 wins over the scope-exit 1000.

The last two drive a hand-rolled server, since a real peer cannot be made
to withhold a close frame or defer a close callback on demand. That
exposed two things worth writing down: connections emitted in the same
tick as resource creation are missed because the accept loop has not
subscribed yet (harmless for a real server, whose sockets arrive via
I/O), and a fake socket has to leave `readyState` OPEN behind for the
"first close wins" rule to mean anything.

Note the port-rebinding hazard that motivated awaiting the close callback
is not reproducible: Node frees the listening socket during `close()`, so
rebinding succeeds either way. The callback reports when connections have
finished, which is what the test asserts instead.
---
 websocket/server.test.ts | 156 ++++++++++++++++++++++++++++++++++++++-
 1 file changed, 155 insertions(+), 1 deletion(-)

diff --git a/websocket/server.test.ts b/websocket/server.test.ts
index 32f3ea99..6ec44fb9 100644
--- a/websocket/server.test.ts
+++ b/websocket/server.test.ts
@@ -1,5 +1,7 @@
+import { EventEmitter } from "node:events";
 import { createServer } from "node:http";
 import type { AddressInfo } from "node:net";
+import { timebox } from "@effectionx/timebox";
 import { describe, it } from "@effectionx/vitest";
 import {
   type Operation,
@@ -7,6 +9,8 @@ import {
   createQueue,
   ensure,
   resource,
+  scoped,
+  sleep,
   spawn,
   suspend,
   withResolvers,
@@ -14,7 +18,11 @@ import {
 import { expect } from "expect";
 import { WebSocketServer, type WebSocket as WsWebSocket } from "ws";
 
-import { type WebSocketServerResource, useWebSocketServer } from "./server.ts";
+import {
+  type WebSocketServerLike,
+  type WebSocketServerResource,
+  useWebSocketServer,
+} from "./server.ts";
 import { type WebSocketResource, useWebSocket } from "./websocket.ts";
 
 describe("WebSocketServer", () => {
@@ -91,6 +99,98 @@ describe("WebSocketServer", () => {
     expect(event.reason).toEqual("server shutting down");
   });
 
+  it("closes every live connection when the server is torn down", function* () {
+    let { httpServer, port } = yield* useHttp();
+    let accepted = createQueue<void, never>();
+
+    let serverTask = yield* spawn(function* () {
+      let server = yield* useWebSocketServer<string>(
+        () => new WebSocketServer({ server: httpServer }),
+      );
+      yield* server.next();
+      yield* server.next();
+      accepted.add();
+      yield* suspend();
+    });
+
+    let clients = [yield* connect(port), yield* connect(port)];
+    let inboxes = [yield* clients[0], yield* clients[1]];
+
+    // both connections are established, so both belong to the live roster
+    yield* accepted.next();
+    yield* serverTask.halt();
+
+    for (let inbox of inboxes) {
+      let event = yield* drain(inbox);
+      expect(event.code).toEqual(1001);
+      expect(event.reason).toEqual("server shutting down");
+      expect(event.wasClean).toEqual(true);
+    }
+  });
+
+  it("does not complete teardown until the server has finished closing", function* () {
+    // this server never invokes its close callback until released, standing in
+    // for one with connections still winding down
+    let server = makeFakeServer({ deferClose: true });
+    let finished = createQueue<void, never>();
+
+    yield* spawn(function* () {
+      yield* scoped(function* () {
+        yield* useWebSocketServer<string>(() => server);
+      });
+      finished.add();
+    });
+
+    // the scope body is done, so teardown has asked the server to close
+    yield* sleep(0);
+    expect(server.closeCalls).toEqual(1);
+
+    // ...but teardown is still pending, because the callback has not fired
+    let early = yield* timebox(100, () => finished.next());
+    expect(early.timeout).toEqual(true);
+
+    server.releaseClose();
+
+    let late = yield* timebox(1_000, () => finished.next());
+    expect(late.timeout).toEqual(false);
+  });
+
+  it("bounds teardown when peers never answer the close handshake", function* () {
+    let server = makeFakeServer();
+    let sockets = [makeSilentSocket(), makeSilentSocket(), makeSilentSocket()];
+
+    let outcome = yield* timebox(2_000, () =>
+      scoped(function* () {
+        let connections = yield* useWebSocketServer<string>(() => server, {
+          closeTimeout: 10,
+        });
+        // emit once the accept loop is subscribed, which a real server's I/O
+        // guarantees but a hand-driven emitter does not
+        yield* spawn(function* () {
+          yield* sleep(0);
+          for (let socket of sockets) {
+            server.emit("connection", socket);
+          }
+        });
+        for (let _ of sockets) {
+          yield* connections.next();
+        }
+      }),
+    );
+
+    // leaving the scope closed all three without waiting on a reply that never
+    // comes, rather than hanging on the first silent peer
+    expect(outcome.timeout).toEqual(false);
+    expect(sockets.map((socket) => socket.closeCalls)).toEqual([1, 1, 1]);
+    // the going-away close won, so the scope-exit 1000 was a no-op
+    expect(sockets.map((socket) => socket.codes)).toEqual([
+      [1001],
+      [1001],
+      [1001],
+    ]);
+    expect(server.closeCalls).toEqual(1);
+  });
+
   it("closes a connection with an explicit code and reason", function* () {
     let { server, port } = yield* useTestServer();
 
@@ -222,6 +322,60 @@ function useHttp(): Operation<{
   });
 }
 
+/**
+ * A server we drive by hand, so a test can hand {@link useWebSocketServer}
+ * sockets that a real peer would never produce.
+ */
+function makeFakeServer({ deferClose = false }: { deferClose?: boolean } = {}) {
+  let pending: (() => void) | undefined;
+  return Object.assign(new EventEmitter(), {
+    closeCalls: 0,
+    close(callback?: () => void) {
+      this.closeCalls += 1;
+      if (deferClose) {
+        pending = callback;
+      } else {
+        callback?.();
+      }
+    },
+    /** Fire a close callback that `deferClose` withheld. */
+    releaseClose() {
+      pending?.();
+    },
+  }) as unknown as EventEmitter &
+    WebSocketServerLike & { closeCalls: number; releaseClose(): void };
+}
+
+/**
+ * An accepted socket that is already open and never emits `open`, `close`, or
+ * `error` — a peer that takes a close frame and never answers it. `close()`
+ * only takes effect while the socket is open, like a real one, so a second
+ * close is the no-op the "first close wins" rule depends on.
+ */
+function makeSilentSocket() {
+  return {
+    readyState: WebSocket.OPEN as number,
+    binaryType: "blob" as BinaryType,
+    bufferedAmount: 0,
+    extensions: "",
+    protocol: "",
+    url: "ws://silent.test",
+    closeCalls: 0,
+    codes: [] as number[],
+    addEventListener() {},
+    removeEventListener() {},
+    send() {},
+    close(code?: number) {
+      if (this.readyState !== WebSocket.OPEN) {
+        return;
+      }
+      this.readyState = WebSocket.CLOSING;
+      this.closeCalls += 1;
+      this.codes.push(code ?? 1000);
+    },
+  };
+}
+
 /**
  * A socket failure throws the DOM `error` event, which is not an `Error`.
  * Effection 4.1 and later box such a value in a `ThrownValueError` that keeps