diff --git a/internal/gateway/fallback.go b/internal/gateway/fallback.go index 1297fb95..79a6c421 100644 --- a/internal/gateway/fallback.go +++ b/internal/gateway/fallback.go @@ -15,6 +15,7 @@ package gateway import ( "bytes" + "encoding/json" "net/http" "regexp" "sort" @@ -316,21 +317,59 @@ func restOf(id string) (Rest, bool) { // status code doesn't: some answer 400 or 403 with it. var quotaWords = regexp.MustCompile(`(?i)quota|insufficient|balance|credit|billing|exceeded|rate.?limit|usage.?limit|limit.?reached|hit your .*limit|limit.{0,24}resets|too many requests|overloaded|余额|额度|欠费|限流|频率|套餐|用量|上限`) +// unservedWords are how a vendor says the model isn't one it serves this +// key, or this way — words another provider, or key, may not answer with. +var unservedWords = regexp.MustCompile(`(?i)model.{0,80}(not (supported|accessible|available|found|enabled|allowed)|unsupported|does ?n[o']t exist|unknown|invalid)|(no such|unknown|invalid|unsupported) model|model_not_found|模型.{0,12}(不存在|不支持|无权|未开通)`) + // retryable says whether another provider may do better with a request -// that failed this way. +// that failed this way: the vendor was busy, out of quota or failing, or +// this key or provider can't serve it — not the request itself at fault. func retryable(status int, body []byte) bool { switch { - case status == 402, status == 408, status == 429, status >= 500: + case status == 401, status == 402, status == 403, status == 404, status == 408, status == 429, status >= 500: return true - case status == 400, status == 401, status == 403: - return quotaWords.Match(body) + case status == 400, status == 422: + return quotaWords.Match(body) || unservedWords.Match(body) } return false } +const ( + // lastRetries is how many times the last one left is tried again after + // a failure that passes — a busy vendor, a dropped connection. + lastRetries = 2 + // longestPause is the longest the vendor's Retry-After is waited for + // before that; longer, and the agent gets the error. + longestPause = 8 * time.Second +) + +// retryPause is the first pause before the last one left is tried again; +// each pause after is twice the one before. +var retryPause = time.Second + +// passing says whether a failure is one that may be gone a moment later, +// and how long to wait before trying the same one again. +func passing(status int, header http.Header, again int) (time.Duration, bool) { + wait := retryPause << again + if d := retryAfter(header, time.Now()); d > 0 { + wait = d + } + switch { + case wait > longestPause: + return 0, false + case status == 408, status == 500, status == 502, status == 503, status == 504, status == 529: + return wait, true + case status == 429: + return wait, retryAfter(header, time.Now()) > 0 // a rate limit says when + } + return 0, false +} + // holdWriter keeps an error reply back while another provider may still // answer: headers and body wait until release, or are dropped for the next -// try. Anything else goes straight through. +// try. Anything else goes straight through — but for a stream, only once +// its first content comes: an error before that is a failure another may +// answer, as an error status is. type holdWriter struct { w http.ResponseWriter hold bool @@ -338,6 +377,12 @@ type holdWriter struct { status int passing bool held bytes.Buffer + + stream bool // a stream held until its first content + since time.Time // when it began + scanned int // how much of held has been read as events + failure int // the status the stream's error stands for + failMsg string } func newHoldWriter(w http.ResponseWriter, hold bool) *holdWriter { @@ -354,6 +399,10 @@ func (h *holdWriter) WriteHeader(code int) { if h.hold && code >= 400 { return } + if h.hold && strings.HasPrefix(h.header.Get("Content-Type"), "text/event-stream") { + h.stream, h.since = true, time.Now() + return + } h.pass() } @@ -370,10 +419,54 @@ func (h *holdWriter) Write(b []byte) (int, error) { if h.status == 0 { h.WriteHeader(http.StatusOK) } - if !h.passing { - return h.held.Write(b) + if h.passing { + return h.w.Write(b) } - return h.w.Write(b) + n, err := h.held.Write(b) + if h.stream && h.failure == 0 { + h.scan() + } + return n, err +} + +// holdLongest is the longest a stream is held waiting for its first +// content, and holdMost the most of it. +const ( + holdLongest = 15 * time.Second + holdMost = 1 << 20 +) + +// scan reads the held stream's events so far: an error before any content +// fails it; content, or waiting too long for it, lets it through. +func (h *holdWriter) scan() { + for { + rest := h.held.Bytes()[h.scanned:] + end := eventEnd(rest) + if end < 0 { + break + } + h.scanned += end + switch kind, status, msg := streamEvent(rest[:end]); kind { + case eventLead: + continue + case eventError: + h.failure, h.failMsg = status, msg + return + } + h.flow() + return + } + if h.held.Len() > holdMost || time.Since(h.since) > holdLongest { + h.flow() + } +} + +// flow lets a held stream through, and what follows it. +func (h *holdWriter) flow() { + h.pass() + h.w.Write(h.held.Bytes()) + h.held.Reset() + h.Flush() } func (h *holdWriter) Flush() { @@ -382,9 +475,26 @@ func (h *holdWriter) Flush() { } } +// code is the status a try failed with: the reply's, or its stream's +// error's. +func (h *holdWriter) code() int { + if h.failure != 0 { + return h.failure + } + return h.status +} + +// errBody is what the vendor said of the failure. +func (h *holdWriter) errBody() []byte { + if h.failure != 0 { + return []byte(h.failMsg) + } + return h.held.Bytes() +} + // failed reports a held error another provider could answer instead. func (h *holdWriter) failed() bool { - return !h.passing && h.status >= 400 && retryable(h.status, h.held.Bytes()) + return !h.passing && h.code() >= 400 && retryable(h.code(), h.errBody()) } // release sends a held reply after all: nobody else is left to try. @@ -394,4 +504,118 @@ func (h *holdWriter) release() { } h.pass() h.w.Write(h.held.Bytes()) + h.Flush() +} + +// eventEnd is where the first whole event in b ends, or -1. +func eventEnd(b []byte) int { + for i := 0; i+1 < len(b); i++ { + if b[i] != '\n' { + continue + } + if b[i+1] == '\n' { + return i + 2 + } + if b[i+1] == '\r' && i+2 < len(b) && b[i+2] == '\n' { + return i + 3 + } + } + return -1 +} + +const ( + eventContent = iota + eventLead // what comes before a reply's content: a start, a ping + eventError +) + +// streamEvent says what one server-sent event of a reply is, in any of the +// protocols an agent speaks: the start of a reply, its content, or an +// error — with the status that error stands for. +func streamEvent(ev []byte) (kind, status int, msg string) { + var name string + var data []byte + for _, ln := range bytes.Split(ev, []byte("\n")) { + ln = bytes.TrimRight(ln, "\r") + switch { + case bytes.HasPrefix(ln, []byte("event:")): + name = strings.TrimSpace(string(ln[6:])) + case bytes.HasPrefix(ln, []byte("data:")): + data = append(data, bytes.TrimSpace(ln[5:])...) + } + } + if name == "" && len(data) == 0 { + return eventLead, 0, "" // a comment, a keep-alive + } + var v struct { + Type string `json:"type"` + Error json.RawMessage `json:"error"` + Message json.RawMessage `json:"message"` // a Responses error's; an Anthropic start's is the message + Response struct { + Error json.RawMessage `json:"error"` + } `json:"response"` + Choices *[]struct { + Delta map[string]any `json:"delta"` + FinishReason *string `json:"finish_reason"` + } `json:"choices"` + } + if json.Unmarshal(data, &v) != nil { + return eventContent, 0, "" // [DONE], or what isn't ours to read + } + typ := v.Type + if typ == "" { + typ = name + } + errOf := func(raw json.RawMessage) (int, int, string) { + msg := string(raw) + var m string + if json.Unmarshal(v.Message, &m) == nil && m != "" { + msg += " " + m + } + return eventError, streamStatus(msg), msg + } + switch { + case typ == "error" || typ == "response.failed": + if len(v.Response.Error) > 0 && string(v.Response.Error) != "null" { + return errOf(v.Response.Error) + } + return errOf(v.Error) + case len(v.Error) > 0 && string(v.Error) != "null": + return errOf(v.Error) + case typ == "ping", typ == "message_start", typ == "response.created", typ == "response.in_progress", typ == "response.queued": + return eventLead, 0, "" + case typ == "" && v.Choices != nil: + // a Chat chunk: the first says only who speaks + for _, c := range *v.Choices { + if c.FinishReason != nil { + return eventContent, 0, "" + } + for k, x := range c.Delta { + if k != "role" && x != nil && x != "" { + return eventContent, 0, "" + } + } + } + return eventLead, 0, "" + } + return eventContent, 0, "" +} + +// streamStatus is the status an error in a stream stands for, as a vendor +// would have answered it before streaming. +func streamStatus(msg string) int { + l := strings.ToLower(msg) + switch { + case strings.Contains(l, "overloaded"): + return 529 + case creditWords.MatchString(msg) && !strings.Contains(l, "rate"): + return 402 + case strings.Contains(l, "rate_limit"), strings.Contains(l, "rate limit"), strings.Contains(l, "too many"), quotaWords.MatchString(msg): + return 429 + case unservedWords.MatchString(msg): + return 400 + case strings.Contains(l, "invalid"), strings.Contains(l, "context_length"), strings.Contains(l, "too long"): + return 400 + } + return 502 } diff --git a/internal/gateway/gateway.go b/internal/gateway/gateway.go index afbec564..a149f759 100644 --- a/internal/gateway/gateway.go +++ b/internal/gateway/gateway.go @@ -401,21 +401,48 @@ func (s *Server) serve(w http.ResponseWriter, r *http.Request, from provider.Pro } tr := s.trace.begin(Route{Time: start, Agent: call.Agent, Model: call.Model, Provider: p.ID, Group: group, Affinity: aff, Order: pl.order, Left: pl.left}) var skipped []string - for i, c := range cands { - last := i == len(cands)-1 || r.Context().Err() != nil - hw := newHoldWriter(w, !last) + again := 0 // times the last one left has been tried again + for i := 0; i < len(cands); i++ { + c := cands[i] + last := i == len(cands)-1 + hw := newHoldWriter(w, !last || again < lastRetries) call.Provider, call.To, call.Usage = c.p.ID, "", Usage{} began := time.Now() s.trace.update(tr, func(t *Route) { t.Tries = append(t.Tries, Try{ID: c.rest, Start: began}) }) call.Status, call.Error = s.attempt(hw, r, from, c.p, c.model, body, &call) + if hw.failure != 0 { // the stream failed before any of it was sent + call.Status, call.Error = hw.failure, c.p.Name+": "+hw.failMsg + } try := Try{ID: c.rest, Start: began, Done: true, Status: call.Status, Millis: time.Since(began).Milliseconds(), Error: call.Error} + if r.Context().Err() != nil { + // the agent went away: nobody failed, and nobody else is asked + call.Status, call.Error = 499, "the agent canceled the request" + try.Status, try.Error, try.Fail = call.Status, call.Error, failCanceled + s.trace.update(tr, func(t *Route) { t.Tries[len(t.Tries)-1] = try }) + break + } if !last && hw.failed() { - rest := s.restAfter(c, hw.status, hw.header, hw.held.Bytes()) + rest := s.restAfter(c, hw.code(), hw.header, hw.errBody()) try.Fail, try.Rest = rest.Why, &rest s.trace.update(tr, func(t *Route) { t.Tries[len(t.Tries)-1] = try }) skipped = append(skipped, c.label()+": "+call.Error) continue } + if wait, ok := passing(hw.code(), hw.header, again); ok && again < lastRetries && hw.failed() { + // nobody else is left: the same one again, after a moment + try.Fail, try.Again = failure(hw.code(), hw.errBody()), wait.Milliseconds() + s.trace.update(tr, func(t *Route) { t.Tries[len(t.Tries)-1] = try }) + skipped = append(skipped, c.label()+": "+call.Error) + again++ + select { + case <-time.After(wait): + i-- + continue + case <-r.Context().Done(): + call.Status, call.Error = 499, "the agent canceled the request" + } + break + } hw.release() model = c.model if call.Status < 400 { diff --git a/internal/gateway/retry_test.go b/internal/gateway/retry_test.go new file mode 100644 index 00000000..650709e6 --- /dev/null +++ b/internal/gateway/retry_test.go @@ -0,0 +1,219 @@ +package gateway + +import ( + "context" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/yetone/magpie/internal/provider" +) + +func init() { retryPause = time.Millisecond } + +// scripted answers each request with the next of its replies, the last +// one over and over. +type scripted struct { + replies []reply + n int + hang chan struct{} // when set, the first request waits on it +} + +type reply struct { + code int + ctype string + body string +} + +func (s *scripted) ServeHTTP(w http.ResponseWriter, r *http.Request) { + io.ReadAll(r.Body) + i := min(s.n, len(s.replies)-1) + s.n++ + if s.hang != nil && i == 0 { + <-s.hang + } + x := s.replies[i] + if x.ctype == "" { + x.ctype = "application/json" + } + w.Header().Set("Content-Type", x.ctype) + if x.code != 0 { + w.WriteHeader(x.code) + } + io.WriteString(w, x.body) +} + +const chatOK = `{"id":"ok","choices":[{"index":0,"message":{"role":"assistant","content":"hello"},"finish_reason":"stop"}]}` + +func scriptedOn(t *testing.T, id string, proto provider.Protocol, s *scripted) { + t.Helper() + up := httptest.NewServer(s) + t.Cleanup(up.Close) + p := provider.Provider{ID: id, Name: strings.ToUpper(id), Key: "k", Models: []string{"m"}} + switch proto { + case provider.Anthropic: + p.Anthropic = up.URL + default: + p.Chat = up.URL + "/v1" + } + if err := provider.Save(p); err != nil { + t.Fatal(err) + } +} + +// A provider that can't serve the model — turned away, not the request at +// fault — hands the request to the next member rather than to the agent. +func TestGroupFailsOverAnUnservedModel(t *testing.T) { + fresh(t) + a := &scripted{replies: []reply{{400, "", `{"error":{"message":"model \"gpt-6-sol\" is not accessible via the /chat/completions endpoint","code":"unsupported_api_for_model"}}`}}} + b := &scripted{replies: []reply{{200, "", chatOK}}} + scriptedOn(t, "a", provider.Chat, a) + scriptedOn(t, "b", provider.Chat, b) + if err := provider.SaveGroup(provider.Group{Name: "G", Members: []string{"a/m", "b/m"}, Routing: provider.Ordered}); err != nil { + t.Fatal(err) + } + s := New() + code, body := postAs(t, s, "", `{"model":"group/g","messages":[{"role":"user","content":"hi"}]}`) + if code != 200 || !strings.Contains(body, "hello") || a.n != 1 || b.n != 1 { + t.Fatalf("%d %s (a %d, b %d)", code, body, a.n, b.n) + } + for _, x := range []struct { + status int + body string + want bool + }{ + {401, `{"error":{"message":"invalid api key"}}`, true}, + {404, `{"error":{"message":"not found"}}`, true}, + {400, `{"error":{"message":"The model gpt-x does not exist"}}`, true}, + {400, `{"error":{"message":"messages: field required"}}`, false}, + {400, `{"error":{"message":"This model's maximum context length is 128000 tokens"}}`, false}, + } { + if got := retryable(x.status, []byte(x.body)); got != x.want { + t.Errorf("retryable(%d, %s) = %v", x.status, x.body, got) + } + } +} + +// An error a stream begins with, before any of its content, is a failure +// the next one answers, as an error status is; one that comes after +// content has begun reaches the agent as it came. +func TestStreamErrorBeforeContentFailsOver(t *testing.T) { + fresh(t) + overloaded := sse(`event: message_start`+"\n"+`data: {"type":"message_start","message":{"id":"m1","role":"assistant","content":[],"usage":{"input_tokens":3}}}`, + `event: ping`+"\n"+`data: {"type":"ping"}`, + `event: error`+"\n"+`data: {"type":"error","error":{"type":"overloaded_error","message":"Overloaded"}}`) + good := sse(`event: message_start`+"\n"+`data: {"type":"message_start","message":{"id":"m2","role":"assistant","content":[],"usage":{"input_tokens":3}}}`, + `event: content_block_start`+"\n"+`data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}`, + `event: content_block_delta`+"\n"+`data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"from b"}}`, + `event: message_stop`+"\n"+`data: {"type":"message_stop"}`) + a := &scripted{replies: []reply{{200, "text/event-stream", overloaded}}} + b := &scripted{replies: []reply{{200, "text/event-stream", good}}} + scriptedOn(t, "a", provider.Anthropic, a) + scriptedOn(t, "b", provider.Anthropic, b) + if err := provider.SaveGroup(provider.Group{Name: "G", Members: []string{"a/m", "b/m"}, Routing: provider.Ordered}); err != nil { + t.Fatal(err) + } + s := New() + send := func() (int, string) { + rec := httptest.NewRecorder() + s.Handler().ServeHTTP(rec, httptest.NewRequest("POST", "/v1/messages", + strings.NewReader(`{"model":"group/g","max_tokens":10,"stream":true,"messages":[{"role":"user","content":"hi"}]}`))) + return rec.Code, rec.Body.String() + } + code, body := send() + if code != 200 || !strings.Contains(body, "from b") || strings.Contains(body, "Overloaded") || strings.Contains(body, `"m1"`) { + t.Fatalf("%d %s %+v", code, body, s.trace.routes[len(s.trace.routes)-1]) + } + r := s.trace.routes[len(s.trace.routes)-1] + if len(r.Tries) != 2 || r.Tries[0].Status != 529 || r.Tries[0].Rest == nil { + t.Fatalf("tries: %+v", r.Tries) + } + + // after content, the error is the agent's to see + fresh(t) + late := sse(`event: message_start`+"\n"+`data: {"type":"message_start","message":{"id":"m1","role":"assistant","content":[]}}`, + `event: content_block_delta`+"\n"+`data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"half"}}`, + `event: error`+"\n"+`data: {"type":"error","error":{"type":"overloaded_error","message":"Overloaded"}}`) + a = &scripted{replies: []reply{{200, "text/event-stream", late}}} + b = &scripted{replies: []reply{{200, "text/event-stream", good}}} + scriptedOn(t, "a", provider.Anthropic, a) + scriptedOn(t, "b", provider.Anthropic, b) + provider.SaveGroup(provider.Group{Name: "G", Members: []string{"a/m", "b/m"}, Routing: provider.Ordered}) + s = New() + if code, body := send(); code != 200 || !strings.Contains(body, "half") || !strings.Contains(body, "Overloaded") || b.n != 0 { + t.Fatalf("late: %d %s (b %d)", code, body, b.n) + } + + for _, x := range []struct { + ev string + kind int + }{ + {`: keep-alive`, eventLead}, + {`data: {"id":"c","choices":[{"index":0,"delta":{"role":"assistant","content":""}}]}`, eventLead}, + {`data: {"id":"c","choices":[{"index":0,"delta":{"content":"hi"}}]}`, eventContent}, + {`data: {"error":{"message":"upstream failed"}}`, eventError}, + {`event: response.created` + "\n" + `data: {"type":"response.created","response":{"id":"r"}}`, eventLead}, + {`event: response.failed` + "\n" + `data: {"type":"response.failed","response":{"error":{"code":"server_error","message":"boom"}}}`, eventError}, + {`event: response.output_text.delta` + "\n" + `data: {"type":"response.output_text.delta","delta":"hi"}`, eventContent}, + {`data: [DONE]`, eventContent}, + } { + if k, _, _ := streamEvent([]byte(x.ev)); k != x.kind { + t.Errorf("streamEvent(%s) = %d, want %d", x.ev, k, x.kind) + } + } +} + +// The last one left — or the only one — is tried again after a failure +// that may pass, a moment later; one that won't is the agent's at once. +func TestLastOneLeftIsTriedAgain(t *testing.T) { + fresh(t) + a := &scripted{replies: []reply{{503, "", `{"error":{"message":"upstream connect error"}}`}, {502, "", `{"error":{"message":"bad gateway"}}`}, {200, "", chatOK}}} + scriptedOn(t, "a", provider.Chat, a) + s := New() + code, body := postAs(t, s, "", `{"model":"a/m","messages":[{"role":"user","content":"hi"}]}`) + if code != 200 || !strings.Contains(body, "hello") || a.n != 3 { + t.Fatalf("%d %s (tried %d times)", code, body, a.n) + } + r := s.trace.routes[len(s.trace.routes)-1] + if len(r.Tries) != 3 || r.Tries[0].Again == 0 || r.Tries[1].Again == 0 || r.Tries[2].Status != 200 { + t.Fatalf("tries: %+v", r.Tries) + } + + fresh(t) + a = &scripted{replies: []reply{{503, "", `{"error":{"message":"down"}}`}}} + scriptedOn(t, "a", provider.Chat, a) + if code, _ := postAs(t, New(), "", `{"model":"a/m","messages":[{"role":"user","content":"hi"}]}`); code != 503 || a.n != 1+lastRetries { + t.Fatalf("gave up with %d after %d tries", code, a.n) + } + + fresh(t) + a = &scripted{replies: []reply{{400, "", `{"error":{"message":"messages: field required"}}`}}} + scriptedOn(t, "a", provider.Chat, a) + if code, _ := postAs(t, New(), "", `{"model":"a/m","messages":[{"role":"user","content":"hi"}]}`); code != 400 || a.n != 1 { + t.Fatalf("a request at fault: %d after %d tries", code, a.n) + } +} + +// An agent that goes away mid-request isn't a provider failing: nobody +// rests for it and nobody else is asked. +func TestCanceledRequestRestsNobody(t *testing.T) { + fresh(t) + a := &scripted{replies: []reply{{200, "", chatOK}}, hang: make(chan struct{})} + b := &scripted{replies: []reply{{200, "", chatOK}}} + scriptedOn(t, "a", provider.Chat, a) + scriptedOn(t, "b", provider.Chat, b) + provider.SaveGroup(provider.Group{Name: "G", Members: []string{"a/m", "b/m"}, Routing: provider.Ordered}) + s := New() + ctx, cancel := context.WithCancel(context.Background()) + go func() { time.Sleep(50 * time.Millisecond); cancel(); time.Sleep(50 * time.Millisecond); close(a.hang) }() + rec := httptest.NewRecorder() + s.Handler().ServeHTTP(rec, httptest.NewRequest("POST", "/v1/chat/completions", + strings.NewReader(`{"model":"group/g","messages":[{"role":"user","content":"hi"}]}`)).WithContext(ctx)) + r := s.trace.routes[len(s.trace.routes)-1] + if b.n != 0 || len(r.Tries) != 1 || r.Tries[0].Fail != failCanceled || r.Tries[0].Rest != nil || s.resting("a") { + t.Fatalf("b tried %d, tries %+v", b.n, r.Tries) + } +} diff --git a/internal/gateway/routing.go b/internal/gateway/routing.go index cad0aefa..f6ed257d 100644 --- a/internal/gateway/routing.go +++ b/internal/gateway/routing.go @@ -53,6 +53,11 @@ func served(rest string, tokens int) { routed.used[rest] = tokenUse{routed.used[rest].now(now) + float64(tokens), now} delete(routed.failures, rest) routed.Unlock() + // one that answered — tried all the same, or again — rests no longer + restingUntil.Lock() + delete(restingUntil.m, rest) + delete(restingUntil.note, rest) + restingUntil.Unlock() } // How long a candidate sits out, by why it failed. @@ -78,6 +83,8 @@ const ( failQuota = "quota" failRate = "rate" failOther = "other" + // failCanceled: the agent went away before the answer came + failCanceled = "canceled" ) // failure says why a reply failed. diff --git a/internal/gateway/trace.go b/internal/gateway/trace.go index 91d35073..2a96a062 100644 --- a/internal/gateway/trace.go +++ b/internal/gateway/trace.go @@ -87,7 +87,8 @@ type Try struct { Millis int64 `json:"ms,omitempty"` Fail string `json:"fail,omitempty"` // why it failed, as rest tells it Error string `json:"error,omitempty"` - Rest *Rest `json:"rest,omitempty"` // how long it now sits out; none when it was the last to try + Rest *Rest `json:"rest,omitempty"` // how long it now sits out; none when it was the last to try + Again int64 `json:"again,omitempty"` // ms waited before it was tried again, the last one left } type planned struct { diff --git a/internal/gui/assets/i18n.js b/internal/gui/assets/i18n.js index 627dd00b..306e7839 100644 --- a/internal/gui/assets/i18n.js +++ b/internal/gui/assets/i18n.js @@ -496,6 +496,9 @@ const I18N = { "out of credit": "余额不足", "quota used up": "额度用尽", "failed": "失败", + "canceled": "已取消", + "{agent} canceled the request while {who} was answering: nobody failed, so nobody rests and nobody else is asked.": "{who} 还在回答时，{agent} 取消了请求：不是谁出了错，所以谁也不用休息，也不再问别人。", + "{who} answered {status} · {fail}, and nobody else is left to ask — a failure that may pass, so it is tried again in {d}, before any of the reply reaches {agent}.": "{who} 回了 {status} · {fail}，而且没有别人可问了——这种失败可能一会儿就过去，所以 {d} 后再试它一次，回复的任何部分都还没交给 {agent}。", "fallback": "备用", "gateway": "网关", "your agent": "你的 Agent", diff --git a/internal/gui/assets/routing.js b/internal/gui/assets/routing.js index f4906dbd..53bb9af8 100644 --- a/internal/gui/assets/routing.js +++ b/internal/gui/assets/routing.js @@ -106,7 +106,7 @@ } const tokens = (n) => n >= 1e6 ? (n / 1e6).toFixed(1) + "M" : n >= 1e3 ? (n / 1e3).toFixed(1) + "k" : String(Math.round(n)); const pct = (n) => Math.round(n) + "%"; - const FAIL = { rate: "rate limited", credit: "out of credit", quota: "quota used up", other: "failed" }; + const FAIL = { rate: "rate limited", credit: "out of credit", quota: "quota used up", other: "failed", canceled: "canceled" }; const failWord = (why) => t(FAIL[why] || "failed"); const API = { anthropic: "Anthropic", chat: "OpenAI", responses: "OpenAI Responses", gemini: "Gemini" }; const MODES = { @@ -267,12 +267,17 @@ ? t("{who} answered in {ms}{tk}. {agent} got one clean reply and never saw the {n} that failed first.", { who: name, ms: took(tr.ms), tk, agent, n: i }) : t("{who} answered in {ms}{tk}.", { who: name, ms: took(tr.ms), tk }); } + if (tr.fail === "canceled") + return t("{agent} canceled the request while {who} was answering: nobody failed, so nobody rests and nobody else is asked.", { who: name, agent }); + if (tr.again) + return t("{who} answered {status} · {fail}, and nobody else is left to ask — a failure that may pass, so it is tried again in {d}, before any of the reply reaches {agent}.", + { who: name, status: tr.status, fail: failWord(tr.fail), d: took(tr.again), agent }); if (tr.rest) { const next = r.tries[i + 1], nw = next && r.order.find((x) => x.id === next.id); return t("{who} answered {status} · {fail}. {how}; the request goes on to {next} before any of the reply reaches {agent}.", { who: name, status: tr.status, fail: failWord(tr.fail), how: restHow(tr.rest, at(tr.start) + (tr.ms || 0)), next: nw ? who(nw) : t("the next"), agent }); } - const last = i === r.order.length - 1; + const last = i >= r.order.length - 1; return last ? t("{who} answered {status} and nobody is left to try, so {agent} gets the error.", { who: name, status: tr.status, agent }) : t("{who} answered {status} — an error another account wouldn't fix, so {agent} gets it.", { who: name, status: tr.status, agent }); @@ -563,9 +568,10 @@ for (const w of r.order) if (w.rest) rests.set(w.id, w.rest); for (const tr of r.tries) { if (tr.done && tr.status < 400) answered.add(tr.id); - else if (tr.done && !tr.rest) gave.set(tr.id, tr); // the error the agent got + else if (tr.done && !tr.rest && !tr.again) gave.set(tr.id, tr); // the error the agent got if (tr.rest) rests.set(tr.id, tr.rest); } + for (const id of answered) rests.delete(id); // it answered: whatever rest it began in is over const w = row.w, rest = rests.get(id), resting = rest && at(rest.until) > n; let s; if (w.unlisted) s = t("its plan doesn't list {model}", { model: w.model }); @@ -696,7 +702,7 @@ }); for (const tr of r.tries) { const a = by.get(tr.id); - if (!a || !tr.done) continue; + if (!a || !tr.done || tr.fail === "canceled") continue; // the agent's doing, not its a.tried++; const end = at(tr.start) + (tr.ms || 0); if (tr.status < 400) { a.ok++; a.last = Math.max(a.last, end); const m = r.order.find((x) => x.id === tr.id)?.model; if (m) a.models.add(m); } @@ -875,7 +881,7 @@ renderAll(); dot.classList.add("back", "err"); back = bird("res"); - if (!tr.rest) { // that was the answer: the agent gets the error + if (!tr.rest && !tr.again) { // that was the answer: the agent gets the error await fly(dot, back, home(row, A), 1350); flying.delete(dot); away(back);