From 0a538a9e0b42343242a87941f843fa23070b346f Mon Sep 17 00:00:00 2001 From: heaven Date: Fri, 3 Jul 2026 00:04:18 +0300 Subject: [PATCH] =?UTF-8?q?feat(ai-bot):=20local=20LLM=20voice=20backend?= =?UTF-8?q?=20=E2=80=94=20OpenAI-compatible=20adapter,=20health-probed=20f?= =?UTF-8?q?ailover=20to=20Grok,=20billing=20by=20actual=20serving=20model?= =?UTF-8?q?=20($0=20local),=20LOCAL=5FLLM=5F*=20config?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- apps/ai-bot/README.md | 30 +++ apps/ai-bot/bot.go | 8 + apps/ai-bot/cascade.go | 38 +++- apps/ai-bot/cascade_test.go | 76 ++++++- apps/ai-bot/config.go | 86 +++++++- apps/ai-bot/config_test.go | 61 +++++- apps/ai-bot/failover.go | 234 ++++++++++++++++++++ apps/ai-bot/failover_test.go | 331 +++++++++++++++++++++++++++++ apps/ai-bot/httpllm.go | 26 ++- apps/ai-bot/httpllm_test.go | 24 +++ apps/ai-bot/llm.go | 11 +- apps/ai-bot/provider_gemini.go | 1 + apps/ai-bot/provider_local.go | 90 ++++++++ apps/ai-bot/provider_local_test.go | 103 +++++++++ apps/ai-bot/provider_xai.go | 1 + apps/ai-bot/router.go | 2 +- docs/ai/ai-bot.md | 28 ++- 17 files changed, 1129 insertions(+), 21 deletions(-) create mode 100644 apps/ai-bot/failover.go create mode 100644 apps/ai-bot/failover_test.go create mode 100644 apps/ai-bot/provider_local.go create mode 100644 apps/ai-bot/provider_local_test.go diff --git a/apps/ai-bot/README.md b/apps/ai-bot/README.md index 05849356..11202727 100644 --- a/apps/ai-bot/README.md +++ b/apps/ai-bot/README.md @@ -186,6 +186,36 @@ offline-eval gate (misroute < 2-3% AND measured saving > the second provider's c | `GEMINI_MODEL` | `gemini-2.5-flash-lite` | cheap model for trivial/classifier | | `GEMINI_BASE_URL` | `…/v1beta/openai` | OpenAI-compat endpoint (native grounding endpoint derived from it) | +### Local LLM backend (voice failover, **default OFF**) + +Phase 1 of `docs/plans/local_llm_backend.md`: while the self-hosted backend is reachable, the +bot's **voice** (every `b.llm` call — grok_direct, reasoning, web/project synthesis) is served by +it **for free**; when it isn't, requests transparently fall back to cloud Grok ([failover.go](failover.go), +active `GET /models` probe + circuit breaker). Works with **any OpenAI-compatible server** +(ollama, llama.cpp `llama-server`, vLLM, LM Studio) — the adapter ([provider_local.go](provider_local.go)) +speaks only the standard `/chat/completions` + `/models` surface, selected purely by env. The +classifier (`b.gemini`) and web fetch (`b.web`) stay cloud until their own phases. A local answer +books **$0** in the ledger (billing follows `LLMResponse.Model` — the model that actually served); +`request_log.models.final` records which backend answered, so the local-vs-cloud share and quality +(via reaction feedback) are measurable. `XAI_API_KEY` stays **required** — the cloud is the fallback leg. + +| Env | Default | Meaning | +|---|---|---| +| `LOCAL_LLM_ENABLED` | false | wrap `b.llm` in the local-first failover (off = byte-identical to today) | +| `LOCAL_LLM_BASE_URL` | — | OpenAI-compatible `/v1` root, e.g. `http://172.18.0.1:11434/v1` (the SSH-tunnel relay on the docker bridge). **Required when enabled** (fail-fast). | +| `LOCAL_LLM_MODEL` | `huihui_ai/qwen3-abliterated:8b` | the local server's model tag. Must **differ** from the cloud model names (its $0 price entry would zero cloud billing — refuses to boot). | +| `LOCAL_LLM_API_KEY` / `_FILE` | — | optional; empty sends **no** Authorization header | +| `LOCAL_LLM_TEMP` | 0.7 | temperature the local adapter sends (Qwen3 non-thinking tuning), **overriding** `XAI_TEMPERATURE` — a request temp beats the Modelfile, so passing Grok's 0.6 through would mistune the local model. The other Qwen3 samplers (top_k/min_p/repeat_penalty) can't cross ollama's `/v1` and live in the Modelfile at home. | +| `LOCAL_LLM_MAX_OUTPUT_TOKENS` | 1024 | replaces `MAX_OUTPUT_TOKENS` on the local leg (0 = inherit). ollama's cap covers **thinking + answer together** (xAI bills thinking on top), so a thinking route (`reasoning_effort` low/high — the web/project synthesis and reason routes) on the cloud-sized 320 would burn the whole budget on reasoning and return empty content. Local tokens are free. | +| `LOCAL_LLM_TIMEOUT_SECONDS` | 45 | the local leg's own request deadline — a half-alive tunnel fails fast enough to leave `REQUEST_BUDGET_SECONDS` room for the cloud fallback | +| `LOCAL_LLM_HEALTH_INTERVAL_SECONDS` | 15 | background health-probe period. Recovery is probe-driven only (anti-flapping); a request-path trip needs 2 consecutive probe passes to re-close. | + +`reasoning_effort` passes through unchanged: ollama's `/v1` maps it onto Qwen3 thinking (`none` = +off — the casual `GROK_REASONING_EFFORT_DIRECT=none` route; `low`/`high` = on; verified live), so +the cascade's existing per-route efforts drive local thinking with **no new knob**. A local +**terminal 4xx** (wrong model tag) fails loud instead of falling back — a config error must not be +masked by paid cloud answers; 429/5xx/timeouts/transport errors fall back within the same request. + ## One-time setup (appservice registration) Like the mautrix bridges (e.g. telegram), the bot **generates its own diff --git a/apps/ai-bot/bot.go b/apps/ai-bot/bot.go index 3b86be41..afd09b60 100644 --- a/apps/ai-bot/bot.go +++ b/apps/ai-bot/bot.go @@ -80,6 +80,14 @@ type Bot struct { func NewBot(ctx context.Context, cfg *Config, logger *slog.Logger) (*Bot, error) { mx := NewMatrixClient(cfg.HomeserverURL, cfg.ASToken, cfg.BotMXID) llm := NewXAIClient(cfg.XAIBaseURL, cfg.XAIAPIKey, logger) + // Local-first voice (Phase 1, docs/plans/local_llm_backend.md): wrap the cloud + // client in the failover decorator — the home backend answers while its tunnel is + // healthy, Grok otherwise. Only b.llm changes; the classifier (b.gemini) and web + // layer (b.web) stay cloud until their own phases. The prober lives on ctx (the + // process signal context), same lifetime as the bot. + if cfg.LocalLLMEnabled { + llm = NewFailoverLLMClient(ctx, NewLocalLLMClient(cfg, logger), llm, cfg, logger) + } st, err := OpenStore(cfg.DatabaseURL) if err != nil { diff --git a/apps/ai-bot/cascade.go b/apps/ai-bot/cascade.go index ec148bf2..0266fdb7 100644 --- a/apps/ai-bot/cascade.go +++ b/apps/ai-bot/cascade.go @@ -209,12 +209,26 @@ func (b *Bot) genGrokDirect(ctx context.Context, msgs []Message, convID string, if err != nil { return err } - res.route, res.finalModel = routeGrokDirect, b.cfg.XAIModel + model := servedModel(b.cfg.XAIModel, resp) + res.route, res.finalModel = routeGrokDirect, model res.text, res.usage, res.providerID = resp.Text, resp.Usage, resp.ProviderRequestID - res.cost.Token += computeUSD(b.cfg.XAIModel, resp.Usage, b.cfg) + res.cost.Token += computeUSD(model, resp.Usage, b.cfg) return nil } +// servedModel is the model a completion must be billed and telemetered as: the one +// that ACTUALLY answered when the adapter reports it (resp.Model — the local/cloud +// failover can answer with a different backend than the cascade requested, and the +// local one is priced $0), else the requested model. Without this, a free local +// answer books at the requested Grok price and the ledger/ceiling lie (§5.4 of the +// local-backend plan). +func servedModel(requested string, resp *LLMResponse) string { + if resp.Model != "" { + return resp.Model + } + return requested +} + // genTrivial answers a trivial message with the cheap Gemini model. An empty reply is // treated as a failure so the caller degrades to Grok rather than sending nothing. func (b *Bot) genTrivial(ctx context.Context, msgs []Message, res *genResult) error { @@ -232,9 +246,10 @@ func (b *Bot) genTrivial(ctx context.Context, msgs []Message, res *genResult) er if strings.TrimSpace(resp.Text) == "" { return fmt.Errorf("trivial: empty Gemini reply") } - res.route, res.finalModel = routeTrivial, b.cfg.GeminiModel + model := servedModel(b.cfg.GeminiModel, resp) + res.route, res.finalModel = routeTrivial, model res.text, res.usage, res.providerID = resp.Text, resp.Usage, resp.ProviderRequestID - res.cost.Token += computeUSD(b.cfg.GeminiModel, resp.Usage, b.cfg) + res.cost.Token += computeUSD(model, resp.Usage, b.cfg) return nil } @@ -257,9 +272,10 @@ func (b *Bot) genReason(ctx context.Context, msgs []Message, convID string, res if strings.TrimSpace(resp.Text) == "" { return fmt.Errorf("reason: empty reply") } - res.route, res.finalModel = routeReason, b.cfg.ReasoningModel + model := servedModel(b.cfg.ReasoningModel, resp) + res.route, res.finalModel = routeReason, model res.text, res.usage, res.providerID = resp.Text, resp.Usage, resp.ProviderRequestID - res.cost.Token += computeUSD(b.cfg.ReasoningModel, resp.Usage, b.cfg) + res.cost.Token += computeUSD(model, resp.Usage, b.cfg) return nil } @@ -288,9 +304,10 @@ func (b *Bot) genProjectThenGrok(ctx context.Context, msgs []Message, convID str if strings.TrimSpace(resp.Text) == "" { return fmt.Errorf("project: empty reply") } - res.route, res.finalModel = routeProject, b.cfg.XAIModel + model := servedModel(b.cfg.XAIModel, resp) + res.route, res.finalModel = routeProject, model res.text, res.usage, res.providerID = resp.Text, resp.Usage, resp.ProviderRequestID - res.cost.Token += computeUSD(b.cfg.XAIModel, resp.Usage, b.cfg) + res.cost.Token += computeUSD(model, resp.Usage, b.cfg) return nil } @@ -370,7 +387,8 @@ func (b *Bot) genWebThenGrok(ctx context.Context, body string, isDM bool, msgs [ if strings.TrimSpace(resp.Text) == "" { return fmt.Errorf("web synth: empty reply") } - res.route, res.finalModel = routeWebThenGrok, b.cfg.XAIModel + model := servedModel(b.cfg.XAIModel, resp) + res.route, res.finalModel = routeWebThenGrok, model res.text, res.providerID = resp.Text, resp.ProviderRequestID // Report BOTH calls' tokens so the analytics token totals match the two-call route. res.usage = Usage{ @@ -379,7 +397,7 @@ func (b *Bot) genWebThenGrok(ctx context.Context, body string, isDM bool, msgs [ CompletionTokens: resp.Usage.CompletionTokens + webUsage.CompletionTokens, ReasoningTokens: resp.Usage.ReasoningTokens + webUsage.ReasoningTokens, } - res.cost.Token += computeUSD(b.cfg.XAIModel, resp.Usage, b.cfg) + res.cost.Token += computeUSD(model, resp.Usage, b.cfg) return nil } diff --git a/apps/ai-bot/cascade_test.go b/apps/ai-bot/cascade_test.go index 3901945e..ff4ba413 100644 --- a/apps/ai-bot/cascade_test.go +++ b/apps/ai-bot/cascade_test.go @@ -11,10 +11,13 @@ import ( func discardLog() *slog.Logger { return slog.New(slog.NewTextHandler(io.Discard, nil)) } -// fakeLLM is a scriptable LLMClient for dispatch/degrade tests. +// fakeLLM is a scriptable LLMClient for dispatch/degrade tests. model, when set, is +// reported as LLMResponse.Model (the responder identity a failover leg would fill); +// empty keeps the pre-Model behaviour (billing falls back to the requested model). type fakeLLM struct { text string usage Usage + model string err error calls int lastReq LLMRequest @@ -26,7 +29,7 @@ func (f *fakeLLM) Complete(_ context.Context, req LLMRequest) (*LLMResponse, err if f.err != nil { return nil, f.err } - return &LLMResponse{Text: f.text, Usage: f.usage, ProviderRequestID: "fake"}, nil + return &LLMResponse{Text: f.text, Usage: f.usage, Model: f.model, ProviderRequestID: "fake"}, nil } type fakeWeb struct { @@ -782,6 +785,75 @@ func TestReserveEstimateProjectNoBump(t *testing.T) { } } +// TestServedModel: billing follows the responder when the adapter reports it, else +// the requested model (fakes / adapters predating LLMResponse.Model). +func TestServedModel(t *testing.T) { + if got := servedModel("grok-x", &LLMResponse{}); got != "grok-x" { + t.Fatalf("empty resp.Model → %q, want the requested model", got) + } + if got := servedModel("grok-x", &LLMResponse{Model: "qwen-local"}); got != "qwen-local" { + t.Fatalf("resp.Model set → %q, want the responder", got) + } +} + +// TestGenerateBillsByServedModel is the §5.4 money invariant: when the failover seam +// answers with the free local model, the cascade books $0 (its price entry) and the +// telemetry final model is the responder — NOT the requested Grok priced at Grok +// rates. And the fallback leg answering keeps booking real cloud money. +func TestGenerateBillsByServedModel(t *testing.T) { + usage := Usage{PromptTokens: 1000, CompletionTokens: 500} + + local := &fakeLLM{text: "local answer", usage: usage, model: "qwen-local"} + cfg := cascadeCfg() + cfg.Prices["qwen-local"] = ModelPrice{} // the $0 entry LoadConfig registers + b := &Bot{cfg: &cfg, llm: local, log: discardLog()} + res, err := b.generate(context.Background(), "привет", msgs("привет"), "", true) + if err != nil { + t.Fatalf("generate: %v", err) + } + if res.cost.Token != 0 { + t.Fatalf("local answer booked $%v, want $0 (the easiest-to-miss bug)", res.cost.Token) + } + if res.finalModel != "qwen-local" { + t.Fatalf("finalModel = %q, want the responder qwen-local (telemetry backend attribution)", res.finalModel) + } + + cloud := &fakeLLM{text: "cloud answer", usage: usage, model: "grok-x"} + cfg2 := cascadeCfg() + b2 := &Bot{cfg: &cfg2, llm: cloud, log: discardLog()} + res2, err := b2.generate(context.Background(), "привет", msgs("привет"), "", true) + if err != nil { + t.Fatalf("generate: %v", err) + } + if res2.cost.Token <= 0 { + t.Fatalf("cloud answer booked $%v, want real Grok money", res2.cost.Token) + } + if res2.finalModel != "grok-x" { + t.Fatalf("finalModel = %q, want grok-x", res2.finalModel) + } +} + +// TestGenerateReasonRouteBillsByServedModel: the $0 invariant holds on the reasoning +// route too (a different computeUSD call-site — the audit found four). +func TestGenerateReasonRouteBillsByServedModel(t *testing.T) { + local := &fakeLLM{text: "deep local answer", usage: Usage{PromptTokens: 100, CompletionTokens: 400}, model: "qwen-local"} + cfg := cascadeCfg() + cfg.ReasoningEnabled = true + cfg.Prices["qwen-local"] = ModelPrice{} + b := &Bot{cfg: &cfg, llm: local, log: discardLog()} + + res, err := b.generate(context.Background(), "подумай глубже про X", msgs("подумай глубже про X"), "", true) + if err != nil { + t.Fatalf("generate: %v", err) + } + if res.route != routeReason { + t.Fatalf("route = %q, want reason", res.route) + } + if res.cost.Token != 0 || res.finalModel != "qwen-local" { + t.Fatalf("cost=%v finalModel=%q, want $0/qwen-local", res.cost.Token, res.finalModel) + } +} + func hedgeContains(ms []Message, sub string) bool { for _, m := range ms { if strings.Contains(m.Content, sub) { diff --git a/apps/ai-bot/config.go b/apps/ai-bot/config.go index 4c1bdca0..8d7c1367 100644 --- a/apps/ai-bot/config.go +++ b/apps/ai-bot/config.go @@ -157,6 +157,33 @@ type Config struct { GeminiAPIKey string GeminiModel string + // --- Local LLM backend (Phase 1 of docs/plans/local_llm_backend.md): the home + // inference server behind the SSH tunnel serves the bot's voice for FREE while + // reachable; cloud Grok stays REQUIRED as the transparent fallback (failover.go). + // Off (default) → b.llm is the plain xAI client, byte-identical to today. --- + + LocalLLMEnabled bool + LocalLLMBaseURL string // OpenAI-compatible /v1 root, e.g. http://172.18.0.1:11434/v1 + LocalLLMAPIKey string // optional; empty = no Authorization header sent + LocalLLMModel string // the local server's model tag, e.g. huihui_ai/qwen3-abliterated:8b + // LocalLLMTemp is the temperature the LOCAL adapter sends (Qwen3's non-thinking + // 0.7), deliberately overriding the cascade's XAI_TEMPERATURE — ollama honours a + // request temperature over its Modelfile, so passing the Grok-tuned 0.6 through + // would silently mistune the local model (provider_local.go). + LocalLLMTemp float64 + // LocalLLMMaxTok replaces MAX_OUTPUT_TOKENS on the local leg (0 = inherit). + // ollama's max_tokens caps thinking+answer TOGETHER (unlike xAI, where thinking + // bills on top of max_tokens), so a thinking route (reasoning_effort low/high) on + // the cloud-sized 320 would burn the whole budget on reasoning and return empty + // content. Local tokens are free — default roomier. + LocalLLMMaxTok int + // LocalLLMTimeout is the local leg's own request deadline: a half-alive tunnel + // must fail fast enough to leave the shared RequestBudget room for the cloud + // fallback (shorter than the transport's 60s per-attempt ceiling). + LocalLLMTimeout time.Duration + // LocalLLMHealthInterval is the background health-probe period (failover.go). + LocalLLMHealthInterval time.Duration + SystemPromptPath string SystemPrompt string StateDir string @@ -292,6 +319,8 @@ func LoadConfig() (*Config, error) { ReasoningEffort: strings.ToLower(strings.TrimSpace(getenv("REASONING_EFFORT", "high"))), GeminiBaseURL: strings.TrimRight(getenv("GEMINI_BASE_URL", "https://generativelanguage.googleapis.com/v1beta/openai"), "/"), GeminiModel: getenv("GEMINI_MODEL", "gemini-2.5-flash-lite"), + LocalLLMBaseURL: strings.TrimRight(getenv("LOCAL_LLM_BASE_URL", ""), "/"), + LocalLLMModel: getenv("LOCAL_LLM_MODEL", "huihui_ai/qwen3-abliterated:8b"), } var problems []string @@ -305,7 +334,8 @@ func LoadConfig() (*Config, error) { {"AS_TOKEN", &cfg.ASToken}, {"HS_TOKEN", &cfg.HSToken}, {"XAI_API_KEY", &cfg.XAIAPIKey}, - {"GEMINI_API_KEY", &cfg.GeminiAPIKey}, // optional; required only if a Gemini layer is on + {"GEMINI_API_KEY", &cfg.GeminiAPIKey}, // optional; required only if a Gemini layer is on + {"LOCAL_LLM_API_KEY", &cfg.LocalLLMAPIKey}, // optional; the local endpoint usually has no auth } { v, err := getSecret(s.key) if err != nil { @@ -454,6 +484,52 @@ func LoadConfig() (*Config, error) { } cfg.Prices[cfg.ReasoningModel] = ModelPrice{InputPerM: rIn, CachedPerM: cfg.PriceCachedPerM, OutputPerM: rOut} + // Local LLM backend (voice failover, Phase 1). + if cfg.LocalLLMEnabled, err = getenvBool("LOCAL_LLM_ENABLED", false); err != nil { + problems = append(problems, err.Error()) + } + if cfg.LocalLLMTemp, err = getenvFloat("LOCAL_LLM_TEMP", 0.7); err != nil { + problems = append(problems, err.Error()) + } + if cfg.LocalLLMMaxTok, err = getenvInt("LOCAL_LLM_MAX_OUTPUT_TOKENS", 1024); err != nil { + problems = append(problems, err.Error()) + } + var localTimeoutSec, localHealthSec int + if localTimeoutSec, err = getenvInt("LOCAL_LLM_TIMEOUT_SECONDS", 45); err != nil { + problems = append(problems, err.Error()) + } + cfg.LocalLLMTimeout = time.Duration(localTimeoutSec) * time.Second + if localHealthSec, err = getenvInt("LOCAL_LLM_HEALTH_INTERVAL_SECONDS", 15); err != nil { + problems = append(problems, err.Error()) + } + cfg.LocalLLMHealthInterval = time.Duration(localHealthSec) * time.Second + // Local fail-fast (mirrors the Gemini block below): a half-configured local layer + // must refuse to boot, not quietly serve everything from the paid cloud. The XAI + // key stays REQUIRED regardless — the cloud is the fallback leg. + if cfg.LocalLLMEnabled { + if cfg.LocalLLMBaseURL == "" { + problems = append(problems, "LOCAL_LLM_BASE_URL is required when LOCAL_LLM_ENABLED is set") + } + if cfg.LocalLLMModel == "" { + problems = append(problems, "LOCAL_LLM_MODEL is required when LOCAL_LLM_ENABLED is set") + } + if localTimeoutSec <= 0 { + problems = append(problems, "LOCAL_LLM_TIMEOUT_SECONDS must be > 0") + } + if localHealthSec <= 0 { + problems = append(problems, "LOCAL_LLM_HEALTH_INTERVAL_SECONDS must be > 0") + } + // The $0 price entry is what books local answers as free: billing prices each + // call by the model that ACTUALLY answered (LLMResponse.Model → computeUSD). + // A name collision with a cloud model would zero THAT model's real price and + // blind the daily ceiling — refuse it. + if cfg.LocalLLMModel == cfg.XAIModel || cfg.LocalLLMModel == cfg.GeminiModel || cfg.LocalLLMModel == cfg.ReasoningModel { + problems = append(problems, "LOCAL_LLM_MODEL must differ from XAI_MODEL / GEMINI_MODEL / REASONING_MODEL (its $0 price entry would zero the cloud model's billing)") + } else if cfg.LocalLLMModel != "" { + cfg.Prices[cfg.LocalLLMModel] = ModelPrice{} // free: home GPU, no per-token cost + } + } + // Fail-fast on broken cascade wiring (§5/F-FUNC-9), at EVERY start (not just // check-config): a layer that needs Gemini but has no key would silently never // fire. Better to refuse to start than to quietly run degraded. @@ -605,6 +681,14 @@ func (c *Config) Summary() string { } return c.ProjectKBPath }()), + fmt.Sprintf(" LOCAL_LLM = enabled=%t url=%s model=%s temp=%g max_out=%d timeout=%s health=%s key=%s", + c.LocalLLMEnabled, func() string { + if c.LocalLLMBaseURL == "" { + return "(unset)" + } + return c.LocalLLMBaseURL + }(), c.LocalLLMModel, c.LocalLLMTemp, c.LocalLLMMaxTok, + c.LocalLLMTimeout, c.LocalLLMHealthInterval, redact(c.LocalLLMAPIKey)), " GEMINI_MODEL = " + c.GeminiModel, " GEMINI_API_KEY = " + redact(c.GeminiAPIKey), }, "\n") diff --git a/apps/ai-bot/config_test.go b/apps/ai-bot/config_test.go index c679f5b4..d66e03b5 100644 --- a/apps/ai-bot/config_test.go +++ b/apps/ai-bot/config_test.go @@ -3,6 +3,7 @@ package main import ( "strings" "testing" + "time" ) // setBaseEnv sets the minimal valid environment (all cascade flags off) so each test @@ -22,6 +23,9 @@ func setBaseEnv(t *testing.T) { "TRIVIAL_OFFLOAD_ENABLED", "WEB_ENABLED", "REASONING_ENABLED", "WEB_PROVIDER", "REASONING_MODEL", "WEB_PARANOID", "WEB_GROUNDING_DAILY_CAP", "GEMINI_GROUNDING_PER_PROMPT_USD", "PROJECT_KB_ENABLED", "PROJECT_KB_PATH", + "LOCAL_LLM_ENABLED", "LOCAL_LLM_BASE_URL", "LOCAL_LLM_MODEL", "LOCAL_LLM_API_KEY", + "LOCAL_LLM_API_KEY_FILE", "LOCAL_LLM_TEMP", "LOCAL_LLM_MAX_OUTPUT_TOKENS", + "LOCAL_LLM_TIMEOUT_SECONDS", "LOCAL_LLM_HEALTH_INTERVAL_SECONDS", } { t.Setenv(k, "") } @@ -41,7 +45,8 @@ func TestConfigAllCascadeFlagsDefaultOff(t *testing.T) { t.Fatalf("%v", err) } if cfg.RouterEnabled || cfg.RouterClassifierEnabled || cfg.TrivialOffloadEnabled || - cfg.WebEnabled || cfg.ReasoningEnabled || cfg.TelemetryEnabled || cfg.GrokPromptCache { + cfg.WebEnabled || cfg.ReasoningEnabled || cfg.TelemetryEnabled || cfg.GrokPromptCache || + cfg.LocalLLMEnabled { t.Fatal("every cascade/telemetry flag must default off (cascade-off == today)") } if cfg.WebProvider != webProviderGrokWebSearch { @@ -157,6 +162,60 @@ func TestConfigGeminiGroundingCapMustBePositive(t *testing.T) { } } +// TestConfigLocalLLMRequiresBaseURL: LOCAL_LLM_ENABLED without a base URL must refuse +// to boot (the fail-fast contract), and a full local config registers the $0 price +// entry for the local model with sane defaults for the rest. +func TestConfigLocalLLMRequiresBaseURL(t *testing.T) { + setBaseEnv(t) + t.Setenv("LOCAL_LLM_ENABLED", "true") + if _, err := LoadConfig(); err == nil || !strings.Contains(err.Error(), "LOCAL_LLM_BASE_URL") { + t.Fatalf("want LOCAL_LLM_BASE_URL error, got %v", err) + } + + t.Setenv("LOCAL_LLM_BASE_URL", "http://172.18.0.1:11434/v1") + cfg, err := LoadConfig() + if err != nil { + t.Fatalf("full local config should be valid: %v", err) + } + if cfg.LocalLLMModel != "huihui_ai/qwen3-abliterated:8b" { + t.Fatalf("LOCAL_LLM_MODEL default = %q", cfg.LocalLLMModel) + } + if p, ok := cfg.Prices[cfg.LocalLLMModel]; !ok || p != (ModelPrice{}) { + t.Fatalf("local model price = %+v (present=%t), want a $0 entry", p, ok) + } + if cfg.LocalLLMTemp != 0.7 || cfg.LocalLLMMaxTok != 1024 || + cfg.LocalLLMTimeout != 45*time.Second || cfg.LocalLLMHealthInterval != 15*time.Second { + t.Fatalf("local defaults = temp %g / max %d / timeout %s / health %s, want 0.7/1024/45s/15s", + cfg.LocalLLMTemp, cfg.LocalLLMMaxTok, cfg.LocalLLMTimeout, cfg.LocalLLMHealthInterval) + } +} + +// TestConfigLocalLLMModelCollision: naming the local model like a cloud one would +// register a $0 price for the CLOUD model and blind the daily ceiling — refuse it. +func TestConfigLocalLLMModelCollision(t *testing.T) { + setBaseEnv(t) + t.Setenv("LOCAL_LLM_ENABLED", "true") + t.Setenv("LOCAL_LLM_BASE_URL", "http://172.18.0.1:11434/v1") + t.Setenv("XAI_MODEL", "grok-4.20") + t.Setenv("LOCAL_LLM_MODEL", "grok-4.20") + if _, err := LoadConfig(); err == nil || !strings.Contains(err.Error(), "LOCAL_LLM_MODEL must differ") { + t.Fatalf("want the model-collision error, got %v", err) + } +} + +// TestConfigLocalLLMDisabledAddsNoPriceEntry: with the flag off nothing local leaks +// into the price table (byte-identical config surface to today). +func TestConfigLocalLLMDisabledAddsNoPriceEntry(t *testing.T) { + setBaseEnv(t) + cfg, err := LoadConfig() + if err != nil { + t.Fatalf("%v", err) + } + if _, ok := cfg.Prices[cfg.LocalLLMModel]; ok { + t.Fatal("a disabled local backend must not register a price entry") + } +} + // The default per-prompt grounding fee is the paid-tier $0.035 (the operator must opt to 0). func TestConfigGroundingFeeDefault(t *testing.T) { setBaseEnv(t) diff --git a/apps/ai-bot/failover.go b/apps/ai-bot/failover.go new file mode 100644 index 00000000..cdb5c087 --- /dev/null +++ b/apps/ai-bot/failover.go @@ -0,0 +1,234 @@ +package main + +import ( + "context" + "errors" + "io" + "log/slog" + "net/http" + "strings" + "sync" + "time" +) + +// failover.go is the local-first routing decorator over the LLMClient seam: prefer +// the (free) home backend while it is reachable, fall back to the cloud transparently +// when it is not. The cascade keeps calling b.llm.Complete and never learns which leg +// answered — LLMResponse.Model carries that to billing/telemetry. +// +// Health is decided by an ACTIVE background probe plus a request-path circuit +// breaker, both feeding one atomic healthy flag: +// +// - unhealthy → requests go straight to the cloud (no per-request timeout tax on a +// dead tunnel); +// - healthy → try local under its own short deadline (LOCAL_LLM_TIMEOUT_SECONDS — +// a half-alive tunnel must fail fast enough to leave the shared RequestBudget +// room for the cloud leg), and on a transport/5xx/timeout failure trip the +// breaker and retry the SAME request on the cloud; +// - recovery is probe-driven ONLY (anti-flapping): a tripped breaker needs +// tripProbesToClose consecutive probe successes to close, so a tunnel that +// accepts TCP but drops requests can't flap traffic between legs. +// +// A terminal 4xx from the local leg (except 429) does NOT fall back: that is a +// config/request error (a wrong LOCAL_LLM_MODEL tag, a malformed body) and the cloud +// would only mask it. 429 (a busy single-slot GPU) falls back like a timeout. +// Per-request fallback logs at DEBUG; only state transitions and a both-legs failure +// are louder. + +const ( + // localProbeTimeout bounds one health probe. The probe is a bare GET /models — + // answered from ollama's process memory, no inference — so anything slower than + // this means the tunnel, not the model, is struggling. + localProbeTimeout = 3 * time.Second + // tripProbesToClose is how many consecutive probe successes re-close a breaker + // tripped by a REQUEST failure. The probe can pass while real requests fail (a + // cold model being loaded, a half-alive tunnel), so one success is not proof; + // two spaced an interval apart also rate-limits the cold-load abort loop (an + // aborted request cancels ollama's model load, keeping every next request cold). + tripProbesToClose = 2 + // neverOnlineWarnAfter: consecutive probe failures — without the local leg EVER + // having come online since boot — that trigger a one-time WARN. A wrong-valued + // LOCAL_LLM_BASE_URL (say, missing the /v1 suffix) passes the boot presence + // check but 404s forever, and per-request fallbacks log only at DEBUG — so + // without this the enabled feature would be silently inoperative, serving 100% + // of traffic from the paid cloud (the exact outcome the config fail-fast exists + // to prevent). + neverOnlineWarnAfter = 3 +) + +type failoverLLMClient struct { + primary LLMClient // local backend (free, preferred) + fallback LLMClient // cloud (paid, always available) + log *slog.Logger + + legTimeout time.Duration // per-request deadline for the local leg + probeURL string // /models + probeKey string // optional bearer for the probe (usually empty) + interval time.Duration // background probe period + httpc *http.Client + + // Breaker state. healthy starts false: until the first probe passes, traffic is + // served by the cloud, so a boot with the home box offline never eats the probe + // timeout on user requests. + mu sync.Mutex + healthy bool + consecOK int // consecutive probe successes so far + needOK int // successes required to close (1 normally, tripProbesToClose after a trip) + consecFail int // consecutive probe failures (the never-online WARN counter) + everHealthy bool // the leg has been online at least once since boot + warnedDown bool // the one-time never-online WARN fired +} + +// NewFailoverLLMClient wraps primary (local) and fallback (cloud) into one LLMClient +// and starts the background health probe, which lives until ctx is cancelled (the +// process signal context — same lifetime as the bot). +func NewFailoverLLMClient(ctx context.Context, primary, fallback LLMClient, cfg *Config, logger *slog.Logger) LLMClient { + c := &failoverLLMClient{ + primary: primary, + fallback: fallback, + log: logger, + legTimeout: cfg.LocalLLMTimeout, + probeURL: cfg.LocalLLMBaseURL + "/models", + probeKey: cfg.LocalLLMAPIKey, + interval: cfg.LocalLLMHealthInterval, + httpc: &http.Client{}, + needOK: 1, + } + // One INFO at construction, so the absence of the follow-up "online" line is a + // documented signal (see the never-online WARN in noteProbe). + logger.Info("local llm failover enabled; cloud serves until the first successful probe", "probe_url", c.probeURL) + go c.probeLoop(ctx) + return c +} + +func (c *failoverLLMClient) Complete(ctx context.Context, req LLMRequest) (*LLMResponse, error) { + if !c.isHealthy() { + c.log.DebugContext(ctx, "local llm unhealthy; serving from cloud") + return c.fallback.Complete(ctx, req) + } + + lctx, cancel := context.WithTimeout(ctx, c.legTimeout) + resp, err := c.primary.Complete(lctx, req) + cancel() + if err == nil { + if strings.TrimSpace(resp.Text) != "" { + return resp, nil + } + // A 2xx with EMPTY content from the local leg (thinking ate the whole token + // budget, a content quirk of the abliterated model). On the plain cloud path + // an empty completion is kept — it was billed — but the local one is free, so + // retry the turn on the cloud instead of shipping the ⚠️-react empty path. + // The backend is up and answering: not a breaker trip. + c.log.DebugContext(ctx, "local llm returned empty content; retrying on cloud") + return c.fallback.Complete(ctx, req) + } + if ctx.Err() != nil { + // The WHOLE request budget is gone, not just the local leg — a cloud call + // would die the same way, so don't burn a paid attempt; and don't trip the + // breaker over our own cancellation. + return nil, err + } + var se *httpStatusError + if errors.As(err, &se) && se.status >= 400 && se.status < 500 && se.status != http.StatusTooManyRequests { + return nil, err // terminal 4xx: a local config error — fail loud, never mask with the cloud + } + + c.trip() + c.log.DebugContext(ctx, "local llm failed; falling back to cloud", "err", err) + resp, ferr := c.fallback.Complete(ctx, req) + if ferr != nil { + // Both legs down — this is the only per-request failure loud enough for WARN + // (the caller will still ERROR+react; this line preserves the local cause). + c.log.WarnContext(ctx, "local llm and cloud fallback both failed", "local_err", err, "cloud_err", ferr) + return nil, ferr + } + return resp, nil +} + +func (c *failoverLLMClient) isHealthy() bool { + c.mu.Lock() + defer c.mu.Unlock() + return c.healthy +} + +// trip opens the breaker after an in-request local failure. Recovery is probe-driven +// only: requests never re-try the local leg until the prober has re-closed it. +func (c *failoverLLMClient) trip() { + c.mu.Lock() + defer c.mu.Unlock() + if c.healthy { + c.log.Info("local llm tripped offline by a request failure; serving from cloud until probes recover") + } + c.healthy = false + c.consecOK = 0 + c.needOK = tripProbesToClose +} + +// noteProbe folds one probe result into the breaker. A failed probe opens it (the +// tunnel is observably down — don't wait for a user request to find out) but does +// not raise needOK: plain down/up recovers on the next success, only a request-path +// trip demands the longer streak. +func (c *failoverLLMClient) noteProbe(ok bool) { + c.mu.Lock() + defer c.mu.Unlock() + if !ok { + if c.healthy { + c.log.Info("local llm probe failed; serving from cloud") + } + c.healthy = false + c.consecOK = 0 + c.consecFail++ + // Never came online since boot → the feature is silently inoperative (every + // request quietly bills the cloud, visible only at DEBUG). Say so ONCE, loud. + if !c.everHealthy && !c.warnedDown && c.consecFail >= neverOnlineWarnAfter { + c.warnedDown = true + c.log.Warn("local llm has never come online since boot; ALL traffic is served by the paid cloud — check LOCAL_LLM_BASE_URL and the tunnel", "probe_url", c.probeURL) + } + return + } + c.consecFail = 0 + c.consecOK++ + if !c.healthy && c.consecOK >= c.needOK { + c.healthy = true + c.everHealthy = true + c.needOK = 1 + c.log.Info("local llm online; serving from local backend") + } +} + +func (c *failoverLLMClient) probeLoop(ctx context.Context) { + t := time.NewTicker(c.interval) + defer t.Stop() + for { + c.noteProbe(c.probe(ctx)) // first probe runs immediately, not an interval late + select { + case <-ctx.Done(): + return + case <-t.C: + } + } +} + +// probe GETs the OpenAI models listing on the local base URL — cheap (no inference) +// and it proves the whole path (SSH tunnel → relay → server) end to end, which is +// exactly what dies when the home box goes away. It does NOT prove the model is +// resident in VRAM; keeping it warm is OLLAMA_KEEP_ALIVE=-1's job (a cold model +// answers slowly, which the request-path breaker handles). +func (c *failoverLLMClient) probe(ctx context.Context) bool { + pctx, cancel := context.WithTimeout(ctx, localProbeTimeout) + defer cancel() + req, err := http.NewRequestWithContext(pctx, http.MethodGet, c.probeURL, nil) + if err != nil { + return false + } + if c.probeKey != "" { + req.Header.Set("Authorization", "Bearer "+c.probeKey) + } + resp, err := c.httpc.Do(req) + if err != nil { + return false + } + defer resp.Body.Close() + _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 4<<10)) // drain for connection reuse + return resp.StatusCode >= 200 && resp.StatusCode < 300 +} diff --git a/apps/ai-bot/failover_test.go b/apps/ai-bot/failover_test.go new file mode 100644 index 00000000..cd5ad927 --- /dev/null +++ b/apps/ai-bot/failover_test.go @@ -0,0 +1,331 @@ +package main + +import ( + "bytes" + "context" + "errors" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" +) + +// funcLLM is an LLMClient driven by a closure — for failover legs that need to +// block, inspect their context, or fail in a specific typed way. +type funcLLM struct { + fn func(ctx context.Context, req LLMRequest) (*LLMResponse, error) + calls int +} + +func (f *funcLLM) Complete(ctx context.Context, req LLMRequest) (*LLMResponse, error) { + f.calls++ + return f.fn(ctx, req) +} + +// newTestFailover builds the decorator without the background prober, so tests +// control the breaker state deterministically. +func newTestFailover(primary, fallback LLMClient, healthy bool) *failoverLLMClient { + return &failoverLLMClient{ + primary: primary, + fallback: fallback, + log: discardLog(), + legTimeout: time.Second, + healthy: healthy, + needOK: 1, + } +} + +// TestFailoverHealthyPrefersLocal: with a healthy breaker the local leg answers and +// the cloud is never touched; the response carries the local leg's Model untouched. +func TestFailoverHealthyPrefersLocal(t *testing.T) { + local := &fakeLLM{text: "local answer", model: "qwen-local"} + cloud := &fakeLLM{text: "cloud answer", model: "grok-x"} + c := newTestFailover(local, cloud, true) + + resp, err := c.Complete(context.Background(), LLMRequest{Model: "grok-x"}) + if err != nil { + t.Fatalf("Complete: %v", err) + } + if resp.Text != "local answer" || resp.Model != "qwen-local" { + t.Fatalf("resp = (%q, model %q), want the local leg's answer", resp.Text, resp.Model) + } + if local.calls != 1 || cloud.calls != 0 { + t.Fatalf("calls local=%d cloud=%d, want 1/0", local.calls, cloud.calls) + } + if !c.isHealthy() { + t.Fatal("a successful local answer must not touch the breaker") + } +} + +// TestFailoverUnhealthyGoesStraightToCloud: an open breaker sends the request to the +// cloud without paying the local leg's timeout (the local client is never called). +func TestFailoverUnhealthyGoesStraightToCloud(t *testing.T) { + local := &fakeLLM{text: "local"} + cloud := &fakeLLM{text: "cloud answer", model: "grok-x"} + c := newTestFailover(local, cloud, false) + + resp, err := c.Complete(context.Background(), LLMRequest{}) + if err != nil { + t.Fatalf("Complete: %v", err) + } + if resp.Text != "cloud answer" { + t.Fatalf("resp = %q, want the cloud answer", resp.Text) + } + if local.calls != 0 || cloud.calls != 1 { + t.Fatalf("calls local=%d cloud=%d, want 0/1", local.calls, cloud.calls) + } +} + +// TestFailoverTransportErrorFallsBackAndTrips: a network-class local failure answers +// from the cloud within the SAME request and opens the breaker, so the next request +// skips the local leg entirely. +func TestFailoverTransportErrorFallsBackAndTrips(t *testing.T) { + local := &fakeLLM{err: errors.New("dial tcp: connection refused")} + cloud := &fakeLLM{text: "cloud answer", model: "grok-x"} + c := newTestFailover(local, cloud, true) + + resp, err := c.Complete(context.Background(), LLMRequest{}) + if err != nil || resp.Text != "cloud answer" { + t.Fatalf("resp=(%v,%v), want the cloud fallback", resp, err) + } + if c.isHealthy() { + t.Fatal("a transport failure must trip the breaker") + } + if _, err := c.Complete(context.Background(), LLMRequest{}); err != nil { + t.Fatalf("second Complete: %v", err) + } + if local.calls != 1 || cloud.calls != 2 { + t.Fatalf("calls local=%d cloud=%d, want 1/2 (tripped breaker skips local)", local.calls, cloud.calls) + } +} + +// TestFailover5xxAnd429FallBack: retryable-class statuses (a dying tunnel's 502, a +// busy single-slot GPU's 429) fall back to the cloud like a transport error. +func TestFailover5xxAnd429FallBack(t *testing.T) { + for _, status := range []int{500, 502, http.StatusTooManyRequests} { + local := &fakeLLM{err: &httpStatusError{provider: "local", status: status, body: "boom"}} + cloud := &fakeLLM{text: "cloud answer"} + c := newTestFailover(local, cloud, true) + resp, err := c.Complete(context.Background(), LLMRequest{}) + if err != nil || resp.Text != "cloud answer" { + t.Fatalf("status %d: resp=(%v,%v), want the cloud fallback", status, resp, err) + } + } +} + +// TestFailoverTerminal4xxFailsLoud: a local 4xx is a config/request error (a wrong +// LOCAL_LLM_MODEL tag) — masking it with a paid cloud answer would hide the +// misconfiguration, so it surfaces as the request's error and the cloud is not called. +func TestFailoverTerminal4xxFailsLoud(t *testing.T) { + local := &fakeLLM{err: &httpStatusError{provider: "local", status: 404, body: "model not found"}} + cloud := &fakeLLM{text: "cloud"} + c := newTestFailover(local, cloud, true) + + if _, err := c.Complete(context.Background(), LLMRequest{}); err == nil { + t.Fatal("want the local 4xx surfaced, got nil") + } + if cloud.calls != 0 { + t.Fatalf("cloud called %d times on a local 4xx, want 0 (fail loud)", cloud.calls) + } + if !c.isHealthy() { + t.Fatal("a 4xx is not an availability failure — the breaker must stay closed") + } +} + +// TestFailoverBudgetGoneNoCloudRetry: when the WHOLE request budget is exhausted +// (parent ctx done), the local error surfaces without burning a doomed cloud attempt +// and without tripping the breaker over our own cancellation. +func TestFailoverBudgetGoneNoCloudRetry(t *testing.T) { + local := &funcLLM{fn: func(ctx context.Context, _ LLMRequest) (*LLMResponse, error) { + <-ctx.Done() + return nil, ctx.Err() + }} + cloud := &fakeLLM{text: "cloud"} + c := newTestFailover(local, cloud, true) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() // the shared RequestBudget is gone before the call + if _, err := c.Complete(ctx, LLMRequest{}); err == nil { + t.Fatal("want an error when the request budget is gone") + } + if cloud.calls != 0 { + t.Fatalf("cloud called %d times after budget exhaustion, want 0", cloud.calls) + } + if !c.isHealthy() { + t.Fatal("our own cancellation must not trip the breaker") + } +} + +// TestFailoverLocalTimeoutFallsBack: the local leg's own deadline (a half-alive +// tunnel that accepts but never answers) expires and the request is served by the +// cloud — the shared budget is still alive. +func TestFailoverLocalTimeoutFallsBack(t *testing.T) { + local := &funcLLM{fn: func(ctx context.Context, _ LLMRequest) (*LLMResponse, error) { + <-ctx.Done() // blocks until the leg deadline fires + return nil, ctx.Err() + }} + cloud := &fakeLLM{text: "cloud answer"} + c := newTestFailover(local, cloud, true) + c.legTimeout = 10 * time.Millisecond + + resp, err := c.Complete(context.Background(), LLMRequest{}) + if err != nil || resp.Text != "cloud answer" { + t.Fatalf("resp=(%v,%v), want the cloud fallback after the leg timeout", resp, err) + } + if c.isHealthy() { + t.Fatal("a local leg timeout must trip the breaker") + } +} + +// TestFailoverEmptyLocalReplyRetriesOnCloud: a 2xx with empty content from the free +// local leg (thinking ate the token budget) is retried on the cloud instead of +// shipping the empty-completion ⚠️ path; the breaker stays closed (the backend is up). +func TestFailoverEmptyLocalReplyRetriesOnCloud(t *testing.T) { + local := &fakeLLM{text: " ", model: "qwen-local"} + cloud := &fakeLLM{text: "cloud answer", model: "grok-x"} + c := newTestFailover(local, cloud, true) + + resp, err := c.Complete(context.Background(), LLMRequest{}) + if err != nil || resp.Text != "cloud answer" { + t.Fatalf("resp=(%v,%v), want the cloud retry on an empty local reply", resp, err) + } + if !c.isHealthy() { + t.Fatal("an empty local reply is not an availability failure — no trip") + } +} + +// TestFailoverBothLegsDown: local and cloud both failing surfaces the cloud error +// (the caller refunds + reacts) — never a silent nil. +func TestFailoverBothLegsDown(t *testing.T) { + local := &fakeLLM{err: errors.New("tunnel down")} + cloudErr := errors.New("xai down") + cloud := &fakeLLM{err: cloudErr} + c := newTestFailover(local, cloud, true) + + _, err := c.Complete(context.Background(), LLMRequest{}) + if !errors.Is(err, cloudErr) { + t.Fatalf("err = %v, want the cloud error surfaced", err) + } +} + +// TestFailoverBreakerRecovery is the anti-flapping state machine: a request-path trip +// needs tripProbesToClose consecutive probe successes to close; a plain probe-down +// re-closes on the next success; a mid-streak probe failure resets the streak. +func TestFailoverBreakerRecovery(t *testing.T) { + c := newTestFailover(&fakeLLM{}, &fakeLLM{}, false) + + // Startup: one successful probe brings the local leg online. + c.noteProbe(true) + if !c.isHealthy() { + t.Fatal("one probe success must close the breaker at startup") + } + + // Request-path trip: one probe success is NOT enough, two are. + c.trip() + c.noteProbe(true) + if c.isHealthy() { + t.Fatal("after a trip one probe success must not re-close the breaker") + } + c.noteProbe(true) + if !c.isHealthy() { + t.Fatalf("after a trip %d probe successes must re-close the breaker", tripProbesToClose) + } + + // A mid-streak failure resets the streak. + c.trip() + c.noteProbe(true) + c.noteProbe(false) + c.noteProbe(true) + if c.isHealthy() { + t.Fatal("a probe failure mid-streak must reset the recovery counter") + } + c.noteProbe(true) + if !c.isHealthy() { + t.Fatal("a full consecutive streak after the reset must re-close the breaker") + } + + // Plain probe-down (no request trip): the next single success re-closes. + c.noteProbe(false) + if c.isHealthy() { + t.Fatal("a probe failure must open the breaker") + } + c.noteProbe(true) + if !c.isHealthy() { + t.Fatal("plain probe-down must recover on one success") + } +} + +// TestFailoverNeverOnlineWarnsOnce: a local leg that has never come online since +// boot (a wrong LOCAL_LLM_BASE_URL 404s every probe) must WARN exactly once — an +// enabled-but-inoperative feature silently billing the cloud is the failure the +// fail-fast config comment promises to prevent — while a leg that WAS online logs +// its down transition at INFO only. +func TestFailoverNeverOnlineWarnsOnce(t *testing.T) { + var buf bytes.Buffer + c := newTestFailover(&fakeLLM{}, &fakeLLM{}, false) + c.log = slog.New(slog.NewTextHandler(&buf, nil)) + + for i := 0; i < neverOnlineWarnAfter-1; i++ { + c.noteProbe(false) + } + if strings.Contains(buf.String(), "never come online") { + t.Fatalf("WARN fired before %d consecutive failures: %s", neverOnlineWarnAfter, buf.String()) + } + c.noteProbe(false) + if n := strings.Count(buf.String(), "never come online"); n != 1 { + t.Fatalf("never-online WARN count = %d, want exactly 1", n) + } + c.noteProbe(false) // still down — must not repeat + if n := strings.Count(buf.String(), "never come online"); n != 1 { + t.Fatalf("never-online WARN repeated: count = %d", n) + } + + // Once the leg HAS been online, a later long outage is a transition (INFO), not + // the never-online WARN. + buf.Reset() + c.noteProbe(true) + for i := 0; i < neverOnlineWarnAfter+1; i++ { + c.noteProbe(false) + } + if strings.Contains(buf.String(), "never come online") { + t.Fatalf("never-online WARN fired for a leg that was online: %s", buf.String()) + } +} + +// TestFailoverProbeLoop exercises the real constructor end to end: the background +// prober sees a live /models endpoint and brings the local leg online; the endpoint +// dying takes it offline again. +func TestFailoverProbeLoop(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/models" { + t.Errorf("probe hit %q, want /models", r.URL.Path) + } + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + cfg := &Config{ + LocalLLMBaseURL: srv.URL, + LocalLLMTimeout: time.Second, + LocalLLMHealthInterval: 10 * time.Millisecond, + } + c := NewFailoverLLMClient(ctx, &fakeLLM{text: "local"}, &fakeLLM{text: "cloud"}, cfg, discardLog()).(*failoverLLMClient) + + waitFor := func(want bool, what string) { + t.Helper() + deadline := time.Now().Add(3 * time.Second) + for c.isHealthy() != want { + if time.Now().After(deadline) { + t.Fatalf("timed out waiting for %s", what) + } + time.Sleep(5 * time.Millisecond) + } + } + waitFor(true, "the prober to bring the local leg online") + srv.Close() + waitFor(false, "the prober to take the dead endpoint offline") +} diff --git a/apps/ai-bot/httpllm.go b/apps/ai-bot/httpllm.go index 76999056..7004fa09 100644 --- a/apps/ai-bot/httpllm.go +++ b/apps/ai-bot/httpllm.go @@ -219,7 +219,11 @@ func (c *openAIClient) attempt(ctx context.Context, payload []byte, reqHeaders m return nil, false, err } req.Header.Set("Content-Type", "application/json") - req.Header.Set("Authorization", "Bearer "+c.key) + // 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) } @@ -240,10 +244,10 @@ func (c *openAIClient) attempt(ctx context.Context, payload []byte, reqHeaders m logLLMExchange(ctx, c.log, c.name, payload, resp.StatusCode, data) if resp.StatusCode == http.StatusTooManyRequests || resp.StatusCode >= 500 { - return nil, true, fmt.Errorf("%s http %d: %s", c.name, resp.StatusCode, snippet(data)) + return nil, true, &httpStatusError{provider: c.name, status: resp.StatusCode, body: snippet(data)} } if resp.StatusCode < 200 || resp.StatusCode >= 300 { - return nil, false, fmt.Errorf("%s http %d: %s", c.name, resp.StatusCode, snippet(data)) + return nil, false, &httpStatusError{provider: c.name, status: resp.StatusCode, body: snippet(data)} } var out openAIResponse @@ -258,6 +262,22 @@ func (c *openAIClient) attempt(ctx context.Context, payload []byte, reqHeaders m return &out, false, nil } +// httpStatusError is a non-2xx completion failure, carrying the status code as a +// typed field so callers (the local/cloud failover) 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. Error() keeps the exact wording the +// plain fmt.Errorf produced, so logs and message-matching (isReasoningEffortUnsupported) +// are unchanged. +type httpStatusError struct { + provider string + status int + body string +} + +func (e *httpStatusError) Error() string { + return fmt.Sprintf("%s http %d: %s", e.provider, e.status, e.body) +} + // isReasoningEffortUnsupported reports whether an xAI error is the specific 400 a // non-reasoning model returns when sent reasoning_effort ("...does not support parameter // reasoningEffort"). Matched loosely so both reasoning_effort and reasoningEffort spellings diff --git a/apps/ai-bot/httpllm_test.go b/apps/ai-bot/httpllm_test.go index 74fd62d0..5b04e4d7 100644 --- a/apps/ai-bot/httpllm_test.go +++ b/apps/ai-bot/httpllm_test.go @@ -96,6 +96,30 @@ func TestCompleteReasoningEffortCachedAfterFirst(t *testing.T) { } } +// TestCompleteAuthHeaderOptional: an empty API key sends NO Authorization header (a +// bare local endpoint — ollama/llama-server — has no auth), while a set key keeps the +// exact Bearer header the cloud providers require. +func TestCompleteAuthHeaderOptional(t *testing.T) { + var gotAuth []string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotAuth = append(gotAuth, r.Header.Get("Authorization")) + w.Header().Set("Content-Type", "application/json") + io.WriteString(w, `{"id":"ok","choices":[{"message":{"content":"hi"},"finish_reason":"stop"}],"usage":{}}`) + })) + defer srv.Close() + + req := openAIRequest{Model: "m", Messages: []openAIMessage{{Role: "user", Content: "hi"}}} + if _, err := newOpenAIClient("local", srv.URL, "", nil, discardLog()).complete(context.Background(), req, nil); err != nil { + t.Fatalf("keyless complete: %v", err) + } + if _, err := newOpenAIClient("xai", srv.URL, "sekret", nil, discardLog()).complete(context.Background(), req, nil); err != nil { + t.Fatalf("keyed complete: %v", err) + } + if len(gotAuth) != 2 || gotAuth[0] != "" || gotAuth[1] != "Bearer sekret" { + t.Fatalf("Authorization headers = %q, want [\"\" \"Bearer sekret\"]", gotAuth) + } +} + // TestCompleteTerminal4xxNoSelfHeal guards that the strip-and-retry is scoped to the // reasoning_effort 400 only: an unrelated 400 still fails fast (no spurious retry). func TestCompleteTerminal4xxNoSelfHeal(t *testing.T) { diff --git a/apps/ai-bot/llm.go b/apps/ai-bot/llm.go index b19d7b04..3b2ddf60 100644 --- a/apps/ai-bot/llm.go +++ b/apps/ai-bot/llm.go @@ -62,8 +62,15 @@ type LLMRequest struct { // LLMResponse is a provider-neutral completion result. type LLMResponse struct { - Text string - Usage Usage + Text string + Usage Usage + // Model is the model that ACTUALLY served this completion. It can differ from + // LLMRequest.Model when a routing decorator (the local/cloud failover) answered + // with a different backend than the cascade asked for — billing and telemetry + // must follow the responder, not the request, or a free local answer books at + // cloud prices. "" = unknown (an adapter predating this field, or a test fake); + // consumers fall back to the requested model. + Model string ProviderRequestID string // the backend's response id, logged for support/debug } diff --git a/apps/ai-bot/provider_gemini.go b/apps/ai-bot/provider_gemini.go index 121685af..2b697483 100644 --- a/apps/ai-bot/provider_gemini.go +++ b/apps/ai-bot/provider_gemini.go @@ -86,6 +86,7 @@ func (c *geminiClient) Complete(ctx context.Context, req LLMRequest) (*LLMRespon CachedTokens: resp.Usage.PromptTokensDetails.CachedTokens, CompletionTokens: resp.Usage.CompletionTokens, }, + Model: req.Model, // Gemini serves exactly the requested model ProviderRequestID: resp.ID, }, nil } diff --git a/apps/ai-bot/provider_local.go b/apps/ai-bot/provider_local.go new file mode 100644 index 00000000..41819a7f --- /dev/null +++ b/apps/ai-bot/provider_local.go @@ -0,0 +1,90 @@ +package main + +import ( + "context" + "log/slog" +) + +// provider_local.go is the thin adapter for the self-hosted inference backend (the +// home ollama/llama-server behind the SSH tunnel — docs/plans/local_llm_backend.md). +// Both speak the OpenAI Chat Completions wire format, so like xAI this is a shell +// over the shared openAIClient transport (httpllm.go). It differs from the cloud +// adapters in three deliberate ways: +// +// - The MODEL is the adapter's own (LOCAL_LLM_MODEL), never the cascade's: the +// cascade asks for grok-* names that don't exist locally. The response's Model +// reports what actually served, so billing prices it at the local $0 entry. +// - The TEMPERATURE is the adapter's own (LOCAL_LLM_TEMP, Qwen3's non-thinking 0.7): +// ollama honours a request temperature over the Modelfile, so passing through the +// cascade's XAI_TEMPERATURE (0.6) would silently mistune the local model. The +// remaining Qwen3 samplers (top_k/min_p/repeat_penalty) can't cross ollama's /v1 +// layer at all and are baked into the Modelfile at home. +// - MAX TOKENS has a local override (LOCAL_LLM_MAX_OUTPUT_TOKENS): unlike xAI — +// where thinking bills on top of max_tokens — ollama's cap covers thinking AND +// answer together, so the cloud-sized MAX_OUTPUT_TOKENS would let a thinking +// route (reasoning_effort low/high) burn the whole budget on reasoning and +// return empty content. Local tokens are free; the override defaults roomier. +// +// ReasoningEffort passes through unchanged: ollama's /v1 maps it onto Qwen3 thinking +// ("none" = off — the casual grok_direct route; "low"/"high" = on — verified live), +// so the cascade's existing per-route efforts drive local thinking with no new knob. +// ConvID (the x-grok-conv-id cache-routing header) and Tools are xAI concepts with no +// local equivalent and are deliberately not forwarded. +type localClient struct { + http *openAIClient + model string + temp float64 + maxTok int // 0 = inherit the request's MaxTokens +} + +// NewLocalLLMClient builds the local-backend adapter. Returns the neutral LLMClient +// so the bot holds no vendor type. The API key is optional (a bare local endpoint +// sends no Authorization header — see openAIClient.attempt). +func NewLocalLLMClient(cfg *Config, logger *slog.Logger) LLMClient { + return &localClient{ + http: newOpenAIClient("local", cfg.LocalLLMBaseURL, cfg.LocalLLMAPIKey, nil, logger), + model: cfg.LocalLLMModel, + temp: cfg.LocalLLMTemp, + maxTok: cfg.LocalLLMMaxTok, + } +} + +func (c *localClient) Complete(ctx context.Context, req LLMRequest) (*LLMResponse, error) { + msgs := make([]openAIMessage, len(req.Messages)) + for i, m := range req.Messages { + msgs[i] = openAIMessage{Role: m.Role, Content: m.Content} + } + maxTok := req.MaxTokens + if c.maxTok > 0 { + maxTok = c.maxTok + } + var respFormat any + if req.JSONOnly { + respFormat = map[string]string{"type": "json_object"} + } + resp, err := c.http.complete(ctx, openAIRequest{ + Model: c.model, + Messages: msgs, + MaxTokens: maxTok, + Temperature: c.temp, + Stream: false, + ReasoningEffort: req.ReasoningEffort, + ResponseFormat: respFormat, + }, nil) + if err != nil { + return nil, err + } + return &LLMResponse{ + Text: resp.Text(), + Usage: Usage{ + PromptTokens: resp.Usage.PromptTokens, + CachedTokens: resp.Usage.PromptTokensDetails.CachedTokens, + CompletionTokens: resp.Usage.CompletionTokens, + // ReasoningTokens deliberately left 0: ollama counts thinking INSIDE + // completion_tokens (subset semantics — see llm.go), and the details + // field is absent anyway. It's all $0 regardless. + }, + Model: c.model, // the local model answered, whatever the cascade asked for + ProviderRequestID: resp.ID, + }, nil +} diff --git a/apps/ai-bot/provider_local_test.go b/apps/ai-bot/provider_local_test.go new file mode 100644 index 00000000..7f60d6d0 --- /dev/null +++ b/apps/ai-bot/provider_local_test.go @@ -0,0 +1,103 @@ +package main + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "testing" +) + +const localOKBody = `{"id":"local-1","model":"qwen-test",` + + `"choices":[{"message":{"content":"привет"},"finish_reason":"stop"}],` + + `"usage":{"prompt_tokens":10,"completion_tokens":5}}` + +func localTestCfg(base string) *Config { + return &Config{ + LocalLLMBaseURL: base, + LocalLLMModel: "qwen-test", + LocalLLMTemp: 0.7, + LocalLLMMaxTok: 1024, + } +} + +// TestLocalClientOverridesModelTempMaxTokens: the adapter's contract with the local +// backend — its OWN model tag and Qwen3 temperature (never the cascade's Grok name / +// XAI_TEMPERATURE), the roomier local max_tokens, reasoning_effort passed through, +// and no xAI-specific conv-id header. The response reports the local model as the +// responder (the billing/telemetry key). +func TestLocalClientOverridesModelTempMaxTokens(t *testing.T) { + var gotBody map[string]any + var gotConvHeader string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + if err := json.Unmarshal(body, &gotBody); err != nil { + t.Errorf("request body: %v", err) + } + gotConvHeader = r.Header.Get("x-grok-conv-id") + w.Header().Set("Content-Type", "application/json") + io.WriteString(w, localOKBody) + })) + defer srv.Close() + + c := NewLocalLLMClient(localTestCfg(srv.URL), discardLog()) + resp, err := c.Complete(context.Background(), LLMRequest{ + Model: "grok-4.20", // what the cascade asks for — must NOT reach the wire + Messages: []Message{{Role: "user", Content: "привет"}}, + MaxTokens: 320, + Temperature: 0.6, // the Grok tuning — must NOT reach the wire + ReasoningEffort: "none", + ConvID: "conv-1", // xAI cache routing — no local equivalent + }) + if err != nil { + t.Fatalf("Complete: %v", err) + } + if gotBody["model"] != "qwen-test" { + t.Fatalf("wire model = %v, want the adapter's own qwen-test", gotBody["model"]) + } + if gotBody["temperature"] != 0.7 { + t.Fatalf("wire temperature = %v, want the local 0.7 (not the cascade's 0.6)", gotBody["temperature"]) + } + if gotBody["max_tokens"] != float64(1024) { + t.Fatalf("wire max_tokens = %v, want the local override 1024", gotBody["max_tokens"]) + } + if gotBody["reasoning_effort"] != "none" { + t.Fatalf("wire reasoning_effort = %v, want the cascade's value passed through", gotBody["reasoning_effort"]) + } + if gotConvHeader != "" { + t.Fatalf("x-grok-conv-id = %q sent to the local backend, want none", gotConvHeader) + } + if resp.Model != "qwen-test" { + t.Fatalf("resp.Model = %q, want qwen-test (the responder identity)", resp.Model) + } + if resp.Text != "привет" || resp.Usage.PromptTokens != 10 || resp.Usage.CompletionTokens != 5 { + t.Fatalf("resp = %+v, want the parsed local completion", resp) + } + if resp.Usage.ReasoningTokens != 0 { + t.Fatal("local ReasoningTokens must stay 0 (subset semantics — see llm.go)") + } +} + +// TestLocalClientInheritsMaxTokensWhenUnset: LOCAL_LLM_MAX_OUTPUT_TOKENS=0 means +// "inherit the request's MaxTokens". +func TestLocalClientInheritsMaxTokensWhenUnset(t *testing.T) { + var gotBody map[string]any + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + _ = json.Unmarshal(body, &gotBody) + w.Header().Set("Content-Type", "application/json") + io.WriteString(w, localOKBody) + })) + defer srv.Close() + + cfg := localTestCfg(srv.URL) + cfg.LocalLLMMaxTok = 0 + c := NewLocalLLMClient(cfg, discardLog()) + if _, err := c.Complete(context.Background(), LLMRequest{MaxTokens: 320}); err != nil { + t.Fatalf("Complete: %v", err) + } + if gotBody["max_tokens"] != float64(320) { + t.Fatalf("wire max_tokens = %v, want the inherited 320", gotBody["max_tokens"]) + } +} diff --git a/apps/ai-bot/provider_xai.go b/apps/ai-bot/provider_xai.go index c0e14c56..be217714 100644 --- a/apps/ai-bot/provider_xai.go +++ b/apps/ai-bot/provider_xai.go @@ -59,6 +59,7 @@ func (c *xaiClient) Complete(ctx context.Context, req LLMRequest) (*LLMResponse, // xAI bills reasoning at the output rate on top of completion_tokens. ReasoningTokens: resp.Usage.CompletionTokensDetails.ReasoningTokens, }, + Model: req.Model, // xAI serves exactly the requested model ProviderRequestID: resp.ID, }, nil } diff --git a/apps/ai-bot/router.go b/apps/ai-bot/router.go index 5e956362..f52470ba 100644 --- a/apps/ai-bot/router.go +++ b/apps/ai-bot/router.go @@ -161,7 +161,7 @@ func (b *Bot) routeLayer1(ctx context.Context, rcx string, l0 rd.Layer0, cost *C if err != nil { return RouterDecision{}, err } - cost.Router += computeUSD(b.cfg.GeminiModel, resp.Usage, b.cfg) + cost.Router += computeUSD(servedModel(b.cfg.GeminiModel, resp), resp.Usage, b.cfg) // The classifier schema IS routedecide.Verdict (tagged), so unmarshal straight into it. var v rd.Verdict diff --git a/docs/ai/ai-bot.md b/docs/ai/ai-bot.md index 3571ad87..c77e34ba 100644 --- a/docs/ai/ai-bot.md +++ b/docs/ai/ai-bot.md @@ -66,8 +66,34 @@ then dispatches; **any layer off or failing degrades to `grok_direct`** (never a [llm.go](../../apps/ai-bot/llm.go) (`Message`/`Usage`/`LLMRequest`/`LLMResponse`/`LLMClient`) + [httpllm.go](../../apps/ai-bot/httpllm.go) (shared OpenAI-compatible transport + retry) + thin adapters [provider_xai.go](../../apps/ai-bot/provider_xai.go) / -[provider_gemini.go](../../apps/ai-bot/provider_gemini.go) + [pricing.go](../../apps/ai-bot/pricing.go) +[provider_gemini.go](../../apps/ai-bot/provider_gemini.go) / +[provider_local.go](../../apps/ai-bot/provider_local.go) + [pricing.go](../../apps/ai-bot/pricing.go) (`priceFor` model→price map). `Bot.llm` is an `LLMClient`, never a concrete vendor type. +**The seam is the extension point** — one adapter per provider per file, cross-provider policy +(failover, billing, telemetry) lives in decorators/neutral fields, never in an adapter: +`LLMResponse.Model` reports which model **actually** answered, and every `computeUSD` call-site +bills by it (`servedModel`), so routing decorators can swap backends without touching accounting. + +## Local LLM backend (voice failover — Phase 1 of `docs/plans/local_llm_backend.md`) + +`LOCAL_LLM_ENABLED` wraps `b.llm` in [failover.go](../../apps/ai-bot/failover.go): primary = the +self-hosted backend ([provider_local.go](../../apps/ai-bot/provider_local.go), any OpenAI-compatible +server — ollama / llama.cpp / vLLM / LM Studio — behind `LOCAL_LLM_BASE_URL`; in prod the home +GPU over an SSH reverse-tunnel relay at `http://172.18.0.1:11434/v1`), fallback = cloud Grok. +**Every voice call** (grok_direct, reason, web/project synthesis) goes local-first while an active +`GET /models` probe + circuit breaker say the tunnel is healthy; otherwise — or on a local +transport/5xx/timeout/429 failure mid-request — the same request is served by Grok. Recovery is +probe-driven only (a request-path trip needs 2 consecutive probe passes — anti-flapping); a local +**terminal 4xx fails loud** (config error, never masked by paid cloud answers); an **empty** local +2xx retries on the cloud (free leg, so no billed-empty dilemma). A local answer books **$0** +(`cfg.Prices[LOCAL_LLM_MODEL]={0,0,0}` + billing by `LLMResponse.Model`); `request_log.models.final` +records the responder, so the local share and quality (reaction feedback) are measurable per +backend. The adapter sends its OWN model/temperature (`LOCAL_LLM_TEMP=0.7`, Qwen3 non-thinking) +and a roomier `LOCAL_LLM_MAX_OUTPUT_TOKENS=1024` (ollama's cap covers thinking+answer TOGETHER, +unlike xAI where thinking bills on top — the cloud 320 would return empty content on thinking +routes); `reasoning_effort` passes through (ollama `/v1`: `none`=off, `low`/`high`=thinking on). +Classifier (`b.gemini`) and web fetch (`b.web`) stay cloud — phases 2–3. Env table in the +[README](../../apps/ai-bot/README.md#local-llm-backend-voice-failover-default-off). ## Money, invariants & store ([store.go](../../apps/ai-bot/store.go))