diff --git a/internal/gateway/affinity.go b/internal/gateway/affinity.go index 92a6b4b3..222337ac 100644 --- a/internal/gateway/affinity.go +++ b/internal/gateway/affinity.go @@ -19,6 +19,7 @@ import ( "encoding/json" "net/http" "regexp" + "slices" "sync" "time" @@ -245,9 +246,17 @@ type sealedItem struct { 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 { +// sealedKinds are the input items a vendor seals for itself: its +// reasoning, and the compaction a conversation's earlier turns were +// folded into (OpenAI's, when Codex compacts on its own models; magpie's +// own is text by the time a request is sent). +var sealedKinds = map[string]bool{"reasoning": true, "compaction": true, "compaction_summary": true} + +var compactionKinds = []string{"compaction", "compaction_summary"} + +// seals are the sealed reasoning and compactions (encrypted_content) in a +// Responses request's input, of these types. +func seals(body []byte, kinds ...string) [][sha256.Size]byte { var q struct { Input []sealedItem `json:"input"` } @@ -256,17 +265,17 @@ func seals(body []byte) [][sha256.Size]byte { } var out [][sha256.Size]byte for _, it := range q.Input { - if it.Type == "reasoning" && it.Enc != "" { + if slices.Contains(kinds, it.Type) && 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) +// refused notes that who turned away the sealed items of these types in +// body, in the conversation key names. +func refused(key, who string, body []byte, kinds ...string) { + ss := seals(body, kinds...) if len(ss) == 0 { return } @@ -293,7 +302,8 @@ func refused(key, who string, body []byte) { } // withoutRefused takes out of a Responses request the sealed reasoning -// who already refused in this conversation; the rest stays as it is. +// and compactions 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() @@ -317,7 +327,7 @@ func withoutRefused(key, who string, body []byte) ([]byte, bool) { 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))] { + if json.Unmarshal(it, &t) == nil && sealedKinds[t.Type] && t.Enc != "" && r.seals[sha256.Sum256([]byte(t.Enc))] { continue } kept = append(kept, it) @@ -334,6 +344,32 @@ func withoutRefused(key, who string, body []byte) ([]byte, bool) { // input — what another account wrote and this one can't read. What was // said and done stays; only the model's private notes to itself go. func withoutReasoning(body []byte) ([]byte, bool) { + return withoutKinds(body, "reasoning") +} + +// withoutCompaction takes out of a Responses request's input the sealed +// compaction of its earlier turns — OpenAI's, which no other vendor or +// account can read (waroy: Grok's "Could not decrypt the provided +// encrypted_content" with the reasoning already gone). What the turns +// since said and did stays; the summary of those before goes. +func withoutCompaction(body []byte) ([]byte, bool) { + var q struct { + Input []sealedItem `json:"input"` + } + if json.Unmarshal(body, &q) != nil { + return nil, false + } + for _, it := range q.Input { + if slices.Contains(compactionKinds, it.Type) && it.Enc != "" { + return withoutKinds(body, compactionKinds...) + } + } + return nil, false +} + +// withoutKinds is a Responses request without its input items of these +// types. +func withoutKinds(body []byte, kinds ...string) ([]byte, bool) { var q map[string]json.RawMessage if json.Unmarshal(body, &q) != nil { return nil, false @@ -347,7 +383,7 @@ func withoutReasoning(body []byte) ([]byte, bool) { var t struct { Type string `json:"type"` } - if json.Unmarshal(it, &t) == nil && t.Type == "reasoning" { + if json.Unmarshal(it, &t) == nil && slices.Contains(kinds, t.Type) { continue } kept = append(kept, it) diff --git a/internal/gateway/foreign_compaction_test.go b/internal/gateway/foreign_compaction_test.go new file mode 100644 index 00000000..e4b78122 --- /dev/null +++ b/internal/gateway/foreign_compaction_test.go @@ -0,0 +1,119 @@ +package gateway + +import ( + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + + "github.com/yetone/magpie/internal/provider" +) + +// grokUpstream is a SuperGrok-like Responses upstream that can decrypt +// only what it sealed itself, in reasoning and in a compaction alike, and +// says so in the stream as xAI does. +type grokUpstream struct { + mu sync.Mutex + got [][]string // the sealed content each request carried, by type +} + +func (g *grokUpstream) 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.Enc != "" { + got = append(got, it.Type+":"+it.Enc) + foreign = foreign || it.Enc != "sealed-by-grok" + } + } + g.mu.Lock() + g.got = append(g.got, got) + g.mu.Unlock() + 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_g","summary":[],"encrypted_content":"sealed-by-grok"}}`, + `data: {"type":"response.output_text.delta","delta":"from grok"}`, + `data: {"type":"response.completed","response":{"id":"r1","usage":{"input_tokens":7,"output_tokens":1}}}`)) +} + +func (g *grokUpstream) sent() [][]string { + g.mu.Lock() + defer g.mu.Unlock() + out := g.got + g.got = nil + return out +} + +// A Codex conversation OpenAI compacted, switched to a Grok model (waroy: +// Grok (SuperGrok): Could not decrypt the provided encrypted_content): +// the compaction OpenAI sealed stays in the input as well as its +// reasoning, and Grok can read neither. With the reasoning taken out it +// still refused the compaction, and that refusal went to Codex. Now the +// compaction goes too once it is still refused, and later turns leave out +// both up front while Grok's own reasoning goes along. +func TestForeignCompactionToGrok(t *testing.T) { + fresh(t) + refusedSeals.Lock() + refusedSeals.m = map[string]sealsRefused{} + refusedSeals.Unlock() + g := &grokUpstream{} + up := httptest.NewServer(g) + t.Cleanup(up.Close) + if err := provider.Save(provider.Provider{ID: "grok1", Name: "Grok (SuperGrok)", Key: "k", Models: []string{"grok-4"}, Responses: up.URL + "/v1"}); err != nil { + t.Fatal(err) + } + s := New() + user := func(text string) string { + return `{"type":"message","role":"user","content":[{"type":"input_text","text":"` + text + `"}]}` + } + turn := func(items ...string) (int, string) { + t.Helper() + rec := httptest.NewRecorder() + req := httptest.NewRequest("POST", CodexPath+"/responses", strings.NewReader(`{"model":"grok1/grok-4","stream":true,"input":[`+strings.Join(items, ",")+`]}`)) + req.Header.Set("session_id", "thread-compacted") + s.Handler().ServeHTTP(rec, req) + return rec.Code, rec.Body.String() + } + compaction := `{"type":"compaction","id":"cmp_1","encrypted_content":"sealed-by-openai-compaction"}` + 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"}` + } + + history := []string{user("start"), compaction, user("go on"), reasoning("openai"), call("c1")} + code, body := turn(history...) + if code != 200 || !strings.Contains(body, "from grok") || strings.Contains(body, "decrypt") { + t.Fatalf("after the switch: %d %s", code, body) + } + got := g.sent() + if len(got) != 3 || len(got[0]) != 2 || strings.Join(got[1], ",") != "compaction:sealed-by-openai-compaction" || len(got[2]) != 0 { + t.Fatalf("grok was sent %q", got) + } + + // the next turn: neither of OpenAI's is sent to be refused again, Grok's + // own reasoning goes along + code, body = turn(append(history, reasoning("grok"), call("c2"))...) + if code != 200 || !strings.Contains(body, "from grok") { + t.Fatalf("next turn: %d %s", code, body) + } + if got := g.sent(); len(got) != 1 || strings.Join(got[0], ",") != "reasoning:sealed-by-grok" { + t.Fatalf("grok was then sent %q", got) + } +} diff --git a/internal/gateway/gateway.go b/internal/gateway/gateway.go index fca426aa..890cfb58 100644 --- a/internal/gateway/gateway.go +++ b/internal/gateway/gateway.go @@ -887,12 +887,12 @@ func (s *Server) serve(w http.ResponseWriter, r *http.Request, from provider.Pro } tr := s.trace.begin(Route{Time: start, Agent: call.Agent, Kind: call.Kind, Model: call.Model, Effort: requestEffort(from, body), Provider: p.ID, Group: group, Rule: hit, Nested: nested, Affinity: shown, Order: pl.order, Left: pl.left}) var skipped []string - sent := "" // the reasoning the last try's model was asked for - where := "" // the last try's provider.Where, for the usage - again := 0 // times the last one left has been tried again - resealed := false // the conversation's reasoning sealed by another account taken out - floored := false // the reply's length raised to what the provider takes - var other *Try // the first failure that wasn't an allowance run out + sent := "" // the reasoning the last try's model was asked for + where := "" // the last try's provider.Where, for the usage + again := 0 // times the last one left has been tried again + resealed := 0 // what of the conversation another account sealed was taken out: its reasoning, then its compaction + floored := false // the reply's length raised to what the provider takes + var other *Try // the first failure that wasn't an allowance run out for i := 0; i < len(cands); i++ { c := cands[i] last := i == len(cands)-1 @@ -983,15 +983,29 @@ 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()) { + if resealed < 2 && 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 + // read as another status. Refused again, or with no reasoning + // to leave out, it's the compaction OpenAI sealed that goes + // (waroy: Codex compacted on GPT, then switched to Grok) + b, ok := []byte(nil), false + if resealed == 0 { + resealed = 1 + if b, ok = withoutReasoning(body); ok { + refused(stuck, c.who(), attemptBody, "reasoning") + } + } + if !ok { + resealed = 2 + if b, ok = withoutCompaction(body); ok { + refused(stuck, c.who(), attemptBody, compactionKinds...) + } + } + if ok { + body = b try.Fail = failForeign s.trace.update(tr, func(t *Route) { t.Tries[len(t.Tries)-1] = try }) i--