gateway: in turn, a new conversation goes to the model or account the fewest live conversations are on, so two sessions of one routing group kept for the session no longer share a model (#946, Moody-Sin) — in turn counts requests, not conversations, so a second session started after the first had sent an even number of requests went to the model the first was kept on. leastHeld counts the conversations of the scope answered in the last 30 minutes per key/account and model and puts the least held first; ties, resting ones and a group with a group in it keep routing's order. TestRotateSpreadsSessions is new and fails (after 2 and 4 requests) without the change. go vet, go build (darwin, GOOS=linux nogui, GOOS=windows), the gateway package and go test -race -count=20 of TestRotateSpreadsSessions/TestAffinity pass; go test -tags nogui ./... under a temp HOME failed only TestWSLProbeFindsBunOmp, TestProvidersAnswerWhileListsComeIn, TestUserPath and TestClaudeSignedOut under the full run's load, each passing on its own.

2026-10-06 03:33:04 +08:00
1 parent 76dc5aed48
commit aab93f9fa8
@@ -42,13 +42,14 @@ served by a plugin are in [Provider and plugin ownership](provider-plugins.md).
 - A fallback happens only while none of the reply has been sent. An agent never gets half a reply from one upstream and the rest from another.
 - 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).
+- In turn (`rotate`), a conversation nobody has answered yet goes to the key, account or model the fewest other conversations of the provider's model or the group answered in the last `heldFor` (30 minutes) are on (`leastHeld`, #946). Ties, resting ones and a group with a group in it keep routing's order. With session affinity, two sessions of one group so stay on different models.
 - Different failures are classified separately: out of credit, out of quota, rate limit and other errors rest for different lengths. A fix must not merge them into one bucket (see #873).
 - Group members are `provider/model[:effort]`. A member with a fixed effort is asked for that effort, bounded by the group's levels (#671).

 ## Verification

 ```sh
-go test -tags nogui ./internal/gateway -run 'TestRouting|TestSmartRouting|TestFallback|TestNoFallbackForOtherErrors|TestLastFallbackErrorReachesTheAgent|TestRateLimitedSinksToTheBack|TestGroupRateLimitedSinks|TestAffinity|TestSeveralKeysOnTakeOverFromEachOther|TestSubscriptionAccountsTakeOver|TestGroup|TestTrace'
+go test -tags nogui ./internal/gateway -run 'TestRotateSpreadsSessions|TestRouting|TestSmartRouting|TestFallback|TestNoFallbackForOtherErrors|TestLastFallbackErrorReachesTheAgent|TestRateLimitedSinksToTheBack|TestGroupRateLimitedSinks|TestAffinity|TestSeveralKeysOnTakeOverFromEachOther|TestSubscriptionAccountsTakeOver|TestGroup|TestTrace'
 go test -tags nogui ./internal/provider -run 'TestGroup'
 ```

@@ -27,6 +27,7 @@ import (
 	"path/filepath"
 	"regexp"
 	"slices"
+	"strings"
 	"sync"
 	"time"

@@ -293,10 +294,71 @@ func affine(scope, mode string, rotate bool, in http.Header, from provider.Proto
 		if rotate {
 			cs, pl = after(cs, pl, at)
 		}
+	case "first":
+		if rotate {
+			cs, pl = leastHeld(scope, cs, pl)
+		}
 	}
 	return cs, pl, a, key
 }

+// heldFor is how long after its last answer a conversation still counts as
+// holding the one that answered it: a session whose user is reading or
+// typing is still going.
+const heldFor = 30 * time.Minute
+
+// leastHeld puts first, for a conversation nobody has answered yet, the
+// one in turn that the fewest other conversations in scope are held on
+// (#946): in turn counts requests, not conversations, so a second session
+// started after the first had sent an even number of them went to the
+// model the first was kept on. Of those held on as few, routing's order
+// stands.
+func leastHeld(scope string, cs []candidate, pl planned) ([]candidate, planned) {
+	if len(cs) < 2 || len(pl.order) < len(cs) {
+		return cs, pl
+	}
+	if strings.HasPrefix(scope, provider.GroupPrefix) {
+		if _, ms, ok := provider.FindGroup(scope); ok {
+			for _, m := range ms {
+				if len(m.Path) > 1 {
+					return cs, pl // a group in the group: its own routing orders its models (#576)
+				}
+			}
+		}
+	}
+	now := time.Now()
+	held := map[string]int{}
+	sticks.Lock()
+	for k, st := range sticks.m {
+		if strings.HasPrefix(k, scope+"|") && now.Sub(st.at) <= heldFor {
+			held[st.who+"/"+provider.WithMemberEffort(st.model, st.effort)]++
+		}
+	}
+	sticks.Unlock()
+	if len(held) == 0 {
+		return cs, pl
+	}
+	seat := func(c candidate) string { return c.who() + "/" + provider.WithMemberEffort(c.model, c.effort) }
+	best := -1
+	for i, c := range cs {
+		if pl.order[i].Rest != nil || pl.order[i].Aside {
+			continue
+		}
+		if best < 0 || held[seat(c)] < held[seat(cs[best])] {
+			best = i
+		}
+	}
+	if best <= 0 {
+		return cs, pl
+	}
+	cs = append(append([]candidate{cs[best]}, cs[:best]...), cs[best+1:]...)
+	pl.order = append(append([]Weighed{pl.order[best]}, pl.order[:best]...), pl.order[best+1:]...)
+	for j := range pl.order {
+		pl.order[j].Turn = j == 0
+	}
+	return cs, pl
+}
+
 // after puts first the one after cs[at], round from the end, that isn't
 // resting.
 func after(cs []candidate, pl planned, at int) ([]candidate, planned) {
@@ -430,3 +430,64 @@ func (k *geminiKeyed) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		`data: [DONE]`,
 	))
 }
+
+// A group in turn keeping each session where it began: a second session
+// goes to the model the first isn't on, however many requests the first
+// sent before it (#946); one resting is still passed over.
+func TestRotateSpreadsSessions(t *testing.T) {
+	pair := func() (*keyed, *keyed, *Server) {
+		fresh(t)
+		a, b := &keyed{}, &keyed{}
+		serveOn(t, "ga", "ka", []string{"m1"}, a)
+		serveOn(t, "gb", "kb", []string{"m2"}, b)
+		if err := provider.SaveGroup(provider.Group{Name: "Pair", Members: []string{"ga/m1", "gb/m2"}, Routing: provider.Rotate, Affinity: provider.AffinitySession}); err != nil {
+			t.Fatal(err)
+		}
+		routed.Lock()
+		delete(routed.turn, provider.GroupPrefix+"pair")
+		routed.Unlock()
+		return a, b, New()
+	}
+	// which of the two the request went to
+	went := func(a, b *keyed, post func()) string {
+		na, nb := len(a.tried), len(b.tried)
+		post()
+		switch {
+		case len(a.tried) > na && len(b.tried) == nb:
+			return "m1"
+		case len(b.tried) > nb && len(a.tried) == na:
+			return "m2"
+		}
+		t.Fatalf("tried %v and %v", a.tried, b.tried)
+		return ""
+	}
+	for n := 1; n <= 4; n++ {
+		a, b, s := pair()
+		msgs := `{"role":"user","content":"hi"}`
+		var first string
+		for i := range n {
+			if i > 0 {
+				msgs += `,{"role":"assistant","content":"yo"},{"role":"user","content":"more"}`
+			}
+			first = went(a, b, func() { postAs(t, s, "s1", `{"model":"group/pair","messages":[`+msgs+`]}`) })
+		}
+		second := went(a, b, func() { postAs(t, s, "s2", `{"model":"group/pair","messages":[{"role":"user","content":"other"}]}`) })
+		if first == second {
+			t.Errorf("after %d requests of s1 on %s, s2 went there too", n, first)
+		}
+		if r := s.trace.routes[len(s.trace.routes)-1]; r.Affinity == nil || r.Affinity.Why != "first" {
+			t.Errorf("s2's affinity: %+v", r.Affinity)
+		}
+	}
+
+	// the one fewest are on is resting: routing's order stands
+	a, b, s := pair()
+	first := went(a, b, func() { postAs(t, s, "s1", `{"model":"group/pair","messages":[{"role":"user","content":"hi"}]}`) })
+	other := map[string]string{"m1": "gb", "m2": "ga"}[first]
+	restingUntil.Lock()
+	restingUntil.m[other] = time.Now().Add(time.Minute)
+	restingUntil.Unlock()
+	if got := went(a, b, func() { postAs(t, s, "s2", `{"model":"group/pair","messages":[{"role":"user","content":"other"}]}`) }); got != first {
+		t.Fatalf("s2 went to %s, where %s rests", got, other)
+	}
+}