diff --git a/.env.example b/.env.example index ceeef71..b6602db 100644 --- a/.env.example +++ b/.env.example @@ -64,10 +64,34 @@ SEED_FIXTURE_PATH=./seed/fixtures/seed.json # mapping below is a deployment decision and changes without editing a single # definition. # -# The key may be left empty outside production: migrations, seeding and every +# WHICH PROVIDER ANSWERS is a deployment decision. Two wire protocols: +# +# anthropic the Claude API. The default, and what an unset value means. +# openai the chat-completions shape — which is NOT only OpenAI. Groq, +# Gemini (through its OpenAI-compatible endpoint), OpenRouter, +# Together, vLLM and a local Ollama all serve it, so moving +# between them is MODEL_BASE_URL and MODEL_* ids, nothing more. +MODEL_PROVIDER=anthropic + +# Where the openai-compatible provider points. IGNORED — and refused at +# startup — unless MODEL_PROVIDER=openai, because a base URL set against the +# anthropic provider is a deployment that believes it has switched and has not: +# every run would still go to Anthropic, and still be billed there. +# +# Groq https://api.groq.com/openai/v1 +# Gemini https://generativelanguage.googleapis.com/v1beta/openai +# OpenRouter https://openrouter.ai/api/v1 +# Ollama http://localhost:11434/v1 (no key needed) +MODEL_BASE_URL= + +# The credential. MODEL_API_KEY is the provider-neutral name and wins; +# ANTHROPIC_API_KEY still works so no existing deployment needs an edit. +# Either may be empty outside production: migrations, seeding and every # endpoint that is not an agent run work without one, and an agent run fails # with a structured `gateway.not_configured` rather than the service refusing -# to boot. APP_ENV=production requires it. +# to boot. APP_ENV=production requires one — unless the model is on localhost, +# which needs no credential at all. +MODEL_API_KEY= ANTHROPIC_API_KEY= # All three tiers default to the same model. They differ by *effort*, which the @@ -82,6 +106,13 @@ MODEL_DEEP=claude-opus-5 # that spans every call in a run and belongs to the runtime. MODEL_MAX_OUTPUT_TOKENS=16000 +# Send the tier's effort level as `reasoning_effort` on the openai-compatible +# wire. OFF by default and it should stay off unless every model named above is +# a reasoning model: the others reject the entire request rather than ignoring +# an unknown key, so turning this on for a non-reasoning model breaks every run +# with a 400. Ignored by the anthropic provider, which always sends effort. +MODEL_REASONING_EFFORT=false + # ── Knowledge layer (retrieval) ───────────────────────────────────────────── # # The dense half of hybrid retrieval needs an embedding model. Three options, diff --git a/CLAUDE.md b/CLAUDE.md index 5dc846d..cf5f544 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -224,7 +224,7 @@ depends on the curated-versus-self-serve decision and is not settled. | Registry | 9 agents + 23 skills as rows; published versions immutable (append-only, trigger-enforced); runs pin the version they started with | | Tools | 19, two of which write (`move_application`, `assign_worker`), behind a bound single-use confirmation | | Knowledge | ACL-tagged ingest, hybrid dense + BM25 fused with RRF, pre-filtered | -| Gateway | tier → model + effort, token accounting, refusal as an outcome | +| Gateway | tier → model + effort, token accounting, refusal as an outcome; two providers behind one interface — `anthropic`, and `openai` for the chat-completions shape that Groq, Gemini, OpenRouter, vLLM and a local Ollama all serve | **Deviations from this document, all deliberate and all flagged in code:** @@ -250,7 +250,13 @@ depends on the curated-versus-self-serve decision and is not settled. Do not resolve these unilaterally. Flag them and ask. - **Who authors agents?** Curated (the team ships specs) vs. self-serve (tenants author their own). Self-serve requires prompt-injection hardening at the authoring boundary, per-tenant cost caps, an approval workflow, and a sandbox — roughly 3× the platform. Current assumption: **curated**, with the registry designed so self-serve is additive later. -- **Model hosting.** Self-hosted vs. API vs. mixed by tier. +- **Model hosting.** Self-hosted vs. API vs. mixed by tier. **Still open** — + but no longer expensive to change: `MODEL_PROVIDER` + `MODEL_BASE_URL` move + the whole platform between Anthropic, Groq, Gemini, OpenRouter and a local + Ollama without a code change, and `make eval-live` runs the suite against + whichever is configured. Decide it on the eval evidence, and weigh the I7 + case heaviest: a cheaper model that follows the planted injection is a + security regression, not a saving. - **Confirmation UX.** Inline in-chat vs. an approval queue. --- diff --git a/Makefile b/Makefile index e0cddef..ab952c7 100644 --- a/Makefile +++ b/Makefile @@ -98,10 +98,16 @@ check-agents: ## Parse every spec in agents/ and report, writing nothing cd go-api && go run ./cmd/importagents --dir ../agents --skills ../skills --org check --dry-run .PHONY: eval-live -eval-live: ## Run the eval suites against the REAL model (needs ANTHROPIC_API_KEY, costs tokens) - @test -n "$$ANTHROPIC_API_KEY" || { \ - echo "eval-live needs ANTHROPIC_API_KEY — it calls the real model and costs tokens."; \ - echo "The scripted suites (make eval) are the gate; this is the confirmation."; exit 1; } +eval-live: ## Run the eval suites against the REAL model (needs a key, costs tokens) + @test -n "$$MODEL_API_KEY" -o -n "$$ANTHROPIC_API_KEY" || { \ + echo "eval-live needs MODEL_API_KEY (or ANTHROPIC_API_KEY) — it calls a real model and costs tokens."; \ + echo "The scripted suites (make eval) are the gate; this is the confirmation."; \ + echo ""; \ + echo "To evaluate a different provider, point it somewhere else:"; \ + echo " MODEL_PROVIDER=openai \\"; \ + echo " MODEL_BASE_URL=https://api.groq.com/openai/v1 \\"; \ + echo " MODEL_API_KEY=... MODEL_BALANCED= make eval-live"; \ + exit 1; } cd go-api && go test ./internal/evals/ -run "TestLive" -v -count=1 -timeout 10m .PHONY: ingest diff --git a/docs/handover.md b/docs/handover.md index 9573d16..4e4f645 100644 --- a/docs/handover.md +++ b/docs/handover.md @@ -182,6 +182,62 @@ configmap before shipping an image that contains the check. --- +## Changing model provider + +The gateway speaks two wire protocols. `anthropic` is the Claude API. +`openai` is the chat-completions shape — and that one is not only OpenAI: +Groq, Gemini's compatibility endpoint, OpenRouter, Together, vLLM and a local +Ollama all serve it, so moving between them is configuration, not code. + +```bash +# Groq +MODEL_PROVIDER=openai +MODEL_BASE_URL=https://api.groq.com/openai/v1 +MODEL_API_KEY= +MODEL_FAST=llama-3.1-8b-instant +MODEL_BALANCED=openai/gpt-oss-120b +MODEL_DEEP=openai/gpt-oss-120b + +# Gemini +MODEL_BASE_URL=https://generativelanguage.googleapis.com/v1beta/openai + +# A model on this machine — no credential at all +MODEL_BASE_URL=http://localhost:11434/v1 +``` + +Four things worth knowing before you do it. + +**Set `MODEL_PROVIDER`, not just the base URL.** The anthropic path has one +endpoint and ignores `MODEL_BASE_URL` entirely, so setting the URL alone is a +deployment that believes it has switched providers and has not — every run +still goes to Anthropic and is still billed there. Config validation refuses +that combination at startup rather than letting it run up a bill quietly. + +**Leave `MODEL_REASONING_EFFORT` off unless every configured model is a +reasoning model.** Reasoning models accept the field; most others reject the +*entire request* with a 400 rather than ignoring an unknown key. + +**Run the evals before trusting it, and read the I7 case first.** + +```bash +MODEL_PROVIDER=openai MODEL_BASE_URL=… MODEL_API_KEY=… MODEL_BALANCED=… make eval-live +``` + +`liveGateway` reads the same environment the service does and logs which +provider and model answered. The handbook corpus contains a planted prompt +injection; Claude refuses it and reports the document as tampered with. **A +model that answers every other case well and follows that injection is not a +cheaper option — it is a security regression.** That case is the gate, not the +cost table. + +**Token accounting differs between the two wires and is already reconciled.** +OpenAI reports `prompt_tokens` *inclusive* of the cached prefix; Anthropic +reports input tokens *exclusive* of it. `oaiUsage.normalise` subtracts, because +`Usage.Total()` sums all four fields and copying both numbers across verbatim +would bill the cached prefix twice — worst on long conversations, which is +exactly where I3's budget matters most. Don't "simplify" that subtraction away; +there is a test named after it. + ## Still outstanding - `ANTHROPIC_API_KEY` was pasted into a chat transcript and is live in a diff --git a/go-api/internal/config/config.go b/go-api/internal/config/config.go index 945a3df..225ec42 100644 --- a/go-api/internal/config/config.go +++ b/go-api/internal/config/config.go @@ -88,11 +88,30 @@ type KnowledgeConfig struct { // first model call, as a structured gateway.not_configured a run can end with, // not at startup as a refusal to boot. type ModelConfig struct { - APIKey string + // Provider names the wire protocol: "anthropic" or "openai". Empty means + // anthropic, so a deployment that predates the second provider keeps + // working with the environment it already has. + // + // "openai" is not only OpenAI. Groq, Gemini's compatibility endpoint, + // OpenRouter, Together, vLLM and a local Ollama all serve that same shape, + // and BaseURL is what chooses between them. + Provider string + + APIKey string + + // BaseURL points the OpenAI-compatible provider at a specific service. + // Ignored by the anthropic provider, which has one endpoint. + BaseURL string + Fast string Balanced string Deep string MaxOutputTokens int + + // ReasoningEffort opts into sending the tier's effort level on the + // OpenAI-compatible wire. Off by default: reasoning models accept the + // field and most others reject the entire request rather than ignoring it. + ReasoningEffort bool } // SeedConfig locates the demo fixture. The file is generated from the frontend @@ -245,7 +264,14 @@ func Load() (*Config, error) { UseLexicalEmbedder: boolDefault("EMBED_USE_LEXICAL", false), }, Model: ModelConfig{ - APIKey: strings.TrimSpace(os.Getenv("ANTHROPIC_API_KEY")), + Provider: strings.ToLower(strings.TrimSpace(os.Getenv("MODEL_PROVIDER"))), + // MODEL_API_KEY first, then the Anthropic-specific name. Two + // spellings because the second provider is not Anthropic and + // ANTHROPIC_API_KEY= would be a lie an operator has to + // keep re-reading; the fallback keeps every existing deployment + // working without an edit. + APIKey: firstSet("MODEL_API_KEY", "ANTHROPIC_API_KEY"), + BaseURL: strings.TrimSpace(os.Getenv("MODEL_BASE_URL")), Fast: withDefault("MODEL_FAST", defaultModel), Balanced: withDefault("MODEL_BALANCED", defaultModel), Deep: withDefault("MODEL_DEEP", defaultModel), @@ -254,6 +280,7 @@ func Load() (*Config, error) { // needs a long answer; this is the ceiling for a single // unstreamed call, not the run's budget. MaxOutputTokens: intDefault("MODEL_MAX_OUTPUT_TOKENS", 16000), + ReasoningEffort: boolDefault("MODEL_REASONING_EFFORT", false), }, DB: DBConfig{ Host: required("DATABASE_HOST"), @@ -306,6 +333,45 @@ const DeepestAgentDeadline = 120 * time.Second // Streaming hides it, and that is the trap. The chat panel uses SSE and // survives, so the product looks healthy while every non-streaming caller — a // webhook, a script, an integration — gets 502 on a slow question. +// validateModel refuses a model configuration that cannot work. +// +// Its own method for the same reason validateWriteTimeout is: these are the +// mistakes that produce a *runtime* symptom far from their cause — a deployment +// that believes it switched providers and is still being billed by the old one, +// or a production install with no credential that fails one run at a time +// instead of once at startup. +func (c *Config) validateModel() error { + switch c.Model.Provider { + case "", "anthropic", "openai": + default: + return fmt.Errorf("MODEL_PROVIDER must be anthropic or openai, got %q", c.Model.Provider) + } + // A local model needs no credential, and demanding one would make the + // zero-cost development path impossible to configure. Everything else does: + // a production deployment without a key fails every run at the gateway, + // which is a misconfiguration wearing a runtime error's clothes. + if c.AppEnv == "production" && c.Model.APIKey == "" && !isLoopback(c.Model.BaseURL) { + return fmt.Errorf("MODEL_API_KEY (or ANTHROPIC_API_KEY) is required when APP_ENV=production; " + + "without it every agent run fails at the model gateway") + } + // A base URL is only read by the OpenAI-compatible provider. Setting one + // while on anthropic is a deployment that believes it has switched + // providers and has not — it would keep calling Claude and keep being + // billed for it, with nothing in the logs to say so. + if c.Model.BaseURL != "" && c.Model.Provider != "openai" { + return fmt.Errorf("MODEL_BASE_URL only applies when MODEL_PROVIDER=openai; "+ + "it is set to %q but the provider is %q, so the base URL would be ignored "+ + "and every run would still go to Anthropic", c.Model.BaseURL, providerName(c.Model.Provider)) + } + if c.Model.BaseURL != "" { + u, err := url.Parse(c.Model.BaseURL) + if err != nil || (u.Scheme != "http" && u.Scheme != "https") || u.Host == "" { + return fmt.Errorf("MODEL_BASE_URL must be an http or https URL, got %q", c.Model.BaseURL) + } + } + return nil +} + func (c *Config) validateWriteTimeout() error { if c.HTTP.WriteTimeout <= 0 { return nil // no deadline set; the server will not cut anything off @@ -355,9 +421,8 @@ func (c *Config) validate() error { // misconfiguration wearing a runtime error's clothes, so it is caught here. // Development is left alone deliberately: working on migrations or the // definitions API must not require a key. - if c.AppEnv == "production" && c.Model.APIKey == "" { - return fmt.Errorf("ANTHROPIC_API_KEY is required when APP_ENV=production; " + - "without it every agent run fails at the model gateway") + if err := c.validateModel(); err != nil { + return err } if c.Model.MaxOutputTokens < 1 { return fmt.Errorf("MODEL_MAX_OUTPUT_TOKENS must be at least 1, got %d", c.Model.MaxOutputTokens) @@ -482,6 +547,46 @@ func withDefault(key, fallback string) string { return fallback } +// firstSet returns the first of several environment variables that has a value. +// +// For settings that have more than one legitimate spelling — a generic name and +// a provider-specific one — where the order expresses which wins rather than +// leaving it to whichever happens to be read last. +func firstSet(keys ...string) string { + for _, k := range keys { + if v := strings.TrimSpace(os.Getenv(k)); v != "" { + return v + } + } + return "" +} + +// isLoopback reports whether a base URL points at this machine. +// +// A model served from localhost needs no credential, and requiring one would +// make the zero-cost local path impossible to configure. Host-only, so a +// remote service that merely mentions "localhost" in a path does not qualify. +func isLoopback(raw string) bool { + if strings.TrimSpace(raw) == "" { + return false + } + u, err := url.Parse(raw) + if err != nil { + return false + } + host := u.Hostname() + return host == "localhost" || host == "127.0.0.1" || host == "::1" +} + +// providerName renders the provider for an error message, naming the default +// rather than showing an empty string an operator then has to interpret. +func providerName(p string) string { + if p == "" { + return "anthropic (the default)" + } + return p +} + func intDefault(key string, fallback int) int { v := strings.TrimSpace(os.Getenv(key)) if v == "" { diff --git a/go-api/internal/config/model_test.go b/go-api/internal/config/model_test.go new file mode 100644 index 0000000..70bf055 --- /dev/null +++ b/go-api/internal/config/model_test.go @@ -0,0 +1,128 @@ +package config + +import ( + "strings" + "testing" +) + +func modelCfg(env string, m ModelConfig) *Config { + c := &Config{AppEnv: env} + c.Model = m + return c +} + +func TestValidateModelProvider(t *testing.T) { + for _, tc := range []struct { + name string + cfg *Config + wantErr bool + }{ + { + "unset provider is anthropic, which is what every existing deployment has", + modelCfg("development", ModelConfig{}), false, + }, + {"anthropic named explicitly", modelCfg("development", ModelConfig{Provider: "anthropic"}), false}, + {"openai", modelCfg("development", ModelConfig{Provider: "openai"}), false}, + {"a typo is caught once at startup, not once per run", + modelCfg("development", ModelConfig{Provider: "openal"}), true}, + {"a provider that does not exist", modelCfg("development", ModelConfig{Provider: "groq"}), true}, + } { + t.Run(tc.name, func(t *testing.T) { + err := tc.cfg.validateModel() + if tc.wantErr != (err != nil) { + t.Fatalf("validateModel() = %v, wantErr = %v", err, tc.wantErr) + } + }) + } +} + +// THE EXPENSIVE MISTAKE. +// +// A deployment that sets MODEL_BASE_URL and forgets MODEL_PROVIDER believes it +// has moved off Claude. It has not: the anthropic path has one endpoint and +// ignores the field entirely, so every run keeps going to Anthropic and keeps +// being billed there, with nothing in the logs to say so. The whole point of +// this change is cost, and that is the one misconfiguration that silently +// defeats it. +func TestBaseURLWithoutOpenAIProviderIsRefused(t *testing.T) { + err := modelCfg("development", ModelConfig{ + BaseURL: "https://api.groq.com/openai/v1", + }).validateModel() + if err == nil { + t.Fatal("a base URL on the anthropic provider was accepted; every run would still go to Anthropic") + } + for _, want := range []string{"MODEL_BASE_URL", "MODEL_PROVIDER=openai", "Anthropic"} { + if !strings.Contains(err.Error(), want) { + t.Errorf("the message does not mention %q:\n %v", want, err) + } + } + + // The same URL with the provider set is exactly the intended configuration. + if err := modelCfg("development", ModelConfig{ + Provider: "openai", BaseURL: "https://api.groq.com/openai/v1", + }).validateModel(); err != nil { + t.Fatalf("the intended configuration was refused: %v", err) + } +} + +func TestBaseURLMustBeAURL(t *testing.T) { + for _, raw := range []string{"api.groq.com", "ftp://x.test", "not a url", "://broken"} { + err := modelCfg("development", ModelConfig{Provider: "openai", BaseURL: raw}).validateModel() + if err == nil { + t.Errorf("MODEL_BASE_URL=%q was accepted", raw) + } + } + for _, raw := range []string{"http://localhost:11434/v1", "https://api.groq.com/openai/v1"} { + if err := modelCfg("development", ModelConfig{Provider: "openai", BaseURL: raw}).validateModel(); err != nil { + t.Errorf("MODEL_BASE_URL=%q was refused: %v", raw, err) + } + } +} + +// Production without a credential fails every run at the gateway, which is a +// misconfiguration wearing a runtime error's clothes. A local model is the one +// exception: it needs no key, and demanding one would make the zero-cost path +// impossible to configure. +func TestProductionCredentialRequirement(t *testing.T) { + for _, tc := range []struct { + name string + cfg *Config + wantErr bool + }{ + {"production with no key", modelCfg("production", ModelConfig{}), true}, + {"production with a key", modelCfg("production", ModelConfig{APIKey: "k"}), false}, + { + "production against a local model needs no key", + modelCfg("production", ModelConfig{Provider: "openai", BaseURL: "http://localhost:11434/v1"}), + false, + }, + { + "production against a hosted provider still does", + modelCfg("production", ModelConfig{Provider: "openai", BaseURL: "https://api.groq.com/openai/v1"}), + true, + }, + {"development needs nothing", modelCfg("development", ModelConfig{}), false}, + } { + t.Run(tc.name, func(t *testing.T) { + err := tc.cfg.validateModel() + if tc.wantErr != (err != nil) { + t.Fatalf("validateModel() = %v, wantErr = %v", err, tc.wantErr) + } + }) + } +} + +func TestIsLoopback(t *testing.T) { + for raw, want := range map[string]bool{ + "http://localhost:11434/v1": true, + "http://127.0.0.1:11434/v1": true, + "https://api.groq.com/v1": false, + "": false, + // A remote host that merely mentions localhost in its path is not local. + "https://x.test/localhost/v1": false, + } { + if got := isLoopback(raw); got != want { + t.Errorf("isLoopback(%q) = %v, want %v", raw, got, want) + } + } +} diff --git a/go-api/internal/evals/live_test.go b/go-api/internal/evals/live_test.go index 31645a9..536c431 100644 --- a/go-api/internal/evals/live_test.go +++ b/go-api/internal/evals/live_test.go @@ -28,19 +28,73 @@ import ( // // Run with: make eval-live +// liveGateway builds the gateway this run is being evaluated against. +// +// PROVIDER-DRIVEN, and that is the point. These cases are the only evidence +// that answers the question a scripted model cannot — whether a real one, given +// these tools and this prompt, actually does the right thing — and that +// question has a different answer for every provider. A helper hardcoded to +// Anthropic could confirm the model this platform already runs and nothing +// else, which is exactly the comparison worth having before changing it. +// +// So the same environment the service reads selects the model here: +// +// MODEL_PROVIDER=openai MODEL_BASE_URL=https://api.groq.com/openai/v1 \ +// MODEL_API_KEY=… MODEL_FAST=… MODEL_BALANCED=… MODEL_DEEP=… make eval-live +// +// The I7 case is the one to watch when comparing. A model that answers the +// other cases well and follows the planted injection is not a cheaper option, +// it is a security regression. func liveGateway(t *testing.T) gateway.Gateway { t.Helper() - key := strings.TrimSpace(os.Getenv("ANTHROPIC_API_KEY")) + + key := strings.TrimSpace(os.Getenv("MODEL_API_KEY")) if key == "" { - t.Skip("no ANTHROPIC_API_KEY; the live suite is skipped") + key = strings.TrimSpace(os.Getenv("ANTHROPIC_API_KEY")) } - return gateway.NewAnthropic(gateway.FromConfig(config.ModelConfig{ + baseURL := strings.TrimSpace(os.Getenv("MODEL_BASE_URL")) + provider := strings.ToLower(strings.TrimSpace(os.Getenv("MODEL_PROVIDER"))) + + // A local model needs no credential; everything else does. Skipping rather + // than failing keeps `go test ./...` green on a machine with no key, which + // is what makes the scripted suites the gate. + if key == "" && !strings.Contains(baseURL, "localhost") && !strings.Contains(baseURL, "127.0.0.1") { + t.Skip("no MODEL_API_KEY or ANTHROPIC_API_KEY; the live suite is skipped") + } + + model := func(env, fallback string) string { + if v := strings.TrimSpace(os.Getenv(env)); v != "" { + return v + } + return fallback + } + // The default stays Claude, so an existing invocation of `make eval-live` + // runs exactly what it ran before this became configurable. + fallback := "claude-opus-5" + + cfg := config.ModelConfig{ + Provider: provider, APIKey: key, - Fast: "claude-opus-5", - Balanced: "claude-opus-5", - Deep: "claude-opus-5", + BaseURL: baseURL, + Fast: model("MODEL_FAST", fallback), + Balanced: model("MODEL_BALANCED", fallback), + Deep: model("MODEL_DEEP", fallback), MaxOutputTokens: 4096, - })) + ReasoningEffort: strings.EqualFold(strings.TrimSpace(os.Getenv("MODEL_REASONING_EFFORT")), "true"), + } + + // Named in the output, because a suite that does not say which model + // answered is a suite whose result cannot be compared with another run's. + t.Logf("live gateway: provider=%s model=%s", providerLabel(provider), cfg.Balanced) + + return gateway.New(gateway.FromConfig(cfg)) +} + +func providerLabel(p string) string { + if p == "" { + return "anthropic" + } + return p } // TestLiveActivityAgentAnswersFromRealData. diff --git a/go-api/internal/gateway/anthropic.go b/go-api/internal/gateway/anthropic.go index e28f889..872450d 100644 --- a/go-api/internal/gateway/anthropic.go +++ b/go-api/internal/gateway/anthropic.go @@ -12,34 +12,6 @@ import ( "github.com/anthropics/anthropic-sdk-go/option" ) -// Routing is how a tier becomes a model and an effort level. -// -// The model per tier is a deployment knob — a tenant on a different contract, -// or a deployment pinning a version through an incident, changes it without a -// spec edit. The *effort* per tier is not: "fast" and "deep" mean something -// specific about how much work an answer is worth, and letting a deployment -// redefine that would make the same spec behave differently in two places -// while claiming the same tier. -type Routing struct { - Model string - Effort anthropic.OutputConfigEffort -} - -// Config is the gateway's whole configuration surface. -// -// Built once at startup from the environment and passed in frozen, per §10. -// Nothing in this package reads the environment itself. -type Config struct { - APIKey string - - Fast Routing - Balanced Routing - Deep Routing - - // MaxOutputTokens applies when a request does not set its own. - MaxOutputTokens int64 -} - // AnthropicGateway calls the Claude API. type AnthropicGateway struct { client anthropic.Client @@ -64,16 +36,23 @@ func NewAnthropic(cfg Config) *AnthropicGateway { return &AnthropicGateway{client: anthropic.NewClient(opts...), cfg: cfg} } -// routing resolves a tier. An unknown tier has already been normalised by -// ParseTier, so the default arm is reached only by a zero value. -func (g *AnthropicGateway) routing(t Tier) Routing { - switch t { - case TierFast: - return g.cfg.Fast - case TierDeep: - return g.cfg.Deep +// routing resolves a tier against this gateway's table. +func (g *AnthropicGateway) routing(t Tier) Routing { return g.cfg.routingFor(t) } + +// sdkEffort maps the platform's effort vocabulary onto Anthropic's. +// +// A one-to-one mapping today, which is exactly why the neutral type is worth +// having: the platform's three levels are a statement about how much a turn is +// worth, and this function is where that statement meets one vendor's spelling +// of it. `max` is not reachable — see FromConfig. +func sdkEffort(e Effort) anthropic.OutputConfigEffort { + switch e { + case EffortLow: + return anthropic.OutputConfigEffortLow + case EffortXhigh: + return anthropic.OutputConfigEffortXhigh default: - return g.cfg.Balanced + return anthropic.OutputConfigEffortHigh } } @@ -108,6 +87,18 @@ var retryBackoff = []time.Duration{400 * time.Millisecond, 1200 * time.Milliseco // does not happen — the deadline belongs to the run, not to this function, and // waiting past it would turn a bounded run into an unbounded one. func (g *AnthropicGateway) Complete(ctx context.Context, req Request) (*Response, error) { + return withRetry(ctx, func() (*Response, error) { return g.complete(ctx, req) }) +} + +// withRetry runs one attempt until it succeeds, fails terminally, or runs out +// of attempts. +// +// SHARED BY EVERY PROVIDER, and it has to be. The retry policy is a property of +// this platform's runs — bounded attempts, short backoff, the caller's deadline +// winning — not of any one vendor's API. Left as a method, the second provider +// would have grown its own copy, and the two would have drifted the first time +// either was tuned. +func withRetry(ctx context.Context, once func() (*Response, error)) (*Response, error) { var last error for attempt := 0; attempt < MaxAttempts; attempt++ { if attempt > 0 { @@ -122,7 +113,7 @@ func (g *AnthropicGateway) Complete(ctx context.Context, req Request) (*Response } } - resp, err := g.complete(ctx, req) + resp, err := once() if err == nil { return resp, nil } @@ -191,7 +182,7 @@ func (g *AnthropicGateway) params(req Request) (anthropic.MessageNewParams, erro Thinking: anthropic.ThinkingConfigParamUnion{ OfAdaptive: &anthropic.ThinkingConfigAdaptiveParam{}, }, - OutputConfig: anthropic.OutputConfigParam{Effort: route.Effort}, + OutputConfig: anthropic.OutputConfigParam{Effort: sdkEffort(route.Effort)}, } if len(req.Tools) > 0 { @@ -233,15 +224,15 @@ func (g *AnthropicGateway) decode(msg *anthropic.Message, req Request) (*Respons // would happily repeat. if msg.StopReason == anthropic.StopReasonRefusal { return &Response{ - StopReason: string(msg.StopReason), - Usage: usage, - Model: route.Model, - Tier: req.Tier, - }, &Error{ - Code: CodeRefused, - Message: "the model declined this request", - Category: string(msg.StopDetails.Category), - } + StopReason: string(msg.StopReason), + Usage: usage, + Model: route.Model, + Tier: req.Tier, + }, &Error{ + Code: CodeRefused, + Message: "the model declined this request", + Category: string(msg.StopDetails.Category), + } } var ( diff --git a/go-api/internal/gateway/gateway_test.go b/go-api/internal/gateway/gateway_test.go index 0fda17a..3dc3877 100644 --- a/go-api/internal/gateway/gateway_test.go +++ b/go-api/internal/gateway/gateway_test.go @@ -113,13 +113,13 @@ func TestFromConfigPinsEffortPerTier(t *testing.T) { MaxOutputTokens: 8000, }) - if cfg.Fast.Effort != anthropic.OutputConfigEffortLow { + if cfg.Fast.Effort != EffortLow { t.Errorf("fast effort = %q, want low", cfg.Fast.Effort) } - if cfg.Balanced.Effort != anthropic.OutputConfigEffortHigh { + if cfg.Balanced.Effort != EffortHigh { t.Errorf("balanced effort = %q, want high", cfg.Balanced.Effort) } - if cfg.Deep.Effort != anthropic.OutputConfigEffortXhigh { + if cfg.Deep.Effort != EffortXhigh { t.Errorf("deep effort = %q, want xhigh", cfg.Deep.Effort) } if cfg.MaxOutputTokens != 8000 { @@ -127,6 +127,23 @@ func TestFromConfigPinsEffortPerTier(t *testing.T) { } } +// The neutral effort vocabulary has to land on the vendor's own enum, and that +// mapping is the one thing FromConfig can no longer assert now that its result +// is provider-independent. Untested, a renamed SDK constant would silently +// route every tier to whatever the default arm returns. +func TestSDKEffortMapsToAnthropic(t *testing.T) { + cases := map[Effort]anthropic.OutputConfigEffort{ + EffortLow: anthropic.OutputConfigEffortLow, + EffortHigh: anthropic.OutputConfigEffortHigh, + EffortXhigh: anthropic.OutputConfigEffortXhigh, + } + for neutral, want := range cases { + if got := sdkEffort(neutral); got != want { + t.Errorf("sdkEffort(%q) = %q, want %q", neutral, got, want) + } + } +} + func TestRoutingSelectsPerTier(t *testing.T) { g := NewAnthropic(Config{ Fast: Routing{Model: "m-fast"}, diff --git a/go-api/internal/gateway/openai.go b/go-api/internal/gateway/openai.go new file mode 100644 index 0000000..2df7e32 --- /dev/null +++ b/go-api/internal/gateway/openai.go @@ -0,0 +1,750 @@ +package gateway + +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "strings" + "time" +) + +// OpenAIGateway calls any service that speaks the OpenAI chat-completions API. +// +// ONE IMPLEMENTATION, MANY PROVIDERS. Groq, Gemini (through its compatibility +// endpoint), OpenRouter, Together, vLLM and a local Ollama all serve this same +// shape, so the difference between them is a base URL and a model id — not a +// package each. That is the whole reason this file exists: the platform needed +// a way off a single vendor's pricing without a rewrite per alternative. +// +// Hand-rolled over net/http rather than an SDK, per §10. The surface actually +// used here is one endpoint and one event stream; a dependency for that buys a +// version to keep current and a second opinion about retries, and this package +// already has its own. +type OpenAIGateway struct { + cfg Config + http *http.Client +} + +// Compile-time proof that this satisfies the boundary and can stream. +var ( + _ Gateway = (*OpenAIGateway)(nil) + _ Streamer = (*OpenAIGateway)(nil) +) + +// DefaultOpenAIBaseURL is where an unconfigured deployment points. +const DefaultOpenAIBaseURL = "https://api.openai.com/v1" + +// openAIHTTPTimeout bounds a single call at the transport. +// +// Above the deepest tier's deadline on purpose. The run's own context is what +// should end a slow call — that failure is a Deadline the runtime can report +// against a budget — and a transport timeout firing first would present the +// same event as an unexplained upstream error instead. +const openAIHTTPTimeout = 10 * time.Minute + +// NewOpenAI builds a gateway over an OpenAI-compatible service. +// +// A missing key is not an error here, for the same reason it is not one for +// Anthropic: the service has to boot without model credentials, and the +// failure belongs at the first Complete as a structured NotConfigured a run +// can end with. A local Ollama legitimately needs no key at all, which is why +// the check is deferred rather than dropped — see complete(). +func NewOpenAI(cfg Config) *OpenAIGateway { + return &OpenAIGateway{cfg: cfg, http: &http.Client{Timeout: openAIHTTPTimeout}} +} + +// endpoint is the chat-completions URL for this deployment. +func (g *OpenAIGateway) endpoint() string { + base := strings.TrimRight(strings.TrimSpace(g.cfg.BaseURL), "/") + if base == "" { + base = DefaultOpenAIBaseURL + } + return base + "/chat/completions" +} + +// routing resolves a tier against this gateway's table. +func (g *OpenAIGateway) routing(t Tier) Routing { return g.cfg.routingFor(t) } + +// needsCredential reports whether this deployment must present a key. +// +// A hosted provider does; a local Ollama does not, and demanding one would +// make the zero-cost development path impossible to configure. The base URL is +// the only signal available — a deployment that has pointed this at its own +// machine has already said the call is not leaving it. +func (g *OpenAIGateway) needsCredential() bool { + base := strings.TrimSpace(g.cfg.BaseURL) + if base == "" { + return true + } + return !strings.Contains(base, "localhost") && !strings.Contains(base, "127.0.0.1") +} + +// Complete calls the model, retrying failures that are worth retrying. +// +// Same policy as every other provider — see withRetry, which is shared +// precisely so the two cannot drift. +func (g *OpenAIGateway) Complete(ctx context.Context, req Request) (*Response, error) { + return withRetry(ctx, func() (*Response, error) { return g.complete(ctx, req) }) +} + +// complete is one attempt. +func (g *OpenAIGateway) complete(ctx context.Context, req Request) (*Response, error) { + body, err := g.params(req, false) + if err != nil { + return nil, err + } + + httpResp, err := g.post(ctx, body) + if err != nil { + return nil, err + } + defer httpResp.Body.Close() + + raw, err := io.ReadAll(httpResp.Body) + if err != nil { + return nil, &Error{Code: CodeUpstream, Message: "the model response could not be read", Cause: err} + } + if httpResp.StatusCode >= 400 { + return nil, translateOpenAI(httpResp.StatusCode, raw) + } + + var decoded oaiResponse + if err := json.Unmarshal(raw, &decoded); err != nil { + return nil, &Error{ + Code: CodeUpstream, + Message: "the model returned a response this gateway could not parse", + Cause: err, + } + } + if len(decoded.Choices) == 0 { + return nil, &Error{Code: CodeUpstream, Message: "the model returned no choices"} + } + + choice := decoded.Choices[0] + return g.decode(req, decoded.Model, choice.FinishReason, choice.Message, decoded.Usage) +} + +// params builds the request body both paths send. +// +// Extracted for the same reason the Anthropic path extracts its own: an answer +// that differed depending on whether it was streamed would be the worst kind of +// bug to chase, because the transport is the last place anybody looks. +func (g *OpenAIGateway) params(req Request, stream bool) (*oaiRequest, error) { + if err := req.Validate(); err != nil { + return nil, err + } + route := g.routing(req.Tier) + + maxTokens := req.MaxOutputTokens + if maxTokens <= 0 { + maxTokens = g.cfg.MaxOutputTokens + } + + body := &oaiRequest{ + Model: route.Model, + Messages: encodeOpenAIMessages(req.System, req.Messages), + MaxTokens: maxTokens, + Tools: encodeOpenAITools(req.Tools), + } + if g.cfg.SendReasoningEffort { + body.ReasoningEffort = openAIEffort(route.Effort) + } + if stream { + body.Stream = true + // Usage is omitted from a stream unless it is asked for, and a call + // whose cost is unknown is a call the run's budget cannot be charged + // for. I3 needs every call measured, so this is not optional. + body.StreamOptions = &oaiStreamOptions{IncludeUsage: true} + } + return body, nil +} + +// post sends the request body. +func (g *OpenAIGateway) post(ctx context.Context, body *oaiRequest) (*http.Response, error) { + if g.cfg.APIKey == "" && g.needsCredential() { + return nil, &Error{ + Code: CodeNotConfigured, + Message: "no model credentials are configured for this deployment", + } + } + + encoded, err := json.Marshal(body) + if err != nil { + return nil, &Error{Code: CodeInvalidRequest, Message: "the request could not be encoded", Cause: err} + } + + httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, g.endpoint(), bytes.NewReader(encoded)) + if err != nil { + return nil, &Error{Code: CodeInvalidRequest, Message: "the request could not be built", Cause: err} + } + httpReq.Header.Set("Content-Type", "application/json") + if g.cfg.APIKey != "" { + httpReq.Header.Set("Authorization", "Bearer "+g.cfg.APIKey) + } + + resp, err := g.http.Do(httpReq) + if err != nil { + if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled) { + return nil, &Error{Code: CodeTimeout, Message: "the model call did not complete in time", Cause: err} + } + return nil, &Error{Code: CodeUpstream, Message: "the model call failed", Cause: err} + } + return resp, nil +} + +// decode turns a finished choice into a Response. +// +// Shared by both paths, so a streamed answer and a non-streamed one are read +// by the same code rather than by two implementations of the same reading. +func (g *OpenAIGateway) decode( + req Request, model, finish string, msg oaiMessage, usage oaiUsage, +) (*Response, error) { + route := g.routing(req.Tier) + if model == "" { + model = route.Model + } + + counted := usage.normalise() + + // A refusal arrives as a successful HTTP response, so it is checked before + // the content is read. It is still billed, and the usage rides on the + // Response rather than being dropped — a refusal that cost nothing on the + // ledger is a refusal the loop would happily repeat. + if refusal := strings.TrimSpace(msg.Refusal); refusal != "" || finish == "content_filter" { + category := finish + if refusal != "" { + category = "refusal" + } + return &Response{ + StopReason: openAIStopReason(finish), + Usage: counted, + Model: model, + Tier: req.Tier, + }, &Error{ + Code: CodeRefused, + Message: "the model declined this request", + Category: category, + } + } + + var calls []ToolCall + for _, c := range msg.ToolCalls { + args := strings.TrimSpace(c.Function.Arguments) + if args == "" { + // An argumentless call is legitimate; an empty string is not valid + // JSON, and the handler's decoder would reject it for a reason that + // has nothing to do with the caller's request. + args = "{}" + } + calls = append(calls, ToolCall{ + ID: c.ID, + Name: c.Function.Name, + // The raw JSON, not a parsed value — handed to the handler's own + // decoder rather than matched on as a string here. + Input: json.RawMessage(args), + }) + } + + return &Response{ + Text: msg.Content, + ToolCalls: calls, + StopReason: openAIStopReason(finish), + Usage: counted, + Model: model, + Tier: req.Tier, + }, nil +} + +/* ── Wire types ─────────────────────────────────────────────────────────── */ + +type oaiRequest struct { + Model string `json:"model"` + Messages []oaiMessage `json:"messages"` + Tools []oaiTool `json:"tools,omitempty"` + MaxTokens int64 `json:"max_tokens,omitempty"` + Stream bool `json:"stream,omitempty"` + StreamOptions *oaiStreamOptions `json:"stream_options,omitempty"` + + // ReasoningEffort is omitted unless a deployment opted in. Most non- + // reasoning models reject the whole request rather than ignoring the key. + ReasoningEffort string `json:"reasoning_effort,omitempty"` +} + +type oaiStreamOptions struct { + IncludeUsage bool `json:"include_usage"` +} + +// oaiMessage is one wire message. It doubles as a streamed delta, because the +// two carry the same fields and differ only in how much of each is present. +type oaiMessage struct { + Role string `json:"role,omitempty"` + Content string `json:"content,omitempty"` + Refusal string `json:"refusal,omitempty"` + ToolCalls []oaiToolCall `json:"tool_calls,omitempty"` + + // ToolCallID is set only on a role:"tool" message, correlating a result + // with the call that asked for it. + ToolCallID string `json:"tool_call_id,omitempty"` +} + +type oaiToolCall struct { + // Index orders a call within a streamed response. Absent when complete, + // which is why it is a pointer: index 0 and "no index" are different + // things, and reading a missing field as 0 merges every streamed call + // into the first one. + Index *int `json:"index,omitempty"` + ID string `json:"id,omitempty"` + Type string `json:"type,omitempty"` + Function oaiFunctionRef `json:"function"` +} + +type oaiFunctionRef struct { + Name string `json:"name,omitempty"` + Arguments string `json:"arguments,omitempty"` +} + +type oaiTool struct { + Type string `json:"type"` + Function oaiFunctionDef `json:"function"` +} + +type oaiFunctionDef struct { + Name string `json:"name"` + Description string `json:"description,omitempty"` + Parameters map[string]any `json:"parameters,omitempty"` +} + +type oaiResponse struct { + Model string `json:"model"` + Choices []oaiChoice `json:"choices"` + Usage oaiUsage `json:"usage"` +} + +type oaiChoice struct { + Message oaiMessage `json:"message"` + Delta oaiMessage `json:"delta"` + FinishReason string `json:"finish_reason"` +} + +type oaiUsage struct { + PromptTokens int64 `json:"prompt_tokens"` + CompletionTokens int64 `json:"completion_tokens"` + PromptTokensDetails struct { + CachedTokens int64 `json:"cached_tokens"` + } `json:"prompt_tokens_details"` +} + +// normalise converts OpenAI's accounting into this platform's. +// +// THE SUBTRACTION IS THE WHOLE FUNCTION, and getting it wrong would corrupt +// every budget quietly. OpenAI reports `prompt_tokens` INCLUSIVE of the cached +// prefix; Anthropic reports input tokens EXCLUSIVE of it, and carries the cache +// separately. Usage.Total() adds all four fields, so copying both numbers +// across verbatim would bill the cached prefix twice — and it would do it +// worst on long conversations, which is exactly where a budget matters most. +// +// Clamped at zero rather than trusted: a provider that reports more cached +// tokens than prompt tokens is wrong, but a negative charge would be a bug +// that hands a run free budget rather than one that shows up as a wrong number. +func (u oaiUsage) normalise() Usage { + cached := u.PromptTokensDetails.CachedTokens + fresh := u.PromptTokens - cached + if fresh < 0 { + fresh = 0 + } + return Usage{ + InputTokens: fresh, + OutputTokens: u.CompletionTokens, + CacheReadTokens: cached, + // No creation figure on this wire. Left at zero rather than guessed: + // an invented number is worse than an absent one, because it looks + // like a measurement. + CacheCreationTokens: 0, + } +} + +/* ── Encoding ───────────────────────────────────────────────────────────── */ + +// openAIEffort maps the platform's effort vocabulary onto OpenAI's. +// +// Three of ours onto three of theirs, preserving the ordering rather than the +// spelling: their scale runs minimal/low/medium/high, so "high" here is their +// "medium" and "xhigh" is their "high". Matching the words instead of the +// positions would have made `fast` and `balanced` nearly indistinguishable. +func openAIEffort(e Effort) string { + switch e { + case EffortLow: + return "low" + case EffortXhigh: + return "high" + default: + return "medium" + } +} + +// openAIStopReason maps a finish_reason onto the vocabulary the trajectories +// already use. +// +// Translated rather than passed through, so a trajectory reads the same +// whichever provider answered. An eval comparing two providers is comparing +// the run, and it should not have to know that one says "tool_calls" where the +// other says "tool_use". +func openAIStopReason(finish string) string { + switch finish { + case "tool_calls", "function_call": + return "tool_use" + case "stop": + return "end_turn" + case "length": + return "max_tokens" + case "content_filter": + return "refusal" + default: + return finish + } +} + +// encodeOpenAITools renders the tool definitions for the wire. +// +// The whole input schema is passed through, not just its properties: this API +// validates arguments against what it is given, so dropping `type`, `enum` or +// a nested object's own required list would let the model send arguments the +// handler then has to reject. +func encodeOpenAITools(defs []ToolDef) []oaiTool { + if len(defs) == 0 { + return nil + } + out := make([]oaiTool, 0, len(defs)) + for _, d := range defs { + params := d.InputSchema + if params == nil { + params = map[string]any{"type": "object", "properties": map[string]any{}} + } else if _, ok := params["type"]; !ok { + // A schema without a type is rejected by some providers and + // silently accepted by others. Copied rather than mutated: the + // caller's map is shared across every call in a run. + cloned := make(map[string]any, len(params)+1) + for k, v := range params { + cloned[k] = v + } + cloned["type"] = "object" + params = cloned + } + out = append(out, oaiTool{ + Type: "function", + Function: oaiFunctionDef{ + Name: d.Name, + Description: d.Description, + Parameters: params, + }, + }) + } + return out +} + +// encodeOpenAIMessages renders a conversation for the wire. +// +// TWO SHAPE DIFFERENCES from the Anthropic path, and both are load-bearing: +// +// - The system prompt is a MESSAGE here, not a top-level field, and it must +// come first. +// - A tool result is its OWN message with role "tool", one per result — +// where Anthropic carries them as blocks inside a single user turn. So the +// grouping the other encoder is careful to preserve has to be undone here, +// in the same order, or a result arrives detached from its call. +// +// Ordering within a turn matters: results are emitted before any text in the +// same message, because they answer the assistant turn that preceded them. +func encodeOpenAIMessages(system string, msgs []Message) []oaiMessage { + out := make([]oaiMessage, 0, len(msgs)+1) + + if s := strings.TrimSpace(system); s != "" { + out = append(out, oaiMessage{Role: "system", Content: s}) + } + + for _, m := range msgs { + for _, r := range m.ToolResults { + // IsError has no home on this wire — there is no error flag on a + // tool message. The handler's own error payload is already in the + // content, per §4, so the model still sees what went wrong; what + // is lost is the structured marker, and inventing a prefix for it + // would put prose in a channel that carries data. + out = append(out, oaiMessage{ + Role: "tool", + ToolCallID: r.CallID, + Content: r.Content, + }) + } + + hasText := strings.TrimSpace(m.Text) != "" + if !hasText && len(m.ToolCalls) == 0 { + continue + } + + msg := oaiMessage{Role: string(m.Role), Content: m.Text} + for _, c := range m.ToolCalls { + args := strings.TrimSpace(string(c.Input)) + if args == "" { + args = "{}" + } + msg.ToolCalls = append(msg.ToolCalls, oaiToolCall{ + ID: c.ID, + Type: "function", + Function: oaiFunctionRef{Name: c.Name, Arguments: args}, + }) + } + out = append(out, msg) + } + return out +} + +/* ── Errors ─────────────────────────────────────────────────────────────── */ + +// translateOpenAI turns an HTTP failure into one the runtime can branch on. +// +// Mapped by status, mirroring the Anthropic path, because the distinction the +// loop needs is the same one either way: whether sending this request again +// could work. The upstream message is carried through when there is one — a +// 400 that says which tool schema is malformed is worth more than "the model +// rejected the request", and the trajectory only records the message. +func translateOpenAI(status int, body []byte) error { + detail := openAIErrorMessage(body) + + withDetail := func(base string) string { + if detail == "" { + return base + } + return base + ": " + detail + } + + switch { + case status == 400 || status == 404 || status == 422: + // 404 belongs here, not with the 5xx: on these providers it almost + // always means the model id does not exist on this endpoint, which is + // a configuration mistake and will fail identically next time. + return &Error{Code: CodeInvalidRequest, Message: withDetail("the model rejected the request"), Status: status} + case status == 401 || status == 403: + return &Error{Code: CodeUnauthorized, Message: withDetail("the model credentials were refused"), Status: status} + case status == 408: + return &Error{Code: CodeTimeout, Message: withDetail("the model call timed out"), Status: status} + case status == 429: + return &Error{Code: CodeRateLimited, Message: withDetail("the model is rate limiting this deployment"), Status: status} + default: + return &Error{ + Code: CodeUpstream, + Message: withDetail(fmt.Sprintf("the model call failed (http %d)", status)), + Status: status, + } + } +} + +// openAIErrorMessage digs the human-readable reason out of an error body. +// +// Best-effort by design: providers agree on the envelope often enough to be +// worth reading and not often enough to depend on, so an unparseable body +// yields nothing rather than failing a failure. +func openAIErrorMessage(body []byte) string { + var envelope struct { + Error struct { + Message string `json:"message"` + } `json:"error"` + Message string `json:"message"` + } + if err := json.Unmarshal(body, &envelope); err != nil { + return "" + } + if m := strings.TrimSpace(envelope.Error.Message); m != "" { + return m + } + return strings.TrimSpace(envelope.Message) +} + +/* ── Streaming ──────────────────────────────────────────────────────────── */ + +// maxSSELine caps a single server-sent-event line. +// +// One event carries one delta, but a tool call's arguments arrive as a single +// field that can be large, and the default scanner limit of 64KB is low enough +// to be hit by a real request. A cap is still wanted: an unbounded line from a +// misbehaving upstream would be read straight into memory. +const maxSSELine = 1 << 20 + +// Stream is Complete, with the assistant's text delivered as it arrives. +// +// §6: "Stream partial assistant text as it arrives; buffer tool calls until +// complete." Both halves matter and they pull in opposite directions. +// +// TEXT IS STREAMED because a fifteen-second wait with nothing on screen reads +// as broken. +// +// TOOL CALLS ARE NOT. On this wire a call's arguments arrive as a JSON string +// assembled across many events, and a half-built argument object is not a +// smaller version of the finished one — it is a different object, usually an +// invalid one. So the fragments are accumulated by index and decoded only once +// the stream closes, by exactly the same code the non-streaming path uses. +// +// onDelta is called from this goroutine, in order, and must not block for long +// — it is on the path between the model and the reader. +func (g *OpenAIGateway) Stream(ctx context.Context, req Request, onDelta func(string)) (*Response, error) { + body, err := g.params(req, true) + if err != nil { + return nil, err + } + + httpResp, err := g.post(ctx, body) + if err != nil { + return nil, err + } + defer httpResp.Body.Close() + + if httpResp.StatusCode >= 400 { + raw, _ := io.ReadAll(httpResp.Body) + return nil, translateOpenAI(httpResp.StatusCode, raw) + } + + acc, err := accumulateSSE(httpResp.Body, onDelta) + if err != nil { + return nil, err + } + return g.decode(req, acc.model, acc.finishReason, acc.message(), acc.usage) +} + +// streamAccumulator assembles a streamed response. +// +// Tool calls are keyed by their wire index rather than appended in arrival +// order: providers interleave the fragments of parallel calls, so arrival +// order is not call order, and appending would splice one call's arguments +// onto another's. +type streamAccumulator struct { + text strings.Builder + refusal strings.Builder + model string + finishReason string + usage oaiUsage + + calls map[int]*oaiToolCall + order []int +} + +// message renders the accumulated stream as the finished message the shared +// decoder reads. +func (a *streamAccumulator) message() oaiMessage { + msg := oaiMessage{ + Role: "assistant", + Content: a.text.String(), + Refusal: a.refusal.String(), + } + for _, idx := range a.order { + msg.ToolCalls = append(msg.ToolCalls, *a.calls[idx]) + } + return msg +} + +// accumulateSSE reads the event stream to its end. +func accumulateSSE(r io.Reader, onDelta func(string)) (*streamAccumulator, error) { + acc := &streamAccumulator{calls: map[int]*oaiToolCall{}} + + scanner := bufio.NewScanner(r) + scanner.Buffer(make([]byte, 0, 64*1024), maxSSELine) + + for scanner.Scan() { + line := strings.TrimSpace(scanner.Text()) + if line == "" { + continue + } + // Some providers emit "data: {...}", others "data:{...}". Comment + // lines beginning ":" are keep-alives and carry nothing. + if !strings.HasPrefix(line, "data:") { + continue + } + payload := strings.TrimSpace(strings.TrimPrefix(line, "data:")) + if payload == "" || payload == "[DONE]" { + continue + } + + var chunk oaiResponse + if err := json.Unmarshal([]byte(payload), &chunk); err != nil { + // One malformed event is not a failed response. Skipping it keeps + // a keep-alive or a provider-specific event from ending a stream + // that is otherwise fine. + continue + } + + if chunk.Model != "" { + acc.model = chunk.Model + } + // The usage chunk arrives last and carries no choices. Guarded rather + // than assumed: a zero usage overwriting a real one would silently + // hand the run a free turn. + if chunk.Usage.PromptTokens > 0 || chunk.Usage.CompletionTokens > 0 { + acc.usage = chunk.Usage + } + if len(chunk.Choices) == 0 { + continue + } + + choice := chunk.Choices[0] + if choice.FinishReason != "" { + acc.finishReason = choice.FinishReason + } + if d := choice.Delta.Content; d != "" { + acc.text.WriteString(d) + if onDelta != nil { + onDelta(d) + } + } + // A refusal is accumulated but never streamed to the reader: it is not + // the answer, and putting it on screen would show a declined request + // as though it were one. + if d := choice.Delta.Refusal; d != "" { + acc.refusal.WriteString(d) + } + acc.addToolCallDeltas(choice.Delta.ToolCalls) + } + + if err := scanner.Err(); err != nil { + return nil, &Error{ + Code: CodeUpstream, + Message: "the streamed response could not be assembled", + Cause: err, + } + } + return acc, nil +} + +// addToolCallDeltas folds one event's tool-call fragments into the accumulator. +func (a *streamAccumulator) addToolCallDeltas(deltas []oaiToolCall) { + for _, d := range deltas { + idx := 0 + if d.Index != nil { + idx = *d.Index + } + + call, seen := a.calls[idx] + if !seen { + call = &oaiToolCall{Type: "function"} + a.calls[idx] = call + a.order = append(a.order, idx) + } + + // The id and name arrive once, on the opening fragment. Assigned only + // when non-empty so a later fragment carrying empty strings — which is + // the common shape — does not erase them. + if d.ID != "" { + call.ID = d.ID + } + if d.Type != "" { + call.Type = d.Type + } + if d.Function.Name != "" { + call.Function.Name = d.Function.Name + } + // Arguments are the fragmented field: concatenated, never replaced. + call.Function.Arguments += d.Function.Arguments + } +} diff --git a/go-api/internal/gateway/openai_stream_test.go b/go-api/internal/gateway/openai_stream_test.go new file mode 100644 index 0000000..ed36bdb --- /dev/null +++ b/go-api/internal/gateway/openai_stream_test.go @@ -0,0 +1,211 @@ +package gateway + +import ( + "context" + "encoding/json" + "errors" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +// sse stands up an endpoint that replays the given event lines. +func sse(t *testing.T, events ...string) *OpenAIGateway { + t.Helper() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + for _, e := range events { + _, _ = io.WriteString(w, e+"\n") + } + })) + t.Cleanup(srv.Close) + return NewOpenAI(Config{ + APIKey: "test-key", + BaseURL: srv.URL, + Balanced: Routing{Model: "m-balanced", Effort: EffortHigh}, + }) +} + +func TestStreamDeliversTextAsItArrives(t *testing.T) { + gw := sse(t, + `data: {"model":"m-1","choices":[{"delta":{"content":"Three "}}]}`, + `data: {"choices":[{"delta":{"content":"are "}}]}`, + `data: {"choices":[{"delta":{"content":"free."},"finish_reason":"stop"}]}`, + `data: {"choices":[],"usage":{"prompt_tokens":40,"completion_tokens":4}}`, + `data: [DONE]`, + ) + + var deltas []string + resp, err := gw.Stream(context.Background(), ask("who is free?"), func(d string) { + deltas = append(deltas, d) + }) + if err != nil { + t.Fatalf("Stream: %v", err) + } + + if strings.Join(deltas, "") != "Three are free." { + t.Errorf("deltas joined to %q", strings.Join(deltas, "")) + } + if len(deltas) != 3 { + t.Errorf("got %d deltas, want 3 — text must arrive in fragments, not in one lump", len(deltas)) + } + if resp.Text != "Three are free." { + t.Errorf("Text = %q", resp.Text) + } + // Usage arrives in a trailing chunk with no choices. Missing it would mean + // a streamed run cost nothing on the ledger, and I3 cannot enforce a budget + // it cannot measure. + if resp.Usage.Total() != 44 { + t.Errorf("Usage.Total() = %d, want 44 — the trailing usage chunk was dropped", resp.Usage.Total()) + } + if resp.StopReason != "end_turn" { + t.Errorf("StopReason = %q", resp.StopReason) + } +} + +// THE ONE THAT IS EASY TO GET WRONG. +// +// Providers interleave the fragments of parallel tool calls, so arrival order +// is not call order. Appending fragments as they land splices one call's +// arguments onto another's — producing two calls that are each valid JSON and +// both wrong, which is the worst possible failure: the tools run, with the +// wrong inputs, and nothing errors. +func TestStreamAccumulatesInterleavedToolCallsByIndex(t *testing.T) { + gw := sse(t, + `data: {"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_a","function":{"name":"find_workers","arguments":"{\"day\""}}]}}]}`, + `data: {"choices":[{"delta":{"tool_calls":[{"index":1,"id":"call_b","function":{"name":"open_shifts","arguments":"{\"week\""}}]}}]}`, + `data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":":\"friday\"}"}}]}}]}`, + `data: {"choices":[{"delta":{"tool_calls":[{"index":1,"function":{"arguments":":\"next\"}"}}]}}]}`, + `data: {"choices":[{"delta":{},"finish_reason":"tool_calls"}]}`, + `data: [DONE]`, + ) + + resp, err := gw.Stream(context.Background(), ask("cover friday"), nil) + if err != nil { + t.Fatalf("Stream: %v", err) + } + if len(resp.ToolCalls) != 2 { + t.Fatalf("got %d tool calls, want 2: %+v", len(resp.ToolCalls), resp.ToolCalls) + } + + want := []struct{ id, name, day string }{ + {"call_a", "find_workers", "friday"}, + {"call_b", "open_shifts", "next"}, + } + for i, w := range want { + got := resp.ToolCalls[i] + if got.ID != w.id || got.Name != w.name { + t.Errorf("call %d = {%s %s}, want {%s %s}", i, got.ID, got.Name, w.id, w.name) + } + // Each must be valid JSON on its own. A spliced pair usually is too, + // which is exactly why the value is asserted and not just the parse. + var args map[string]string + if err := json.Unmarshal(got.Input, &args); err != nil { + t.Fatalf("call %d input %q is not valid JSON: %v", i, got.Input, err) + } + if len(args) != 1 { + t.Errorf("call %d carried %d args, want 1 — fragments from another call were spliced in: %v", + i, len(args), args) + } + for _, v := range args { + if v != w.day { + t.Errorf("call %d arg = %q, want %q", i, v, w.day) + } + } + } + if resp.StopReason != "tool_use" { + t.Errorf("StopReason = %q, want tool_use", resp.StopReason) + } +} + +// A tool call is buffered until the stream closes: a half-built argument object +// is not a smaller version of the finished one, and dispatching on it would run +// a tool with arguments the model had not finished choosing. +func TestStreamNeverEmitsPartialToolArguments(t *testing.T) { + gw := sse(t, + `data: {"choices":[{"delta":{"tool_calls":[{"index":0,"id":"c","function":{"name":"t","arguments":"{\"a\":"}}]}}]}`, + `data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"1}"}}]}}]}`, + `data: {"choices":[{"delta":{},"finish_reason":"tool_calls"}]}`, + `data: [DONE]`, + ) + + var streamed strings.Builder + resp, err := gw.Stream(context.Background(), ask("go"), func(d string) { streamed.WriteString(d) }) + if err != nil { + t.Fatalf("Stream: %v", err) + } + if streamed.String() != "" { + t.Errorf("tool-call JSON reached the reader as text: %q", streamed.String()) + } + if string(resp.ToolCalls[0].Input) != `{"a":1}` { + t.Errorf("Input = %q, want the assembled object", resp.ToolCalls[0].Input) + } +} + +// Keep-alives, comment lines and provider-specific events are not failures. A +// stream that died on one would fail against providers that are working fine. +func TestStreamIgnoresNoiseEvents(t *testing.T) { + gw := sse(t, + `: keep-alive`, + ``, + `event: ping`, + `data: {"not":"a chunk"`, + `data:{"choices":[{"delta":{"content":"ok"},"finish_reason":"stop"}]}`, + `data: [DONE]`, + ) + + resp, err := gw.Stream(context.Background(), ask("hi"), nil) + if err != nil { + t.Fatalf("Stream: %v", err) + } + if resp.Text != "ok" { + t.Errorf("Text = %q, want ok", resp.Text) + } +} + +// A streamed refusal must come back as the same structured outcome the +// non-streaming path produces, and must not be shown to the reader as though +// it were the answer. +func TestStreamRefusalIsNotShownToTheReader(t *testing.T) { + gw := sse(t, + `data: {"choices":[{"delta":{"refusal":"I cannot help with that."},"finish_reason":"stop"}]}`, + `data: [DONE]`, + ) + + var streamed strings.Builder + _, err := gw.Stream(context.Background(), ask("do something disallowed"), + func(d string) { streamed.WriteString(d) }) + + var gwErr *Error + if !errors.As(err, &gwErr) || gwErr.Code != CodeRefused { + t.Fatalf("err = %v, want a %s", err, CodeRefused) + } + if streamed.String() != "" { + t.Errorf("a refusal was streamed to the reader as an answer: %q", streamed.String()) + } +} + +// StreamComplete has to reach the streaming path for a gateway that has one. +// The fallback exists for gateways that do not, and silently taking it here +// would turn every streamed answer into one lump with no error to trace it to. +func TestStreamCompleteUsesTheStreamingPath(t *testing.T) { + gw := sse(t, + `data: {"choices":[{"delta":{"content":"a"}}]}`, + `data: {"choices":[{"delta":{"content":"b"},"finish_reason":"stop"}]}`, + `data: [DONE]`, + ) + + var deltas int + resp, err := StreamComplete(context.Background(), gw, ask("hi"), func(string) { deltas++ }) + if err != nil { + t.Fatalf("StreamComplete: %v", err) + } + if deltas != 2 { + t.Errorf("got %d deltas, want 2 — the non-streaming fallback was taken", deltas) + } + if resp.Text != "ab" { + t.Errorf("Text = %q", resp.Text) + } +} diff --git a/go-api/internal/gateway/openai_test.go b/go-api/internal/gateway/openai_test.go new file mode 100644 index 0000000..9dc7749 --- /dev/null +++ b/go-api/internal/gateway/openai_test.go @@ -0,0 +1,341 @@ +package gateway + +import ( + "context" + "encoding/json" + "errors" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +// serve stands up a fake OpenAI-compatible endpoint and returns a gateway +// pointed at it, plus a pointer to the last request body it received. +func serve(t *testing.T, handler func(w http.ResponseWriter, body *oaiRequest)) (*OpenAIGateway, *oaiRequest) { + t.Helper() + var captured oaiRequest + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + raw, _ := io.ReadAll(r.Body) + if err := json.Unmarshal(raw, &captured); err != nil { + t.Errorf("request body was not valid JSON: %v", err) + } + handler(w, &captured) + })) + t.Cleanup(srv.Close) + + gw := NewOpenAI(Config{ + Provider: ProviderOpenAI, + APIKey: "test-key", + BaseURL: srv.URL, + Fast: Routing{Model: "m-fast", Effort: EffortLow}, + Balanced: Routing{Model: "m-balanced", Effort: EffortHigh}, + Deep: Routing{Model: "m-deep", Effort: EffortXhigh}, + MaxOutputTokens: 4096, + }) + return gw, &captured +} + +func ask(text string) Request { + return Request{Tier: TierBalanced, Messages: []Message{{Role: RoleUser, Text: text}}} +} + +// THE REGRESSION THIS FILE EXISTS FOR. +// +// OpenAI reports prompt_tokens INCLUSIVE of the cached prefix; Anthropic +// reports input tokens EXCLUSIVE of it. Usage.Total() adds all four fields, so +// copying both numbers across verbatim bills the cached prefix twice — and it +// does it worst on long conversations, which is exactly where I3's budget +// matters most. A wrong total here is invisible: the run still answers, it just +// terminates BudgetExceeded earlier than it should. +func TestUsageDoesNotDoubleCountCachedTokens(t *testing.T) { + usage := oaiUsage{PromptTokens: 1000, CompletionTokens: 200} + usage.PromptTokensDetails.CachedTokens = 800 + + got := usage.normalise() + + if got.InputTokens != 200 { + t.Errorf("InputTokens = %d, want 200 (1000 prompt less 800 cached)", got.InputTokens) + } + if got.CacheReadTokens != 800 { + t.Errorf("CacheReadTokens = %d, want 800", got.CacheReadTokens) + } + if got.Total() != 1200 { + t.Errorf("Total() = %d, want 1200 — the wire billed 1000 prompt + 200 output, "+ + "and anything higher is the cached prefix counted twice", got.Total()) + } +} + +// A provider reporting more cached tokens than prompt tokens is wrong, but the +// failure must not hand the run free budget: a negative charge would reduce the +// total, which is the one direction a bug must never go. +func TestUsageClampsImpossibleCacheReport(t *testing.T) { + usage := oaiUsage{PromptTokens: 100, CompletionTokens: 10} + usage.PromptTokensDetails.CachedTokens = 500 + + got := usage.normalise() + if got.InputTokens < 0 { + t.Fatalf("InputTokens = %d, want no negative charge", got.InputTokens) + } + if got.Total() < got.OutputTokens { + t.Errorf("Total() = %d is below OutputTokens = %d", got.Total(), got.OutputTokens) + } +} + +// Tool results are blocks inside one user turn on the Anthropic wire and +// standalone role:"tool" messages here. Getting the split wrong detaches a +// result from the call it answers, which most providers reject outright and +// some silently mis-attribute. +func TestEncodeMessagesSplitsToolResults(t *testing.T) { + msgs := []Message{ + {Role: RoleUser, Text: "who is free friday?"}, + {Role: RoleAssistant, ToolCalls: []ToolCall{ + {ID: "call_1", Name: "find_workers", Input: json.RawMessage(`{"day":"friday"}`)}, + {ID: "call_2", Name: "open_shifts", Input: json.RawMessage(`{}`)}, + }}, + {Role: RoleUser, ToolResults: []ToolResult{ + {CallID: "call_1", Content: `{"workers":3}`}, + {CallID: "call_2", Content: `{"shifts":1}`}, + }}, + } + + got := encodeOpenAIMessages("you are a scheduler", msgs) + + wantRoles := []string{"system", "user", "assistant", "tool", "tool"} + if len(got) != len(wantRoles) { + t.Fatalf("got %d messages, want %d: %+v", len(got), len(wantRoles), got) + } + for i, want := range wantRoles { + if got[i].Role != want { + t.Errorf("messages[%d].Role = %q, want %q", i, got[i].Role, want) + } + } + if got[0].Content != "you are a scheduler" { + t.Errorf("system message = %q", got[0].Content) + } + if len(got[2].ToolCalls) != 2 { + t.Fatalf("assistant turn carried %d tool calls, want 2", len(got[2].ToolCalls)) + } + // The call id is the model's own handle. A result carrying a different one + // is a result attached to the wrong question. + if got[3].ToolCallID != "call_1" || got[4].ToolCallID != "call_2" { + t.Errorf("tool results correlated to %q and %q, want call_1 and call_2", + got[3].ToolCallID, got[4].ToolCallID) + } +} + +// A turn that is only tool results carries no text, and dropping it would strip +// every answer the tools produced. +func TestEncodeMessagesKeepsResultOnlyTurn(t *testing.T) { + got := encodeOpenAIMessages("", []Message{ + {Role: RoleUser, Text: "hi"}, + {Role: RoleUser, ToolResults: []ToolResult{{CallID: "c1", Content: "{}"}}}, + }) + if len(got) != 2 || got[1].Role != "tool" { + t.Fatalf("result-only turn was not encoded: %+v", got) + } +} + +func TestCompleteDecodesTextAndUsage(t *testing.T) { + gw, captured := serve(t, func(w http.ResponseWriter, _ *oaiRequest) { + _, _ = io.WriteString(w, `{ + "model":"m-balanced-0625", + "choices":[{"message":{"role":"assistant","content":"Three are free."}, + "finish_reason":"stop"}], + "usage":{"prompt_tokens":120,"completion_tokens":8} + }`) + }) + + resp, err := gw.Complete(context.Background(), ask("who is free?")) + if err != nil { + t.Fatalf("Complete: %v", err) + } + if resp.Text != "Three are free." { + t.Errorf("Text = %q", resp.Text) + } + // The id ACTUALLY used, not the tier that was asked for — a change of + // routing has to be visible in the trajectory rather than inferred. + if resp.Model != "m-balanced-0625" { + t.Errorf("Model = %q, want the id the provider reported", resp.Model) + } + if resp.StopReason != "end_turn" { + t.Errorf("StopReason = %q, want end_turn", resp.StopReason) + } + if resp.Usage.Total() != 128 { + t.Errorf("Usage.Total() = %d, want 128", resp.Usage.Total()) + } + if captured.Model != "m-balanced" { + t.Errorf("requested model = %q, want the balanced tier's", captured.Model) + } +} + +func TestCompleteDecodesToolCalls(t *testing.T) { + gw, _ := serve(t, func(w http.ResponseWriter, _ *oaiRequest) { + _, _ = io.WriteString(w, `{ + "choices":[{"message":{"role":"assistant","tool_calls":[ + {"id":"call_x","type":"function", + "function":{"name":"find_workers","arguments":"{\"day\":\"friday\"}"}}]}, + "finish_reason":"tool_calls"}], + "usage":{"prompt_tokens":10,"completion_tokens":5} + }`) + }) + + resp, err := gw.Complete(context.Background(), ask("who is free?")) + if err != nil { + t.Fatalf("Complete: %v", err) + } + if len(resp.ToolCalls) != 1 { + t.Fatalf("got %d tool calls, want 1", len(resp.ToolCalls)) + } + call := resp.ToolCalls[0] + if call.ID != "call_x" || call.Name != "find_workers" { + t.Errorf("call = %+v", call) + } + // The loop branches on len(ToolCalls), but the trajectory records the stop + // reason, and it has to read the same as the Anthropic path's. + if resp.StopReason != "tool_use" { + t.Errorf("StopReason = %q, want tool_use", resp.StopReason) + } + var args map[string]string + if err := json.Unmarshal(call.Input, &args); err != nil { + t.Fatalf("tool input was not valid JSON: %v", err) + } + if args["day"] != "friday" { + t.Errorf("args = %v", args) + } +} + +// An argumentless call arrives as "" on this wire, which is not valid JSON. The +// handler's decoder would reject it for a reason that has nothing to do with +// the request. +func TestEmptyToolArgumentsBecomeEmptyObject(t *testing.T) { + gw, _ := serve(t, func(w http.ResponseWriter, _ *oaiRequest) { + _, _ = io.WriteString(w, `{"choices":[{"message":{"tool_calls":[ + {"id":"c1","function":{"name":"workspace_summary","arguments":""}}]}, + "finish_reason":"tool_calls"}]}`) + }) + + resp, err := gw.Complete(context.Background(), ask("summarise")) + if err != nil { + t.Fatalf("Complete: %v", err) + } + if string(resp.ToolCalls[0].Input) != "{}" { + t.Errorf("Input = %q, want {}", resp.ToolCalls[0].Input) + } +} + +// A refusal is a successful HTTP response and one of the six terminations. It +// is still billed: a refusal that cost nothing on the ledger is one the loop +// would happily repeat. +func TestRefusalIsStructuredAndStillBilled(t *testing.T) { + gw, _ := serve(t, func(w http.ResponseWriter, _ *oaiRequest) { + _, _ = io.WriteString(w, `{"choices":[{"message":{"role":"assistant", + "refusal":"I cannot help with that."},"finish_reason":"stop"}], + "usage":{"prompt_tokens":50,"completion_tokens":6}}`) + }) + + resp, err := gw.Complete(context.Background(), ask("do something disallowed")) + + var gwErr *Error + if !errors.As(err, &gwErr) || gwErr.Code != CodeRefused { + t.Fatalf("err = %v, want a %s", err, CodeRefused) + } + if gwErr.Retryable() { + t.Error("a refusal must not be retryable — re-sending it burns the budget on one turn") + } + if resp == nil { + t.Fatal("a refusal must still carry its usage") + } + if resp.Usage.Total() != 56 { + t.Errorf("Usage.Total() = %d, want 56", resp.Usage.Total()) + } +} + +func TestErrorsMapToRetryability(t *testing.T) { + cases := []struct { + status int + wantCode string + retryable bool + }{ + {400, CodeInvalidRequest, false}, + // A model id that does not exist on this endpoint is a configuration + // mistake and will fail identically next time. + {404, CodeInvalidRequest, false}, + {401, CodeUnauthorized, false}, + {429, CodeRateLimited, true}, + {500, CodeUpstream, true}, + {503, CodeUpstream, true}, + } + for _, c := range cases { + err := translateOpenAI(c.status, []byte(`{"error":{"message":"upstream detail"}}`)) + var gwErr *Error + if !errors.As(err, &gwErr) { + t.Fatalf("http %d: not a gateway error", c.status) + } + if gwErr.Code != c.wantCode { + t.Errorf("http %d: code = %s, want %s", c.status, gwErr.Code, c.wantCode) + } + if gwErr.Retryable() != c.retryable { + t.Errorf("http %d: Retryable() = %v, want %v", c.status, gwErr.Retryable(), c.retryable) + } + // The upstream reason has to survive: the trajectory records only the + // message, and "the model call failed" costs an hour to diagnose. + if !strings.Contains(gwErr.Message, "upstream detail") { + t.Errorf("http %d: message %q dropped the upstream detail", c.status, gwErr.Message) + } + } +} + +// Most non-reasoning models reject the whole request rather than ignoring an +// unknown key, so the field must be absent unless a deployment opted in. +func TestReasoningEffortIsOptIn(t *testing.T) { + gw, captured := serve(t, func(w http.ResponseWriter, _ *oaiRequest) { + _, _ = io.WriteString(w, `{"choices":[{"message":{"content":"ok"},"finish_reason":"stop"}]}`) + }) + if _, err := gw.Complete(context.Background(), ask("hi")); err != nil { + t.Fatalf("Complete: %v", err) + } + if captured.ReasoningEffort != "" { + t.Errorf("reasoning_effort = %q, want it omitted by default", captured.ReasoningEffort) + } + + gw.cfg.SendReasoningEffort = true + if _, err := gw.Complete(context.Background(), Request{ + Tier: TierDeep, Messages: []Message{{Role: RoleUser, Text: "hi"}}, + }); err != nil { + t.Fatalf("Complete: %v", err) + } + // Ordering preserved, not spelling: their scale runs minimal/low/medium/ + // high, so the platform's xhigh is their high. + if captured.ReasoningEffort != "high" { + t.Errorf("deep tier sent reasoning_effort = %q, want high", captured.ReasoningEffort) + } +} + +// A local model needs no credential. Requiring one would make the zero-cost +// development path impossible to configure. +func TestLocalEndpointNeedsNoCredential(t *testing.T) { + local := NewOpenAI(Config{BaseURL: "http://localhost:11434/v1"}) + if local.needsCredential() { + t.Error("a localhost endpoint must not require a key") + } + hosted := NewOpenAI(Config{BaseURL: "https://api.groq.com/openai/v1"}) + if !hosted.needsCredential() { + t.Error("a hosted endpoint must require a key") + } + if _, err := NewOpenAI(Config{BaseURL: "https://api.groq.com/openai/v1"}). + Complete(context.Background(), ask("hi")); err == nil { + t.Error("a hosted call without a key must fail as NotConfigured") + } +} + +func TestBaseURLDefaultsAndTrimsSlash(t *testing.T) { + if got := NewOpenAI(Config{}).endpoint(); got != DefaultOpenAIBaseURL+"/chat/completions" { + t.Errorf("endpoint = %q", got) + } + if got := NewOpenAI(Config{BaseURL: "https://x.test/v1/"}).endpoint(); got != "https://x.test/v1/chat/completions" { + t.Errorf("endpoint = %q, want the trailing slash collapsed", got) + } +} diff --git a/go-api/internal/gateway/routing.go b/go-api/internal/gateway/routing.go index dc2d772..7b4a2a1 100644 --- a/go-api/internal/gateway/routing.go +++ b/go-api/internal/gateway/routing.go @@ -1,11 +1,90 @@ package gateway import ( - "github.com/anthropics/anthropic-sdk-go" - "github.com/krow/krow-backend/go-api/internal/config" ) +// Provider names the wire protocol a deployment talks. +// +// Two, not two hundred: "anthropic" is the Claude API, and "openai" is the +// chat-completions shape that Groq, Gemini, OpenRouter, Together, vLLM and +// Ollama all serve. That second one is the reason this constant exists at all +// — supporting those five providers is one implementation and five different +// base URLs, and pretending otherwise would grow a package per vendor. +const ( + ProviderAnthropic = "anthropic" + ProviderOpenAI = "openai" +) + +// Effort is how hard a tier is allowed to think. +// +// PROVIDER-NEUTRAL ON PURPOSE. This was `anthropic.OutputConfigEffort` until a +// second provider existed, which meant the vendor's enum was baked into the +// routing table that every provider has to read. Nothing was wrong with it +// while there was one implementation; it became wrong the moment there were +// two, because the OpenAI path would have had to import the Anthropic SDK to +// learn how hard to think. +// +// The three values are the platform's own vocabulary. Each implementation maps +// them onto whatever its API calls the same idea, and a provider with no such +// concept ignores them — the tier still selects the model, which is the larger +// lever anyway. +type Effort string + +const ( + EffortLow Effort = "low" + EffortHigh Effort = "high" + EffortXhigh Effort = "xhigh" +) + +// Routing is how a tier becomes a model and an effort level. +// +// The model per tier is a deployment knob — a tenant on a different contract, +// or a deployment pinning a version through an incident, changes it without a +// spec edit. The *effort* per tier is not: "fast" and "deep" mean something +// specific about how much work an answer is worth, and letting a deployment +// redefine that would make the same spec behave differently in two places +// while claiming the same tier. +type Routing struct { + Model string + Effort Effort +} + +// Config is the gateway's whole configuration surface. +// +// Built once at startup from the environment and passed in frozen, per §10. +// Nothing in this package reads the environment itself. +type Config struct { + // Provider selects the implementation. Empty means anthropic, so a + // deployment that predates the second provider keeps working untouched. + Provider string + + APIKey string + + // BaseURL points the OpenAI-compatible path at a specific service. Empty + // means OpenAI itself. This is the field that turns one implementation + // into a choice between Groq, Gemini, OpenRouter and a local Ollama. + BaseURL string + + Fast Routing + Balanced Routing + Deep Routing + + // MaxOutputTokens applies when a request does not set its own. + MaxOutputTokens int64 + + // SendReasoningEffort controls whether the OpenAI path transmits the + // effort level as `reasoning_effort`. + // + // OFF BY DEFAULT, and that default is the careful one. Reasoning models + // accept the field; most others reject the whole request with a 400 rather + // than ignoring an unknown key. A run that dies on a malformed request is + // worse than a run that thinks at the model's own default, so a deployment + // on a reasoning-capable model opts in rather than every other deployment + // opting out. + SendReasoningEffort bool +} + // FromConfig builds the gateway's routing table from validated settings. // // The effort per tier is fixed here rather than configured, and that is the @@ -24,10 +103,41 @@ import ( // about a deployment, not one an agent author makes about a page. func FromConfig(c config.ModelConfig) Config { return Config{ - APIKey: c.APIKey, - Fast: Routing{Model: c.Fast, Effort: anthropic.OutputConfigEffortLow}, - Balanced: Routing{Model: c.Balanced, Effort: anthropic.OutputConfigEffortHigh}, - Deep: Routing{Model: c.Deep, Effort: anthropic.OutputConfigEffortXhigh}, - MaxOutputTokens: int64(c.MaxOutputTokens), + Provider: c.Provider, + APIKey: c.APIKey, + BaseURL: c.BaseURL, + Fast: Routing{Model: c.Fast, Effort: EffortLow}, + Balanced: Routing{Model: c.Balanced, Effort: EffortHigh}, + Deep: Routing{Model: c.Deep, Effort: EffortXhigh}, + MaxOutputTokens: int64(c.MaxOutputTokens), + SendReasoningEffort: c.ReasoningEffort, + } +} + +// New builds the gateway a deployment's configuration asks for. +// +// The one place that maps a provider name to an implementation, so a caller +// wires a gateway without knowing which vendor answers. An unrecognised +// provider cannot reach here — config.validate rejects it at startup, where a +// typo is one loud failure instead of one per run. +func New(cfg Config) Gateway { + if cfg.Provider == ProviderOpenAI { + return NewOpenAI(cfg) + } + return NewAnthropic(cfg) +} + +// routingFor resolves a tier against a table. +// +// Shared by both implementations: an unknown tier has already been normalised +// by ParseTier, so the default arm is reached only by a zero value. +func (c Config) routingFor(t Tier) Routing { + switch t { + case TierFast: + return c.Fast + case TierDeep: + return c.Deep + default: + return c.Balanced } } diff --git a/go-api/internal/runtime/wire.go b/go-api/internal/runtime/wire.go index 0379d55..31c3f45 100644 --- a/go-api/internal/runtime/wire.go +++ b/go-api/internal/runtime/wire.go @@ -21,7 +21,10 @@ import ( // carries. Handing SkillExec a model would create a second, unbounded path to // one — which is exactly the shape I3 exists to prevent. func NewModelEngine(db repo.Querier, cfg config.Config) *Engine { - gw := gateway.NewAnthropic(gateway.FromConfig(cfg.Model)) + // gateway.New, not NewAnthropic: which provider answers is a deployment + // decision now, and hardcoding the constructor here would have meant every + // alternative provider needed an edit to this file to be reachable. + gw := gateway.New(gateway.FromConfig(cfg.Model)) retriever := knowledge.NewRetriever(db, NewEmbedder(cfg)) exec := NewModelExecutor(gw, NewPostgresSink(db), DefaultTools(db, retriever)). WithRetriever(retriever). diff --git a/infrastructure/docker-compose.yml b/infrastructure/docker-compose.yml index ff6c948..88e8803 100644 --- a/infrastructure/docker-compose.yml +++ b/infrastructure/docker-compose.yml @@ -92,6 +92,12 @@ services: # without it answers 404 on /agents/{id}/runs and reports three fewer # endpoints on /version. Empty by default: absent is a working API # without Owliver, which is a legitimate way to run this. + # Provider selection. Empty MODEL_PROVIDER means anthropic, so a stack + # that predates the second provider comes up exactly as it did. + MODEL_PROVIDER: ${MODEL_PROVIDER:-} + MODEL_BASE_URL: ${MODEL_BASE_URL:-} + MODEL_API_KEY: ${MODEL_API_KEY:-} + MODEL_REASONING_EFFORT: ${MODEL_REASONING_EFFORT:-} ANTHROPIC_API_KEY: ${ANTHROPIC_API_KEY:-} MODEL_FAST: ${MODEL_FAST:-} MODEL_BALANCED: ${MODEL_BALANCED:-}