package llm import ( "bytes" "context" "encoding/json" "fmt" "io" "log/slog" "math/rand" "net/http" "strconv" "time" "textmachine/backend/internal/obs" ) // httpllm.go is the shared OpenAI-compatible Chat Completions transport: one // HTTP+retry implementation reused by every OpenAI-compatible adapter // (DeepSeek, GLM, Kimi, xAI, ru-агрегаторы, llama-server). Порт vojo // httpllm.go с правками Фазы 0 (03-implementation-notes §2 п.2): // // - все таймауты/ретраи — ПАРАМЕТРЫ клиента из профиля models.yaml, не // константы: захардкоженный 60-секундный per-attempt дедлайн донора убивал // бы реальные чанки 63–278 с и трижды бесплатно их ретраил; // - Retry-After на 429 читается и уважается (донор капил backoff на 8 с — // молоток по rate-limit'ам массового прогона глав); // - тело ответа читается через LimitReader (у донора кап был только в // gemini-native адаптере). // RetryProfile bounds one client's retry loop. Zero fields fall back to // defaults; the profile comes from models.yaml per provider (per-role // overrides — Фаза 1). type RetryProfile struct { AttemptTimeout time.Duration // deadline for ONE HTTP attempt MaxAttempts int BackoffBase time.Duration // first backoff; doubles per attempt BackoffCap time.Duration } func (p RetryProfile) withDefaults() RetryProfile { if p.AttemptTimeout <= 0 { // Полигон намерил 63–278 с на чанк локальной моделью; облачные быстрее, // но дефолт обязан переживать длинный чанк. Провайдер-специфика — в // models.yaml, это только страховочный потолок. p.AttemptTimeout = 300 * time.Second } if p.MaxAttempts <= 0 { p.MaxAttempts = 3 } if p.BackoffBase <= 0 { p.BackoffBase = 500 * time.Millisecond } if p.BackoffCap <= 0 { p.BackoffCap = 30 * time.Second } return p } // maxResponseBytes caps one completion body read. A translated chapter chunk // is ~10–50 KiB; 16 MiB leaves two orders of magnitude of headroom while // keeping a misbehaving endpoint from exhausting memory. const maxResponseBytes = 16 << 20 // openAIClient performs OpenAI-compatible /chat/completions calls with retry. type openAIClient struct { name string // provider label for logs/errors ("deepseek", "glm", "local") base string key string http *http.Client profile RetryProfile headers map[string]string // extra static headers (provider-specific), may be nil log *slog.Logger } // newOpenAIClient builds the shared transport. httpc may be nil (default // client); the local provider passes an explicit no-proxy client (прокси-грабли // стенда: env-прокси перехватывает не-loopback локальные адреса вроде 172.x). func newOpenAIClient(name, base, key string, profile RetryProfile, headers map[string]string, httpc *http.Client, logger *slog.Logger) *openAIClient { if httpc == nil { httpc = &http.Client{} } return &openAIClient{ name: name, base: base, key: key, http: httpc, profile: profile.withDefaults(), headers: headers, log: logger, } } // --- OpenAI-compatible wire types ------------------------------------------------- type openAIMessage struct { Role string `json:"role"` Content string `json:"content"` } type openAIRequest struct { Model string `json:"model"` Messages []openAIMessage `json:"messages"` MaxTokens int `json:"max_tokens"` Temperature float64 `json:"temperature,omitempty"` Stream bool `json:"stream"` // Optional; omitempty keeps the plain body minimal. ReasoningEffort string `json:"reasoning_effort,omitempty"` ResponseFormat any `json:"response_format,omitempty"` // extra carries per-model provider-specific fields from models.yaml // (например GLM {"thinking":{"type":"disabled"}}). Merged into the JSON // body at marshal time — the neutral request stays vendor-free. extra map[string]any } // MarshalJSON merges extra into the standard body. Standard fields win on // key collision (extra is config-supplied tuning, not an override channel). func (r openAIRequest) MarshalJSON() ([]byte, error) { type plain openAIRequest base, err := json.Marshal(plain(r)) if err != nil { return nil, err } if len(r.extra) == 0 { return base, nil } var m map[string]any if err := json.Unmarshal(base, &m); err != nil { return nil, err } for k, v := range r.extra { if _, exists := m[k]; !exists { m[k] = v } } return json.Marshal(m) } type openAIUsage struct { PromptTokens int `json:"prompt_tokens"` CompletionTokens int `json:"completion_tokens"` PromptTokensDetails struct { CachedTokens int `json:"cached_tokens"` } `json:"prompt_tokens_details"` // xAI reports reasoning tokens here SEPARATELY from completion_tokens and // bills them at the output rate; OpenAI-spec providers count them inside // completion_tokens. The adapter decides which semantics apply. CompletionTokensDetails struct { ReasoningTokens int `json:"reasoning_tokens"` } `json:"completion_tokens_details"` // DeepSeek historically reports cache usage in its own top-level fields // instead of prompt_tokens_details. Parse both so the Phase-0 cache // experiment can't show a false zero (03-implementation-notes §3.5). PromptCacheHitTokens int `json:"prompt_cache_hit_tokens"` PromptCacheMissTokens int `json:"prompt_cache_miss_tokens"` } // cacheRead returns the cache-hit token count whichever field the provider // used. func (u openAIUsage) cacheRead() int { if u.PromptTokensDetails.CachedTokens > 0 { return u.PromptTokensDetails.CachedTokens } return u.PromptCacheHitTokens } type openAIResponse struct { ID string `json:"id"` Model string `json:"model"` Choices []struct { Message struct { Content string `json:"content"` } `json:"message"` FinishReason string `json:"finish_reason"` } `json:"choices"` Usage openAIUsage `json:"usage"` } func (r *openAIResponse) Text() string { if len(r.Choices) == 0 { return "" } return r.Choices[0].Message.Content } func (r *openAIResponse) FinishReason() string { if len(r.Choices) == 0 { return "" } // OpenAI-compat wire values map 1:1 onto the neutral constants. return r.Choices[0].FinishReason } // complete calls Chat Completions with retry on transient failures (429 / 5xx / // network, exponential backoff + jitter, Retry-After honoured). Non-retryable // 4xx fail immediately. On exhaustion the caller releases the reservation, so // a transient failure is never silently swallowed. func (c *openAIClient) complete(ctx context.Context, reqBody openAIRequest) (*openAIResponse, error) { payload, err := json.Marshal(reqBody) if err != nil { return nil, err } var lastErr error for attempt := 0; attempt < c.profile.MaxAttempts; attempt++ { if attempt > 0 { backoff := c.profile.BackoffBase << uint(attempt-1) if backoff > c.profile.BackoffCap { backoff = c.profile.BackoffCap } // A server-provided Retry-After overrides our schedule: the provider // knows its rate-limit window better than our exponential guess. if ra := retryAfterOf(lastErr); ra > 0 { backoff = ra } backoff += time.Duration(rand.Intn(250)) * time.Millisecond select { case <-ctx.Done(): return nil, ctx.Err() case <-time.After(backoff): } } resp, retryable, err := c.attempt(ctx, payload) if err == nil { return resp, nil } lastErr = err if ctx.Err() != nil { return nil, ctx.Err() } if !retryable { return nil, err } if c.log != nil { c.log.WarnContext(ctx, c.name+" attempt failed, will retry", "attempt", attempt+1, "max", c.profile.MaxAttempts, "err", err) } } return nil, fmt.Errorf("%s: exhausted %d attempts: %w", c.name, c.profile.MaxAttempts, lastErr) } // attempt performs one HTTP call. Returns retryable=true for 429/5xx and // network errors, false for other non-2xx (terminal 4xx). The per-attempt // deadline bounds a single hung connection; the overall per-request deadline // (set by the caller via ctx) bounds the whole retry loop. func (c *openAIClient) attempt(ctx context.Context, payload []byte) (*openAIResponse, bool, error) { attemptCtx, cancel := context.WithTimeout(ctx, c.profile.AttemptTimeout) defer cancel() req, err := http.NewRequestWithContext(attemptCtx, http.MethodPost, c.base+"/chat/completions", bytes.NewReader(payload)) if err != nil { return nil, false, err } req.Header.Set("Content-Type", "application/json") // A local backend (ollama / llama-server) usually runs without auth; an // empty key means "no Authorization header", not "Bearer " with an empty token. if c.key != "" { req.Header.Set("Authorization", "Bearer "+c.key) } for k, v := range c.headers { req.Header.Set(k, v) } resp, err := c.http.Do(req) if err != nil { // Network error / timeout — retryable (unless the parent ctx is done). return nil, ctx.Err() == nil, err } defer resp.Body.Close() data, _ := io.ReadAll(io.LimitReader(resp.Body, maxResponseBytes)) obs.LogLLMExchange(ctx, c.log, c.name, payload, resp.StatusCode, data) if resp.StatusCode == http.StatusTooManyRequests || resp.StatusCode >= 500 { return nil, true, &HTTPStatusError{Provider: c.name, Status: resp.StatusCode, Body: snippet(data), RetryAfter: parseRetryAfter(resp.Header.Get("Retry-After"))} } if resp.StatusCode < 200 || resp.StatusCode >= 300 { return nil, false, &HTTPStatusError{Provider: c.name, Status: resp.StatusCode, Body: snippet(data)} } var out openAIResponse if err := json.Unmarshal(data, &out); err != nil { return nil, false, fmt.Errorf("%s decode: %w", c.name, err) } // A 2xx is a billed call even when the model returns empty content // (content filter, finish_reason=length with no text). Return it as a // success so the caller settles the real cost via the ledger instead of // releasing the reservation and losing the spend — which would let empty // replies bypass the book/day ceilings. Gates deal with the empty text. return &out, false, nil } // HTTPStatusError is a non-2xx completion failure, carrying the status code as // a typed field so callers (the failover decorator) can classify it // structurally — a terminal 4xx is a config/request error that must fail loud, // not be masked by a fallback — instead of parsing the message. type HTTPStatusError struct { Provider string Status int Body string RetryAfter time.Duration // from the Retry-After header; 0 = absent } func (e *HTTPStatusError) Error() string { return fmt.Sprintf("%s http %d: %s", e.Provider, e.Status, e.Body) } func retryAfterOf(err error) time.Duration { if se, ok := err.(*HTTPStatusError); ok { return se.RetryAfter } return 0 } func parseRetryAfter(v string) time.Duration { if v == "" { return 0 } if secs, err := strconv.Atoi(v); err == nil && secs > 0 { return time.Duration(secs) * time.Second } if t, err := http.ParseTime(v); err == nil { if d := time.Until(t); d > 0 { return d } } return 0 } func snippet(b []byte) string { const max = 300 if len(b) > max { return string(b[:max]) + "…" } return string(b) }