diff --git a/CLAUDE.md b/CLAUDE.md index 9a0889c3c..c12100e7b 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -616,9 +616,26 @@ 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`). 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. 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/cmd/agentop/tui/detail_pane.go b/cmd/agentop/tui/detail_pane.go index 036073bdc..0dbd31d80 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,36 @@ 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 + } + // 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), bodyText, 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..519f5ae5e 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,51 @@ 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) + } +} + +// 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). +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..52d827302 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(), @@ -671,6 +673,22 @@ func (s *Server) handleResponseHeaders(ctx context.Context, headers *corev3.Head 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 86d7bece1..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) @@ -124,18 +129,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 +149,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/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 ca41b2603..8438c98bf 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 @@ -587,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. @@ -718,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 { @@ -739,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 @@ -1654,6 +1673,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 +2070,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's plain +// dial, 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..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) @@ -711,6 +718,197 @@ 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() + + // 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) + } + 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_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() + + // 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) + } + 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 +// 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) + } + + // 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) + } + 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/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/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/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.go b/core/listener/reverseproxy/server.go index 0268abc51..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" @@ -487,6 +488,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 @@ -496,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) } @@ -623,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 @@ -651,9 +673,21 @@ 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), }) } + // 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 } @@ -844,10 +878,38 @@ 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), }) } +// 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..1a9a35846 100644 --- a/core/listener/reverseproxy/server_test.go +++ b/core/listener/reverseproxy/server_test.go @@ -720,6 +720,76 @@ 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() + + // 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) + } + 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/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/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 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..3584bb864 100644 --- a/docs/session-api.md +++ b/docs/session-api.md @@ -251,6 +251,20 @@ 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). 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. + 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. - `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`. @@ -263,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: