From aee6d2ff1a6c973db47f944f14d67d4e198075ae 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 1/9] =?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 1617f317..21e5a76b 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.3",
+  "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 6af4d245..d022e811 100644
--- a/websocket/websocket.test.ts
+++ b/websocket/websocket.test.ts
@@ -20,7 +20,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();
 
@@ -32,7 +32,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 fbbfa372..95b0182c 100644
--- a/websocket/websocket.ts
+++ b/websocket/websocket.ts
@@ -30,7 +30,7 @@ export interface WebSocketResource<T>
   readonly protocol: string;
   readonly readyState: number;
   readonly url: string;
-  send(data: WebSocketData): void;
+  send(data: WebSocketData): Operation<void>;
 }
 
 /**
@@ -158,7 +158,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 1319a81f0209d595acb27cc1cde999ca25271f0a 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 2/9] =?UTF-8?q?=F0=9F=90=9B=20Fix=20strict=20typecheck=20i?=
 =?UTF-8?q?n=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 61f849d18dac8d33038c2c55747b1d3700ac958d 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 3/9] =?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 fb995c29..89ccb691 100644
--- a/pnpm-lock.yaml
+++ b/pnpm-lock.yaml
@@ -420,6 +420,10 @@ importers:
         version: 4.0.2
 
   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 21e5a76b..8c8acea5 100644
--- a/websocket/package.json
+++ b/websocket/package.json
@@ -14,6 +14,9 @@
       "default": "./dist/mod.js"
     }
   },
+  "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 af5c837ac3fc516c21b11b28f96efd310093dbf0 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 4/9] =?UTF-8?q?=F0=9F=8F=B7=EF=B8=8F=20Add=20explicit=20ty?=
 =?UTF-8?q?pes=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 95b0182c..5dce7c62 100644
--- a/websocket/websocket.ts
+++ b/websocket/websocket.ts
@@ -158,7 +158,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 118813ff7bcbbf71e244419353fe29f24d489ab4 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 5/9] =?UTF-8?q?=E2=99=BB=EF=B8=8F=20Drop=20redundant=20per?=
 =?UTF-8?q?-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 9191267fadd284c6f3ad49d6166b56dae37278ac 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 6/9] =?UTF-8?q?=E2=9C=85=20Cover=20close=20code/reason=20p?=
 =?UTF-8?q?ropagation=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 e4921e911923b70b25564231c657e635b139aa43 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 7/9] =?UTF-8?q?=E2=9C=A8=20Bound=20websocket=20teardown=20?=
 =?UTF-8?q?with=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 89ccb691..4b694c2f 100644
--- a/pnpm-lock.yaml
+++ b/pnpm-lock.yaml
@@ -424,6 +424,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 8c8acea5..7436b4a0 100644
--- a/websocket/package.json
+++ b/websocket/package.json
@@ -15,7 +15,8 @@
     }
   },
   "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 d022e811..d860ce86 100644
--- a/websocket/websocket.test.ts
+++ b/websocket/websocket.test.ts
@@ -5,6 +5,7 @@ import {
   type Subscription,
   createQueue,
   resource,
+  scoped,
   suspend,
   useScope,
   withResolvers,
@@ -62,8 +63,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 5dce7c62..3635f97c 100644
--- a/websocket/websocket.ts
+++ b/websocket/websocket.ts
@@ -7,6 +7,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
@@ -166,7 +173,9 @@ export function useWebSocket<T>(
       ]);
     } finally {
       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 d1f246f9d5d982e2be752e061b1c062dc83fbc23 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 8/9] =?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   | 34 ++++++++++++++++++++++++++++------
 3 files changed, 55 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 3635f97c..faf29947 100644
--- a/websocket/websocket.ts
+++ b/websocket/websocket.ts
@@ -23,8 +23,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> {
@@ -38,6 +40,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>;
 }
 
 /**
@@ -140,6 +153,13 @@ export function useWebSocket<T>(
       close(next.value);
     });
 
+    // Close the socket and wait (bounded) for the close handshake. The first
+    // close wins, so whoever calls this first picks the code the peer sees.
+    function* closeSocket(code: number, reason: string): Operation<void> {
+      socket.close(code, reason);
+      yield* timebox(CLOSE_TIMEOUT_MS, () => closed);
+    }
+
     try {
       socket.addEventListener("message", messages.send);
       socket.addEventListener("close", messages.close);
@@ -168,14 +188,16 @@ 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],
         }),
       ]);
     } finally {
-      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);
+      // Default scope-exit close; a no-op if the caller already closed with an
+      // explicit code via `close()`. Bounded so a silent peer can't hang.
+      yield* closeSocket(1000, "released");
       socket.removeEventListener("message", messages.send);
       socket.removeEventListener("close", messages.close);
     }

From 3822ca5b35b2e309f70b116734c808857104aca9 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 9/9] =?UTF-8?q?=E2=9C=A8=20Isolate=20per-connection=20erro?=
 =?UTF-8?q?rs=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();