From a870e3b7c028b8129368afe1b114eaafbccef150 Mon Sep 17 00:00:00 2001 From: Mariusz Sabath Date: Wed, 28 Oct 2026 15:41:20 -0400 Subject: [PATCH 1/4] fix: Answer both halves of the long-duration question (#1251) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The sorting half of #1251 landed in #1371; this addresses the two it left open. The 30-second 502s: connectDialTimeout bounded the CONNECT tunnel's upstream TCP dial at 30s, and the only dials that wait it out are black-holed ones — a refused destination answers a SYN with an RST in well under a second, while a host that silently drops it (off the VPN that reaches it, behind a dead NAT) consumes the whole budget and the agent spends it staring at nothing. By 10s the initial SYN and its first four scheduled retransmits (1s, 2s, 4s, 8s) have all gone unanswered, so the remaining 20s bought no connection, only latency before the 502. 10s also matches upstreamVerifyTimeout and the mTLS dialer, which had settled on it independently. The unexplained 11-minute 200s: durationMs on a response row measures the whole exchange, so an operator sorting by duration could not tell a slow-to-answer origin from one that answered and then streamed slowly. Response rows now carry ttfbMs — request entry to the upstream's response HEADERS arriving — and durationMs − ttfbMs is the body's share, the stream time for a streamed completion. All three listeners stamp pctx.ResponseHeadersAt at their own headers-arrival moment (client.Do returning for the proxy listeners, Envoy's response-headers callback for extproc) and every response-event builder derives the figure the same way (pipeline.DurationBetween), so it means one thing wherever a row reports it. Absent when the exchange failed before headers, on request rows and on tunnel rows; omitempty keeps the wire, the session archive and an older agentop unaffected. agentop's detail pane renders the split as a timing: line. Assisted-By: Claude (Anthropic AI) Signed-off-by: Mariusz Sabath --- CLAUDE.md | 10 ++ cmd/agentop/tui/detail_pane.go | 24 ++++- cmd/agentop/tui/detail_pane_test.go | 31 ++++++ cmd/agentop/tui/events_pane.go | 13 ++- core/listener/extproc/server.go | 8 ++ core/listener/forwardproxy/server.go | 23 ++++- core/listener/forwardproxy/server_test.go | 114 ++++++++++++++++++++++ core/listener/reverseproxy/server.go | 10 ++ core/pipeline/context.go | 13 +++ core/pipeline/session.go | 29 +++++- core/pipeline/session_test.go | 18 +++- core/pipeline/snapshot.go | 15 +++ docs/laptop-service.md | 2 +- docs/session-api.md | 8 ++ 14 files changed, 304 insertions(+), 14 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 9a0889c3c..1a29adbf9 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -616,6 +616,16 @@ Every event on `/v1/sessions/{id}` and `/v1/events` carries: open and a close saying 200. An opaque CONNECT that could not be dialed reports the same situation one layer down as `tunnelReason: "dial-failed"` with `error.kind: "dial_failed"`. +- `ttfbMs` — on a delivered response row, the wall time from request entry into the + listener to the upstream's response **headers** arriving, so `durationMs` − `ttfbMs` + is how long the body took — for a streamed completion, the stream time. The pair + separates a slow-to-answer origin from one that answered and then trickled, which + `durationMs` alone cannot (#1251). Absent when the exchange failed before headers, on + request rows, and on tunnel rows (an opaque tunnel has no headers to time). All three + listeners set `pctx.ResponseHeadersAt` at their own headers-arrival moment — `client.Do` + returning (forward and reverse proxy), Envoy's response-headers callback (extproc) — + and every response-event builder derives the figure the same way + (`pipeline.DurationBetween`). agentop's detail pane renders the split as a `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, and so does a response relayed straight through — both proxies copy that one after the row is appended. 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. All 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. diff --git a/cmd/agentop/tui/detail_pane.go b/cmd/agentop/tui/detail_pane.go index 036073bdc..1eadd3270 100644 --- a/cmd/agentop/tui/detail_pane.go +++ b/cmd/agentop/tui/detail_pane.go @@ -50,7 +50,7 @@ func (m *model) showDetail(r eventRow, resetScroll bool) { if r.tunnel != nil { tunnel = tunnelHeader(r.tunnel) } - for _, h := range []string{tlsHeader(e.TLS), rewriteHeader(e), tunnel} { + for _, h := range []string{tlsHeader(e.TLS), rewriteHeader(e), timingHeader(e), tunnel} { if h != "" { lines = append(lines, strings.Split(h, "\n")...) lines = append(lines, "") @@ -75,6 +75,28 @@ func (m *model) showDetail(r eventRow, resetScroll bool) { m.detailVp.SetYOffset(m.detailVp.YOffset) } +// timingHeader decomposes a response row's duration the way #1251 asked for: +// how long the origin took to answer (ttfb) against how long the body then +// took. Duration alone cannot tell an 11-minute response with 217 tokens +// apart from a fast one — the two need different fixes, and the reader +// sorting by duration is asking exactly that question. The body's share is +// Duration − Ttfb rather than a third measurement: the row already holds +// both ends of the span. Response rows only, and only when ttfb exists — a +// failed exchange has no headers to time, and a request row no duration. +// +// timing: ttfb 412ms · body 10m32s of 11m07s total +func timingHeader(e *pipeline.SessionEvent) string { + if e.Phase != pipeline.SessionResponse || e.Ttfb <= 0 || e.Duration == 0 { + return "" + } + body := e.Duration - e.Ttfb + if body < 0 { + body = 0 + } + return fmt.Sprintf("timing: ttfb %s · body %s of %s total", + formatDuration(e.Ttfb), formatDuration(body), formatDuration(e.Duration)) +} + // tunnelHeader summarizes the CONNECT tunnel folded into a TLS-bridged // request row: the bridged origin (host:port) and any gate invocations that // ran on the tunnel-open before the bytes were decrypted. Lets an operator diff --git a/cmd/agentop/tui/detail_pane_test.go b/cmd/agentop/tui/detail_pane_test.go index 9608deed3..a34b939c7 100644 --- a/cmd/agentop/tui/detail_pane_test.go +++ b/cmd/agentop/tui/detail_pane_test.go @@ -4,6 +4,7 @@ import ( "encoding/json" "strings" "testing" + "time" "github.com/rossoctl/cortex/core/cost/event" "github.com/rossoctl/cortex/core/pipeline" @@ -244,6 +245,36 @@ func TestRewriteHeader_StacksTheRedirectAndTheModel(t *testing.T) { } } +// The #1251 shape: an 11-minute response whose ttfb is small says "the origin +// answered, then the stream trickled" — the header exists to make that legible +// without dividing two millisecond figures in one's head. +func TestTimingHeader_SplitsDurationIntoTtfbAndBody(t *testing.T) { + got := timingHeader(&pipeline.SessionEvent{ + Phase: pipeline.SessionResponse, + Duration: 667 * time.Second, + Ttfb: 5 * time.Second, + }) + want := "timing: ttfb 5.00s · body 11m02s of 11m07s total" + if got != want { + t.Errorf("timingHeader = %q, want %q", got, want) + } +} + +// No header where the decomposition would be fiction: a request row (no +// duration), a failed exchange (no ttfb — nothing answered), and a tunnel +// close (no headers to time). +func TestTimingHeader_EmptyWhereTheFiguresAreNot(t *testing.T) { + for _, e := range []*pipeline.SessionEvent{ + {Phase: pipeline.SessionRequest, Duration: time.Second, Ttfb: 100 * time.Millisecond}, + {Phase: pipeline.SessionResponse, Duration: 30 * time.Second}, + {Phase: pipeline.SessionResponse}, + } { + if got := timingHeader(e); got != "" { + t.Errorf("timingHeader = %q for phase %v duration %v ttfb %v, want empty", got, e.Phase, e.Duration, e.Ttfb) + } + } +} + // The JSON below the header keeps requestedModel on both phases, beside model. func TestFilterForDetail_KeepsRequestedModel(t *testing.T) { wire := []byte(`{"inference":{"model":"glm-5.3","requestedModel":"claude-opus-5-5","completion":"hi"}}`) diff --git a/cmd/agentop/tui/events_pane.go b/cmd/agentop/tui/events_pane.go index 5cee95318..fd2aebfeb 100644 --- a/cmd/agentop/tui/events_pane.go +++ b/cmd/agentop/tui/events_pane.go @@ -6,6 +6,7 @@ import ( "slices" "sort" "strings" + "time" "github.com/charmbracelet/lipgloss" @@ -1021,10 +1022,18 @@ func formatBytes(n int64) string { // Seconds hand over to minutes at 59.995s rather than 60s, because %.2f rounds // 59.999s up to "60.00s", a seconds figure that is already a minute. func durationCell(e pipeline.SessionEvent) string { - if e.Duration == 0 { + return formatDuration(e.Duration) +} + +// formatDuration is durationCell's body, split out so the detail pane's timing +// header can format ttfb and the body's share of a response at the same +// precision the DURATION column uses — one rule for the same unit, everywhere +// a reader sees it. +func formatDuration(d time.Duration) string { + if d == 0 { return "" } - ms := e.Duration.Milliseconds() + ms := d.Milliseconds() switch { case ms < 1000: return fmt.Sprintf("%dms", ms) diff --git a/core/listener/extproc/server.go b/core/listener/extproc/server.go index e9e05c547..f1867fe28 100644 --- a/core/listener/extproc/server.go +++ b/core/listener/extproc/server.go @@ -456,6 +456,7 @@ func (s *Server) recordInboundResponseSession(pctx *pipeline.Context) { HTTPMethod: pctx.Method, HTTPPath: pctx.Path, Duration: pipeline.DurationSince(pctx.StartedAt), + Ttfb: pipeline.DurationBetween(pctx.StartedAt, pctx.ResponseHeadersAt), Client: pctx.ClientInfo(), Subagent: pctx.Subagent(), ParentSubagent: pctx.ParentSubagent(), @@ -492,6 +493,7 @@ func (s *Server) recordOutboundResponseSession(pctx *pipeline.Context) { HTTPMethod: pctx.Method, HTTPPath: pctx.Path, Duration: pipeline.DurationSince(pctx.StartedAt), + Ttfb: pipeline.DurationBetween(pctx.StartedAt, pctx.ResponseHeadersAt), Client: pctx.ClientInfo(), Subagent: pctx.Subagent(), ParentSubagent: pctx.ParentSubagent(), @@ -665,6 +667,12 @@ func (s *Server) handleResponseHeaders(ctx context.Context, headers *corev3.Head statusStr := getHeader(headers, ":status") pctx.StatusCode, _ = strconv.Atoi(statusStr) pctx.ResponseHeaders = headerMapToHTTP(headers) + // Envoy delivering response headers is this listener's headers-arrival + // moment — the same point in the exchange where the proxy listeners set + // theirs — so ttfb means one thing on a row from any listener (#1251). + // The body phase and end-of-stream recording run later, which is what + // makes this headers-arrival rather than body-completion. + pctx.ResponseHeadersAt = time.Now() p := s.OutboundPipeline if direction == "inbound" { diff --git a/core/listener/forwardproxy/server.go b/core/listener/forwardproxy/server.go index ca41b2603..538eab737 100644 --- a/core/listener/forwardproxy/server.go +++ b/core/listener/forwardproxy/server.go @@ -577,6 +577,14 @@ func (s *Server) serveOutbound(w http.ResponseWriter, r *http.Request, tl *tunne // Response phase: populate pctx and run plugins in reverse order. pctx.StatusCode = resp.StatusCode pctx.ResponseHeaders = resp.Header.Clone() + // Taken HERE and not where client.Do returned, because a resend + // (resendIfAsked above) discards that response and re-issues the exchange: + // the figure the response row should carry is for the response the client + // actually got. Both proxy listeners measure at the same point in their + // own flow — the moment the delivered response's headers are in hand — + // so ttfb means one thing wherever a row reports it. Stays zero on the + // error path above, where no headers ever arrived (#1251). + pctx.ResponseHeadersAt = time.Now() // BytesDown for the response row (#1309), counted HERE and nowhere else. Every // arm below reads this one body — buffered whole, relayed SSE block by block, relayed @@ -1654,6 +1662,7 @@ func (s *Server) recordOutboundResponseEvent(pctx *pipeline.Context, statusCode StatusCode: statusCode, Error: pipeline.EventErrorOr(pctx, fail), Duration: pipeline.DurationSince(pctx.StartedAt), + Ttfb: pipeline.DurationBetween(pctx.StartedAt, pctx.ResponseHeadersAt), Client: pctx.ClientInfo(), Subagent: pctx.Subagent(), ParentSubagent: pctx.ParentSubagent(), @@ -2050,7 +2059,19 @@ func (s *Server) recordOutboundRejectIn(tl *tunnelLog, pctx *pipeline.Context, a // connectDialTimeout bounds the upstream TCP dial for a CONNECT tunnel. // Once the tunnel is open the timeout no longer applies — the agent's TLS // handshake and subsequent traffic flow at their own pace. -const connectDialTimeout = 30 * time.Second +// +// 10s, not the 30s it long was, because the only dials that wait it out are +// black-holed ones (#1251): a refused or unroutable destination answers a TCP +// SYN with an RST in well under a second, while a host that drops the SYN +// silently — off the VPN that reaches it, behind a dead NAT — consumes the +// whole budget and the agent spends it staring at nothing. By 10s the initial +// SYN and its first four scheduled retransmits (1s, 2s, 4s, 8s) have all gone +// unanswered; a handshake that misses five straight tries is not being saved +// by a sixth at 16s, so the rest of the budget bought no connection, only +// latency before the 502. Matches the budget the bridge already trusts for +// its own upstream probe (upstreamVerifyTimeout) and the mTLS dialer, which +// had settled on 10s independently. +const connectDialTimeout = 10 * time.Second // handleConnect tunnels HTTPS (and any other TLS-wrapped protocol) through // the forward proxy as raw TCP. Mirrors the TLS-passthrough behavior of diff --git a/core/listener/forwardproxy/server_test.go b/core/listener/forwardproxy/server_test.go index d6be7a995..ade3f3a16 100644 --- a/core/listener/forwardproxy/server_test.go +++ b/core/listener/forwardproxy/server_test.go @@ -711,6 +711,120 @@ func TestForwardProxy_RecordsMessageWithNoPluginActivity(t *testing.T) { } } +// TestForwardProxy_ResponseEventCarriesTtfb pins the decomposition #1251 asked +// for: a response row's Duration must split into ttfb (request entry to the +// upstream's headers arriving) and the remainder (the body, which for a +// streamed completion is the stream time). The backend flushes its headers and +// then waits before finishing the body, so a row that reports ttfb at all — and +// one smaller than its duration — proves both halves were measured where the +// exchange actually spent its time. An 11-minute response with a small ttfb +// says "the origin answered, then trickled", which Duration alone could not. +func TestForwardProxy_ResponseEventCarriesTtfb(t *testing.T) { + store := session.New(5*time.Minute, 100, 0) + defer store.Close() + + const drip = 60 * time.Millisecond + backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + w.WriteHeader(http.StatusOK) + w.(http.Flusher).Flush() + time.Sleep(drip) + w.Write([]byte("data: done\n\n")) + })) + defer backend.Close() + + p, err := pipeline.New([]pipeline.Plugin{}) + if err != nil { + t.Fatal(err) + } + srv := &Server{OutboundPipeline: pipeline.NewHolder(p), Sessions: store, Client: http.DefaultClient} + proxy := httptest.NewServer(srv.Handler()) + defer proxy.Close() + + req, _ := http.NewRequest("GET", backend.URL+"/stream", nil) + proxyClient := &http.Client{Transport: &http.Transport{Proxy: http.ProxyURL(mustParseURL(proxy.URL))}} + resp, err := proxyClient.Do(req) + if err != nil { + t.Fatalf("request failed: %v", err) + } + io.Copy(io.Discard, resp.Body) + resp.Body.Close() + + v := store.View(session.DefaultSessionID) + if v == nil || len(v.Events) != 2 { + t.Fatalf("expected 2 events (request + response), got %+v", v) + } + respEv := v.Events[1] + if respEv.Phase != pipeline.SessionResponse { + t.Fatalf("second event phase = %v, want SessionResponse", respEv.Phase) + } + if respEv.Ttfb <= 0 { + t.Errorf("response event ttfb = %v, want > 0 — headers arrived, so the figure exists", respEv.Ttfb) + } + if respEv.Ttfb > respEv.Duration { + t.Errorf("response event ttfb %v > duration %v — headers cannot arrive after the row is recorded", respEv.Ttfb, respEv.Duration) + } + // The drip happens entirely after the flush, so the body's share of the + // duration is at least most of it. The bound is deliberately loose: the + // claim is the ORDER of the two phases, not a stopwatch. + if body := respEv.Duration - respEv.Ttfb; body < drip/2 { + t.Errorf("duration−ttfb = %v, want ≥ %v — the body dripped for %v after the headers flushed", body, drip/2, drip) + } +} + +// TestForwardProxy_ResponseEventOmitsTtfbOnFailedExchange pins the other half +// of the field's contract: an exchange that never got response headers reports +// NO ttfb rather than a zero one. The dead-backend 502 is the #1251 shape — a +// black-holed or refused upstream — and a ttfbMs of 0 on that row would read +// as "the origin answered instantly" instead of "it never answered". +func TestForwardProxy_ResponseEventOmitsTtfbOnFailedExchange(t *testing.T) { + store := session.New(5*time.Minute, 100, 0) + defer store.Close() + + // A server that is already closed: its port answers with a refusal, which + // is the fast half of the failure family. The 502 the proxy synthesizes is + // the same row shape a timed-out (black-holed) dial produces. + dead := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {})) + deadURL := dead.URL + dead.Close() + + p, err := pipeline.New([]pipeline.Plugin{}) + if err != nil { + t.Fatal(err) + } + srv := &Server{OutboundPipeline: pipeline.NewHolder(p), Sessions: store, Client: http.DefaultClient} + proxy := httptest.NewServer(srv.Handler()) + defer proxy.Close() + + req, _ := http.NewRequest("GET", deadURL+"/gone", nil) + proxyClient := &http.Client{Transport: &http.Transport{Proxy: http.ProxyURL(mustParseURL(proxy.URL))}} + resp, err := proxyClient.Do(req) + if err != nil { + t.Fatalf("request failed: %v", err) + } + resp.Body.Close() + if resp.StatusCode != http.StatusBadGateway { + t.Fatalf("status = %d, want 502", resp.StatusCode) + } + + v := store.View(session.DefaultSessionID) + if v == nil || len(v.Events) != 2 { + t.Fatalf("expected 2 events (request + response), got %+v", v) + } + respEv := v.Events[1] + if respEv.Ttfb != 0 { + t.Errorf("response event ttfb = %v, want 0 — no headers ever arrived", respEv.Ttfb) + } + // The wire half of the same contract: absence, not a serialized zero. + data, err := json.Marshal(respEv) + if err != nil { + t.Fatalf("marshal: %v", err) + } + if strings.Contains(string(data), `"ttfbMs"`) { + t.Errorf("failed exchange serialized a ttfbMs key: %s", data) + } +} + // schemeCapturePlugin captures pctx.Scheme for the scheme-wiring // test below. type schemeCapturePlugin struct { diff --git a/core/listener/reverseproxy/server.go b/core/listener/reverseproxy/server.go index 0268abc51..a4e93dcd4 100644 --- a/core/listener/reverseproxy/server.go +++ b/core/listener/reverseproxy/server.go @@ -487,6 +487,14 @@ func (s *Server) modifyResponse(resp *http.Response) error { pctx.StatusCode = resp.StatusCode pctx.ResponseHeaders = resp.Header.Clone() + // The forward proxy's handleRequest sets its ResponseHeadersAt at the same + // point in its own flow — the moment the delivered response's headers are + // in hand — so ttfb means the same thing on a row from either listener. + // Runs for the streaming arm too: modifyResponse executes before the body + // is relayed, which is exactly what makes this headers-arrival, not + // body-completion. Stays zero on errorHandler paths, where no headers + // arrived (#1251). + pctx.ResponseHeadersAt = time.Now() // BytesDown for the response row (#1309), counted HERE and nowhere else — at // the wire, before either arm below gets at the body. The streaming arm sees @@ -651,6 +659,7 @@ func (s *Server) modifyResponse(resp *http.Response) error { StatusCode: resp.StatusCode, Error: pipeline.DeriveError(pctx), Duration: pipeline.DurationSince(pctx.StartedAt), + Ttfb: pipeline.DurationBetween(pctx.StartedAt, pctx.ResponseHeadersAt), TLS: eventTLS(pctx), }) } @@ -844,6 +853,7 @@ func (s *Server) recordInboundResponseEvent(pctx *pipeline.Context, statusCode i StatusCode: statusCode, Error: pipeline.EventErrorOr(pctx, fail), Duration: pipeline.DurationSince(pctx.StartedAt), + Ttfb: pipeline.DurationBetween(pctx.StartedAt, pctx.ResponseHeadersAt), TLS: eventTLS(pctx), }) } diff --git a/core/pipeline/context.go b/core/pipeline/context.go index b80176246..a00315031 100644 --- a/core/pipeline/context.go +++ b/core/pipeline/context.go @@ -155,6 +155,19 @@ type Context struct { // compute SessionEvent.Duration without walking the event history. StartedAt time.Time + // ResponseHeadersAt is the wall-clock time the upstream's response HEADERS + // arrived — when client.Do returned (both proxy listeners) or Envoy delivered + // the response-headers callback (extproc). Set by the listener, once, on the + // arm that actually got a response; left zero on a failed exchange, so + // SessionEvent.Ttfb stays absent rather than reporting a figure for a call + // that never answered (#1251). Together with StartedAt it decomposes a + // response's Duration: ttfb is how long the origin took to answer, and the + // remainder is the body — for a streamed completion, how long it took to + // finish streaming. Duration alone cannot tell a slow-to-answer origin from + // a slow-to-finish stream, which is the distinction an operator sorting by + // duration is usually asking the timeline for. + ResponseHeadersAt time.Time + // requestID backs RequestID(), which generates it on first use. See // requestid.go for why it is lazy rather than a constructor argument. requestID string diff --git a/core/pipeline/session.go b/core/pipeline/session.go index f29d6c7e8..a9aa73ddb 100644 --- a/core/pipeline/session.go +++ b/core/pipeline/session.go @@ -169,6 +169,18 @@ type SessionEvent struct { // it has a duration even though no response was ever recorded. Duration time.Duration + // Ttfb is the wall-clock time from request entry into the listener + // (pctx.StartedAt) to the upstream's response HEADERS arriving — not to + // the first body byte, which no listener measures. Zero when the exchange + // failed before headers (and on every request-phase event), so Duration − + // Ttfb on a delivered response is how long the body took: for a streamed + // completion, the stream time. Duration alone cannot separate an origin + // that was slow to answer from one that answered and then streamed slowly + // (#1251) — an 11-minute response with a small ttfb says "the origin + // answered, then trickled", which is a different problem than a large one. + // Tunnel rows carry none: an opaque tunnel has no headers to time. + Ttfb time.Duration + // TLS, when non-nil, carries connection-level identity for events // that arrived over TLS: negotiated version, cipher, and the peer's // SPIFFE ID extracted from the URI SAN. Nil for plaintext events @@ -514,10 +526,17 @@ type sessionEventWire struct { Host string `json:"host,omitempty"` // omitempty for the same skew reason as TunnelReason: an old agentop ignores the // key, and an event from a proxy that predates redirects decodes to "". - RequestedHost string `json:"requestedHost,omitempty"` - DurationMs int64 `json:"durationMs,omitempty"` - TLS *EventTLS `json:"tls,omitempty"` - Tunnel bool `json:"tunnel,omitempty"` + RequestedHost string `json:"requestedHost,omitempty"` + DurationMs int64 `json:"durationMs,omitempty"` + // TtfbMs omitempty, like DurationMs: absence is the meaningful value. Zero + // means the exchange never got as far as response headers — the upstream + // timed out, was refused, or failed TLS — and a ttfb of 0ms would read as + // an instant answer instead. Same wire-skew safety as DurationMs: an old + // agentop ignores the key, and an event from a proxy that predates the + // field decodes to zero and omits. + TtfbMs int64 `json:"ttfbMs,omitempty"` + TLS *EventTLS `json:"tls,omitempty"` + Tunnel bool `json:"tunnel,omitempty"` // omitempty so both skew directions are safe: an old agentop ignores an unknown // key, and a new agentop against an old proxy sees "" and renders exactly what it // renders today. @@ -568,6 +587,7 @@ func (e SessionEvent) MarshalJSON() ([]byte, error) { Host: e.Host, RequestedHost: e.RequestedHost, DurationMs: e.Duration.Milliseconds(), + TtfbMs: e.Ttfb.Milliseconds(), TLS: e.TLS, Tunnel: e.Tunnel, TunnelReason: e.TunnelReason, @@ -608,6 +628,7 @@ func (e *SessionEvent) UnmarshalJSON(data []byte) error { Host: w.Host, RequestedHost: w.RequestedHost, Duration: time.Duration(w.DurationMs) * time.Millisecond, + Ttfb: time.Duration(w.TtfbMs) * time.Millisecond, TLS: w.TLS, Tunnel: w.Tunnel, TunnelReason: w.TunnelReason, diff --git a/core/pipeline/session_test.go b/core/pipeline/session_test.go index e67727142..b4d6b924f 100644 --- a/core/pipeline/session_test.go +++ b/core/pipeline/session_test.go @@ -27,6 +27,7 @@ func TestSessionEvent_MarshalJSON_ReadableEnums(t *testing.T) { Direction: Inbound, Phase: SessionResponse, Duration: 250 * time.Millisecond, + Ttfb: 80 * time.Millisecond, Host: "weather-tool-mcp:9090", A2A: &A2AExtension{Method: "message/stream"}, } @@ -48,6 +49,12 @@ func TestSessionEvent_MarshalJSON_ReadableEnums(t *testing.T) { if !strings.Contains(s, `"durationMs":250`) { t.Errorf("durationMs missing or wrong: %s", s) } + // Ttfb likewise, and separately from Duration — a ttfb serialized in + // nanoseconds, or as the same figure as durationMs, would make the + // decomposition #1251 added unreadable. + if !strings.Contains(s, `"ttfbMs":80`) { + t.Errorf("ttfbMs missing or wrong: %s", s) + } } // TestSessionEvent_JSONRoundTrip locks in the round-trip contract between @@ -64,6 +71,7 @@ func TestSessionEvent_JSONRoundTrip(t *testing.T) { Direction: Outbound, Phase: SessionResponse, Duration: 1600 * time.Millisecond, + Ttfb: 412 * time.Millisecond, Host: "api.openai.com", A2A: &A2AExtension{ Method: "message/stream", RPCID: "r-1", SessionID: "ctx-1", @@ -117,7 +125,7 @@ func TestSessionEvent_MarshalJSON_OmitsEmpty(t *testing.T) { for _, field := range []string{ "a2a", "mcp", "inference", "auth", "plugins", "identity", "statusCode", - "error", "host", "durationMs", + "error", "host", "durationMs", "ttfbMs", // The omitempty half of the version-skew contract: a new proxy that // recorded none of these must not emit the keys at all, so a client // predating them sees the same bytes it always did. @@ -268,10 +276,10 @@ func TestSessionEventWire_HasEveryDomainField(t *testing.T) { domain := reflect.TypeOf(SessionEvent{}) wire := reflect.TypeOf(sessionEventWire{}) - // Deliberate renames, kept explicit so the check stays strict. Duration is - // milliseconds on the wire, and the DTO field is named for the unit rather than - // silently changing what a same-named field means. - renamed := map[string]string{"Duration": "DurationMs"} + // Deliberate renames, kept explicit so the check stays strict. Duration and + // Ttfb are milliseconds on the wire, and the DTO fields are named for the + // unit rather than silently changing what a same-named field means. + renamed := map[string]string{"Duration": "DurationMs", "Ttfb": "TtfbMs"} for i := 0; i < domain.NumField(); i++ { f := domain.Field(i) diff --git a/core/pipeline/snapshot.go b/core/pipeline/snapshot.go index b6011bc6b..913cc3046 100644 --- a/core/pipeline/snapshot.go +++ b/core/pipeline/snapshot.go @@ -184,6 +184,21 @@ func DurationSince(start time.Time) time.Duration { return time.Since(start) } +// DurationBetween returns to − from, or 0 when either endpoint is zero. +// For the optional response-side timestamp (ResponseHeadersAt), whose absence +// means "no headers ever arrived", the zero guard keeps an unset endpoint from +// publishing a nonsense figure — a negative span against time.Now() would read +// as a real duration on the row. The inverse of DurationSince: that one spans +// start-to-now for a figure read at record time, this one spans two recorded +// moments, which is what SessionEvent.Ttfb is (StartedAt → ResponseHeadersAt, +// measured when the row is written long after both). +func DurationBetween(from, to time.Time) time.Duration { + if from.IsZero() || to.IsZero() { + return 0 + } + return to.Sub(from) +} + // DeriveError constructs an EventError from response-side signals. // Returns nil for 2xx / no guardrail block / no parser error. func DeriveError(pctx *Context) *EventError { diff --git a/docs/laptop-service.md b/docs/laptop-service.md index e3af42e41..e3a5af568 100644 --- a/docs/laptop-service.md +++ b/docs/laptop-service.md @@ -790,7 +790,7 @@ The other reasons you may see, and what each one asks of you: | `client-hung-up` | The client vanished mid-handshake. Often a cancelled request; not evidence about trust, which is why it carries no advice. OpenCode hangs up rather than rejecting the CA; see [its page](agents/opencode.md#ca-trust). | usually no | | `handshake-failed` | Some other handshake failure — a version, cipher or ALPN mismatch, or Cortex failing to mint a certificate. | check `error=` in the log | | `origin-unverified` | **Cortex** could not verify the destination's certificate, so it declined to vouch for it. Bridging would have meant terminating TLS for a server we could not authenticate. | investigate the destination | -| `dial-failed` | Cortex could not reach the destination at all — a DNS failure, a refused connection or a timeout — so no tunnel opened. Its response row is a `502` whose `error` carries the dial error. | check the destination and the network path to it | +| `dial-failed` | Cortex could not reach the destination at all — a DNS failure, a refused connection or a timeout — so no tunnel opened. Its response row is a `502` whose `error` carries the dial error. The TCP dial is bounded at 10s, so a destination that silently drops the SYN (off the VPN that reaches it, a dead NAT) fails within that rather than hanging the agent for longer. | check the destination and the network path to it | Only a CA rejection asks you to restart anything: `client-rejected-ca` itself, and `program-refused` when the earlier failure it remembers was one. The others are either diff --git a/docs/session-api.md b/docs/session-api.md index eb40f0daa..51320f007 100644 --- a/docs/session-api.md +++ b/docs/session-api.md @@ -251,6 +251,14 @@ Every event on `/v1/sessions/{id}` and `/v1/events` carries: OpenAI's Responses API (`chatgpt.com` or `api.openai.com`): nothing is forwarded, and a client such as Codex reads it as "use HTTPS" and retries at once over HTTPS. It is not counted as an error in `/v1/usage`. +- `ttfbMs` — on a delivered response row, the wall time from request entry to + the upstream's response **headers** arriving. `durationMs` − `ttfbMs` is then + how long the body took — for a streamed completion, the stream time. The pair + separates a slow-to-answer origin from one that answered and then trickled, + which `durationMs` alone cannot (#1251). Absent when the exchange failed + before headers (the 502 cases above), on request rows, and on tunnel rows — + an opaque tunnel has no headers to time. agentop's detail pane renders the + split as a `timing:` line. - `requestedHost` — present only when a plugin redirected the request: the host the client asked for. `host` is where the request actually went, and usage, the cost ledger and pricing all follow `host`. From e1eb1ad1b4c2c810e2e8715a48fe025b31531775 Mon Sep 17 00:00:00 2001 From: Mariusz Sabath Date: Wed, 28 Oct 2026 16:05:06 -0400 Subject: [PATCH 2/4] test: record ttfb's decision in the sessionapi shape guard MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SessionEvent grew to 29 fields with Ttfb and the shape guard fired, which is its job: the field needs its keep-or-drop decision recorded. It is kept, for the same reason RequestedHost is — agentop's detail pane draws its timing: line from the projected event (#1251), and a scalar survives summarizeEvent's struct copy. Fixture, kept-field assertion, and the guard count with its lineage comment. Also name the third 10s precedent in connectDialTimeout's comment: the plain dialer in the same file was already at 10s, which made the 30s CONNECT dial the outlier in its own file. Assisted-By: Claude (Anthropic AI) Signed-off-by: Mariusz Sabath --- core/listener/forwardproxy/server.go | 5 +++-- core/sessionapi/summary_test.go | 13 ++++++++++++- 2 files changed, 15 insertions(+), 3 deletions(-) diff --git a/core/listener/forwardproxy/server.go b/core/listener/forwardproxy/server.go index 538eab737..f8c2acad5 100644 --- a/core/listener/forwardproxy/server.go +++ b/core/listener/forwardproxy/server.go @@ -2069,8 +2069,9 @@ func (s *Server) recordOutboundRejectIn(tl *tunnelLog, pctx *pipeline.Context, a // unanswered; a handshake that misses five straight tries is not being saved // by a sixth at 16s, so the rest of the budget bought no connection, only // latency before the 502. Matches the budget the bridge already trusts for -// its own upstream probe (upstreamVerifyTimeout) and the mTLS dialer, which -// had settled on 10s independently. +// its own upstream probe (upstreamVerifyTimeout), the mTLS dialer, which had +// settled on 10s independently, and this file's own plain dialer — the 30s +// CONNECT dial was the outlier in its own file. const connectDialTimeout = 10 * time.Second // handleConnect tunnels HTTPS (and any other TLS-wrapped protocol) through diff --git a/core/sessionapi/summary_test.go b/core/sessionapi/summary_test.go index 17de745e9..b9907c4e8 100644 --- a/core/sessionapi/summary_test.go +++ b/core/sessionapi/summary_test.go @@ -24,6 +24,7 @@ func fullEvent() pipeline.SessionEvent { Host: "api.example.com", StatusCode: 200, Duration: 1200 * time.Millisecond, + Ttfb: 412 * time.Millisecond, BytesUp: 4210, BytesDown: 18230, Identity: &pipeline.EventIdentity{}, @@ -150,6 +151,12 @@ func TestSummarizeEvent_DropsPayloadsKeepsTimelineFields(t *testing.T) { t.Errorf("requestedHost = %q, want %q — the detail pane's redirected: line reads it", got.RequestedHost, full.RequestedHost) } + // ttfb is kept for the same reason as requestedHost: the detail pane draws its + // timing: line from the projected event, splitting a response's duration into the + // origin's answer time and the body's (#1251). + if got.Ttfb != full.Ttfb { + t.Errorf("ttfb = %v, want %v — the detail pane's timing: line reads it", got.Ttfb, full.Ttfb) + } // upgradeStripped is kept for the same reason as requestedHost: the detail pane's // upgrade: line reads it off the projected event, and a 405 whose explanation only // appears after a round trip is the confusion this field exists to end. @@ -360,7 +367,11 @@ func TestSummarizeEvent_ShapeIsGuarded(t *testing.T) { // 27 since Subagent, which is TIMELINE data: it says which agent made the row. A // scalar capped at maxClientLen, so the struct copy keeps it; asserted above. // 28 since ParentSubagent, TIMELINE data for the same reason; asserted above. - {"SessionEvent", reflect.TypeOf(pipeline.SessionEvent{}), 28}, + // 29 since Ttfb, which is kept as RequestedHost is: agentop's detail pane draws + // its timing: line from the projected event, splitting a response's duration + // into the origin's answer time and the body's (#1251). A scalar, so the struct + // copy keeps it; asserted above. + {"SessionEvent", reflect.TypeOf(pipeline.SessionEvent{}), 29}, // 25 since AgentRole, which is TIMELINE data: agentop's CONTEXT gauge reads it on every // row the timeline serves. It needs no assertion of its own in the projection test // beyond the equality one there — a scalar survives the struct copy, unlike the two From df9ce704b73c78dc5d569c9ae3c434b9a29f1ada Mon Sep 17 00:00:00 2001 From: Mariusz Sabath Date: Wed, 28 Oct 2026 19:47:19 -0400 Subject: [PATCH 3/4] fix: record unbuffered response rows after the body relay MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three CodeRabbit findings on the #1251 split: - agentop's timing: line printed an empty body share when the computed remainder was zero (formatDuration(0) is ""), reading as "body of 412ms total". A computed zero is now rendered as an explicit 0ms — distinct from the counted-nothing case, which stays blank. - forwardproxy's serveOutbound appended the response row before the final io.Copy, so an unbuffered arm's Duration stopped at the headers and duration−ttfb read as a zero body while the proxy was still relaying one. The row is now recorded after the copy, matching the streamed arms, and BytesDown is complete for the same reason. - reverseproxy's modifyResponse had the same shape: the row was appended before httputil.ReverseProxy copies the body. It is now deferred to the end of that copy (recordOnEOFBody fires on EOF or Close, once), except a 101 — ReverseProxy hands the connection to the upgraded protocol and never copies or closes, so a failed upgrade would leave no row at all; the 101 is recorded inline. The parity consequence is pinned rather than papered over: with no plugin reading bodies, extproc asks Envoy for body mode NONE and never receives the bytes, so it reports ↓0 where the proxies now report the relayed count — a structural split (every shipped pipeline reads bodies, so it is custom body-less pipelines only). TestParity_ResponseEventEmission moves to a bodyless fixture — its claim is about the plugin event, which a headers-only plugin sees the same either way — and the body-carrying arm is documented in divergence_test.go with per-listener rows and reasons, via a new bytes field on divergentExpectation. Signed-off-by: Mariusz Sabath Assisted-By: Claude (Anthropic AI) --- CLAUDE.md | 6 +- cmd/agentop/tui/detail_pane.go | 10 +- cmd/agentop/tui/detail_pane_test.go | 15 ++ core/listener/forwardproxy/bytescount_test.go | 27 ++-- core/listener/forwardproxy/server.go | 25 +++- core/listener/forwardproxy/server_test.go | 63 ++++++++ core/listener/parity/divergence_test.go | 139 ++++++++++++++++++ core/listener/parity/parity_test.go | 15 +- core/listener/reverseproxy/bytescount_test.go | 28 ++-- .../reverseproxy/finalize_context_test.go | 6 + core/listener/reverseproxy/server.go | 64 +++++++- core/listener/reverseproxy/server_test.go | 64 ++++++++ docs/session-api.md | 5 +- 13 files changed, 422 insertions(+), 45 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 1a29adbf9..15d4c8e2b 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -625,7 +625,11 @@ Every event on `/v1/sessions/{id}` and `/v1/events` carries: listeners set `pctx.ResponseHeadersAt` at their own headers-arrival moment — `client.Do` returning (forward and reverse proxy), Envoy's response-headers callback (extproc) — and every response-event builder derives the figure the same way - (`pipeline.DurationBetween`). agentop's detail pane renders the split as a `timing:` line. + (`pipeline.DurationBetween`). The row itself is recorded when the listener is + finished with the body — after the read on a buffered path, after the relay on a + streamed or unbuffered one — so the difference really is the body's time and not a + remainder cut short at the headers. agentop's detail pane renders the split as a + `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, and so does a response relayed straight through — both proxies copy that one after the row is appended. 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. All 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. diff --git a/cmd/agentop/tui/detail_pane.go b/cmd/agentop/tui/detail_pane.go index 1eadd3270..0dbd31d80 100644 --- a/cmd/agentop/tui/detail_pane.go +++ b/cmd/agentop/tui/detail_pane.go @@ -93,8 +93,16 @@ func timingHeader(e *pipeline.SessionEvent) string { if body < 0 { body = 0 } + // formatDuration renders zero blank because a zero in the DURATION column + // means "the listener counted nothing" — an unknown. This zero is a computed + // difference and means the body took under a millisecond, which the sentence + // wants said rather than elided: a blank would print "body of 412ms total". + bodyText := formatDuration(body) + if body == 0 { + bodyText = "0ms" + } return fmt.Sprintf("timing: ttfb %s · body %s of %s total", - formatDuration(e.Ttfb), formatDuration(body), formatDuration(e.Duration)) + formatDuration(e.Ttfb), bodyText, formatDuration(e.Duration)) } // tunnelHeader summarizes the CONNECT tunnel folded into a TLS-bridged diff --git a/cmd/agentop/tui/detail_pane_test.go b/cmd/agentop/tui/detail_pane_test.go index a34b939c7..519f5ae5e 100644 --- a/cmd/agentop/tui/detail_pane_test.go +++ b/cmd/agentop/tui/detail_pane_test.go @@ -260,6 +260,21 @@ func TestTimingHeader_SplitsDurationIntoTtfbAndBody(t *testing.T) { } } +// A body that landed inside the same millisecond as the headers is a computed +// zero, not an unknown one: the column convention renders counted-nothing +// blank, but a blank here would leave a hole in the middle of the sentence. +func TestTimingHeader_RendersAZeroBodyExplicitly(t *testing.T) { + got := timingHeader(&pipeline.SessionEvent{ + Phase: pipeline.SessionResponse, + Duration: 412 * time.Millisecond, + Ttfb: 412 * time.Millisecond, + }) + want := "timing: ttfb 412ms · body 0ms of 412ms total" + if got != want { + t.Errorf("timingHeader = %q, want %q", got, want) + } +} + // No header where the decomposition would be fiction: a request row (no // duration), a failed exchange (no ttfb — nothing answered), and a tunnel // close (no headers to time). diff --git a/core/listener/forwardproxy/bytescount_test.go b/core/listener/forwardproxy/bytescount_test.go index 86d7bece1..a8d18843a 100644 --- a/core/listener/forwardproxy/bytescount_test.go +++ b/core/listener/forwardproxy/bytescount_test.go @@ -124,18 +124,18 @@ func TestBytes_PassthroughSSECountsTheFramingToo(t *testing.T) { } } -// TestBytes_UnbufferedResponseReportsZeroRatherThanAGuess pins the one arm the -// counting wrapper cannot reach: a plain response that no plugin asked to buffer -// and that is not an event stream is relayed by an io.Copy at the bottom of -// handleHTTP, which runs AFTER the response row has been recorded. +// TestBytes_UnbufferedResponseReportsTheFullCount pins the arm this file used +// to excuse: a plain response that no plugin asked to buffer and that is not an +// event stream is relayed by an io.Copy at the bottom of serveOutbound, which +// runs AFTER the row used to be recorded — so this arm reported zero. The row is +// now recorded below that copy, and the count the wrapper finishes during it is +// the figure the row carries. // -// Zero is the honest answer and the event contract's own: zero means the listener -// counted nothing, not that the body was empty, and agentop renders it blank. The -// tempting alternative is Content-Length, which is -1 on anything chunked and -// deleted outright on every streaming arm — a guess that would be wrong exactly -// where it mattered. Pinned so that if the ordering here ever changes, it changes -// deliberately. -func TestBytes_UnbufferedResponseReportsZeroRatherThanAGuess(t *testing.T) { +// Still counted at the wire, not guessed from Content-Length, which is -1 on +// anything chunked and deleted outright on every streaming arm. reverseproxy's +// unbuffered relay defers its row to the end of its copy the same way +// (recordOnEOFBody). +func TestBytes_UnbufferedResponseReportsTheFullCount(t *testing.T) { req, _ := http.NewRequest("GET", "http://placeholder/plain", nil) _, respRow := bytesOf(t, []pipeline.Plugin{}, req, @@ -144,7 +144,8 @@ func TestBytes_UnbufferedResponseReportsZeroRatherThanAGuess(t *testing.T) { fmt.Fprint(w, `{"never":"buffered"}`) }) - if respRow.BytesDown != 0 { - t.Errorf("response row ↓%d, want ↓0 — the relay copies after the row is recorded, so any figure here is counted too late to be this row's", respRow.BytesDown) + if respRow.BytesDown != int64(len(`{"never":"buffered"}`)) { + t.Errorf("response row ↓%d, want ↓%d — the row is recorded after the copy, so the count is complete when the figure is read", + respRow.BytesDown, len(`{"never":"buffered"}`)) } } diff --git a/core/listener/forwardproxy/server.go b/core/listener/forwardproxy/server.go index f8c2acad5..3a2cc9e73 100644 --- a/core/listener/forwardproxy/server.go +++ b/core/listener/forwardproxy/server.go @@ -595,12 +595,13 @@ func (s *Server) serveOutbound(w http.ResponseWriter, r *http.Request, tl *tunne // with what extproc reports for the same response without each having to // count: see bodycount. // - // What the wrapper does NOT reach is the final io.Copy at the bottom of this - // function — the arm a response takes when nothing buffered it and it was not - // streamed. The row is recorded above that copy, so those bytes are counted - // after the figure has been read and the row reports zero, which agentop - // renders blank rather than wrong. reverseproxy's unbuffered relay has the same - // shape for the same reason. + // The final io.Copy at the bottom of this function — the arm a response takes + // when nothing buffered it and it was not streamed — reads through this + // wrapper, and the row is recorded below that copy, so the count is complete + // when the figure is read. What the wrapper still does not reach is a SKIPPED + // response's row: skip_hosts records nothing at all. reverseproxy's + // unbuffered relay defers its row to the end of its copy the same way + // (recordOnEOFBody). resp.Body = bodycount.Wrap(resp.Body, &pctx.ResponseBytes) // SkipHosts: bypass response-phase pipeline + recording entirely. @@ -726,7 +727,6 @@ func (s *Server) serveOutbound(w http.ResponseWriter, r *http.Request, tl *tunne resp.Header.Del("Content-Encoding") } - s.recordOutboundResponseEvent(pctx, resp.StatusCode, nil) } for key, values := range resp.Header { @@ -747,6 +747,17 @@ func (s *Server) serveOutbound(w http.ResponseWriter, r *http.Request, tl *tunne if _, err := io.Copy(dst, resp.Body); err != nil { slog.Debug("response copy error", "host", r.Host, "error", err) } + // Recorded after the copy, not before it: the arm that reaches this io.Copy + // unbuffered is still receiving its body while the copy runs, and a row filed + // above it would stop Duration at the headers — reporting a body share of zero + // for a body the proxy was still relaying, the exact misreading #1251's split + // exists to prevent. After the copy the figure covers the whole exchange, + // matching the streamed arms (which record when their relay loop ends), and + // ResponseBytes is complete for the same reason. Skipped responses stay + // unrecorded, as skip_hosts promises. + if !skipped { + s.recordOutboundResponseEvent(pctx, resp.StatusCode, nil) + } } // flushWriter flushes after every write, so a relayed stream reaches the client as the diff --git a/core/listener/forwardproxy/server_test.go b/core/listener/forwardproxy/server_test.go index ade3f3a16..9e328ba82 100644 --- a/core/listener/forwardproxy/server_test.go +++ b/core/listener/forwardproxy/server_test.go @@ -772,6 +772,69 @@ func TestForwardProxy_ResponseEventCarriesTtfb(t *testing.T) { } } +// TestForwardProxy_ResponseEventDurationIncludesUnbufferedBody pins the arm the +// SSE test above cannot reach: a plain response no plugin asked to buffer is +// relayed by the io.Copy at the bottom of serveOutbound, and the row used to be +// filed before that copy — Duration stopped at the headers, so duration−ttfb +// read as a zero body while the body was still arriving (#1251's misreading, on +// exactly the rows the split was built to diagnose). The row now waits for the +// copy, which also completes ResponseBytes for the same arm. +func TestForwardProxy_ResponseEventDurationIncludesUnbufferedBody(t *testing.T) { + store := session.New(5*time.Minute, 100, 0) + defer store.Close() + + const drip = 60 * time.Millisecond + backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + // Plain JSON, not event-stream: no streaming responder, and an empty + // pipeline buffers nothing — the final io.Copy arm. + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + w.(http.Flusher).Flush() + time.Sleep(drip) + w.Write([]byte(`{"ok":true}`)) + })) + defer backend.Close() + + p, err := pipeline.New([]pipeline.Plugin{}) + if err != nil { + t.Fatal(err) + } + srv := &Server{OutboundPipeline: pipeline.NewHolder(p), Sessions: store, Client: http.DefaultClient} + proxy := httptest.NewServer(srv.Handler()) + defer proxy.Close() + + req, _ := http.NewRequest("GET", backend.URL+"/plain", nil) + proxyClient := &http.Client{Transport: &http.Transport{Proxy: http.ProxyURL(mustParseURL(proxy.URL))}} + resp, err := proxyClient.Do(req) + if err != nil { + t.Fatalf("request failed: %v", err) + } + io.Copy(io.Discard, resp.Body) + resp.Body.Close() + + v := store.View(session.DefaultSessionID) + if v == nil || len(v.Events) != 2 { + t.Fatalf("expected 2 events (request + response), got %+v", v) + } + respEv := v.Events[1] + if respEv.Phase != pipeline.SessionResponse { + t.Fatalf("second event phase = %v, want SessionResponse", respEv.Phase) + } + if respEv.Ttfb <= 0 { + t.Errorf("response event ttfb = %v, want > 0 — headers arrived, so the figure exists", respEv.Ttfb) + } + // The drip happens entirely after the flush, so the body's share is at least + // most of it — the same loose bound as the SSE test: the claim is the ORDER + // of the two phases, not a stopwatch. + if body := respEv.Duration - respEv.Ttfb; body < drip/2 { + t.Errorf("duration−ttfb = %v, want ≥ %v — the body dripped for %v after the headers flushed", + body, drip/2, drip) + } + if respEv.BytesDown == 0 { + t.Errorf("response event BytesDown = 0, want the relayed body counted — the row is recorded after the copy") + } +} + // TestForwardProxy_ResponseEventOmitsTtfbOnFailedExchange pins the other half // of the field's contract: an exchange that never got response headers reports // NO ttfb rather than a zero one. The dead-backend 502 is the #1251 shape — a diff --git a/core/listener/parity/divergence_test.go b/core/listener/parity/divergence_test.go index 509252540..411f97570 100644 --- a/core/listener/parity/divergence_test.go +++ b/core/listener/parity/divergence_test.go @@ -164,6 +164,14 @@ type divergentExpectation struct { // therefore asserts the event carried no plugin payload — which is the // whole claim on the reverse proxy's leg below, where OnResponse never ran. pluginEvents map[string]string + + // bytes, when non-nil, pins BytesUp/BytesDown on this listener's row — the + // one field here that has to be optional against this struct's "every field + // is asserted on every row" rule, because zero is itself a value a row may + // want to pin (the extproc leg of the unread-body fixtures below claims + // exactly ↓0), so a plain pair could not tell "assert nothing" from "assert + // zero". A pointer, for the same reason fixture.expectedBytes is one. + bytes *bytesSummary } // assertDivergence runs the fixture through every listener and checks each one @@ -293,6 +301,12 @@ func assertDivergence(t *testing.T, f fixture, wantPhase pipeline.SessionPhase, t.Errorf("%s: unexpected plugin event %q: %s", where, key, gotJSON) } } + if e.bytes != nil { + got := bytesSummary{Up: obs.BytesUp, Down: obs.BytesDown} + if got != *e.bytes { + t.Errorf("%s: bytes counted\n got: %+v\n want: %+v", where, got, *e.bytes) + } + } } } @@ -320,6 +334,15 @@ func sameRow(a, b divergentExpectation) bool { a.eventStatusCode != b.eventStatusCode || a.errorKind != b.errorKind || a.errorCode != b.errorCode { return false } + // bytes participates too, with nil and an empty summary kept distinct for + // the reason the field's own comment gives: ↓0 is a claim, so a row pinning + // it diverges from a row making no claim about bytes at all. + if (a.bytes == nil) != (b.bytes == nil) { + return false + } + if a.bytes != nil && *a.bytes != *b.bytes { + return false + } if len(a.pluginEvents) != len(b.pluginEvents) { return false } @@ -671,3 +694,119 @@ func TestDivergence_OutboundResponseBodyCapOnShippedEnvoy(t *testing.T) { }, }) } + +// pricedEvents is the emit both unread-body fixtures declare on every row: the +// spy's OnResponse publish, which must surface identically regardless of what +// the listener did with the bytes. +var pricedEvents = map[string]string{ + spyPluginA: jsonOf(spyEvent{Marker: "priced", Count: 3}), +} + +// TestDivergence_InboundResponseBodyCountWhenNoPluginReadsIt: a response body +// no plugin asked to see, and the one field the listeners now honestly +// disagree on — BytesDown. +// +// The split is #1251's fix making it visible. Before it, the reverse proxy +// appended its response row inside modifyResponse, BEFORE httputil.ReverseProxy +// copies the body to the client, so this arm reported ↓0; extproc reported ↓0 +// too, because with no body-reading plugin it asks Envoy for body mode NONE and +// Envoy never sends it the bytes. Three zeros agreed — the shared gap +// observationDiff's BytesDown comment describes — so parity held for the wrong +// reason. The fix moved the proxies' row to after the relay, where the body they +// relay through their own process is counted, and the agreement broke into the +// split this file exists to document. +// +// It cannot be closed on the extproc side. Matching the proxies would mean +// requesting bodies no plugin needs — defeating body mode NONE, which is what +// keeps unobserved responses unbuffered — so the honest options were restoring +// the shared zero (un-fixing the misread duration #1251 exists to fix) or +// pinning the divergence with its reason. Pinned. +// +// Narrower than it looks: every shipped pipeline includes inference-parser, +// whose undirected ReadsBody counts on both directions, so NeedsBody() is true +// there and extproc buffers, counts and records at body end like the proxies. +// This divergence is the custom body-less pipeline — an auth-only gate, say — +// not any shape a shipped profile runs. +// +// Duration has the same shape — extproc's stops at the headers, the proxies' +// now covers the relay — and is deliberately not pinned here: a timing figure +// needs sleeps to assert against, and the proxies' own packages pin their side +// (TestReverseProxy_ResponseEventDurationIncludesUnbufferedBody and the +// forwardproxy twin). The parity suite's HasDuration check is presence-only. +// +// The spy emits — the same priced emit the parity fixture this arm used to live +// in carried — for two reasons: the row must exist on every leg (extproc's +// recorders write nothing for a pipeline whose plugins published no A2A, no +// Invocation and no Custom entry), and the emit surfacing identically in both +// rows keeps a slice of that fixture's old claim, TestParity_ +// ResponseEventEmission's, pinned here rather than dropped with the move. +func TestDivergence_InboundResponseBodyCountWhenNoPluginReadsIt(t *testing.T) { + f := fixture{ + name: "inbound-unread-response-body", + direction: pipeline.Inbound, + entries: []config.PluginEntry{spyEntry(spyPluginA, spyConfig{ + EmitOnResponse: true, + ResponseEvent: &spyEvent{Marker: "priced", Count: 3}, + })}, + method: "GET", + path: "/divergence/unread-response", + upstreamStatus: 200, + upstreamBody: []byte(`{"reply":"ok"}`), + } + assertDivergence(t, f, pipeline.SessionResponse, inboundListeners, map[string]divergentExpectation{ + "extproc": { + why: "no plugin reads bodies, so the listener asks Envoy for body mode NONE and Envoy never sends it the response body; the row is recorded at the headers with ↓0 — not bytes it saw and failed to count, but bytes it asked not to see", + pipelineRan: true, + wireStatus: 0, // success path; this listener has no HTTP transport to report from + eventStatusCode: 200, + bytes: &bytesSummary{}, + pluginEvents: pricedEvents, + }, + "reverseproxy": { + why: "the unbuffered relay copies this body through the listener's own process, so it is counted at the wire and the row recorded after the copy (#1251); before that fix this leg reported ↓0 too and the two shapes agreed for the wrong reason", + pipelineRan: true, + wireStatus: 200, + eventStatusCode: 200, + bytes: &bytesSummary{Down: int64(len(`{"reply":"ok"}`))}, + pluginEvents: pricedEvents, + }, + }) +} + +// TestDivergence_OutboundResponseBodyCountWhenNoPluginReadsIt is the egress +// twin, and exists for the same reason the cap fixtures come in pairs: the +// forward proxy's unbuffered relay is separate code from the reverse proxy's +// (serveOutbound's io.Copy against recordOnEOFBody), so the post-copy recording +// could regress on one side only. Same body, same split, different recorder. +func TestDivergence_OutboundResponseBodyCountWhenNoPluginReadsIt(t *testing.T) { + f := fixture{ + name: "outbound-unread-response-body", + direction: pipeline.Outbound, + entries: []config.PluginEntry{spyEntry(spyPluginA, spyConfig{ + EmitOnResponse: true, + ResponseEvent: &spyEvent{Marker: "priced", Count: 3}, + })}, + method: "GET", + path: "/divergence/unread-response", + upstreamStatus: 200, + upstreamBody: []byte(`{"reply":"ok"}`), + } + assertDivergence(t, f, pipeline.SessionResponse, outboundListeners, map[string]divergentExpectation{ + "extproc": { + why: "same mechanism as the inbound twin: body mode NONE means Envoy never sends the bytes, so the row is recorded at the headers with ↓0", + pipelineRan: true, + wireStatus: 0, // success path; this listener has no HTTP transport to report from + eventStatusCode: 200, + bytes: &bytesSummary{}, + pluginEvents: pricedEvents, + }, + "forwardproxy": { + why: "serveOutbound relays the body through this process and records the row after its io.Copy (#1251); the count is complete for the same reason the duration is", + pipelineRan: true, + wireStatus: 200, + eventStatusCode: 200, + bytes: &bytesSummary{Down: int64(len(`{"reply":"ok"}`))}, + pluginEvents: pricedEvents, + }, + }) +} diff --git a/core/listener/parity/parity_test.go b/core/listener/parity/parity_test.go index feb892789..27df4f1a8 100644 --- a/core/listener/parity/parity_test.go +++ b/core/listener/parity/parity_test.go @@ -77,6 +77,19 @@ func TestParity_DenyOnRequest(t *testing.T) { // TestParity_ResponseEventEmission: an OnResponse emit to Extensions. // Custom must surface identically on SessionEvent.Plugins from both // listeners. +// +// Bodyless deliberately. The fixture's claim is about the plugin event, and a +// headers-only plugin sees the same OnResponse either way — but with a body +// present and no plugin reading it, the listeners no longer agree on +// BytesDown: the proxies relay that body through their own process and count +// it (their row is recorded after the relay, #1251's fix), while extproc asked +// Envoy for body mode NONE and never received the bytes at all. That split is +// structural — closing it would mean making extproc request bodies no plugin +// needs — so it is pinned with its reason in divergence_test.go +// (TestDivergence_InboundResponseBodyCountWhenNoPluginReadsIt) rather than +// silently failing here. Dropping the body drops only the divergent figure; +// the emit fires in OnResponse either way, and the absolute anchor below keeps +// the bodyless zero meaning "no bytes", not "no reader". func TestParity_ResponseEventEmission(t *testing.T) { f := fixture{ name: "priced-response-headers-only", @@ -88,7 +101,7 @@ func TestParity_ResponseEventEmission(t *testing.T) { method: "GET", path: "/parity/priced", upstreamStatus: 200, - upstreamBody: []byte(`{"reply":"ok"}`), + expectedBytes: &bytesSummary{}, } assertParity(t, f, pipeline.SessionResponse, inboundListeners) } diff --git a/core/listener/reverseproxy/bytescount_test.go b/core/listener/reverseproxy/bytescount_test.go index 0b7169d96..6050d0652 100644 --- a/core/listener/reverseproxy/bytescount_test.go +++ b/core/listener/reverseproxy/bytescount_test.go @@ -88,27 +88,25 @@ func TestBytes_BufferedPairCountsEachDirectionOnItsOwnRow(t *testing.T) { } } -// TestBytes_UnbufferedResponseReportsZeroRatherThanAGuess pins the one arm the -// counting wrapper cannot reach: with no plugin asking for the response body, -// httputil.ReverseProxy copies it straight to the client — and it does that AFTER -// modifyResponse returns, which is after the response row has been appended -// inside it. +// TestBytes_UnbufferedResponseReportsTheFullCount pins the arm this file used +// to excuse: with no plugin asking for the response body, httputil.ReverseProxy +// copies it straight to the client — AFTER modifyResponse returns. The row used +// to be appended inside modifyResponse, before the copy, so this arm reported +// zero; it is now deferred to the end of the copy (recordOnEOFBody), and the +// count the wrapper finishes during that copy is the figure the row carries. // -// Zero is the honest answer and the event contract's own: zero means the listener -// counted nothing, not that the body was empty, and agentop renders it blank. The -// tempting alternative is Content-Length, which is -1 on anything chunked and -// deleted outright on the streaming arm — a guess that would be wrong exactly -// where it mattered. Pinned so that if the ordering here ever changes, it changes -// deliberately. forwardproxy's unbuffered relay has the same shape for the same -// reason. -func TestBytes_UnbufferedResponseReportsZeroRatherThanAGuess(t *testing.T) { +// Still counted at the wire, not guessed from Content-Length, which is -1 on +// anything chunked and deleted outright on the streaming arm. forwardproxy's +// unbuffered relay records after its copy for the same reason. +func TestBytes_UnbufferedResponseReportsTheFullCount(t *testing.T) { _, respRow := bytesOf(t, []pipeline.Plugin{}, "GET", "/plain", "", func(w http.ResponseWriter, _ *http.Request) { w.Header().Set("Content-Type", "application/json") fmt.Fprint(w, `{"never":"buffered"}`) }) - if respRow.BytesDown != 0 { - t.Errorf("response row ↓%d, want ↓0 — ReverseProxy copies after the row is recorded, so any figure here is counted too late to be this row's", respRow.BytesDown) + if respRow.BytesDown != int64(len(`{"never":"buffered"}`)) { + t.Errorf("response row ↓%d, want ↓%d — the row is recorded after the copy, so the count is complete when the figure is read", + respRow.BytesDown, len(`{"never":"buffered"}`)) } } diff --git a/core/listener/reverseproxy/finalize_context_test.go b/core/listener/reverseproxy/finalize_context_test.go index 900d28240..f4e351fcd 100644 --- a/core/listener/reverseproxy/finalize_context_test.go +++ b/core/listener/reverseproxy/finalize_context_test.go @@ -219,6 +219,12 @@ func TestModifyResponse_BufferedArmSurvivesAHangup(t *testing.T) { if err := srv.modifyResponse(resp); err != nil { t.Fatalf("modifyResponse: %v — a cancelled request context produced a Deny this arm cannot tell from a policy reject, so a complete response was rendered as a rejection", err) } + // What ReverseProxy does next: relay the body. The unbuffered arm's row is + // deferred to the end of that copy (recordOnEOFBody), so drain the body as + // the relay would — a hung-up client's copy fails and Close fires the same + // record, which is the telemetry-survival claim under test. + io.Copy(io.Discard, resp.Body) + resp.Body.Close() if terminals, ctxErr, _, _ := probe.snapshot(); terminals != 1 || ctxErr != nil { t.Errorf("terminal dispatches = %d, ctx.Err = %v, want 1 and nil: nothing finalized for a response that arrived complete", terminals, ctxErr) } diff --git a/core/listener/reverseproxy/server.go b/core/listener/reverseproxy/server.go index a4e93dcd4..dd3f3b5f3 100644 --- a/core/listener/reverseproxy/server.go +++ b/core/listener/reverseproxy/server.go @@ -16,6 +16,7 @@ import ( "net/url" "strconv" "strings" + "sync" "time" "github.com/rossoctl/cortex/core/listener/httpx" @@ -504,9 +505,10 @@ func (s *Server) modifyResponse(resp *http.Response) error { // bytes. The buffered arm's replacement of resp.Body with a reader over what it // already read means nothing is counted twice. See bodycount. // - // What this does NOT reach is the fully-unbuffered relay: ReverseProxy copies - // that body after modifyResponse returns, which is after the response row is - // appended below, so it stays at zero and renders blank rather than wrong. + // The fully-unbuffered relay gets its full count too: ReverseProxy copies + // that body after modifyResponse returns, and that arm's row is deferred to + // the end of the copy (recordOnEOFBody below), so the count is complete when + // the figure is read. if resp.Body != nil { resp.Body = bodycount.Wrap(resp.Body, &pctx.ResponseBytes) } @@ -631,9 +633,21 @@ func (s *Server) modifyResponse(resp *http.Response) error { // SSE responses still get recorded — the body is whatever the // pipeline saw at this point (may be empty for streamed bodies), // but the status code and plugin invocations are always meaningful. - plugins := pipeline.SnapshotPlugins(pctx.Extensions.Custom) - // Always pair every inbound request with a response row (carries StatusCode). - if s.Sessions != nil { + // + // The buffered arm has the whole body in hand by the time it gets here, so + // its row is complete at this point. The unbuffered arm does not: ReverseProxy + // copies resp.Body after modifyResponse returns, and a row filed now would + // stop Duration at the headers — a body share of zero for a body still + // arriving, the exact misreading #1251's split exists to prevent. That arm + // defers the row to the end of the copy via recordOnEOFBody, which is also + // when ResponseBytes finishes counting (see the bodycount note above). The + // streaming arm returned earlier and records from its own onClose. + record := func() { + if s.Sessions == nil { + return + } + plugins := pipeline.SnapshotPlugins(pctx.Extensions.Custom) + // Always pair every inbound request with a response row (carries StatusCode). sid := inboundSessionID(pctx) // Inference is the token report, and it is what a cost consumer reads: without it // this listener served a whole `configured` cost for zero tokens, because the cost @@ -663,6 +677,17 @@ func (s *Server) modifyResponse(resp *http.Response) error { TLS: eventTLS(pctx), }) } + // A 101 is recorded here, not deferred: ReverseProxy never copies its body — + // it hands the connection to the upgraded protocol — so on a failed upgrade + // (protocol mismatch, failed hijack) nothing would ever fire the deferred + // record. The row is the 101 itself, and ErrorHandler must not add a second + // one after it. + if !s.InboundPipeline.NeedsBody() && resp.Body != nil && + resp.StatusCode != http.StatusSwitchingProtocols { + resp.Body = &recordOnEOFBody{inner: resp.Body, onClose: record} + return nil + } + record() return nil } @@ -858,6 +883,33 @@ func (s *Server) recordInboundResponseEvent(pctx *pipeline.Context, statusCode i }) } +// recordOnEOFBody defers the response row of the fully-unbuffered relay to +// the moment ReverseProxy finishes with the body. modifyResponse returns +// before that copy starts, so a row appended there would stop Duration at the +// headers and report a body share of zero for a body still arriving (#1251) — +// and a ResponseBytes figure the bodycount wrapper had not finished counting. +// Records exactly once, on EOF or Close, whichever arrives first: a copy that +// runs to the end fires from Read, a client that hangs up mid-copy still gets +// its row from ReverseProxy's Close. +type recordOnEOFBody struct { + inner io.ReadCloser + onClose func() + once sync.Once +} + +func (b *recordOnEOFBody) Read(p []byte) (int, error) { + n, err := b.inner.Read(p) + if err == io.EOF { + b.once.Do(b.onClose) + } + return n, err +} + +func (b *recordOnEOFBody) Close() error { + b.once.Do(b.onClose) + return b.inner.Close() +} + // streamingResponseBody is the io.ReadCloser ReverseProxy ferries // downstream. Each Read call pulls one SSE frame from the upstream, // dispatches it through the pipeline, and copies the block's upstream diff --git a/core/listener/reverseproxy/server_test.go b/core/listener/reverseproxy/server_test.go index 00f2d201b..783cc6674 100644 --- a/core/listener/reverseproxy/server_test.go +++ b/core/listener/reverseproxy/server_test.go @@ -720,6 +720,70 @@ func TestReverseProxy_ModifyResponse_RekeyAndResponseEvent(t *testing.T) { } } +// TestReverseProxy_ResponseEventDurationIncludesUnbufferedBody pins the arm the +// buffered tests cannot reach: when no inbound plugin asks for the body, +// modifyResponse returns before ReverseProxy copies it, and the row used to be +// appended right there — Duration stopped at the headers, so duration−ttfb read +// as a zero body while the body was still being relayed (#1251's misreading). +// The row is now deferred to the end of the copy (recordOnEOFBody), which also +// completes BytesDown for the same arm. +func TestReverseProxy_ResponseEventDurationIncludesUnbufferedBody(t *testing.T) { + store := session.New(5*time.Minute, 100, 0) + defer store.Close() + + const drip = 60 * time.Millisecond + backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + // Plain JSON, not event-stream, and the pipeline below is empty: nothing + // buffers the body, so ReverseProxy's own copy is the relay. + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + w.(http.Flusher).Flush() + time.Sleep(drip) + _, _ = w.Write([]byte(`{"ok":true}`)) + })) + defer backend.Close() + + p, err := pipeline.New([]pipeline.Plugin{}) + if err != nil { + t.Fatal(err) + } + srv, err := NewServer(pipeline.NewHolder(p), store, backend.URL, nil) + if err != nil { + t.Fatalf("NewServer: %v", err) + } + proxy := httptest.NewServer(srv.Handler()) + defer proxy.Close() + + resp, err := http.DefaultClient.Get(proxy.URL + "/api") + if err != nil { + t.Fatalf("Get: %v", err) + } + io.Copy(io.Discard, resp.Body) + resp.Body.Close() + + v := store.View(session.DefaultSessionID) + if v == nil || len(v.Events) != 2 { + t.Fatalf("expected 2 events (request + response) in the default bucket, got %+v", v) + } + respEv := v.Events[1] + if respEv.Phase != pipeline.SessionResponse { + t.Fatalf("second event phase = %v, want SessionResponse", respEv.Phase) + } + if respEv.Ttfb <= 0 { + t.Errorf("response event ttfb = %v, want > 0 — headers arrived, so the figure exists", respEv.Ttfb) + } + // The drip happens entirely after the flush, so the body's share is at least + // most of it. Loose bound on purpose: the claim is the ORDER of the two + // phases, not a stopwatch. + if body := respEv.Duration - respEv.Ttfb; body < drip/2 { + t.Errorf("duration−ttfb = %v, want ≥ %v — the body dripped for %v after the headers flushed", + body, drip/2, drip) + } + if respEv.BytesDown == 0 { + t.Errorf("response event BytesDown = 0, want the relayed body counted — the row is recorded after the copy") + } +} + // TestInboundSessionID_NoActiveSessionFallback locks in the // reverseproxy's bucketing rule: when the A2A request has no // contextId (first turn of a conversation), inboundSessionID returns diff --git a/docs/session-api.md b/docs/session-api.md index 51320f007..ba836e549 100644 --- a/docs/session-api.md +++ b/docs/session-api.md @@ -255,7 +255,10 @@ Every event on `/v1/sessions/{id}` and `/v1/events` carries: the upstream's response **headers** arriving. `durationMs` − `ttfbMs` is then how long the body took — for a streamed completion, the stream time. The pair separates a slow-to-answer origin from one that answered and then trickled, - which `durationMs` alone cannot (#1251). Absent when the exchange failed + which `durationMs` alone cannot (#1251). The row is recorded when the listener + is finished with the body — after the read on a buffered path, after the relay + on a streamed or unbuffered one — so the difference really is the body's time. + Absent when the exchange failed before headers (the 502 cases above), on request rows, and on tunnel rows — an opaque tunnel has no headers to time. agentop's detail pane renders the split as a `timing:` line. From ea8de741fbc046b5028532b7bde35b9a43d75800 Mon Sep 17 00:00:00 2001 From: Mariusz Sabath Date: Wed, 28 Oct 2026 21:18:04 -0400 Subject: [PATCH 4/4] fix: wait for deferred response rows in tests, omit extproc ttfb on the unread arm MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review findings on #1532: - TestReverseProxy_RecordsInboundAllowWithoutA2A read the store right after the client returned, but the row it asserts on is deferred to recordOnEOFBody — appended while ReverseProxy finishes its copy. That was the Go CI (core) flake (~1 in 200-500 locally). The test now waits (waitForResponseEvent); so do the other tests the row move exposed: reverseproxy's unbuffered-duration test and, in forwardproxy, the four server_test round trips, bytesOf, both resend tests and the tool-prune chain — with waitForResponseEventCount for the multi-row resend shapes, where waiting for the first response row would return while the last attempt's is still pending behind the relay. - extproc's headers-recorded arm (!NeedsBody, body still coming) had the misreading df9ce704 fixed for the proxies: Envoy relays the body unseen, Duration stops at the headers, and ttfb == duration would read as "body 0ms". ResponseHeadersAt is now left unset on that arm so ttfb is absent — the contract's "not measured" — pinned in both directions by ttfb_headers_test.go. - connectDialTimeout's comment counted the mTLS dialer's plain dial twice; dropped the last clause. - CLAUDE.md's bytesDown bullet and docs/session-api.md's bytes paragraph still said a relayed response reports zero and that every arm agrees; both now state the post-df9ce704 split, and the ttfb docs carry the carve-out for extproc's body-free arm. Signed-off-by: Mariusz Sabath Assisted-By: Claude (Anthropic AI) --- CLAUDE.md | 7 +- core/listener/extproc/server.go | 22 ++- core/listener/extproc/ttfb_headers_test.go | 144 ++++++++++++++++++ core/listener/forwardproxy/bytescount_test.go | 5 + .../forwardproxy/chainedwriters_test.go | 3 + core/listener/forwardproxy/resend_test.go | 7 +- core/listener/forwardproxy/server.go | 5 +- core/listener/forwardproxy/server_test.go | 21 +++ .../forwardproxy/upstreamfailure_test.go | 22 +++ core/listener/reverseproxy/recording_test.go | 21 +-- core/listener/reverseproxy/server_test.go | 6 + docs/session-api.md | 12 +- 12 files changed, 251 insertions(+), 24 deletions(-) create mode 100644 core/listener/extproc/ttfb_headers_test.go diff --git a/CLAUDE.md b/CLAUDE.md index 15d4c8e2b..c12100e7b 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -628,11 +628,14 @@ Every event on `/v1/sessions/{id}` and `/v1/events` carries: (`pipeline.DurationBetween`). The row itself is recorded when the listener is finished with the body — after the read on a buffered path, after the relay on a streamed or unbuffered one — so the difference really is the body's time and not a - remainder cut short at the headers. agentop's detail pane renders the split as a + remainder cut short at the headers. The one arm that breaks the pattern is extproc + with a body-free pipeline: the row is recorded at the headers while Envoy relays the + body unseen, so ttfb is left absent there rather than reported as a figure equal to + the whole duration. agentop's detail pane renders the split as a `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, and so does a response relayed straight through — both proxies copy that one after the row is appended. 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. All 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 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. - `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 (`tunnelReason: "dial-failed"`) — 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. - `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. - `subagent` — present on rows from a Claude Code subagent or agent-team teammate: the `X-Claude-Code-Agent-Id` the request carried. For a subagent that is also the `` in `~/.claude/projects///subagents/agent-.meta.json`; for a teammate the join is unverified. Resolved by `Context.Subagent`, pinned with the User-Agent by `ResolveClient` so no plugin can re-file a row, and **refused rather than sanitised** when unusable (control runes, over 128 bytes), because it is a join key and a cleaned-up id joins nothing. It does not move the row: a subagent's requests stay in their parent's session, so the header must not be added to `session.id_headers`. **It names an agent, it does not classify the row:** `inference.agentRole` stays the field that says whether a row is a subagent's, and a teammate's row is `agentRole: main` with `subagent` set — Claude Code writes `cc_is_subagent=true` only for `agentType: subagent`. diff --git a/core/listener/extproc/server.go b/core/listener/extproc/server.go index f1867fe28..52d827302 100644 --- a/core/listener/extproc/server.go +++ b/core/listener/extproc/server.go @@ -667,18 +667,28 @@ func (s *Server) handleResponseHeaders(ctx context.Context, headers *corev3.Head statusStr := getHeader(headers, ":status") pctx.StatusCode, _ = strconv.Atoi(statusStr) pctx.ResponseHeaders = headerMapToHTTP(headers) - // Envoy delivering response headers is this listener's headers-arrival - // moment — the same point in the exchange where the proxy listeners set - // theirs — so ttfb means one thing on a row from any listener (#1251). - // The body phase and end-of-stream recording run later, which is what - // makes this headers-arrival rather than body-completion. - pctx.ResponseHeadersAt = time.Now() p := s.OutboundPipeline if direction == "inbound" { p = s.InboundPipeline } + // Envoy delivering response headers is this listener's headers-arrival + // moment — the same point in the exchange where the proxy listeners set + // theirs — so ttfb means one thing on a row from any listener (#1251). + // Set it, though, only when the row is NOT recorded right here while the + // body is still coming: with !p.NeedsBody() the body phase below is never + // requested, Envoy relays the body without us, and the row lands at + // headers time — so ttfb ≈ duration and duration−ttfb would read as + // "body 0ms" for a body this listener never saw, the exact misreading + // #1251's split exists to prevent. Absent is the contract's "not + // measured", so leave the timestamp zero on that arm. endOfStream means + // no body is coming at all, and p.NeedsBody() means the body phase + // records later — in both, headers arrival is a real, reportable figure. + if p.NeedsBody() || endOfStream { + pctx.ResponseHeadersAt = time.Now() + } + // Defer to the body phase — but only when there IS one. // // AND !endOfStream IS THE FIX, not a refinement. NeedsBody() alone is unconditionally true for diff --git a/core/listener/extproc/ttfb_headers_test.go b/core/listener/extproc/ttfb_headers_test.go new file mode 100644 index 000000000..c05326e6e --- /dev/null +++ b/core/listener/extproc/ttfb_headers_test.go @@ -0,0 +1,144 @@ +package extproc + +import ( + "context" + "testing" + + extprocv3 "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3" + + "github.com/rossoctl/cortex/core/pipeline" + "github.com/rossoctl/cortex/core/session" +) + +// ttfbProbePlugin publishes one plugin-public Custom entry and asks for no +// body in either direction, so a pipeline holding it has NeedsBody() false — +// the arm where handleResponseHeaders records the response row itself because +// the body phase is never requested. The entry is what makes the row exist at +// all: recordOutboundResponseSession gates on a protocol payload, an +// Invocation or a Custom entry, and this test needs the row without any of +// the body-reading plugins that would move the recording to the body phase. +type ttfbProbePlugin struct{} + +func (ttfbProbePlugin) Name() string { return "ttfb-probe" } +func (ttfbProbePlugin) Capabilities() pipeline.PluginCapabilities { + return pipeline.PluginCapabilities{} +} +func (ttfbProbePlugin) OnRequest(_ context.Context, _ *pipeline.Context) pipeline.Action { + return pipeline.Action{Type: pipeline.Continue} +} +func (ttfbProbePlugin) OnResponse(_ context.Context, pctx *pipeline.Context) pipeline.Action { + if pctx.Extensions.Custom == nil { + pctx.Extensions.Custom = map[string]any{} + } + pctx.Extensions.Custom["ttfb-probe/event"] = map[string]any{"seen": true} + return pipeline.Action{Type: pipeline.Continue} +} + +// newTtfbProbeServer builds a Server whose outbound pipeline holds only the +// probe above, so a response is recorded on the headers path rather than the +// body phase. +func newTtfbProbeServer(t *testing.T) (*Server, *session.Store) { + t.Helper() + p, err := pipeline.New([]pipeline.Plugin{ttfbProbePlugin{}}) + if err != nil { + t.Fatalf("pipeline.New: %v", err) + } + if p.NeedsBody() { + t.Fatal("NeedsBody() = true for a pipeline holding only the probe; this file's premise is the body-free arm") + } + empty, err := pipeline.New(nil) + if err != nil { + t.Fatalf("pipeline.New(nil): %v", err) + } + store := session.New(0, 100, 100) + t.Cleanup(func() { store.Close() }) + return &Server{ + OutboundPipeline: pipeline.NewHolder(p), + InboundPipeline: pipeline.NewHolder(empty), + Sessions: store, + }, store +} + +// ttfbProbeStream drives one outbound exchange: a bodyless request, then +// response headers with the given end_of_stream. Driving Server.Process is +// the listener's real message loop, same choice as server_headeronly_test.go. +func ttfbProbeStream(srv *Server, endOfStream bool) *mockStream { + reqs := []*extprocv3.ProcessingRequest{ + {Request: &extprocv3.ProcessingRequest_RequestHeaders{ + RequestHeaders: &extprocv3.HttpHeaders{ + Headers: makeHeaders( + "x-authbridge-direction", "outbound", + ":method", "GET", + ":path", "/health", + ":authority", "internal.local", + ), + EndOfStream: true, + }, + }}, + {Request: &extprocv3.ProcessingRequest_ResponseHeaders{ + ResponseHeaders: &extprocv3.HttpHeaders{ + Headers: makeHeaders( + ":status", "200", + "content-type", "application/json", + "content-length", "2", + ), + EndOfStream: endOfStream, + }, + }}, + } + return &mockStream{ctx: context.Background(), requests: reqs} +} + +// TestExtProc_HeadersRecordedRowWithBodyComingOmitsTtfb pins the arm where a +// body-free pipeline makes handleResponseHeaders record the row itself while +// the body is still coming: Envoy relays it without this listener ever seeing +// the bytes, so Duration stops at the headers and a ttfb measured there would +// equal Duration — agentop would render "body 0ms" for a body that crossed +// Envoy, the exact misreading #1251's split exists to prevent. The row +// reports NO ttfb instead: absent is the contract's "not measured". +func TestExtProc_HeadersRecordedRowWithBodyComingOmitsTtfb(t *testing.T) { + srv, store := newTtfbProbeServer(t) + + stream := ttfbProbeStream(srv, false) // headers only; the body is still coming + _ = srv.Process(stream) + if stream.recvIdx != 2 { + t.Fatalf("consumed %d of 2 messages; the listener bailed", stream.recvIdx) + } + // No ModeOverride on the response-headers reply is what puts this exchange + // on the arm under test: the body phase was never requested, so the row was + // recorded at the headers. + if mo := stream.responses[1].GetModeOverride(); mo != nil { + t.Fatalf("the response-header phase asked Envoy for a body (%v); this test's premise is the body-free arm", mo.GetResponseBodyMode()) + } + + ev := responseEvent(t, store) + if ev == nil { + t.Fatal("no outbound response row recorded; the probe's Custom entry should have qualified it") + } + if ev.Ttfb != 0 { + t.Errorf("response row ttfb = %v, want absent (0) — the body is relayed by Envoy and never seen here, so duration−ttfb would read as a zero body it did not measure", ev.Ttfb) + } +} + +// TestExtProc_HeadersRecordedRowOnEndOfStreamCarriesTtfb is the positive +// control: when the headers END the stream there is no body at all, so +// headers arrival is the whole answer and the row carries a real ttfb — +// distinguishing "not measured" (above) from "measured, and it is +// everything". +func TestExtProc_HeadersRecordedRowOnEndOfStreamCarriesTtfb(t *testing.T) { + srv, store := newTtfbProbeServer(t) + + stream := ttfbProbeStream(srv, true) // headers end the stream: a bodyless response + _ = srv.Process(stream) + if stream.recvIdx != 2 { + t.Fatalf("consumed %d of 2 messages; the listener bailed", stream.recvIdx) + } + + ev := responseEvent(t, store) + if ev == nil { + t.Fatal("no outbound response row recorded; the probe's Custom entry should have qualified it") + } + if ev.Ttfb <= 0 { + t.Errorf("response row ttfb = %v, want > 0 — the headers ended the stream, so their arrival is a real, reportable figure", ev.Ttfb) + } +} diff --git a/core/listener/forwardproxy/bytescount_test.go b/core/listener/forwardproxy/bytescount_test.go index a8d18843a..0d74c49ec 100644 --- a/core/listener/forwardproxy/bytescount_test.go +++ b/core/listener/forwardproxy/bytescount_test.go @@ -54,6 +54,11 @@ func bytesOf(t *testing.T, plugins []pipeline.Plugin, req *http.Request, upstrea // undrained body would under-report by whatever was still in flight. drainAndClose(t, resp) + // The row is recorded after the relay (io.Copy or the streaming arms' loop), + // which can finish after the client has drained and returned — wait for it + // rather than racing the recording. + waitForResponseEvent(t, store) + v := store.View(session.DefaultSessionID) if v == nil || len(v.Events) != 2 { t.Fatalf("want a request row and a response row, got %+v", v) diff --git a/core/listener/forwardproxy/chainedwriters_test.go b/core/listener/forwardproxy/chainedwriters_test.go index 01439d420..efde6263c 100644 --- a/core/listener/forwardproxy/chainedwriters_test.go +++ b/core/listener/forwardproxy/chainedwriters_test.go @@ -214,6 +214,9 @@ func TestForwardProxy_ToolPruneAndTheRouterChain(t *testing.T) { } // The session: one request event and one response event for the request. + // Two response rows land (glm's refusal, then the answer), the last after + // the relay — wait for both rather than racing the recording. + waitForResponseEventCount(t, store, 2) v := store.View("s1") if v == nil { t.Fatal("no session s1 recorded") diff --git a/core/listener/forwardproxy/resend_test.go b/core/listener/forwardproxy/resend_test.go index 87ddfca53..fe95e57b5 100644 --- a/core/listener/forwardproxy/resend_test.go +++ b/core/listener/forwardproxy/resend_test.go @@ -208,6 +208,9 @@ func TestResend_BothAttemptsAreRecordedAndPairByRequestID(t *testing.T) { defer backend.Close() _, _, store := postVia(t, &substituter{target: backend.URL, sub: "glm-5-3", resend: true}, "exo-free") + // Two response rows (the refusal and the answer), the second recorded + // after the relay — wait for both rather than racing the recording. + waitForResponseEventCount(t, store, 2) v := store.View(session.DefaultSessionID) if v == nil || len(v.Events) != 4 { t.Fatalf("rows = %+v, want request, refusal, re-sent request, answer", v) @@ -353,7 +356,9 @@ func TestResend_ASecondRoundReachesTheThirdAttempt(t *testing.T) { } // Request, refusal, request, refusal, request, answer: each attempt paired with // its own answer, as TestResend_BothAttemptsAreRecordedAndPairByRequestID checks - // for one round. + // for one round. Three response rows, the last recorded after the relay — + // wait for all three rather than racing the recording. + waitForResponseEventCount(t, store, 3) v := store.View(session.DefaultSessionID) if v == nil || len(v.Events) != 6 { t.Fatalf("rows = %+v, want three requests and three answers", v) diff --git a/core/listener/forwardproxy/server.go b/core/listener/forwardproxy/server.go index 3a2cc9e73..8438c98bf 100644 --- a/core/listener/forwardproxy/server.go +++ b/core/listener/forwardproxy/server.go @@ -2080,9 +2080,8 @@ func (s *Server) recordOutboundRejectIn(tl *tunnelLog, pctx *pipeline.Context, a // unanswered; a handshake that misses five straight tries is not being saved // by a sixth at 16s, so the rest of the budget bought no connection, only // latency before the 502. Matches the budget the bridge already trusts for -// its own upstream probe (upstreamVerifyTimeout), the mTLS dialer, which had -// settled on 10s independently, and this file's own plain dialer — the 30s -// CONNECT dial was the outlier in its own file. +// its own upstream probe (upstreamVerifyTimeout) and the mTLS dialer's plain +// dial, which had settled on 10s independently. const connectDialTimeout = 10 * time.Second // handleConnect tunnels HTTPS (and any other TLS-wrapped protocol) through diff --git a/core/listener/forwardproxy/server_test.go b/core/listener/forwardproxy/server_test.go index 9e328ba82..f6cb0f3e6 100644 --- a/core/listener/forwardproxy/server_test.go +++ b/core/listener/forwardproxy/server_test.go @@ -695,6 +695,13 @@ func TestForwardProxy_RecordsMessageWithNoPluginActivity(t *testing.T) { t.Errorf("status = %d, want 404", resp.StatusCode) } + // Empty pipeline: the row is recorded after the final io.Copy, which can + // finish after the client has read the body and returned — wait for it + // rather than racing the recording (waitForResponseEvent, + // upstreamfailure_test.go). The request row is appended before the + // exchange, so it is on the view once the response row is. + waitForResponseEvent(t, store) + v := store.View(session.DefaultSessionID) if v == nil || len(v.Events) != 2 { t.Fatalf("expected 2 events (request + response) with no plugin activity, got %+v", v) @@ -750,6 +757,11 @@ func TestForwardProxy_ResponseEventCarriesTtfb(t *testing.T) { io.Copy(io.Discard, resp.Body) resp.Body.Close() + // The row is recorded after the relay ends, which can be after the client + // has read the stream and returned — wait for it rather than racing the + // recording. + waitForResponseEvent(t, store) + v := store.View(session.DefaultSessionID) if v == nil || len(v.Events) != 2 { t.Fatalf("expected 2 events (request + response), got %+v", v) @@ -812,6 +824,11 @@ func TestForwardProxy_ResponseEventDurationIncludesUnbufferedBody(t *testing.T) io.Copy(io.Discard, resp.Body) resp.Body.Close() + // The row is recorded after the final io.Copy, which can finish after the + // client has read the body and returned — wait for it rather than racing + // the recording. + waitForResponseEvent(t, store) + v := store.View(session.DefaultSessionID) if v == nil || len(v.Events) != 2 { t.Fatalf("expected 2 events (request + response), got %+v", v) @@ -870,6 +887,10 @@ func TestForwardProxy_ResponseEventOmitsTtfbOnFailedExchange(t *testing.T) { t.Fatalf("status = %d, want 502", resp.StatusCode) } + // The 502 row is appended just after the response is written, so the + // client can beat it — same wait as the success-path tests. + waitForResponseEvent(t, store) + v := store.View(session.DefaultSessionID) if v == nil || len(v.Events) != 2 { t.Fatalf("expected 2 events (request + response), got %+v", v) diff --git a/core/listener/forwardproxy/upstreamfailure_test.go b/core/listener/forwardproxy/upstreamfailure_test.go index cc1afeaa1..4eeb829f4 100644 --- a/core/listener/forwardproxy/upstreamfailure_test.go +++ b/core/listener/forwardproxy/upstreamfailure_test.go @@ -588,6 +588,28 @@ func waitForResponseEvent(t *testing.T, store *session.Store) pipeline.SessionEv return pipeline.SessionEvent{} } +// waitForResponseEventCount polls until at least n response-phase events are +// recorded — the multi-attempt shape of a resend, where waiting for the first +// (waitForResponseEvent) would return while the last attempt's row is still +// pending behind the relay. +func waitForResponseEventCount(t *testing.T, store *session.Store, n int) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + count := 0 + for _, ev := range allEvents(t, store) { + if ev.Phase == pipeline.SessionResponse { + count++ + } + } + if count >= n { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatalf("fewer than %d response events recorded; events = %+v", n, allEvents(t, store)) +} + // truncatingBackend returns the URL of an upstream that promises a 1000-byte // body, sends 11 bytes of it, and closes — a real mid-body failure, read through // the real transport. diff --git a/core/listener/reverseproxy/recording_test.go b/core/listener/reverseproxy/recording_test.go index 69b0791f5..70ab4a927 100644 --- a/core/listener/reverseproxy/recording_test.go +++ b/core/listener/reverseproxy/recording_test.go @@ -69,6 +69,14 @@ func TestReverseProxy_RecordsInboundAllowWithoutA2A(t *testing.T) { } resp.Body.Close() + // No plugin reads bodies here, so the response row is deferred to + // recordOnEOFBody — appended while ReverseProxy finishes its copy, which + // can be after the client has already read the body and returned. Wait + // for the row rather than racing the recording (waitForResponseEvent, + // upstreamfailure_test.go); the request row is appended before the + // exchange, so it is on the view once the response row is. + respEvent := waitForResponseEvent(t, store) + // No A2A means no contextId, so the event lands in the default bucket. v := store.View(session.DefaultSessionID) if v == nil { @@ -105,16 +113,9 @@ func TestReverseProxy_RecordsInboundAllowWithoutA2A(t *testing.T) { // The response-phase gate in modifyResponse is widened the same way. // Assert the inbound response event was recorded with its response-phase // invocation, locking in the second gate against regression too. - var respEvent *pipeline.SessionEvent - for i := range v.Events { - ev := v.Events[i] - if ev.Direction == pipeline.Inbound && ev.Phase == pipeline.SessionResponse { - respEvent = &v.Events[i] - break - } - } - if respEvent == nil { - t.Fatalf("no inbound response event recorded; events=%+v", v.Events) + // respEvent is the row waitForResponseEvent handed us above. + if respEvent.Direction != pipeline.Inbound { + t.Fatalf("response event direction = %v, want Inbound", respEvent.Direction) } if respEvent.Invocations == nil || len(respEvent.Invocations.Inbound) == 0 { t.Fatalf("response event has no inbound invocations: %+v", respEvent) diff --git a/core/listener/reverseproxy/server_test.go b/core/listener/reverseproxy/server_test.go index 783cc6674..1a9a35846 100644 --- a/core/listener/reverseproxy/server_test.go +++ b/core/listener/reverseproxy/server_test.go @@ -761,6 +761,12 @@ func TestReverseProxy_ResponseEventDurationIncludesUnbufferedBody(t *testing.T) io.Copy(io.Discard, resp.Body) resp.Body.Close() + // The row is deferred to the end of ReverseProxy's copy, which can finish + // after the client has read the body and returned — wait for it rather + // than racing the recording (see recording_test.go for the flake this + // shape produced in CI). + waitForResponseEvent(t, store) + v := store.View(session.DefaultSessionID) if v == nil || len(v.Events) != 2 { t.Fatalf("expected 2 events (request + response) in the default bucket, got %+v", v) diff --git a/docs/session-api.md b/docs/session-api.md index ba836e549..3584bb864 100644 --- a/docs/session-api.md +++ b/docs/session-api.md @@ -258,7 +258,10 @@ Every event on `/v1/sessions/{id}` and `/v1/events` carries: which `durationMs` alone cannot (#1251). The row is recorded when the listener is finished with the body — after the read on a buffered path, after the relay on a streamed or unbuffered one — so the difference really is the body's time. - Absent when the exchange failed + The one exception is the envoy-sidecar listener with a pipeline no plugin of + which reads bodies: the row is recorded at the headers while Envoy relays the + body unseen, and ttfb is left absent there rather than reported as a figure + equal to the whole duration. Absent when the exchange failed before headers (the 502 cases above), on request rows, and on tunnel rows — an opaque tunnel has no headers to time. agentop's detail pane renders the split as a `timing:` line. @@ -274,7 +277,12 @@ Every event on `/v1/sessions/{id}` and `/v1/events` carries: destination. 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** — every listener buffers bodies only when a plugin asks, so an - unbuffered request reports zero. + unbuffered request reports zero. A response relayed straight through is + counted, not zero: both proxies record the row after they finish relaying + it (#1251), so the bytes that crossed their own process are on the row. The + one arm that still reports zero is the envoy-sidecar listener with a + body-free pipeline — it asked Envoy not to send it the bytes, so zero there + means "not asked to see", not "not counted". - `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 the close (`phase: