| 1 | package provider |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "io" |
| 8 | "net" |
| 9 | "net/http" |
| 10 | "strconv" |
| 11 | "strings" |
| 12 | "sync/atomic" |
| 13 | "syscall" |
| 14 | "time" |
| 15 | ) |
| 16 | |
| 17 | // errorBodyReadTimeout bounds how long draining a non-OK response body may |
| 18 | // block. Proxies and gateways under load (502/524 storms) can send headers and |
| 19 | // then stall the body on a half-open connection; http.Client has no Timeout |
| 20 | // and ResponseHeaderTimeout no longer applies once headers arrive, so without |
| 21 | // this deadline error reporting blocks in io.ReadAll indefinitely with no |
| 22 | // user-visible progress — the turn looks frozen until the process is killed |
| 23 | // (#6607). A var, not a const, so tests can shrink it. |
| 24 | var errorBodyReadTimeout = 10 * time.Second |
| 25 | |
| 26 | // SendOptions carries the per-request identity used to label failures. |
| 27 | type SendOptions struct { |
| 28 | Provider string // stable provider instance id |
| 29 | ProviderDisplayName string // user-editable display label |
| 30 | Protocol string // configured wire adapter id |
| 31 | KeyEnv string // api_key_env the key is read from, when known |
| 32 | KeySource string // human-readable source of KeyEnv, when known |
| 33 | KeyPresent bool // a non-empty key is being sent — separates "rejected" from "missing" |
| 34 | RetryAuth bool // retained for compatibility; authentication failures are terminal |
| 35 | } |
| 36 | |
| 37 | // RetryInfo is retained for callers of the retired transport retry callback. |
| 38 | type RetryInfo struct { |
| 39 | Attempt int |
| 40 | Max int |
| 41 | Delay time.Duration |
| 42 | Err error |
| 43 | } |
| 44 | |
| 45 | type RetryNotify func(RetryInfo) |
| 46 | |
| 47 | type retryNotifyKey struct{} |
| 48 | |
| 49 | type requestAttemptCounterKey struct{} |
| 50 | |
| 51 | type requestAttemptCounter struct { |
| 52 | count atomic.Int64 |
| 53 | } |
| 54 | |
| 55 | // WithRetryNotify retains compatibility with older callers. HTTP requests no |
| 56 | // longer retry automatically, so the callback is never invoked. |
| 57 | func WithRetryNotify(ctx context.Context, fn RetryNotify) context.Context { |
| 58 | if fn == nil { |
| 59 | return ctx |
| 60 | } |
| 61 | return context.WithValue(ctx, retryNotifyKey{}, fn) |
| 62 | } |
| 63 | |
| 64 | // WithRequestAttemptCounter returns a context that counts every HTTP request |
| 65 | // SendWithRetry starts. An existing counter is reused so a caller can observe |
| 66 | // attempts even when the provider returns before producing a Usage chunk. |
| 67 | // Provider implementations attach the count to usage, including explicit |
| 68 | // protocol/context repair requests made within the same logical model round. |
| 69 | func WithRequestAttemptCounter(ctx context.Context) context.Context { |
| 70 | if ctx == nil { |
| 71 | ctx = context.Background() |
| 72 | } |
| 73 | if counter, _ := ctx.Value(requestAttemptCounterKey{}).(*requestAttemptCounter); counter != nil { |
| 74 | return ctx |
| 75 | } |
| 76 | return context.WithValue(ctx, requestAttemptCounterKey{}, &requestAttemptCounter{}) |
| 77 | } |
| 78 | |
| 79 | // WithIndependentRequestAttemptCounter gives an auxiliary call its own usage |
| 80 | // count while preserving cancellation and other context values from its parent. |
| 81 | func WithIndependentRequestAttemptCounter(ctx context.Context) context.Context { |
| 82 | return context.WithValue(ctx, requestAttemptCounterKey{}, &requestAttemptCounter{}) |
| 83 | } |
| 84 | |
| 85 | // RequestAttemptCount returns the number of HTTP requests started through |
| 86 | // SendWithRetry for the counter attached to ctx. |
| 87 | func RequestAttemptCount(ctx context.Context) int { |
| 88 | if ctx == nil { |
| 89 | return 0 |
| 90 | } |
| 91 | counter, _ := ctx.Value(requestAttemptCounterKey{}).(*requestAttemptCounter) |
| 92 | if counter == nil { |
| 93 | return 0 |
| 94 | } |
| 95 | return int(counter.count.Load()) |
| 96 | } |
| 97 | |
| 98 | // ApplyRequestAttemptCount copies the stream's exact HTTP request count into a |
| 99 | // Usage record. Contexts without a counter leave the record unchanged so custom |
| 100 | // providers keep the zero-means-one compatibility contract. |
| 101 | func ApplyRequestAttemptCount(ctx context.Context, usage *Usage) { |
| 102 | if usage == nil { |
| 103 | return |
| 104 | } |
| 105 | if count := RequestAttemptCount(ctx); count > 0 { |
| 106 | usage.RequestCount = count |
| 107 | } |
| 108 | } |
| 109 | |
| 110 | // UsageWithRequestAttemptCount returns a copy of usage carrying the exact |
| 111 | // number of HTTP requests observed through ctx. When a provider request fails |
| 112 | // before producing token usage, it returns a request-only Usage record so |
| 113 | // callers can still account for the API calls. If neither usage nor attempts |
| 114 | // exist, it returns nil. |
| 115 | func UsageWithRequestAttemptCount(ctx context.Context, usage *Usage) *Usage { |
| 116 | count := RequestAttemptCount(ctx) |
| 117 | if usage == nil { |
| 118 | if count <= 0 { |
| 119 | return nil |
| 120 | } |
| 121 | return &Usage{RequestCount: count, Unknown: true} |
| 122 | } |
| 123 | result := *usage |
| 124 | if count > 0 { |
| 125 | result.RequestCount = count |
| 126 | } |
| 127 | return &result |
| 128 | } |
| 129 | |
| 130 | func recordRequestAttempt(ctx context.Context) { |
| 131 | if ctx == nil { |
| 132 | return |
| 133 | } |
| 134 | counter, _ := ctx.Value(requestAttemptCounterKey{}).(*requestAttemptCounter) |
| 135 | if counter != nil { |
| 136 | counter.count.Add(1) |
| 137 | } |
| 138 | } |
| 139 | |
| 140 | // APIError reports a non-OK HTTP status that isn't an auth failure. Status |
| 141 | // carries the code so the display layer can map it to an actionable, localized |
| 142 | // message; Body is a trimmed snippet of the response. |
| 143 | type APIError struct { |
| 144 | RetryAfter time.Duration // uncapped server delay for managed recovery |
| 145 | ShouldRetry string // explicit provider retry hint |
| 146 | Provider string // stable provider instance id |
| 147 | ProviderDisplayName string |
| 148 | Protocol string |
| 149 | Status int |
| 150 | Body string |
| 151 | TraceID string // provider trace identifier from the response headers, when present |
| 152 | RequestPath string // path only; query and URL userinfo are never retained |
| 153 | ToolContext string // resolved Reasonix/MCP identity for provider-indexed tool schema errors |
| 154 | } |
| 155 | |
| 156 | func (e *APIError) Error() string { |
| 157 | label := ProviderDisplayLabel(e.Provider, e.ProviderDisplayName, e.Protocol) |
| 158 | var base string |
| 159 | if e.Body == "" { |
| 160 | base = fmt.Sprintf("%s: status %d", label, e.Status) |
| 161 | } else { |
| 162 | base = fmt.Sprintf("%s: status %d: %s", label, e.Status, e.Body) |
| 163 | } |
| 164 | if e.ToolContext != "" { |
| 165 | return base + "\n" + e.ToolContext |
| 166 | } |
| 167 | return base |
| 168 | } |
| 169 | |
| 170 | // RetryableStatus reports whether a backoff can plausibly recover from status s: |
| 171 | // 408 (request timeout), 429 (rate limit) and 5xx (incl. Anthropic's 529). Other |
| 172 | // 4xx (400/401/402/422, …) are caller/config problems retrying can't fix. |
| 173 | func RetryableStatus(s int) bool { |
| 174 | return s == http.StatusRequestTimeout || s == http.StatusTooManyRequests || (s >= 500 && s <= 599) |
| 175 | } |
| 176 | |
| 177 | // IsConnReset distinguishes connection failures (peer reset, truncated body, |
| 178 | // closed socket) from protocol or caller errors for failure reporting. A common |
| 179 | // trigger is a proxy idle-closing SSE during a reasoner's first-token gap. |
| 180 | func IsConnReset(err error) bool { |
| 181 | if err == nil { |
| 182 | return false |
| 183 | } |
| 184 | if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) || errors.Is(err, ErrNonStreamingResponse) { |
| 185 | return false |
| 186 | } |
| 187 | if errors.Is(err, io.ErrUnexpectedEOF) || errors.Is(err, io.EOF) || |
| 188 | errors.Is(err, net.ErrClosed) || |
| 189 | errors.Is(err, syscall.ECONNRESET) || errors.Is(err, syscall.ECONNABORTED) { |
| 190 | return true |
| 191 | } |
| 192 | // net.Error alone does not prove a network failure: url.Error wraps every |
| 193 | // HTTP client error, and filesystem errors/syscall.Errno can implement it. |
| 194 | // Require a socket/DNS cause or an actual timeout to label a network failure. |
| 195 | var op *net.OpError |
| 196 | var dns *net.DNSError |
| 197 | if errors.As(err, &op) || errors.As(err, &dns) { |
| 198 | return true |
| 199 | } |
| 200 | var netErr net.Error |
| 201 | return errors.As(err, &netErr) && netErr.Timeout() |
| 202 | } |
| 203 | |
| 204 | func parseRetryAfter(resp *http.Response) time.Duration { |
| 205 | v := strings.TrimSpace(resp.Header.Get("Retry-After")) |
| 206 | if v == "" { |
| 207 | return 0 |
| 208 | } |
| 209 | if secs, err := strconv.Atoi(v); err == nil && secs >= 0 { |
| 210 | return time.Duration(secs) * time.Second |
| 211 | } |
| 212 | // RFC 9110 also allows an HTTP-date; gateways in front of rate-limited |
| 213 | // backends use it more often than the delta-seconds form. |
| 214 | if when, err := http.ParseTime(v); err == nil { |
| 215 | if d := time.Until(when); d > 0 { |
| 216 | return d |
| 217 | } |
| 218 | } |
| 219 | return 0 |
| 220 | } |
| 221 | |
| 222 | // readErrorBody drains a non-OK response body under a hard deadline and |
| 223 | // returns up to the first 4 KiB for the error message. Context cancellation |
| 224 | // already unblocks the read (the transport aborts body reads when the request |
| 225 | // context is canceled); the timer covers the case nobody cancels — a half-open |
| 226 | // upstream that sent headers and then went silent. Closing the body from the |
| 227 | // timer goroutine is the documented way to unblock an in-flight Read; it |
| 228 | // tears down the connection, which is the right call for a stalled peer. |
| 229 | func readErrorBody(resp *http.Response) []byte { |
| 230 | timer := time.AfterFunc(errorBodyReadTimeout, func() { resp.Body.Close() }) |
| 231 | msg, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) |
| 232 | // Drain the rest so a healthy connection can be reused; the timer still |
| 233 | // arms this read, so a body that stalls after the first 4 KiB cannot |
| 234 | // wedge the retry loop either. |
| 235 | _, _ = io.Copy(io.Discard, resp.Body) |
| 236 | timer.Stop() |
| 237 | resp.Body.Close() |
| 238 | return msg |
| 239 | } |
| 240 | |
| 241 | // SendWithRetry retains its historical name but sends one HTTP request, plus a |
| 242 | // single resend on a fresh connection when a pooled idle connection closed |
| 243 | // before answering (see staleIdleConnection). Every other failure returns to |
| 244 | // the caller without backoff; the user decides whether to try again. |
| 245 | func SendWithRetry(ctx context.Context, httpClient *http.Client, opts SendOptions, newReq func(context.Context) (*http.Request, error)) (*http.Response, error) { |
| 246 | identity := RequestIdentity{Provider: opts.Provider, DisplayName: opts.ProviderDisplayName, Protocol: opts.Protocol} |
| 247 | resp, stale, err := sendAttempt(ctx, httpClient, newReq) |
| 248 | if stale { |
| 249 | httpClient.CloseIdleConnections() |
| 250 | resp, _, err = sendAttempt(ctx, httpClient, newReq) |
| 251 | } |
| 252 | var build *buildRequestError |
| 253 | if errors.As(err, &build) { |
| 254 | return nil, &RequestFailure{Identity: identity, Operation: "build request", Err: build.err} |
| 255 | } |
| 256 | if err != nil { |
| 257 | return nil, &RequestFailure{Identity: identity, Operation: "request failed", Err: err} |
| 258 | } |
| 259 | if resp.StatusCode == http.StatusOK { |
| 260 | return resp, nil |
| 261 | } |
| 262 | msg := readErrorBody(resp) |
| 263 | if quota := QuotaErrorFromResponseWithIdentity(opts.Provider, opts.ProviderDisplayName, opts.Protocol, resp.StatusCode, string(msg)); quota != nil { |
| 264 | return nil, quota |
| 265 | } |
| 266 | if resp.StatusCode == http.StatusUnauthorized || resp.StatusCode == http.StatusForbidden { |
| 267 | return nil, &AuthError{Provider: opts.Provider, ProviderDisplayName: opts.ProviderDisplayName, Protocol: opts.Protocol, KeyEnv: opts.KeyEnv, KeySource: opts.KeySource, Status: resp.StatusCode, HasKey: opts.KeyPresent, Body: strings.TrimSpace(string(msg))} |
| 268 | } |
| 269 | apiErr := &APIError{ |
| 270 | RetryAfter: parseRetryAfter(resp), ShouldRetry: resp.Header.Get("x-should-retry"), |
| 271 | Provider: opts.Provider, ProviderDisplayName: opts.ProviderDisplayName, Protocol: opts.Protocol, |
| 272 | Status: resp.StatusCode, Body: strings.TrimSpace(string(msg)), |
| 273 | TraceID: responseTraceID(resp.Header), RequestPath: responseRequestPath(resp), |
| 274 | } |
| 275 | if !RetryableStatus(resp.StatusCode) { |
| 276 | if limitErr := ParseOutputLimitError(apiErr); limitErr != nil { |
| 277 | return nil, limitErr |
| 278 | } |
| 279 | if limitErr := ParseContextLimitError(apiErr); limitErr != nil { |
| 280 | return nil, limitErr |
| 281 | } |
| 282 | if replayErr := ParseReasoningReplayError(apiErr); replayErr != nil { |
| 283 | return nil, replayErr |
| 284 | } |
| 285 | } |
| 286 | return nil, apiErr |
| 287 | } |
| 288 | |
| 289 | type buildRequestError struct{ err error } |
| 290 | |
| 291 | func (e *buildRequestError) Error() string { return e.err.Error() } |
| 292 | func (e *buildRequestError) Unwrap() error { return e.err } |
| 293 | |
| 294 | // sendAttempt issues one counted, observed request. stale reports that it |
| 295 | // failed on a dead pooled idle connection. |
| 296 | func sendAttempt(ctx context.Context, httpClient *http.Client, newReq func(context.Context) (*http.Request, error)) (resp *http.Response, stale bool, err error) { |
| 297 | requestCtx, observation := observeRequest(ctx) |
| 298 | probe := &connectionProbe{} |
| 299 | req, err := newReq(probe.attach(requestCtx)) |
| 300 | if err != nil { |
| 301 | observation.finish(err, "build_error") |
| 302 | return nil, false, &buildRequestError{err: err} |
| 303 | } |
| 304 | observation.request(req) |
| 305 | recordRequestAttempt(ctx) |
| 306 | resp, err = httpClient.Do(req) |
| 307 | if err != nil { |
| 308 | observation.finish(err, "request_error") |
| 309 | return nil, probe.staleIdleConnection(ctx, err), err |
| 310 | } |
| 311 | observation.response(resp) |
| 312 | return resp, false, nil |
| 313 | } |
| 314 | |
| 315 | func responseRequestPath(resp *http.Response) string { |
| 316 | if resp == nil || resp.Request == nil || resp.Request.URL == nil { |
| 317 | return "" |
| 318 | } |
| 319 | path := resp.Request.URL.EscapedPath() |
| 320 | if len(path) > 512 { |
| 321 | return path[:512] |
| 322 | } |
| 323 | return path |
| 324 | } |
| 325 | |
| 326 | func responseTraceID(header http.Header) string { |
| 327 | for _, name := range []string{"trace_id", "trace-id", "x-trace-id"} { |
| 328 | if value := strings.TrimSpace(header.Get(name)); value != "" { |
| 329 | return value |
| 330 | } |
| 331 | } |
| 332 | return "" |
| 333 | } |
| 334 |