dream: websockets — upgrade + send/receive/close/broadcast + 16 tests

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

lib/dream/conformance.sh

@@ -28,6 +28,7 @@ MODULES=(
   "lib/dream/session.sx"
   "lib/dream/flash.sx"
   "lib/dream/form.sx"
+  "lib/dream/websocket.sx"
 )
 
 # Suites: NAME RUNNER-FN PATH
@@ -38,6 +39,7 @@ SUITES=(
   "session     dream-ss-tests-run!  lib/dream/tests/session.sx"
   "flash       dream-fl-tests-run!  lib/dream/tests/flash.sx"
   "form        dream-fo-tests-run!  lib/dream/tests/form.sx"
+  "websocket   dream-ws-tests-run!  lib/dream/tests/websocket.sx"
 )
 
 TMPFILE=$(mktemp); trap "rm -f $TMPFILE" EXIT

lib/dream/tests/websocket.sx

@@ -0,0 +1,94 @@
+;; lib/dream/tests/websocket.sx — upgrade, send/receive/close, broadcast.
+
+(define dream-ws-pass 0)
+(define dream-ws-fail 0)
+(define dream-ws-fails (list))
+
+(define
+  dream-ws-test
+  (fn
+    (name actual expected)
+    (if
+      (= actual expected)
+      (set! dream-ws-pass (+ dream-ws-pass 1))
+      (begin
+        (set! dream-ws-fail (+ dream-ws-fail 1))
+        (append! dream-ws-fails {:name name :actual actual :expected expected})))))
+
+;; ── upgrade response ───────────────────────────────────────────────
+(define dream-ws-echo (fn (ws) (dream-text "unused")))
+(define
+  dream-ws-up
+  ((dream-websocket dream-ws-echo) (dream-request "GET" "/ws" {} "")))
+(dream-ws-test "upgrade status 101" (dream-status dream-ws-up) 101)
+(dream-ws-test "is a websocket response" (dream-websocket? dream-ws-up) true)
+(dream-ws-test
+  "plain response is not ws"
+  (dream-websocket? (dream-html "x"))
+  false)
+(dream-ws-test
+  "upgrade header"
+  (dream-resp-header dream-ws-up "upgrade")
+  "websocket")
+
+;; ── basic send / receive / close on a mock ─────────────────────────
+(define dream-ws-w1 (dream-mock-ws (list "hi" "there")))
+(dream-ws-test "open initially" (dream-ws-open? dream-ws-w1) true)
+(dream-ws-test "receive first" (dream-receive dream-ws-w1) "hi")
+(dream-ws-test "receive second" (dream-receive dream-ws-w1) "there")
+(dream-ws-test "receive empty -> nil" (dream-receive dream-ws-w1) nil)
+(dream-send dream-ws-w1 "out1")
+(dream-send dream-ws-w1 "out2")
+(dream-ws-test
+  "sent recorded"
+  (dream-ws-sent dream-ws-w1)
+  (list "out1" "out2"))
+(dream-close dream-ws-w1)
+(dream-ws-test "closed flag" (dream-ws-closed? dream-ws-w1) true)
+(dream-ws-test "open? false after close" (dream-ws-open? dream-ws-w1) false)
+
+;; ── echo handler driven over the upgrade response ──────────────────
+(define
+  dream-ws-echo-h
+  (fn
+    (ws)
+    (let
+      ((m (dream-receive ws)))
+      (if
+        (nil? m)
+        (dream-close ws)
+        (begin (dream-send ws (str "echo:" m)) (dream-ws-echo-h ws))))))
+(define
+  dream-ws-echo-up
+  ((dream-websocket dream-ws-echo-h)
+    (dream-request "GET" "/ws" {} "")))
+(define dream-ws-echo-conn (dream-mock-ws (list "a" "b" "c")))
+(dream-ws-run dream-ws-echo-up dream-ws-echo-conn)
+(dream-ws-test
+  "echo all messages"
+  (dream-ws-sent dream-ws-echo-conn)
+  (list "echo:a" "echo:b" "echo:c"))
+(dream-ws-test
+  "echo closes at end"
+  (dream-ws-closed? dream-ws-echo-conn)
+  true)
+
+;; ── broadcast to a room ────────────────────────────────────────────
+(define dream-ws-c1 (dream-mock-ws (list)))
+(define dream-ws-c2 (dream-mock-ws (list)))
+(define dream-ws-c3 (dream-mock-ws (list)))
+(dream-ws-broadcast (list dream-ws-c1 dream-ws-c2 dream-ws-c3) "hello room")
+(dream-ws-test
+  "broadcast c1"
+  (dream-ws-sent dream-ws-c1)
+  (list "hello room"))
+(dream-ws-test
+  "broadcast c2"
+  (dream-ws-sent dream-ws-c2)
+  (list "hello room"))
+(dream-ws-test
+  "broadcast c3"
+  (dream-ws-sent dream-ws-c3)
+  (list "hello room"))
+
+(define dream-ws-tests-run! (fn () {:total (+ dream-ws-pass dream-ws-fail) :passed dream-ws-pass :failed dream-ws-fail :fails dream-ws-fails}))

lib/dream/websocket.sx

@@ -0,0 +1,42 @@
+;; lib/dream/websocket.sx — Dream-on-SX WebSockets.
+;; dream-websocket wraps a (fn (ws) ...) handler into an ordinary handler that
+;; returns a 101 upgrade response carrying the ws handler. The host detects the
+;; upgrade, builds a ws backed by host IO, and runs the handler. The ws carries an
+;; injectable io fn — a mock in-memory ws for tests, (perform op) in production.
+;; Depends on types.sx.
+
+;; ── upgrade response ───────────────────────────────────────────────
+(define dream-websocket (fn (handler) (fn (req) {:websocket handler :body "" :headers {:connection "Upgrade" :upgrade "websocket"} :status 101})))
+
+(define
+  dream-websocket?
+  (fn (resp) (and (dict? resp) (has-key? resp :websocket))))
+(define dream-ws-handler (fn (resp) (get resp :websocket)))
+
+;; ── ws operations (over an injectable io) ──────────────────────────
+(define dream-send (fn (ws msg) ((get ws :io) {:op "ws/send" :msg msg})))
+(define dream-receive (fn (ws) ((get ws :io) {:op "ws/receive"})))
+(define dream-close (fn (ws) ((get ws :io) {:op "ws/close"})))
+(define dream-ws-open? (fn (ws) ((get ws :io) {:op "ws/open?"})))
+(define
+  dream-ws-broadcast
+  (fn (wss msg) (for-each (fn (ws) (dream-send ws msg)) wss)))
+
+;; production io: every op suspends to the host
+(define dream-ws-perform-io (fn (op) (perform op)))
+(define dream-ws-from-io (fn (io) {:io io}))
+
+;; ── in-memory mock ws (tests + demos) ──────────────────────────────
+;; incoming is a list of messages dream-receive will yield in order.
+(define
+  dream-mock-ws
+  (fn
+    (incoming)
+    (let ((inbox incoming) (outbox (list)) (closed false)) {:closed? (fn () closed) :outbox (fn () outbox) :io (fn (op) (cond ((= (get op :op) "ws/send") (begin (set! outbox (concat outbox (list (get op :msg)))) true)) ((= (get op :op) "ws/receive") (if (empty? inbox) nil (let ((m (first inbox))) (begin (set! inbox (rest inbox)) m)))) ((= (get op :op) "ws/close") (begin (set! closed true) true)) ((= (get op :op) "ws/open?") (not closed)) (else nil)))})))
+
+;; test/demo introspection
+(define dream-ws-sent (fn (ws) ((get ws :outbox))))
+(define dream-ws-closed? (fn (ws) ((get ws :closed?))))
+
+;; drive a ws handler (from an upgrade response) against a ws
+(define dream-ws-run (fn (resp ws) ((dream-ws-handler resp) ws)))

plans/dream-on-sx.md

@@ -69,7 +69,7 @@ The five types: `request`, `response`, `handler = request -> response`, `middlew
       - [x] `dream-multipart req` — multipart form data (in-memory, not yet streaming).
       - [x] CSRF middleware: stateless signed tokens, session-scoped.
       - [x] `dream-csrf-tag req` — returns hidden input fragment for SX templates.
-- [ ] **WebSockets** in `lib/dream/websocket.sx`:
+- [x] **WebSockets** in `lib/dream/websocket.sx`:
       - `dream-websocket handler` — upgrades request; handler `(fn (ws) ...)`.
       - `dream-send ws msg`, `dream-receive ws`, `dream-close ws`.
 - [ ] **Static files:** `dream-static root-path` — serves files, ETags, range requests.
@@ -174,6 +174,14 @@ Confirm scope before starting; some of these may be addable as Dream-internal he
   in file content preserved). `dream-multipart-field` / `dream-multipart-file`
   accessors. In-memory, not streaming (noted for future). `\r`/`\n` string escapes
   work in SX literals.
+- **2026-06-07 — WebSockets** (`lib/dream/websocket.sx`, 16 tests). `dream-websocket
+  handler` wraps a `(fn (ws) …)` into an ordinary handler returning a 101 upgrade
+  response carrying the ws handler (`dream-websocket?` / `dream-ws-handler` for the
+  host to detect + dispatch). `dream-send` / `dream-receive` / `dream-close` /
+  `dream-ws-open?` / `dream-ws-broadcast` operate over an injectable io; production io
+  is `(perform op)`, tests use `dream-mock-ws` (in-memory inbox/outbox/closed via the
+  cell pattern) with `dream-ws-sent` / `dream-ws-closed?` introspection and
+  `dream-ws-run` to drive a handler. Echo loop + room broadcast verified.
 
 ## Blockers