diff --git a/internal/gateway/affinity.go b/internal/gateway/affinity.go index c3ba4020..5061a2b2 100644 --- a/internal/gateway/affinity.go +++ b/internal/gateway/affinity.go @@ -15,6 +15,7 @@ package gateway // and not yet gone cold. import ( + "crypto/sha256" "encoding/json" "net/http" "regexp" @@ -211,10 +212,121 @@ func answered(key string, c candidate, turn, cacheRead int) { } } -// foreignReasoning is how OpenAI refuses reasoning another account (or -// organization) sealed: "The encrypted content for item rs_… could not be -// verified", invalid_encrypted_content. -var foreignReasoning = regexp.MustCompile(`(?i)invalid_encrypted_content|encrypted content.{0,80}could not be (verified|decrypted)`) +// foreignReasoning is how a vendor refuses reasoning another account (or +// organization, or vendor) sealed: OpenAI's "The encrypted content for +// item rs_… could not be verified", invalid_encrypted_content; xAI's +// (Grok's API and a SuperGrok account alike) "Could not decrypt the +// provided encrypted_content. Ensure the value is the unmodified +// encrypted_content from a previous response." +var foreignReasoning = regexp.MustCompile(`(?i)invalid_encrypted_content|encrypted[ _]content.{0,80}could not be (verified|decrypted)|could not (decrypt|verify).{0,40}encrypted[ _]content`) + +// refusedSeals remembers, per conversation and the key or account that +// refused it, the sealed reasoning another one wrote. Codex hands every +// earlier turn's reasoning back on each request, so once a conversation +// has moved, what was sealed before it moved is left out of its later +// requests there up front, rather than refused and asked again each turn. +// Reasoning the one it moved to sealed itself still goes along, for its +// cache and its train of thought. +var refusedSeals = struct { + sync.Mutex + m map[string]sealsRefused // scope|conversation|who → what it refused +}{m: map[string]sealsRefused{}} + +type sealsRefused struct { + at time.Time + seals map[[sha256.Size]byte]bool +} + +// sealedItem is what's read of an input item to tell sealed reasoning. +type sealedItem struct { + Type string `json:"type"` + Enc string `json:"encrypted_content"` +} + +// seals are the sealed reasoning (encrypted_content) in a Responses +// request's input. +func seals(body []byte) [][sha256.Size]byte { + var q struct { + Input []sealedItem `json:"input"` + } + if json.Unmarshal(body, &q) != nil { + return nil + } + var out [][sha256.Size]byte + for _, it := range q.Input { + if it.Type == "reasoning" && it.Enc != "" { + out = append(out, sha256.Sum256([]byte(it.Enc))) + } + } + return out +} + +// refused notes that who turned away the sealed reasoning in body, in the +// conversation key names. +func refused(key, who string, body []byte) { + ss := seals(body) + if len(ss) == 0 { + return + } + now := time.Now() + refusedSeals.Lock() + defer refusedSeals.Unlock() + k := key + "|" + who + r := refusedSeals.m[k] + if r.seals == nil { + r.seals = map[[sha256.Size]byte]bool{} + } + for _, s := range ss { + r.seals[s] = true + } + r.at = now + refusedSeals.m[k] = r + if len(refusedSeals.m) > 4096 { + for k, r := range refusedSeals.m { + if now.Sub(r.at) > stickKeep { + delete(refusedSeals.m, k) + } + } + } +} + +// withoutRefused takes out of a Responses request the sealed reasoning +// who already refused in this conversation; the rest stays as it is. +func withoutRefused(key, who string, body []byte) ([]byte, bool) { + k := key + "|" + who + refusedSeals.Lock() + r, ok := refusedSeals.m[k] + if ok && time.Since(r.at) > stickKeep { + delete(refusedSeals.m, k) + ok = false + } + refusedSeals.Unlock() + if !ok { + return nil, false + } + var q map[string]json.RawMessage + if json.Unmarshal(body, &q) != nil { + return nil, false + } + var items []json.RawMessage + if json.Unmarshal(q["input"], &items) != nil { + return nil, false + } + kept := items[:0:0] + for _, it := range items { + var t sealedItem + if json.Unmarshal(it, &t) == nil && t.Type == "reasoning" && t.Enc != "" && r.seals[sha256.Sum256([]byte(t.Enc))] { + continue + } + kept = append(kept, it) + } + if len(kept) == len(items) { + return nil, false + } + q["input"], _ = json.Marshal(kept) + b, err := json.Marshal(q) + return b, err == nil +} // withoutReasoning takes the sealed reasoning out of a Responses request's // input — what another account wrote and this one can't read. What was diff --git a/internal/gateway/foreign_seal_test.go b/internal/gateway/foreign_seal_test.go new file mode 100644 index 00000000..a2e9a5dc --- /dev/null +++ b/internal/gateway/foreign_seal_test.go @@ -0,0 +1,175 @@ +package gateway + +import ( + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + + "github.com/yetone/magpie/internal/provider" +) + +// xaiUndecryptable is what xAI answers — a Grok API key or a SuperGrok +// account alike — for reasoning another account or vendor sealed. +const xaiUndecryptable = "Could not decrypt the provided encrypted_content. Ensure the value is the unmodified encrypted_content from a previous response." + +// sealer is a Grok-like Responses upstream: it seals its reasoning as +// "sealed-by-" and can read only its own. +type sealer struct { + name string + down bool // answering 429, for the conversation to move on + stream bool // refusing in the stream (response.failed), not as a 400 + mu sync.Mutex + got [][]string // the sealed reasoning each request carried +} + +func (s *sealer) ServeHTTP(w http.ResponseWriter, r *http.Request) { + b, _ := io.ReadAll(r.Body) + var q struct { + Input []sealedItem `json:"input"` + } + json.Unmarshal(b, &q) + var got []string + foreign := false + for _, it := range q.Input { + if it.Type == "reasoning" { + got = append(got, it.Enc) + foreign = foreign || it.Enc != "sealed-by-"+s.name + } + } + s.mu.Lock() + s.got = append(s.got, got) + s.mu.Unlock() + switch { + case s.down: + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(429) + io.WriteString(w, `{"error":{"message":"slow down"}}`) + return + case foreign && !s.stream: + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(400) + io.WriteString(w, `{"code":"Client specified an invalid argument","error":"`+xaiUndecryptable+`"}`) + return + } + w.Header().Set("Content-Type", "text/event-stream") + if foreign { + io.WriteString(w, sse( + `data: {"type":"response.created","response":{"id":"r0","model":"grok-4"}}`, + `data: {"type":"response.failed","response":{"id":"r0","error":{"code":"invalid_request_error","message":"`+xaiUndecryptable+`"}}}`)) + return + } + io.WriteString(w, sse( + `data: {"type":"response.created","response":{"id":"r1","model":"grok-4"}}`, + `data: {"type":"response.output_item.done","output_index":0,"item":{"type":"reasoning","id":"rs_1","summary":[],"encrypted_content":"sealed-by-`+s.name+`"}}`, + `data: {"type":"response.output_text.delta","delta":"from `+s.name+`"}`, + `data: {"type":"response.completed","response":{"id":"r1","usage":{"input_tokens":7,"output_tokens":1}}}`)) +} + +func (s *sealer) sent() [][]string { + s.mu.Lock() + defer s.mu.Unlock() + out := s.got + s.got = nil + return out +} + +// Switching a Codex conversation between two Grok accounts of a routing +// group mid-way (waroy): the second can't decrypt the first's sealed +// reasoning. It's asked again without it, and later turns leave out what +// it refused while its own sealed reasoning still goes along. +func TestForeignSealAcrossGroupMembers(t *testing.T) { + for _, inStream := range []bool{false, true} { + fresh(t) + refusedSeals.Lock() + refusedSeals.m = map[string]sealsRefused{} + refusedSeals.Unlock() + a, b := &sealer{name: "a"}, &sealer{name: "b", stream: inStream} + for _, u := range []struct { + id string + h http.Handler + }{{"grok1", a}, {"grok2", b}} { + up := httptest.NewServer(u.h) + t.Cleanup(up.Close) + if err := provider.Save(provider.Provider{ID: u.id, Name: "Grok (SuperGrok)", Key: "k-" + u.id, Models: []string{"grok-4"}, Responses: up.URL + "/v1"}); err != nil { + t.Fatal(err) + } + } + if err := provider.SaveGroup(provider.Group{Name: "Grok", Members: []string{"grok1/grok-4", "grok2/grok-4"}, Routing: provider.Ordered}); err != nil { + t.Fatal(err) + } + s := New() + turn := func(items ...string) (int, string) { + t.Helper() + in := `{"type":"message","role":"user","content":[{"type":"input_text","text":"count the lines"}]}` + for _, it := range items { + in += "," + it + } + rec := httptest.NewRecorder() + req := httptest.NewRequest("POST", "/v1/responses", strings.NewReader(`{"model":"group/grok","stream":true,"input":[`+in+`]}`)) + req.Header.Set("session_id", "thread-1") + s.Handler().ServeHTTP(rec, req) + return rec.Code, rec.Body.String() + } + reasoning := func(by string) string { + return `{"type":"reasoning","id":"rs_` + by + `","summary":[],"encrypted_content":"sealed-by-` + by + `"}` + } + call := func(id string) string { + return `{"type":"function_call","call_id":"` + id + `","name":"shell","arguments":"{}"},{"type":"function_call_output","call_id":"` + id + `","output":"ok"}` + } + + // a answers, and reads back what it sealed + if code, body := turn(); code != 200 || !strings.Contains(body, "from a") { + t.Fatalf("stream=%v first: %d %s", inStream, code, body) + } + if code, body := turn(reasoning("a"), call("c1")); code != 200 || !strings.Contains(body, "from a") { + t.Fatalf("stream=%v second: %d %s", inStream, code, body) + } + if got := a.sent(); len(got) != 2 || strings.Join(got[1], ",") != "sealed-by-a" { + t.Fatalf("stream=%v a was sent %q", inStream, got) + } + + // a goes down: b takes the conversation over, without a's seal + a.down = true + code, body := turn(reasoning("a"), call("c1"), reasoning("a"), call("c2")) + if code != 200 || !strings.Contains(body, "from b") { + t.Fatalf("stream=%v after the switch: %d %s", inStream, code, body) + } + if got := b.sent(); len(got) != 2 || len(got[0]) != 2 || len(got[1]) != 0 { + t.Fatalf("stream=%v b was sent %q", inStream, got) + } + + // the next turn doesn't hit it again: a's seal is left out up front, + // b's own goes along + code, body = turn(reasoning("a"), call("c1"), reasoning("a"), call("c2"), reasoning("b"), call("c3")) + if code != 200 || !strings.Contains(body, "from b") { + t.Fatalf("stream=%v next turn: %d %s", inStream, code, body) + } + if got := b.sent(); len(got) != 1 || strings.Join(got[0], ",") != "sealed-by-b" { + t.Fatalf("stream=%v b was then sent %q", inStream, got) + } + a.sent() + } +} + +// Each vendor's words for reasoning it can't read are known; a request +// at fault some other way is not taken for one. +func TestForeignReasoningWords(t *testing.T) { + for _, m := range []string{ + xaiUndecryptable, + `{"error":{"message":"The encrypted content for item rs_1 could not be verified.","code":"invalid_encrypted_content"}}`, + "Grok (SuperGrok): Could not decrypt the provided encrypted_content.", + } { + if !foreignReasoning.MatchString(m) { + t.Errorf("not matched: %s", m) + } + } + for _, m := range []string{"Invalid value for reasoning.effort", "model not found", "could not parse the JSON body"} { + if foreignReasoning.MatchString(m) { + t.Errorf("matched: %s", m) + } + } +} diff --git a/internal/gateway/gateway.go b/internal/gateway/gateway.go index b7e602a5..4f98423c 100644 --- a/internal/gateway/gateway.go +++ b/internal/gateway/gateway.go @@ -805,6 +805,13 @@ func (s *Server) serve(w http.ResponseWriter, r *http.Request, from provider.Pro where = c.p.Where() began := time.Now() attemptBody := body + if from == provider.Responses { + // reasoning another sealed, which this one refused earlier in + // the conversation, isn't sent to be refused again; its own is + if b, ok := withoutRefused(stuck, c.who(), attemptBody); ok { + attemptBody = b + } + } if isGroup { if in := membersImageInput([]provider.Member{{Provider: c.p, Model: c.model}}, nil); in != nil && !*in { attemptBody, _ = textOnlyBody(from, body) // omit images in prior turns and tool results @@ -849,10 +856,14 @@ func (s *Server) serve(w http.ResponseWriter, r *http.Request, from provider.Pro s.trace.update(tr, func(t *Route) { t.Tries[len(t.Tries)-1] = try }) break } - if !resealed && from == provider.Responses && !hw.passing && hw.code() == 400 && foreignReasoning.Match(hw.errBody()) { - // the conversation moved here from another account, whose - // sealed reasoning this one can't read: asked again without it + if !resealed && from == provider.Responses && !hw.passing && hw.code() >= 400 && foreignReasoning.Match(hw.errBody()) { + // the conversation moved here from another account or vendor, + // whose sealed reasoning this one can't read: asked again + // without it, and what it refused isn't sent here again. xAI + // says so as a 400, or as the stream's error, which may be + // read as another status if b, ok := withoutReasoning(body); ok { + refused(stuck, c.who(), attemptBody) resealed, body = true, b try.Fail = failForeign s.trace.update(tr, func(t *Route) { t.Tries[len(t.Tries)-1] = try })