Decide a socket handshake before its handler; bound every tool read

A socket op's handshake is decided before its handler runs, so a
handler with effects runs only for a connection it will get: 426 for
another protocol version, 400 without a key, and 403 for a browser on
an origin the op does not admit — its own, plus zip.Origins(...). The
origin policy is the op's, not the answer's, so Socket carries only
Serve, Subprotocols and Compression, and wsx's own refusals are back
to the library's. The in-process CLI refuses a socket op before
invoking it.

An op that answers JSON or bytes keeps JSON's media for its JSON when
Produces names the bytes'. 204, 304 and a redirecting op's 3xx carry
no body in the document; any other declared status, 3xx included,
carries the Out and the doc comment's example, as before. An MCP read
of a Reader is bounded in time as a stream is, and the reader closed.

TestConnMemory counts the goroutines holding connections added, as it
counts the heap they added, so goroutines other tests leave winding
down are not charged to the server.

LLM.md

@@ -73,7 +73,7 @@ Patch tags, one above the highest `v*` on GitHub. Push the commit to main, read
 - The `/docs` page loads Swagger UI from cdn.jsdelivr.net. Embedding swagger-ui-dist 5 would add about 1.74 MB (bundle 1,552,209 bytes, CSS 185,784 bytes, Apache-2.0) to every binary that links zip.
 
 - A manifest op may say `"stream": "sse"`, `"bytes"` or `"socket"` (the answer is written as produced, or the connection upgrades; `ProjectOpenAPI` publishes `text/event-stream` with `x-events`, binary media, or `101` with `x-socket`) and `"raw": true` (the request body is bytes, bound to no type; the Rust front end's). A stream beside an `out` means either. `Check` refuses a socket beside an `out` and a raw body beside an `in`.
-- Every wire is a typed op. A request body that is not JSON is a `zip.Body` field (bytes and Content-Type as sent), a `Parser` In (decodes itself; the document keeps In's schema), or a form: `*zip.File` parts make multipart, and `form:` fields bind only when the op `Consumes` a form media type (otherwise the tags are another binder's and JSON binds as before). A `cookie:` field is an `in: cookie` parameter set by the request's cookie and nothing else: the body, an argument object and the URL never set it, and no schema lists it as a body field. An answer that is not one JSON value is a `Body` (bytes, filename, its own status and headers; a `Reader` streams chunked; no bytes and no Type sends no Content-Type), `Verbatim[T]` (a relay's bytes documented as T; the upstream's 4xx/5xx pass undeclared), `Sse[T]` (`Send(ctx, emit)` on its own goroutine; ctx ends with the stream), `Socket[M]` (GET only; 426 without an upgrade and 501 over ZAP, both before the handler runs; a nil `CheckOrigin` is same-origin, 403 otherwise), `Redirect` or a `CookieCoder`; a type embedding `Body` or `Redirect` is that answer, so one with a `Cookies` method sets them beside it. A JSON answer keeps fiber's `application/json; charset=utf-8` unless the op `Produces` another. Every declared status carries the Out's schema, except one a union's alternative claims with its value-receiver `StatusCode()`. A union is a type whose `OneOf()` result types are its alternatives; `Or[A,B]` is the stock one. Over MCP bytes are base64 (in) and tool content (out), bounded at `toolBound`; a stream is collected as JSON lines up to `toolBound` or `toolWait`, and a socket op is no tool (`_meta.refused`). The call plane writes an `Sse` as the REST door does, refuses a `Socket` before its handler, and carries a union that holds a value as the union. A header field is a CLI flag under its field's name, as before; there are no header flags.
+- Every wire is a typed op. A request body that is not JSON is a `zip.Body` field (bytes and Content-Type as sent), a `Parser` In (decodes itself; the document keeps In's schema), or a form: `*zip.File` parts make multipart, and `form:` fields bind only when the op `Consumes` a form media type (otherwise the tags are another binder's and JSON binds as before). A `cookie:` field is an `in: cookie` parameter set by the request's cookie and nothing else: the body, an argument object and the URL never set it, and no schema lists it as a body field. An answer that is not one JSON value is a `Body` (bytes, filename, its own status and headers; a `Reader` streams chunked; no bytes and no Type sends no Content-Type), `Verbatim[T]` (a relay's bytes documented as T; the upstream's 4xx/5xx pass undeclared), `Sse[T]` (`Send(ctx, emit)` on its own goroutine; ctx ends with the stream), `Socket[M]` (GET only; the handshake is decided before the handler runs: 501 over ZAP, 426 without an upgrade or for another version, 400 without a key, 403 for a browser origin the op does not admit — its own, plus `zip.Origins(...)`, `"*"` for any), `Redirect` or a `CookieCoder`; a type embedding `Body` or `Redirect` is that answer, so one with a `Cookies` method sets them beside it. A JSON answer keeps fiber's `application/json; charset=utf-8` unless the op `Produces` another. Every declared status carries the Out's schema, except one a union's alternative claims with its value-receiver `StatusCode()`. A union is a type whose `OneOf()` result types are its alternatives; `Or[A,B]` is the stock one. Over MCP bytes are base64 (in) and tool content (out); bytes, a Reader and a stream (as JSON lines) are bounded at `toolBound` and a Reader or stream at `toolWait`, and a socket op is no tool (`_meta.refused`). 204, 304 and a redirecting op's 3xx carry no body in the document; any other declared status carries the Out and its example. The call plane writes an `Sse` as the REST door does, refuses a `Socket` before its handler, and carries a union that holds a value as the union. A header field is a CLI flag under its field's name, as before; there are no header flags.
 
 ## Rust
 

answer.go

@@ -16,6 +16,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/valyala/fasthttp"
 	"github.com/zap-proto/fiber/v3"
 
 	"github.com/zap-proto/zip/internal/jsonenc"
@@ -63,10 +64,9 @@ func (Verbatim[T]) documents() reflect.Type { return reflect.TypeOf((*T)(nil)).E
 //	}}, nil
 //
 // Send runs after the handler returns, once the status and headers are out, on
-// a goroutine of its own. Its ctx ends when the stream does — the client has
-// gone or the server closed the connection (found by a write, so Keep finds it
-// in a quiet stream), or a tool call has collected all it will — and a producer
-// that waits on anything selects on it. An error from emit means the stream
+// a goroutine of its own. Its ctx ends when the stream does — a write found the
+// client gone (so in a quiet stream only Keep finds it), or a tool call has
+// collected all it will — and a producer that waits on anything selects on it. An error from emit means the stream
 // has ended, and Send should return it. An error Send returns ends the stream
 // where it stands: the status already went out, so it cannot be a refusal.
 //
@@ -264,50 +264,77 @@ func writeEvents(ctx context.Context, s stream, w *bufio.Writer) error {
 //
 //	return &zip.Socket[Frame]{Serve: func(conn *wsx.Conn) error { … }}, nil
 //
-// A request that does not ask to upgrade is answered 426 with Upgrade:
-// websocket; over ZAP, which carries no upgrade, 501. The document publishes
-// 101 with M as x-events and marks the op x-socket: websocket. A connection is
-// not a call, so the op is no MCP tool and has no call-plane method; the CLI
-// bridges it to stdin and stdout.
+// A request that cannot become a connection is refused before the handler
+// runs, so a handler with effects runs only for a connection it will get: over
+// ZAP, which carries no upgrade, 501; one that does not ask to upgrade, 426 with
+// Upgrade: websocket; one asking for another protocol version, 426 with
+// Sec-WebSocket-Version: 13; one with no Sec-WebSocket-Key, 400; and a browser
+// on an origin the op does not admit ([Origins]), 403 — a socket op may read a
+// cookie, which a browser sends cross-site too. The document publishes 101 with
+// M as x-events and marks the op x-socket: websocket. A connection is not a
+// call, so the op is no MCP tool and has no call-plane method; the CLI bridges
+// it to stdin and stdout.
 type Socket[M any] struct {
 	// Serve runs on the upgraded connection.
 	Serve func(conn *ws.Conn) error `zap:"-"`
-	// Config configures the upgrade (wsx.Config). A nil CheckOrigin admits a
-	// browser only from the address's own origin, since a socket op may read
-	// a cookie, which a browser sends cross-site too.
-	Config ws.Config `zap:"-"`
+	// Subprotocols are the protocols the op speaks, in its order of
+	// preference; the first the client also offers is selected.
+	Subprotocols []string `zap:"-"`
+	// Compression negotiates per-message deflate when the client offers it.
+	Compression bool `zap:"-"`
+}
+
+// Origins names the browser origins, besides the address's own, that may open
+// an op's WebSocket: "https://app.example", or "*" for any. Without it only the
+// address's own origin may; a request carrying no Origin is not a browser's
+// and is always admitted. The handler runs only for an origin admitted.
+func Origins(origins ...string) OpOption {
+	return func(op *registeredOp) { op.Origins = append([]string(nil), origins...) }
 }
 
 // frame is the type one message carries.
 func (Socket[M]) frame() reflect.Type { return reflect.TypeOf((*M)(nil)).Elem() }
 
+// serve upgrades a connection the REST door already found upgradable.
 func (s *Socket[M]) serve(c fiber.Ctx) error {
-	if err := upgradable(c); err != nil {
-		return err
-	}
 	serve := s.Serve
-	err := ws.Upgrade(c.RequestCtx(), s.Config, func(conn *ws.Conn) {
+	cfg := ws.Config{
+		Subprotocols:      s.Subprotocols,
+		EnableCompression: s.Compression,
+		// The origin was admitted before the handler ran ([upgradable]).
+		CheckOrigin: func(*fasthttp.RequestCtx) bool { return true },
+	}
+	return ws.Upgrade(c.RequestCtx(), cfg, func(conn *ws.Conn) {
 		if serve != nil {
 			_ = serve(conn)
 		}
 	})
-	if errors.Is(err, ws.ErrOrigin) {
-		return Errorf(http.StatusForbidden, "%s", ws.ErrOrigin.Error())
-	}
-	return err
 }
 
-// upgradable refuses a request that cannot become a WebSocket: one carried
-// over ZAP, which runs no upgrade (501), and one that does not ask (426, with
-// Upgrade: websocket).
-func upgradable(c fiber.Ctx) error {
-	if carriedByZAP(c.RequestCtx()) {
+// upgradable refuses a request that cannot become a WebSocket the op will
+// serve, before its handler runs: one carried over ZAP, which runs no upgrade
+// (501); one that does not ask (426, with Upgrade: websocket); one asking for
+// another protocol version (426, with the one spoken); one with no key (400);
+// and a browser on an origin the op does not admit (403).
+func upgradable(c fiber.Ctx, origins []string) error {
+	rc := c.RequestCtx()
+	if carriedByZAP(rc) {
 		return Errorf(http.StatusNotImplemented, "%s %s upgrades to a WebSocket, which does not cross ZAP; connect over HTTP", c.Method(), c.Path())
 	}
-	if !ws.Is(c.RequestCtx()) {
+	if !ws.Is(rc) {
 		c.Set("Upgrade", "websocket")
 		return Errorf(http.StatusUpgradeRequired, ws.Refusal)
 	}
+	if !ws.Speaks(rc) {
+		c.Set("Sec-WebSocket-Version", ws.Version)
+		return Errorf(http.StatusUpgradeRequired, "this address speaks WebSocket version %s", ws.Version)
+	}
+	if len(rc.Request.Header.Peek("Sec-WebSocket-Key")) == 0 {
+		return ErrBadRequest("a WebSocket handshake carries a Sec-WebSocket-Key")
+	}
+	if !ws.Admits(rc, origins) {
+		return Errorf(http.StatusForbidden, "%s may not open a connection here", rc.Request.Header.Peek("Origin"))
+	}
 	return nil
 }
 
@@ -614,8 +641,8 @@ func writeAnswer(c fiber.Ctx, op *registeredOp, out any) error {
 		}
 		c.Set(fiber.HeaderContentType, "text/event-stream")
 		// The stream outlives the handler, so its ctx is its own: it ends when a
-		// write finds the client gone — a shutdown closes the connection too —
-		// or when the producer is done.
+		// write finds the client gone, or when the producer is done. A quiet
+		// stream finds a client that left only through Keep.
 		return c.SendStreamWriter(func(w *bufio.Writer) {
 			defer func() { _ = recover() }()
 			_ = writeEvents(context.Background(), x, w)
@@ -644,9 +671,11 @@ func writeAnswer(c fiber.Ctx, op *registeredOp, out any) error {
 		return c.Send(b.Bytes)
 	}
 	// The framework's own JSON media (with its charset) unless the op names
-	// another: an op that declares nothing answers as it always has.
-	if len(op.Produces) > 0 {
-		return c.JSON(out, op.Produces[0])
+	// another for its JSON: an op that declares nothing answers as it always
+	// has, and one whose Produces names the media of its bytes keeps JSON for
+	// its JSON.
+	if media := produces(op, mimeJSON); media != mimeJSON {
+		return c.JSON(out, media)
 	}
 	return c.JSON(out)
 }
@@ -753,14 +782,59 @@ func collect(ctx context.Context, out any) (data []byte, media string, truncated
 		}
 		return b.Bytes, b.Type, false, nil
 	}
-	if c, ok := b.Reader.(io.Closer); ok {
+	ctx, cancel := context.WithTimeout(ctx, toolWait)
+	defer cancel()
+	data, truncated, err = readBounded(ctx, b.Reader)
+	return data, b.Type, truncated, err
+}
+
+// readBounded reads r to its end, toolBound bytes or ctx's end, whichever
+// comes first, and closes it when it can. The read runs beside the wait, so a
+// reader that never yields cannot hold the call; closing it is what ends a read
+// left blocked.
+func readBounded(ctx context.Context, r io.Reader) (data []byte, truncated bool, err error) {
+	c, closes := r.(io.Closer)
+	if closes {
 		defer func() { _ = c.Close() }()
 	}
-	data, err = io.ReadAll(io.LimitReader(b.Reader, toolBound+1))
-	if len(data) > toolBound {
-		return data[:toolBound], b.Type, true, err
+	chunks := make(chan []byte)
+	ended := make(chan error, 1)
+	go func() {
+		buf := make([]byte, 32<<10)
+		for {
+			n, rerr := r.Read(buf)
+			if n > 0 {
+				select {
+				case chunks <- append([]byte(nil), buf[:n]...):
+				case <-ctx.Done():
+					return
+				}
+			}
+			if rerr != nil {
+				if rerr == io.EOF {
+					rerr = nil
+				}
+				ended <- rerr
+				return
+			}
+		}
+	}()
+	for {
+		select {
+		case chunk := <-chunks:
+			if len(data)+len(chunk) > toolBound {
+				return append(data, chunk[:toolBound-len(data)]...), true, nil
+			}
+			data = append(data, chunk...)
+		case err := <-ended:
+			return data, false, err
+		case <-ctx.Done():
+			if errors.Is(ctx.Err(), context.DeadlineExceeded) {
+				return data, true, nil
+			}
+			return data, false, ctx.Err()
+		}
 	}
-	return data, b.Type, false, err
 }
 
 // textual reports whether bytes of a media type read as text.

cli.go

@@ -395,6 +395,9 @@ func LocalInvoke(ctx context.Context, c Command, path map[string]string, body []
 	if c.op == nil || c.op.invoke == nil {
 		return nil, fmt.Errorf("%s %s is not registered in this process — give the CLI a Remote invoker", c.Service, c.Name)
 	}
+	if c.op.ans.stream == streamSocket {
+		return nil, fmt.Errorf("%s %s: %s; reach it over HTTP", c.Service, c.Name, NotACall)
+	}
 	rest, cookie, err := c.split(body)
 	if err != nil {
 		return nil, err

conn_memory_test.go

@@ -75,6 +75,10 @@ func TestConnMemory(t *testing.T) {
 	var baseline, peak runtime.MemStats
 	runtime.GC()
 	runtime.ReadMemStats(&baseline)
+	// The goroutines a connection costs are the ones holding them added, as the
+	// heap a connection costs is the heap they added: what earlier tests in the
+	// process left winding down is not this server's.
+	before := runtime.NumGoroutine()
 
 	req := []byte("GET /hold HTTP/1.1\r\nHost: x\r\n\r\n")
 	conns := make([]net.Conn, 0, n)
@@ -124,7 +128,7 @@ func TestConnMemory(t *testing.T) {
 	delta := int64(peak.HeapAlloc) - int64(baseline.HeapAlloc)
 	perConn := float64(delta) / float64(n)
 	totalGoroutines := runtime.NumGoroutine()
-	goroutinesPerConn := float64(totalGoroutines) / float64(n)
+	goroutinesPerConn := float64(totalGoroutines-before) / float64(n)
 
 	fmt.Printf("\n=== Per-connection memory profile (zip) ===\n")
 	fmt.Printf("conns held       : %d\n", n)

internal/ws/ws.go

@@ -1,10 +1,8 @@
 // Package ws is the one WebSocket upgrade zip performs. wsx.Upgrade (an untyped
-// route) and a typed op answering zip.Socket both reach it, so a connection
-// is accepted, refused and configured the same way whichever door it came in.
+// route) and a typed op answering zip.Socket both reach it.
 package ws
 
 import (
-	"errors"
 	"net/url"
 	"strings"
 
@@ -27,47 +25,55 @@ type Config struct {
 	// EnableCompression negotiates per-message deflate when the client offers it.
 	EnableCompression bool
 	// CheckOrigin decides whether a request's Origin may open a connection.
-	// Nil admits a request with no Origin, or one whose Origin is the
-	// request's own host: a browser on another site is refused.
+	// Nil is the library's: an Origin, when there is one, must name the
+	// request's own host.
 	CheckOrigin func(ctx *fasthttp.RequestCtx) bool
 }
 
 // Is reports whether the request asks to become a WebSocket.
 func Is(rc *fasthttp.RequestCtx) bool { return websocket.FastHTTPIsWebSocketUpgrade(rc) }
 
-// ErrOrigin is what Upgrade answers when the request's Origin may not open a
-// connection, before anything is written, so the caller refuses it with a
-// status of its own.
-var ErrOrigin = errors.New("this origin may not open a connection here")
-
 // Upgrade answers 101 and runs fn on the connection once the handler that
 // called it returns. The caller has already checked [Is].
 func Upgrade(rc *fasthttp.RequestCtx, cfg Config, fn func(*Conn)) error {
-	check := cfg.CheckOrigin
-	if check == nil {
-		check = sameOrigin
-	}
-	if !check(rc) {
-		return ErrOrigin
-	}
 	up := &websocket.FastHTTPUpgrader{
 		ReadBufferSize:    cfg.ReadBufferSize,
 		WriteBufferSize:   cfg.WriteBufferSize,
 		Subprotocols:      cfg.Subprotocols,
 		EnableCompression: cfg.EnableCompression,
-		CheckOrigin:       func(*fasthttp.RequestCtx) bool { return true }, // asked above
+		CheckOrigin:       cfg.CheckOrigin,
 	}
 	return up.Upgrade(rc, fn)
 }
 
-// sameOrigin admits a request with no Origin — not a browser — and one whose
-// Origin names the host the request was sent to.
-func sameOrigin(rc *fasthttp.RequestCtx) bool {
-	origin := rc.Request.Header.Peek("Origin")
-	if len(origin) == 0 {
+// Version is the protocol version a handshake must ask for (RFC 6455).
+const Version = "13"
+
+// Speaks reports whether a handshake's Sec-WebSocket-Version list names
+// [Version], read as the library reads it: comma-separated, case aside.
+func Speaks(rc *fasthttp.RequestCtx) bool {
+	for _, v := range strings.Split(string(rc.Request.Header.Peek("Sec-WebSocket-Version")), ",") {
+		if strings.EqualFold(strings.TrimSpace(v), Version) {
 			return true
 		}
-	u, err := url.Parse(string(origin))
+	}
+	return false
+}
+
+// Admits reports whether a request's Origin may open a connection: a request
+// with no Origin (not a browser), one whose Origin names the request's own host,
+// and one whose Origin is among origins, "*" being any.
+func Admits(rc *fasthttp.RequestCtx, origins []string) bool {
+	origin := string(rc.Request.Header.Peek("Origin"))
+	if origin == "" {
+		return true
+	}
+	for _, o := range origins {
+		if o == "*" || strings.EqualFold(o, origin) {
+			return true
+		}
+	}
+	u, err := url.Parse(origin)
 	return err == nil && strings.EqualFold(u.Host, string(rc.Host()))
 }
 

kinds_wire_test.go

@@ -108,11 +108,13 @@ func TestKinds_APanickingReaderEndsOnlyItsAnswer(t *testing.T) {
 	})
 	app.Get("/v1/good", func(context.Context, *struct{}) (*mkOut, error) { return &mkOut{ID: "ok"}, nil })
 	addr := serveHTTP(t, app)
-	if resp, err := http.Get("http://" + addr + "/v1/bad"); err == nil {
+	// No keep-alive: a connection left idle would outlive the test.
+	client := &http.Client{Transport: &http.Transport{DisableKeepAlives: true}}
+	if resp, err := client.Get("http://" + addr + "/v1/bad"); err == nil {
 		_, _ = io.Copy(io.Discard, resp.Body)
 		_ = resp.Body.Close()
 	}
-	resp, err := http.Get("http://" + addr + "/v1/good")
+	resp, err := client.Get("http://" + addr + "/v1/good")
 	if err != nil {
 		t.Fatalf("the process did not survive: %v", err)
 	}
@@ -151,15 +153,19 @@ func TestKinds_ToolBytesAreBoundedAndTyped(t *testing.T) {
 	}
 }
 
-// A WebSocket op refuses a request that cannot upgrade before its handler
-// runs, and a browser on another site cannot open it.
+// A WebSocket op refuses a request that cannot become its connection before
+// its handler runs — no upgrade, another version, no key, a browser on an
+// origin it does not admit, the in-process CLI — and admits the origins it
+// names.
 func TestKinds_ASocketRefusesBeforeItsHandler(t *testing.T) {
 	var ran atomic.Int32
-	app := zip.New(zip.Config{AppName: "ws", DisableStartupMessage: true})
-	app.Get("/v1/live", func(context.Context, *struct{}) (*zip.Socket[line], error) {
+	live := func(context.Context, *struct{}) (*zip.Socket[line], error) {
 		ran.Add(1)
 		return &zip.Socket[line]{Serve: func(*websocket.Conn) error { return nil }}, nil
-	})
+	}
+	app := zip.New(zip.Config{AppName: "ws", DisableStartupMessage: true})
+	app.Get("/v1/live", live)
+	app.Get("/v1/open", live, zip.Origins("https://app.example"))
 	resp, err := app.Test(httptest.NewRequest("GET", "/v1/live", nil))
 	if err != nil {
 		t.Fatal(err)
@@ -167,20 +173,56 @@ func TestKinds_ASocketRefusesBeforeItsHandler(t *testing.T) {
 	if resp.StatusCode != 426 || ran.Load() != 0 {
 		t.Errorf("a plain GET answered %d and ran the handler %d times", resp.StatusCode, ran.Load())
 	}
+	for _, c := range []struct {
+		name   string
+		header map[string]string
+		code   int
+	}{
+		{"version 8", map[string]string{"Sec-WebSocket-Version": "8"}, 426},
+		{"versions 8 and 7", map[string]string{"Sec-WebSocket-Version": "8, 7"}, 426},
+		{"no key", map[string]string{"Sec-WebSocket-Key": ""}, 400},
+		{"another site", map[string]string{"Origin": "https://elsewhere.example"}, 403},
+		{"opaque origin", map[string]string{"Origin": "null"}, 403},
+	} {
+		req := httptest.NewRequest("GET", "/v1/live", nil)
+		req.Header.Set("Connection", "Upgrade")
+		req.Header.Set("Upgrade", "websocket")
+		req.Header.Set("Sec-WebSocket-Version", "13")
+		req.Header.Set("Sec-WebSocket-Key", "dGhlIHNhbXBsZSBub25jZQ==")
+		for k, v := range c.header {
+			req.Header.Set(k, v)
+		}
+		resp, err := app.Test(req)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if resp.StatusCode != c.code || ran.Load() != 0 {
+			t.Errorf("%s answered %d, want %d; the handler ran %d times", c.name, resp.StatusCode, c.code, ran.Load())
+		}
+	}
+	if err := app.CLI().Run(context.Background(), argv(t, app.Commands(), "GET", "/v1/live")); err == nil || ran.Load() != 0 {
+		t.Errorf("the in-process CLI answered %v and ran the handler %d times", err, ran.Load())
+	}
+
 	addr := serveHTTP(t, app)
 	d := websocket.Dialer{HandshakeTimeout: 5 * time.Second}
-	h := http.Header{"Origin": {"https://elsewhere.example"}}
-	if conn, resp, err := d.Dial("ws://"+addr+"/v1/live", h); err == nil {
+	for _, c := range []struct {
+		path, origin string
+		ok           bool
+	}{
+		{"/v1/live", "http://" + addr, true},
+		{"/v1/live", "https://app.example", false},
+		{"/v1/open", "https://app.example", true},
+		{"/v1/open", "https://elsewhere.example", false},
+	} {
+		conn, resp, err := d.Dial("ws://"+addr+c.path, http.Header{"Origin": {c.origin}})
+		if err == nil {
 			_ = conn.Close()
-		t.Error("a cross-site origin opened the socket")
-	} else if resp == nil || resp.StatusCode != 403 {
-		t.Errorf("a cross-site origin got %v, %v", resp, err)
 		}
-	conn, _, err := d.Dial("ws://"+addr+"/v1/live", http.Header{"Origin": {"http://" + addr}})
-	if err != nil {
-		t.Fatalf("the socket's own origin was refused: %v", err)
+		if c.ok != (err == nil) || !c.ok && (resp == nil || resp.StatusCode != 403) {
+			t.Errorf("%s from %s: %v, %v", c.path, c.origin, resp, err)
+		}
 	}
-	_ = conn.Close()
 }
 
 // A 3xx that carries a body is read as an answer by the remote CLI; one that
@@ -230,3 +272,54 @@ func TestKinds_AValueOrStreamCrossesTheCallPlane(t *testing.T) {
 		t.Errorf("got %+v", got)
 	}
 }
+
+type metaOut struct {
+	// W is the width.
+	W int `json:"w"`
+}
+
+// An op that answers JSON or image bytes names the bytes' media with Produces,
+// and its JSON still goes out as JSON.
+func TestKinds_JSONBesideBytesStaysJSON(t *testing.T) {
+	app := zip.New(zip.Config{AppName: "img", DisableStartupMessage: true})
+	app.Get("/v1/thumb", func(context.Context, *struct{}) (*zip.Or[metaOut, zip.Body], error) {
+		return &zip.Or[metaOut, zip.Body]{A: &metaOut{W: 64}}, nil
+	}, zip.Produces("image/png"))
+	resp, err := app.Test(httptest.NewRequest("GET", "/v1/thumb", nil))
+	if err != nil {
+		t.Fatal(err)
+	}
+	b, _ := io.ReadAll(resp.Body)
+	if ct := resp.Header.Get("Content-Type"); ct != "application/json; charset=utf-8" || string(b) != `{"w":64}` {
+		t.Errorf("answered %q %s", ct, b)
+	}
+}
+
+// A 3xx that is not a redirect carries the body it sends, 204 and 304 carry
+// none, and a redirect's 3xx carries a Location and no body. The doc comment's
+// example rides every status that answers the Out.
+func TestKinds_WhichStatusesCarryABody(t *testing.T) {
+	app := zip.New(zip.Config{AppName: "st", DisableStartupMessage: true})
+	zip.Describe("POST /v1/see", zip.Doc{Description: "See answers.", Response: json.RawMessage(`{"id":"x"}`)})
+	app.Post("/v1/see", func(context.Context, *struct{}) (*multiOut, error) {
+		return &multiOut{ID: "x", code: 302}, nil
+	}, zip.WithStatus(200, 201, 302, 304))
+	app.Get("/v1/go", func(context.Context, *struct{}) (*zip.Redirect, error) {
+		return &zip.Redirect{To: "/"}, nil
+	}, zip.WithStatus(307))
+	see := opIn(t, app, "/v1/see", "post")["responses"].(map[string]any)
+	for code, want := range map[string]bool{"200": true, "201": true, "302": true, "304": false} {
+		entry := see[code].(map[string]any)
+		content, has := entry["content"].(map[string]any)
+		if has != want {
+			t.Errorf("%s carries content %v, want %v", code, has, want)
+			continue
+		}
+		if want && content["application/json"].(map[string]any)["example"] == nil {
+			t.Errorf("%s carries no example", code)
+		}
+	}
+	if r := opIn(t, app, "/v1/go", "get")["responses"].(map[string]any)["307"].(map[string]any); r["content"] != nil {
+		t.Errorf("a redirect's 307 carries content: %v", r)
+	}
+}

manifest.go

@@ -134,6 +134,10 @@ type ManifestOp struct {
 	// Verbatim says the answer is an upstream's, relayed as it was sent, so a
 	// refusal may be the upstream's own rather than this service's.
 	Verbatim bool `json:"verbatim,omitempty"`
+
+	// Redirect says the op may answer a redirect, so its 3xx statuses carry a
+	// Location and no body.
+	Redirect bool `json:"redirect,omitempty"`
 }
 
 // streams are the answers a manifest may say an op streams, by media type.
@@ -386,6 +390,7 @@ func (a *App) Manifest() Manifest {
 			Consumes:        op.Consumes,
 			Produces:        op.Produces,
 			Verbatim:        op.ans.verbatim,
+			Redirect:        op.ans.redirect,
 		}
 		if hasDoc {
 			mo.Description = doc.Description

project.go

@@ -301,20 +301,29 @@ func (p *projector) responses(op ManifestOp, reg *schemaRegistry) map[string]any
 		dflt = 204
 	}
 	codes := statuses(op.Statuses, dflt)
+	// A status that carries no body: 204 and 304 by definition, and a 3xx of
+	// an op that redirects, which carries a Location instead.
+	bodyless := func(code int) bool {
+		return code == 204 || code == 304 || op.Redirect && code >= 300 && code < 400
+	}
 	primary := codes[0]
+	var bodied []int
 	for _, code := range codes {
-		if code < 300 || code >= 400 {
-			primary = code
-			break
+		if !bodyless(code) {
+			bodied = append(bodied, code)
 		}
 	}
-	byStatus := p.alternatives(op.Out, codes)
+	if len(bodied) > 0 {
+		primary = bodied[0]
+	}
+	byStatus := p.alternatives(op.Out, bodied)
+	union := p.types[op.Out] != nil && p.types[op.Out].Kind == "union"
 	for _, code := range codes {
 		entry := map[string]any{"description": statusText(code)}
 		if h := headerDecls(op); h != nil {
 			entry["headers"] = h
 		}
-		if code >= 300 && code < 400 {
+		if bodyless(code) {
 			resp[strconv.Itoa(code)] = entry
 			continue
 		}
@@ -331,7 +340,9 @@ func (p *projector) responses(op ManifestOp, reg *schemaRegistry) map[string]any
 				schema = map[string]any{"oneOf": one}
 			}
 			media := map[string]any{"schema": schema}
-			if len(op.Response) > 0 && code == primary {
+			// The doc comment's Response example is the Out's, at every status
+			// that answers it; a union's is the primary answer's.
+			if len(op.Response) > 0 && (code == primary || !union) {
 				media["example"] = json.RawMessage(op.Response)
 			}
 			jsonMedia := []string{mimeJSON}
@@ -376,9 +387,9 @@ func (p *projector) answers(out string) bool {
 
 // alternatives files an op's JSON answer by status. An alternative of a
 // union that states its own status is filed under it; the answer, or each
-// alternative that states none, is filed under every declared status that is
-// not a redirect and that no alternative claimed — an op that declares 200 and
-// 201 answers the same value under either. A named union none of whose
+// alternative that states none, is filed under every status in codes (the
+// ones that carry a body) that no alternative claimed — an op that declares 200
+// and 201 answers the same value under either. A named union none of whose
 // alternatives states a status is its own schema at each, one component every
 // op that answers it shares.
 func (p *projector) alternatives(out string, codes []int) map[int][]TypeRef {
@@ -405,10 +416,7 @@ func (p *projector) alternatives(out string, codes []int) map[int][]TypeRef {
 		by[code] = alts
 	}
 	for _, code := range codes {
-		if code >= 300 && code < 400 || len(unstated) == 0 {
-			continue
-		}
-		if _, taken := stated[code]; taken {
+		if _, taken := stated[code]; taken || len(unstated) == 0 {
 			continue
 		}
 		by[code] = append(by[code], unstated...)

pump_internal_test.go

@@ -6,6 +6,7 @@ package zip
 
 import (
 	"context"
+	"io"
 	"testing"
 	"time"
 )
@@ -65,3 +66,20 @@ func TestCollect_AStuckProducerCannotHoldTheCall(t *testing.T) {
 		t.Fatal("collect waited on a producer that never emits")
 	}
 }
+
+// A reader that never yields cannot hold a tool call: the read is bounded in
+// time as a stream is, and the reader is closed.
+func TestReadBounded_AQuietReaderEnds(t *testing.T) {
+	r, w := io.Pipe()
+	defer func() { _ = w.Close() }()
+	go func() { _, _ = w.Write([]byte("first")) }()
+	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+	defer cancel()
+	data, truncated, err := readBounded(ctx, r)
+	if err != nil || !truncated || string(data) != "first" {
+		t.Errorf("readBounded = %q %v %v", data, truncated, err)
+	}
+	if _, err := w.Write([]byte("x")); err == nil {
+		t.Error("the reader was left open")
+	}
+}

typed.go

@@ -60,6 +60,9 @@ type registeredOp struct {
 	// [Produces].
 	Consumes []string
 	Produces []string
+	// Origins are the browser origins besides its own that may open the op's
+	// WebSocket ([Origins]).
+	Origins []string
 	// req is how the input arrives and ans what the answer may be, each read
 	// once from In and Out at registration.
 	req intake
@@ -867,7 +870,7 @@ func registerTyped[In, Out any](depth int, on *App, method, path string, fn Type
 		// before its handler runs, so a handler with effects (a one-time ticket)
 		// runs only for a connection it will get.
 		if op.ans.stream == streamSocket {
-			if err := upgradable(c); err != nil {
+			if err := upgradable(c, op.Origins); err != nil {
 				return err
 			}
 		}

wsx/wsx.go

@@ -10,8 +10,7 @@
 //	}))
 //
 // A typed op that answers a WebSocket returns a [zip.Socket] instead; both run
-// the same upgrade (internal/ws), so a connection is accepted and refused the
-// same way at either kind of route.
+// the same upgrade (internal/ws).
 package wsx
 
 import (