diff --git a/docs/subsystems/gateway-routing.md b/docs/subsystems/gateway-routing.md index 5f7d0720..b80ddfc5 100644 --- a/docs/subsystems/gateway-routing.md +++ b/docs/subsystems/gateway-routing.md @@ -23,7 +23,7 @@ served by a plugin are in [Provider and plugin ownership](provider-plugins.md). | Limits | Each key or account (a candidate's `who()`) has at most `MaxConcurrency` requests out at once (its own `AccountConcurrency` over it); one more waits in its lane's queue, bounded by `QueueLimit` and `QueueWait` and turned away with a 429 past either (`lanes.take`). `MaxRPM` (coeo91 on Discord: OpenRouter's free models take 20 a minute) bounds how many requests it sends the vendor in any rolling 60 seconds (`rpms`): one more waits for room in the minute, in the order they came, after it has its slot and right before it is sent. A wait that would be longer than 2 minutes (or the provider's `QueueWait`, when shorter: `rpmWait`) is turned away at once with a 429 whose `Retry-After` is when the minute has room; nothing rests for it. An agent gone while it waits gives its place back. Every request that reaches the vendor counts: each try of the candidate loop (a retry, the next key, a fallback) waits in `serve`, and the requests a try makes on the way and the gateway's own asks (count_tokens, System One's decisions, embeddings, drawing, Kimi's search) are counted by `rpmTransport` on `s.client` (`metered`, `paidFor`). count_tokens never waits: a candidate with no room is skipped, and the local estimate answers. A full lane or a minute with no room gives way to a ready key or account of the provider or the group (`laneMate`, `ready`); a fallback is never asked for want of either. A plugin's account sends through its own transport, so only its tries are counted, not a plugin's own extra requests. Set in the provider editor (Concurrency, Requests per minute, Queue size, Queue wait) or with `magpie provider account-concurrency`, `queue` and `rpm`; a subscription's are kept with its picks. | [`concurrency.go`](../../internal/gateway/concurrency.go), [`rpm.go`](../../internal/gateway/rpm.go), [`internal/provider/concurrency.go`](../../internal/provider/concurrency.go) | | Sink | With `Sink` on, a key or account that answers 429 while it still has quota goes to the back, behind every one not rate limited since. Held in memory until restart. | [`sink.go`](../../internal/gateway/sink.go) | | Affinity | A conversation stays with the key or account that answered it, so the vendor's prompt cache is read again. Modes: auto (""), `session`, `turn`, `off`. Stored in `affinity.json` next to `providers.json`. | [`affinity.go`](../../internal/gateway/affinity.go) | -| One try | `attempt` sends one candidate the request. It goes through the proxy of the account or key it uses, else the provider's (`p.Via`, `ProxyChoice`: `AccountProxies` holds an account's by its name, a key's by its `KeyID`). A subscription with its own client is served specially, such as Claude through the genuine Claude Code. The request is relayed as is (`passthrough`) when the model is usable on the agent's API, except that a Claude model where the provider has an Anthropic URL is translated to Messages from Chat or Responses too (`OnMessages`: OpenCode speaks Chat for every model, and a relay drops cache_control there), unless the user set the model's API or the vendor's list names its APIs. A request offering web search to a provider that doesn't search by itself is translated too, so magpie can search for it (`search.go`), but on the agent's own API where the provider serves it and no `ResponsesFirst`/`MessagesFirst` rule prefers another: Codex offers web_search on every turn, and each went out as Chat (#997). Otherwise it is translated (`translate`) to the first API `usable` gives: Chat, except Responses for OpenAI's models where their maker serves them (`ResponsesFirst`) and Anthropic's Messages for a Claude model where the provider has an Anthropic URL (`MessagesFirst`, which keeps its cache breakpoints, #997). `Provider.Native`, the API agents with a per-model API (Pi, omp, droid) are wired to ask on, follows the same order. A custom provider with a Gemini URL (#1346) has Gemini last among the APIs it speaks: a Gemini client's request is relayed there, and any other is translated to it when it has no other. Upstream it is always asked `models/{model}:streamGenerateContent?alt=sse` under the base (`upstreamPath`), the model only in the path (`forwardOnce` drops the body's), so a Gemini client that asked for one JSON body is translated, as for Factory's generate route. | [`gateway.go`](../../internal/gateway/gateway.go) | +| One try | `attempt` sends one candidate the request. It goes through the proxy of the account or key it uses, else the provider's (`p.Via`, `ProxyChoice`: `AccountProxies` holds an account's by its name, a key's by its `KeyID`). A subscription with its own client is served specially, such as Claude through the genuine Claude Code. The request is relayed as is (`passthrough`) when the model is usable on the agent's API, except that a Claude model where the provider has an Anthropic URL is translated to Messages from Chat or Responses too (`OnMessages`: OpenCode speaks Chat for every model, and a relay drops cache_control there), unless the user set the model's API or the vendor's list names its APIs. A request offering web search to a provider that doesn't search by itself is translated too, so magpie can search for it (`search.go`), but on the agent's own API where the provider serves it and no `ResponsesFirst`/`MessagesFirst` rule prefers another: Codex offers web_search on every turn, and each went out as Chat (#997). Otherwise it is translated (`translate`) to the first API `usable` gives: Chat, except Responses for OpenAI's models where their maker serves them (`ResponsesFirst`) and Anthropic's Messages for a Claude model where the provider has an Anthropic URL (`MessagesFirst`, which keeps its cache breakpoints, #997). An API that answers the model is not served there (`wrongEndpoint`) is remembered in `unfit` until magpie restarts. When every API has turned the model away, the API the user picked for it (`ModelAPI`) is asked again, and without a pick each API the provider has a URL for (`Speaks`), Chat first. `Provider.Native`, the API agents with a per-model API (Pi, omp, droid) are wired to ask on, follows the same order. A custom provider with a Gemini URL (#1346) has Gemini last among the APIs it speaks: a Gemini client's request is relayed there, and any other is translated to it when it has no other. Upstream it is always asked `models/{model}:streamGenerateContent?alt=sse` under the base (`upstreamPath`), the model only in the path (`forwardOnce` drops the body's), so a Gemini client that asked for one JSON body is translated, as for Factory's generate route. | [`gateway.go`](../../internal/gateway/gateway.go) | | Trace | The Routing view's record of each decision, written where the decision is made: candidates, rests, which try answered. | [`trace.go`](../../internal/gateway/trace.go) | | Prompt | What each request's prompt held, part by part: the agent's system prompt, tool definitions (an MCP server's tools as one item), instruction files and skills, files its tools read, their other output, and the conversation. `inspectPrompt` reads it from the body as the agent sent it, before translation; `calibrated` scales the estimate to the prompt the vendor counted and sets the window of the model that answered. Only names are kept, never the text. It goes on the trace as `Route.Prompt`, with `Route.Conv` naming a conversation that has no session id. The GUI's `/api/context` scores each agent from it (`internal/gui/context.go`). | [`prompt.go`](../../internal/gateway/prompt.go) | diff --git a/internal/gateway/gateway.go b/internal/gateway/gateway.go index c3facb9e..55d8507e 100644 --- a/internal/gateway/gateway.go +++ b/internal/gateway/gateway.go @@ -2615,7 +2615,15 @@ func (s *Server) attempt(w http.ResponseWriter, r *http.Request, from provider.P } to := s.usable(p, model) if len(to) == 0 { - to = p.Speaks() + // every API turned the model away: the one the user picked for it + // is asked again, not the first the provider has a URL for, which + // they said it isn't asked on (01huadalang: picked for Responses, + // sent as chat after one refusal there) + if proto, ok := p.ModelAPI(model); ok { + to = []provider.Protocol{proto} + } else { + to = p.Speaks() + } } if len(to) == 0 { msg := p.Name + " has no endpoint configured" diff --git a/internal/gateway/model_api_test.go b/internal/gateway/model_api_test.go index 6ff02ba0..1e1c0f7c 100644 --- a/internal/gateway/model_api_test.go +++ b/internal/gateway/model_api_test.go @@ -91,3 +91,82 @@ data: {"type":"message_stop"}`)) } } } + +// A model the user picked Responses for stays on Responses after the relay +// once answers it there with "use /v1/chat/completions" (01huadalang on +// Discord: 为什么会转成 chat,我都设置的 response): the next request was +// sent on chat completions, the first API the provider has a URL for, +// for as long as magpie ran. Now it is asked on Responses again, and the +// relay's error reaches the agent rather than a request on an API the +// user said the model isn't asked on. +func TestModelAPIPickHoldsAfterWrongEndpoint(t *testing.T) { + var mu sync.Mutex + var hits []string + refused := false + up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + hits = append(hits, r.URL.Path) + first := !refused + refused = true + mu.Unlock() + if r.URL.Path != "/v1/responses" { + http.Error(w, `{"error":{"message":"no available channel for model resp under group default"}}`, 503) + return + } + if first { + http.Error(w, `{"error":{"message":"this request is not supported in /v1/responses, use /v1/chat/completions","type":"invalid_request_error"}}`, 400) + return + } + w.Header().Set("Content-Type", "text/event-stream") + io.WriteString(w, sse(`event: response.created +data: {"type":"response.created","response":{"id":"r","object":"response","status":"in_progress","model":"resp","output":[]}}`, + `event: response.output_text.delta +data: {"type":"response.output_text.delta","item_id":"m","output_index":0,"content_index":0,"delta":"from responses"}`, + `event: response.completed +data: {"type":"response.completed","response":{"id":"r","object":"response","status":"completed","model":"resp","output":[{"type":"message","id":"m","role":"assistant","content":[{"type":"output_text","text":"from responses"}]}],"usage":{"input_tokens":3,"output_tokens":2,"total_tokens":5}}}`)) + })) + defer up.Close() + t.Setenv("XDG_CONFIG_HOME", t.TempDir()) + t.Setenv("XDG_CACHE_HOME", t.TempDir()) + if err := provider.Save(provider.Provider{ID: "relay", Name: "Relay", Key: "k", Chat: up.URL + "/v1", Responses: up.URL + "/v1", Models: []string{"resp"}}); err != nil { + t.Fatal(err) + } + if err := provider.SetModelAPI("relay/resp", "responses"); err != nil { + t.Fatal(err) + } + // each agent API on a gateway of its own, whose memory of the + // refusal starts empty + for _, c := range []struct{ path, body string }{ + {"/v1/responses", `{"model":"relay/resp","stream":true,"input":"hi"}`}, + {"/v1/chat/completions", `{"model":"relay/resp","stream":true,"messages":[{"role":"user","content":"hi"}]}`}, + } { + mu.Lock() + hits, refused = nil, false + mu.Unlock() + srv := New() + gw := httptest.NewServer(srv.Handler()) + for i := range 2 { + res, err := http.Post(gw.URL+c.path, "application/json", strings.NewReader(c.body)) + if err != nil { + t.Fatal(err) + } + b, _ := io.ReadAll(res.Body) + res.Body.Close() + if i == 1 && (res.StatusCode != 200 || !strings.Contains(string(b), "from responses")) { + mu.Lock() + h := append([]string(nil), hits...) + mu.Unlock() + t.Errorf("%s, after the refusal: %d %s (upstream %v)", c.path, res.StatusCode, b, h) + } + } + gw.Close() + mu.Lock() + for _, h := range hits { + if h != "/v1/responses" { + t.Errorf("%s: asked on %s, though the model is picked for Responses (upstream %v)", c.path, h, hits) + break + } + } + mu.Unlock() + } +}