From 51e675acb07668fcdc73704ad4a6c6fb0cad6f89 Mon Sep 17 00:00:00 2001 From: Lukasz Raczylo Date: Sun, 23 Aug 2026 02:40:14 +0100 Subject: [PATCH] fix(providers): address adversarial review findings F1, F4, F6, F9 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit F1: TestModelRegistry_MaybeRefresh_Throttled_OnePerInterval mutated a plain `now` variable while a background refresh goroutine read it through nowFn, racing under -race in about 14% of full-suite runs once this branch's timer-heavy tests shared the parallel phase. Guards the clock with a mutex instead. registry.go is untouched. F4: a mid-body watchdog timeout never reached provider-health accounting. limiter.recordProviderAttempt only ran at header-arrival time, so a provider that reliably stalled mid-body reported attempts=1 failures=0 forever. watchdogBody.fire now invokes the same attemptRecorder a second time, as a failure, when it fires. Reuses the existing recorder rather than adding new plumbing; this does mean two attempts are recorded for one logical request (one success at header time, one failure at watchdog-fire time), not one attempt marked failed outright. F6: watchdogBody.Read reset the timer on any err == nil read, regardless of n. A reader returning (0, nil) forever would have held the watchdog open indefinitely. Reset now only fires when n > 0. F9 (three items): - proxyUpstream passed logPrefix as a bare providerName, which newWatchdogBody wrapped in a hardcoded `provider %q` — labeling an MCP/A2A target as a provider. watchdogBody now takes an already-formatted label from the caller instead. - errUpstream was wrapped twice: once in watchdogBody's own error, once again by every caller that already wraps it. watchdogBody now wraps only errProviderTimeout; callers' existing wraps still make errors.Is(err, errUpstream) hold. - timeout_test.go read wb.timedOut directly in one place while every other access is atomic; switched to atomic.LoadInt32. Also rewords errProviderTimeout's message from "provider request timed out" to "upstream progress stalled" (finding F3, documented not fixed per coordinator ruling): the watchdog measures time since the last successful upstream read, which can go quiet because of a slow client write between reads, not only a slow provider. The new wording no longer asserts which side caused it. New tests: TestChatCompletion_Stream_StallsMidBody_RecordsProviderFailure, TestWatchdogBody_ZeroByteRead_DoesNotResetTimer, TestWatchdogBody_TimeoutError_UsesCallerLabel_NotHardcodedProvider. Each mutation-verified per its own doc comment. --- providers.go | 13 ++- registry_test.go | 24 +++++- routes_passthrough.go | 15 +++- timeout.go | 138 +++++++++++++++++++++++-------- timeout_test.go | 185 +++++++++++++++++++++++++++++++++++++++++- 5 files changed, 332 insertions(+), 43 deletions(-) diff --git a/providers.go b/providers.go index 9e87b1d..e440abd 100644 --- a/providers.go +++ b/providers.go @@ -310,6 +310,17 @@ func upstreamRawBytes(ctx context.Context, client *http.Client, method, url stri // "application/json" only when bodyBytes is non-nil and hdr set none of // its own. func upstreamBytes(ctx context.Context, client *http.Client, method, url string, hdr http.Header, bodyBytes []byte, policy *retryPolicy, timeout time.Duration, providerName string) (*http.Response, error) { + // rec is looked up once, outside the retry loop's call closure: every + // attempt shares the identical ctx, so the recorder (nil when no + // caller ever wired one via withAttemptRecorder) is the same across + // attempts. watchdogBody uses it too (coordinator adversarial review, + // 2026-08-23, finding F4): retryPolicy.do's own rec(resp, err) call + // (retry.go) already ran, at header-arrival time, before any byte of + // the body was read — a mid-body stall was otherwise invisible to + // provider-health accounting, since nothing else ever called rec + // again for that attempt. + rec := attemptRecorderFromContext(ctx) + call := func() (*http.Response, error) { var r io.Reader if bodyBytes != nil { @@ -355,7 +366,7 @@ func upstreamBytes(ctx context.Context, client *http.Client, method, url string, cancel() return nil, fmt.Errorf("%w: %w", errUpstream, err) } - resp.Body = newWatchdogBody(resp.Body, cancel, timeout, providerName) + resp.Body = newWatchdogBody(resp.Body, cancel, timeout, fmt.Sprintf("provider %q", providerName), rec) return resp, nil } diff --git a/registry_test.go b/registry_test.go index 08e9272..e92ff58 100644 --- a/registry_test.go +++ b/registry_test.go @@ -505,8 +505,26 @@ func TestModelRegistry_MaybeRefresh_Throttled_OnePerInterval(t *testing.T) { if err != nil { t.Fatalf("newModelRegistry: %v", err) } - now := time.Now() - reg.nowFn = func() time.Time { return now } + // clockMu guards clock: reg.nowFn runs on maybeRefresh's own + // background refresh goroutine (registry.go), concurrently with this + // test's own goroutine advancing the clock below a few lines down. A + // plain `now` variable closed over by both was a genuine data race + // (coordinator adversarial review, 2026-08-23) — the 10ms sleep below + // gave the background goroutine a chance to finish reading it first, + // but that is a race won by scheduling luck, not a happens-before + // guarantee, and -race caught it in roughly 14% of full-suite runs. + var clockMu sync.Mutex + clock := time.Now() + reg.nowFn = func() time.Time { + clockMu.Lock() + defer clockMu.Unlock() + return clock + } + advanceClock := func(d time.Duration) { + clockMu.Lock() + clock = clock.Add(d) + clockMu.Unlock() + } reg.maybeRefresh(context.Background()) waitUntil(t, time.Second, func() bool { return atomic.LoadInt32(&calls) == 1 }) @@ -518,7 +536,7 @@ func TestModelRegistry_MaybeRefresh_Throttled_OnePerInterval(t *testing.T) { t.Errorf("listModels called %d times within one interval, want 1", got) } - now = now.Add(2 * time.Hour) // past the interval: next entry should refetch + advanceClock(2 * time.Hour) // past the interval: next entry should refetch reg.maybeRefresh(context.Background()) waitUntil(t, time.Second, func() bool { return atomic.LoadInt32(&calls) == 2 }) } diff --git a/routes_passthrough.go b/routes_passthrough.go index 71c248d..d0f33dd 100644 --- a/routes_passthrough.go +++ b/routes_passthrough.go @@ -1014,8 +1014,13 @@ func (g *Gateway) proxyUpstream(w http.ResponseWriter, r *http.Request, upstream // carries a recorder (attemptRecorderFromContext, providers.go) — // handleTargetProxy (mcp_a2a.go), this function's other caller, never // wraps r's context this way, so an MCP/A2A target proxy attempt is - // correctly never accounted as provider traffic. - if rec := attemptRecorderFromContext(r.Context()); rec != nil { + // correctly never accounted as provider traffic. rec is kept in scope + // (not just checked inline) so the watchdogBody below can reuse it + // too — a mid-body stall must reach the SAME provider-health + // accounting a build/send failure already does (coordinator + // adversarial review, 2026-08-23, finding F4). + rec := attemptRecorderFromContext(r.Context()) + if rec != nil { rec(resp, err) } if err != nil { @@ -1028,7 +1033,11 @@ func (g *Gateway) proxyUpstream(w http.ResponseWriter, r *http.Request, upstream writeOAIError(w, http.StatusBadGateway, "server_error", "upstream connection error") return proxyResult{}, false } - resp.Body = newWatchdogBody(resp.Body, cancel, timeout, logPrefix) + // logPrefix, not a hardcoded "provider %q": handleTargetProxy's own + // calls here (mcp_a2a.go) pass "mcp target (name ...)"/"a2a target + // (name ...)" — an MCP/A2A target is not a provider, and the watchdog + // error text must not claim it is (finding F9). + resp.Body = newWatchdogBody(resp.Body, cancel, timeout, logPrefix, rec) defer resp.Body.Close() //nolint:errcheck // read-side close; nothing actionable on failure copyHeadersExcept(w.Header(), resp.Header, hopByHopHeaders, dangerousResponseHeaders) diff --git a/timeout.go b/timeout.go index e020b70..b7a5913 100644 --- a/timeout.go +++ b/timeout.go @@ -30,12 +30,22 @@ const defaultRequestTimeout = 5 * time.Minute // disconnect, classified separately by handleAdapterErrorEnvelope // (routes_unified.go) and proxyUpstream (routes_passthrough.go), both of // which check errors.Is(err, context.Canceled) ahead of their generic -// upstream-error fallback — a watchdog firing is always this gateway -// giving up on an upstream that stopped making progress: a provider -// failure to attribute to that provider, never the client's own doing. -// watchdogBody.Read never lets a timeout error wrap context.Canceled, so -// the two stay mutually exclusive in every caller's classification. -var errProviderTimeout = errors.New("llmgateway: provider request timed out") +// upstream-error fallback — watchdogBody.Read never lets a timeout error +// wrap context.Canceled, so the two stay mutually exclusive in every +// caller's classification. +// +// The message deliberately says "upstream progress stalled", not +// "provider timed out" (coordinator adversarial review, 2026-08-23, +// finding F3): the watchdog measures time since this gateway's last +// successful READ from the upstream body, which can go quiet either +// because the provider itself stalled, or because this gateway spent the +// whole window blocked WRITING to a slow/stalled client between reads (a +// streaming forwarder reads one chunk, writes it out, then reads again — +// the write is not itself watchdog-guarded). Aborting in either case is +// still correct (holding an upstream connection open for a stalled +// client is its own problem), but asserting the PROVIDER specifically +// caused it is not always true, so the message no longer claims it. +var errProviderTimeout = errors.New("llmgateway: upstream progress stalled") // resolveRequestTimeout parses raw (a Go duration string, e.g. "5m" or // "90s" — matching RetryConfig.Backoff/CacheConfig.TTL's own convention) @@ -94,9 +104,9 @@ func resolveProviderTimeout(providerName, providerRaw string, globalTimeout time // Deliberately NOT http.Client.Timeout: that bounds the whole request // including the body read, which would truncate a healthy, long-running // stream — exactly what this feature must never do. A watchdogBody -// instead resets on every successful read, so a stream (or a large -// non-streaming body) that keeps making progress runs indefinitely; only -// SILENCE, never total duration, ends it. +// instead resets on every successful, non-empty read, so a stream (or a +// large non-streaming body) that keeps making progress runs indefinitely; +// only SILENCE, never total duration, ends it. // // The mechanism: cancel is the CancelFunc for the *http.Request's own // context, derived via context.WithCancel by the caller that built the @@ -105,25 +115,47 @@ func resolveProviderTimeout(providerName, providerRaw string, globalTimeout time // to fire timeout in the future again; if timer fires first, fire cancels // the request context, which unblocks whatever Read is currently in // flight (or the next one) with an error. Read then substitutes that -// error for one wrapping errProviderTimeout, naming providerName, instead -// of passing through whatever the canceled context's own error text says -// — a real client disconnect (the parent context canceling instead of -// this watchdog) never sets timedOut, so it is never misreported this way. +// error for one wrapping errProviderTimeout, naming label, instead of +// passing through whatever the canceled context's own error text says — +// a real client disconnect (the parent context canceling instead of this +// watchdog) never sets timedOut, so it is never misreported this way. // Close stops the timer (a no-op once already fired) and always calls // cancel, so this watchdog never leaves a live timer or an un-canceled // derived context behind on the hot path — every upstream request this // gateway makes. // +// onTimeout, when non-nil, is invoked once from fire (coordinator +// adversarial review, 2026-08-23, finding F4): a provider that sends +// headers and then stalls mid-body previously left provider-health +// accounting blind to the failure entirely — limiter.recordProviderAttempt +// is only ever reached from retryPolicy.do's own attemptRecorder call (at +// header-arrival time, before any body byte is read) and from +// proxyUpstream's identical call — so a chronic mid-body staller looked +// like 100% successful attempts to the discovery breaker and any +// request-path health signal that consumes those counters. onTimeout is +// that same attemptRecorder, reused rather than reinvented: callers look +// it up once via attemptRecorderFromContext and pass it straight through. +// This does mean a request that stalls mid-body is recorded as TWO +// attempts (one success at header time, one failure when the watchdog +// fires) rather than one attempt marked failed outright — deliberately: +// the alternative was deferring the FIRST recorder call until the body +// finishes, which would change when every other (already-tested, +// already-shipped) success path reports its attempt too, a much larger +// blast radius for what this fix needs. Two real, honestly-reported +// signals (headers arrived; the body then never finished) beats the +// previous zero. +// // timedOut is accessed via sync/atomic's function API (atomic.StoreInt32/ // LoadInt32), not the newer atomic.Bool type, per this plugin's // yaegi-interpreted-stdlib guardrail. type watchdogBody struct { - rc io.ReadCloser - cancel context.CancelFunc - timer *time.Timer - providerName string - timeout time.Duration - timedOut int32 + rc io.ReadCloser + cancel context.CancelFunc + timer *time.Timer + onTimeout attemptRecorder + label string + timeout time.Duration + timedOut int32 } // newWatchdogBody returns a watchdogBody wrapping rc, arming its watchdog @@ -132,8 +164,19 @@ type watchdogBody struct { // headers then stalls mid-body" is caught here, not left to // ResponseHeaderTimeout, which has already been satisfied by the time a // body exists to wrap). -func newWatchdogBody(rc io.ReadCloser, cancel context.CancelFunc, timeout time.Duration, providerName string) *watchdogBody { - wb := &watchdogBody{rc: rc, cancel: cancel, timeout: timeout, providerName: providerName} +// +// label names what this watchdog guards, already formatted by the +// caller: providers.go passes `provider "openai"`; routes_passthrough.go +// passes its own logPrefix verbatim (`passthrough (provider "openai")` or +// `mcp target (name "foo")`) — NOT reworded here into a hardcoded +// "provider %q", which previously mislabeled an MCP/A2A target as a +// provider (coordinator adversarial review, 2026-08-23, finding F9). +// onTimeout may be nil (no attemptRecorder was ever wired onto this +// request's context — the common case for discovery's own listModels +// calls and the MCP/A2A target proxy, neither of which Feature A +// accounts). +func newWatchdogBody(rc io.ReadCloser, cancel context.CancelFunc, timeout time.Duration, label string, onTimeout attemptRecorder) *watchdogBody { + wb := &watchdogBody{rc: rc, cancel: cancel, timeout: timeout, label: label, onTimeout: onTimeout} wb.timer = time.AfterFunc(timeout, wb.fire) return wb } @@ -142,32 +185,59 @@ func newWatchdogBody(rc io.ReadCloser, cancel context.CancelFunc, timeout time.D // no goroutine is ever spawned for a request whose timer is Stopped // before it fires, which is the common, non-timed-out case), the moment // timeout passes with no successful Read: it marks timedOut so the next -// Read error substitutes the provider-attributed timeout error, then -// cancels the request context, unblocking whatever Read is currently -// parked waiting on the network. +// Read error substitutes the timeout error, cancels the request context +// (unblocking whatever Read is currently parked waiting on the network), +// and — finding F4 — notifies onTimeout, if this watchdog has one, so +// provider-health accounting sees the failure even though no caller will +// ever read a body byte past this point. func (wb *watchdogBody) fire() { atomic.StoreInt32(&wb.timedOut, 1) wb.cancel() + if wb.onTimeout != nil { + wb.onTimeout(nil, wb.timeoutError()) + } } -// Read implements io.Reader. A successful read (err == nil) resets the -// watchdog for another full timeout window — this is what lets a stream -// that keeps producing chunks run indefinitely: only silence, never total -// duration, ends it. A failed read while timedOut is set (fire already -// ran) is reported as errProviderTimeout, naming providerName and the -// configured timeout, instead of whatever error canceling the request -// context actually produced. Any other error — a real client disconnect, -// a clean EOF, a genuine network failure that has nothing to do with this +// timeoutError builds the error Read returns once timedOut is set, and +// the error fire hands onTimeout — one construction, reused both places, +// so the two can never drift apart. Wraps errProviderTimeout only, not +// errUpstream: every caller that reads a body to completion already +// wraps errUpstream around whatever Read returns (forwardJSON/ +// forwardStream/forwardMediaBody/listModels, providers.go and the +// translate-aware adapters alike), so wrapping it again here produced a +// doubled "upstream request failed: ... upstream request failed: ..." +// message (finding F9) with no compensating benefit — errors.Is(err, +// errUpstream) still holds once the caller's own wrap runs. +func (wb *watchdogBody) timeoutError() error { + return fmt.Errorf("%w: %s: no upstream progress for %s", errProviderTimeout, wb.label, wb.timeout) +} + +// Read implements io.Reader. A successful, non-empty read (err == nil and +// n > 0) resets the watchdog for another full timeout window — this is +// what lets a stream that keeps producing chunks run indefinitely: only +// silence, never total duration, ends it. n == 0 with err == nil (a +// legal, if unusual, io.Reader response — not currently produced by +// net/http's own body reader against a non-empty buffer, but not +// forbidden by the io.Reader contract either) does NOT reset the timer +// (finding F6): resetting on a read that delivered nothing would let a +// reader that keeps returning (0, nil) hold the watchdog open forever, +// exactly the naive-watchdog hole this feature exists to close. A failed +// read while timedOut is set (fire already ran) is reported via +// timeoutError, instead of whatever error canceling the request context +// actually produced. Any other error — a real client disconnect, a clean +// EOF, a genuine network failure that has nothing to do with this // watchdog — passes through unchanged. func (wb *watchdogBody) Read(p []byte) (int, error) { n, err := wb.rc.Read(p) if err != nil { if atomic.LoadInt32(&wb.timedOut) == 1 { - return n, fmt.Errorf("%w: %w: provider %q: no upstream progress for %s", errUpstream, errProviderTimeout, wb.providerName, wb.timeout) + return n, wb.timeoutError() } return n, err } - wb.timer.Reset(wb.timeout) + if n > 0 { + wb.timer.Reset(wb.timeout) + } return n, nil } diff --git a/timeout_test.go b/timeout_test.go index c3ffdd3..cbce1eb 100644 --- a/timeout_test.go +++ b/timeout_test.go @@ -9,6 +9,8 @@ import ( "net/http/httptest" "runtime" "strings" + "sync" + "sync/atomic" "testing" "time" ) @@ -521,18 +523,25 @@ func (f *fakeReadCloser) Close() error { // (timeout.go) made this test fail — timedOut read back as 1 after the // sleep, since the un-stopped timer still fired on schedule. Reverted // before committing. +// +// wb.timedOut is read via atomic.LoadInt32 below, not a bare field read +// (coordinator adversarial review, 2026-08-23, finding F9): every other +// access in timeout.go itself is atomic, so a bare read here was a +// latent -race report waiting on a scheduling difference, even though +// Close's own Stop happens-before makes this specific read safe in +// practice. func TestWatchdogBody_Close_StopsTimer_FireNeverRuns(t *testing.T) { const timeout = 30 * time.Millisecond frc := newFakeReadCloser() ctx, cancel := context.WithCancel(context.Background()) - wb := newWatchdogBody(frc, cancel, timeout, "p1") + wb := newWatchdogBody(frc, cancel, timeout, "p1", nil) if err := wb.Close(); err != nil { t.Fatalf("Close: %v", err) } time.Sleep(3 * timeout) // well past when an un-stopped timer would fire - if got := wb.timedOut; got != 0 { + if got := atomic.LoadInt32(&wb.timedOut); got != 0 { t.Errorf("wb.timedOut = %d after Close, want 0 — the timer must not fire once stopped", got) } select { @@ -703,3 +712,175 @@ func TestNewGateway_TargetTimeout_ResolvedFromConfig(t *testing.T) { t.Errorf("targetClient Transport.ResponseHeaderTimeout = %v, want 90s", tr.ResponseHeaderTimeout) } } + +// TestChatCompletion_Stream_StallsMidBody_RecordsProviderFailure proves +// finding F4 (coordinator adversarial review, 2026-08-23): a mid-body +// watchdog timeout must reach provider-health accounting through the +// SAME attemptRecorder mechanism retryPolicy.do already uses at +// header-arrival time. Before this fix, limiter.recordProviderAttempt was +// only ever invoked once per request — at header time — so a provider +// that reliably stalls mid-body looked 100% successful to any consumer +// of those counters (the discovery breaker, cross-provider failover) +// forever: attempts=1 failures=0, identical to a real success. +// +// MUTATION VERIFIED: removing the `if wb.onTimeout != nil { ... }` block +// from watchdogBody.fire (timeout.go) made this test fail — the spy +// recorder was invoked exactly once (the header-time success), never a +// second time recording the failure. Reverted before committing. +func TestChatCompletion_Stream_StallsMidBody_RecordsProviderFailure(t *testing.T) { + const timeout = 60 * time.Millisecond + srv, release := newStallingServer(func(w http.ResponseWriter) { + w.Header().Set("Content-Type", "text/event-stream") + w.WriteHeader(http.StatusOK) + w.(http.Flusher).Flush() + }) + defer release() + + a := newTimeoutTestAdapter(srv.URL, timeout) + + type recorded struct { + resp *http.Response + err error + } + var mu sync.Mutex + var calls []recorded + rec := attemptRecorder(func(resp *http.Response, err error) { + mu.Lock() + calls = append(calls, recorded{resp, err}) + mu.Unlock() + }) + ctx := withAttemptRecorder(context.Background(), rec) + + rw := httptest.NewRecorder() + if _, err := a.chatCompletion(ctx, rw, map[string]any{"model": "gpt-5", "messages": []any{}, "stream": true}); err == nil { + t.Fatal("chatCompletion: want an error for a stream that stalls mid-body, got nil") + } + + mu.Lock() + defer mu.Unlock() + if len(calls) != 2 { + t.Fatalf("attemptRecorder invoked %d times, want 2 (header-time success, then watchdog-fire failure): %+v", len(calls), calls) + } + if calls[0].err != nil { + t.Errorf("first call err = %v, want nil (the header-arrival attempt succeeded)", calls[0].err) + } + if calls[1].resp != nil { + t.Errorf("second call resp = %v, want nil", calls[1].resp) + } + if !errors.Is(calls[1].err, errProviderTimeout) { + t.Errorf("second call err = %v, want errors.Is(err, errProviderTimeout)", calls[1].err) + } +} + +// zeroByteReadCloser always returns (0, nil) — a legal io.Reader response +// that delivers no bytes without signaling an error or EOF — until +// Close, which switches it to (0, io.EOF). Used by +// TestWatchdogBody_ZeroByteRead_DoesNotResetTimer (finding F6) to prove a +// Read that "succeeds" but delivers nothing does not keep the watchdog +// alive. +type zeroByteReadCloser struct { + closed chan struct{} +} + +func newZeroByteReadCloser() *zeroByteReadCloser { + return &zeroByteReadCloser{closed: make(chan struct{})} +} + +func (z *zeroByteReadCloser) Read([]byte) (int, error) { + select { + case <-z.closed: + return 0, io.EOF + default: + return 0, nil + } +} + +func (z *zeroByteReadCloser) Close() error { + select { + case <-z.closed: + default: + close(z.closed) + } + return nil +} + +// TestWatchdogBody_ZeroByteRead_DoesNotResetTimer proves finding F6 +// (coordinator adversarial review, 2026-08-23): watchdogBody.Read +// resetting on every err == nil read, regardless of n, let a reader that +// keeps returning (0, nil) hold the watchdog open forever — the +// coordinator's own probe measured this surviving past 20x the +// configured timeout across 1044 reads. Not reachable from net/http +// against a non-empty buffer today, but exactly the naive-watchdog hole +// this feature exists to close, so it is fixed and pinned regardless. +// +// This checks wb.timedOut directly rather than expecting wb.Read to +// surface an error: zeroByteReadCloser never inspects the watchdog's +// canceled context (unlike a real net.Conn, whose blocked Read a +// cancellation actually unblocks with an error), so it would return +// (0, nil) forever either way — the property under test is whether the +// watchdog's internal deadline advances, not what Read eventually +// returns. +// +// MUTATION VERIFIED: reverting watchdogBody.Read's `if n > 0 { ... }` +// guard (timeout.go) to an unconditional `wb.timer.Reset(wb.timeout)` +// made this test fail — timedOut stayed 0 for the full 10-timeout poll +// window, since every (0, nil) read kept re-arming the timer. Reverted +// before committing. +func TestWatchdogBody_ZeroByteRead_DoesNotResetTimer(t *testing.T) { + const timeout = 20 * time.Millisecond + zrc := newZeroByteReadCloser() + defer zrc.Close() + _, cancel := context.WithCancel(context.Background()) + wb := newWatchdogBody(zrc, cancel, timeout, "p1", nil) + defer wb.Close() + + buf := make([]byte, 16) + deadline := time.Now().Add(10 * timeout) + for time.Now().Before(deadline) && atomic.LoadInt32(&wb.timedOut) == 0 { + n, _ := wb.Read(buf) // always (0, nil) from this fixture, until Close + if n != 0 { + t.Fatalf("Read returned n=%d, want 0 (fixture bug)", n) + } + time.Sleep(time.Millisecond) + } + + if got := atomic.LoadInt32(&wb.timedOut); got != 1 { + t.Fatalf("wb.timedOut = %d after %s of continuous (0, nil) reads (timeout=%s), want 1 — a Read returning (0, nil) must not keep resetting the watchdog", got, 10*timeout, timeout) + } +} + +// TestWatchdogBody_TimeoutError_UsesCallerLabel_NotHardcodedProvider +// proves finding F9 (coordinator adversarial review, 2026-08-23): +// proxyUpstream (routes_passthrough.go) previously passed logPrefix as a +// bare "providerName" argument that watchdogBody wrapped in a hardcoded +// `provider %q` — producing a timeout error reading `provider +// "mcp target (name foo)"` for an MCP/A2A target proxy request, which is +// not a provider at all. watchdogBody now takes an already-formatted +// label from the caller and uses it verbatim. +// +// MUTATION VERIFIED: changing watchdogBody.timeoutError's format string +// back to `"%w: provider %q: no upstream progress for %s"` (timeout.go) +// made this test fail both assertions — the caller's label no longer +// appeared verbatim, and the hardcoded "provider \"mcp target" prefix +// reappeared. Reverted before committing. +func TestWatchdogBody_TimeoutError_UsesCallerLabel_NotHardcodedProvider(t *testing.T) { + const timeout = 20 * time.Millisecond + const label = `mcp target (name "foo")` + frc := newFakeReadCloser() + _, cancel := context.WithCancel(context.Background()) + wb := newWatchdogBody(frc, cancel, timeout, label, nil) + defer wb.Close() + + time.Sleep(3 * timeout) // let the watchdog fire + close(frc.unblock) // simulate the underlying connection erroring out once canceled + _, err := wb.Read(make([]byte, 16)) + if err == nil { + t.Fatal("Read: want an error once the watchdog has fired, got nil") + } + if !strings.Contains(err.Error(), label) { + t.Errorf("err = %v, want it to contain the caller's own label (%q) verbatim", err, label) + } + if strings.Contains(err.Error(), `provider "mcp target`) { + t.Errorf(`err = %v, must not wrap the label in a hardcoded "provider %%q" — an MCP/A2A target is not a provider`, err) + } +}