Documentation
¶
Index ¶
- Constants
- Variables
- func ApplyAuthHeaders(request *http.Request, options AuthHeaders)
- func Backoff(ctx context.Context, attempt int, retryAfter time.Duration) bool
- func ClassifiedError(statusCode int, message string, secrets ...string) string
- func ContentStallTimeout(idleTimeout time.Duration) time.Duration
- func CopyHeaders(headers map[string]string) map[string]string
- func HTTPClient(client *http.Client) *http.Client
- func NormalizeBaseURL(baseURL string, defaultBaseURL string, label string) (string, error)
- func PositiveOrDefault(value int, fallback int, label string) (int, error)
- func Redact(message string, secrets ...string) string
- func ResolveStreamIdleTimeout(option time.Duration) time.Duration
- func RetryAfter(response *http.Response) time.Duration
- func ScanSSEData(reader io.Reader, handle func(data string) bool) error
- func ScanSSEDataWithContext(ctx context.Context, cancel context.CancelFunc, reader io.Reader, ...) error
- func SendEvent(ctx context.Context, events chan<- zeroruntime.StreamEvent, ...)
- func SendWithAuthRetry(ctx context.Context, client *http.Client, method string, url string, ...) (*http.Response, error)
- func SendWithRetry(ctx context.Context, client *http.Client, method string, url string, ...) (*http.Response, error)
- func ShouldRetryStatus(code int) bool
- func StreamTimeoutMessage(err error, idleTimeout time.Duration) string
- func UpstreamUnreachable(message string) (string, bool)
- type AuthHeaders
- type TokenResolver
Constants ¶
const DefaultStreamIdleTimeout = 5 * time.Minute
DefaultStreamIdleTimeout is the single source of truth for how long every provider waits on a silent stream before aborting it. The watchdog only fires on genuine silence — SSE keep-alive comments reset it (see ScanSSEDataWithContext) — so this needs to be generous enough for slow cloud and reasoning backends that pause for minutes without heartbeating, while still bounding a truly hung connection. 90s was too aggressive and killed healthy long generations; 5 minutes is the floor a real stall must cross. Override globally with ZERO_STREAM_IDLE_TIMEOUT.
Variables ¶
var ErrStreamIdle = errors.New("idle timeout (upstream stopped sending data)")
ErrStreamIdle reports that a streaming upstream stopped sending data without closing the connection. Callers surface it as an idle-timeout error.
var ErrStreamStalled = errors.New("stream stalled (upstream kept the connection alive but produced no output)")
ErrStreamStalled reports that a streaming upstream kept the connection alive (SSE keep-alives reset the idle watchdog, so it never fired) but produced no actual output for ContentStallTimeout(idle). Without this an upstream that heartbeats-but-stalls — observed on chatgpt/gpt-5.x and ollama reasoning models — would hang the agent indefinitely.
Functions ¶
func ApplyAuthHeaders ¶
func ApplyAuthHeaders(request *http.Request, options AuthHeaders)
func Backoff ¶
Backoff waits before retry attempt N (1-based), returning false if the context is cancelled during the wait. A server-supplied (positive) Retry-After wins; otherwise the wait doubles from retryBackoffBase per attempt. Either way the wait is capped at maxBackoff.
func ClassifiedError ¶
ClassifiedError normalizes provider HTTP/stream errors and redacts secrets.
func ContentStallTimeout ¶
ContentStallTimeout bounds a heartbeat-but-no-output stream: keep-alives reset the idle watchdog (a heartbeating upstream is not "dead"), but if NO real data line arrives for this long the stream is aborted (ErrStreamStalled). Scaled above idleTimeout (so a slow-but-producing request that emits data between heartbeats is never killed — the watchdog only ever fires when NOTHING real arrives), but only 1.2× rather than the old 2×: a genuine heartbeat-pause stall on chatgpt/gpt-5.x rarely recovers, so a ~10-minute dead wait (at the 5m idle default) was a terrible UX for a doomed turn. At the default idle this is 6 minutes; it still scales with ZERO_STREAM_IDLE_TIMEOUT. A returned value <= 0 (idle watchdog disabled) leaves the content watchdog off too.
func HTTPClient ¶
HTTPClient returns the configured client or the shared, stall-hardened default.
func NormalizeBaseURL ¶
NormalizeBaseURL trims trailing slashes and validates an HTTP API base URL.
func PositiveOrDefault ¶
PositiveOrDefault validates optional max token settings.
func ResolveStreamIdleTimeout ¶
ResolveStreamIdleTimeout selects the effective stream idle timeout. Precedence: an explicit positive option (e.g. set by a test) wins; otherwise the ZERO_STREAM_IDLE_TIMEOUT env override if set and valid; otherwise DefaultStreamIdleTimeout. A returned value <= 0 disables the idle watchdog.
func RetryAfter ¶
RetryAfter parses a response's Retry-After header (delay-seconds or an HTTP date) into a positive duration, or 0 when absent/unparseable. The result is capped at maxBackoff by Backoff.
func ScanSSEData ¶
ScanSSEData parses Server-Sent Event data fields from a streaming response.
func ScanSSEDataWithContext ¶
func ScanSSEDataWithContext( ctx context.Context, cancel context.CancelFunc, reader io.Reader, idleTimeout time.Duration, handle func(data string) bool, ) error
ScanSSEDataWithContext parses SSE data payloads while enforcing an idle timeout and honoring ctx cancellation. The blocking scan runs on a goroutine that forwards each completed payload over a buffered channel; this consumer selects on ctx.Done, the idle timer, and incoming payloads. When the upstream goes silent for idleTimeout, cancel is invoked to abort the in-flight request (unblocking the reader) and ErrStreamIdle is returned. On ctx cancellation ctx.Err() is returned. A non-positive idleTimeout disables the watchdog.
func SendEvent ¶
func SendEvent(ctx context.Context, events chan<- zeroruntime.StreamEvent, event zeroruntime.StreamEvent)
SendEvent writes a provider event without blocking cancellation cleanup.
func SendWithAuthRetry ¶
func SendWithAuthRetry( ctx context.Context, client *http.Client, method string, url string, body []byte, base AuthHeaders, resolver TokenResolver, setExtra func(*http.Request), maxAttempts int, ) (*http.Response, error)
SendWithAuthRetry issues a request authenticated with an OAuth bearer (when the resolver yields one) or the API key, retrying ONCE with a force-refreshed token after an upstream 401. A 401 means the server rejected the request (auth failed) WITHOUT processing a completion, so replaying it after a refresh is safe — the same "no duplicate billable work" guarantee SendWithRetry relies on for 429/503. With a nil resolver (or one that yields ok=false) this behaves exactly like SendWithRetry with API-key auth.
base carries the API-key auth config (key + default header/scheme). setExtra (optional) sets the provider's non-auth headers on every attempt. Transient retries (429/503/529) are handled by the inner SendWithRetry.
func SendWithRetry ¶
func SendWithRetry( ctx context.Context, client *http.Client, method string, url string, body []byte, setHeader func(*http.Request), maxAttempts int, ) (*http.Response, error)
SendWithRetry issues an HTTP request, retrying ONLY the safe-to-replay server responses (429 and 503, see ShouldRetryStatus) up to maxAttempts — backing off between tries and honoring a server Retry-After header and context cancellation. Other 5xx and transport/network errors are returned immediately, never replayed (see the package note). The request is rebuilt from body each attempt; setHeader (if non-nil) sets headers on every attempt.
It returns the final *http.Response (which the caller inspects for a non-2xx status, exactly as before) or an error for a network failure / context cancellation. Retries exhausted on a retryable status return that response, not an error, so the caller's existing HTTP-error path still runs.
func ShouldRetryStatus ¶
ShouldRetryStatus reports whether an HTTP status is safe to retry for a non-idempotent completion POST: 429 (Too Many Requests), 503 (Service Unavailable), and 529 (Anthropic's "overloaded"). All mean the server explicitly did NOT accept the request — it was rate-limited or the service was unavailable — so replaying it cannot duplicate work. Other 5xx (500/502/504) are deliberately NOT retried: they do not guarantee the request had no effect (e.g. a 504 gateway timeout may follow an upstream that already produced a billable completion), so replaying them risks duplicate work.
func StreamTimeoutMessage ¶
StreamTimeoutMessage returns the human-readable detail for a stream-timeout error (ErrStreamIdle or ErrStreamStalled), given the configured idle timeout. Callers prepend their own "provider stream error: " prefix. A stalled stream gets a distinct, actionable message — it did NOT stop sending data (keep-alives kept arriving); it just produced no output, which usually means the model is stuck or very slow.
func UpstreamUnreachable ¶
UpstreamUnreachable detects a provider error that is really a connectivity failure to an upstream host rather than a model/request error, and rewrites it into a clear, actionable message. The common case is a local Ollama daemon serving a "-cloud" model: it answers on localhost but returns HTTP 502 because it cannot reach its own cloud backend, surfacing an opaque proxied string like `Post "https://ollama.com:443/...": net/http: TLS handshake timeout`. It matches only when both a transport failure marker and a concrete host are present, so the agent's own request-deadline cancellations are left untouched. Non-matching messages are returned unchanged with false.
Types ¶
type AuthHeaders ¶
type TokenResolver ¶
type TokenResolver func(ctx context.Context, forceRefresh bool) (header string, value string, ok bool, err error)
TokenResolver yields a fresh OAuth credential for one request, or ok=false to fall back to the API key (e.g. there is no OAuth login for this provider). header is the header name to set ("" => Authorization); value is the full header value (for example "Bearer abc"). forceRefresh asks the resolver to bypass any cached token; it is set only on the single retry after a 401.
Source Files