diff --git a/docs/subsystems/gateway-routing.md b/docs/subsystems/gateway-routing.md index c85cfcf5..0a898fc8 100644 --- a/docs/subsystems/gateway-routing.md +++ b/docs/subsystems/gateway-routing.md @@ -49,7 +49,7 @@ served by a plugin are in [Provider and plugin ownership](provider-plugins.md). - A 2xx that can't be an API's answer is the 502 it stands for (`notAnAPIReply`, #1012): a web page (`text/html`, or a body that begins as one: a sign-in page, Cloudflare's challenge), a body with nothing in it, or one sent as JSON that doesn't begin as JSON. Every upstream request goes through it (`forwardOnce`, and the built-in clients' own: the ChatGPT backend's `/responses`, Kiro, Zed, Qoder, Devin, Cursor, Command Code), so it fails over, rests and is logged as a 502 does, its reason naming the page's title. Only the first bytes are peeked, which a stream waits for anyway, and a compressed body is left as it came. - A streamed request is never left silent while its vendor says nothing. `holdWriter.watch` runs beside each try: past `keepHeldAfter` (15s) it sends a held try's agent the stream's 200 and SSE comments (`keepAlive`), none of the try, so another candidate may still answer in the same stream; once the reply passes, it puts a comment between two of its lines after `keepaliveEvery` of quiet. It stops after `keepaliveLongest` (5 minutes) with nothing from the vendor. A try waiting for its key's or account's slot or for room in its minute (`lanes.take`, `rpms.wait`) is kept alive the same way (`keepQueued`), for as long as it waits; once the 200 is out, the try is held, and its failure or its queue turning it away reaches the agent as the stream's error (`failTo`), not a status (coeo91 on Discord). A wait shorter than `keepHeldAfter` sends nothing, so its 429 keeps its `Retry-After`. A proxy in front such as Cloudflare (125s) otherwise ends the request (#947). Gemini streams get no comments (#934). A vendor that then ignores `stream:true` and answers 200 with its reply whole has it sent as the agent's stream (`wholeAsStream`: Chat, Responses, Anthropic), not as the stream's error, while a stream under a type that says neither JSON nor a stream goes through as it comes (`sseStart` reads its first bytes); a 200 that is no reply of the protocol's fails as a translated reply that didn't stream does (502, "… did not stream"), for the next candidate to answer, and the last one's reaches the agent as the stream's error. - Claude's, GPT's and Gemini's reasoning is held (`refusesAfterThinking`, up to `holdThinking`, 4 minutes) so that a refusal after it goes to the next candidate unseen (#248). It is held only while one could: on the last candidate a refusal isn't asked again, so Claude's and GPT's reasoning streams as it comes there, and a refusal after it reaches the agent as the vendor's own. Held there, a lone relay's Claude or GPT showed Claude Code its thinking all at once with the text, and Codex none until then (iTianbao on X). Gemini's stays held on the last candidate too, for a reply that only reasoned to be asked again (#667). Any other model's reasoning is never held. -- A streamed reply stuck in a loop is ended, not let run to its output limit (#1359, Moody-Sin: WorkBuddy's DeepSeek v4.1 flash reasoned round "Now." "Writing." "Go." "Let me write." "OK." until max tokens while dsh and Pi waited). `holdWriter.Write` feeds every streamed try, passed through or translated, a key's, a built-in subscription's or a plugin's (a moved WorkBuddy's fetch in the plugin host, `TestLoopEndsWorkBuddysReply`), in each of the agent's protocols, to a `loopGuard` ([`loop_guard.go`](../../internal/gateway/loop_guard.go)). It cuts the reasoning and the text, each apart, into units, its lines and sentences (a list's `1.` or a decimal is no sentence's end), compared without case or a closing stop. A unit with no word in it (a letter run with no digit: not a brace, a rule, `0x00`, a hash or a row of numbers) isn't counted. When the last `loopWindow` (256) counted units of either are among no more than `loopDistinct` (24) different ones, the try is cut (`cutLoop`): what came with that write goes out, then the protocol's stream error (`streamError`, 502) naming the loop and quoting its units, and the vendor's request is cancelled. It falls back to nobody and rests nobody (`failLoop`); its usage record's error type is `reply_loop`. A cycle need not be exact. Code, tables, lists, logs, CSV and base64 differ line to line or have no words, and don't trip it (`TestLoopGuardLeavesOrdinaryRepliesAlone` feeds magpie's own sources and docs through it). A reply sent whole (no stream, or Gemini without `alt=sse`) isn't read. Settings › Models › Replies › *Stop looping replies* (`noLoopGuard`) turns it off. +- A streamed reply stuck in a loop is ended, not let run to its output limit (#1359, Moody-Sin: WorkBuddy's DeepSeek v4.1 flash reasoned round "Now." "Writing." "Go." "Let me write." "OK." until max tokens while dsh and Pi waited). `holdWriter.Write` feeds every streamed try, passed through or translated, a key's, a built-in subscription's or a plugin's (a moved WorkBuddy's fetch in the plugin host, `TestLoopEndsWorkBuddysReply`), in each of the agent's protocols, to a `loopGuard` ([`loop_guard.go`](../../internal/gateway/loop_guard.go)). It cuts the reasoning and the text, each apart, into units, its lines and sentences (a list's `1.` or a decimal is no sentence's end), compared without case or a closing stop. A unit with no word in it (a letter run with no digit: not a brace, a rule, `0x00`, a hash or a row of numbers) isn't counted. When the last `loopWindow` (256) counted units of either are among no more than `loopDistinct` (24) different ones, the try is cut (`cutLoop`): what came with that write goes out, then the protocol's stream error (`streamError`, 502) naming the loop and quoting its units, and the vendor's request is cancelled. It falls back to nobody and rests nobody (`failLoop`); its usage record's error type is `reply_loop`. A reply magpie translates into the agent's protocol (`streamTranslated`) whose *reasoning* loops before it has said a text or a call, and isn't a continuation, is asked again once in place instead (`loopReasks`, through `holdWriter.onLoop`; cdredfox on X, WorkBuddy AI's DeepSeek v4 flash): the agent keeps the reasoning it has, the next try's follows it in the same stream, as an empty reply's re-ask does (#667), and only a loop on that try ends the reply with the error (`TestReasoningLoopIsAskedAgain`). A loop in the text, or after a text or call, is ended at once (`TestLoopAfterTextIsNotAskedAgain`), and so is a Chat stream passed through as it comes, whose bytes aren't magpie's to splice. A cycle need not be exact. Code, tables, lists, logs, CSV and base64 differ line to line or have no words, and don't trip it (`TestLoopGuardLeavesOrdinaryRepliesAlone` feeds magpie's own sources and docs through it). A reply sent whole (no stream, or Gemini without `alt=sse`) isn't read. Settings › Models › Replies › *Stop looping replies* (`noLoopGuard`) turns it off. - The order of the steps a request's accounts go through, how much the accounts' own order (*Make first*) counts under each routing, and how often an allowance is read are described for users in [How magpie picks an account](../reference.md#how-magpie-picks-an-account). Smart compares renewals window by window, biggest first, each truncated to the hour (`weighRouted`), so the order decides only when every window falls in the same hour. Affinity keeps a conversation on an account at 90–97%: only `SpentShareOf` (98%, 100% In order) stops it (`affine`), while the 90% tier (`lowShare`) orders only conversations nobody has answered. - A resting candidate is tried last, never dropped (`restLast`). A lone candidate is tried even while it rests, because there is no other. - Routing decides who goes first only when nobody has answered the conversation yet. Affinity keeps the answerer while its cache is worth it. Auto mode keeps it across turns while at least `cacheWorth` (1024) tokens were read from the cache and the cache isn't cold (`cacheCold`, 5 minutes). diff --git a/internal/gateway/continuation.go b/internal/gateway/continuation.go index c6623d05..35919b46 100644 --- a/internal/gateway/continuation.go +++ b/internal/gateway/continuation.go @@ -47,6 +47,11 @@ func emptyRetries(w http.ResponseWriter, ctx context.Context) int { return streamRetries } +// loopReasks is how many times a reply whose reasoning was stuck in a +// loop (loopGuard) before it said anything is asked again in place, rather +// than ended with the loop's error. +const loopReasks = 1 + // errStreamCut ends a reply's read once its error event is in hand, // without waiting on a vendor that keeps the connection open after it. var errStreamCut = errors.New("stream cut mid-reply") @@ -260,7 +265,30 @@ func (s *Server) streamTranslated(w http.ResponseWriter, r *http.Request, p prov said, stop := false, "" var kept []Event // the reply's end, while nothing is said in it var before, this Usage // what the tries before this one billed, and this try + // spoke: the reply has said something, a text or a call; relooped: this + // try's reasoning was found stuck in a loop before it had, and it is + // let go to ask again (loopReasks) + var spoke, relooped bool + reasks := 0 + if h, ok := w.(*holdWriter); ok { + h.onLoop(func(t loopTrip) bool { + // only reasoning nobody acts on: a text or a call the agent + // has would be said twice, and a continuation's prefill would + // be asked to go on with the loop + if !t.reasoning || spoke || cont.resume || reasks >= loopReasks || r.Context().Err() != nil { + return false + } + reasks++ + relooped = true + return true + }) + defer h.onLoop(nil) + } emit := func(ev Event) { + if relooped { + return // the rest of a try let go for its loop + } + spoke = spoke || saysSomething(ev) switch ev.Kind { case KError: if ev.Code == "" && cont.possible() && !inGroupTry(r.Context()) { @@ -385,7 +413,7 @@ func (s *Server) streamTranslated(w http.ResponseWriter, r *http.Request, p prov if err := dec(data, attemptSee); err != nil { return err } - if cut { + if cut || relooped { return errStreamCut } return nil @@ -394,7 +422,7 @@ func (s *Server) streamTranslated(w http.ResponseWriter, r *http.Request, p prov if errors.Is(serr, errStreamCut) { serr = nil } - if serr == nil && failed == "" && !ended && actual == provider.Anthropic && r.Context().Err() == nil { + if serr == nil && failed == "" && !ended && !relooped && actual == provider.Anthropic && r.Context().Err() == nil { // an Anthropic stream that just stopped — no // stop_reason, no message_stop — is a reply cut // short, not a finished one: a relay's (蓝猫 on @@ -404,6 +432,23 @@ func (s *Server) streamTranslated(w http.ResponseWriter, r *http.Request, p prov } } } + if relooped { + // the model's reasoning ran into a loop before it said a word + // (#1359), which DeepSeek v4 flash on WorkBuddy did now and + // then in a long turn; asked again, it mostly doesn't. The + // agent keeps the reasoning it has and the next try's follows + // it, as an empty reply's does (#667); a loop on that try too + // ends the reply with the loop's error. + relooped = false + before, this = before.plus(this, false), Usage{} + kept, stop = nil, "" + enc.keepalive() + select { + case <-time.After(retryPause): + case <-r.Context().Done(): + } + continue + } if serr == nil && failed == "" { if held != nil { held() diff --git a/internal/gateway/fallback.go b/internal/gateway/fallback.go index 3dce590a..7e33f6c4 100644 --- a/internal/gateway/fallback.go +++ b/internal/gateway/fallback.go @@ -1099,6 +1099,19 @@ type holdWriter struct { // ended there loop *loopGuard looped string + // reask, when the try's reply can be asked again in place + // (streamTranslated), is told of a loop first, and says whether the + // reply is asked again instead of ended: the guard then reads the + // next try's afresh + reask func(loopTrip) bool +} + +// onLoop sets what is told of a loop before it ends the reply (reask), +// nil for nobody. +func (h *holdWriter) onLoop(f func(loopTrip) bool) { + h.mu.Lock() + defer h.mu.Unlock() + h.reask = f } // errSlowStart is what a try's writes get once it was let go for taking @@ -1190,7 +1203,10 @@ func (h *holdWriter) Write(b []byte) (int, error) { } if h.loop != nil && !h.whole && h.status < 400 { if t, ok := h.loop.feed(b); ok { - return h.cutLoop(b, t) + if h.reask == nil || !h.reask(t) { + return h.cutLoop(b, t) + } + h.loop = &loopGuard{} } } h.see(b) diff --git a/internal/gateway/loop_guard_test.go b/internal/gateway/loop_guard_test.go index ca1f3b3a..e37b331a 100644 --- a/internal/gateway/loop_guard_test.go +++ b/internal/gateway/loop_guard_test.go @@ -264,8 +264,11 @@ func TestLoopGuardLeavesOrdinaryRepliesAlone(t *testing.T) { type loopingVendor struct { reasoning string // what it reasons, else the reporter's loop lines int // 0: loop until let go + loops int // how many of its requests loop, 0 for every one + said string // a text it says before its reasoning loops + asked atomic.Int64 sent atomic.Int64 - letGo chan bool + letGo chan bool // each request's, sized for every one it gets } // loopMost is how many lines of its loop loopingVendor sends before it @@ -274,6 +277,7 @@ const loopMost = 2000 func (v *loopingVendor) ServeHTTP(w http.ResponseWriter, r *http.Request) { io.ReadAll(r.Body) + asked := v.asked.Add(1) w.Header().Set("Content-Type", "text/event-stream") w.WriteHeader(200) f := w.(http.Flusher) @@ -286,7 +290,12 @@ func (v *loopingVendor) ServeHTTP(w http.ResponseWriter, r *http.Request) { return r.Context().Err() == nil } send(chatEvent("reasoning_content", reasoningBefore)) - if v.reasoning != "" { + if v.said != "" { + send(chatEvent("content", v.said)) + } + if v.loops > 0 && asked > int64(v.loops) { + send(chatEvent("reasoning_content", "So: a case for .toml in parse(), and its test.")) + } else if v.reasoning != "" { for _, c := range pieces(v.reasoning, 11) { if !send(chatEvent("reasoning_content", c)) { v.letGo <- true @@ -321,19 +330,23 @@ func (v *loopingVendor) ServeHTTP(w http.ResponseWriter, r *http.Request) { } // loopAsks are a streamed request on each of the agents' protocols, and -// how each one's stream says it ended well, and with magpie's error. +// how each one's stream says it ended well, and with magpie's error; asks +// is how many times the vendor is asked when every reply loops: a reply +// translated to the agent's protocol is asked again once (loopReasks), a +// Chat stream passed through as it comes is not. var loopAsks = []struct { name, path, body string done, failed string + asks int }{ {"chat", "/v1/chat/completions", `{"model":"ds/deepseek-v4-max","stream":true,"messages":[{"role":"user","content":"add TOML"}]}`, - "data: [DONE]", `data: {"error":{"message":"magpie ended this reply`}, + "data: [DONE]", `data: {"error":{"message":"magpie ended this reply`, 1}, {"anthropic", "/v1/messages", `{"model":"ds/deepseek-v4-max","max_tokens":32000,"stream":true,"messages":[{"role":"user","content":"add TOML"}]}`, - `"type":"message_stop"`, "event: error\ndata: {\"error\":{"}, + `"type":"message_stop"`, "event: error\ndata: {\"error\":{", 2}, {"responses", "/v1/responses", `{"model":"ds/deepseek-v4-max","stream":true,"input":"add TOML"}`, - `"type":"response.completed"`, `"type":"response.failed"`}, + `"type":"response.completed"`, `"type":"response.failed"`, 2}, {"gemini", "/v1beta/models/ds/deepseek-v4-max:streamGenerateContent?alt=sse", `{"contents":[{"role":"user","parts":[{"text":"add TOML"}]}]}`, - `"finishReason":"STOP"`, `"status":"UNAVAILABLE"`}, + `"finishReason":"STOP"`, `"status":"UNAVAILABLE"`, 2}, } func deepseekOn(t *testing.T, v *loopingVendor) string { @@ -382,17 +395,18 @@ func askStream(t *testing.T, url, body string) string { func TestLoopEndsTheReplyOnEveryProtocol(t *testing.T) { for _, x := range loopAsks { t.Run(x.name, func(t *testing.T) { - v := &loopingVendor{letGo: make(chan bool, 1)} + v := &loopingVendor{letGo: make(chan bool, 4)} gw := deepseekOn(t, v) - loopCut(t, v, askStream(t, gw+x.path, x.body), x.done, x.failed, "ds") + loopCut(t, v, askStream(t, gw+x.path, x.body), x.done, x.failed, "ds", x.asks) }) } } // loopCut checks what the agent got of a looping reply, got, and what came -// of it: the stream's error, not its end (done); the vendor let go; nobody -// resting; the usage record's reply_loop, for provider. -func loopCut(t *testing.T, v *loopingVendor, got, done, failed, provider string) { +// of it: the stream's error, not its end (done); the vendor asked asks +// times and let go each time; nobody resting; the usage record's +// reply_loop, for provider. +func loopCut(t *testing.T, v *loopingVendor, got, done, failed, provider string, asks int) { t.Helper() if !strings.Contains(got, failed) || !strings.Contains(got, "stuck in a loop") || strings.Contains(got, done) { t.Fatalf("the stream doesn't end with magpie's error:\n%s", got[max(len(got)-1500, 0):]) @@ -400,19 +414,24 @@ func loopCut(t *testing.T, v *loopingVendor, got, done, failed, provider string) if !strings.Contains(got, "Producing") { t.Fatal("the reasoning before the loop was found didn't reach the agent") } - select { - case letGo := <-v.letGo: - if !letGo { - t.Fatalf("the vendor's request was not let go (%d lines)", v.sent.Load()) + for range asks { + select { + case letGo := <-v.letGo: + if !letGo { + t.Fatalf("the vendor's request was not let go (%d lines)", v.sent.Load()) + } + case <-time.After(30 * time.Second): + t.Fatal("the vendor's request was not let go") } - case <-time.After(30 * time.Second): - t.Fatal("the vendor's request was not let go") + } + if n := v.asked.Load(); n != int64(asks) { + t.Fatalf("the vendor was asked %d times, want %d", n, asks) } // the guard cut within its window of the loop's start: what reached // the agent of the loop is that window and at most one 32K read of the // vendor's stream more, which the cut's write sends whole. It doesn't // depend on how far the vendor got ahead of the gateway (#1443). - if n := loopLines(got); n > loopWindow+(32<<10)/100 { + if n := loopLines(got); n > asks*(loopWindow+(32<<10)/100) { t.Fatalf("the loop was cut only after %d of its lines reached the agent", n) } restingUntil.Lock() @@ -472,7 +491,7 @@ func TestLoopEndsWorkBuddysReply(t *testing.T) { for _, rt := range routes { for _, x := range loopAsks { t.Run(rt.name+"/"+x.name, func(t *testing.T) { - v := &loopingVendor{letGo: make(chan bool, 1)} + v := &loopingVendor{letGo: make(chan bool, 4)} id := rt.on(t, v) m := model if rt.name == "plugin" { @@ -482,12 +501,57 @@ func TestLoopEndsWorkBuddysReply(t *testing.T) { t.Cleanup(gw.Close) path := strings.Replace(x.path, "ds/deepseek-v4-max", id+"/"+m, 1) body := strings.Replace(x.body, `"model":"ds/deepseek-v4-max"`, `"model":"`+id+"/"+m+`","reasoning_effort":"max"`, 1) - loopCut(t, v, askStream(t, gw.URL+path, body), x.done, x.failed, id) + loopCut(t, v, askStream(t, gw.URL+path, body), x.done, x.failed, id, x.asks) }) } } } +// A reply whose reasoning loops before it says anything is asked again +// once, in place, when magpie translates it to the agent's protocol: the +// agent keeps the reasoning it has, the next try's follows it, and the +// reply ends whole, the loop no error (cdredfox on X: WorkBuddy's DeepSeek +// v4 flash now and then looping in a long turn, the agent's turn ended). +// A Chat stream passed through as it comes is ended with the loop's error, +// as before: its events aren't magpie's to splice. +func TestReasoningLoopIsAskedAgain(t *testing.T) { + for _, x := range loopAsks { + t.Run(x.name, func(t *testing.T) { + v := &loopingVendor{loops: 1, letGo: make(chan bool, 4)} + gw := deepseekOn(t, v) + got := askStream(t, gw+x.path, x.body) + if x.asks == 1 { + loopCut(t, v, got, x.done, x.failed, "ds", 1) + return + } + if !strings.Contains(got, x.done) || strings.Contains(got, "magpie ended this reply") || !strings.Contains(got, "parse() now takes .toml") { + t.Fatalf("the reply asked again didn't end whole:\n%s", got[max(len(got)-1500, 0):]) + } + if !strings.Contains(got, "Producing") || !strings.Contains(got, "a case for .toml in parse()") { + t.Fatal("the agent didn't get both tries' reasoning") + } + if n := v.asked.Load(); n != 2 { + t.Fatalf("the vendor was asked %d times, want 2", n) + } + if first, second := <-v.letGo, <-v.letGo; !first || second { + t.Fatalf("let go: the looping try %v, the next %v", first, second) + } + }) + } +} + +// A reply that has said something before its reasoning loops is ended, not +// asked again: the agent has its text, which a second try would say again. +func TestLoopAfterTextIsNotAskedAgain(t *testing.T) { + for _, x := range loopAsks { + t.Run(x.name, func(t *testing.T) { + v := &loopingVendor{said: "Looking at parse() first.", letGo: make(chan bool, 4)} + gw := deepseekOn(t, v) + loopCut(t, v, askStream(t, gw+x.path, x.body), x.done, x.failed, "ds", 1) + }) + } +} + // A long reply that doesn't loop goes through whole, on every protocol: // magpie's own gateway.go as the model's reasoning, 4,900 lines of code // with its braces, returns and comments. @@ -502,7 +566,7 @@ func TestLongReplyThatDoesNotLoopGoesThrough(t *testing.T) { src = src[:bytes.LastIndexByte(src[:40<<10], '\n')+1] for _, x := range loopAsks { t.Run(x.name, func(t *testing.T) { - v := &loopingVendor{reasoning: string(src), letGo: make(chan bool, 1)} + v := &loopingVendor{reasoning: string(src), letGo: make(chan bool, 4)} gw := deepseekOn(t, v) got := askStream(t, gw+x.path, x.body) if !strings.Contains(got, x.done) || strings.Contains(got, "magpie ended this reply") || !strings.Contains(got, "parse() now takes .toml") { @@ -518,7 +582,7 @@ func TestLongReplyThatDoesNotLoopGoesThrough(t *testing.T) { // With Stop looping replies off (Settings' NoLoopGuard), a loop runs on // as the vendor sends it, to its end. func TestLoopGuardOff(t *testing.T) { - v := &loopingVendor{lines: 3000, letGo: make(chan bool, 1)} + v := &loopingVendor{lines: 3000, letGo: make(chan bool, 4)} gw := deepseekOn(t, v) if err := settings.Save(settings.Settings{NoLoopGuard: true}); err != nil { t.Fatal(err)