feat(gateway): optional quota failover for existing conversations

PRODUCT.md

@@ -107,6 +107,8 @@ Gateway admission allows model changes with reasoning, compaction and reconstruc
 
 Resource groups offer optional session stickiness, disabled by default. When enabled, requests that carry no gateway session derive a conversation key from client conversation headers, Codex turn metadata, Claude Code session metadata or, as a fallback, a hash of the system prompt and first user message, following CLIProxyAPI's session-affinity order. Derived keys create one-hour preferences in their own namespace, so they never read or rewrite an explicit binding: the preferred resource ranks first only after every policy, quota, allowance, cooldown and capacity check, and unavailable resources fall back to normal scheduling. Preferences may hold at most a quarter of the shared affinity table, are skipped rather than rejected when it is full, and are evicted first when an explicit conversation needs a slot. Subagents identified by Codex parent threads, Claude parent sessions or `X-Parent-Session-Id` prefer their parent's resource through read-only lookups of the parent's explicit binding or preference. Claude Code subagents share their main session key; a conversation keeps whichever raw or normalized key already holds its binding whether the setting is on or off; the setting only decides whether an unbound conversation creates a normalized binding. Explicit sessions keep the existing strict binding rules.
 
+Resource groups also offer optional quota failover, disabled by default. When enabled, an existing Responses conversation whose bound account has exhausted its quota moves to another admitted account of the same provider instead of failing. The binding records 16-byte fingerprints (at most 4,096) of every encrypted reasoning or compaction item present at the move, and those items are withheld from the new account on that and every later turn, so only readable transcript and the new account's own state reach it. Requests that reference provider-held state (`previous_response_id`, `conversation`, `container` or `item_reference`) never move, and disablement, cooldown, saturation or model mismatch still never move a conversation.
+
 API channels can be renamed and given a replacement key in place at `/channels`; leaving the key blank keeps the stored secret. Base URL, platform and protocol remain fixed so a stored key never reaches another upstream; those changes still require a separate channel. Edits retain the channel ID, group assignments, limits and request history; a key change resets verification and invalidates the model catalog in the same transaction. Resource groups can be deleted when no active personal key or resource allowance references them. Deletion retires the group: routing, grants, resources, model policy and bindings are removed in one audited transaction, while request history and usage aggregates keep their workspace scope. Retired groups keep their names; uniqueness applies only to active groups, so the name can be reused.
 
 Network proxies are shown as compact cards in a responsive grid whose column count follows the page width. Each card shows its status and latency, an IP purity panel, and the exit IP, location, network, verdict and bound accounts exactly once, merging the proxy check with the IP report. Opening the page only reads reports the server already holds; an administrator must explicitly request a lookup, which sends the exit IP observed by the last successful check, never the proxy endpoint, to the public ip.net.coffee lookup used by its IP scoring page. The panel shows its trust score with the same five purity tiers, network type, risk flags, ASN, ISP, location and verdict. Only public exit IPs are sent; results are cached per IP for six hours, concurrent lookups for one IP share a single request, attempts are paced to one per 30 seconds per IP from when they start, rate-limited responses back off for five minutes, and a new IP is refused while all 256 cache slots hold in-flight lookups. The report is informational and does not affect routing or proxy checks.

internal/backup/backup_test.go

@@ -212,6 +212,8 @@ ALTER TABLE accounts DROP COLUMN reset_check_at;
 ALTER TABLE accounts DROP COLUMN usage_revision;
 ALTER TABLE account_affinity DROP COLUMN rebound;
 ALTER TABLE account_affinity DROP COLUMN model;
+ALTER TABLE account_groups DROP COLUMN quota_failover;
+ALTER TABLE account_affinity DROP COLUMN foreign_state;
 DROP INDEX account_groups_active_name;
 ALTER TABLE allocation_entries DROP COLUMN image_input_tokens;
 ALTER TABLE account_affinity DROP COLUMN derived;

internal/gateway/failover.go

@@ -0,0 +1,150 @@
+package gateway
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/json"
+)
+
+// Quota failover lets a resource group move an existing Responses conversation off a
+// quota-exhausted account. Encrypted reasoning and compaction state belongs to the account that
+// produced it, so the binding remembers fingerprints of every encrypted item present when it moved
+// and those items are withheld from the new account on this and later turns.
+
+const (
+	stateDigestSize = 16
+	// Bounds the fingerprint column (migration 035 allows 65,536 bytes).
+	foreignStateLimit = 4096
+)
+
+type stateDigest [stateDigestSize]byte
+
+// quotaFailover is owned by one Open call; selection fills Strip for the account it leased.
+type quotaFailover struct {
+	// Movable is false when the request references provider-held state SubLane cannot replay.
+	Movable bool
+	// Present fingerprints encrypted items carried by the current request.
+	Present []stateDigest
+	// Strip lists fingerprints the leased account must not receive.
+	Strip map[stateDigest]bool
+}
+
+type failoverKey struct{}
+
+func withQuotaFailover(ctx context.Context, carrier *quotaFailover) context.Context {
+	return context.WithValue(ctx, failoverKey{}, carrier)
+}
+
+func quotaFailoverFrom(ctx context.Context) *quotaFailover {
+	carrier, _ := ctx.Value(failoverKey{}).(*quotaFailover)
+	return carrier
+}
+
+func fingerprint(encrypted json.RawMessage) stateDigest {
+	// Hash the decoded token so equivalent JSON escaping yields one fingerprint.
+	value := []byte(encrypted)
+	var text string
+	if json.Unmarshal(encrypted, &text) == nil {
+		value = []byte(text)
+	}
+	sum := sha256.Sum256(value)
+	var digest stateDigest
+	copy(digest[:], sum[:])
+	return digest
+}
+
+type stateItem struct {
+	Type             string          `json:"type"`
+	EncryptedContent json.RawMessage `json:"encrypted_content"`
+}
+
+func (item stateItem) encrypted() bool {
+	return len(item.EncryptedContent) > 0 && string(item.EncryptedContent) != "null"
+}
+
+func newQuotaFailover(input map[string]json.RawMessage, kind Kind) *quotaFailover {
+	carrier := &quotaFailover{}
+	if kind != Responses && kind != Compact {
+		return carrier
+	}
+	for _, key := range []string{"previous_response_id", "conversation", "container"} {
+		if value := input[key]; len(value) > 0 && string(value) != "null" && string(value) != `""` {
+			return carrier
+		}
+	}
+	var items []stateItem
+	if json.Unmarshal(input["input"], &items) != nil {
+		// A string input carries no provider state.
+		var text string
+		carrier.Movable = json.Unmarshal(input["input"], &text) == nil || len(input["input"]) == 0
+		return carrier
+	}
+	seen := map[stateDigest]bool{}
+	for _, item := range items {
+		// Stored item references live on the original account and cannot be replayed elsewhere.
+		if item.Type == "item_reference" {
+			return carrier
+		}
+		if item.encrypted() {
+			if digest := fingerprint(item.EncryptedContent); !seen[digest] {
+				seen[digest] = true
+				carrier.Present = append(carrier.Present, digest)
+			}
+		}
+	}
+	carrier.Movable = true
+	return carrier
+}
+
+func decodeForeignState(raw []byte) map[stateDigest]bool {
+	set := make(map[stateDigest]bool, len(raw)/stateDigestSize)
+	for len(raw) >= stateDigestSize {
+		var digest stateDigest
+		copy(digest[:], raw[:stateDigestSize])
+		set[digest] = true
+		raw = raw[stateDigestSize:]
+	}
+	return set
+}
+
+// mergeForeignState returns the stored fingerprints plus the request's, or false past the bound.
+func mergeForeignState(stored []byte, present []stateDigest) (map[stateDigest]bool, []byte, bool) {
+	set := decodeForeignState(stored)
+	encoded := append([]byte(nil), stored[:len(stored)-len(stored)%stateDigestSize]...)
+	for _, digest := range present {
+		if !set[digest] {
+			set[digest] = true
+			encoded = append(encoded, digest[:]...)
+		}
+	}
+	return set, encoded, len(set) <= foreignStateLimit
+}
+
+// stripForeignState removes encrypted items another account produced. Visible messages and tool
+// calls stay, so the new account continues from the readable transcript.
+func stripForeignState(input map[string]json.RawMessage, strip map[stateDigest]bool) error {
+	if len(strip) == 0 {
+		return nil
+	}
+	var items []json.RawMessage
+	if json.Unmarshal(input["input"], &items) != nil {
+		return nil
+	}
+	kept := items[:0]
+	for _, raw := range items {
+		var item stateItem
+		if json.Unmarshal(raw, &item) == nil && item.encrypted() && strip[fingerprint(item.EncryptedContent)] {
+			continue
+		}
+		kept = append(kept, raw)
+	}
+	if len(kept) == len(items) {
+		return nil
+	}
+	encoded, err := json.Marshal(kept)
+	if err != nil {
+		return err
+	}
+	input["input"] = encoded
+	return nil
+}

internal/gateway/failover_test.go

@@ -0,0 +1,136 @@
+package gateway
+
+import (
+	"context"
+	"errors"
+	"io"
+	"net/http"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/murongg/SubLane/internal/accounts"
+	"github.com/murongg/SubLane/internal/upstream"
+)
+
+type capturedRequest struct{ account, body string }
+
+// failoverGateway returns the account a new conversation was bound to and the other pool member.
+func failoverGateway(t *testing.T, enabled bool) (*Service, string, string, func() []capturedRequest) {
+	t.Helper()
+	var mu sync.Mutex
+	var seen []capturedRequest
+	s, ids := codexGateway(t, transportFunc(func(r *http.Request) (*http.Response, error) {
+		if r.Body != nil && strings.HasSuffix(r.URL.Path, "/responses") {
+			raw, _ := io.ReadAll(r.Body)
+			mu.Lock()
+			seen = append(seen, capturedRequest{r.Header.Get("Chatgpt-Account-Id"), string(raw)})
+			mu.Unlock()
+		}
+		return syntheticStream(), nil
+	}))
+	ctx := context.Background()
+	second, err := s.accounts.Authorize(ctx, "Synthetic second", accounts.Credential{AccountID: "failover-second", AccessToken: "synthetic-access", RefreshToken: "synthetic-refresh", ExpiresAt: time.Now().Add(time.Hour).Unix()}, "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := s.accounts.SaveCatalog(ctx, second.ID, 0, []string{"synthetic-model"}, s.now().Unix(), upstream.CatalogSource("codex")); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := s.db.Exec("INSERT INTO group_accounts(group_id,account_id) VALUES(1,?)", second.ID); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := s.db.Exec("UPDATE account_groups SET quota_failover=? WHERE id=1", enabled); err != nil {
+		t.Fatal(err)
+	}
+	if err := openTurn(t, s, failoverUser); err != nil {
+		t.Fatal("initial binding", err)
+	}
+	first, other := boundAccount(t, s), second.ID
+	if first == second.ID {
+		other = ids["codex"]
+	}
+	return s, first, other, func() []capturedRequest {
+		mu.Lock()
+		defer mu.Unlock()
+		return append([]capturedRequest(nil), seen...)
+	}
+}
+
+func openTurn(t *testing.T, s *Service, input string) error {
+	t.Helper()
+	response, err := s.Open(context.Background(), 1, 1, []byte(`{"model":"synthetic-model","input":[`+input+`]}`), http.Header{"Session_id": {"synthetic-failover"}}, Responses)
+	if err == nil {
+		io.Copy(io.Discard, response.Body)
+		response.Body.Close()
+	}
+	return err
+}
+
+func boundAccount(t *testing.T, s *Service) string {
+	t.Helper()
+	var account string
+	if err := s.db.QueryRow("SELECT account_id FROM account_affinity").Scan(&account); err != nil {
+		t.Fatal(err)
+	}
+	return account
+}
+
+const (
+	failoverUser     = `{"role":"user","content":"Synthetic question"}`
+	failoverOld      = `{"type":"reasoning","summary":[],"encrypted_content":"gAAAA-synthetic-old"}`
+	failoverReply    = `{"role":"assistant","content":[{"type":"output_text","text":"Synthetic answer"}]}`
+	failoverNew      = `{"type":"reasoning","summary":[],"encrypted_content":"gAAAA-synthetic-new"}`
+	failoverFollowUp = `{"role":"user","content":"Synthetic follow-up"}`
+)
+
+func TestQuotaFailoverMovesConversationAndWithholdsForeignState(t *testing.T) {
+	s, first, second, requests := failoverGateway(t, true)
+	saveExhaustedQuota(t, s, first)
+	if err := openTurn(t, s, strings.Join([]string{failoverUser, failoverOld, failoverReply, failoverFollowUp}, ",")); err != nil {
+		t.Fatal("exhausted conversation did not fail over", err)
+	}
+	if boundAccount(t, s) != second {
+		t.Fatal("binding did not move")
+	}
+	// A restart must keep withholding the previous account's state while new state passes through.
+	restarted := New(context.Background(), s.db, s.accounts, s.provider)
+	defer restarted.Close()
+	if err := openTurn(t, restarted, strings.Join([]string{failoverUser, failoverOld, failoverReply, failoverFollowUp, failoverNew, failoverReply, failoverFollowUp}, ",")); err != nil {
+		t.Fatal(err)
+	}
+	seen := requests()
+	secondSubject := seen[1].account
+	if secondSubject == seen[0].account {
+		t.Fatal("moved turn reused the exhausted account")
+	}
+	if len(seen) != 3 {
+		t.Fatal("unexpected upstream requests", len(seen))
+	}
+	for i, request := range seen[1:] {
+		if request.account != secondSubject || strings.Contains(request.body, "synthetic-old") || !strings.Contains(request.body, "Synthetic follow-up") {
+			t.Fatal("moved turn leaked foreign state or lost transcript", i, request.account, request.body)
+		}
+	}
+	// The SDK may drop synthetic token text, so count reasoning items: only the current account's remains.
+	if strings.Count(seen[1].body, `"type":"reasoning"`) != 0 || strings.Count(seen[2].body, `"type":"reasoning"`) != 1 {
+		t.Fatal("state from the current account was withheld", seen[2].body)
+	}
+}
+
+func TestQuotaFailoverDisabledKeepsStrictBinding(t *testing.T) {
+	s, first, _, _ := failoverGateway(t, false)
+	saveExhaustedQuota(t, s, first)
+	if err := openTurn(t, s, strings.Join([]string{failoverUser, failoverOld, failoverReply, failoverFollowUp}, ",")); !errors.Is(err, ErrQuotaExhausted) || boundAccount(t, s) != first {
+		t.Fatal("conversation moved without opt-in", err)
+	}
+}
+
+func TestQuotaFailoverRefusesServerHeldState(t *testing.T) {
+	s, first, _, _ := failoverGateway(t, true)
+	saveExhaustedQuota(t, s, first)
+	if err := openTurn(t, s, `{"type":"item_reference","id":"synthetic-item"},`+failoverFollowUp); !errors.Is(err, ErrQuotaExhausted) || boundAccount(t, s) != first {
+		t.Fatal("stored provider state moved accounts", err)
+	}
+}

internal/gateway/gateway.go

@@ -222,6 +222,8 @@ func (s *Service) Open(ctx context.Context, userID, groupID int64, raw []byte, h
 	}
 	// Hints are inert unless the group enables stickiness; selection reads that policy in its transaction.
 	ctx = withStickyHints(ctx, requestStickyHints(headers, input, kind, session, fromMetadata))
+	failover := newQuotaFailover(input, kind)
+	ctx = withQuotaFailover(ctx, failover)
 	id, digest, err := s.leaseAccount(ctx, entry, session, provider, model, kind, affinity, reviewParentThread(input, headers, model, kind))
 	if id != "" {
 		if account, lookupErr := s.accounts.Get(ctx, id); lookupErr == nil {
@@ -244,6 +246,10 @@ func (s *Service) Open(ctx context.Context, userID, groupID int64, raw []byte, h
 	} else {
 		input["prompt_cache_key"], _ = json.Marshal(hex.EncodeToString(digest[:]))
 	}
+	// Selection names encrypted items from accounts this conversation left; the leased account cannot read them.
+	if stripForeignState(input, failover.Strip) != nil {
+		return nil, upstream.ErrInput
+	}
 	raw, err = json.Marshal(input)
 	if err != nil {
 		return nil, upstream.ErrInput

internal/gateway/selection.go

@@ -161,6 +161,13 @@ func (s *Service) selectAllocationAccount(ctx context.Context, userID, groupID i
 	bindingModel := upstream.CatalogModelID(model)
 	var replacement *db.ListSessionAffinitiesRow
 	stickyAccount := ""
+	carrier := quotaFailoverFrom(ctx)
+	if carrier != nil {
+		carrier.Strip = nil
+	}
+	// failoverProvider is set when an exhausted binding moves; foreignState is the binding's new fingerprint set.
+	failoverProvider := ""
+	var foreignState []byte
 	if session != "" && derived {
 		// A derived binding is only a preference: unavailable resources fall back to normal scheduling.
 		existing, err := sessionBinding(ctx, q, db.ListSessionAffinitiesParams{GroupID: groupID, UserID: userID, SessionHash: digest[:], Now: now}, scope)
@@ -237,8 +244,19 @@ func (s *Service) selectAllocationAccount(ctx context.Context, userID, groupID i
 					return id, digest, err
 				}
 				if err := quotaAdmission(*bound, model); err != nil {
-					return id, digest, err
+					var quota *QuotaError
+					// Opt-in only: the default keeps the strict rule that conversations never move.
+					if !errors.As(err, &quota) || !policy.Routing.QuotaFailover || carrier == nil || !carrier.Movable || existing.Provider != scope {
+						return id, digest, err
+					}
+					set, encoded, ok := mergeForeignState(existing.ForeignState, carrier.Present)
+					if !ok {
+						return id, digest, err
+					}
+					carrier.Strip, foreignState, failoverProvider, replacement = set, encoded, bound.Provider, existing
 				}
+			}
+			if failoverProvider == "" && !reselect {
 				if scheme != 0 {
 					if _, err := allocations.Check(ctx, q, scheme, userID, groupID, id, model, now, s.location()); err != nil {
 						return id, digest, err
@@ -258,6 +276,10 @@ func (s *Service) selectAllocationAccount(ctx context.Context, userID, groupID i
 				if err := tx.Commit(); err != nil {
 					return "", digest, err
 				}
+				if carrier != nil && len(existing.ForeignState) > 0 {
+					// State written by accounts this conversation left stays off the current one.
+					carrier.Strip = decodeForeignState(existing.ForeignState)
+				}
 				route.Candidates = 1
 				return id, digest, nil
 			}
@@ -293,7 +315,7 @@ func (s *Service) selectAllocationAccount(ctx context.Context, userID, groupID i
 	var coolingModel string
 	var allocationErr error
 	for _, account := range available {
-		if !s.accounts.ProviderEnabled(account.Provider) || !policy.AllowsProvider(account.Provider) || !allowed[account.ID] || !account.Enabled || account.Status == "reauth_required" || (provider != "" && account.Provider != provider) || (kind.codexOnly() && account.Provider != "codex") {
+		if !s.accounts.ProviderEnabled(account.Provider) || !policy.AllowsProvider(account.Provider) || !allowed[account.ID] || !account.Enabled || account.Status == "reauth_required" || (provider != "" && account.Provider != provider) || (kind.codexOnly() && account.Provider != "codex") || (failoverProvider != "" && account.Provider != failoverProvider) {
 			if !s.accounts.ProviderEnabled(account.Provider) || !account.Enabled || account.Status == "reauth_required" {
 				route.Exclusions.Unavailable = min(maxRouteCount, route.Exclusions.Unavailable+1)
 			} else {
@@ -453,7 +475,12 @@ func (s *Service) selectAllocationAccount(ctx context.Context, userID, groupID i
 	}
 	s.next[cursor] = (s.next[cursor] + 1) % period
 	if session != "" {
-		if replacement != nil && replacement.Provider == scope {
+		if failoverProvider != "" {
+			if bindingModel == "" {
+				bindingModel = replacement.Model
+			}
+			err = q.MoveAccountAffinity(ctx, db.MoveAccountAffinityParams{UserID: userID, GroupID: groupID, SessionHash: digest[:], Provider: scope, AccountID: id, Model: bindingModel, ExpiresAt: expires, ForeignState: foreignState})
+		} else if replacement != nil && replacement.Provider == scope {
 			err = q.ReplaceAccountAffinity(ctx, db.ReplaceAccountAffinityParams{UserID: userID, GroupID: groupID, SessionHash: digest[:], Provider: scope, AccountID: id, Model: bindingModel, ExpiresAt: expires})
 		} else {
 			// Creating an auto binding from a legacy scope can also move the account.