Feat: Forward WebSocket upgrades on intercepted connections

CLAUDE.md

@@ -618,11 +618,42 @@ Every event on `/v1/sessions/{id}` and `/v1/events` carries:
   disagree about the status on purpose. **A 426 with `error.kind:
   "upgrade_declined"` is not one of these:** it is the forward proxy refusing a
   WebSocket on OpenAI's Responses API (a path ending `/responses` on `chatgpt.com` or
-  `api.openai.com`; other hosts keep the strip), because the origin
-  would only answer the stripped GET with a 405, which Codex retries for ~6s, while a
-  426 sends it straight to HTTPS. Nothing is forwarded, the Finisher sees an allow with
+  `api.openai.com`) — a deliberate steer, since a 426 sends Codex straight to HTTPS
+  where its traffic stays parseable, while a forwarded WebSocket's frames are opaque
+  to every plugin. Nothing is forwarded, the Finisher sees an allow with
   status 426, and `/v1/usage` does not count it as an error. On by default;
-  `listener.decline_responses_websockets: false` forwards the upgrade instead.
+  `listener.decline_responses_websockets: false` forwards the upgrade for real
+  instead. **Every other WebSocket is forwarded** (#1488): the upgrade crosses the
+  proxy (normalized to `Connection: Upgrade` / `Upgrade: websocket`, which net/http
+  pins to HTTP/1.1 even on the bridge's h2-negotiating client), the origin's 101 is
+  hijacked and spliced, and the session records the request row plus a paired 101
+  response row carrying both directions' byte counts — the frames themselves are
+  never inspected or buffered. **An upgrade is both headers, on both sides**, and
+  since the proxy WRITES the normalized pair a loose predicate forges one rather
+  than missing one: a request needs `Connection: upgrade` as well as `Upgrade:
+  websocket` (RFC 9110 §7.8, and `Request.requiresHTTP1`'s own test) or it is
+  stripped and takes the ordinary path, and a 101 that switches to some other
+  protocol than the one requested is refused 502 before the hijack instead of
+  spliced. Either check missing turns a plain request into an opaque pipe no
+  plugin inspects. **The forward proxy's 101 row is kept out of `/v1/usage` for the
+  same reasons a tunnel row is** — it is recorded after both splice directions end,
+  so its
+  `durationMs` is the WebSocket SESSION's lifetime, and folding it as a request
+  latency moved one host's mean from 500ms to 6.4s in testing; its bytes are opaque
+  frames with no model, token or cost to attribute; and counting it would add one
+  Request whose latency is its whole lifetime. The `durationMs` stays on the
+  session timeline, where the lifetime is the feature — only the aggregation
+  misread it. **The exclusion is scoped to outbound, and a 101 is not excluded
+  everywhere.** The reverse proxy records one too, but *at handshake time* —
+  `ReverseProxy` never copies an upgraded body, so a deferred record would never
+  fire — which makes that row's `durationMs` the handshake latency rather than a
+  session lifetime. An inbound 101 is therefore an ordinary request and still
+  counts as one; excluding it on the status alone would change what `Requests`
+  means on every inbound listener. **A relay that broke mid-session carries `error.kind:
+  "proxy_error"`** on that row, so a WebSocket that died is distinguishable from
+  one that closed cleanly; an ordinary close (`net.ErrClosed`, from the direction
+  that lost the close race) is deliberately not marked, or every healthy session
+  would read as broken.
   **A 403 with `error.kind: "destination_private"` is the egress guard's
   answer, not the upstream's** (`listener.egress_guard`, off by default): the
   destination resolved into loopback, link-local, private or this host's own
@@ -675,7 +706,7 @@ Every event on `/v1/sessions/{id}` and `/v1/events` carries:
   `timing:` line.
 - `requestedHost` — present only when a plugin redirected the request (`pctx.Redirect`): the host the client asked for. `host` is where the request actually went, and usage, the cost ledger and pricing all follow `host`. agentop's detail pane shows both on a `redirected:` line.
 - `inference.requestedModel` — present only when a plugin changed the model (`pctx.SetRequestModel`): the model the client asked for. `inference.model` is the model the request was sent for, which settlement prices; agentop's detail pane shows both on a `model:` line.
-- `bytesUp`, `bytesDown` — **body** bytes, one direction per row: a request row carries `bytesUp`, the body **as forwarded** (so a `tool-prune`-style rewrite is reflected, since the count is taken after the pipeline ran), and a response row carries `bytesDown`, the body **as it came back from the destination** — not as delivered, because extproc records the row before it emits a response mutator's replacement. Headers are not counted. A tunnel close is the one row carrying both. **Zero means the listener counted nothing, not that the body was empty**, and there is no third value for "unknown": every listener buffers bodies only when a plugin asks, so an unbuffered request reports zero. A response relayed straight through is counted, not zero — both proxies record the row after the copy (#1251's fix), so the bytes they relayed through their own process are on it. The one arm that still reports ↓0 is extproc with a body-free pipeline: it asked Envoy for body mode NONE, so the bytes it never saw are bytes it asked not to see — `core/listener/parity`'s `TestDivergence_InboundResponseBodyCountWhenNoPluginReadsIt` pins that disagreement with its reason. agentop renders a zero side blank rather than as `0B`, and that is why its BYTES column can be trusted as a size when it shows one. Everywhere else the three listeners count the same thing for the same response, which took some doing: extproc sums the chunks Envoy hands it, while the proxies count off the upstream body at one choke point each, so every arm — buffered, block-by-block SSE, or chunk copy — reports the same figure. `core/listener/internal/bodycount` and `core/listener/parity`'s bytes fixtures hold that line.
+- `bytesUp`, `bytesDown` — **body** bytes, one direction per row: a request row carries `bytesUp`, the body **as forwarded** (so a `tool-prune`-style rewrite is reflected, since the count is taken after the pipeline ran), and a response row carries `bytesDown`, the body **as it came back from the destination** — not as delivered, because extproc records the row before it emits a response mutator's replacement. Headers are not counted. A tunnel close and a forwarded WebSocket's 101 response row are the rows carrying both. **Zero means the listener counted nothing, not that the body was empty**, and there is no third value for "unknown": every listener buffers bodies only when a plugin asks, so an unbuffered request reports zero. A response relayed straight through is counted, not zero — both proxies record the row after the copy (#1251's fix), so the bytes they relayed through their own process are on it. The one arm that still reports ↓0 is extproc with a body-free pipeline: it asked Envoy for body mode NONE, so the bytes it never saw are bytes it asked not to see — `core/listener/parity`'s `TestDivergence_InboundResponseBodyCountWhenNoPluginReadsIt` pins that disagreement with its reason. agentop renders a zero side blank rather than as `0B`, and that is why its BYTES column can be trusted as a size when it shows one. Everywhere else the three listeners count the same thing for the same response, which took some doing: extproc sums the chunks Envoy hands it, while the proxies count off the upstream body at one choke point each, so every arm — buffered, block-by-block SSE, or chunk copy — reports the same figure. `core/listener/internal/bodycount` and `core/listener/parity`'s bytes fixtures hold that line.
 - `tunnel`, `tunnelReason` — an opaque CONNECT (or transparent-redirect) tunnel records two rows sharing a `requestId`: the open (`phase: "request"`, `tunnelReason` saying why the bytes stayed opaque) and, when the tunnel ends, the close (`phase: "response"`). The close carries the CONNECT's own `statusCode` — 200, or 502 with the dial error in `error` when the destination could not be reached and no TLS bridge could take the connection (`tunnelReason: "dial-failed"`; where the bridge is enabled for the port the CONNECT is answered 200 first, because a redirect can still apply, and a bridge that then declines closes with that 200 and the dial error) — plus `durationMs` for how long the tunnel stayed open and, in `bytesUp`/`bytesDown`, the bytes it carried each way (up = client to destination; opaque bytes have no body/header split, so these are everything that crossed). It is not the destination's status: that travels inside the client's end-to-end TLS. A bridged tunnel's open is recorded with its first decrypted request — in that request's session, directly before it, stamped with its time — and records no close, because the request carries its own response; a bridged tunnel that recorded no request gets its open and a close when it ends. Tunnel rows are kept out of `/v1/usage`: a tunnel's lifetime is not a request latency. `tunnelReason: "dest-private"` with a 403 close and `error.kind: "destination_private"` is the egress guard refusing the destination before any dial — never confused with `dial-failed`, which attempted.
 - `dialAddr` — the address the egress guard checked and the proxy dialed, `"ip:port"`, on a tunnel's open row when `listener.egress_guard` is on. With `host` it is the audit trail against DNS rebinding: under the guard the proxy resolves once, checks that answer, and dials exactly it, so the two fields together say "asked for the name, reached this address". Absent when the guard is off and on rows that dialed nothing.
 - `httpMethod`, `httpPath` — the HTTP verb and path, so a request no parser recognized is still identifiable rather than showing only a host. Distinct from the `method` inside `a2a` / `mcp`, which is a protocol method name. On an opaque tunnel `httpMethod` is `CONNECT` and `httpPath` is absent — opaque bytes carry no request line. The path is query-stripped and percent-decoded, so query-borne credentials never reach the timeline, but a secret in a path *segment* (a bot token, a webhook path) does survive on this unauthenticated surface — worth knowing before exporting events off-box.

core/config/config.go

@@ -1045,8 +1045,8 @@ type ListenerConfig struct {
 
 	// DeclineResponsesWebSockets: whether the forward proxy answers a WebSocket
 	// upgrade on an OpenAI Responses API path (one ending /responses, on chatgpt.com
-	// or api.openai.com) with 426 itself, instead of forwarding the request with its Upgrade stripped. nil means
-	// unset, and the default is on; explicit false restores the forwarding. See
+	// or api.openai.com) with 426 itself, instead of forwarding the upgrade. nil
+	// means unset, and the default is on; explicit false forwards it. See
 	// DeclinesResponsesWebSockets for why on is the default.
 	//
 	// A pointer for the reason SessionConfig.Enabled is one: with a plain bool,
@@ -1100,13 +1100,14 @@ type EgressGuardConfig struct {
 // DeclinesResponsesWebSockets reports whether the forward proxy answers a WebSocket
 // upgrade on a Responses API path with 426 rather than forwarding it.
 //
-// On unless set false, in every deployment. The proxy must strip Upgrade (RFC 9110
-// makes it hop-by-hop), so a forwarded upgrade reaches the origin as a bare GET, and
-// OpenAI's Responses endpoint answers that 405. Codex reads a 405 as transient and
-// retries for about six seconds before falling back to HTTPS; it reads a 426 as "use
-// HTTP" and falls back at once. The forwarded request can never become a WebSocket, so
-// nothing is lost by answering it here. False is the escape hatch for a client that
-// reads 426 some other way.
+// On unless set false, in every deployment. The proxy CAN forward a WebSocket (it
+// does for every other host, #1488), but a Responses session on one is opaque to
+// every parser — no token counts, no cost — so the 426 keeps Codex on HTTPS where
+// its traffic stays observable. That, and history: before forwarding existed the
+// stripped handshake drew a 405 that Codex retried for about six seconds before
+// falling back to HTTPS, while it reads a 426 as "use HTTP" and falls back at once.
+// False is the escape hatch for a client that reads 426 some other way — or wants
+// the WebSocket for real.
 func (l ListenerConfig) DeclinesResponsesWebSockets() bool {
 	return l.DeclineResponsesWebSockets == nil || *l.DeclineResponsesWebSockets
 }

core/cost/usage/replay_test.go

@@ -2,6 +2,7 @@ package usage
 
 import (
 	"encoding/json"
+	"net/http"
 	"reflect"
 	"strings"
 	"testing"
@@ -154,3 +155,45 @@ func TestAbandonReplay_StopsLogging(t *testing.T) {
 		t.Fatalf("after AbandonReplay: replaying=%v, %d ops logged", live.replaying, len(live.replayOps))
 	}
 }
+
+// A 101 row is excluded on the REPLAY path too, which only holds because
+// ReplayCopy is a whole-struct copy (`c := *e`) and so carries every field the
+// exclusion reads. There are THREE: StatusCode and Tunnel, and — since the
+// exclusion was scoped to outbound — Direction. If ReplayCopy ever became a
+// field-by-field copy that dropped or mangled any of them, the exclusion would
+// silently stop applying to archived WebSocket sessions and their lifetimes
+// would reappear in the latency mean on every restart — see
+// TestRecord_WebSocket101IsNotARequestLatency for what that costs.
+//
+// The first two are asserted directly below; Direction is covered by the
+// end-to-end half, which is why that half is not redundant: mangling Direction
+// to Inbound moves the mean from 500ms to 300s and fails there.
+func TestReplayCopy_KeepsWhatExcludesA101Row(t *testing.T) {
+	now := time.Date(2026, 10, 5, 12, 0, 0, 0, time.UTC)
+	ws := respEvent(now, http.StatusSwitchingProtocols, 10*time.Minute, "", 0)
+	if c := ReplayCopy(ws); c.StatusCode != http.StatusSwitchingProtocols {
+		t.Fatalf("ReplayCopy dropped StatusCode (%d); the 101 exclusion in Record keys on it",
+			c.StatusCode)
+	}
+	tun := respEvent(now, 200, 4*time.Minute, "", 0)
+	tun.Tunnel = true
+	if c := ReplayCopy(tun); !c.Tunnel {
+		t.Fatal("ReplayCopy dropped Tunnel; the tunnel exclusion in Record keys on it")
+	}
+
+	// And end to end through the ring: a replayed 101 changes nothing.
+	a := New(WithClock(func() time.Time { return now }))
+	a.Record("s1", respEvent(now, 200, 500*time.Millisecond, "", 0))
+	before := a.Snapshot(time.Minute, BucketWidth, "", GroupNone)
+	c := ReplayCopy(ws)
+	a.Record("s1", &c)
+	after := a.Snapshot(time.Minute, BucketWidth, "", GroupNone)
+	if after.Totals.Requests != before.Totals.Requests {
+		t.Errorf("requests = %d, want %d — a replayed 101 must not count either",
+			after.Totals.Requests, before.Totals.Requests)
+	}
+	if weightedLatMean(after) != weightedLatMean(before) {
+		t.Errorf("latency mean moved %.1f -> %.1f on replay",
+			weightedLatMean(before), weightedLatMean(after))
+	}
+}

core/cost/usage/usage.go

@@ -16,6 +16,7 @@ package usage
 import (
 	"math"
 	"net"
+	"net/http"
 	"sort"
 	"strconv"
 	"strings"
@@ -1056,7 +1057,37 @@ func (a *Aggregator) Record(sessionID string, e *pipeline.SessionEvent) {
 	// were opaque, so there is no model, token or cost to attribute. Counting it would
 	// also quietly change what Requests means for every reader of /v1/usage. A bridged
 	// tunnel's decrypted requests are ordinary events and are counted as usual.
-	if e.Tunnel {
+	//
+	// A FORWARDED WEBSOCKET'S 101 ROW IS THE SAME SHAPE, and is excluded for the
+	// same three reasons (#1488, #1549 review). It is recorded after both splice
+	// directions end, so its Duration is the WebSocket SESSION's lifetime, not a
+	// request latency: measured, one 10-minute WS session against a host with a
+	// hundred 500ms requests moved that host's latency mean from 500ms to 6.4s
+	// (TestRecord_WebSocket101IsNotARequestLatency). Its bytes are opaque frames
+	// with no model, token or cost to attribute. And counting it would make
+	// Requests mean something new, inconsistently: the upgrade GET's own request
+	// row is already not counted (Invocations nil), so one WebSocket would add
+	// exactly one Request whose latency is its whole lifetime.
+	//
+	// Keyed on the status rather than a new event field because 101 IS the
+	// condition on the forward proxy — it records that status for no other
+	// reason, and a 101 that reached a client always means a spliced upgrade
+	// (the relay refuses any other switched protocol before the hijack). The row
+	// keeps its durationMs on the session timeline, where the WebSocket's
+	// lifetime is the feature; only the aggregation misread it.
+	//
+	// SCOPED TO OUTBOUND, because the status alone does not identify the
+	// condition across listeners. The reverse proxy records a 101 too
+	// (ModifyResponse's StatusSwitchingProtocols arm, Direction: Inbound) and
+	// records it AT HANDSHAKE TIME — ReverseProxy never copies an upgraded body,
+	// so a deferred record would never fire — which makes that row's Duration
+	// the handshake latency, not a session lifetime. It is an ordinary request
+	// with an ordinary latency, and excluding it would change what Requests
+	// means on every inbound listener: the same silent redefinition this comment
+	// warns about, in the other direction. Pinned both ways by
+	// TestRecord_WebSocket101IsNotARequestLatency and
+	// TestRecord_InboundWebSocket101IsStillARequest (#1549 review).
+	if e.Tunnel || (e.Direction == pipeline.Outbound && e.StatusCode == http.StatusSwitchingProtocols) {
 		return
 	}