fix(gateway): normalize standalone Responses tool outputs (#544)

* fix(gateway): normalize standalone Responses tool outputs

* fix(gateway): preserve native notifications and tool result ordering
This commit is contained in:
Zhaofeng authored and GitHub committed 2026-10-02 18:26:23 +08:00
1 parent 723b0f34f5
commit 1a283b95fc
7 files changed
+536 -5

No files matched your search

+5
View File
@@ -1561,6 +1561,11 @@ func sinceStart(before, d time.Duration) int64 {
// attempt sends a request to one provider. call.To stays empty when the
// provider has no endpoint to send it to.
func (s *Server) attempt(w http.ResponseWriter, r *http.Request, from provider.Protocol, p provider.Provider, model string, body []byte, call *Call) (int, string) {
// Normalize for the actual destination, separately on each fallback.
// Native ChatGPT accounts accept standalone tool outputs themselves.
if from == provider.Responses && (p.Account == nil || p.Account.Agent != "codex") {
body = orphanedToolOutputs(body)
}
// every request to the provider goes through its own proxy, if it has
// one (#237)
r = r.WithContext(p.Via(r.Context()))
+4 -3
View File
@@ -43,9 +43,10 @@ type Part struct {
Args json.RawMessage // a JSON object
// tool_result
CallID string
IsError bool
Images []Part // the images the tool returned beside its text
CallID string
IsError bool
Images []Part // the images the tool returned beside its text
Standalone map[string]any // native Responses notification with no call ID
// thinking
Signature string
+175
View File
@@ -0,0 +1,175 @@
package gateway
import (
"bytes"
"encoding/json"
"io"
"net/http"
"strings"
"testing"
"github.com/yetone/magpie/internal/provider"
)
// Exercise both real ingress handlers and the two Responses forwarding paths.
// A search-capable provider relays; offering web_search to one that cannot
// search makes the gateway translate even when its upstream is Responses.
func TestStandaloneToolOutputResponsesRoutes(t *testing.T) {
for _, endpoint := range []string{"/v1/responses", CodexPath + "/responses"} {
for _, relay := range []bool{true, false} {
route := "translated"
if relay {
route = "relayed"
}
for _, tc := range []struct {
name, input string
standalone bool
}{
{"ordinary", `[{"type":"message","role":"user","content":[{"type":"input_text","text":"ordinary"}]}]`, false},
{"paired", `[{"type":"function_call","call_id":"call_synthetic","name":"echo","arguments":"{}"},{"type":"function_call_output","call_id":"call_synthetic","output":"paired result"}]`, false},
{"standalone", `[{"type":"function_call","call_id":"call_synthetic","name":"echo","arguments":"{}"},{"type":"function_call_output","call_id":"call_synthetic","output":"paired result"},{"type":"function_call_output","id":"fco_synthetic","name":"send_message_to_thread","namespace":"codex_app","output":"<codex_delegation>synthetic delivery</codex_delegation>"}]`, true},
{"custom_missing", `[{"type":"custom_tool_call_output","output":"synthetic delivery"}]`, true},
{"custom_empty", `[{"type":"custom_tool_call_output","call_id":"","output":"synthetic delivery"}]`, true},
{"empty_id", `[{"type":"function_call_output","call_id":"","output":"synthetic delivery"}]`, true},
} {
t.Run(endpoint+"/"+route+"/"+tc.name, func(t *testing.T) {
f := &fake{t: t, reply: sse(`data: {"type":"response.completed","response":{"id":"r1","status":"completed","output":[]}}`)}
f.refuse = func(body []byte) (int, string) {
// Use the wire spelling of call_id, not Go's default field matching.
var sent struct {
Input []map[string]any `json:"input"`
}
if err := json.Unmarshal(body, &sent); err != nil {
t.Fatal(err)
}
for _, item := range sent.Input {
if item["type"] == "function_call_output" && (item["call_id"] == nil || item["call_id"] == "") {
return 400, `{"error":{"message":"call_id required"}}`
}
}
return 0, ""
}
setup(t, provider.Responses, f)
p, err := provider.Find("fake")
if err != nil {
t.Fatal(err)
}
p.Searches = relay
if err := provider.Save(*p); err != nil {
t.Fatal(err)
}
body := `{"model":"fake/m1","stream":true,"tools":[{"type":"web_search"}],"input":` + tc.input + `}`
var code int
var response string
if endpoint == CodexPath+"/responses" {
code, response = codexPost(t, body)
} else {
code, response = post(t, endpoint, body)
}
if code != 200 {
t.Fatalf("status %d: %s", code, response)
}
var sent struct {
Input []map[string]any `json:"input"`
}
if err := json.Unmarshal(f.got, &sent); err != nil {
t.Fatal(err)
}
calls, outputs := 0, 0
for _, item := range sent.Input {
switch item["type"] {
case "function_call":
calls++
if item["call_id"] != "call_synthetic" {
t.Fatalf("call changed: %v", item)
}
case "function_call_output":
outputs++
if item["call_id"] != "call_synthetic" || item["output"] != "paired result" {
t.Fatalf("paired result changed: %v", item)
}
}
}
if tc.name == "paired" || tc.name == "standalone" {
if calls != 1 || outputs != 1 {
t.Fatalf("lost pair: %s", f.got)
}
} else if calls != 0 || outputs != 0 {
t.Fatalf("invented pair: %s", f.got)
}
if tc.standalone {
if !strings.Contains(string(f.got), "synthetic delivery") {
t.Fatalf("lost delivery: %s", f.got)
}
last := sent.Input[len(sent.Input)-1]
if last["type"] != "message" || last["role"] != "user" {
t.Fatalf("not user context: %v", last)
}
} else if tc.name == "ordinary" && !strings.Contains(string(f.got), "ordinary") {
t.Fatalf("lost ordinary message: %s", f.got)
}
})
}
}
}
}
func TestOrphanToolOutputsContentAndNoops(t *testing.T) {
for _, body := range []string{`not json`, `{}`, `{"input":"hello"}`, `{"input":[{"type":"function_call_output","call_id":"paired","output":"ok"}]}`, `{"input":[{"role":"user","content":"ordinary"}]}`} {
if got := orphanedToolOutputs([]byte(body)); string(got) != body {
t.Fatalf("unchanged input rewritten: %s", got)
}
}
body := []byte(`{"model":"fake/m1","previous_response_id":"resp_synthetic","input":[{"type":"function_call_output","output":[{"type":"text","text":"look"},{"type":"input_image","image_url":"data:image/png;base64,AQID"}]},{"type":"function_call_output","output":""}]}`)
original := bytes.Clone(body)
fixed := orphanedToolOutputs(body)
if !bytes.Equal(body, original) {
t.Fatal("mutated caller's request buffer")
}
var q struct {
Previous string `json:"previous_response_id"`
Input []struct {
Content []map[string]any `json:"content"`
} `json:"input"`
}
if err := json.Unmarshal(fixed, &q); err != nil {
t.Fatal(err)
}
if q.Previous != "resp_synthetic" || len(q.Input) != 2 || len(q.Input[0].Content) != 2 {
t.Fatalf("lost context: %s", fixed)
}
if q.Input[0].Content[0]["text"] != "look" || q.Input[0].Content[1]["image_url"] != "data:image/png;base64,AQID" || q.Input[1].Content[0]["text"] != "Tool result received." {
t.Fatalf("lost output content: %s", fixed)
}
if !bytes.Equal(orphanedToolOutputs(fixed), fixed) {
t.Fatal("normalization is not idempotent")
}
}
// OpenAI's native backend supports standalone outputs; do not rewrite that
// path merely because Codex happened to send it through openai_base_url.
func TestCodexNativeStandaloneOutputUnchanged(t *testing.T) {
var got []byte
chatgpt(t, func(w http.ResponseWriter, r *http.Request) {
got, _ = io.ReadAll(r.Body)
w.Header().Set("Content-Type", "text/event-stream")
io.WriteString(w, sse(`data: {"type":"response.completed","response":{"id":"r1","status":"completed","output":[]}}`))
})
body := `{"model":"gpt-5.5","stream":true,"input":[{"type":"function_call_output","name":"send_message_to_thread","namespace":"codex_app","output":"synthetic native notification"}]}`
code, response := codexPost(t, body)
if code != 200 {
t.Fatalf("%d %s", code, response)
}
var q struct {
Input []map[string]any `json:"input"`
}
if err := json.Unmarshal(got, &q); err != nil {
t.Fatal(err)
}
if len(q.Input) != 1 || q.Input[0]["type"] != "function_call_output" || q.Input[0]["name"] != "send_message_to_thread" || q.Input[0]["namespace"] != "codex_app" {
t.Fatalf("native notification changed: %s", got)
}
if _, ok := q.Input[0]["call_id"]; ok {
t.Fatalf("native call ID invented: %s", got)
}
}
+100 -2
View File
@@ -221,7 +221,8 @@ func parseResponses(body []byte) (*Request, error) {
// written to the vendor's cache again (Anthropic's system comes
// first; a Claude subscription's waiting run is found by it) (#502)
replied := false
for _, it := range items {
var rawItems []json.RawMessage // decoded only for native standalone outputs
for i, it := range items {
if it.Role == "assistant" || strings.HasPrefix(it.Type, "function_call") || strings.HasPrefix(it.Type, "custom_tool_call") || strings.HasPrefix(it.Type, "tool_search") || it.Type == "reasoning" {
replied = true
}
@@ -276,7 +277,17 @@ func parseResponses(body []byte) (*Request, error) {
r.Messages = append(r.Messages, Message{Role: "user", Parts: []Part{{Kind: ToolResult, CallID: it.CallID, Text: searchFound(it.Tools)}}})
case it.Type == "function_call_output" || it.Type == "custom_tool_call_output":
out, images := toolOutput(it.Output)
r.Messages = append(r.Messages, Message{Role: "user", Parts: []Part{{Kind: ToolResult, CallID: it.CallID, Text: out, Images: images}}})
part := Part{Kind: ToolResult, CallID: it.CallID, Text: out, Images: images}
if it.CallID == "" {
// Native account requests may still need translation to
// collect a streamed response for a non-streaming client.
// Retain the original notification, including absent IDs.
if rawItems == nil {
_ = json.Unmarshal(q.Input, &rawItems)
}
_ = json.Unmarshal(rawItems[i], &part.Standalone)
}
r.Messages = append(r.Messages, Message{Role: "user", Parts: []Part{part}})
case it.Type == "reasoning":
// the reasoning itself when the item carries it, else its
// summary (all magpie gives a client of a translated reply)
@@ -401,6 +412,89 @@ func parseResponses(body []byte) (*Request, error) {
return r, nil
}
// orphanedToolOutputs turns a tool result with no call ID into a user message.
// Codex uses standalone outputs for cross-thread notifications. Backends
// requiring paired outputs still need the delivered text, without a fake call ID.
func orphanedToolOutputs(body []byte) []byte {
var q map[string]json.RawMessage
if json.Unmarshal(body, &q) != nil {
return body
}
var input []json.RawMessage
if json.Unmarshal(q["input"], &input) != nil {
return body
}
changed := false
out := make([]json.RawMessage, 0, len(input))
var pending []json.RawMessage
inCalls, hadResults := false, false
flush := func() {
out = append(out, pending...)
pending = nil
}
for _, raw := range input {
var item struct {
Type string `json:"type"`
CallID string `json:"call_id"`
Output json.RawMessage `json:"output"`
}
if json.Unmarshal(raw, &item) != nil {
flush()
inCalls = false
out = append(out, raw)
continue
}
isOutput := item.Type == "function_call_output" || item.Type == "custom_tool_call_output"
if !isOutput || item.CallID != "" {
switch item.Type {
case "function_call", "custom_tool_call", "tool_search_call":
if hadResults {
flush()
}
inCalls, hadResults = true, false
case "function_call_output", "custom_tool_call_output", "tool_search_output":
// Keep a run of tool results directly after its calls.
hadResults = true
default:
flush()
inCalls = false
}
out = append(out, raw)
continue
}
text, images := toolOutput(item.Output)
content := []map[string]any{}
if text != "" {
content = append(content, map[string]any{"type": "input_text", "text": text})
}
for _, image := range images {
content = append(content, map[string]any{"type": "input_image", "image_url": dataURL(image)})
}
if len(content) == 0 {
content = append(content, map[string]any{"type": "input_text", "text": "Tool result received."})
}
message, _ := marshalPlain(map[string]any{"type": "message", "role": "user", "content": content})
if inCalls {
// Chat rejects a user message between tool_calls and results,
// including between the results of parallel calls.
pending = append(pending, message)
} else {
out = append(out, message)
}
changed = true
}
flush()
if !changed {
return body
}
q["input"], _ = marshalPlain(out)
encoded, err := marshalPlain(q)
if err != nil {
return body
}
return encoded
}
// mergeTurns joins consecutive messages of the same role, since the
// Responses API splits an assistant turn into one item per part.
func mergeTurns(msgs []Message) []Message {
@@ -496,6 +590,10 @@ func buildResponses(r *Request, model, host string, rejectTemp bool) []byte {
input = append(input, map[string]any{"type": "function_call", "call_id": id, "name": p.Name, "arguments": argsString(p)})
case ToolResult:
flushMsg()
if p.Standalone != nil {
input = append(input, p.Standalone)
continue
}
var output any = p.Text
if len(p.Images) > 0 {
// an output can be a list of text and images
+223
View File
@@ -0,0 +1,223 @@
package gateway
import (
"bytes"
"encoding/json"
"fmt"
"io"
"net/http"
"net/http/httptest"
"strings"
"testing"
"github.com/yetone/magpie/internal/provider"
)
func TestStandaloneChatOrdering(t *testing.T) {
for _, endpoint := range []string{"/v1/responses", CodexPath + "/responses"} {
for _, kind := range []string{"function", "custom_tool"} {
call := func(id string) string {
if kind == "function" {
return fmt.Sprintf(`{"type":"function_call","call_id":%q,"name":"echo","arguments":"{}"}`, id)
}
return fmt.Sprintf(`{"type":"custom_tool_call","call_id":%q,"name":"echo","input":"echo"}`, id)
}
output := func(id, text string) string {
return fmt.Sprintf(`{"type":%q,"call_id":%q,"output":%q}`, kind+"_call_output", id, text)
}
notice := fmt.Sprintf(`{"type":%q,"output":"notice"}`, kind+"_call_output")
for _, tc := range []struct{ name, input, roles string }{
{"before_calls", notice + "," + call("A") + "," + output("A", "result A"), "user,assistant,tool"},
{"before_result", call("A") + "," + notice + "," + output("A", "result A"), "assistant,tool,user"},
{"between_parallel_results", call("A") + "," + call("B") + "," + output("A", "result A") + "," + notice + "," + output("B", "result B"), "assistant,tool,tool,user"},
{"empty_id", call("A") + "," + output("", "notice") + "," + output("A", "result A"), "assistant,tool,user"},
} {
t.Run(endpoint+"/"+kind+"/"+tc.name, func(t *testing.T) {
f := &fake{t: t, reply: sse(`data: {"id":"ok","choices":[{"index":0,"delta":{"content":"ok"},"finish_reason":"stop"}]}`, `data: [DONE]`)}
var roles []string
f.refuse = func(body []byte) (int, string) {
var q struct {
Messages []struct {
Role string `json:"role"`
Calls []struct {
ID string `json:"id"`
} `json:"tool_calls"`
ID string `json:"tool_call_id"`
} `json:"messages"`
}
if err := json.Unmarshal(body, &q); err != nil {
t.Error(err)
return 400, `{"error":{"message":"bad JSON"}}`
}
pending := map[string]bool{}
for _, m := range q.Messages {
roles = append(roles, m.Role)
if len(pending) > 0 && m.Role != "tool" {
return 400, `{"error":{"message":"interrupted tool results"}}`
}
if m.Role == "tool" {
if !pending[m.ID] {
return 400, `{"error":{"message":"unpaired result"}}`
}
delete(pending, m.ID)
}
for _, c := range m.Calls {
pending[c.ID] = true
}
}
if len(pending) > 0 {
return 400, `{"error":{"message":"missing result"}}`
}
return 0, ""
}
setup(t, provider.Chat, f)
body := `{"model":"fake/m1","input":[` + tc.input + `]}`
var code int
var response string
if endpoint == "/v1/responses" {
code, response = post(t, endpoint, body)
} else {
code, response = codexPost(t, body)
}
if code != 200 {
t.Fatalf("%d: %s", code, response)
}
if got := strings.Join(roles, ","); got != tc.roles {
t.Fatalf("roles %s, want %s: %s", got, tc.roles, f.got)
}
for _, text := range []string{"notice", "result A"} {
if !bytes.Contains(f.got, []byte(text)) {
t.Fatalf("lost %s: %s", text, f.got)
}
}
})
}
}
}
}
func TestStandaloneNativeAccounts(t *testing.T) {
for _, kind := range []string{"function_call_output", "custom_tool_call_output"} {
for _, id := range []string{"", `,"call_id":""`} {
for _, tc := range []struct {
name, path, model, pin, account string
saved bool
}{
{"direct", CodexPath + "/responses", "gpt-5.5", "", "acct-1", false},
{"saved", CodexPath + "/responses", "gpt-5.5", "", "acct-1", true},
{"fixed", CodexPath + "/responses", "gpt-5.5", "spare@example.com", "acct-2", true},
{"namespaced", CodexPath + "/responses", "codex/gpt-5.5", "spare@example.com", "acct-2", true},
{"v1_saved", "/v1/responses", "codex/gpt-5.5", "", "", true},
{"v1_fixed", "/v1/responses", "codex/gpt-5.5", "spare@example.com", "acct-2", true},
} {
for _, stream := range []bool{true, false} {
if !stream && !tc.saved {
continue
}
t.Run(fmt.Sprintf("%s/%s/%s/stream=%t", kind, id, tc.name, stream), func(t *testing.T) {
if tc.saved {
codexSignedIn(t, "spare@example.com")
} else {
fresh(t)
provider.ForgetAccounts()
t.Cleanup(provider.ForgetAccounts)
}
var got []byte
var account string
up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
got, _ = io.ReadAll(r.Body)
account = r.Header.Get("chatgpt-account-id")
w.Header().Set("Content-Type", "text/event-stream")
io.WriteString(w, sse(`data: {"type":"response.completed","response":{"id":"r1","status":"completed","output":[]}}`))
}))
defer up.Close()
old := provider.CodexBase
provider.CodexBase = up.URL + "/backend-api/codex"
defer func() { provider.CodexBase = old }()
body := fmt.Sprintf(`{"model":%q,"stream":%t,"input":[{"type":%q,"id":"standalone_synthetic","name":"send_message_to_thread","namespace":"codex_app","output":"native notice"%s}]}`, tc.model, stream, kind, id)
req := httptest.NewRequest("POST", tc.path, strings.NewReader(body))
req.Header.Set("Authorization", "Bearer chatgpt-token")
req.Header.Set("chatgpt-account-id", "acct-1")
req.Header.Set(AccountHeader, tc.pin)
rec := httptest.NewRecorder()
New().Handler().ServeHTTP(rec, req)
if rec.Code != 200 {
t.Fatalf("%d %s", rec.Code, rec.Body.String())
}
if tc.account != "" && account != tc.account {
t.Fatalf("account %q, want %q", account, tc.account)
}
var sent struct {
Input []map[string]any `json:"input"`
}
json.Unmarshal(got, &sent)
if len(sent.Input) != 1 {
t.Fatalf("input: %s", got)
}
item := sent.Input[0]
if item["id"] != "standalone_synthetic" || item["type"] != kind || item["name"] != "send_message_to_thread" || item["namespace"] != "codex_app" || item["output"] != "native notice" {
t.Fatalf("native notification changed: %v", item)
}
value, present := item["call_id"]
if (id != "") != present || present && value != "" {
t.Fatalf("native ID changed: %v", item)
}
})
}
}
}
}
}
func TestStandalonePlainEncoding(t *testing.T) {
for _, kind := range []string{"function_call_output", "custom_tool_call_output"} {
for _, body := range []string{`{ "z": "<keep>&", "input": [ { "role": "user", "content": "hi" } ], "a": 1 }`, fmt.Sprintf(`{ "input": [{"type":%q,"call_id":"A","output":"<paired>"}], "instructions":"<keep>&" }`, kind)} {
if got := orphanedToolOutputs([]byte(body)); string(got) != body {
t.Fatalf("no-op bytes changed: %s", got)
}
}
body := []byte(fmt.Sprintf(`{ "z": 1, "instructions": "<keep>&", "input": [{"type":%q,"output":"<notice>&"}], "a": 2 }`, kind))
got := orphanedToolOutputs(body)
for _, literal := range []string{"<keep>&", "<notice>&"} {
if !bytes.Contains(got, []byte(literal)) {
t.Fatalf("escaped %s: %s", literal, got)
}
}
if !bytes.Equal(got, orphanedToolOutputs(got)) {
t.Fatal("not idempotent")
}
}
}
// A failed third-party attempt must not carry its normalized body into a
// native ChatGPT fallback for the same request.
func TestStandaloneNativeFallback(t *testing.T) {
codexSignedIn(t, "spare@example.com")
var third, native []byte
up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
b, _ := io.ReadAll(r.Body)
if r.URL.Path == "/v1/responses" {
third = b
w.WriteHeader(429)
io.WriteString(w, `{"error":{"message":"rate limited"}}`)
return
}
native = b
w.Header().Set("Content-Type", "text/event-stream")
io.WriteString(w, sse(`data: {"type":"response.completed","response":{"id":"r1","status":"completed","output":[]}}`))
}))
defer up.Close()
old := provider.CodexBase
provider.CodexBase = up.URL + "/backend-api/codex"
defer func() { provider.CodexBase = old }()
if err := provider.Save(provider.Provider{ID: "third", Name: "Third", Key: "synthetic", Models: []string{"m1"}, Responses: up.URL + "/v1", Fallback: []string{"codex/gpt-5.5"}}); err != nil {
t.Fatal(err)
}
code, response := post(t, "/v1/responses", `{"model":"third/m1","stream":true,"input":[{"type":"function_call_output","name":"send_message_to_thread","namespace":"codex_app","output":"notice"}]}`)
if code != 200 {
t.Fatalf("%d %s", code, response)
}
if !bytes.Contains(third, []byte(`"role":"user"`)) || !bytes.Contains(native, []byte(`"type":"function_call_output"`)) || !bytes.Contains(native, []byte(`"namespace":"codex_app"`)) {
t.Fatalf("third=%s native=%s", third, native)
}
}
+7
View File
@@ -379,6 +379,13 @@ func codexInput(input []any) []any {
if it["type"] == "item_reference" {
continue
}
// Codex delivers standalone notifications without a call ID. The
// native backend understands their name/namespace and item ID;
// they are not historical results whose calls have gone missing.
if (it["type"] == "function_call_output" || it["type"] == "custom_tool_call_output") && (it["call_id"] == nil || it["call_id"] == "") {
out = append(out, it)
continue
}
delete(it, "id")
switch t, _ := it["type"].(string); t {
case "function_call_output", "custom_tool_call_output", "local_shell_call_output":
+22
View File
@@ -204,3 +204,25 @@ func TestCodexBodyServiceTier(t *testing.T) {
t.Errorf("none became %v", v)
}
}
// Standalone notifications retain the metadata the native backend recognizes;
// paired outputs and ID-bearing historical orphans keep their existing rules.
func TestCodexStandaloneNotifications(t *testing.T) {
t.Setenv("CODEX_HOME", t.TempDir())
t.Setenv("XDG_CACHE_HOME", t.TempDir())
for _, kind := range []string{"function_call_output", "custom_tool_call_output"} {
for _, id := range []string{"", `,"call_id":""`} {
body := []byte(`{"model":"gpt-5.5","input":[{"type":"` + kind + `","id":"standalone","name":"send_message_to_thread","namespace":"codex_app","output":"native notice"` + id + `}]}`)
var before, after struct {
Input []map[string]any `json:"input"`
}
json.Unmarshal(body, &before)
json.Unmarshal(codexBody(body), &after)
b, _ := json.Marshal(before.Input)
a, _ := json.Marshal(after.Input)
if string(a) != string(b) {
t.Fatalf("standalone changed: %s -> %s", b, a)
}
}
}
}