gateway: a request an upstream fails is failed over rather than handed back — a model the provider says it doesn't serve (Copilot's "not accessible via the /chat/completions endpoint", "model does not exist" and the like) goes on to the next one, as a 401, 404, 429 or 5xx already did; an error a stream begins with, before any of its content (Anthropic's overloaded error after message_start, a Chat error chunk, Responses' response.failed), is held and failed over as an error status is, while one after content has begun reaches the agent as it came; the last one left, or the only one, is tried again twice after a failure that may pass (408, 5xx, 529, a 429 that says when), a second and then two apart or as Retry-After says; and an agent that goes away mid-request is counted as canceled, resting nobody and asking nobody else. One that answers stops resting. The Routing view tells each of these

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
 }

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 {

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)
+	}
+}

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.

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 {

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",

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);