diff --git a/.gitignore b/.gitignore index 3b80565..1791c84 100644 --- a/.gitignore +++ b/.gitignore @@ -3,3 +3,5 @@ bin/ *.out coverage.* .DS_Store +.odek-artifacts/ +.tmp-spincheck/ diff --git a/PLAN.md b/PLAN.md index ad260a7..24a4da7 100644 --- a/PLAN.md +++ b/PLAN.md @@ -15,7 +15,7 @@ inference endpoints, with: 1. **Multiple authenticated endpoints simultaneously**, auto-discovered from the environment via `_API_KEY` (e.g. `OPENAI_API_KEY`, `GEMINI_API_KEY`, `DEEPSEEK_API_KEY`, `ZAI_API_KEY`, - `KIM_API_KEY`, `ANTHROPIC_API_KEY`). + `KIMI_API_KEY`, `ANTHROPIC_API_KEY`). 2. **Dynamic model discovery on the fly** — `ListModels(ctx)` hits each provider's models endpoint and returns what the account can actually access. **No static model profile tables** (replaces odek's `KnownProfiles` / `ModelProfile`). @@ -101,7 +101,7 @@ instance, exactly as odek does today. OpenAI-compatible shape is the canonical type system (odek's loop already speaks it): `Message{Role, Content, ReasoningContent, ToolCalls, ToolCallID}`, `ToolDef`, -`ChatRequest{Model, Messages, Tools, Thinking, ThinkingBudget, MaxTokens, Temperature, SystemBlocks}`, +`ChatRequest{Model, Messages, Tools, Thinking, ThinkingBudget, MaxTokens, Temperature, System}`, `ChatResult{Content, ReasoningContent, ToolCalls, FinishReason, Usage}`. Adapters translate to/from Anthropic (`system` top-level, content blocks, `tool_use`/`tool_result`, `thinking` blocks) and Gemini (`contents`/`parts`, `functionDeclarations`, `functionCall`/`functionResponse`, @@ -121,7 +121,7 @@ Thinking control maps per format: `reasoning_effort` (openai) / `thinking:{type, - **Gemini**: system prompts go to `systemInstruction`; `functionResponse` parts ride `role:"user"` per current API docs; model in URL path (`/v1beta/models/{m}:generateContent`); no tool-call IDs — SDK synthesizes `call_` ids; `finishReason` mapping (STOP→stop, MAX_TOKENS→length, - SAFETY/RECISATION→content_filter); reasoning = parts with `thought:true` + `includeThoughts`. + SAFETY/RECITATION→content_filter); reasoning = parts with `thought:true` + `includeThoughts`. - **Canonical**: multi-part content (images/audio) is **out of scope for v0** and documented as a limitation — `Content` is plain text; if a provider returns multi-part, text parts concatenate. @@ -257,3 +257,13 @@ TDD: every milestone lands RED tests first, then implementation (house conventio 6. No static fallback metadata: unknown context window stays unknown (0) + optional consumer override. 7. v0.x until odek migration lands; then v1.0. 8. Stdlib only — SSE parsing, JSON, HTTP pooling hand-rolled, exactly like odek today. + +## 11. Implementation status (2026-09-04, pre-v0.1.0) + +Shipped on `feat/multi-provider-sdk` ahead of v0.1.0: + +- **Hardening passes** — three sequential adversarial reviews (API contract vs docs, concurrency/streaming bug hunt, wire-fidelity + registry) with every finding personally verified, reproduced RED-first, and fixed. Highlights: SSE parser goroutine-leak fix (abort-safe `done` protocol), real 1 MiB SSE line cap (was effectively 64 KiB), no-retry-after-partial-output branch in `callStream`, premature-close detection (no silent empty successes), buffered 429 preserving `*RateLimitError` under deadline pressure, narrowed `streamRejected` classifier, Anthropic `after_id` pagination, `max_completion_tokens` routing for o-series/gpt-5, Gemini usage-only-chunk guard and functionResponse name resolution, nested OpenAI error-envelope parsing, `RateLimitError.Unwrap`, learn-once state hoisted to `Provider` (shared across all ChatClients), wiring-time validation for overridden built-ins, in-band role validation at the SDK boundary. +- **Extended-thinking round-trip** — `ThinkingSignature` added to `Message`/`ChatResult`; Anthropic thinking blocks are captured (buffered + `signature_delta`) and replayed first-block-with-signature, unblocking extended-thinking tool loops. +- **Canonical vocabulary** — unmapped provider finish reasons map to `""` on all three formats. +- **Coverage** — 81% → 97.7% of statements, including the previously zero-coverage `CallStream` failure-orchestration paths; race detector and golangci-lint clean. +- **Deferred (known, intentional)** — `ReasoningContent` is serialized request-side only for Anthropic (signature-gated); DeepSeek/GLM/Gemini reasoning stays advisory. The 2.3% coverage residual is unreachable defensive code (documented in the review record). diff --git a/README.md b/README.md index 74f3a14..1a103ae 100644 --- a/README.md +++ b/README.md @@ -1,11 +1,25 @@ # go-llm-sdk +[![CI](https://github.com/BackendStack21/go-llm-sdk/actions/workflows/ci.yml/badge.svg)](https://github.com/BackendStack21/go-llm-sdk/actions/workflows/ci.yml) +[![Go Reference](https://pkg.go.dev/badge/github.com/BackendStack21/go-llm-sdk.svg)](https://pkg.go.dev/github.com/BackendStack21/go-llm-sdk) +[![License: MIT](https://img.shields.io/badge/License-MIT-blue.svg)](LICENSE) +![Go](https://img.shields.io/badge/Go-1.25%2B-00ADD8) + Multi-provider Go SDK for LLM inference endpoints — **OpenAI, Google Gemini, DeepSeek, Z.ai, Kimi (Moonshot) and Anthropic**, plus any custom OpenAI-compatible gateway. Stdlib only, zero external dependencies. - **Multiple authenticated endpoints at once** — auto-discovered from `_API_KEY` environment variables (aliases supported). - **Dynamic model discovery** — `ListModels` returns what the account can actually access, on the fly. No static model tables, ever. - **One canonical API** — OpenAI-shaped requests/responses; Anthropic and Gemini wire formats are translated for you. -- **Production streaming semantics** (ported from [odek](https://github.com/BackendStack21/odek)'s battle-tested client): SSE with idle watchdog + hard wall-clock deadline, abort-with-partial-result, retries that never duplicate partial output, and learn-once fallbacks for providers that reject `stream_options`, streaming, or `reasoning_effort`+tools. +- **Production streaming semantics** (ported from [odek](https://github.com/BackendStack21/odek)'s battle-tested client): SSE with idle watchdog + hard wall-clock deadline, abort-with-partial-result, retries that never duplicate partial output, premature-close detection, and learn-once fallbacks for providers that reject `stream_options`, streaming, or `reasoning_effort`+tools. +- **Hardened by adversarial review** — three sequential adversarial review passes (contract, concurrency, wire fidelity) with every finding reproduced and fixed RED-first. Goroutine-leak-free streaming, race-clean shared state, canonical-only error vocabulary. + +## Install + +```bash +go get github.com/BackendStack21/go-llm-sdk@v0.1.0 +``` + +Requires Go 1.25+. No dependencies beyond the standard library. ## Quickstart @@ -25,7 +39,11 @@ res, err := chat.Call(ctx, &llm.ChatRequest{ Messages: []llm.Message{{Role: llm.RoleUser, Content: "Hello"}}, Tools: []llm.ToolDef{{Name: "get_weather", Parameters: schema}}, }) +``` +Streaming: + +```go res, err = chat.CallStream(ctx, req, func(d llm.Delta) error { switch d.Kind { case llm.DeltaReasoning: // thinking fragment @@ -47,7 +65,7 @@ res, err = chat.CallStream(ctx, req, func(d llm.Delta) error { | `kimi` | openai | `https://api.moonshot.ai/v1` | `KIMI_API_KEY` (`MOONSHOT_API_KEY`) | `KIMI_BASE_URL` | | `anthropic` | anthropic | `https://api.anthropic.com` | `ANTHROPIC_API_KEY` | `ANTHROPIC_BASE_URL` | -Primary env var beats its alias. Explicit keys (`WithAPIKey`) beat env. Base-URL overrides accept any gateway speaking the provider's format. +Primary env var beats its alias. Explicit keys (`WithAPIKey`) beat env. Base-URL overrides accept any gateway speaking the provider's format. Override validation runs at wiring time — a bad base URL fails loudly at `New`, not per request. Custom gateways: @@ -59,26 +77,142 @@ sdk := llm.New(llm.WithProvider("my-gateway", )) ``` -## Provider quirks, handled +## Canonical API + +Requests and results are provider-neutral. Unknown message roles are rejected at the SDK boundary (never silently dropped or reinterpreted). + +```go +type ChatRequest struct { + Model string // optional; ChatClient's model wins when both set + Messages []Message // RoleUser | RoleAssistant | RoleSystem | RoleTool + System []SystemBlock // {Text, Cache} — Cache marks Anthropic prompt-cache blocks + Tools []ToolDef // {Name, Description, Parameters json.RawMessage} + Thinking string // "", "enabled", "disabled", "low", "medium", "high" + ThinkingBudget int // explicit token budget where the provider supports it + MaxTokens int // routed to max_completion_tokens on o-series/gpt-5 + Temperature float64 // 0 = provider default; negative = explicit 0 +} + +type ChatResult struct { + Content string + ReasoningContent string // provider thinking text (advisory) + ThinkingSignature string // Anthropic: replay via Message.ThinkingSignature + ToolCalls []ToolCall // {ID, Name, Arguments} + FinishReason string // stop | length | tool_calls | content_filter | "" + Usage Usage // {PromptTokens, CompletionTokens, ReasoningTokens} +} +``` + +Finish reasons are canonical: anything a provider reports outside the vocabulary maps to `""` (unknown) rather than leaking provider-specific strings. + +## Streaming semantics + +`CallStream` enforces four guarantees, each covered by regression tests: + +1. **Idle watchdog** — a stream silent longer than `streamIdleTimeout` (120s default) fails with `ErrIdleTimeout`. Keepalive comments reset it. +2. **Hard wall-clock deadline** — the whole stream is bounded by the per-request timeout (`WithRequestTimeout`, default 120s; per-client via `SetRequestTimeout`). +3. **Retries never duplicate output** — retries happen only before the first emitted delta. A failure after partial output returns the partial `*ChatResult` plus a wrapped error and is never retried. +4. **No silent empty successes** — a provider that closes the stream before its completion signal (before `[DONE]` / `message_stop`) yields a retryable error, not an empty result. Gemini, whose streams legitimately end at EOF, is exempt. + +Aborting from the delta handler returns the partial result alongside `*StreamAbortedError` — the parser goroutine is always released, so aborted streams leak nothing. + +### Tool-call loop + +```go +for { + res, err := chat.CallStream(ctx, req, func(d llm.Delta) error { return nil }) + if err != nil { + return err + } + if len(res.ToolCalls) == 0 { + return nil + } + req.Messages = append(req.Messages, + Message{Role: RoleAssistant, Content: res.Content, + ReasoningContent: res.ReasoningContent, ThinkingSignature: res.ThinkingSignature, + ToolCalls: res.ToolCalls}) + for _, tc := range res.ToolCalls { + req.Messages = append(req.Messages, + Message{Role: RoleTool, ToolCallID: tc.ID, ToolName: tc.Name, Content: execute(tc)}) + } +} +``` + +On Gemini, a tool result's `ToolName` may be omitted — the SDK recovers the function name from the assistant `ToolCall` it answers, and errors loudly if it cannot. + +## Extended thinking -The registry carries explicit quirk flags (no URL sniffing): Anthropic/DeepSeek/GLM thinking objects, OpenAI/GLM `reasoning_effort`, GLM-5.3 forced-thinking models (`disabled` → `enabled` + `reasoning_effort: low`), the `anthropic-version` header, Gemini `systemInstruction`/`thinkingConfig`, and temperature-forbidding models (`o1/o3/o4/gpt-5*/kimi-for-coding*/k3*` never receive an explicit temperature). Learn-once fallbacks fix provider rejections at runtime without repeated failed round-trips. +- **Anthropic** — `thinking` blocks are parsed in both buffered and streaming modes. `ChatResult.ThinkingSignature` carries the provider signature; for tool loops, replay it on the assistant message (`Message.ReasoningContent` + `Message.ThinkingSignature`) — the SDK re-serializes it as the first block, as Anthropic's API requires. Omitting it makes extended-thinking tool loops fail mid-conversation. +- **DeepSeek / GLM** — reasoning streams as `DeltaReasoning` fragments and lands in `ReasoningContent` (advisory; not replayed). +- **Gemini** — `thought: true` parts map to reasoning deltas; `thinkingConfig` is derived from `Thinking`/`ThinkingBudget`. + +## Learn-once fallbacks + +When a provider rejects a request pattern, the SDK learns the constraint **once per provider** (shared across every `ChatClient` you mint) and never re-pays the failed round-trip: + +| Trigger (provider 400) | Learned fallback | +|---|---| +| Rejects `stream_options` | omit `stream_options` from streaming requests | +| Rejects `reasoning_effort` + tools | pin `reasoning_effort: "none"` | +| Rejects streaming itself | downgrade to buffered calls permanently | +| Answers a streamed request with a non-SSE body | downgrade to buffered calls permanently | ## Retry policy -8 attempts, exponential backoff capped at 30s with ±20% jitter, `Retry-After` (seconds or HTTP-date) honored, context cancellation between attempts. Persistent 429s surface as `*llm.RateLimitError{Attempts, RetryAfter}`. Streaming retries happen only before the first emitted delta. +8 attempts, exponential backoff capped at 30s with ±20% jitter, `Retry-After` (seconds or HTTP-date) honored, context cancellation honored between and during attempts. Persistent 429s surface as `*llm.RateLimitError{Attempts, RetryAfter}` — including when the retry sleep is cut short by a deadline, so the caller never loses the retry signal. `RateLimitError` unwraps to `*APIError` for `errors.As` access to `Status`/`Retryable`. + +## Timeouts & cancellation + +- Buffered calls: per-request timeout on the HTTP client (`WithRequestTimeout`, per-client `SetRequestTimeout` — race-safe, swap is atomic). +- Streaming: the same timeout becomes the hard wall-clock deadline via context; per-attempt SSE reads are additionally bounded by the idle watchdog. +- Every wait (backoff, Retry-After, stream reads) selects on the caller's context — cancellation propagates everywhere, and a cancelled call never misreports as "retry exhausted". +- Response bodies are capped (50 MB chat, 8 MB listings, 1 MiB SSE lines, 4 MiB SSE events) as an OOM bound. + +## Error handling -## Errors +`*ConfigError` (unknown/unauthenticated provider, invalid wiring, unknown role, no model), `*APIError{Provider, Status, Code, Message, Retryable}`, `*RateLimitError{Attempts, RetryAfter}` (unwraps to `*APIError`), `*StreamAbortedError` (returned together with the partial `*ChatResult`). A stream failure after partial output returns the partial `*ChatResult` plus a wrapped error and is never retried; the idle watchdog surfaces as `ErrIdleTimeout` (retried only before the first delta); wall-clock deadlines surface as context deadline errors. Recommended classification: + +```go +var abort *llm.StreamAbortedError +var rl *llm.RateLimitError +var ae *llm.APIError +switch { +case errors.As(err, &abort): // consumer abort (partial result returned) +case errors.As(err, &rl): // back off rl.RetryAfter +case errors.As(err, &ae): // provider said no (ae.Status) +case errors.Is(err, llm.ErrIdleTimeout): // stream went silent +case errors.Is(err, context.DeadlineExceeded): // wall-clock budget spent +} +``` + +API keys never appear in any error text. Provider error bodies are parsed per format (nested OpenAI envelope, Anthropic `error.type/message`, Gemini `error.status/message`) with a 512-byte raw-body fallback. + +## Thread safety + +`SDK` and `Provider` are safe for concurrent use. `ChatClient` is safe for concurrent `Call`/`CallStream`; `SetRequestTimeout` is race-safe (atomic swap) but should still be called before the first request so in-flight calls use one timeout. Learn-once state is shared per provider via atomics — monotonic, converging, race-free. + +## Model discovery + +`ListModels` hits each provider's models endpoint (Anthropic paginates with `after_id`, Gemini with `pageToken`), caches per SDK for 5 minutes (`WithModelCacheTTL(0)` disables, `ForceRefresh()` bypasses), and retries transient failures 3×. Fields the provider does not report stay zero — the SDK never guesses. + +## Testing + +```bash +make quality # fmt + vet + tests +make test-race # race detector +make lint # golangci-lint (v2 config) +``` -`*ConfigError` (unknown/unauthenticated provider), `*APIError{Provider, Status, Code, Message, Retryable}`, `*RateLimitError`, `*StreamAbortedError` (returned together with the partial `*ChatResult`). API keys never appear in any error text. +Coverage sits at **97.7%** of statements, including the streaming failure-orchestration paths (deadline, 429, premature close, partial-output) that are usually the blind spot of SDK test suites. The residual ~2% is provably unreachable defensive code (documented in the review record). ## Design record -See [PLAN.md](PLAN.md) for the architecture and the odek migration path. +See [PLAN.md](PLAN.md) for the architecture, the provider-quirk table, and the odek migration path. ## Status -v0 — API may shift until the odek integration lands, then v1.0. +v0.1.0 — API may shift until the odek integration lands, then v1.0. ## License -MIT — see [LICENSE](LICENSE). +[MIT](LICENSE) diff --git a/anthropic.go b/anthropic.go index d34c4fc..0f131d1 100644 --- a/anthropic.go +++ b/anthropic.go @@ -28,6 +28,10 @@ type anSysBlock struct { type anBlock struct { Type string `json:"type"` // "text" | "tool_use" | "tool_result" Text string `json:"text,omitempty"` + // thinking (replayed assistant turns; must be the FIRST block and + // carry the provider signature) + Thinking string `json:"thinking,omitempty"` + Signature string `json:"signature,omitempty"` // tool_use ID string `json:"id,omitempty"` Name string `json:"name,omitempty"` @@ -153,6 +157,12 @@ func buildAnthropicRequest(req *ChatRequest, model string, stream bool) ([]byte, }) case RoleAssistant: var blocks []anBlock + if m.ReasoningContent != "" && m.ThinkingSignature != "" { + // Anthropic requires a replayed thinking block to be the + // FIRST block and to carry its signature; extended-thinking + // tool loops are otherwise rejected mid-conversation. + blocks = append(blocks, anBlock{Type: "thinking", Thinking: m.ReasoningContent, Signature: m.ThinkingSignature}) + } if m.Content != "" { blocks = append(blocks, anBlock{Type: "text", Text: m.Content}) } @@ -194,12 +204,13 @@ func buildAnthropicRequest(req *ChatRequest, model string, stream bool) ([]byte, // ── response ───────────────────────────────────────────────────────────── type anRespBlock struct { - Type string `json:"type"` // "text" | "thinking" | "tool_use" - Text string `json:"text"` - Thinking string `json:"thinking"` - ID string `json:"id"` - Name string `json:"name"` - Input json.RawMessage `json:"input"` + Type string `json:"type"` // "text" | "thinking" | "tool_use" + Text string `json:"text"` + Thinking string `json:"thinking"` + Signature string `json:"signature"` + ID string `json:"id"` + Name string `json:"name"` + Input json.RawMessage `json:"input"` } type anResponse struct { @@ -226,7 +237,9 @@ func mapAnthropicStopReason(s string) string { case "refusal": return FinishContentFilter default: - return s + // Provider-specific reasons (pause_turn, model_context_window_exceeded, + // ...) stay out of the canonical vocabulary. + return "" } } @@ -257,6 +270,11 @@ func parseAnthropicResponse(body []byte) (*ChatResult, error) { } res.Content = strings.Join(content, "") res.ReasoningContent = strings.Join(thinking, "") + for _, b := range r.Content { + if b.Type == "thinking" && b.Signature != "" { + res.ThinkingSignature = b.Signature + } + } res.Usage = Usage{ PromptTokens: r.Usage.InputTokens, CompletionTokens: r.Usage.OutputTokens, @@ -281,10 +299,11 @@ type anStreamEvent struct { // delta and usage as top-level siblings, not nested under a // "message_delta" key. Delta struct { - Type string `json:"type"` // "text_delta" | "thinking_delta" | "input_json_delta" + Type string `json:"type"` // "text_delta" | "thinking_delta" | "signature_delta" | "input_json_delta" StopReason string `json:"stop_reason"` Text string `json:"text"` Thinking string `json:"thinking"` + Signature string `json:"signature"` PartialJSON string `json:"partial_json"` } `json:"delta"` // message_delta usage (top-level sibling of delta) @@ -327,6 +346,8 @@ func mapAnthropicStreamEvent(data []byte, acc *streamAccum) ([]Delta, bool, erro case "thinking_delta": acc.reasoning.WriteString(ev.Delta.Thinking) deltas = append(deltas, Delta{Kind: DeltaReasoning, Text: ev.Delta.Thinking}) + case "signature_delta": + acc.thinkingSignature += ev.Delta.Signature case "input_json_delta": c := acc.call(ev.Index) c.args.WriteString(ev.Delta.PartialJSON) @@ -378,7 +399,7 @@ func listModelsAnthropic(ctx context.Context, pc *providerClient) ([]Model, erro for page := 0; page < 10; page++ { url := pc.base + "/v1/models?limit=100" if pageID != "" { - url += "&page_id=" + pageID + url += "&after_id=" + pageID } data, _, err := pc.get(ctx, url) if err != nil { diff --git a/anthropic_test.go b/anthropic_test.go index a29bc3a..b0da9a0 100644 --- a/anthropic_test.go +++ b/anthropic_test.go @@ -224,7 +224,7 @@ func TestListModelsAnthropic_Pagination(t *testing.T) { w.WriteHeader(400) return } - if r.URL.Query().Get("page_id") == "" { + if r.URL.Query().Get("after_id") == "" { fmt.Fprint(w, `{"data":[{"id":"claude-a","display_name":"Claude A","created_at":"2026-01-02T15:04:05Z"}],"has_more":true,"last_id":"claude-a"}`) return } @@ -244,7 +244,7 @@ func TestListModelsAnthropic_Pagination(t *testing.T) { if models[0].CreatedAt.IsZero() { t.Errorf("CreatedAt = %v, want parsed RFC3339", models[0].CreatedAt) } - if len(paths) != 2 || paths[1] != "/v1/models?limit=100&page_id=claude-a" { - t.Errorf("requests = %v, want page_id follow-up", paths) + if len(paths) != 2 || paths[1] != "/v1/models?limit=100&after_id=claude-a" { + t.Errorf("requests = %v, want after_id follow-up", paths) } } diff --git a/chat.go b/chat.go index d47e737..2579868 100644 --- a/chat.go +++ b/chat.go @@ -28,26 +28,47 @@ import ( // - Learn-once fallbacks: drop stream_options (field level), fall back to // the buffered path (provider rejects streaming), pin reasoning_effort // "none" (provider rejects effort combined with tools). -type providerClient struct { - cfg ProviderConfig - http *http.Client // buffered, whole-request timeout - streamHTTP *http.Client // no deadline; SSE body reads - base string // trimmed base URL - - // Learn-once, atomic: set after the provider taught us a constraint; - // subsequent calls avoid the failed round-trip. +// +// learnOnce holds the learn-once fallback flags. They live on the Provider +// (shared across every ChatClient minted from it) so a constraint the +// provider teaches one client is honored by all of them. +type learnOnce struct { dropStreamOptions atomic.Bool forceBuffered atomic.Bool forceNoneEffort atomic.Bool } +type providerClient struct { + cfg ProviderConfig + bufPtr atomic.Pointer[http.Client] // buffered client; SetRequestTimeout swaps it atomically + streamHTTP *http.Client // no deadline; SSE body reads + base string // trimmed base URL + learn *learnOnce // shared learn-once fallback state +} + +// buffered returns the current buffered-path HTTP client. +func (pc *providerClient) buffered() *http.Client { return pc.bufPtr.Load() } + func newProviderClient(cfg ProviderConfig, buffered, stream *http.Client) *providerClient { - return &providerClient{ + return newProviderClientWithLearn(cfg, buffered, stream, &learnOnce{}) +} + +// newProviderClientWithLearn builds a client that shares learn-once state. +func newProviderClientWithLearn(cfg ProviderConfig, buffered, stream *http.Client, learn *learnOnce) *providerClient { + if learn == nil { + learn = &learnOnce{} + } + if buffered == nil { + buffered = newBufferedHTTP(nil, 0) + } + pc := &providerClient{ cfg: cfg, - http: buffered, streamHTTP: stream, base: strings.TrimRight(cfg.BaseURL, "/"), + learn: learn, } + pc.bufPtr.Store(buffered) + return pc } // Response body read caps (DoS/OOM bound). @@ -65,6 +86,13 @@ var streamIdleTimeout = 120 * time.Second // errStreamStop is the internal sentinel for a clean stream end. var errStreamStop = errors.New("llm: stream complete") +// errPrematureClose marks a 200+SSE stream the provider closed before its +// completion signal; retryable only before the first delta. +var errPrematureClose = errors.New("llm: provider closed the stream before completion") + +// errNonSSE marks a streamed request answered with a regular body. +var errNonSSE = errors.New("llm: provider answered a streamed request with a non-event-stream body") + // consumerAbort wraps a delta-handler error through pumpSSE. type consumerAbort struct{ err error } @@ -74,7 +102,7 @@ func (c *consumerAbort) Unwrap() error { return c.err } // requestTimeout is the per-request wall-clock budget; streaming uses it // as the hard overall deadline. func (pc *providerClient) requestTimeout() time.Duration { - if t := pc.http.Timeout; t > 0 { + if t := pc.buffered().Timeout; t > 0 { return t } return DefaultTimeout @@ -85,6 +113,15 @@ func (pc *providerClient) requestTimeout() time.Duration { // buildChatRequest dispatches format-specific serialization. stream=false // yields the buffered request; stream=true the SSE request. func (pc *providerClient) buildChatRequest(req *ChatRequest, model string, stream bool) ([]byte, string, error) { + // Reject unknown roles loudly: OpenAI would silently send them as user + // messages and Anthropic/Gemini would silently drop them. + for i, m := range req.Messages { + switch m.Role { + case RoleUser, RoleAssistant, RoleSystem, RoleTool: + default: + return nil, "", &ConfigError{Msg: fmt.Sprintf("message %d: unknown role %q", i, string(m.Role))} + } + } if model == "" { model = req.Model } @@ -102,8 +139,8 @@ func (pc *providerClient) buildChatRequest(req *ChatRequest, model string, strea } return body, fmt.Sprintf("%s/v1beta/models/%s:generateContent", pc.base, model), err default: // FormatOpenAI - oa := buildOpenAIRequest(pc.cfg, req, model, stream, !pc.dropStreamOptions.Load()) - if pc.forceNoneEffort.Load() && len(req.Tools) > 0 { + oa := buildOpenAIRequest(pc.cfg, req, model, stream, !pc.learn.dropStreamOptions.Load()) + if pc.learn.forceNoneEffort.Load() && len(req.Tools) > 0 { oa = reasoningEffortNonePatched(oa) } body, err := json.Marshal(oa) @@ -160,12 +197,27 @@ func (pc *providerClient) httpError(status int, body []byte) *APIError { msg, code = eb.Error.Message, eb.Error.Status } default: - var eb oaErrorBody - if json.Unmarshal(body, &eb) == nil && eb.Message != "" { - msg = eb.Message - if s, ok := eb.Code.(string); ok { + // OpenAI's canonical envelope nests the error object; some + // gateways send the fields top-level. Try nested first. + var nested struct { + Error struct { + Message string `json:"message"` + Code any `json:"code"` + } `json:"error"` + } + if json.Unmarshal(body, &nested) == nil && nested.Error.Message != "" { + msg = nested.Error.Message + if s, ok := nested.Error.Code.(string); ok { code = s } + } else { + var eb oaErrorBody + if json.Unmarshal(body, &eb) == nil && eb.Message != "" { + msg = eb.Message + if s, ok := eb.Code.(string); ok { + code = s + } + } } } if msg == "" && len(body) > 0 { @@ -194,7 +246,7 @@ func (pc *providerClient) post(ctx context.Context, client *http.Client, url str if err != nil { return nil, 0, err } - defer resp.Body.Close() + defer func() { _ = resp.Body.Close() }() data, err = io.ReadAll(io.LimitReader(resp.Body, maxResponseSize+1)) if err != nil { return nil, 0, err @@ -227,12 +279,24 @@ func streamOptionsRejected(e *APIError) bool { strings.Contains(e.Message, "stream_options") } -// streamRejected classifies a 400 rejecting streaming itself. +// streamRejected classifies a 400 rejecting streaming itself. Deliberately +// narrow: the message must pair "stream" with an explicit rejection +// phrase, so unrelated 400s that merely mention streaming (context-length +// errors, parameter limits) never trigger the permanent buffered downgrade. func streamRejected(e *APIError) bool { if e == nil || e.Status != http.StatusBadRequest { return false } - return strings.Contains(e.Message, "stream") + m := strings.ToLower(e.Message) + if !strings.Contains(m, "stream") || strings.Contains(m, "stream_options") { + return false + } + for _, phrase := range []string{"not support", "unsupported", "does not support", "not allowed", "disabled", "reject"} { + if strings.Contains(m, phrase) { + return true + } + } + return false } // retryDelay picks Retry-After when present, else exponential backoff. @@ -248,10 +312,9 @@ func retryDelay(ra time.Duration, attempt int) time.Duration { // call runs a buffered chat completion with retries. func (pc *providerClient) call(ctx context.Context, req *ChatRequest, model string) (*ChatResult, error) { var ( - lastErr error - rateErr *APIError // last 429 - rateRA time.Duration - attempts429 int + lastErr error + rateErr *APIError // last 429 + rateRA time.Duration ) for attempt := 0; attempt <= maxRetries; attempt++ { if err := ctx.Err(); err != nil { @@ -261,17 +324,19 @@ func (pc *providerClient) call(ctx context.Context, req *ChatRequest, model stri if err != nil { return nil, err } - data, ra, err := pc.post(ctx, pc.http, url, body) + data, ra, err := pc.post(ctx, pc.buffered(), url, body) if err != nil { var apiErr *APIError if errors.As(err, &apiErr) { switch { case apiErr.Status == http.StatusTooManyRequests: - rateErr, rateRA, attempts429 = apiErr, ra, attempts429+1 - lastErr = apiErr + rateErr, rateRA, lastErr = apiErr, ra, apiErr if attempt < maxRetries { if !retrySleep(ctx, retryDelay(ra, attempt)) { - return nil, ctx.Err() + // The sleep died with the context (deadline or + // cancel); still surface the 429 — the caller + // needs Status/RetryAfter to plan the retry. + return nil, &RateLimitError{APIError: *rateErr, Attempts: attempt + 1, RetryAfter: rateRA} } continue } @@ -282,11 +347,15 @@ func (pc *providerClient) call(ctx context.Context, req *ChatRequest, model stri } continue case apiErr.Status == http.StatusBadRequest && len(req.Tools) > 0 && - !pc.forceNoneEffort.Load() && reasoningEffortRejected(apiErr): + !pc.learn.forceNoneEffort.Load() && reasoningEffortRejected(apiErr): // Learn the constraint once; retry immediately with // effort pinned to "none". - pc.forceNoneEffort.Store(true) - continue + pc.learn.forceNoneEffort.Store(true) + lastErr = apiErr + if attempt < maxRetries { + continue + } + return nil, apiErr } if rateErr != nil && !apiErr.Retryable && apiErr.Status != http.StatusTooManyRequests { // A definitive failure after earlier 429s. @@ -348,7 +417,7 @@ func (pc *providerClient) mapper() streamEventMapper { // fragments; returning an error from it aborts with the partial result and // *StreamAbortedError. func (pc *providerClient) callStream(ctx context.Context, req *ChatRequest, model string, onDelta func(Delta) error) (*ChatResult, error) { - if pc.forceBuffered.Load() { + if pc.learn.forceBuffered.Load() { return pc.call(ctx, req, model) } deadlineCtx, cancel := context.WithTimeout(ctx, pc.requestTimeout()) @@ -364,7 +433,7 @@ func (pc *providerClient) callStream(ctx context.Context, req *ChatRequest, mode if err := deadlineCtx.Err(); err != nil { break } - if pc.forceBuffered.Load() { + if pc.learn.forceBuffered.Load() { cancel() return pc.call(ctx, req, model) } @@ -373,14 +442,28 @@ func (pc *providerClient) callStream(ctx context.Context, req *ChatRequest, mode return nil, err } - out := pc.attemptStream(deadlineCtx, url, body, mapper, onDelta) + out := pc.attemptStream(deadlineCtx, url, body, mapper, onDelta, len(req.Tools) > 0) switch { case out.success(): return out.result, nil case out.abort != nil: return out.result, out.abort + case out.err != nil && out.result != nil: + // Partial output was delivered: never retry, surface with the + // partial result (a retry would duplicate user-visible output). + return out.result, out.err case out.learnRetry: // learn-once applied; retry without backoff - continue + if out.apiErr != nil { + lastErr = out.apiErr + } else if out.err != nil { + lastErr = out.err + } + if attempt < maxRetries { + continue + } + // Learn trigger fired on the final attempt: surface the cause + // instead of falling off the loop with a nil error. + return nil, lastErr case out.apiErr != nil && out.apiErr.Retryable: if out.apiErr.Status == http.StatusTooManyRequests { rateErr, rateRA = out.apiErr, out.retryAfter @@ -400,6 +483,9 @@ func (pc *providerClient) callStream(ctx context.Context, req *ChatRequest, mode if retrySleep(deadlineCtx, retryDelay(0, attempt)) { continue } + if err := deadlineCtx.Err(); err != nil { + return nil, err // interrupted by deadline/cancel, not exhaustion + } return nil, fmt.Errorf("llm: retry exhausted (%d attempts): %w", attempt+1, lastErr) default: return nil, out.err @@ -429,8 +515,9 @@ func (o *streamOutcome) success() bool { return o.err == nil && o.abort == nil && o.apiErr == nil && !o.learnRetry } -// attemptStream performs one streaming attempt. -func (pc *providerClient) attemptStream(ctx context.Context, url string, body []byte, mapper streamEventMapper, onDelta func(Delta) error) streamOutcome { +// attemptStream performs one streaming attempt. learnEffort enables the +// reasoning_effort learn-once trigger (set when the request carries tools). +func (pc *providerClient) attemptStream(ctx context.Context, url string, body []byte, mapper streamEventMapper, onDelta func(Delta) error, learnEffort bool) streamOutcome { req, rerr := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(body)) if rerr != nil { return streamOutcome{err: &ConfigError{Msg: "build request: " + rerr.Error()}} @@ -443,7 +530,7 @@ func (pc *providerClient) attemptStream(ctx context.Context, url string, body [] if derr != nil { return streamOutcome{err: derr} // transport error: retryable } - defer resp.Body.Close() + defer func() { _ = resp.Body.Close() }() if resp.StatusCode < 200 || resp.StatusCode >= 300 { data, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) @@ -451,11 +538,14 @@ func (pc *providerClient) attemptStream(ctx context.Context, url string, body [] e := pc.httpError(resp.StatusCode, data) switch { case streamOptionsRejected(e): - pc.dropStreamOptions.Store(true) - return streamOutcome{learnRetry: true} + pc.learn.dropStreamOptions.Store(true) + return streamOutcome{learnRetry: true, apiErr: e} + case learnEffort && !pc.learn.forceNoneEffort.Load() && reasoningEffortRejected(e): + pc.learn.forceNoneEffort.Store(true) + return streamOutcome{learnRetry: true, apiErr: e} case streamRejected(e): - pc.forceBuffered.Store(true) - return streamOutcome{learnRetry: true} + pc.learn.forceBuffered.Store(true) + return streamOutcome{learnRetry: true, apiErr: e} default: return streamOutcome{apiErr: e, retryAfter: ra} } @@ -463,8 +553,8 @@ func (pc *providerClient) attemptStream(ctx context.Context, url string, body [] if ct := resp.Header.Get("Content-Type"); !strings.Contains(ct, "text/event-stream") { // Provider answered a streamed request with a regular body. - pc.forceBuffered.Store(true) - return streamOutcome{learnRetry: true} + pc.learn.forceBuffered.Store(true) + return streamOutcome{learnRetry: true, err: errNonSSE} } acc := newStreamAccum() @@ -490,6 +580,18 @@ func (pc *providerClient) attemptStream(ctx context.Context, url string, body [] switch { case perr == nil, errors.Is(perr, errStreamStop): + if perr == nil && pc.cfg.Format != FormatGemini && acc.finishReason == "" { + // Gemini completes at EOF; every other format has an explicit + // completion signal that never arrived — the provider dropped + // the stream. Never surface this as an empty success. + if acc.emitted { + return streamOutcome{ + result: acc.result(), + err: fmt.Errorf("llm: stream closed before completion: %w", errPrematureClose), + } + } + return streamOutcome{err: errPrematureClose} + } return streamOutcome{result: acc.result()} default: var ca *consumerAbort @@ -522,13 +624,14 @@ type toolCallAccum struct { // streamAccum assembles a ChatResult from SSE chunks across formats. type streamAccum struct { - content strings.Builder - reasoning strings.Builder - calls []*toolCallAccum - callIndex map[int]*toolCallAccum - finishReason string - usage Usage - emitted bool // any delta delivered to the consumer + content strings.Builder + reasoning strings.Builder + calls []*toolCallAccum + callIndex map[int]*toolCallAccum + finishReason string + usage Usage + emitted bool // any delta delivered to the consumer + thinkingSignature string // anthropic signature_delta capture } func newStreamAccum() *streamAccum { @@ -548,10 +651,11 @@ func (a *streamAccum) call(idx int) *toolCallAccum { func (a *streamAccum) result() *ChatResult { res := &ChatResult{ - Content: a.content.String(), - ReasoningContent: a.reasoning.String(), - FinishReason: a.finishReason, - Usage: a.usage, + Content: a.content.String(), + ReasoningContent: a.reasoning.String(), + ThinkingSignature: a.thinkingSignature, + FinishReason: a.finishReason, + Usage: a.usage, } for _, c := range a.calls { res.ToolCalls = append(res.ToolCalls, ToolCall{ID: c.id, Name: c.name, Arguments: c.args.String()}) diff --git a/dispatch_edges_test.go b/dispatch_edges_test.go new file mode 100644 index 0000000..023f519 --- /dev/null +++ b/dispatch_edges_test.go @@ -0,0 +1,1004 @@ +package llm + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "runtime" + "strings" + "testing" + "time" +) + +// ── callStream arms ────────────────────────────────────────────────────── + +// A stream transport failure (dropped connection) retries and succeeds. +func TestCallStreamTransportErrorRetries(t *testing.T) { + var n int + srv := httptestNewServer(func(w http.ResponseWriter, r *http.Request) { + n++ + if n == 1 { + panic(http.ErrAbortHandler) + } + w.Header().Set("Content-Type", "text/event-stream") + fmt.Fprint(w, "data: {\"choices\":[{\"delta\":{\"content\":\"ok\"}}]}\n\n") + fmt.Fprint(w, "data: [DONE]\n\n") + }) + defer srv.Close() + cc := newTestClient(t, ProviderConfig{ID: "openai", Format: FormatOpenAI, BaseURL: srv.URL, APIKey: "k"}, srv) + res, err := cc.CallStream(context.Background(), &ChatRequest{Messages: []Message{{Role: RoleUser, Content: "hi"}}}, func(Delta) error { + return nil + }) + if err != nil || res.Content != "ok" { + t.Fatalf("res=%q err=%v", res.Content, err) + } + if n != 2 { + t.Errorf("requests = %d, want 2", n) + } +} + +// Persistent transport failure exhausts the budget with a wrapped cause. +func TestCallStreamTransportExhausted(t *testing.T) { + srv := httptestNewServer(func(w http.ResponseWriter, r *http.Request) { + panic(http.ErrAbortHandler) + }) + defer srv.Close() + cc := newTestClient(t, ProviderConfig{ID: "openai", Format: FormatOpenAI, BaseURL: srv.URL, APIKey: "k"}, srv) + _, err := cc.CallStream(context.Background(), &ChatRequest{Messages: []Message{{Role: RoleUser, Content: "hi"}}}, func(Delta) error { + return nil + }) + // The final attempt's raw transport error surfaces as-is. + if err == nil || !errors.Is(err, io.EOF) { + t.Fatalf("err = %v, want the terminal transport error", err) + } +} + +// A non-retryable status during streaming surfaces immediately. +func TestCallStreamImmediateNonRetryable(t *testing.T) { + var n int + srv := httptestNewServer(func(w http.ResponseWriter, r *http.Request) { + n++ + w.WriteHeader(401) + fmt.Fprint(w, `{"error":{"message":"bad key"}}`) + }) + defer srv.Close() + cc := newTestClient(t, ProviderConfig{ID: "openai", Format: FormatOpenAI, BaseURL: srv.URL, APIKey: "k"}, srv) + _, err := cc.CallStream(context.Background(), &ChatRequest{Messages: []Message{{Role: RoleUser, Content: "hi"}}}, func(Delta) error { + return nil + }) + var ae *APIError + if !errors.As(err, &ae) || ae.Status != 401 { + t.Fatalf("err = %v, want 401 APIError", err) + } + if n != 1 { + t.Errorf("attempts = %d, want 1", n) + } +} + +// The wall-clock deadline fires while the provider never responds: the +// deadline error surfaces (no partial output, nothing to retry). +func TestCallStreamDeadlineNoOutput(t *testing.T) { + srv := httptestNewServer(func(w http.ResponseWriter, r *http.Request) { + select { + case <-r.Context().Done(): + case <-time.After(5 * time.Second): + } + }) + defer srv.Close() + cc := newTestClient(t, ProviderConfig{ID: "openai", Format: FormatOpenAI, BaseURL: srv.URL, APIKey: "k"}, srv) + cc.SetRequestTimeout(120 * time.Millisecond) + _, err := cc.CallStream(context.Background(), &ChatRequest{Messages: []Message{{Role: RoleUser, Content: "hi"}}}, func(Delta) error { + return nil + }) + if err == nil { + t.Fatal("expected deadline error") + } +} + +// Anthropic and Gemini streaming dispatch arms via the real client path. +func TestCallStreamDispatchArms(t *testing.T) { + t.Run("anthropic", func(t *testing.T) { + srv := httptestNewServer(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + fmt.Fprint(w, "data: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"text\"}}\n\n") + fmt.Fprint(w, "data: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\"A\"}}\n\n") + fmt.Fprint(w, "data: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"end_turn\"},\"usage\":{\"output_tokens\":2}}\n\n") + fmt.Fprint(w, "data: {\"type\":\"message_stop\"}\n\n") + }) + defer srv.Close() + cc := newTestClient(t, ProviderConfig{ID: "a", Format: FormatAnthropic, BaseURL: srv.URL, APIKey: "k"}, srv) + res, err := cc.CallStream(context.Background(), &ChatRequest{Messages: []Message{{Role: RoleUser, Content: "hi"}}}, func(Delta) error { + return nil + }) + if err != nil || res.Content != "A" || res.FinishReason != FinishStop { + t.Fatalf("res=%+v err=%v", res, err) + } + }) + t.Run("gemini", func(t *testing.T) { + srv := httptestNewServer(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + fmt.Fprint(w, "data: {\"candidates\":[{\"content\":{\"parts\":[{\"text\":\"G\"}]},\"finishReason\":\"STOP\"}]}\n\n") + }) + defer srv.Close() + cc := newTestClient(t, ProviderConfig{ID: "g", Format: FormatGemini, BaseURL: srv.URL, APIKey: "k"}, srv) + res, err := cc.CallStream(context.Background(), &ChatRequest{Messages: []Message{{Role: RoleUser, Content: "hi"}}}, func(Delta) error { + return nil + }) + if err != nil || res.Content != "G" || res.FinishReason != FinishStop { + t.Fatalf("res=%+v err=%v", res, err) + } + }) +} + +// ── small unit arms ────────────────────────────────────────────────────── + +func TestReasoningEffortRejectedShape(t *testing.T) { + if reasoningEffortRejected(errors.New("plain")) { + t.Error("plain error must not classify") + } + if reasoningEffortRejected(&APIError{Status: 500, Message: "reasoning_effort"}) { + t.Error("non-400 must not classify") + } + if !reasoningEffortRejected(&APIError{Status: 400, Message: "reasoning_effort unsupported"}) { + t.Error("400 + reasoning_effort must classify") + } +} + +func TestReasoningEffortNonePatchedClearsThinking(t *testing.T) { + withThinking := reasoningEffortNonePatched(oaRequest{ReasoningEffort: "high", Thinking: &oaThinking{Type: "enabled"}}) + if withThinking.ReasoningEffort != "none" || withThinking.Thinking != nil { + t.Errorf("patched = %+v, want effort none and thinking cleared", withThinking) + } + plain := reasoningEffortNonePatched(oaRequest{ReasoningEffort: "high"}) + if plain.ReasoningEffort != "none" || plain.Thinking != nil { + t.Errorf("patched = %+v", plain) + } +} + +func TestNewProviderClientDefaults(t *testing.T) { + pc := newProviderClientWithLearn(ProviderConfig{ID: "x", Format: FormatOpenAI, BaseURL: "http://127.0.0.1:1"}, nil, nil, nil) + if pc.buffered() == nil { + t.Fatal("nil buffered client must default to a usable client") + } + if pc.learn == nil { + t.Fatal("nil learn state must default to fresh state") + } +} + +func TestSetRequestTimeoutIgnoresNonPositive(t *testing.T) { + pc := newProviderClient(ProviderConfig{ID: "x", Format: FormatOpenAI}, nil, nil) + before := pc.buffered() + cc := &ChatClient{pc: pc, model: "m", parent: &Provider{cfg: pc.cfg, sdk: New()}} + cc.SetRequestTimeout(0) + cc.SetRequestTimeout(-1 * time.Second) + if pc.buffered() != before { + t.Error("non-positive timeout must not swap the client") + } +} + +// ── builders: remaining arms ───────────────────────────────────────────── + +func TestBuildAnthropicRequestArms(t *testing.T) { + req := &ChatRequest{ + System: []SystemBlock{{Text: "be terse", Cache: true}}, + Messages: []Message{ + {Role: RoleUser, Content: "q"}, + {Role: RoleAssistant, Content: " "}, // empty assistant → placeholder + {Role: RoleTool, ToolCallID: "tu_1", Content: "r1"}, + {Role: RoleTool, ToolCallID: "tu_2", Content: "r2"}, + }, + Tools: []ToolDef{{Name: "f"}}, // empty schema → default {} + Thinking: "enabled", + ThinkingBudget: 4096, + Temperature: -1, + } + b, err := buildAnthropicRequest(req, "claude-x", false) + if err != nil { + t.Fatal(err) + } + s := string(b) + for _, want := range []string{ + "be terse", "cache_control", // system block cache marking + `"budget_tokens":4096`, // explicit budget (maxInt arm) + "tool_result", `"tool_use_id":"tu_1"`, `"tool_use_id":"tu_2"`, + `"input_schema":{"type":"object"}`, // defaulted schema + } { + if !strings.Contains(s, want) { + t.Errorf("anthropic request missing %q:\n%s", want, s) + } + } + if !strings.Contains(s, `"temperature":0`) { + t.Error("negative temperature must normalize to an explicit 0") + } +} + +func TestBuildGeminiRequestArms(t *testing.T) { + cases := []struct { + thinking string + budget int + want string + notWant string + }{ + {"enabled", 0, `"thinkingConfig"`, ""}, + {"high", 0, `"thinkingConfig"`, ""}, + {"disabled", 0, `"thinkingConfig":{"thinkingBudget":0}`, ""}, // Gemini encodes "off" as budget 0 + {"", 0, "", `"thinkingConfig"`}, + } + for _, tc := range cases { + req := &ChatRequest{ + Messages: []Message{{Role: RoleSystem, Content: "sys"}, {Role: RoleUser, Content: "q"}}, + Thinking: tc.thinking, + ThinkingBudget: tc.budget, + Temperature: -1, + MaxTokens: 55, + } + b, err := buildGeminiRequest(req, "gemini-2.5-pro", false) + if err != nil { + t.Fatal(err) + } + s := string(b) + if tc.want != "" && !strings.Contains(s, tc.want) { + t.Errorf("thinking %q: missing %q in %s", tc.thinking, tc.want, s) + } + if tc.notWant != "" && strings.Contains(s, tc.notWant) { + t.Errorf("thinking %q: unexpected %q in %s", tc.thinking, tc.notWant, s) + } + if !strings.Contains(s, "sys") || !strings.Contains(s, `"maxOutputTokens":55`) { + t.Errorf("system fold / maxOutputTokens missing: %s", s) + } + } +} + +// ── model listing edges ────────────────────────────────────────────────── + +func TestModelsGetEdgePaths(t *testing.T) { + // Malformed base URL → request-build ConfigError. + pc := newProviderClient(ProviderConfig{ID: "x", Format: FormatOpenAI, BaseURL: "ht tp://bad", APIKey: "k"}, nil, nil) + if _, _, err := pc.get(context.Background(), pc.base+"/models"); err == nil { + t.Fatal("malformed URL must error") + } + // Unreachable endpoint → transport error. + pc2 := newProviderClient(ProviderConfig{ID: "x", Format: FormatOpenAI, BaseURL: "http://127.0.0.1:1", APIKey: "k"}, nil, nil) + if _, _, err := pc2.get(context.Background(), pc2.base+"/models"); err == nil { + t.Fatal("unreachable endpoint must error") + } +} + +func TestListModelsGeminiPagination(t *testing.T) { + var n int + srv := httptestNewServer(func(w http.ResponseWriter, r *http.Request) { + n++ + if n == 1 { + fmt.Fprint(w, `{"models":[{"name":"models/a"}],"nextPageToken":"PAGE2"}`) + return + } + if !strings.Contains(r.URL.RawQuery, "pageToken=PAGE2") { + t.Errorf("second page missing pageToken: %s", r.URL.RawQuery) + } + fmt.Fprint(w, `{"models":[{"name":"models/b"}]}`) + }) + got, err := newListModels(srv.URL, FormatGemini) + if err != nil || len(got) != 2 { + t.Fatalf("models = %+v err %v", got, err) + } +} + +// ── final tail: reachable-but-uncovered arms ───────────────────────────── + +func TestAnthropicThinkingBudgetArms(t *testing.T) { + // Explicit budget at the floor: maxInt takes the >= branch. + b, err := buildAnthropicRequest(&ChatRequest{ + Messages: []Message{{Role: RoleUser, Content: "q"}}, + Thinking: "enabled", ThinkingBudget: 1024, + }, "claude-x", false) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(string(b), `"budget_tokens":1024`) { + t.Errorf("floor budget missing: %s", b) + } +} + +func TestParseAnthropicResponseArms(t *testing.T) { + res, err := parseAnthropicResponse([]byte(`{"content":[{"type":"tool_use","id":"tu_9","name":"f","input":{"a":1}}],"stop_reason":"tool_use","usage":{"input_tokens":1,"output_tokens":2}}`)) + if err != nil { + t.Fatal(err) + } + if len(res.ToolCalls) != 1 || res.ToolCalls[0].ID != "tu_9" || res.ToolCalls[0].Arguments != `{"a":1}` { + t.Fatalf("tool call parse = %+v", res.ToolCalls) + } + if _, err := parseAnthropicResponse([]byte(`{not-json`)); err == nil { + t.Fatal("unparseable body must error") + } +} + +func TestListModelsAnthropicErrorAndBadDates(t *testing.T) { + srv := httptestNewServer(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(401) + }) + if _, err := newListModels(srv.URL, FormatAnthropic); err == nil { + t.Fatal("401 listing must error") + } + srv2 := httptestNewServer(func(w http.ResponseWriter, r *http.Request) { + fmt.Fprint(w, `{"data":[{"id":"x","created_at":"not-a-date"}],"has_more":false}`) + }) + got, err := newListModels(srv2.URL, FormatAnthropic) + if err != nil || len(got) != 1 || !got[0].CreatedAt.IsZero() { + t.Fatalf("bad date handling = %+v err %v", got, err) + } +} + +func TestListModelsOpenAIBadJSON(t *testing.T) { + srv := httptestNewServer(func(w http.ResponseWriter, r *http.Request) { + fmt.Fprint(w, `{oops`) + }) + if _, err := newListModels(srv.URL, FormatOpenAI); err == nil { + t.Fatal("bad listing JSON must error") + } +} + +func TestListModelsGeminiError(t *testing.T) { + srv := httptestNewServer(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(403) + }) + if _, err := newListModels(srv.URL, FormatGemini); err == nil { + t.Fatal("403 listing must error") + } +} + +func TestListModelsCancelledContext(t *testing.T) { + pc := newProviderClient(ProviderConfig{ID: "x", Format: FormatOpenAI, BaseURL: "http://127.0.0.1:1", APIKey: "k"}, nil, nil) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if _, err := pc.listModels(ctx); !errors.Is(err, context.Canceled) { + t.Fatalf("err = %v, want context.Canceled", err) + } +} + +func TestModelsGetOversizedBody(t *testing.T) { + srv := httptestNewServer(func(w http.ResponseWriter, r *http.Request) { + w.Write(make([]byte, maxModelsResponseSize+2)) + }) + pc := newProviderClient(ProviderConfig{ID: "x", Format: FormatOpenAI, BaseURL: srv.URL, APIKey: "k"}, srv.Client(), srv.Client()) + if _, _, err := pc.get(context.Background(), pc.base+"/models"); err == nil || !strings.Contains(err.Error(), "exceeds") { + t.Fatalf("err = %v, want size-cap error", err) + } +} + +func TestBufferedCallPostBuildError(t *testing.T) { + pc := newProviderClient(ProviderConfig{ID: "x", Format: FormatOpenAI, BaseURL: "http://127.0.0.1:1", APIKey: "k"}, nil, nil) + // NUL is rejected by net/url at request-build time, before any dial. + _, _, err := pc.post(context.Background(), pc.buffered(), "http://127.0.0.1:1/\x00chat", nil) + var ce *ConfigError + if !errors.As(err, &ce) { + t.Fatalf("err = %v (%T), want request-build ConfigError", err, err) + } +} + +func TestBufferedCallCancelledContext(t *testing.T) { + srv := httptestNewServer(func(w http.ResponseWriter, r *http.Request) { t.Error("must not dial") }) + defer srv.Close() + cc := newTestClient(t, ProviderConfig{ID: "openai", Format: FormatOpenAI, BaseURL: srv.URL, APIKey: "k"}, srv) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if _, err := cc.Call(ctx, &ChatRequest{Messages: []Message{{Role: RoleUser, Content: "hi"}}}); !errors.Is(err, context.Canceled) { + t.Fatalf("err = %v, want context.Canceled", err) + } +} + +func TestCallStreamRetryable500ThenSuccess(t *testing.T) { + var n int + srv := httptestNewServer(func(w http.ResponseWriter, r *http.Request) { + n++ + if n == 1 { + w.WriteHeader(500) + fmt.Fprint(w, `{"error":{"message":"blip"}}`) + return + } + w.Header().Set("Content-Type", "text/event-stream") + fmt.Fprint(w, "data: {\"choices\":[{\"delta\":{\"content\":\"ok\"}}]}\n\n") + fmt.Fprint(w, "data: [DONE]\n\n") + }) + defer srv.Close() + cc := newTestClient(t, ProviderConfig{ID: "openai", Format: FormatOpenAI, BaseURL: srv.URL, APIKey: "k"}, srv) + res, err := cc.CallStream(context.Background(), &ChatRequest{Messages: []Message{{Role: RoleUser, Content: "hi"}}}, func(Delta) error { + return nil + }) + if err != nil || res.Content != "ok" || n != 2 { + t.Fatalf("res=%q n=%d err=%v", res.Content, n, err) + } +} + +func TestMapOpenAIStreamEventBadJSON(t *testing.T) { + acc := newStreamAccum() + if _, _, err := mapOpenAIStreamEvent([]byte(`{bad`), acc); err == nil { + t.Fatal("bad chunk JSON must error") + } +} + +func TestParseGeminiResponseBadJSON(t *testing.T) { + if _, err := parseGeminiResponse([]byte(`{bad`)); err == nil { + t.Fatal("bad body must error") + } +} + +func TestConsumerAbortError(t *testing.T) { + ca := &consumerAbort{err: errors.New("inner")} + if ca.Error() != "inner" { + t.Errorf("consumerAbort.Error() = %q", ca.Error()) + } +} + +func TestBuildOpenAIRequestThinkingArms(t *testing.T) { + // GLM forced-thinking models upgrade disabled → enabled + low effort. + req := &ChatRequest{Messages: []Message{{Role: RoleUser, Content: "hi"}}, Thinking: "disabled"} + b, err := json.Marshal(buildOpenAIRequest(ProviderConfig{ + ID: "zai", Format: FormatOpenAI, + Quirks: Quirks{ThinkingObject: true, ForceThinking: []string{"glm"}, ReasoningEffort: true}, + }, req, "glm-5.3", false, false)) + if err != nil { + t.Fatal(err) + } + s := string(b) + if !strings.Contains(s, `"thinking":{"type":"enabled"}`) || !strings.Contains(s, `"reasoning_effort":"low"`) { + t.Errorf("forced-thinking upgrade missing: %s", s) + } + // Providers with no thinking support ignore the field entirely. + b2, err := json.Marshal(buildOpenAIRequest(ProviderConfig{ID: "kimi", Format: FormatOpenAI}, req, "kimi-x", false, false)) + if err != nil { + t.Fatal(err) + } + if strings.Contains(string(b2), "thinking") || strings.Contains(string(b2), "reasoning_effort") { + t.Errorf("no-quirk provider must not receive thinking fields: %s", b2) + } +} + +func TestGeminiThinkingExplicitBudget(t *testing.T) { + b, err := buildGeminiRequest(&ChatRequest{ + Messages: []Message{{Role: RoleUser, Content: "q"}}, + Thinking: "enabled", + ThinkingBudget: 2048, + }, "gemini-2.5-pro", false) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(string(b), `"thinkingBudget":2048`) { + t.Errorf("explicit budget missing: %s", b) + } +} + +func TestHTTPErrorTopLevelOpenAIAndEmptyNested(t *testing.T) { + pc := newProviderClient(ProviderConfig{ID: "o", Format: FormatOpenAI}, nil, nil) + e := pc.httpError(400, []byte(`{"message":"top","code":"c1"}`)) + if e.Message != "top" || e.Code != "c1" { + t.Fatalf("top-level = %+v", e) + } + e = pc.httpError(400, []byte(`{"error":{"type":"x"}}`)) + if e.Message != `{"error":{"type":"x"}}` { + t.Fatalf("empty nested must fall back to raw body: %+v", e) + } +} + +// A build-time rejection (unknown role) inside CallStream surfaces before +// any network activity. +func TestCallStreamUnknownRoleRejected(t *testing.T) { + var dialed bool + srv := httptestNewServer(func(w http.ResponseWriter, r *http.Request) { + dialed = true + }) + defer srv.Close() + cc := newTestClient(t, ProviderConfig{ID: "openai", Format: FormatOpenAI, BaseURL: srv.URL, APIKey: "k"}, srv) + _, err := cc.CallStream(context.Background(), &ChatRequest{Messages: []Message{{Role: Role("wizard"), Content: "hi"}}}, func(Delta) error { + return nil + }) + var ce *ConfigError + if !errors.As(err, &ce) || dialed { + t.Fatalf("err = %v dialed=%v, want pre-dial ConfigError", err, dialed) + } +} + +func TestParseAnthropicResponseUnknownBlockIgnored(t *testing.T) { + res, err := parseAnthropicResponse([]byte(`{"content":[{"type":"tool_result","content":"weird"},{"type":"text","text":"ok"}],"stop_reason":"end_turn"}`)) + if err != nil { + t.Fatal(err) + } + if res.Content != "ok" { + t.Fatalf("res = %+v", res) + } +} + +func TestAnthropicThinkingBudgetBelowFloor(t *testing.T) { + b, err := buildAnthropicRequest(&ChatRequest{ + Messages: []Message{{Role: RoleUser, Content: "q"}}, + Thinking: "enabled", ThinkingBudget: 200, // below floor → clamped to 1024 + }, "claude-x", false) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(string(b), `"budget_tokens":1024`) { + t.Errorf("floor clamp missing: %s", b) + } +} + +// ── third tail: small reachable arms ───────────────────────────────────── + +func TestChatModelFallbackArms(t *testing.T) { + var seenModel string + srv := httptestNewServer(func(w http.ResponseWriter, r *http.Request) { + var req struct { + Model string `json:"model"` + } + b, _ := io.ReadAll(r.Body) + _ = json.Unmarshal(b, &req) + seenModel = req.Model + fmt.Fprint(w, `{"choices":[{"message":{"content":"ok"},"finish_reason":"stop"}]}`) + }) + defer srv.Close() + cc := newTestClient(t, ProviderConfig{ID: "openai", Format: FormatOpenAI, BaseURL: srv.URL, APIKey: "k"}, srv) + + // Empty ChatClient model: falls back to ChatRequest.Model. + fallback := &ChatClient{pc: cc.pc, model: "", parent: cc.parent} + if _, err := fallback.Call(context.Background(), &ChatRequest{Model: "req-model", Messages: []Message{{Role: RoleUser, Content: "hi"}}}); err != nil { + t.Fatal(err) + } + if seenModel != "req-model" { + t.Errorf("model = %q, want req-model fallback", seenModel) + } + // Neither set: loud config error. + if _, err := fallback.Call(context.Background(), &ChatRequest{Messages: []Message{{Role: RoleUser, Content: "hi"}}}); err == nil || !strings.Contains(err.Error(), "no model set") { + t.Fatalf("err = %v, want no-model ConfigError", err) + } +} + +func TestCallCtxCancelDuringRetryableSleep(t *testing.T) { + srv := httptestNewServer(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(500) + }) + defer srv.Close() + cc := newTestClient(t, ProviderConfig{ID: "openai", Format: FormatOpenAI, BaseURL: srv.URL, APIKey: "k"}, srv) + backoffUnit = 500 * time.Millisecond // first retry backoff (1s) outlives ctx + t.Cleanup(func() { backoffUnit = time.Second }) + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + _, err := cc.Call(ctx, &ChatRequest{Messages: []Message{{Role: RoleUser, Content: "hi"}}}) + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("err = %v, want context deadline", err) + } +} + +func TestBufferedResponseOversized(t *testing.T) { + srv := httptestNewServer(func(w http.ResponseWriter, r *http.Request) { + w.Write(make([]byte, maxResponseSize+2)) + }) + defer srv.Close() + cc := newTestClient(t, ProviderConfig{ID: "openai", Format: FormatOpenAI, BaseURL: srv.URL, APIKey: "k"}, srv) + _, err := cc.Call(context.Background(), &ChatRequest{Messages: []Message{{Role: RoleUser, Content: "hi"}}}) + if err == nil || !strings.Contains(err.Error(), "exceeds") { + t.Fatalf("err = %v, want response size-cap error", err) + } +} + +func TestCallStreamBuildErrorNULURL(t *testing.T) { + old := backoffUnit + backoffUnit = time.Millisecond + t.Cleanup(func() { backoffUnit = old }) + pc := newProviderClient(ProviderConfig{ID: "x", Format: FormatOpenAI, BaseURL: "http://127.0.0.1:1/\x00", APIKey: "k"}, nil, nil) + cc := &ChatClient{pc: pc, model: "m", parent: &Provider{cfg: pc.cfg, sdk: New()}} + _, err := cc.CallStream(context.Background(), &ChatRequest{Messages: []Message{{Role: RoleUser, Content: "hi"}}}, func(Delta) error { + return nil + }) + var ce *ConfigError + if !errors.As(err, &ce) { + t.Fatalf("err = %v (%T), want request-build ConfigError", err, err) + } +} + +func TestOpenAIParseBadJSON(t *testing.T) { + if _, err := parseOpenAIResponse([]byte(`{oops`)); err == nil { + t.Fatal("bad body must error") + } +} + +func TestCallStreamLearnsBufferedFromStreamRejected400(t *testing.T) { + var n int + srv := httptestNewServer(func(w http.ResponseWriter, r *http.Request) { + n++ + var req struct { + Stream bool `json:"stream"` + } + b, _ := io.ReadAll(r.Body) + _ = json.Unmarshal(b, &req) + if req.Stream { + w.WriteHeader(400) + fmt.Fprint(w, `{"error":{"message":"Streaming is not supported for this model"}}`) + return + } + fmt.Fprint(w, `{"choices":[{"message":{"content":"buffered"},"finish_reason":"stop"}]}`) + }) + defer srv.Close() + cc := newTestClient(t, ProviderConfig{ID: "openai", Format: FormatOpenAI, BaseURL: srv.URL, APIKey: "k"}, srv) + res, err := cc.CallStream(context.Background(), &ChatRequest{Messages: []Message{{Role: RoleUser, Content: "hi"}}}, func(Delta) error { + return nil + }) + if err != nil || res.Content != "buffered" || n != 2 { + t.Fatalf("res=%q n=%d err=%v", res.Content, n, err) + } +} + +func TestGeminiThinkingMediumArm(t *testing.T) { + b, err := buildGeminiRequest(&ChatRequest{ + Messages: []Message{{Role: RoleUser, Content: "q"}}, + Thinking: "medium", + }, "gemini-2.5-pro", false) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(string(b), `"thinkingBudget":8192`) { + t.Errorf("medium budget missing: %s", b) + } +} + +func TestListModelsGeminiBadJSON(t *testing.T) { + srv := httptestNewServer(func(w http.ResponseWriter, r *http.Request) { + fmt.Fprint(w, `{oops`) + }) + if _, err := newListModels(srv.URL, FormatGemini); err == nil { + t.Fatal("bad listing JSON must error") + } +} + +func TestParseSSELeadingBlankLines(t *testing.T) { + done := make(chan struct{}) + ch := make(chan sseItem, 4) + go parseSSEStream(strings.NewReader("\n\n\ndata: x\n\n"), ch, done) + it := <-ch + if it.kind != sseData || string(it.data) != "x" { + t.Fatalf("leading blank lines produced %v", it) + } + if next := <-ch; next.kind != sseEnd { + t.Fatalf("expected clean end, got %v", next) + } +} + +func TestStreamAccumResultCarriesToolCalls(t *testing.T) { + acc := newStreamAccum() + c := acc.call(0) + c.id, c.name = "call_0", "f" + c.args.WriteString(`{"a":1}`) + res := acc.result() + if len(res.ToolCalls) != 1 || res.ToolCalls[0].Name != "f" || res.ToolCalls[0].Arguments != `{"a":1}` { + t.Fatalf("result tool calls = %+v", res.ToolCalls) + } +} + +// ── fourth tail: last reachable arms ───────────────────────────────────── + +func TestBuildAnthropicRequestArmsTail(t *testing.T) { + req := &ChatRequest{ + Messages: []Message{ + {Role: RoleSystem, Content: "in-band system"}, + {Role: RoleUser, Content: "q"}, + {Role: RoleAssistant, ToolCalls: []ToolCall{{ID: "tu_1", Name: "f"}}}, // empty args → {} + }, + Thinking: "enabled", // no explicit budget → 5000 default + } + b, err := buildAnthropicRequest(req, "claude-x", false) + if err != nil { + t.Fatal(err) + } + s := string(b) + for _, want := range []string{`"budget_tokens":5000`, "in-band system", `"input":{}`} { + if !strings.Contains(s, want) { + t.Errorf("missing %q in %s", want, s) + } + } +} + +func TestParseAnthropicResponseArmsTail(t *testing.T) { + res, err := parseAnthropicResponse([]byte(`{"content":[{"type":"tool_use","id":"t","name":"f"}],"stop_reason":"tool_use"}`)) + if err != nil { + t.Fatal(err) + } + if len(res.ToolCalls) != 1 || res.ToolCalls[0].Arguments != "{}" { + t.Fatalf("empty input must default to {}: %+v", res.ToolCalls) + } +} + +func TestBuildGeminiRequestArmsTail(t *testing.T) { + req := &ChatRequest{Messages: []Message{ + {Role: RoleUser, Content: "q"}, + {Role: RoleAssistant, ToolCalls: []ToolCall{{ID: "call_0", Name: "f"}}}, // empty content + empty args + }} + b, err := buildGeminiRequest(req, "gemini-2.5-pro", false) + if err != nil { + t.Fatal(err) + } + s := string(b) + if !strings.Contains(s, `"name":"f"`) || !strings.Contains(s, `"args":{}`) { + t.Errorf("empty args must default to {}: %s", s) + } + if strings.Count(s, `"text":" "`) != 0 && !strings.Contains(s, `"functionCall"`) { + t.Errorf("placeholder displaced the function call: %s", s) + } +} + +func TestMapAnthropicStreamEventBadJSON(t *testing.T) { + acc := newStreamAccum() + if _, _, err := mapAnthropicStreamEvent([]byte(`{bad`), acc); err == nil { + t.Fatal("bad event JSON must error") + } +} + +func TestListModelsAnthropicBadJSON(t *testing.T) { + srv := httptestNewServer(func(w http.ResponseWriter, r *http.Request) { + fmt.Fprint(w, `{oops`) + }) + if _, err := newListModels(srv.URL, FormatAnthropic); err == nil { + t.Fatal("bad listing JSON must error") + } +} + +func TestListModelsAnthropicPageLimitExit(t *testing.T) { + srv := httptestNewServer(func(w http.ResponseWriter, r *http.Request) { + fmt.Fprint(w, `{"data":[{"id":"x"}],"has_more":true,"last_id":"x"}`) + }) + got, err := newListModels(srv.URL, FormatAnthropic) + if err != nil { + t.Fatal(err) + } + if len(got) != 10 { + t.Errorf("models = %d, want 10 (page cap)", len(got)) + } +} + +func TestProviderListModelsErrorPropagates(t *testing.T) { + srv := httptestNewServer(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(401) + }) + sdk := New(WithProvider("openai", WithBaseURL(srv.URL), WithAPIKey("k"))) + p, _ := sdk.Provider("openai") + if _, err := p.ListModels(context.Background()); err == nil { + t.Fatal("listing failure must propagate") + } +} + +func TestFoldGeminiPartsEmptyArgs(t *testing.T) { + acc := newStreamAccum() + chunk := []byte(`{"candidates":[{"content":{"parts":[{"functionCall":{"name":"f"}}]}}]}`) + if _, _, err := mapGeminiStreamEvent(chunk, acc); err != nil { + t.Fatal(err) + } + if len(acc.calls) != 1 || acc.calls[0].args.String() != "{}" { + t.Fatalf("empty functionCall args must default to {}: %+v", acc.calls) + } +} + +// ── fifth tail: last reachable edges ───────────────────────────────────── + +// EOF with a pending un-terminated event must still deliver it. +func TestParseSSETrailingEventWithoutBlankLine(t *testing.T) { + done := make(chan struct{}) + ch := make(chan sseItem, 4) + go parseSSEStream(strings.NewReader("data: trailing"), ch, done) + it := <-ch + if it.kind != sseData || string(it.data) != "trailing" { + t.Fatalf("trailing event = %v (%q)", it.kind, it.data) + } + if next := <-ch; next.kind != sseEnd { + t.Fatalf("expected end after trailing event, got %v", next) + } +} + +// Some OpenAI-compatible providers wrap listings in "models" instead of "data". +func TestListModelsOpenAIModelsWrapper(t *testing.T) { + srv := httptestNewServer(func(w http.ResponseWriter, r *http.Request) { + fmt.Fprint(w, `{"models":[{"id":"wrapped","context_length":4096}]}`) + }) + got, err := newListModels(srv.URL, FormatOpenAI) + if err != nil || len(got) != 1 || got[0].ID != "wrapped" || got[0].ContextWindow != 4096 { + t.Fatalf("wrapped listing = %+v err %v", got, err) + } +} + +func TestBuildOpenAIRequestSystemAndToolRoles(t *testing.T) { + req := &ChatRequest{ + System: []SystemBlock{{Text: "sys"}}, + Messages: []Message{{Role: RoleSystem, Content: "band"}, {Role: RoleUser, Content: "q"}, {Role: RoleAssistant, Content: "a"}, {Role: RoleTool, ToolCallID: "t1", Content: "r"}}, + } + b, err := json.Marshal(buildOpenAIRequest(ProviderConfig{ID: "openai", Format: FormatOpenAI}, req, "m", false, false)) + if err != nil { + t.Fatal(err) + } + s := string(b) + if !strings.Contains(s, `"role":"system","content":"sys\nband"`) { + t.Errorf("system fold missing: %s", s) + } + if !strings.Contains(s, `"role":"tool","content":"r","tool_call_id":"t1"`) { + t.Errorf("tool role missing: %s", s) + } +} + +func TestListModelsRetryBreakOnCancelledContext(t *testing.T) { + old := backoffUnit + backoffUnit = time.Second + t.Cleanup(func() { backoffUnit = old }) + srv := httptestNewServer(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(500) + }) + pc := newProviderClient(ProviderConfig{ID: "x", Format: FormatOpenAI, BaseURL: srv.URL, APIKey: "k"}, srv.Client(), srv.Client()) + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + _, err := pc.listModels(ctx) + if err == nil || !strings.Contains(err.Error(), "list models failed") { + t.Fatalf("err = %v, want exhausted-listing wrap", err) + } +} + +func TestPumpSSEKeepaliveWithoutWatchdog(t *testing.T) { + before := runtime.NumGoroutine() + err := pumpSSE(context.Background(), io.NopCloser(strings.NewReader(": ping\n\ndata: x\n\n")), 0, func([]byte) error { + return nil + }) + if err != nil { + t.Fatalf("pump: %v", err) + } + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if runtime.NumGoroutine() <= before { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatal("parser goroutine did not exit") +} + +// Buffered: persistent transport failure exhausts the budget with the +// wrapped "retry exhausted" error. +func TestCallTransportExhausted(t *testing.T) { + srv := httptestNewServer(func(w http.ResponseWriter, r *http.Request) { + panic(http.ErrAbortHandler) + }) + defer srv.Close() + cc := newTestClient(t, ProviderConfig{ID: "openai", Format: FormatOpenAI, BaseURL: srv.URL, APIKey: "k"}, srv) + _, err := cc.Call(context.Background(), &ChatRequest{Messages: []Message{{Role: RoleUser, Content: "hi"}}}) + if err == nil || !strings.Contains(err.Error(), "retry exhausted") { + t.Fatalf("err = %v, want exhausted wrap", err) + } +} + +// Stream: a transport failure whose backoff consumes the remaining wall +// clock surfaces the deadline (or the exhausted wrap — either is honest). +func TestCallStreamDeadlineBetweenAttempts(t *testing.T) { + old := backoffUnit + backoffUnit = 80 * time.Millisecond + t.Cleanup(func() { backoffUnit = old }) + + var n int + srv := httptestNewServer(func(w http.ResponseWriter, r *http.Request) { + n++ + if n == 1 { + panic(http.ErrAbortHandler) // fast transport error + } + select { // hang until the wall clock kills the attempt + case <-r.Context().Done(): + case <-time.After(5 * time.Second): + } + }) + defer srv.Close() + cc := newTestClient(t, ProviderConfig{ID: "openai", Format: FormatOpenAI, BaseURL: srv.URL, APIKey: "k"}, srv) + // AFTER newTestClient (which pins a 1ms unit): a 80ms unit makes the + // post-error backoff outlive the 150ms wall clock. + backoffUnit = 80 * time.Millisecond + cc.SetRequestTimeout(150 * time.Millisecond) + res, err := cc.CallStream(context.Background(), &ChatRequest{Messages: []Message{{Role: RoleUser, Content: "hi"}}}, func(Delta) error { + return nil + }) + if err == nil || res != nil { + t.Fatalf("res=%v err=%v, want an error with no result", res, err) + } +} + +// Anthropic streaming error events surface as errors pre-delta (retryable). +func TestCallStreamAnthropicErrorEvent(t *testing.T) { + var n int + srv := httptestNewServer(func(w http.ResponseWriter, r *http.Request) { + n++ + if n == 1 { + w.Header().Set("Content-Type", "text/event-stream") + fmt.Fprint(w, "data: {\"type\":\"error\",\"error\":{\"type\":\"overloaded_error\",\"message\":\"boom\"}}\n\n") + return + } + w.Header().Set("Content-Type", "text/event-stream") + fmt.Fprint(w, "data: {\"type\":\"message_start\",\"message\":{\"usage\":{\"input_tokens\":1}}}\n\n") + fmt.Fprint(w, "data: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\"A\"}}\n\n") + fmt.Fprint(w, "data: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"end_turn\"}}\n\n") + fmt.Fprint(w, "data: {\"type\":\"message_stop\"}\n\n") + }) + defer srv.Close() + cc := newTestClient(t, ProviderConfig{ID: "a", Format: FormatAnthropic, BaseURL: srv.URL, APIKey: "k"}, srv) + res, err := cc.CallStream(context.Background(), &ChatRequest{Messages: []Message{{Role: RoleUser, Content: "hi"}}}, func(Delta) error { + return nil + }) + if err != nil || res.Content != "A" { + t.Fatalf("res=%+v err=%v", res, err) + } + if n != 2 { + t.Errorf("requests = %d, want 2 (error event retried pre-delta)", n) + } +} + +func TestMapGeminiStreamEventBadJSON(t *testing.T) { + acc := newStreamAccum() + if _, _, err := mapGeminiStreamEvent([]byte(`{bad`), acc); err == nil { + t.Fatal("bad chunk JSON must error") + } +} + +// parseSSEStream propagates reader errors verbatim (non-EOF). +type errReader struct{} + +func (errReader) Read([]byte) (int, error) { return 0, errors.New("socket gone") } + +func TestParseSSEStreamReadError(t *testing.T) { + done := make(chan struct{}) + ch := make(chan sseItem, 2) + go parseSSEStream(errReader{}, ch, done) + it := <-ch + if it.kind != sseErr || it.err == nil || strings.Contains(fmt.Sprint(it.err), "EOF") { + t.Fatalf("item = %+v, want reader error", it) + } +} + +func TestBackoffDelayNegativeClamp(t *testing.T) { + if d := backoffDelay(-1); d != 0 { + t.Errorf("backoffDelay(-1) = %v, want 0", d) + } +} + +func TestParseOpenAIResponseToolCalls(t *testing.T) { + res, err := parseOpenAIResponse([]byte(`{"choices":[{"message":{"role":"assistant","content":"","tool_calls":[{"id":"c1","type":"function","function":{"name":"f","arguments":"{\"x\":1}"}}]},"finish_reason":"tool_calls"}],"usage":{"prompt_tokens":1,"completion_tokens":2,"completion_tokens_details":{"reasoning_tokens":7}}}`)) + if err != nil { + t.Fatal(err) + } + if len(res.ToolCalls) != 1 || res.ToolCalls[0].Name != "f" || res.FinishReason != FinishToolCalls { + t.Fatalf("res = %+v", res) + } + if res.Usage.ReasoningTokens != 7 { + t.Errorf("reasoning tokens = %d", res.Usage.ReasoningTokens) + } +} + +func TestListModelsAnthropicMidPageError(t *testing.T) { + var n int + srv := httptestNewServer(func(w http.ResponseWriter, r *http.Request) { + n++ + if n == 1 { + fmt.Fprint(w, `{"data":[{"id":"a"}],"has_more":true,"last_id":"a"}`) + return + } + w.WriteHeader(500) + }) + _, err := newListModels(srv.URL, FormatAnthropic) + var ae *APIError + if err == nil || !errors.As(err, &ae) { + t.Fatalf("err = %v, want the mid-page APIError", err) + } +} + +func TestListModelsGeminiMidPageError(t *testing.T) { + var n int + srv := httptestNewServer(func(w http.ResponseWriter, r *http.Request) { + n++ + if n == 1 { + fmt.Fprint(w, `{"models":[{"name":"models/a"}],"nextPageToken":"P"}`) + return + } + w.WriteHeader(500) + }) + if _, err := newListModels(srv.URL, FormatGemini); err == nil { + t.Fatal("mid-page failure must error") + } +} diff --git a/errors.go b/errors.go index 810422f..453e5a0 100644 --- a/errors.go +++ b/errors.go @@ -46,6 +46,10 @@ type RateLimitError struct { RetryAfter time.Duration } +// Unwrap exposes the embedded APIError so errors.As(err, *APIError) +// reaches Status/Retryable without a type switch. +func (e *RateLimitError) Unwrap() error { return &e.APIError } + func (e *RateLimitError) Error() string { s := fmt.Sprintf("llm: %s: rate limited after %d attempts", e.Provider, e.Attempts) if e.RetryAfter > 0 { diff --git a/gemini.go b/gemini.go index a0f182c..c6e8530 100644 --- a/gemini.go +++ b/gemini.go @@ -123,6 +123,7 @@ func buildGeminiRequest(req *ChatRequest, model string, stream bool) ([]byte, er out.SystemInstruction = &gmContent{Parts: sysParts} } + toolNameByID := make(map[string]string) for i := 0; i < len(req.Messages); i++ { m := req.Messages[i] switch m.Role { @@ -134,6 +135,11 @@ func buildGeminiRequest(req *ChatRequest, model string, stream bool) ([]byte, er Parts: []gmPart{{Text: m.Content}}, }) case RoleAssistant: + for _, tc := range m.ToolCalls { + if tc.ID != "" { + toolNameByID[tc.ID] = tc.Name + } + } parts := []gmPart{} if m.Content != "" { parts = append(parts, gmPart{Text: m.Content}) @@ -155,9 +161,19 @@ func buildGeminiRequest(req *ChatRequest, model string, stream bool) ([]byte, er var parts []gmPart for ; i < len(req.Messages) && req.Messages[i].Role == RoleTool; i++ { tm := req.Messages[i] + name := tm.ToolName + if name == "" { + // Consumers ported from the OpenAI format set only + // ToolCallID; recover the function name from the + // assistant tool_call it answers. + name = toolNameByID[tm.ToolCallID] + } + if name == "" { + return nil, &ConfigError{Msg: "tool result for \"" + tm.ToolCallID + "\" has no ToolName and no matching assistant tool_call"} + } parts = append(parts, gmPart{ FunctionResponse: &gmFnResp{ - Name: tm.ToolName, + Name: name, Response: wrapToolResponse(tm.Content), }, }) @@ -170,11 +186,7 @@ func buildGeminiRequest(req *ChatRequest, model string, stream bool) ([]byte, er if len(req.Tools) > 0 { g := gmToolGroup{} for _, t := range req.Tools { - g.FunctionDeclarations = append(g.FunctionDeclarations, gmFnDecl{ - Name: t.Name, - Description: t.Description, - Parameters: t.Parameters, - }) + g.FunctionDeclarations = append(g.FunctionDeclarations, gmFnDecl(t)) } out.Tools = []gmToolGroup{g} } @@ -237,7 +249,8 @@ func mapGeminiFinishReason(s string) string { case "": return "" default: - return strings.ToLower(s) + // Provider-specific reasons stay out of the canonical vocabulary. + return "" } } @@ -307,7 +320,7 @@ func parseGeminiResponse(body []byte) (*ChatResult, error) { type gmStreamChunk struct { Candidates []gmCandidate `json:"candidates"` - UsageMetadata gmUsage `json:"usageMetadata"` + UsageMetadata *gmUsage `json:"usageMetadata"` // nil = chunk carries no usage update } // mapGeminiStreamEvent folds one Gemini SSE chunk into acc. Chunks carry @@ -318,6 +331,9 @@ func mapGeminiStreamEvent(data []byte, acc *streamAccum) ([]Delta, bool, error) if err := json.Unmarshal(data, &c); err != nil { return nil, false, fmt.Errorf("llm: parse stream chunk: %w", err) } + if c.UsageMetadata != nil { + acc.usage = mapGeminiUsage(*c.UsageMetadata) + } if len(c.Candidates) == 0 { return nil, false, nil } @@ -326,7 +342,6 @@ func mapGeminiStreamEvent(data []byte, acc *streamAccum) ([]Delta, bool, error) if cand.FinishReason != "" { acc.finishReason = mapGeminiFinishReason(cand.FinishReason) } - acc.usage = mapGeminiUsage(c.UsageMetadata) return deltas, false, nil } diff --git a/gemini_test.go b/gemini_test.go index 37818be..87d4479 100644 --- a/gemini_test.go +++ b/gemini_test.go @@ -250,3 +250,18 @@ func TestListModelsGemini_PaginationAndLimits(t *testing.T) { t.Errorf("capabilities = %v", m0.Capabilities) } } + +// Regression: a trailing stream chunk without usageMetadata must not wipe +// usage accumulated from earlier chunks. +func TestMapGeminiStreamEventUsageNotWiped(t *testing.T) { + acc := newStreamAccum() + if _, _, err := mapGeminiStreamEvent([]byte(`{"candidates":[{"content":{"parts":[{"text":"a"}]}}],"usageMetadata":{"promptTokenCount":10,"candidatesTokenCount":5}}`), acc); err != nil { + t.Fatal(err) + } + if _, _, err := mapGeminiStreamEvent([]byte(`{"candidates":[{"finishReason":"STOP","content":{"parts":[{"text":"b"}]}}]}`), acc); err != nil { + t.Fatal(err) + } + if acc.usage.PromptTokens != 10 || acc.usage.CompletionTokens != 5 { + t.Fatalf("usage wiped by trailing chunk: %+v, want {10 5 0}", acc.usage) + } +} diff --git a/llm.go b/llm.go index 757f46e..4dc7db8 100644 --- a/llm.go +++ b/llm.go @@ -3,6 +3,7 @@ package llm import ( "context" "net/http" + "strings" "sync" "time" ) @@ -90,9 +91,9 @@ func FromEnv() Option { return WithEnv(lookupEnv) } // custom provider (requires WithFormat and WithBaseURL, plus auth). func WithProvider(id string, popts ...ProviderOption) Option { return func(s *SDK) { - existing, exists := s.providers[id] + existing := s.providers[id] cfg := ProviderConfig{ID: id} - if exists { + if existing != nil { cfg = existing.cfg } for _, po := range popts { @@ -101,11 +102,10 @@ func WithProvider(id string, popts ...ProviderOption) Option { } } s.put(cfg) - if !exists { - // Custom entry: validate before it can poison requests. - if p, ok := s.get(id); ok { - p.invalid = validateProviderConfig(p.cfg) != nil - } + // Validate every entry — custom or overridden built-in — so config + // typos fail at wiring time instead of per request. + if p, ok := s.get(id); ok { + p.invalid = validateProviderConfig(p.cfg) != nil } } } @@ -195,13 +195,14 @@ func (s *SDK) Chat(providerID, model string) (*ChatClient, error) { return nil, err } if !p.Authenticated() { - return nil, &ConfigError{Msg: providerID + " has no API key (set " + providerID + "_API_KEY or use WithAPIKey)"} + return nil, &ConfigError{Msg: providerID + " has no API key (set " + strings.ToUpper(providerID) + "_API_KEY or use WithAPIKey)"} } if p.invalid { return nil, &ConfigError{Msg: providerID + " has an invalid configuration"} } + p.chatLearnOnce.Do(func() { p.chatLearn = &learnOnce{} }) return &ChatClient{ - pc: newProviderClient(p.cfg, newBufferedHTTP(p.sdk.rt, p.sdk.timeout), newStreamHTTP(p.sdk.rt)), + pc: newProviderClientWithLearn(p.cfg, newBufferedHTTP(p.sdk.rt, p.sdk.timeout), newStreamHTTP(p.sdk.rt), p.chatLearn), model: model, parent: p, }, nil @@ -218,6 +219,11 @@ type Provider struct { clientOnce sync.Once listClient *providerClient + // Learn-once fallback state shared by every Chat() client of this + // provider: a constraint one client learns, all of them honor. + chatLearnOnce sync.Once + chatLearn *learnOnce + mu sync.Mutex cached []Model cachedAt time.Time @@ -303,7 +309,7 @@ func (c *ChatClient) SetRequestTimeout(d time.Duration) { if d <= 0 { return } - c.pc.http = newBufferedHTTP(c.pc.http.Transport, d) + c.pc.bufPtr.Store(newBufferedHTTP(c.pc.buffered().Transport, d)) } // ProviderID returns the bound provider's id. diff --git a/message.go b/message.go index 0bd6deb..6129f22 100644 --- a/message.go +++ b/message.go @@ -47,15 +47,21 @@ type ToolCall struct { // Message is one canonical chat message. For RoleTool messages, ToolCallID // and ToolName identify the call being answered and Content carries the // tool result. ReasoningContent is provider-reported thinking text -// (deepseek-reasoner, anthropic thinking, gemini thoughts); it is echoed -// back to providers only where they accept it and is otherwise advisory. +// (deepseek-reasoner, anthropic thinking, gemini thoughts). It is consumer +// metadata; the SDK replays it back only where a provider requires it for +// conversation continuity — Anthropic, and only when ThinkingSignature is +// also set (extended-thinking tool loops mandate the signed thinking block +// as the first block of the replayed assistant turn). type Message struct { Role Role Content string ReasoningContent string - ToolCalls []ToolCall - ToolCallID string - ToolName string + // ThinkingSignature authenticates ReasoningContent for providers that + // require thinking to be replayed verbatim (Anthropic signature). + ThinkingSignature string + ToolCalls []ToolCall + ToolCallID string + ToolName string } // SystemBlock is one system-prompt segment. On Anthropic each block maps to @@ -102,9 +108,13 @@ type ChatRequest struct { type ChatResult struct { Content string ReasoningContent string - ToolCalls []ToolCall - FinishReason string - Usage Usage + // ThinkingSignature authenticates ReasoningContent (Anthropic extended + // thinking). Consumers must carry it back on the next assistant Message + // for tool loops to stay valid. + ThinkingSignature string + ToolCalls []ToolCall + FinishReason string + Usage Usage } // DeltaKind discriminates streamed fragments. diff --git a/models.go b/models.go index 71ff505..741d69f 100644 --- a/models.go +++ b/models.go @@ -62,11 +62,11 @@ func (pc *providerClient) get(ctx context.Context, url string) ([]byte, time.Dur req.Header.Set("Accept", "application/json") pc.setAuthHeaders(req.Header) - resp, err := pc.http.Do(req) + resp, err := pc.buffered().Do(req) if err != nil { return nil, 0, err } - defer resp.Body.Close() + defer func() { _ = resp.Body.Close() }() data, err := io.ReadAll(io.LimitReader(resp.Body, maxModelsResponseSize+1)) if err != nil { return nil, 0, err diff --git a/models_flow_test.go b/models_flow_test.go new file mode 100644 index 0000000..7153e1e --- /dev/null +++ b/models_flow_test.go @@ -0,0 +1,319 @@ +package llm + +import ( + "context" + "errors" + "fmt" + "net/http" + "strings" + "testing" + "time" +) + +// ── buffered dispatch per wire format (parseResponse arms) ─────────────── + +func TestCallBufferedPerFormatDispatch(t *testing.T) { + cases := []struct { + name string + format Format + quirks Quirks + body string + check func(*testing.T, *ChatResult) + }{ + { + name: "openai", + format: FormatOpenAI, + body: `{"choices":[{"message":{"role":"assistant","content":"hey","reasoning_content":"thought"},"finish_reason":"tool_calls","tool_calls":[]}],"usage":{"prompt_tokens":2,"completion_tokens":3}}`, + check: func(t *testing.T, r *ChatResult) { + if r.Content != "hey" || r.ReasoningContent != "thought" || r.FinishReason != FinishToolCalls { + t.Fatalf("openai parse = %+v", r) + } + }, + }, + { + name: "anthropic", + format: FormatAnthropic, + body: `{"content":[{"type":"text","text":"A"}],"stop_reason":"max_tokens","usage":{"input_tokens":4,"output_tokens":6}}`, + check: func(t *testing.T, r *ChatResult) { + if r.Content != "A" || r.FinishReason != FinishLength || r.Usage.CompletionTokens != 6 { + t.Fatalf("anthropic parse = %+v", r) + } + }, + }, + { + name: "gemini", + format: FormatGemini, + body: `{"candidates":[{"content":{"parts":[{"text":"G"}]},"finishReason":"STOP"}],"usageMetadata":{"promptTokenCount":1,"candidatesTokenCount":2}}`, + check: func(t *testing.T, r *ChatResult) { + if r.Content != "G" || r.FinishReason != FinishStop || r.Usage.PromptTokens != 1 { + t.Fatalf("gemini parse = %+v", r) + } + }, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + srv := httptestNewServer(func(w http.ResponseWriter, r *http.Request) { + w.Write([]byte(tc.body)) + }) + defer srv.Close() + cc := newTestClient(t, ProviderConfig{ID: "x", Format: tc.format, BaseURL: srv.URL, APIKey: "k", Quirks: tc.quirks}, srv) + res, err := cc.Call(context.Background(), &ChatRequest{Messages: []Message{{Role: RoleUser, Content: "hi"}}}) + if err != nil { + t.Fatal(err) + } + tc.check(t, res) + }) + } +} + +// Parse failures must surface as descriptive errors, not empty results. +func TestBufferedParseErrorPaths(t *testing.T) { + cases := []struct { + name string + format Format + body string + want string + }{ + {"openai provider error", FormatOpenAI, `{"error":{"message":"boom"}}`, "provider error"}, + {"openai no choices", FormatOpenAI, `{"choices":[]}`, "no choices"}, + {"gemini no candidates", FormatGemini, `{"candidates":[]}`, "no candidates"}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + srv := httptestNewServer(func(w http.ResponseWriter, r *http.Request) { + w.Write([]byte(tc.body)) + }) + defer srv.Close() + cc := newTestClient(t, ProviderConfig{ID: "x", Format: tc.format, BaseURL: srv.URL, APIKey: "k"}, srv) + _, err := cc.Call(context.Background(), &ChatRequest{Messages: []Message{{Role: RoleUser, Content: "hi"}}}) + if err == nil || !strings.Contains(err.Error(), tc.want) { + t.Fatalf("err = %v, want containing %q", err, tc.want) + } + }) + } +} + +// ── retry state machine arms ───────────────────────────────────────────── + +// A transport-level failure (dropped connection) retries and then succeeds. +func TestCallTransportErrorRetries(t *testing.T) { + var n int + srv := httptestNewServer(func(w http.ResponseWriter, r *http.Request) { + n++ + if n <= 2 { + panic(http.ErrAbortHandler) // drop the connection + } + fmt.Fprint(w, `{"choices":[{"message":{"content":"ok"},"finish_reason":"stop"}]}`) + }) + defer srv.Close() + cc := newTestClient(t, ProviderConfig{ID: "openai", Format: FormatOpenAI, BaseURL: srv.URL, APIKey: "k"}, srv) + res, err := cc.Call(context.Background(), &ChatRequest{Messages: []Message{{Role: RoleUser, Content: "hi"}}}) + if err != nil { + t.Fatalf("Call: %v", err) + } + if res.Content != "ok" || n != 3 { + t.Fatalf("res=%q attempts=%d", res.Content, n) + } +} + +// A definitive 4xx after earlier 429s surfaces that 4xx (the request is +// definitively broken — retrying the rate limit was pointless). +func TestCallDefinitiveFailureAfter429(t *testing.T) { + var n int + srv := httptestNewServer(func(w http.ResponseWriter, r *http.Request) { + n++ + if n == 1 { + w.Header().Set("Retry-After", "0") + w.WriteHeader(429) + fmt.Fprint(w, `{"error":{"message":"slow"}}`) + return + } + w.WriteHeader(401) + fmt.Fprint(w, `{"error":{"message":"bad key"}}`) + }) + defer srv.Close() + cc := newTestClient(t, ProviderConfig{ID: "openai", Format: FormatOpenAI, BaseURL: srv.URL, APIKey: "k"}, srv) + _, err := cc.Call(context.Background(), &ChatRequest{Messages: []Message{{Role: RoleUser, Content: "hi"}}}) + var ae *APIError + if !errors.As(err, &ae) || ae.Status != 401 { + t.Fatalf("err = %v, want the definitive 401", err) + } + if n != 2 { + t.Errorf("attempts = %d, want 2", n) + } +} + +// A streamed request answered with a non-SSE body learns the buffered +// fallback mid-stream and completes on the buffered path. +func TestCallStreamLearnsBufferedFromNonSSEBody(t *testing.T) { + var n int + srv := httptestNewServer(func(w http.ResponseWriter, r *http.Request) { + n++ + w.Header().Set("Content-Type", "application/json") + fmt.Fprint(w, `{"choices":[{"message":{"content":"buffered"},"finish_reason":"stop"}]}`) + }) + defer srv.Close() + cc := newTestClient(t, ProviderConfig{ID: "openai", Format: FormatOpenAI, BaseURL: srv.URL, APIKey: "k"}, srv) + res, err := cc.CallStream(context.Background(), &ChatRequest{Messages: []Message{{Role: RoleUser, Content: "hi"}}}, func(Delta) error { + return nil + }) + if err != nil { + t.Fatalf("CallStream: %v", err) + } + if res.Content != "buffered" { + t.Errorf("res = %q", res.Content) + } + if n != 2 { + t.Errorf("requests = %d, want 2 (learn, then buffered retry)", n) + } + // The learned fallback must short-circuit the NEXT CallStream of the + // same client straight onto the buffered path. + res2, err := cc.CallStream(context.Background(), &ChatRequest{Messages: []Message{{Role: RoleUser, Content: "hi"}}}, func(Delta) error { + return nil + }) + if err != nil || res2.Content != "buffered" { + t.Fatalf("second stream: res=%q err=%v", res2.Content, err) + } + if n != 3 { + t.Errorf("requests = %d, want 3 (entry fast-path, no re-learn)", n) + } +} + +// ── provider error-body parsing per format ─────────────────────────────── + +func TestHTTPErrorPerFormat(t *testing.T) { + cases := []struct { + name string + format Format + body string + message string + code string + }{ + {"anthropic", FormatAnthropic, `{"error":{"type":"invalid_request_error","message":"am"}}`, "am", "invalid_request_error"}, + {"gemini", FormatGemini, `{"error":{"message":"gm","status":"INVALID_ARGUMENT"}}`, "gm", "INVALID_ARGUMENT"}, + {"openai numeric code", FormatOpenAI, `{"error":{"message":"om","code":42}}`, "om", ""}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + pc := newProviderClient(ProviderConfig{ID: "p", Format: tc.format}, nil, nil) + e := pc.httpError(400, []byte(tc.body)) + if e.Message != tc.message || e.Code != tc.code { + t.Fatalf("APIError = %+v, want message %q code %q", e, tc.message, tc.code) + } + }) + } +} + +// ── auth headers per format ────────────────────────────────────────────── + +func TestSetAuthHeaders(t *testing.T) { + anthropic := newProviderClient(ProviderConfig{ID: "a", Format: FormatAnthropic, APIKey: "ak"}, nil, nil) + h := http.Header{} + anthropic.setAuthHeaders(h) + if h.Get("x-api-key") != "ak" || h.Get("anthropic-version") != "2023-06-01" { + t.Errorf("anthropic headers = %v", h) + } + anthropicCustom := newProviderClient(ProviderConfig{ID: "a", Format: FormatAnthropic, APIKey: "ak", Quirks: Quirks{AnthropicVersion: "2044-01-01"}}, nil, nil) + h = http.Header{} + anthropicCustom.setAuthHeaders(h) + if h.Get("anthropic-version") != "2044-01-01" { + t.Errorf("custom version = %q", h.Get("anthropic-version")) + } + gemini := newProviderClient(ProviderConfig{ID: "g", Format: FormatGemini, APIKey: "gk"}, nil, nil) + h = http.Header{} + gemini.setAuthHeaders(h) + if h.Get("x-goog-api-key") != "gk" { + t.Errorf("gemini headers = %v", h) + } + bearer := newProviderClient(ProviderConfig{ID: "o", Format: FormatOpenAI, APIKey: "ok"}, nil, nil) + h = http.Header{} + bearer.setAuthHeaders(h) + if h.Get("Authorization") != "Bearer ok" { + t.Errorf("bearer = %q", h.Get("Authorization")) + } +} + +// ── canonical finish-reason vocabulary ─────────────────────────────────── + +func TestFinishReasonTables(t *testing.T) { + for _, tc := range []struct{ in, want string }{ + {"stop", FinishStop}, {"length", FinishLength}, + {"tool_calls", FinishToolCalls}, {"function_call", FinishToolCalls}, + {"content_filter", FinishContentFilter}, {"", ""}, + {"mystery_reason", ""}, // non-canonical values never leak + } { + if got := mapOpenAIFinishReason(tc.in); got != tc.want { + t.Errorf("mapOpenAIFinishReason(%q) = %q, want %q", tc.in, got, tc.want) + } + } + for _, tc := range []struct{ in, want string }{ + {"end_turn", FinishStop}, {"stop_sequence", FinishStop}, + {"max_tokens", FinishLength}, {"tool_use", FinishToolCalls}, + {"refusal", FinishContentFilter}, {"", ""}, {"pause_turn", ""}, + } { + if got := mapAnthropicStopReason(tc.in); got != tc.want { + t.Errorf("mapAnthropicStopReason(%q) = %q, want %q", tc.in, got, tc.want) + } + } + for _, tc := range []struct{ in, want string }{ + {"STOP", FinishStop}, {"MAX_TOKENS", FinishLength}, + {"SAFETY", FinishContentFilter}, {"RECITATION", FinishContentFilter}, + {"BLOCKLIST", FinishContentFilter}, {"", ""}, {"OTHER", ""}, + } { + if got := mapGeminiFinishReason(tc.in); got != tc.want { + t.Errorf("mapGeminiFinishReason(%q) = %q, want %q", tc.in, got, tc.want) + } + } +} + +// ── SSE edges ──────────────────────────────────────────────────────────── + +// A single event assembled from multiple data lines may reach the 4 MiB +// event cap; oversized events are rejected with errSSEOversized. +func TestParseSSEStreamOversizedEvent(t *testing.T) { + chunk := strings.Repeat("x", 700*1024) // 700 KiB per line, under the line cap + var b strings.Builder + for i := 0; i < 7; i++ { // 4.9 MiB accumulated into ONE event + b.WriteString("data: " + chunk + "\n") + } + b.WriteString("\n") + done := make(chan struct{}) + ch := make(chan sseItem, 2) + go parseSSEStream(strings.NewReader(b.String()), ch, done) + it := <-ch + if it.kind != sseErr || !errors.Is(it.err, errSSEOversized) { + t.Fatalf("oversized event: kind %v err %v", it.kind, it.err) + } +} + +// data payload whitespace: "data:x" keeps x, "data: x" strips one space. +func TestSSEPayloadTrimming(t *testing.T) { + ch := make(chan sseItem, 4) + done := make(chan struct{}) + go parseSSEStream(strings.NewReader("data:nospace\n\ndata: withspace\n\n"), ch, done) + first := <-ch + if string(first.data) != "nospace" { + t.Fatalf("first = %q", first.data) + } + second := <-ch + if string(second.data) != "withspace" { + t.Fatalf("second = %q", second.data) + } +} + +// ── backoff bounds ─────────────────────────────────────────────────────── + +func TestBackoffDelayCap(t *testing.T) { + old := backoffUnit + backoffUnit = 10 * time.Second + t.Cleanup(func() { backoffUnit = old }) + for attempt := 1; attempt <= 8; attempt++ { + if d := backoffDelay(attempt); d > maxRetryBackoff { + t.Fatalf("backoffDelay(%d) = %v exceeds cap", attempt, d) + } + } + if d := backoffDelay(5); d != maxRetryBackoff { + t.Errorf("backoffDelay(5) = %v, want capped %v", d, maxRetryBackoff) + } +} diff --git a/openai.go b/openai.go index 215a9df..52d80bc 100644 --- a/openai.go +++ b/openai.go @@ -50,24 +50,31 @@ type oaStreamOptions struct { } type oaRequest struct { - Model string `json:"model"` - Messages []oaMessage `json:"messages"` - Tools []oaToolDef `json:"tools,omitempty"` - MaxTokens int `json:"max_tokens,omitempty"` - Temperature *float64 `json:"temperature,omitempty"` - Stream bool `json:"stream,omitempty"` - StreamOptions *oaStreamOptions `json:"stream_options,omitempty"` - ReasoningEffort string `json:"reasoning_effort,omitempty"` - Thinking *oaThinking `json:"thinking,omitempty"` + Model string `json:"model"` + Messages []oaMessage `json:"messages"` + Tools []oaToolDef `json:"tools,omitempty"` + MaxTokens int `json:"max_tokens,omitempty"` + MaxCompletionTokens int `json:"max_completion_tokens,omitempty"` + Temperature *float64 `json:"temperature,omitempty"` + Stream bool `json:"stream,omitempty"` + StreamOptions *oaStreamOptions `json:"stream_options,omitempty"` + ReasoningEffort string `json:"reasoning_effort,omitempty"` + Thinking *oaThinking `json:"thinking,omitempty"` } // buildOpenAIRequest renders the canonical request in OpenAI format. func buildOpenAIRequest(cfg ProviderConfig, req *ChatRequest, model string, stream, includeStreamOptions bool) oaRequest { q := cfg.Quirks out := oaRequest{ - Model: model, - MaxTokens: req.MaxTokens, - Stream: stream, + Model: model, + Stream: stream, + } + // OpenAI o-series/gpt-5 models reject max_tokens in favor of + // max_completion_tokens; everyone else keeps the classic parameter. + if modelUsesCompletionTokens(model) { + out.MaxCompletionTokens = req.MaxTokens + } else { + out.MaxTokens = req.MaxTokens } // System prompt: canonical blocks (+ any in-band system messages) @@ -119,7 +126,7 @@ func buildOpenAIRequest(cfg ProviderConfig, req *ChatRequest, model string, stre out.Messages = msgs for _, t := range req.Tools { - fn, _ := json.Marshal(oaToolFn{Name: t.Name, Description: t.Description, Parameters: t.Parameters}) + fn, _ := json.Marshal(oaToolFn(t)) out.Tools = append(out.Tools, oaToolDef{Type: "function", Function: fn}) } @@ -245,7 +252,8 @@ func mapOpenAIFinishReason(s string) string { case "": return "" default: - return s + // Non-canonical reasons never leak into the public vocabulary. + return "" } } diff --git a/openai_test.go b/openai_test.go index 13e6b34..27992a8 100644 --- a/openai_test.go +++ b/openai_test.go @@ -328,3 +328,33 @@ func TestListModelsOpenAI_ModelsWrapper(t *testing.T) { t.Fatalf("models = %+v", models) } } + +// Models in the o-series/gpt-5 family reject max_tokens in favor of +// max_completion_tokens; everyone else keeps the classic parameter. +func TestBuildOpenAIRequestTokenParamRouting(t *testing.T) { + req := &ChatRequest{Messages: []Message{{Role: RoleUser, Content: "hi"}}, MaxTokens: 100} + for _, model := range []string{"o3", "o4-mini", "gpt-5-mini"} { + b, err := json.Marshal(buildOpenAIRequest(ProviderConfig{ID: "openai", Format: FormatOpenAI}, req, model, false, false)) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(string(b), `"max_completion_tokens":100`) { + t.Errorf("%s: body missing max_completion_tokens: %s", model, b) + } + if strings.Contains(string(b), `"max_tokens"`) { + t.Errorf("%s: body must not send max_tokens: %s", model, b) + } + } + for _, model := range []string{"gpt-4o", "deepseek-v4"} { + b, err := json.Marshal(buildOpenAIRequest(ProviderConfig{ID: "openai", Format: FormatOpenAI}, req, model, false, false)) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(string(b), `"max_tokens":100`) { + t.Errorf("%s: body missing max_tokens: %s", model, b) + } + if strings.Contains(string(b), "max_completion_tokens") { + t.Errorf("%s: body must not send max_completion_tokens: %s", model, b) + } + } +} diff --git a/provider.go b/provider.go index 946710b..bef1004 100644 --- a/provider.go +++ b/provider.go @@ -100,6 +100,24 @@ func modelForbidsTemperature(model string) bool { return false } +// completionTokenModels lists model prefixes that reject max_tokens in +// favor of max_completion_tokens (OpenAI o-series and gpt-5 families). +// Deliberately narrower than temperatureForbiddenModels: Z.ai/Moonshot +// models documented against max_tokens keep it. +var completionTokenModels = []string{"o1", "o3", "o4", "gpt-5"} + +// modelUsesCompletionTokens reports whether this model's token limit must +// be sent as max_completion_tokens. +func modelUsesCompletionTokens(model string) bool { + m := strings.ToLower(model) + for _, prefix := range completionTokenModels { + if strings.HasPrefix(m, prefix) { + return true + } + } + return false +} + // builtinProviders returns the built-in registry, in stable order. func builtinProviders() []ProviderConfig { return []ProviderConfig{ diff --git a/provider_registry_test.go b/provider_registry_test.go new file mode 100644 index 0000000..caef53a --- /dev/null +++ b/provider_registry_test.go @@ -0,0 +1,110 @@ +package llm + +import ( + "context" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" +) + +// Regression: learn-once fallbacks must live on the Provider (shared), not +// on each ChatClient — every sdk.Chat() minted a fresh client that re-paid +// the provider's rejection round-trip. +func TestLearnOnceSharedAcrossChatClients(t *testing.T) { + var mu sync.Mutex + var bodies []string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + b, _ := io.ReadAll(r.Body) + mu.Lock() + bodies = append(bodies, string(b)) + n := len(bodies) + mu.Unlock() + if n == 1 { + w.WriteHeader(400) + fmt.Fprint(w, `{"error":{"message":"'stream_options.include_usage' is not supported by this model"}}`) + return + } + w.Header().Set("Content-Type", "text/event-stream") + fmt.Fprint(w, "data: {\"choices\":[{\"delta\":{\"content\":\"ok\"}}]}\n\n") + fmt.Fprint(w, "data: [DONE]\n\n") + })) + defer srv.Close() + + sdk := New(WithProvider("openai", + WithFormat(FormatOpenAI), WithBaseURL(srv.URL), WithAPIKey("k"))) + req := &ChatRequest{Messages: []Message{{Role: RoleUser, Content: "hi"}}} + + c1, err := sdk.Chat("openai", "test-model") + if err != nil { + t.Fatal(err) + } + if _, err := c1.CallStream(context.Background(), req, func(Delta) error { return nil }); err != nil { + t.Fatalf("first client stream: %v", err) + } + + c2, err := sdk.Chat("openai", "test-model") // fresh client, same provider + if err != nil { + t.Fatal(err) + } + if _, err := c2.CallStream(context.Background(), req, func(Delta) error { return nil }); err != nil { + t.Fatalf("second client stream: %v", err) + } + + mu.Lock() + defer mu.Unlock() + streamOptionsReqs := 0 + for _, b := range bodies { + if strings.Contains(b, "stream_options") { + streamOptionsReqs++ + } + } + if streamOptionsReqs != 1 { + t.Errorf("%d/%d requests carried stream_options, want exactly 1 (the learning request)", streamOptionsReqs, len(bodies)) + } + if len(bodies) != 3 { + t.Errorf("requests = %d, want 3 (learn once, then two clean streams)", len(bodies)) + } +} + +// Regression: overriding a built-in provider must run the same wiring-time +// validation as a custom registration — config typos should fail loudly at +// startup, not per request. +func TestWithProviderOverrideInvalidConfigRejected(t *testing.T) { + sdk := New(WithProvider("openai", WithBaseURL("htps://typo.example"), WithAPIKey("k"))) + _, err := sdk.Chat("openai", "gpt-4o") + var ce *ConfigError + if !errors.As(err, &ce) { + t.Fatalf("err = %v (%T), want *ConfigError at wiring time", err, err) + } +} + +// The unauthenticated-provider hint must name the real env var +// (DEEPSEEK_API_KEY), not the mixed-case literal provider id. +func TestUnauthenticatedHintUsesUppercaseEnvName(t *testing.T) { + sdk := New(WithEnv(func(string) (string, bool) { return "", false })) + _, err := sdk.Chat("deepseek", "m") + if err == nil || !strings.Contains(err.Error(), "DEEPSEEK_API_KEY") { + t.Fatalf("err = %v, want hint naming DEEPSEEK_API_KEY", err) + } +} + +// FromEnv is the production entry point; it must resolve keys via the real +// environment. +func TestFromEnvRegistersAuthenticatedProviders(t *testing.T) { + t.Setenv("DEEPSEEK_API_KEY", "test-key") + sdk := New(FromEnv()) + found := false + for _, p := range sdk.Providers() { + if p.ID() == "deepseek" { + found = p.Authenticated() + } + } + if !found { + t.Fatal("DEEPSEEK_API_KEY set but deepseek provider not authenticated") + } +} diff --git a/registry_options_test.go b/registry_options_test.go new file mode 100644 index 0000000..2020fb2 --- /dev/null +++ b/registry_options_test.go @@ -0,0 +1,288 @@ +package llm + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" +) + +// httptestNewServer is a shorthand for the handler-shaped server used +// throughout these tests. +func httptestNewServer(h func(w http.ResponseWriter, r *http.Request)) *httptest.Server { + return httptest.NewServer(http.HandlerFunc(h)) +} + +// newListModels drives providerClient.listModels against a test server. +func newListModels(baseURL string, f Format) ([]Model, error) { + pc := newProviderClient(ProviderConfig{ID: "x", Format: f, BaseURL: baseURL, APIKey: "k"}, nil, nil) + return pc.listModels(context.Background()) +} + +// ── error rendering ────────────────────────────────────────────────────── + +func TestErrorStrings(t *testing.T) { + if got := (&ConfigError{Msg: "bad"}).Error(); got != "llm: config: bad" { + t.Errorf("ConfigError = %q", got) + } + ae := &APIError{Provider: "openai", Status: 429, Code: "rate_limit", Message: "slow", Retryable: true} + s := ae.Error() + for _, want := range []string{"openai", "429", "rate_limit", "slow", "retryable"} { + if !strings.Contains(s, want) { + t.Errorf("APIError.Error() = %q, missing %q", s, want) + } + } + minimal := (&APIError{Provider: "x", Status: 500}).Error() + if strings.Contains(minimal, "(") || strings.Contains(minimal, "[retryable]") { + t.Errorf("minimal APIError = %q, want bare status form", minimal) + } + rl := (&RateLimitError{APIError: APIError{Provider: "x", Message: "m"}, Attempts: 2, RetryAfter: 3 * time.Second}).Error() + for _, want := range []string{"x", "2 attempts", "3s", "m"} { + if !strings.Contains(rl, want) { + t.Errorf("RateLimitError = %q, missing %q", rl, want) + } + } + rlBare := (&RateLimitError{APIError: APIError{Provider: "x"}, Attempts: 1}).Error() + if strings.Contains(rlBare, "retry after") { + t.Errorf("RateLimitError without hint = %q", rlBare) + } + if got := (&StreamAbortedError{Reason: errors.New("stop")}).Error(); got != "llm: stream aborted by consumer: stop" { + t.Errorf("StreamAbortedError = %q", got) + } + if got := errSSEOversized.Error(); got != "llm: sse: frame exceeds size limit" { + t.Errorf("sseError = %q", got) + } + ca := &consumerAbort{err: errors.New("inner")} + if !errors.As(func() error { return ca }(), new(*consumerAbort)) || ca.Unwrap().Error() != "inner" { + t.Error("consumerAbort unwrap broken") + } +} + +// ── options ────────────────────────────────────────────────────────────── + +func TestSDKOptions(t *testing.T) { + rt := &countingRoundTripper{} + sdk := New( + WithRequestTimeout(7*time.Second), + WithModelCacheTTL(0), + WithTransport(rt), + WithProvider("mygw", + WithFormat(FormatOpenAI), + WithBaseURL("http://127.0.0.1:1"), + WithAPIKey("k"), + WithQuirks(Quirks{ReasoningEffort: true}), + WithEnvKeys("A_KEY", "B_KEY"), + ), + ) + cc, err := sdk.Chat("mygw", "m") + if err != nil { + t.Fatal(err) + } + if cc.RequestTimeout() != 7*time.Second { + t.Errorf("RequestTimeout = %v, want 7s from WithRequestTimeout", cc.RequestTimeout()) + } + cfg := sdk.Providers()[0].Config() + if !cfg.Quirks.ReasoningEffort { + t.Error("WithQuirks not applied") + } + if len(cfg.EnvKeys) != 2 || cfg.EnvKeys[0] != "A_KEY" { + t.Errorf("WithEnvKeys = %v", cfg.EnvKeys) + } + // WithTransport: a chat call must flow through the replaced transport. + // A short context keeps the (retryable) transport failure cheap. + ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond) + defer cancel() + _, srvErr := cc.Call(ctx, &ChatRequest{Messages: []Message{{Role: RoleUser, Content: "hi"}}}) + if srvErr == nil { + t.Fatal("expected transport-level failure from unreachable base URL") + } + if rt.calls == 0 { + t.Error("WithTransport ignored: request bypassed the replaced transport") + } + // nil guards must not panic. + New(WithTransport(nil), WithRequestTimeout(0), WithEnv(nil)) +} + +type countingRoundTripper struct{ calls int } + +func (r *countingRoundTripper) RoundTrip(*http.Request) (*http.Response, error) { + r.calls++ + return nil, errors.New("unreachable by design") +} + +// ── provider config validation + env resolution ────────────────────────── + +func TestValidateProviderConfigTable(t *testing.T) { + cases := []struct { + name string + cfg ProviderConfig + wantErr string + }{ + {"ok", ProviderConfig{ID: "x", Format: FormatOpenAI, BaseURL: "https://api.example.com"}, ""}, + {"empty id", ProviderConfig{Format: FormatOpenAI, BaseURL: "https://x"}, "ID is empty"}, + {"space id", ProviderConfig{ID: "a b", Format: FormatOpenAI, BaseURL: "https://x"}, "whitespace"}, + {"bad format", ProviderConfig{ID: "x", Format: Format("nope"), BaseURL: "https://x"}, "unknown format"}, + {"empty url", ProviderConfig{ID: "x", Format: FormatOpenAI}, "empty base URL"}, + {"bad scheme", ProviderConfig{ID: "x", Format: FormatOpenAI, BaseURL: "ftp://x"}, "http://"}, + } + for _, tc := range cases { + err := validateProviderConfig(tc.cfg) + if tc.wantErr == "" { + if err != nil { + t.Errorf("%s: %v", tc.name, err) + } + continue + } + if err == nil || !strings.Contains(err.Error(), tc.wantErr) { + t.Errorf("%s: err = %v, want containing %q", tc.name, err, tc.wantErr) + } + } +} + +func TestResolveEnvProvider(t *testing.T) { + base := ProviderConfig{ID: "zai", Format: FormatOpenAI, BaseURL: "https://api.z.ai/api/paas/v4", EnvKeys: []string{"ZAI_API_KEY", "ALIAS_KEY"}} + + if _, ok := resolveEnvProvider(base, nil); ok { + t.Error("nil lookup must not authenticate") + } + if _, ok := resolveEnvProvider(base, func(string) (string, bool) { return "", false }); ok { + t.Error("missing keys must not authenticate") + } + // Primary beats alias; whitespace is trimmed. + cfg, ok := resolveEnvProvider(base, func(k string) (string, bool) { + if k == "ZAI_API_KEY" { + return " primary ", true + } + return "alias", true + }) + if !ok || cfg.APIKey != "primary" { + t.Errorf("primary = %q ok=%v", cfg.APIKey, ok) + } + // Alias wins when primary absent; _BASE_URL override applies. + cfg, ok = resolveEnvProvider(base, func(k string) (string, bool) { + switch k { + case "ALIAS_KEY": + return "alias-key", true + case "ZAI_BASE_URL": + return "https://custom.endpoint/v4", true + } + return "", false + }) + if !ok || cfg.APIKey != "alias-key" || cfg.BaseURL != "https://custom.endpoint/v4" { + t.Errorf("alias/baseURL = %q %q ok=%v", cfg.APIKey, cfg.BaseURL, ok) + } + // Blank-string values are ignored. + if _, ok = resolveEnvProvider(base, func(k string) (string, bool) { return " ", true }); ok { + t.Error("whitespace-only key must not authenticate") + } + if envBaseURLKey("zai") != "ZAI_BASE_URL" { + t.Errorf("envBaseURLKey = %q", envBaseURLKey("zai")) + } +} + +// ── ProviderConfig rendering ───────────────────────────────────────────── + +func TestProviderConfigString(t *testing.T) { + withKey := ProviderConfig{ID: "openai", Format: FormatOpenAI, BaseURL: "https://api.openai.com/v1", APIKey: "secret"} + if s := withKey.String(); !strings.Contains(s, "Authenticated:true") || strings.Contains(s, "secret") { + t.Errorf("String() = %q (must flag auth without leaking the key)", s) + } + without := ProviderConfig{ID: "openai", Format: FormatOpenAI, BaseURL: "https://api.openai.com/v1"} + if s := without.String(); !strings.Contains(s, "Authenticated:false") { + t.Errorf("String() = %q", s) + } +} + +// ── model listing orchestration ────────────────────────────────────────── + +func TestListModelsDispatchAndRetries(t *testing.T) { + old := backoffUnit + backoffUnit = time.Millisecond + t.Cleanup(func() { backoffUnit = old }) + + // Retryable failures exhaust the 3-attempt budget and wrap the cause. + var n int + srv := httptestNewServer(func(w http.ResponseWriter, r *http.Request) { + n++ + w.WriteHeader(500) + }) + if _, err := newListModels(srv.URL, FormatOpenAI); err == nil || !strings.Contains(err.Error(), "list models failed") { + t.Fatalf("err = %v, want exhausted-listing wrap", err) + } + if n != 3 { + t.Errorf("attempts = %d, want 3", n) + } + // Non-retryable failures return immediately. + n = 0 + srv2 := httptestNewServer(func(w http.ResponseWriter, r *http.Request) { + n++ + w.WriteHeader(401) + }) + if _, err := newListModels(srv2.URL, FormatOpenAI); err == nil { + t.Fatal("expected 401 to surface") + } + if n != 1 { + t.Errorf("attempts after 401 = %d, want 1 (no retry on definitive error)", n) + } + // Gemini dispatch arm. + srv3 := httptestNewServer(func(w http.ResponseWriter, r *http.Request) { + w.Write([]byte(`{"models":[{"name":"models/gem","displayName":"Gem","inputTokenLimit":10,"outputTokenLimit":5,"supportedGenerationMethods":["generateContent"]}]}`)) + }) + got, err := newListModels(srv3.URL, FormatGemini) + if err != nil || len(got) != 1 || got[0].ID != "gem" { + t.Fatalf("gemini listing = %+v err %v", got, err) + } + // Anthropic dispatch arm. + srv4 := httptestNewServer(func(w http.ResponseWriter, r *http.Request) { + w.Write([]byte(`{"data":[{"id":"claude-x"}],"has_more":false}`)) + }) + got, err = newListModels(srv4.URL, FormatAnthropic) + if err != nil || len(got) != 1 || got[0].ID != "claude-x" { + t.Fatalf("anthropic listing = %+v err %v", got, err) + } +} + +func TestProviderListModelsCache(t *testing.T) { + var n int + srv := httptestNewServer(func(w http.ResponseWriter, r *http.Request) { + n++ + w.Write([]byte(`{"data":[{"id":"m"}]}`)) + }) + sdk := New(WithProvider("openai", WithBaseURL(srv.URL), WithAPIKey("k"))) + p, _ := sdk.Provider("openai") + + if _, err := p.ListModels(context.Background()); err != nil { + t.Fatal(err) + } + if _, err := p.ListModels(context.Background()); err != nil { + t.Fatal(err) + } + if n != 1 { + t.Fatalf("cached: server hits = %d, want 1", n) + } + if _, err := p.ListModels(context.Background(), ForceRefresh()); err != nil { + t.Fatal(err) + } + if n != 2 { + t.Errorf("ForceRefresh: server hits = %d, want 2", n) + } + + // TTL 0 disables caching. + sdk2 := New(WithModelCacheTTL(0), WithProvider("openai", WithBaseURL(srv.URL), WithAPIKey("k"))) + p2, _ := sdk2.Provider("openai") + _, _ = p2.ListModels(context.Background()) + _, _ = p2.ListModels(context.Background()) + if n != 4 { + t.Errorf("ttl0: server hits = %d, want 4", n) + } + + // Unauthenticated listing is a config error, not a network call. + sdk3 := New(WithEnv(func(string) (string, bool) { return "", false })) + p3, _ := sdk3.Provider("openai") + if _, err := p3.ListModels(context.Background()); err == nil { + t.Fatal("unauthenticated ListModels must fail") + } +} diff --git a/sse.go b/sse.go index 6ea9948..a7add8e 100644 --- a/sse.go +++ b/sse.go @@ -32,12 +32,14 @@ type sseItem struct { err error } -// parseSSEStream reads SSE frames from r and emits items on ch. It closes ch -// when done. A data event is every accumulated "data:" line set terminated -// by a blank line; comment lines (':') and retry fields count as keepalive -// activity. Runs in its own goroutine; terminates when r errors (e.g. the -// caller closes the response body). -func parseSSEStream(r io.Reader, ch chan<- sseItem) { +// parseSSEStream reads SSE frames from r and emits items on ch. It closes +// ch when done. A data event is every accumulated "data:" line set +// terminated by a blank line; comment lines (':') and retry fields count as +// keepalive activity. Runs in its own goroutine; terminates when r errors +// (e.g. the caller closes the response body) or the consumer closes done — +// a pump that exits early (abort, cancel, timeout) must never strand this +// goroutine on a channel send. +func parseSSEStream(r io.Reader, ch chan<- sseItem, done <-chan struct{}) { defer close(ch) br := bufio.NewReaderSize(r, 64*1024) var data []byte @@ -52,7 +54,7 @@ func parseSSEStream(r io.Reader, ch chan<- sseItem) { case hasSSEPrefix(line, "data:"): payload := trimSSEPayload(line[5:]) if len(data)+len(payload) > sseMaxEventSize { - ch <- sseItem{kind: sseErr, err: errSSEOversized} + sendSSE(ch, done, sseItem{kind: sseErr, err: errSSEOversized}) return } if len(data) > 0 { @@ -63,46 +65,75 @@ func parseSSEStream(r io.Reader, ch chan<- sseItem) { } if err != nil { if err == io.EOF && len(data) == 0 { - ch <- sseItem{kind: sseEnd} + sendSSE(ch, done, sseItem{kind: sseEnd}) return } if err == io.EOF { // Trailing event without blank line: deliver it. - ch <- sseItem{kind: sseData, data: data} - ch <- sseItem{kind: sseEnd} + if sendSSE(ch, done, sseItem{kind: sseData, data: data}) { + sendSSE(ch, done, sseItem{kind: sseEnd}) + } return } - ch <- sseItem{kind: sseErr, err: err} + sendSSE(ch, done, sseItem{kind: sseErr, err: err}) return } if len(line) == 0 { // Blank line: event boundary. if len(data) > 0 { - ch <- sseItem{kind: sseData, data: data} + if !sendSSE(ch, done, sseItem{kind: sseData, data: data}) { + return + } data = nil + sawActivity = false } else if sawActivity { - ch <- sseItem{kind: sseKeepalive} + if !sendSSE(ch, done, sseItem{kind: sseKeepalive}) { + return + } sawActivity = false } } } } +// sendSSE delivers one item unless the consumer abandoned the stream +// (done closed); reports whether delivery succeeded. +func sendSSE(ch chan<- sseItem, done <-chan struct{}, it sseItem) bool { + select { + case ch <- it: + return true + case <-done: + return false + } +} + // readSSELine reads one line including the trailing newline; returns the -// line without it (also strips a leading \r for CRLF frames). +// line without it (also strips a trailing \r\n for CRLF frames). Lines +// longer than the bufio buffer are accumulated across ReadSlice calls, +// bounded by sseMaxLineSize. func readSSELine(br *bufio.Reader) ([]byte, error) { - line, err := br.ReadSlice('\n') - if len(line) > sseMaxLineSize { - return nil, errSSEOversized - } - l := len(line) - if l > 0 && line[l-1] == '\n' { - l-- - if l > 0 && line[l-1] == '\r' { + var buf []byte + for { + frag, err := br.ReadSlice('\n') + if len(buf)+len(frag) > sseMaxLineSize { + return nil, errSSEOversized + } + buf = append(buf, frag...) + if err == bufio.ErrBufferFull { + continue // line spans the buffer boundary; keep accumulating + } + if err != nil { + return buf, err // EOF or read error; trailing line lacks '\n' + } + l := len(buf) + if l > 0 && buf[l-1] == '\n' { l-- + if l > 0 && buf[l-1] == '\r' { + l-- + } } + return buf[:l], nil } - return line[:l], err } func hasSSEPrefix(line []byte, prefix string) bool { @@ -129,11 +160,15 @@ var errSSEOversized = &sseError{"frame exceeds size limit"} // (nil error) on EOF without a trailing partial event. func pumpSSE(ctx context.Context, r io.ReadCloser, idle time.Duration, handle func(data []byte) error) error { ch := make(chan sseItem, 8) - go parseSSEStream(r, ch) + done := make(chan struct{}) + defer close(done) // always release the parser, even on abort/timeout + go parseSSEStream(r, ch, done) var timer *time.Timer var timeout <-chan time.Time if idle > 0 { + // No drain-before-Reset dance is needed: since Go 1.23 a Timer's + // channel never delivers stale values after Reset (go.mod: 1.25). timer = time.NewTimer(idle) defer timer.Stop() timeout = timer.C diff --git a/sse_test.go b/sse_test.go new file mode 100644 index 0000000..6224467 --- /dev/null +++ b/sse_test.go @@ -0,0 +1,59 @@ +package llm + +import ( + "context" + "errors" + "io" + "runtime" + "strings" + "testing" + "time" +) + +// Regression: pumpSSE returning early (consumer abort, cancellation, idle +// timeout) must signal the parser goroutine to exit. Without the signal the +// parser blocks forever on a channel send once the 8-slot buffer fills — +// one leaked goroutine per aborted stream. +func TestPumpSSEAbortDoesNotLeakParser(t *testing.T) { + events := strings.Repeat("data: x\n\n", 500) // far more than the 8-slot buffer + before := runtime.NumGoroutine() + err := pumpSSE(context.Background(), io.NopCloser(strings.NewReader(events)), 0, func([]byte) error { + return errors.New("abort now") + }) + if err == nil { + t.Fatal("expected the handler abort to surface") + } + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if runtime.NumGoroutine() <= before { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatalf("goroutine leak: before=%d after=%d (parser stuck on channel send)", before, runtime.NumGoroutine()) +} + +// Regression: a single data line larger than the 64 KiB bufio buffer must +// still be delivered (up to the 1 MiB line cap); anything larger must be +// rejected with errSSEOversized. +func TestParseSSEStreamLongLines(t *testing.T) { + line := strings.Repeat("a", 100*1024) // 100 KiB: over the bufio buffer, under the cap + done := make(chan struct{}) + ch := make(chan sseItem, 2) + go parseSSEStream(strings.NewReader("data: "+line+"\n\n"), ch, done) + it := <-ch + if it.kind == sseErr { + t.Fatalf("100KiB single-line event failed: %v (oversized=%v)", it.err, errors.Is(it.err, errSSEOversized)) + } + if it.kind != sseData || len(it.data) != len(line) { + t.Fatalf("data item = kind %v, %d bytes, want sseData with %d bytes", it.kind, len(it.data), len(line)) + } + + huge := strings.Repeat("b", (1<<20)+1) // 1 MiB + 1: over the line cap + ch2 := make(chan sseItem, 2) + go parseSSEStream(strings.NewReader("data: "+huge+"\n\n"), ch2, done) + it2 := <-ch2 + if it2.kind != sseErr || !errors.Is(it2.err, errSSEOversized) { + t.Fatalf("oversized line: kind %v err %v, want errSSEOversized", it2.kind, it2.err) + } +} diff --git a/stream_failure_test.go b/stream_failure_test.go new file mode 100644 index 0000000..62fb19d --- /dev/null +++ b/stream_failure_test.go @@ -0,0 +1,338 @@ +package llm + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" + "time" +) + +// streamRejected is a heuristic classifier; these cases pin its intended +// shape: a 400 that rejects streaming itself, not any 400 that merely +// mentions the word "stream". +func TestStreamRejectedClassification(t *testing.T) { + cases := []struct { + msg string + want bool + }{ + {"Streaming is not supported for this model.", true}, + {"This model does not support streaming via the API.", true}, + {"streaming unsupported for model", true}, + {"Streaming rejected for this model", true}, + {"Context length exceeded: the streaming limit is 8192 tokens.", false}, + {"Reasoning effort cannot be combined with tools while streaming.", false}, + {"'stream_options.include_usage' is not supported by this model.", false}, // separate learn-once class + {"max_tokens is too large for this model.", false}, + } + for _, tc := range cases { + e := &APIError{Status: http.StatusBadRequest, Message: tc.msg} + if got := streamRejected(e); got != tc.want { + t.Errorf("streamRejected(%q) = %v, want %v", tc.msg, got, tc.want) + } + } + if streamRejected(&APIError{Status: http.StatusInternalServerError, Message: "streaming is not supported"}) { + t.Error("non-400 must never classify as stream rejection") + } +} + +// Contract: partial output followed by a mid-stream failure is returned as +// the partial result plus a wrapped error, and is never retried — a silent +// retry would duplicate user-visible output. +func TestCallStreamPartialFailureNotRetried(t *testing.T) { + var reqs int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + atomic.AddInt32(&reqs, 1) + w.Header().Set("Content-Type", "text/event-stream") + fmt.Fprint(w, "data: {\"choices\":[{\"delta\":{\"content\":\"a\"}}]}\n\n") + fmt.Fprint(w, "data: {\"choices\":[{\"delta\":{\"content\":\"b\"}}]}\n\n") + w.(http.Flusher).Flush() + panic(http.ErrAbortHandler) // hard connection drop mid-stream + })) + defer srv.Close() + + cc := newTestClient(t, ProviderConfig{ID: "openai", Format: FormatOpenAI, BaseURL: srv.URL, APIKey: "k"}, srv) + deltas := 0 + res, err := cc.CallStream(context.Background(), &ChatRequest{Messages: []Message{{Role: RoleUser, Content: "hi"}}}, func(Delta) error { + deltas++ + return nil + }) + if err == nil { + t.Fatal("expected a mid-stream failure error") + } + if res == nil { + t.Fatalf("partial result lost (err = %v)", err) + } + if res.Content != "ab" { + t.Errorf("partial content = %q, want \"ab\"", res.Content) + } + if deltas != 2 { + t.Errorf("deltas delivered = %d, want 2", deltas) + } + if got := atomic.LoadInt32(&reqs); got != 1 { + t.Errorf("requests = %d, want 1 (never retry after partial output)", got) + } + if strings.Contains(err.Error(), "retry exhausted") { + t.Errorf("partial failure misclassified as retry exhaustion: %v", err) + } +} + +// Contract: a stream that stays 429 through the whole retry budget surfaces +// as *RateLimitError with Attempts = maxRetries+1. +func TestCallStreamPersistent429ReturnsRateLimitError(t *testing.T) { + var mu sync.Mutex + var n int + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + n++ + mu.Unlock() + w.Header().Set("Retry-After", "0") + w.WriteHeader(429) + fmt.Fprint(w, `{"error":{"message":"slow down"}}`) + })) + defer srv.Close() + + cc := newTestClient(t, ProviderConfig{ID: "openai", Format: FormatOpenAI, BaseURL: srv.URL, APIKey: "k"}, srv) + _, err := cc.CallStream(context.Background(), &ChatRequest{Messages: []Message{{Role: RoleUser, Content: "hi"}}}, func(Delta) error { + return nil + }) + var rl *RateLimitError + if !errors.As(err, &rl) { + t.Fatalf("err = %v (%T), want *RateLimitError", err, err) + } + if rl.Attempts != maxRetries+1 { + t.Errorf("Attempts = %d, want %d", rl.Attempts, maxRetries+1) + } + mu.Lock() + defer mu.Unlock() + if n != maxRetries+1 { + t.Errorf("server saw %d requests, want %d", n, maxRetries+1) + } +} + +// A 429 followed by a healthy stream must succeed after exactly one retry. +func TestCallStream429ThenSuccess(t *testing.T) { + var mu sync.Mutex + var n int + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + n++ + first := n == 1 + mu.Unlock() + if first { + w.Header().Set("Retry-After", "0") + w.WriteHeader(429) + fmt.Fprint(w, `{"error":{"message":"slow down"}}`) + return + } + sse(w, + `{"choices":[{"delta":{"content":"ok"}}]}`, + `{"choices":[{"delta":{},"finish_reason":"stop"}],"usage":{"prompt_tokens":3,"completion_tokens":2}}`, + "[DONE]", + ) + })) + defer srv.Close() + + cc := newTestClient(t, ProviderConfig{ID: "openai", Format: FormatOpenAI, BaseURL: srv.URL, APIKey: "k"}, srv) + res, err := cc.CallStream(context.Background(), &ChatRequest{Messages: []Message{{Role: RoleUser, Content: "hi"}}}, func(Delta) error { + return nil + }) + if err != nil { + t.Fatalf("CallStream: %v", err) + } + if res.Content != "ok" || res.Usage.PromptTokens != 3 { + t.Errorf("res = %q usage %+v", res.Content, res.Usage) + } + mu.Lock() + defer mu.Unlock() + if n != 2 { + t.Errorf("requests = %d, want 2", n) + } +} + +// The wall-clock deadline after partial output must return the partial +// result wrapped in the deadline error — not a bare context error, and not +// a retry. +func TestCallStreamDeadlineAfterPartialKeepsPartial(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + sse(w, `{"choices":[{"delta":{"content":"a"}}]}`) + time.Sleep(500 * time.Millisecond) // deadline fires long before + })) + defer srv.Close() + + cc := newTestClient(t, ProviderConfig{ID: "openai", Format: FormatOpenAI, BaseURL: srv.URL, APIKey: "k"}, srv) + cc.SetRequestTimeout(100 * time.Millisecond) + res, err := cc.CallStream(context.Background(), &ChatRequest{Messages: []Message{{Role: RoleUser, Content: "hi"}}}, func(Delta) error { + return nil + }) + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("err = %v, want context deadline", err) + } + if res == nil || res.Content != "a" { + t.Errorf("partial result lost after deadline: res=%v err=%v", res, err) + } + if strings.Contains(err.Error(), "retry exhausted") { + t.Errorf("cancellation mislabeled as retry exhaustion: %v", err) + } +} + +// Buffered path: a 429 whose Retry-After outlives the request context must +// still surface as *RateLimitError (the stream path already does this) — +// the caller needs Status/RetryAfter to plan the retry. +func TestCallBuffered429DeadlineKeepsRateLimitError(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Retry-After", "3600") + w.WriteHeader(429) + fmt.Fprint(w, `{"error":{"message":"quota exhausted, retry in an hour"}}`) + })) + defer srv.Close() + + cc := newTestClient(t, ProviderConfig{ID: "openai", Format: FormatOpenAI, BaseURL: srv.URL, APIKey: "k"}, srv) + ctx, cancel := context.WithTimeout(context.Background(), 150*time.Millisecond) + defer cancel() + _, err := cc.Call(ctx, &ChatRequest{Messages: []Message{{Role: RoleUser, Content: "hi"}}}) + var rl *RateLimitError + if !errors.As(err, &rl) { + t.Fatalf("err = %v (%T), want *RateLimitError (a bare ctx error hides the 429)", err, err) + } + if rl.Attempts != 1 { + t.Errorf("Attempts = %d, want 1", rl.Attempts) + } + if rl.RetryAfter != 3600*time.Second { + t.Errorf("RetryAfter = %v, want 1h", rl.RetryAfter) + } + if rl.Status != http.StatusTooManyRequests { + t.Errorf("Status = %d, want 429", rl.Status) + } +} + +// RateLimitError embeds APIError; errors.As must reach Status/Retryable. +func TestRateLimitErrorUnwrapsToAPIError(t *testing.T) { + rl := &RateLimitError{APIError: APIError{Provider: "openai", Status: 429, Message: "slow down"}, Attempts: 3} + var ae *APIError + if !errors.As(rl, &ae) || ae.Status != 429 { + t.Fatalf("errors.As(*APIError) failed on RateLimitError (ae=%v)", ae) + } +} + +// The most common OpenAI error envelope is the nested error object; it must +// be parsed into Message/Code, not left as the raw-body fallback. +func TestHTTPErrorParsesNestedOpenAIError(t *testing.T) { + pc := newProviderClient(ProviderConfig{ID: "openai", Format: FormatOpenAI}, nil, nil) + e := pc.httpError(400, []byte(`{"error":{"message":"nested boom","code":"invalid_request_error"}}`)) + if e.Message != "nested boom" || e.Code != "invalid_request_error" { + t.Fatalf("APIError = %+v, want parsed nested message/code", e) + } +} + +// Unparseable error bodies degrade to the raw preview, capped. +func TestHTTPErrorTruncatesUnparseableBody(t *testing.T) { + pc := newProviderClient(ProviderConfig{ID: "openai", Format: FormatOpenAI}, nil, nil) + e := pc.httpError(500, []byte(strings.Repeat("x", 5000))) + if len(e.Message) != maxErrorBodyPreview { + t.Fatalf("preview = %d bytes, want %d", len(e.Message), maxErrorBodyPreview) + } +} + +// A 200 + SSE headers response that closes with no events and no completion +// signal is a transport failure (retryable), not a silent empty success. +func TestCallStreamPrematureCloseNoEventsIsRetryable(t *testing.T) { + var mu sync.Mutex + var n int + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + n++ + mu.Unlock() + w.Header().Set("Content-Type", "text/event-stream") + w.(http.Flusher).Flush() + // close: no events, no [DONE] + })) + defer srv.Close() + + cc := newTestClient(t, ProviderConfig{ID: "openai", Format: FormatOpenAI, BaseURL: srv.URL, APIKey: "k"}, srv) + res, err := cc.CallStream(context.Background(), &ChatRequest{Messages: []Message{{Role: RoleUser, Content: "hi"}}}, func(Delta) error { + return nil + }) + if err == nil { + t.Fatalf("premature close returned empty success (res=%+v)", res) + } + mu.Lock() + defer mu.Unlock() + if n != maxRetries+1 { + t.Errorf("requests = %d, want %d (retryable: nothing was emitted)", n, maxRetries+1) + } +} + +// Deltas followed by a premature close (no completion signal) must return +// the partial result with a wrapped error, never retried. +func TestCallStreamPrematureCloseAfterDeltasKeepsPartial(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + fmt.Fprint(w, "data: {\"choices\":[{\"delta\":{\"content\":\"a\"}}]}\n\n") + w.(http.Flusher).Flush() + // close: no finish_reason, no [DONE] + })) + defer srv.Close() + + cc := newTestClient(t, ProviderConfig{ID: "openai", Format: FormatOpenAI, BaseURL: srv.URL, APIKey: "k"}, srv) + res, err := cc.CallStream(context.Background(), &ChatRequest{Messages: []Message{{Role: RoleUser, Content: "hi"}}}, func(Delta) error { + return nil + }) + if res == nil || res.Content != "a" { + t.Fatalf("partial result lost after premature close: res=%v err=%v", res, err) + } + if err == nil || !strings.Contains(err.Error(), "before completion") { + t.Fatalf("err = %v, want premature-completion error", err) + } +} + +// The stream path must learn the force-none-effort fallback exactly like +// the buffered path does. +func TestCallStreamLearnsEffortNone(t *testing.T) { + var mu sync.Mutex + var bodies []string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + b, _ := io.ReadAll(r.Body) + mu.Lock() + bodies = append(bodies, string(b)) + mu.Unlock() + if strings.Contains(string(b), `"reasoning_effort"`) && !strings.Contains(string(b), `"reasoning_effort":"none"`) { + w.WriteHeader(400) + fmt.Fprint(w, `{"error":{"message":"reasoning_effort is not supported with tools on this model"}}`) + return + } + w.Header().Set("Content-Type", "text/event-stream") + fmt.Fprint(w, "data: {\"choices\":[{\"delta\":{\"content\":\"ok\"}}]}\n\n") + fmt.Fprint(w, "data: [DONE]\n\n") + })) + defer srv.Close() + + cc := newTestClient(t, ProviderConfig{ID: "openai", Format: FormatOpenAI, BaseURL: srv.URL, APIKey: "k", Quirks: Quirks{ReasoningEffort: true}}, srv) + req := &ChatRequest{ + Messages: []Message{{Role: RoleUser, Content: "hi"}}, + Tools: []ToolDef{{Name: "f", Parameters: json.RawMessage(`{"type":"object"}`)}}, + Thinking: "high", + } + _, err := cc.CallStream(context.Background(), req, func(Delta) error { return nil }) + if err != nil { + t.Fatalf("CallStream: %v", err) + } + mu.Lock() + defer mu.Unlock() + if len(bodies) != 2 { + t.Fatalf("requests = %d, want 2", len(bodies)) + } + if !strings.Contains(bodies[0], `"reasoning_effort":"high"`) { + t.Errorf("first body missing effort: %s", bodies[0]) + } + if !strings.Contains(bodies[1], `"reasoning_effort":"none"`) { + t.Errorf("second body must pin effort none: %s", bodies[1]) + } +} diff --git a/translation_test.go b/translation_test.go new file mode 100644 index 0000000..ecd9a85 --- /dev/null +++ b/translation_test.go @@ -0,0 +1,231 @@ +package llm + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" +) + +// ── Anthropic extended-thinking round-trip ─────────────────────────────── + +// Anthropic requires an assistant turn that ended in tool_use to replay its +// thinking block (with signature) as the first block. The canonical types +// must therefore carry the signature through results AND back into requests. +func TestAnthropicThinkingRoundTripBuffered(t *testing.T) { + body := []byte(`{"content":[{"type":"thinking","thinking":"hmm","signature":"SIG1"},{"type":"text","text":"hi"}],"stop_reason":"tool_use","usage":{"input_tokens":3,"output_tokens":5}}`) + res, err := parseAnthropicResponse(body) + if err != nil { + t.Fatal(err) + } + if res.ReasoningContent != "hmm" { + t.Fatalf("ReasoningContent = %q, want %q", res.ReasoningContent, "hmm") + } + if res.ThinkingSignature != "SIG1" { + t.Fatalf("ThinkingSignature = %q, want %q (signature must survive the round trip)", res.ThinkingSignature, "SIG1") + } + + req := &ChatRequest{Messages: []Message{ + {Role: RoleUser, Content: "q"}, + {Role: RoleAssistant, Content: "hi", ReasoningContent: "hmm", ThinkingSignature: "SIG1", + ToolCalls: []ToolCall{{ID: "tu_1", Name: "f", Arguments: `{}`}}}, + {Role: RoleTool, ToolCallID: "tu_1", Content: "r"}, + }} + b, err := buildAnthropicRequest(req, "m", false) + if err != nil { + t.Fatal(err) + } + var out struct { + Messages []struct { + Role string `json:"role"` + Content []struct { + Type string `json:"type"` + Thinking string `json:"thinking,omitempty"` + Signature string `json:"signature,omitempty"` + } `json:"content"` + } `json:"messages"` + } + if err := json.Unmarshal(b, &out); err != nil { + t.Fatal(err) + } + var asst *struct { + Role string `json:"role"` + Content []struct { + Type string `json:"type"` + Thinking string `json:"thinking,omitempty"` + Signature string `json:"signature,omitempty"` + } `json:"content"` + } + for i := range out.Messages { + if out.Messages[i].Role == "assistant" { + asst = &out.Messages[i] + } + } + if asst == nil || len(asst.Content) == 0 { + t.Fatalf("no assistant message in %s", b) + } + if asst.Content[0].Type != "thinking" { + t.Fatalf("first assistant block = %s, want thinking replayed first (Anthropic contract)", asst.Content[0].Type) + } + if asst.Content[0].Thinking != "hmm" || asst.Content[0].Signature != "SIG1" { + t.Fatalf("thinking block = %+v, want thinking/signature replayed", asst.Content[0]) + } +} + +// Streaming: signature_delta must be captured into the result's +// ThinkingSignature, mirroring the buffered path. +func TestAnthropicStreamSignatureCapture(t *testing.T) { + acc := newStreamAccum() + evs := []string{ + `{"type":"message_start","message":{"usage":{"input_tokens":2}}}`, + `{"type":"content_block_start","index":0,"content_block":{"type":"thinking"}}`, + `{"type":"content_block_delta","index":0,"delta":{"type":"thinking_delta","thinking":"let me"}}`, + `{"type":"content_block_delta","index":0,"delta":{"type":"signature_delta","signature":"SIG9"}}`, + `{"type":"content_block_stop","index":0}`, + `{"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":4}}`, + `{"type":"message_stop"}`, + } + for _, e := range evs { + if _, _, err := mapAnthropicStreamEvent([]byte(e), acc); err != nil { + t.Fatal(err) + } + } + res := acc.result() + if res.ReasoningContent != "let me" { + t.Fatalf("ReasoningContent = %q", res.ReasoningContent) + } + if res.ThinkingSignature != "SIG9" { + t.Fatalf("ThinkingSignature = %q, want %q (signature_delta dropped)", res.ThinkingSignature, "SIG9") + } +} + +// ── role validation ───────────────────────────────────────────────────── + +// A zero-value (or unknown) Role must be rejected loudly at the SDK +// boundary, not silently dropped (Anthropic/Gemini) or reinterpreted as a +// user message (OpenAI). +func TestUnknownRoleRejectedAtBoundary(t *testing.T) { + var reached atomicBoolFlag + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + reached.Store(true) + w.WriteHeader(500) + })) + defer srv.Close() + + cc := newTestClient(t, ProviderConfig{ID: "openai", Format: FormatOpenAI, BaseURL: srv.URL, APIKey: "k"}, srv) + _, err := cc.Call(context.Background(), &ChatRequest{Messages: []Message{ + {Role: Role("wizard"), Content: "hi"}, + }}) + var ce *ConfigError + if !errors.As(err, &ce) || !strings.Contains(err.Error(), "role") { + t.Fatalf("err = %v (%T), want *ConfigError naming the bad role", err, err) + } + if reached.Load() { + t.Error("request reached the network despite invalid role") + } +} + +// atomicBoolFlag is a tiny helper so tests can flag handler entry safely. +type atomicBoolFlag struct { + mu sync.Mutex + v bool +} + +func (b *atomicBoolFlag) Store(v bool) { b.mu.Lock(); b.v = v; b.mu.Unlock() } +func (b *atomicBoolFlag) Load() bool { b.mu.Lock(); defer b.mu.Unlock(); return b.v } + +// ── Gemini fidelity ───────────────────────────────────────────────────── + +// A trailing usage-only chunk (no candidates) must still update usage. +func TestGeminiUsageOnlyChunkRetained(t *testing.T) { + acc := newStreamAccum() + if _, _, err := mapGeminiStreamEvent([]byte(`{"candidates":[{"content":{"parts":[{"text":"a"}]}}]}`), acc); err != nil { + t.Fatal(err) + } + if _, _, err := mapGeminiStreamEvent([]byte(`{"usageMetadata":{"promptTokenCount":7,"candidatesTokenCount":3}}`), acc); err != nil { + t.Fatal(err) + } + if acc.usage.PromptTokens != 7 || acc.usage.CompletionTokens != 3 { + t.Fatalf("usage = %+v, want {7 3 0} (usage-only chunk dropped)", acc.usage) + } +} + +// Gemini functionResponse requires the function NAME; consumers ported from +// the OpenAI format set only ToolCallID, so the name must be recovered from +// the assistant tool_call — and a truly unresolvable name must be a loud +// config error instead of a guaranteed provider 400. +func TestGeminiToolResultNameFallback(t *testing.T) { + req := &ChatRequest{Messages: []Message{ + {Role: RoleUser, Content: "weather?"}, + {Role: RoleAssistant, ToolCalls: []ToolCall{{ID: "call_0", Name: "get_weather", Arguments: `{"city":"SF"}`}}}, + {Role: RoleTool, ToolCallID: "call_0", Content: `{"temp":70}`}, + }} + b, err := buildGeminiRequest(req, "gemini-2.5-pro", false) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(string(b), `"name":"get_weather"`) { + t.Fatalf("functionResponse name unresolved in %s", b) + } + + bad := &ChatRequest{Messages: []Message{ + {Role: RoleTool, ToolCallID: "ghost", Content: "x"}, + }} + if _, err := buildGeminiRequest(bad, "m", false); err == nil { + t.Fatal("unresolvable tool result name must error, not silently 400 at the provider") + } +} + +// ── canonical finish reasons ──────────────────────────────────────────── + +// Provider-specific stop reasons must not leak into the canonical +// FinishReason; unknown values map to "" (unknown), like documented gaps. +func TestUnknownStopReasonsMapToCanonicalEmpty(t *testing.T) { + if got := mapAnthropicStopReason("pause_turn"); got != "" { + t.Errorf(`anthropic "pause_turn" = %q, want ""`, got) + } + if got := mapGeminiFinishReason("OTHER"); got != "" { + t.Errorf(`gemini "OTHER" = %q, want ""`, got) + } + if mapAnthropicStopReason("tool_use") != FinishToolCalls { + t.Error("anthropic tool_use regression") + } + if mapGeminiFinishReason("STOP") != FinishStop { + t.Error("gemini STOP regression") + } +} + +// ── SetRequestTimeout race safety ─────────────────────────────────────── + +// SetRequestTimeout concurrent with RequestTimeout must be race-free (the +// buffered client is swapped atomically, not field-assigned). +// +// NOTE: this deliberately uses a deadline loop instead of time.After + +// select-default spinning — that pattern was observed to never observe its +// timer on this machine (Go 1.26.5 darwin/arm64, reproducible in a +// standalone program), which would hang the test unrelated to the SDK. +func TestSetRequestTimeoutRacesAreSafe(t *testing.T) { + pc := newProviderClient(ProviderConfig{ID: "x", Format: FormatOpenAI, BaseURL: "http://127.0.0.1:1"}, nil, nil) + cc := &ChatClient{pc: pc, model: "m", parent: &Provider{cfg: pc.cfg, sdk: New()}} + deadline := time.Now().Add(80 * time.Millisecond) + var wg sync.WaitGroup + wg.Add(2) + go func() { + defer wg.Done() + for time.Now().Before(deadline) { + cc.SetRequestTimeout(30 * time.Millisecond) + } + }() + go func() { + defer wg.Done() + for time.Now().Before(deadline) { + _ = cc.RequestTimeout() + } + }() + wg.Wait() +}