diff --git a/packages/agentproxy/proxy.go b/packages/agentproxy/proxy.go
index aa7df7c9..06bd1300 100644
--- a/packages/agentproxy/proxy.go
+++ b/packages/agentproxy/proxy.go
@@ -387,7 +387,15 @@ func (ps *proxyServer) handleConnect(w http.ResponseWriter, r *http.Request) {
 	if err != nil {
 		return
 	}
-	defer clientConn.Close()
+	// A handler inside the tunnel may hijack the conn again (a WebSocket pipe, which outlives the tunnel
+	// server). Serve returns the moment that happens, while the handler is still running, so closing
+	// unconditionally here would cut a live WebSocket out from under itself.
+	tunnelOwnsConn := true
+	defer func() {
+		if tunnelOwnsConn {
+			_ = clientConn.Close()
+		}
+	}()
 
 	if _, err := clientConn.Write([]byte("HTTP/1.1 200 Connection Established\r\n\r\n")); err != nil {
 		return
@@ -405,13 +413,18 @@ func (ps *proxyServer) handleConnect(w http.ResponseWriter, r *http.Request) {
 	}
 	_ = tlsConn.SetDeadline(time.Time{})
 
-	ps.serveTunnel(tlsConn, hostname, port, jwt, scope)
+	if ps.serveTunnel(tlsConn, hostname, port, jwt, scope) {
+		tunnelOwnsConn = false
+	}
 }
 
 // serveTunnel serves HTTP/1.1 requests off the decrypted MITM connection using a fresh http.Server over a
-// one-shot listener, so the tunnel gets the same header/timeout enforcement as the ingress.
-func (ps *proxyServer) serveTunnel(tlsConn *tls.Conn, hostname, port, jwt string, scope agentScope) {
+// one-shot listener, so the tunnel gets the same header/timeout enforcement as the ingress. It reports whether
+// the inner handler hijacked the connection, in which case that handler owns it and the caller must not close
+// it: Serve returns immediately on hijack while the handler is still running.
+func (ps *proxyServer) serveTunnel(tlsConn *tls.Conn, hostname, port, jwt string, scope agentScope) bool {
 	listener := newOneShotListener(tlsConn)
+	var hijacked atomic.Bool
 	srv := &http.Server{
 		Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 			ps.forwardHTTP(w, r, "https", hostname, port, jwt, scope)
@@ -424,12 +437,16 @@ func (ps *proxyServer) serveTunnel(tlsConn *tls.Conn, hostname, port, jwt string
 		// The one-shot listener yields the single conn once, then blocks; closing it on terminal conn state
 		// makes Serve return. The conn is owned by http.Server (Closed) or the hijack handler, not closed here.
 		ConnState: func(_ net.Conn, state http.ConnState) {
+			if state == http.StateHijacked {
+				hijacked.Store(true)
+			}
 			if state == http.StateHijacked || state == http.StateClosed {
 				_ = listener.Close()
 			}
 		},
 	}
 	_ = srv.Serve(listener)
+	return hijacked.Load()
 }
 
 // Only http:// absolute-form is served; https:// is rejected so the proxy can never be used to silently TLS-strip (HTTPS must arrive as CONNECT).
@@ -472,6 +489,13 @@ func (ps *proxyServer) forwardHTTP(w http.ResponseWriter, r *http.Request, schem
 		return
 	}
 
+	// A WebSocket upgrade cannot be relayed through the ResponseWriter: after 101 the connection stops being
+	// request/response and the proxy has to own both sockets.
+	if isWebSocketUpgrade(r) {
+		ps.serveWebSocket(w, r, scheme, hostname, port, jwt, scope)
+		return
+	}
+
 	// EscapedPath keeps the path encoded (no log injection); the cap bounds record size.
 	method := r.Method
 	reqPath := r.URL.EscapedPath()
@@ -606,9 +630,25 @@ type forwardOutcome struct {
 }
 
 func (ps *proxyServer) forward(req *http.Request, scheme, hostname, port, jwt string, scope agentScope) (*http.Response, forwardOutcome, error) {
+	outcome, _, err := ps.prepareUpstream(req, scheme, hostname, port, jwt, scope)
+	if err != nil {
+		return nil, outcome, err
+	}
+
+	resp, err := ps.transport.RoundTrip(req)
+	if err != nil {
+		return nil, outcome, err
+	}
+	return resp, outcome, nil
+}
+
+// Stops short of dispatching, because the HTTP path hands req to the pooled transport while the WebSocket path
+// has to own the connection itself. The resolved credentials come back so that path can arm frame
+// substitution.
+func (ps *proxyServer) prepareUpstream(req *http.Request, scheme, hostname, port, jwt string, scope agentScope) (forwardOutcome, []resolvedCredential, error) {
 	services, err := ps.resolver.get(jwt, scope)
 	if err != nil {
-		return nil, forwardOutcome{}, fmt.Errorf("failed to resolve agent permissions: %w", err)
+		return forwardOutcome{}, nil, fmt.Errorf("failed to resolve agent permissions: %w", err)
 	}
 
 	// Capture identity now; reading it after the round trip would race a cache eviction and drop the record.
@@ -618,7 +658,7 @@ func (ps *proxyServer) forward(req *http.Request, scheme, hostname, port, jwt st
 	svc := bestMatch(services, hostname, port, req.URL.Path)
 
 	if svc == nil && ps.opts.UnmatchedHost == UnmatchedBlock && !ps.hostAllowlisted(hostname) {
-		return nil, outcome, fmt.Errorf("host %q has no matching proxied service: %w", hostname, errHostBlocked)
+		return outcome, nil, fmt.Errorf("host %q has no matching proxied service: %w", hostname, errHostBlocked)
 	}
 
 	outcome.service = svc
@@ -629,23 +669,36 @@ func (ps *proxyServer) forward(req *http.Request, scheme, hostname, port, jwt st
 	req.Host = hostHeaderForScheme(scheme, req.URL.Host)
 	req.RequestURI = ""
 
+	// The upgrade headers are hop-by-hop, so snapshot them before the strip: without Upgrade, Connection and
+	// the Sec-WebSocket-* set the upstream never switches protocols.
+	var handshake http.Header
+	if isWebSocketUpgrade(req) {
+		handshake = captureWebSocketHandshakeHeaders(req.Header)
+	}
+
 	// Strip hop-by-hop before injecting so a client's Connection header cannot delete the injected credential (injected always wins).
 	stripHopByHopHeaders(req.Header)
 
+	// Restored before injection, for the same reason: a header rewrite naming one of these still wins.
+	if handshake != nil {
+		restoreHeaders(req.Header, handshake)
+		// Connection is rebuilt, not restored. Every token an agent puts here is hop-by-hop to the upstream, so
+		// forwarding the agent's own value would let it send "Connection: Upgrade, Authorization" and have the
+		// service strip the injected credential before seeing it. Upgrade is all the handshake needs.
+		req.Header.Set("Connection", "Upgrade")
+	}
+
+	var creds []resolvedCredential
 	if svc != nil {
-		creds := ps.materializeCredentials(svc)
+		creds = ps.materializeCredentials(svc)
 		applied, err := applyCredentials(req, creds)
 		if err != nil {
-			return nil, outcome, fmt.Errorf("failed to apply credentials: %w", err)
+			return outcome, nil, fmt.Errorf("failed to apply credentials: %w", err)
 		}
 		outcome.applied = applied
 	}
 
-	resp, err := ps.transport.RoundTrip(req)
-	if err != nil {
-		return nil, outcome, err
-	}
-	return resp, outcome, nil
+	return outcome, creds, nil
 }
 
 // hostAllowlisted reports whether hostname is in AllowedHosts (case-insensitive).
diff --git a/packages/agentproxy/rewrite.go b/packages/agentproxy/rewrite.go
index 1549b555..7471c71a 100644
--- a/packages/agentproxy/rewrite.go
+++ b/packages/agentproxy/rewrite.go
@@ -18,9 +18,58 @@ const (
 	surfacePath        = "path"
 	surfaceQuery       = "query"
 	surfaceBody        = "body"
+	surfaceWebSocket   = "websocket"
 	maxBodyRewriteSize = 10 * 1024 * 1024
 )
 
+// Unlike the other surfaces this one cannot be applied to the request: the placeholder appears in frames sent
+// after the upgrade, so it is carried to the frame pipe instead.
+type wsSubstitution struct {
+	placeholder string
+	value       string
+	label       AppliedCredential
+}
+
+// wsReplacement is one direction's swap. Outbound turns the placeholder into the real value; inbound turns it
+// back, so a value the service reflects never reaches the agent.
+type wsReplacement struct {
+	from string
+	to   string
+}
+
+func forwardReplacements(subs []wsSubstitution) []wsReplacement {
+	out := make([]wsReplacement, 0, len(subs))
+	for _, sub := range subs {
+		out = append(out, wsReplacement{from: sub.placeholder, to: sub.value})
+	}
+	return out
+}
+
+func reverseReplacements(subs []wsSubstitution) []wsReplacement {
+	out := make([]wsReplacement, 0, len(subs))
+	for _, sub := range subs {
+		out = append(out, wsReplacement{from: sub.value, to: sub.placeholder})
+	}
+	return out
+}
+
+func websocketSubstitutions(creds []resolvedCredential) []wsSubstitution {
+	var out []wsSubstitution
+	for _, cred := range creds {
+		if cred.role != roleCredentialSub || cred.placeholder == "" {
+			continue
+		}
+		if !hasSurface(cred.surfaces, surfaceWebSocket) {
+			continue
+		}
+		label := credLabel(cred)
+		label.Role = roleCredentialSub
+		label.Surfaces = []string{surfaceWebSocket}
+		out = append(out, wsSubstitution{placeholder: cred.placeholder, value: cred.value, label: label})
+	}
+	return out
+}
+
 type AppliedCredential struct {
 	Key                string   `json:"key,omitempty"`
 	DynamicSecretName  string   `json:"dynamicSecretName,omitempty"`
diff --git a/packages/agentproxy/websocket.go b/packages/agentproxy/websocket.go
new file mode 100644
index 00000000..d26be090
--- /dev/null
+++ b/packages/agentproxy/websocket.go
@@ -0,0 +1,622 @@
+package agentproxy
+
+import (
+	"bufio"
+	"context"
+	"crypto/rand"
+	"crypto/tls"
+	"encoding/binary"
+	"errors"
+	"fmt"
+	"io"
+	"math"
+	"net"
+	"net/http"
+	"net/textproto"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/rs/zerolog"
+	"github.com/rs/zerolog/log"
+)
+
+const (
+	// Without an idle bound a stalled or abandoned connection pins a goroutine pair, an upstream conn, and one
+	// of the proxy's maxConcurrentConns slots indefinitely. Real-time keepalives sit well inside this window.
+	wsIdleTimeout = 10 * time.Minute
+
+	wsResponseTimeout = 30 * time.Second
+
+	// The response timeout only starts once TCP is up, so without a dial bound an agent can aim a WebSocket at a
+	// black-holed address and pin a connection slot for however long the OS takes to give up.
+	wsDialTimeout = 10 * time.Second
+
+	// Larger text frames stream through untouched: frame length is attacker-controlled and auth payloads are
+	// tiny, so buffering by the declared length would be a memory-exhaustion lever.
+	maxWSSubstitutionPayload = 1 << 20
+)
+
+// WebSocket frame opcodes (RFC 6455 section 11.8).
+const (
+	wsOpText  = 0x1
+	wsOpClose = 0x8
+)
+
+// These have to survive stripHopByHopHeaders, or the upstream never switches protocols. Connection is
+// deliberately absent: prepareUpstream rebuilds it rather than carrying the agent's value over.
+var websocketHandshakeHeaderNames = []string{
+	"Origin",
+	"Sec-Websocket-Extensions",
+	"Sec-Websocket-Key",
+	"Sec-Websocket-Protocol",
+	"Sec-Websocket-Version",
+	"Upgrade",
+}
+
+func isWebSocketUpgrade(r *http.Request) bool {
+	if !strings.EqualFold(r.Header.Get("Upgrade"), "websocket") {
+		return false
+	}
+	for _, header := range r.Header.Values("Connection") {
+		for _, token := range strings.Split(header, ",") {
+			if strings.EqualFold(strings.TrimSpace(token), "upgrade") {
+				return true
+			}
+		}
+	}
+	return false
+}
+
+func captureWebSocketHandshakeHeaders(src http.Header) http.Header {
+	captured := make(http.Header)
+	for _, name := range websocketHandshakeHeaderNames {
+		for _, value := range src.Values(name) {
+			captured.Add(name, value)
+		}
+	}
+	return captured
+}
+
+func restoreHeaders(dst, src http.Header) {
+	for name, values := range src {
+		dst.Del(name)
+		for _, value := range values {
+			dst.Add(name, value)
+		}
+	}
+}
+
+// Serves both wss (off the MITM tunnel) and ws (a plaintext forward request). After 101 the proxy owns both
+// sockets for the connection's lifetime rather than relaying through the ResponseWriter.
+func (ps *proxyServer) serveWebSocket(w http.ResponseWriter, r *http.Request, scheme, hostname, port, jwt string, scope agentScope) {
+	reqPath := r.URL.EscapedPath()
+	if len(reqPath) > maxLoggedPathLen {
+		reqPath = reqPath[:maxLoggedPathLen] + "...[truncated]"
+	}
+
+	outcome, creds, err := ps.prepareUpstream(r, scheme, hostname, port, jwt, scope)
+	if err != nil {
+		decision, status := decisionError, http.StatusBadGateway
+		if errors.Is(err, errHostBlocked) {
+			decision, status = decisionBlocked, http.StatusForbidden
+		}
+		ps.emitActivity(r.Method, reqPath, hostname, port, decision, status, scope, outcome, err)
+		http.Error(w, err.Error(), status)
+		return
+	}
+
+	// Settled before the handshake goes out. A permessage-deflate frame carries RSV1 and is never substituted,
+	// and most clients offer compression by default, so leaving the offer in would silently turn substitution
+	// into a no-op. Costs bandwidth, and only where the surface is actually used.
+	wsSubs := websocketSubstitutions(creds)
+	if len(wsSubs) > 0 && dropPerMessageDeflate(r.Header) {
+		log.Debug().
+			Str("host", hostname).
+			Msg("dropped permessage-deflate from the websocket upgrade offer so frame substitution can read message text")
+	}
+
+	upstreamConn, upstreamReader, resp, err := ps.dialWebSocketUpstream(r.Context(), r)
+	if err != nil {
+		ps.emitActivity(r.Method, reqPath, hostname, port, decisionError, http.StatusBadGateway, scope, outcome, err)
+		http.Error(w, "failed to reach upstream for websocket upgrade", http.StatusBadGateway)
+		return
+	}
+
+	decision := decisionPassthrough
+	if outcome.service != nil {
+		decision = decisionBrokered
+		ps.recordUsage(outcome.service.id)
+	}
+
+	// Relay a refusal like any other response, so the agent sees the real 401 or 404 rather than a proxy error.
+	if resp.StatusCode != http.StatusSwitchingProtocols {
+		defer func() {
+			_ = resp.Body.Close()
+			_ = upstreamConn.Close()
+		}()
+		ps.emitActivity(r.Method, reqPath, hostname, port, decision, resp.StatusCode, scope, outcome, nil)
+
+		stripHopByHopHeaders(resp.Header)
+		dst := w.Header()
+		for name, values := range resp.Header {
+			for _, v := range values {
+				dst.Add(name, v)
+			}
+		}
+		w.WriteHeader(resp.StatusCode)
+		_, _ = io.Copy(flushingWriter{w}, resp.Body)
+		return
+	}
+
+	// Reports only what was applied to the handshake. Credentials armed for frame substitution are not claimed
+	// here: no frame has been rewritten yet, and the agent may never send the placeholder at all.
+	ps.emitActivity(r.Method, reqPath, hostname, port, decision, http.StatusSwitchingProtocols, scope, outcome, nil)
+
+	hijacker, ok := w.(http.Hijacker)
+	if !ok {
+		_ = upstreamConn.Close()
+		http.Error(w, "connection hijacking unsupported", http.StatusInternalServerError)
+		return
+	}
+	clientConn, clientBuf, err := hijacker.Hijack()
+	if err != nil {
+		_ = upstreamConn.Close()
+		return
+	}
+
+	// Hijack leaves the server's ReadTimeout/WriteTimeout deadlines on the conn, which would kill a long-lived
+	// WebSocket at tunnelReadTimeout. The pipe below applies its own idle deadline instead.
+	_ = clientConn.SetDeadline(time.Time{})
+
+	_ = clientConn.SetWriteDeadline(time.Now().Add(10 * time.Second))
+	if err := writeWebSocketSwitchingResponse(clientConn, resp); err != nil {
+		_ = clientConn.Close()
+		_ = upstreamConn.Close()
+		return
+	}
+	_ = clientConn.SetWriteDeadline(time.Time{})
+
+	counts := pipeWebSocket(clientConn, clientBuf.Reader, upstreamConn, upstreamReader, wsSubs)
+
+	// A connection that rewrote a credential is an audit event, so it must not be filtered out with the
+	// passthrough noise.
+	level := zerolog.DebugLevel
+	if counts.substituted > 0 || counts.redacted > 0 {
+		level = zerolog.InfoLevel
+	}
+	ev := log.WithLevel(level).
+		Str("event", activityEventName).
+		Str("decision", decision).
+		Str("agentId", outcome.agentID).
+		Str("projectId", scope.projectID).
+		Str("environment", scope.environment).
+		Str("secretPath", scope.secretPath).
+		Str("host", hostname).
+		Str("path", reqPath).
+		Int("framesSubstituted", counts.substituted).
+		Int("framesRedacted", counts.redacted)
+	// Named only once a frame was actually rewritten, so the record cannot claim a credential was applied to a
+	// connection that never carried one.
+	if counts.substituted > 0 {
+		labels := make([]AppliedCredential, 0, len(wsSubs))
+		for _, sub := range wsSubs {
+			labels = append(labels, sub.label)
+		}
+		ev = ev.Interface("credentials", labels)
+	}
+	ev.Msg("websocket closed")
+}
+
+// Offers are comma separated and their parameters semicolon separated, so splitting on commas is safe.
+func dropPerMessageDeflate(h http.Header) bool {
+	values := h.Values("Sec-Websocket-Extensions")
+	if len(values) == 0 {
+		return false
+	}
+
+	var kept []string
+	changed := false
+	for _, value := range values {
+		for _, offer := range strings.Split(value, ",") {
+			offer = strings.TrimSpace(offer)
+			if offer == "" {
+				continue
+			}
+			name := strings.ToLower(strings.TrimSpace(strings.SplitN(offer, ";", 2)[0]))
+			if name == "permessage-deflate" || name == "x-webkit-deflate-frame" {
+				changed = true
+				continue
+			}
+			kept = append(kept, offer)
+		}
+	}
+	if !changed {
+		return false
+	}
+
+	h.Del("Sec-Websocket-Extensions")
+	if len(kept) > 0 {
+		h.Set("Sec-Websocket-Extensions", strings.Join(kept, ", "))
+	}
+	return true
+}
+
+// The WebSocket dial needs the raw conn, so it cannot inherit the transport's settings by going through it.
+// Reading them off explicitly is what keeps upstream verification identical to the plain HTTP path.
+func (ps *proxyServer) upstreamTLSConfig(serverName string) *tls.Config {
+	cfg := &tls.Config{}
+	if t, ok := ps.transport.(*http.Transport); ok && t.TLSClientConfig != nil {
+		cfg = t.TLSClientConfig.Clone()
+	}
+	if cfg.ServerName == "" {
+		cfg.ServerName = serverName
+	}
+	if cfg.MinVersion == 0 {
+		cfg.MinVersion = tls.VersionTLS12
+	}
+	// WebSocket is HTTP/1.1 only; pin ALPN so the upstream cannot select h2.
+	cfg.NextProtos = []string{"http/1.1"}
+	return cfg
+}
+
+// A hijacked WebSocket cannot be driven through http.Transport's pooled round tripper, so the handshake runs
+// on a connection the proxy owns.
+func (ps *proxyServer) dialWebSocketUpstream(ctx context.Context, outReq *http.Request) (net.Conn, *bufio.Reader, *http.Response, error) {
+	dialer := &net.Dialer{Timeout: wsDialTimeout}
+	rawConn, err := dialer.DialContext(ctx, "tcp", outReq.URL.Host)
+	if err != nil {
+		return nil, nil, nil, err
+	}
+
+	// ws:// upstream: no TLS. pipeWebSocket and copyWithIdleTimeout only use net.Conn methods, so a
+	// *net.TCPConn substitutes for a *tls.Conn.
+	conn := net.Conn(rawConn)
+	if outReq.URL.Scheme != "http" {
+		tlsConn := tls.Client(rawConn, ps.upstreamTLSConfig(outReq.URL.Hostname()))
+		_ = tlsConn.SetDeadline(time.Now().Add(tlsHandshakeTimeout))
+		if err := tlsConn.HandshakeContext(ctx); err != nil {
+			_ = rawConn.Close()
+			return nil, nil, nil, err
+		}
+		_ = tlsConn.SetDeadline(time.Time{})
+		conn = tlsConn
+	}
+
+	_ = conn.SetDeadline(time.Now().Add(wsResponseTimeout))
+	if err := outReq.Write(conn); err != nil {
+		_ = conn.Close()
+		return nil, nil, nil, err
+	}
+	reader := bufio.NewReader(conn)
+	resp, err := http.ReadResponse(reader, outReq)
+	if err != nil {
+		_ = conn.Close()
+		return nil, nil, nil, err
+	}
+	_ = conn.SetDeadline(time.Time{})
+
+	return conn, reader, resp, nil
+}
+
+// The ResponseWriter is already hijacked, so Go's response machinery cannot frame this.
+func writeWebSocketSwitchingResponse(w io.Writer, resp *http.Response) error {
+	proto := resp.Proto
+	if proto == "" {
+		proto = "HTTP/1.1"
+	}
+	status := resp.Status
+	if status == "" {
+		status = fmt.Sprintf("%d %s", resp.StatusCode, http.StatusText(resp.StatusCode))
+	}
+	if _, err := fmt.Fprintf(w, "%s %s\r\n", proto, status); err != nil {
+		return err
+	}
+
+	header := make(http.Header)
+	for name, values := range resp.Header {
+		if !isSafeWebSocketSwitchHeader(name) {
+			continue
+		}
+		for _, v := range values {
+			header.Add(name, v)
+		}
+	}
+	header.Set("Connection", "Upgrade")
+	header.Set("Upgrade", "websocket")
+
+	for name, values := range header {
+		canonical := textproto.CanonicalMIMEHeaderKey(name)
+		for _, v := range values {
+			if _, err := fmt.Fprintf(w, "%s: %s\r\n", canonical, v); err != nil {
+				return err
+			}
+		}
+	}
+	_, err := io.WriteString(w, "\r\n")
+	return err
+}
+
+// Unknown Sec-* headers are dropped so the upstream cannot smuggle extension state the proxy has not
+// accounted for. Body-framing headers are dropped too: a 101 carries no body, and a stray Content-Length or
+// Transfer-Encoding would desynchronize the client's frame parser.
+func isSafeWebSocketSwitchHeader(name string) bool {
+	canonical := http.CanonicalHeaderKey(name)
+	switch canonical {
+	case "Connection", "Upgrade", "Sec-Websocket-Accept", "Sec-Websocket-Extensions", "Sec-Websocket-Protocol":
+		return true
+	case "Content-Length", "Transfer-Encoding":
+		return false
+	}
+	if strings.HasPrefix(canonical, "Sec-") {
+		return false
+	}
+	// Every other hop-by-hop header is meaningless to the client and belongs to the upstream connection.
+	for _, name := range hopByHopHeaders {
+		if http.CanonicalHeaderKey(name) == canonical {
+			return false
+		}
+	}
+	return true
+}
+
+// wsIdle extends both read deadlines together. Per-direction deadlines would tear down a live connection: a
+// subscribe-mostly stream can receive for hours while sending nothing, and its own deadline would expire and
+// take the whole connection with it. Traffic either way counts as activity, so a timeout means both directions
+// have gone quiet.
+type wsIdle struct {
+	mu    sync.Mutex
+	conns [2]net.Conn
+}
+
+func (w *wsIdle) extend() {
+	deadline := time.Now().Add(wsIdleTimeout)
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	for _, c := range w.conns {
+		_ = c.SetReadDeadline(deadline)
+	}
+}
+
+type wsCounts struct {
+	substituted int
+	redacted    int
+}
+
+func pipeWebSocket(clientConn net.Conn, clientReader *bufio.Reader, upstreamConn net.Conn, upstreamReader *bufio.Reader, wsSubs []wsSubstitution) wsCounts {
+	done := make(chan struct{}, 2)
+	var closeOnce sync.Once
+	closeBoth := func() {
+		closeOnce.Do(func() {
+			_ = clientConn.Close()
+			_ = upstreamConn.Close()
+		})
+	}
+
+	idle := &wsIdle{conns: [2]net.Conn{clientConn, upstreamConn}}
+	idle.extend()
+
+	var counts wsCounts
+	go func() {
+		defer func() {
+			done <- struct{}{}
+			closeBoth()
+		}()
+		src := io.MultiReader(clientReader, clientConn)
+		if len(wsSubs) > 0 {
+			counts.substituted = copyWSFrames(upstreamConn, src, idle, forwardReplacements(wsSubs))
+		} else {
+			copyWithIdleTimeout(upstreamConn, src, idle)
+		}
+	}()
+	go func() {
+		defer func() {
+			done <- struct{}{}
+			closeBoth()
+		}()
+		src := io.MultiReader(upstreamReader, upstreamConn)
+		// The agent picks where its placeholder goes, so it can plant one in a field the service echoes (a
+		// correlation id, an error message) and read the real credential out of the reply. Swapping the value
+		// back on the way in closes that, and also stops a service that quotes the credential in an error from
+		// leaking it by accident.
+		if len(wsSubs) > 0 {
+			counts.redacted = copyWSFrames(clientConn, src, idle, reverseReplacements(wsSubs))
+		} else {
+			copyWithIdleTimeout(clientConn, src, idle)
+		}
+	}()
+
+	<-done
+	<-done
+	return counts
+}
+
+func copyWithIdleTimeout(dst io.Writer, src io.Reader, idle *wsIdle) {
+	buf := make([]byte, 32*1024)
+	for {
+		n, err := src.Read(buf)
+		if n > 0 {
+			idle.extend()
+			if _, werr := dst.Write(buf[:n]); werr != nil {
+				return
+			}
+		}
+		if err != nil {
+			return
+		}
+	}
+}
+
+// Binary, fragmented, compressed (RSV1) and oversized frames are forwarded byte-for-byte, so a shape the
+// parser cannot safely rewrite degrades to passthrough instead of corrupting the stream.
+func copyWSFrames(dst io.Writer, src io.Reader, idle *wsIdle, reps []wsReplacement) int {
+	r := bufio.NewReaderSize(src, 32*1024)
+	rewritten := 0
+
+	for {
+		hdr := make([]byte, 2)
+		if _, err := io.ReadFull(r, hdr); err != nil {
+			return rewritten
+		}
+
+		fin := hdr[0]&0x80 != 0
+		rsv1 := hdr[0]&0x40 != 0
+		opcode := hdr[0] & 0x0F
+		masked := hdr[1]&0x80 != 0
+		payloadLen := uint64(hdr[1] & 0x7F)
+
+		var extHdr []byte
+		switch payloadLen {
+		case 126:
+			extHdr = make([]byte, 2)
+			if _, err := io.ReadFull(r, extHdr); err != nil {
+				return rewritten
+			}
+			payloadLen = uint64(binary.BigEndian.Uint16(extHdr))
+		case 127:
+			extHdr = make([]byte, 8)
+			if _, err := io.ReadFull(r, extHdr); err != nil {
+				return rewritten
+			}
+			payloadLen = binary.BigEndian.Uint64(extHdr)
+			// RFC 6455 section 5.2: the MSB must be 0. Rejecting here prevents an int64 overflow in io.CopyN
+			// that would desynchronize the frame parser.
+			if payloadLen > math.MaxInt64 {
+				return rewritten
+			}
+		}
+
+		var maskKey [4]byte
+		if masked {
+			if _, err := io.ReadFull(r, maskKey[:]); err != nil {
+				return rewritten
+			}
+		}
+
+		writeFrameHeader := func() bool {
+			if _, err := dst.Write(hdr); err != nil {
+				return false
+			}
+			if len(extHdr) > 0 {
+				if _, err := dst.Write(extHdr); err != nil {
+					return false
+				}
+			}
+			if masked {
+				if _, err := dst.Write(maskKey[:]); err != nil {
+					return false
+				}
+			}
+			return true
+		}
+
+		if opcode != wsOpText || !fin || rsv1 || payloadLen > maxWSSubstitutionPayload {
+			if opcode == wsOpText && (!fin || rsv1 || payloadLen > maxWSSubstitutionPayload) {
+				log.Warn().
+					Bool("fragmented", !fin).
+					Bool("compressed", rsv1).
+					Uint64("payloadLen", payloadLen).
+					Msg("websocket text frame not eligible for substitution; forwarding unchanged")
+			}
+			if !writeFrameHeader() {
+				return rewritten
+			}
+			if payloadLen > 0 {
+				if _, err := io.CopyN(dst, r, int64(payloadLen)); err != nil {
+					return rewritten
+				}
+			}
+			idle.extend()
+			if opcode == wsOpClose {
+				return rewritten
+			}
+			continue
+		}
+
+		payload := make([]byte, payloadLen)
+		if _, err := io.ReadFull(r, payload); err != nil {
+			return rewritten
+		}
+		idle.extend()
+
+		raw := make([]byte, len(payload))
+		copy(raw, payload)
+
+		if masked {
+			for i := range payload {
+				payload[i] ^= maskKey[i%4]
+			}
+		}
+
+		text := string(payload)
+		for _, rep := range reps {
+			if replaced, ok := replaceWithinLimit(text, rep.from, rep.to, maxWSSubstitutionPayload); ok {
+				text = replaced
+			}
+		}
+
+		// Nothing matched: forward the original bytes rather than re-encoding an identical frame.
+		if text == string(payload) {
+			if !writeFrameHeader() {
+				return rewritten
+			}
+			if _, err := dst.Write(raw); err != nil {
+				return rewritten
+			}
+			continue
+		}
+
+		if err := writeSubstitutedFrame(dst, hdr[0], []byte(text), masked); err != nil {
+			return rewritten
+		}
+		rewritten++
+	}
+}
+
+// RFC 6455 section 5.3 requires client masks be unpredictable, and reusing the client's key would leak the
+// XOR relationship between the placeholder and the real credential, so a fresh key is generated here.
+func writeSubstitutedFrame(dst io.Writer, firstByte byte, payload []byte, masked bool) error {
+	newLen := uint64(len(payload))
+
+	var frame []byte
+	switch {
+	case newLen <= 125:
+		second := byte(newLen)
+		if masked {
+			second |= 0x80
+		}
+		frame = append(frame, firstByte, second)
+	case newLen <= 65535:
+		second := byte(126)
+		if masked {
+			second |= 0x80
+		}
+		frame = append(frame, firstByte, second)
+		frame = binary.BigEndian.AppendUint16(frame, uint16(newLen))
+	default:
+		second := byte(127)
+		if masked {
+			second |= 0x80
+		}
+		frame = append(frame, firstByte, second)
+		frame = binary.BigEndian.AppendUint64(frame, newLen)
+	}
+
+	if masked {
+		var newMask [4]byte
+		if _, err := rand.Read(newMask[:]); err != nil {
+			return err
+		}
+		maskedPayload := make([]byte, len(payload))
+		for i := range payload {
+			maskedPayload[i] = payload[i] ^ newMask[i%4]
+		}
+		frame = append(frame, newMask[:]...)
+		frame = append(frame, maskedPayload...)
+	} else {
+		frame = append(frame, payload...)
+	}
+
+	_, err := dst.Write(frame)
+	return err
+}
diff --git a/packages/agentproxy/websocket_test.go b/packages/agentproxy/websocket_test.go
new file mode 100644
index 00000000..14767a4d
--- /dev/null
+++ b/packages/agentproxy/websocket_test.go
@@ -0,0 +1,875 @@
+package agentproxy
+
+import (
+	"bufio"
+	"bytes"
+	"crypto/tls"
+	"crypto/x509"
+	"encoding/binary"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"strings"
+	"testing"
+	"time"
+)
+
+// wsFakeConn satisfies net.Conn for the frame copiers, which only ever call SetReadDeadline on it.
+type wsFakeConn struct {
+	io.Reader
+	io.Writer
+}
+
+func (wsFakeConn) Close() error                     { return nil }
+func (wsFakeConn) LocalAddr() net.Addr              { return nil }
+func (wsFakeConn) RemoteAddr() net.Addr             { return nil }
+func (wsFakeConn) SetDeadline(time.Time) error      { return nil }
+func (wsFakeConn) SetReadDeadline(time.Time) error  { return nil }
+func (wsFakeConn) SetWriteDeadline(time.Time) error { return nil }
+
+func writeWSTextFrame(w io.Writer, text string, masked bool) error {
+	payload := []byte(text)
+	header := []byte{0x81, byte(len(payload))}
+	if masked {
+		header[1] |= 0x80
+	}
+	if _, err := w.Write(header); err != nil {
+		return err
+	}
+	mask := []byte{1, 2, 3, 4}
+	if masked {
+		if _, err := w.Write(mask); err != nil {
+			return err
+		}
+		for i := range payload {
+			payload[i] ^= mask[i%len(mask)]
+		}
+	}
+	_, err := w.Write(payload)
+	return err
+}
+
+func readWSTextFrame(r io.Reader) (string, error) {
+	header := make([]byte, 2)
+	if _, err := io.ReadFull(r, header); err != nil {
+		return "", err
+	}
+	if header[0]&0x0F != wsOpText {
+		return "", fmt.Errorf("opcode = %d, want text", header[0]&0x0F)
+	}
+
+	masked := header[1]&0x80 != 0
+	length := int(header[1] & 0x7F)
+	switch length {
+	case 126:
+		extended := make([]byte, 2)
+		if _, err := io.ReadFull(r, extended); err != nil {
+			return "", err
+		}
+		length = int(binary.BigEndian.Uint16(extended))
+	case 127:
+		return "", fmt.Errorf("large frames are not supported by this test helper")
+	}
+
+	mask := []byte{0, 0, 0, 0}
+	if masked {
+		if _, err := io.ReadFull(r, mask); err != nil {
+			return "", err
+		}
+	}
+	payload := make([]byte, length)
+	if _, err := io.ReadFull(r, payload); err != nil {
+		return "", err
+	}
+	if masked {
+		for i := range payload {
+			payload[i] ^= mask[i%len(mask)]
+		}
+	}
+	return string(payload), nil
+}
+
+func maskedTextFrame(t *testing.T, text string) []byte {
+	t.Helper()
+	var buf bytes.Buffer
+	if err := writeWSTextFrame(&buf, text, true); err != nil {
+		t.Fatalf("writeWSTextFrame: %v", err)
+	}
+	return buf.Bytes()
+}
+
+// maskedCloseFrame is opcode 0x8, FIN set, masked, empty payload.
+func maskedCloseFrame() []byte {
+	return []byte{0x88, 0x80, 0x00, 0x00, 0x00, 0x00}
+}
+
+func maskedBinaryFrame(payload []byte, mask [4]byte) []byte {
+	masked := make([]byte, len(payload))
+	for i := range payload {
+		masked[i] = payload[i] ^ mask[i%4]
+	}
+	hdr := []byte{0x82, byte(len(masked)) | 0x80}
+	hdr = append(hdr, mask[:]...)
+	return append(hdr, masked...)
+}
+
+// pingFrame is opcode 0x9, FIN set, unmasked.
+func pingFrame(payload []byte) []byte {
+	return append([]byte{0x89, byte(len(payload))}, payload...)
+}
+
+func wsSubs(placeholder, value string) []wsSubstitution {
+	return []wsSubstitution{{placeholder: placeholder, value: value}}
+}
+
+func runFrameCopy(t *testing.T, frames []byte, subs []wsSubstitution) ([]byte, int) {
+	t.Helper()
+	var dst bytes.Buffer
+	idle := &wsIdle{conns: [2]net.Conn{wsFakeConn{}, wsFakeConn{}}}
+	n := copyWSFrames(&dst, bytes.NewReader(frames), idle, forwardReplacements(subs))
+	return dst.Bytes(), n
+}
+
+// runReverseFrameCopy drives the upstream-to-client direction, where the swap runs the other way.
+func runReverseFrameCopy(t *testing.T, frames []byte, subs []wsSubstitution) ([]byte, int) {
+	t.Helper()
+	var dst bytes.Buffer
+	idle := &wsIdle{conns: [2]net.Conn{wsFakeConn{}, wsFakeConn{}}}
+	n := copyWSFrames(&dst, bytes.NewReader(frames), idle, reverseReplacements(subs))
+	return dst.Bytes(), n
+}
+
+// unmaskedTextFrame is what a server sends: RFC 6455 forbids masking server-to-client frames.
+func unmaskedTextFrame(t *testing.T, text string) []byte {
+	t.Helper()
+	var buf bytes.Buffer
+	if err := writeWSTextFrame(&buf, text, false); err != nil {
+		t.Fatalf("writeWSTextFrame: %v", err)
+	}
+	return buf.Bytes()
+}
+
+func TestWebSocketSubstitutesTextFrame(t *testing.T) {
+	frames := append(
+		maskedTextFrame(t, `{"op":2,"d":{"token":"__slack_token__","intents":513}}`),
+		maskedCloseFrame()...,
+	)
+
+	out, n := runFrameCopy(t, frames, wsSubs("__slack_token__", "xoxb-real-token"))
+	if n != 1 {
+		t.Fatalf("substituted count = %d, want 1", n)
+	}
+
+	got, err := readWSTextFrame(bytes.NewReader(out))
+	if err != nil {
+		t.Fatalf("readWSTextFrame: %v", err)
+	}
+	want := `{"op":2,"d":{"token":"xoxb-real-token","intents":513}}`
+	if got != want {
+		t.Fatalf("got %q, want %q", got, want)
+	}
+}
+
+// A rewritten client frame must carry a fresh mask, not the client's: reusing it would expose the XOR
+// relationship between the placeholder the agent sent and the real credential.
+func TestWebSocketSubstitutionRemasks(t *testing.T) {
+	frames := append(maskedTextFrame(t, "__tk__"), maskedCloseFrame()...)
+
+	out, _ := runFrameCopy(t, frames, wsSubs("__tk__", "abcdef"))
+	if len(out) < 6 {
+		t.Fatalf("output too short: %d bytes", len(out))
+	}
+	if out[1]&0x80 == 0 {
+		t.Fatal("rewritten frame lost its mask bit")
+	}
+	if bytes.Equal(out[2:6], []byte{1, 2, 3, 4}) {
+		t.Fatal("rewritten frame reused the client's masking key")
+	}
+	if got, err := readWSTextFrame(bytes.NewReader(out)); err != nil || got != "abcdef" {
+		t.Fatalf("got %q (err %v), want %q", got, err, "abcdef")
+	}
+}
+
+func TestWebSocketNoMatchPassesThrough(t *testing.T) {
+	payload := `{"op":11,"d":null}`
+	frames := append(maskedTextFrame(t, payload), maskedCloseFrame()...)
+
+	out, n := runFrameCopy(t, frames, wsSubs("__slack_token__", "real"))
+	if n != 0 {
+		t.Fatalf("substituted count = %d, want 0", n)
+	}
+	got, err := readWSTextFrame(bytes.NewReader(out))
+	if err != nil {
+		t.Fatalf("readWSTextFrame: %v", err)
+	}
+	if got != payload {
+		t.Fatalf("got %q, want %q", got, payload)
+	}
+}
+
+func TestWebSocketBinaryFramePassesThrough(t *testing.T) {
+	frames := append(
+		maskedBinaryFrame([]byte{0x01, 0x02, 0x03, 0x04}, [4]byte{5, 6, 7, 8}),
+		maskedCloseFrame()...,
+	)
+
+	out, n := runFrameCopy(t, frames, wsSubs("__token__", "real"))
+	if n != 0 {
+		t.Fatalf("substituted count = %d, want 0", n)
+	}
+	if out[0]&0x0F != 0x2 {
+		t.Fatalf("opcode = 0x%x, want binary", out[0]&0x0F)
+	}
+}
+
+func TestWebSocketPingPassesThrough(t *testing.T) {
+	frames := append(pingFrame([]byte("hello")), maskedCloseFrame()...)
+
+	out, _ := runFrameCopy(t, frames, wsSubs("__token__", "real"))
+	if out[0]&0x0F != 0x9 {
+		t.Fatalf("opcode = 0x%x, want ping", out[0]&0x0F)
+	}
+}
+
+// A fragmented text frame cannot be substituted without reassembly, so it must pass through untouched rather
+// than be rewritten in pieces.
+func TestWebSocketFragmentedTextPassesThrough(t *testing.T) {
+	// FIN clear, opcode text, masked, payload "__tk__" under mask {1,2,3,4}.
+	payload := []byte("__tk__")
+	for i := range payload {
+		payload[i] ^= []byte{1, 2, 3, 4}[i%4]
+	}
+	frag := append([]byte{0x01, byte(len(payload)) | 0x80, 1, 2, 3, 4}, payload...)
+	frames := append(frag, maskedCloseFrame()...)
+
+	out, n := runFrameCopy(t, frames, wsSubs("__tk__", "real"))
+	if n != 0 {
+		t.Fatalf("substituted count = %d, want 0", n)
+	}
+	if out[0]&0x80 != 0 {
+		t.Fatal("FIN bit was set on a forwarded fragment")
+	}
+}
+
+func TestWebSocketCloseFrameExits(t *testing.T) {
+	done := make(chan struct{})
+	go func() {
+		runFrameCopy(t, maskedCloseFrame(), wsSubs("__token__", "real"))
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("frame copier did not exit after a close frame")
+	}
+}
+
+// Substitution can push a payload across a length-encoding boundary, so the rewritten header has to switch
+// from the 7-bit form to the 16-bit form.
+func TestWebSocketLengthEncodingTransition(t *testing.T) {
+	prefix := strings.Repeat("A", 115)
+	frames := append(maskedTextFrame(t, prefix+"__tk__"), maskedCloseFrame()...)
+
+	out, _ := runFrameCopy(t, frames, wsSubs("__tk__", strings.Repeat("X", 100)))
+	if len(out) < 4 {
+		t.Fatalf("output too short: %d bytes", len(out))
+	}
+	if got := out[1] & 0x7F; got != 126 {
+		t.Fatalf("length byte = %d, want 126 (16-bit form)", got)
+	}
+	if got, want := int(binary.BigEndian.Uint16(out[2:4])), 215; got != want {
+		t.Fatalf("encoded length = %d, want %d", got, want)
+	}
+	got, err := readWSTextFrame(bytes.NewReader(out))
+	if err != nil {
+		t.Fatalf("readWSTextFrame: %v", err)
+	}
+	if got != prefix+strings.Repeat("X", 100) {
+		t.Fatal("payload after substitution is not what was expected")
+	}
+}
+
+func TestWebSocketOversizedFramePassesThrough(t *testing.T) {
+	body := []byte(strings.Repeat("A", maxWSSubstitutionPayload+100) + "__token__")
+	mask := [4]byte{1, 2, 3, 4}
+
+	var frame bytes.Buffer
+	frame.WriteByte(0x81)
+	frame.WriteByte(127 | 0x80)
+	lenBytes := make([]byte, 8)
+	binary.BigEndian.PutUint64(lenBytes, uint64(len(body)))
+	frame.Write(lenBytes)
+	frame.Write(mask[:])
+	for i := range body {
+		body[i] ^= mask[i%4]
+	}
+	frame.Write(body)
+	frame.Write(maskedCloseFrame())
+
+	out, n := runFrameCopy(t, frame.Bytes(), wsSubs("__token__", "real"))
+	if n != 0 {
+		t.Fatalf("substituted count = %d, want 0", n)
+	}
+	if out[0]&0x0F != wsOpText {
+		t.Fatalf("opcode = 0x%x, want text", out[0]&0x0F)
+	}
+	if got := out[1] & 0x7F; got != 127 {
+		t.Fatalf("length byte = %d, want 127 (frame was re-encoded)", got)
+	}
+}
+
+func TestIsWebSocketUpgrade(t *testing.T) {
+	cases := []struct {
+		name    string
+		headers map[string]string
+		want    bool
+	}{
+		{"upgrade", map[string]string{"Upgrade": "websocket", "Connection": "Upgrade"}, true},
+		{"mixed case", map[string]string{"Upgrade": "WebSocket", "Connection": "keep-alive, upgrade"}, true},
+		{"no connection token", map[string]string{"Upgrade": "websocket", "Connection": "keep-alive"}, false},
+		{"other protocol", map[string]string{"Upgrade": "h2c", "Connection": "Upgrade"}, false},
+		{"plain request", nil, false},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			r, err := http.NewRequest(http.MethodGet, "https://example.com/ws", nil)
+			if err != nil {
+				t.Fatalf("NewRequest: %v", err)
+			}
+			for k, v := range tc.headers {
+				r.Header.Set(k, v)
+			}
+			if got := isWebSocketUpgrade(r); got != tc.want {
+				t.Fatalf("isWebSocketUpgrade = %v, want %v", got, tc.want)
+			}
+		})
+	}
+}
+
+// The upgrade headers are hop-by-hop, so the strip would kill the handshake unless they are restored.
+func TestHandshakeHeadersSurviveHopByHopStrip(t *testing.T) {
+	r, err := http.NewRequest(http.MethodGet, "https://example.com/ws", nil)
+	if err != nil {
+		t.Fatalf("NewRequest: %v", err)
+	}
+	r.Header.Set("Upgrade", "websocket")
+	r.Header.Set("Connection", "Upgrade")
+	r.Header.Set("Sec-WebSocket-Key", "dGhlIHNhbXBsZSBub25jZQ==")
+	r.Header.Set("Sec-WebSocket-Version", "13")
+
+	captured := captureWebSocketHandshakeHeaders(r.Header)
+	stripHopByHopHeaders(r.Header)
+	if r.Header.Get("Upgrade") != "" {
+		t.Fatal("expected the strip to remove Upgrade")
+	}
+	restoreHeaders(r.Header, captured)
+
+	if got := r.Header.Get("Upgrade"); got != "websocket" {
+		t.Fatalf("Upgrade = %q after restore, want websocket", got)
+	}
+	if got := r.Header.Get("Sec-WebSocket-Key"); got != "dGhlIHNhbXBsZSBub25jZQ==" {
+		t.Fatalf("Sec-WebSocket-Key = %q after restore", got)
+	}
+
+	// Connection is not part of the restore set; prepareUpstream rebuilds it. See
+	// TestUpgradeConnectionHeaderIsRebuilt for why it is not carried over from the agent.
+	if got := r.Header.Get("Connection"); got != "" {
+		t.Fatalf("Connection = %q after restore, want it left for prepareUpstream to rebuild", got)
+	}
+	r.Header.Set("Connection", "Upgrade")
+	if !isWebSocketUpgrade(r) {
+		t.Fatal("the rebuilt request is not recognised as an upgrade")
+	}
+}
+
+// End to end over the plaintext forward path: the agent sends a placeholder in the handshake and in the
+// first frame, and the upstream must see the real credential in both. This is what proves the handshake
+// headers survive the hop-by-hop strip and that the hijacked conn outlives the server's read deadline.
+func TestWebSocketEndToEndBrokersHandshakeAndFrame(t *testing.T) {
+	type result struct {
+		auth  string
+		proto string
+		frame string
+	}
+	results := make(chan result, 1)
+
+	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if !isWebSocketUpgrade(r) {
+			http.Error(w, "expected a websocket upgrade", http.StatusBadRequest)
+			return
+		}
+		got := result{auth: r.Header.Get("Authorization"), proto: r.Header.Get("Sec-Websocket-Protocol")}
+
+		conn, buf, err := w.(http.Hijacker).Hijack()
+		if err != nil {
+			return
+		}
+		defer func() { _ = conn.Close() }()
+
+		if _, err := io.WriteString(conn, "HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=\r\n\r\n"); err != nil {
+			return
+		}
+		text, err := readWSTextFrame(buf)
+		if err != nil {
+			return
+		}
+		got.frame = text
+		results <- got
+	}))
+	defer upstream.Close()
+
+	u, err := url.Parse(upstream.URL)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	jwt := "test.jwt.token"
+	scope := agentScope{projectID: "proj", environment: "prod", secretPath: "/"}
+	services := []*resolvedService{{
+		id:           "svc-ws",
+		name:         "realtime",
+		hostPatterns: parseHostPatterns(u.Hostname()),
+		isEnabled:    true,
+		credentials: []resolvedCredential{
+			{role: roleHeaderRewrite, headerName: "Authorization", headerPrefix: "Bearer", value: "real_secret"},
+			{role: roleCredentialSub, placeholder: "__tk__", value: "xoxb-real", surfaces: []string{surfaceWebSocket}},
+		},
+	}}
+
+	client := newTestProxy(t, UnmatchedAllow, jwt, scope, services)
+	reader := bufio.NewReader(client)
+
+	_, err = fmt.Fprintf(client,
+		"GET http://%s/ws HTTP/1.1\r\nHost: %s\r\nProxy-Authorization: %s\r\n"+
+			"Upgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n"+
+			"Sec-WebSocket-Version: 13\r\nSec-WebSocket-Protocol: chat\r\nAuthorization: Bearer placeholder\r\n\r\n",
+		u.Host, u.Host, proxyAuthHeader("proj", "prod", "/", jwt))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	resp, err := http.ReadResponse(reader, nil)
+	if err != nil {
+		t.Fatalf("reading upgrade response: %v", err)
+	}
+	if resp.StatusCode != http.StatusSwitchingProtocols {
+		t.Fatalf("status = %d, want 101", resp.StatusCode)
+	}
+
+	if err := writeWSTextFrame(client, `{"token":"__tk__"}`, true); err != nil {
+		t.Fatalf("writing client frame: %v", err)
+	}
+
+	select {
+	case got := <-results:
+		if got.auth != "Bearer real_secret" {
+			t.Errorf("handshake Authorization = %q, want the injected credential", got.auth)
+		}
+		if got.proto != "chat" {
+			t.Errorf("Sec-WebSocket-Protocol = %q, want chat", got.proto)
+		}
+		if want := `{"token":"xoxb-real"}`; got.frame != want {
+			t.Errorf("frame = %q, want %q", got.frame, want)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("upstream never received the upgrade and frame")
+	}
+}
+
+// An upstream that refuses to upgrade must have its real response relayed, so the agent sees the actual
+// rejection rather than a proxy error.
+func TestWebSocketUpgradeRefusalIsRelayed(t *testing.T) {
+	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusUnauthorized)
+		_, _ = io.WriteString(w, "nope")
+	}))
+	defer upstream.Close()
+
+	u, err := url.Parse(upstream.URL)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	jwt := "test.jwt.token"
+	scope := agentScope{projectID: "proj", environment: "prod", secretPath: "/"}
+	client := newTestProxy(t, UnmatchedAllow, jwt, scope, nil)
+	reader := bufio.NewReader(client)
+
+	_, err = fmt.Fprintf(client,
+		"GET http://%s/ws HTTP/1.1\r\nHost: %s\r\nProxy-Authorization: %s\r\n"+
+			"Upgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n"+
+			"Sec-WebSocket-Version: 13\r\n\r\n",
+		u.Host, u.Host, proxyAuthHeader("proj", "prod", "/", jwt))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	resp, err := http.ReadResponse(reader, nil)
+	if err != nil {
+		t.Fatalf("reading response: %v", err)
+	}
+	if resp.StatusCode != http.StatusUnauthorized {
+		t.Fatalf("status = %d, want 401 relayed from upstream", resp.StatusCode)
+	}
+	body, _ := io.ReadAll(resp.Body)
+	_ = resp.Body.Close()
+	if string(body) != "nope" {
+		t.Fatalf("body = %q, want the upstream body", string(body))
+	}
+}
+
+// The wss path: CONNECT, MITM TLS terminate, then an upgrade inside the tunnel. This is the shape a real
+// agent uses, and it is the path where the inner tunnel server's ReadTimeout would otherwise kill the
+// connection after 60s, so it covers the hijack deadline reset as well as TLS to the upstream.
+func TestWebSocketOverConnectTunnel(t *testing.T) {
+	frames := make(chan string, 1)
+
+	upstream := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if !isWebSocketUpgrade(r) {
+			http.Error(w, "expected an upgrade", http.StatusBadRequest)
+			return
+		}
+		conn, buf, err := w.(http.Hijacker).Hijack()
+		if err != nil {
+			return
+		}
+		defer func() { _ = conn.Close() }()
+		if _, err := io.WriteString(conn, "HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=\r\n\r\n"); err != nil {
+			return
+		}
+		text, err := readWSTextFrame(buf)
+		if err != nil {
+			return
+		}
+		frames <- text
+		// The return direction is a separate copier: a client that never receives frames is a broken WebSocket
+		// even when the outbound substitution is perfect.
+		_ = writeWSTextFrame(conn, "reply-from-upstream", false)
+		// Hold the handler open so the reply is not raced by the deferred close.
+		time.Sleep(2 * time.Second)
+	}))
+	defer upstream.Close()
+
+	u, err := url.Parse(upstream.URL)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	jwt := "test.jwt.token"
+	scope := agentScope{projectID: "proj", environment: "prod", secretPath: "/"}
+	services := []*resolvedService{{
+		id:           "svc-wss",
+		name:         "realtime",
+		hostPatterns: parseHostPatterns(u.Hostname()),
+		isEnabled:    true,
+		credentials: []resolvedCredential{
+			{role: roleCredentialSub, placeholder: "__tk__", value: "xoxb-real", surfaces: []string{surfaceWebSocket}},
+		},
+	}}
+
+	ca, interCert := newTestCA(t)
+	cache := newAgentCache(func() string { return "" }, newLeaseStore(func() string { return "" }))
+	cache.entries[cacheKey(jwt, scope)] = &agentEntry{jwt: jwt, scope: scope, services: services, lastSeen: time.Now()}
+
+	// The proxy has to trust the test upstream's self-signed cert, which is exactly the transport TLS config
+	// the WebSocket dial now reads instead of hardcoding system roots.
+	upstreamRoots := x509.NewCertPool()
+	upstreamRoots.AddCert(upstream.Certificate())
+	ps := &proxyServer{
+		opts:      Options{UnmatchedHost: UnmatchedAllow},
+		ca:        ca,
+		resolver:  cache,
+		transport: &http.Transport{TLSClientConfig: &tls.Config{RootCAs: upstreamRoots}},
+	}
+
+	client, server := net.Pipe()
+	l := newOneShotListener(server)
+	srv := ps.newFrontServer()
+	srv.ConnState = func(_ net.Conn, s http.ConnState) {
+		if s == http.StateClosed || s == http.StateHijacked {
+			_ = l.Close()
+		}
+	}
+	go func() { _ = srv.Serve(l) }()
+	t.Cleanup(func() { _ = client.Close() })
+
+	_ = client.SetDeadline(time.Now().Add(15 * time.Second))
+
+	if _, err := fmt.Fprintf(client, "CONNECT %s HTTP/1.1\r\nHost: %s\r\nProxy-Authorization: %s\r\n\r\n",
+		u.Host, u.Host, proxyAuthHeader("proj", "prod", "/", jwt)); err != nil {
+		t.Fatal(err)
+	}
+	established := "HTTP/1.1 200 Connection Established\r\n\r\n"
+	buf := make([]byte, len(established))
+	if _, err := io.ReadFull(client, buf); err != nil {
+		t.Fatalf("reading CONNECT response: %v", err)
+	}
+	if string(buf) != established {
+		t.Fatalf("CONNECT response = %q", buf)
+	}
+
+	mitmRoots := x509.NewCertPool()
+	mitmRoots.AddCert(interCert)
+	tlsClient := tls.Client(client, &tls.Config{ServerName: u.Hostname(), RootCAs: mitmRoots})
+	if err := tlsClient.Handshake(); err != nil {
+		t.Fatalf("MITM TLS handshake: %v", err)
+	}
+
+	// permessage-deflate is offered here deliberately: the proxy must drop it, or the upstream could negotiate
+	// compression and the substitution below would silently never fire.
+	if _, err := fmt.Fprintf(tlsClient,
+		"GET /ws HTTP/1.1\r\nHost: %s\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n"+
+			"Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\nSec-WebSocket-Version: 13\r\n"+
+			"Sec-WebSocket-Extensions: permessage-deflate\r\n\r\n", u.Host); err != nil {
+		t.Fatal(err)
+	}
+
+	resp, err := http.ReadResponse(bufio.NewReader(tlsClient), nil)
+	if err != nil {
+		t.Fatalf("reading upgrade response: %v", err)
+	}
+	if resp.StatusCode != http.StatusSwitchingProtocols {
+		t.Fatalf("status = %d, want 101", resp.StatusCode)
+	}
+	if ext := resp.Header.Get("Sec-Websocket-Extensions"); strings.Contains(ext, "permessage-deflate") {
+		t.Fatalf("compression was negotiated (%q); substitution would be silently skipped", ext)
+	}
+
+	if err := writeWSTextFrame(tlsClient, `{"token":"__tk__"}`, true); err != nil {
+		t.Fatalf("writing frame: %v", err)
+	}
+
+	select {
+	case got := <-frames:
+		if want := `{"token":"xoxb-real"}`; got != want {
+			t.Fatalf("upstream frame = %q, want %q", got, want)
+		}
+	case <-time.After(10 * time.Second):
+		t.Fatal("upstream never received the substituted frame")
+	}
+
+	// The return direction: frames from the upstream must reach the client untouched.
+	back, err := readWSTextFrame(tlsClient)
+	if err != nil {
+		t.Fatalf("reading the upstream's reply: %v", err)
+	}
+	if back != "reply-from-upstream" {
+		t.Fatalf("reply = %q, want %q", back, "reply-from-upstream")
+	}
+}
+
+// Every token in Connection is hop-by-hop to the upstream, so forwarding the agent's own value would let it
+// name the injected credential and have the service strip it before seeing it.
+func TestUpgradeConnectionHeaderIsRebuilt(t *testing.T) {
+	jwt := "test.jwt.token"
+	scope := agentScope{projectID: "proj", environment: "prod", secretPath: "/"}
+	services := []*resolvedService{{
+		name:         "realtime",
+		hostPatterns: parseHostPatterns("example.com"),
+		isEnabled:    true,
+		credentials: []resolvedCredential{
+			{role: roleHeaderRewrite, headerName: "Authorization", headerPrefix: "Bearer", value: "real_secret"},
+		},
+	}}
+
+	cache := newAgentCache(func() string { return "" }, newLeaseStore(func() string { return "" }))
+	cache.entries[cacheKey(jwt, scope)] = &agentEntry{jwt: jwt, scope: scope, services: services, lastSeen: time.Now()}
+	ps := &proxyServer{opts: Options{UnmatchedHost: UnmatchedAllow}, resolver: cache, transport: &http.Transport{}}
+
+	r, err := http.NewRequest(http.MethodGet, "http://example.com/ws", nil)
+	if err != nil {
+		t.Fatalf("NewRequest: %v", err)
+	}
+	r.Header.Set("Upgrade", "websocket")
+	// A hostile value: naming Authorization would have the upstream drop the injected credential.
+	r.Header.Set("Connection", "Upgrade, Authorization")
+	r.Header.Set("Sec-WebSocket-Key", "dGhlIHNhbXBsZSBub25jZQ==")
+	r.Header.Set("Sec-WebSocket-Version", "13")
+
+	if _, _, err := ps.prepareUpstream(r, "http", "example.com", "80", jwt, scope); err != nil {
+		t.Fatalf("prepareUpstream: %v", err)
+	}
+
+	if got := r.Header.Get("Connection"); got != "Upgrade" {
+		t.Fatalf("Connection = %q, want exactly %q", got, "Upgrade")
+	}
+	if got := r.Header.Get("Upgrade"); got != "websocket" {
+		t.Fatalf("Upgrade = %q, want websocket", got)
+	}
+	if got := r.Header.Get("Authorization"); got != "Bearer real_secret" {
+		t.Fatalf("Authorization = %q, want the injected credential", got)
+	}
+	if got := r.Header.Get("Sec-WebSocket-Key"); got == "" {
+		t.Fatal("Sec-WebSocket-Key was lost, the handshake would fail")
+	}
+}
+
+// The agent chooses where its placeholder goes, so it can plant one in a field the service echoes and read the
+// real credential out of the reply. The return path has to swap it back.
+func TestWebSocketReverseSubstitutionRedactsReflectedCredential(t *testing.T) {
+	reply := `{"correlation_id":"xoxb-real","error":"bad token xoxb-real"}`
+	frames := append(unmaskedTextFrame(t, reply), []byte{0x88, 0x00}...)
+
+	out, n := runReverseFrameCopy(t, frames, wsSubs("__tk__", "xoxb-real"))
+	if n != 1 {
+		t.Fatalf("redacted count = %d, want 1", n)
+	}
+	got, err := readWSTextFrame(bytes.NewReader(out))
+	if err != nil {
+		t.Fatalf("readWSTextFrame: %v", err)
+	}
+	if strings.Contains(got, "xoxb-real") {
+		t.Fatalf("the real credential reached the agent: %q", got)
+	}
+	if want := `{"correlation_id":"__tk__","error":"bad token __tk__"}`; got != want {
+		t.Fatalf("got %q, want %q", got, want)
+	}
+}
+
+// A server frame must stay unmasked after being rewritten, or the client will reject it.
+func TestWebSocketReverseSubstitutionKeepsServerFramesUnmasked(t *testing.T) {
+	frames := append(unmaskedTextFrame(t, "token xoxb-real"), []byte{0x88, 0x00}...)
+
+	out, _ := runReverseFrameCopy(t, frames, wsSubs("__tk__", "xoxb-real"))
+	if out[1]&0x80 != 0 {
+		t.Fatal("rewritten server frame was masked")
+	}
+}
+
+func TestWebSocketReverseSubstitutionLeavesUnrelatedFrames(t *testing.T) {
+	frames := append(unmaskedTextFrame(t, `{"ok":true}`), []byte{0x88, 0x00}...)
+
+	out, n := runReverseFrameCopy(t, frames, wsSubs("__tk__", "xoxb-real"))
+	if n != 0 {
+		t.Fatalf("redacted count = %d, want 0", n)
+	}
+	got, err := readWSTextFrame(bytes.NewReader(out))
+	if err != nil || got != `{"ok":true}` {
+		t.Fatalf("got %q (err %v)", got, err)
+	}
+}
+
+// Traffic either way is activity. A per-direction deadline would tear down a subscribe-mostly stream that
+// receives for hours while sending nothing.
+func TestWSIdleExtendsBothDirections(t *testing.T) {
+	a, b := &deadlineRecorder{}, &deadlineRecorder{}
+	idle := &wsIdle{conns: [2]net.Conn{a, b}}
+
+	idle.extend()
+	if a.calls != 1 || b.calls != 1 {
+		t.Fatalf("extend touched %d and %d deadlines, want 1 each", a.calls, b.calls)
+	}
+	if a.last.IsZero() || b.last.IsZero() {
+		t.Fatal("extend did not set a deadline on both conns")
+	}
+	if a.last.Before(time.Now().Add(wsIdleTimeout / 2)) {
+		t.Fatalf("deadline %v is not roughly wsIdleTimeout out", a.last)
+	}
+}
+
+type deadlineRecorder struct {
+	wsFakeConn
+	calls int
+	last  time.Time
+}
+
+func (d *deadlineRecorder) SetReadDeadline(t time.Time) error {
+	d.calls++
+	d.last = t
+	return nil
+}
+
+// A compressed frame carries RSV1 and is never substituted, so the offer has to be dropped or the surface
+// silently does nothing on any client that negotiates compression.
+func TestDropPerMessageDeflate(t *testing.T) {
+	cases := []struct {
+		name    string
+		offer   string
+		want    string
+		changed bool
+	}{
+		{"only deflate", "permessage-deflate", "", true},
+		{"deflate with params", "permessage-deflate; client_max_window_bits=15", "", true},
+		{"keeps other extensions", "permessage-deflate, x-custom-ext", "x-custom-ext", true},
+		{"webkit variant", "x-webkit-deflate-frame", "", true},
+		{"case insensitive", "PerMessage-Deflate", "", true},
+		{"nothing to drop", "x-custom-ext", "x-custom-ext", false},
+		{"no header", "", "", false},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			h := make(http.Header)
+			if tc.offer != "" {
+				h.Set("Sec-WebSocket-Extensions", tc.offer)
+			}
+			if got := dropPerMessageDeflate(h); got != tc.changed {
+				t.Fatalf("dropPerMessageDeflate = %v, want %v", got, tc.changed)
+			}
+			if got := h.Get("Sec-WebSocket-Extensions"); got != tc.want {
+				t.Fatalf("remaining extensions = %q, want %q", got, tc.want)
+			}
+		})
+	}
+}
+
+// A 101 carries no body, so upstream framing headers must not reach the client: they would desynchronize its
+// frame parser.
+func TestSwitchingResponseDropsFramingHeaders(t *testing.T) {
+	resp := &http.Response{
+		StatusCode: http.StatusSwitchingProtocols,
+		Status:     "101 Switching Protocols",
+		Proto:      "HTTP/1.1",
+		Header: http.Header{
+			"Sec-Websocket-Accept": []string{"s3pPLMBiTxaQ9kYGzzhZRbK+xOo="},
+			"Content-Length":       []string{"1234"},
+			"Transfer-Encoding":    []string{"chunked"},
+			"Keep-Alive":           []string{"timeout=5"},
+			"Sec-Something-Else":   []string{"smuggled"},
+			"Set-Cookie":           []string{"session=abc"},
+		},
+	}
+
+	var buf bytes.Buffer
+	if err := writeWebSocketSwitchingResponse(&buf, resp); err != nil {
+		t.Fatalf("writeWebSocketSwitchingResponse: %v", err)
+	}
+	out := buf.String()
+
+	for _, banned := range []string{"Content-Length", "Transfer-Encoding", "Keep-Alive", "Sec-Something-Else"} {
+		if strings.Contains(out, banned) {
+			t.Errorf("101 response leaked %s:\n%s", banned, out)
+		}
+	}
+	for _, required := range []string{"101 Switching Protocols", "Sec-Websocket-Accept", "Upgrade: websocket", "Set-Cookie"} {
+		if !strings.Contains(out, required) {
+			t.Errorf("101 response is missing %s:\n%s", required, out)
+		}
+	}
+}
+
+func TestWebsocketSubstitutionsSelectsBySurface(t *testing.T) {
+	creds := []resolvedCredential{
+		{role: roleCredentialSub, placeholder: "__a__", value: "a", surfaces: []string{surfaceHeader, surfaceWebSocket}},
+		{role: roleCredentialSub, placeholder: "__b__", value: "b", surfaces: []string{surfaceBody}},
+		{role: roleCredentialSub, placeholder: "__c__", value: "c", surfaces: []string{surfaceWebSocket}},
+		{role: roleHeaderRewrite, placeholder: "__d__", value: "d", surfaces: []string{surfaceWebSocket}},
+		{role: roleCredentialSub, placeholder: "", value: "e", surfaces: []string{surfaceWebSocket}},
+	}
+
+	got := websocketSubstitutions(creds)
+	if len(got) != 2 {
+		t.Fatalf("got %d websocket substitutions, want 2", len(got))
+	}
+	if got[0].placeholder != "__a__" || got[1].placeholder != "__c__" {
+		t.Fatalf("unexpected selection: %+v", got)
+	}
+	if got[0].label.Role != roleCredentialSub || len(got[0].label.Surfaces) != 1 || got[0].label.Surfaces[0] != surfaceWebSocket {
+		t.Fatalf("unexpected activity label: %+v", got[0].label)
+	}
+}