Merge pull request 'Add native Responses API upstream proxy' (#2) from fix/native-openai-responses into main

Reviewed-on: #2

internal/gateway/gateway_test.go

@@ -19,15 +19,35 @@ import (
 type fakeUpstream struct {
 	lastModel   string
 	lastAuth    string
+	lastPath    string
 	lastRequest map[string]any
 }
 
 func (f *fakeUpstream) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	f.lastPath = r.URL.Path
 	if r.Method == http.MethodGet && r.URL.Path == "/models" {
 		w.Header().Set("Content-Type", "application/json")
 		_, _ = io.WriteString(w, `{"object":"list","data":[{"id":"ds-chat"},{"id":"ds-haiku"}]}`)
 		return
 	}
+	if r.Method == http.MethodPost && r.URL.Path == "/v1/responses" {
+		f.lastAuth = r.Header.Get("Authorization")
+		body, _ := io.ReadAll(r.Body)
+		_ = json.Unmarshal(body, &f.lastRequest)
+		f.lastModel, _ = f.lastRequest["model"].(string)
+		stream, _ := f.lastRequest["stream"].(bool)
+		if stream {
+			w.Header().Set("Content-Type", "text/event-stream")
+			_, _ = io.WriteString(w, "event: response.completed\ndata: {\"type\":\"response.completed\"}\n\n")
+			return
+		}
+		w.Header().Set("Content-Type", "application/json")
+		_ = json.NewEncoder(w).Encode(map[string]any{
+			"id": "resp_native", "object": "response", "status": "completed", "model": f.lastModel,
+			"output": []any{}, "usage": map[string]any{"input_tokens": 9, "output_tokens": 2, "total_tokens": 11},
+		})
+		return
+	}
 	if r.URL.Path != "/chat/completions" {
 		w.WriteHeader(http.StatusNotFound)
 		return
@@ -290,6 +310,74 @@ func TestResponsesStream(t *testing.T) {
 	}
 }
 
+func TestNativeResponsesProtocol(t *testing.T) {
+	for _, stream := range []bool{false, true} {
+		name := "non-streaming"
+		if stream {
+			name = "streaming"
+		}
+		t.Run(name, func(t *testing.T) {
+			env := newCodexTestEnv(t)
+			provider, err := env.store.AddProvider("native-responses", "", env.upstreamURL+"/v1", "up-key",
+				config.Models{Default: "native-model"}, config.TargetCodex, config.ProtocolOpenAIResponses)
+			if err != nil {
+				t.Fatalf("AddProvider: %v", err)
+			}
+			if _, err := env.store.ActivateCodex(provider.ID); err != nil {
+				t.Fatalf("ActivateCodex: %v", err)
+			}
+
+			body := map[string]any{
+				"model":     "gpt-5-codex",
+				"input":     "What model are you?",
+				"stream":    stream,
+				"reasoning": map[string]any{"effort": "high"},
+				"tools": []any{
+					map[string]any{"type": "web_search_preview"},
+					map[string]any{"type": "function", "name": "read_file", "parameters": map[string]any{"type": "object"}},
+				},
+			}
+			resp := env.post(t, "/v1/responses", body, env.store.Config().RelayToken)
+			defer resp.Body.Close()
+			if resp.StatusCode != http.StatusOK {
+				data, _ := io.ReadAll(resp.Body)
+				t.Fatalf("status = %d, body = %s", resp.StatusCode, data)
+			}
+			data, err := io.ReadAll(resp.Body)
+			if err != nil {
+				t.Fatalf("read response: %v", err)
+			}
+			if stream {
+				if !strings.Contains(string(data), "event: response.completed") {
+					t.Fatalf("stream response = %q", data)
+				}
+			} else if !strings.Contains(string(data), `"id":"resp_native"`) {
+				t.Fatalf("native response = %s", data)
+			}
+
+			if env.upstream.lastPath != "/v1/responses" {
+				t.Errorf("upstream path = %q, want /v1/responses", env.upstream.lastPath)
+			}
+			if env.upstream.lastAuth != "Bearer up-key" {
+				t.Errorf("upstream auth = %q", env.upstream.lastAuth)
+			}
+			if env.upstream.lastModel != "native-model" {
+				t.Errorf("upstream model = %q, want model mapping", env.upstream.lastModel)
+			}
+			if env.upstream.lastRequest["input"] != "What model are you?" || env.upstream.lastRequest["messages"] != nil {
+				t.Errorf("request was converted instead of passed through: %v", env.upstream.lastRequest)
+			}
+			if env.upstream.lastRequest["reasoning"] == nil {
+				t.Errorf("native reasoning field was dropped: %v", env.upstream.lastRequest)
+			}
+			tools := env.upstream.lastRequest["tools"].([]any)
+			if len(tools) != 2 || tools[0].(map[string]any)["type"] != "web_search_preview" {
+				t.Errorf("native tools were changed: %v", tools)
+			}
+		})
+	}
+}
+
 func TestResponsesRejectsBadToken(t *testing.T) {
 	env := newCodexTestEnv(t)
 	resp := env.post(t, "/v1/responses", responsesBody("gpt-5-codex", false), "crl_wrong")

internal/gateway/responses.go

@@ -1,20 +1,30 @@
 package gateway
 
 import (
+	"bytes"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
 	"net/http"
+	"strings"
+	"time"
 
 	"gitea.rutua.cn/kate522/CrossRelay/internal/chatcmpl"
+	"gitea.rutua.cn/kate522/CrossRelay/internal/config"
 	"gitea.rutua.cn/kate522/CrossRelay/internal/responses"
 )
 
 // handleResponses serves POST /v1/responses in both streaming and
 // non-streaming modes.
 func (s *Server) handleResponses(w http.ResponseWriter, r *http.Request) {
+	body, err := io.ReadAll(io.LimitReader(r.Body, 64<<20))
+	if err != nil {
+		writeResponsesError(w, http.StatusBadRequest, "invalid_request_error", "invalid request body: "+err.Error())
+		return
+	}
 	var req responses.Request
-	if err := decodeJSON(r, &req, 64<<20); err != nil {
+	if err := json.Unmarshal(body, &req); err != nil {
 		writeResponsesError(w, http.StatusBadRequest, "invalid_request_error", "invalid request body: "+err.Error())
 		return
 	}
@@ -25,6 +35,20 @@ func (s *Server) handleResponses(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 	model := provider.ResolveModel(req.Model)
+	rec := recordFrom(r)
+	if provider.Protocol == config.ProtocolOpenAIResponses {
+		rec.update(func(e *LogEntry) {
+			e.Model = model
+			e.InputTokens = (len(body) + 3) / 4
+		})
+		s.proxyNativeResponses(w, r, provider, body, req.Model, model, req.Stream, rec)
+		return
+	}
+	if provider.Protocol != config.ProtocolOpenAIChat {
+		writeResponsesError(w, http.StatusNotImplemented, "unsupported_protocol",
+			fmt.Sprintf("upstream protocol %q is not supported for Codex Responses requests", provider.Protocol))
+		return
+	}
 
 	chatReq, err := responses.ConvertRequest(&req, model)
 	if err != nil {
@@ -32,7 +56,6 @@ func (s *Server) handleResponses(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	rec := recordFrom(r)
 	rec.update(func(e *LogEntry) {
 		e.Model = model
 		e.InputTokens = responses.EstimateTokens(chatReq)
@@ -58,6 +81,134 @@ func (s *Server) handleResponses(w http.ResponseWriter, r *http.Request) {
 	writeJSON(w, http.StatusOK, out)
 }
 
+func (s *Server) proxyNativeResponses(w http.ResponseWriter, r *http.Request, provider *config.Provider,
+	body []byte, requestedModel, model string, stream bool, rec *record) {
+	if model != requestedModel {
+		var payload map[string]json.RawMessage
+		if err := json.Unmarshal(body, &payload); err != nil || payload == nil {
+			writeResponsesError(w, http.StatusBadRequest, "invalid_request_error", "invalid request body: expected a JSON object")
+			return
+		}
+		encodedModel, err := json.Marshal(model)
+		if err != nil {
+			writeResponsesError(w, http.StatusInternalServerError, "api_error", "encode model: "+err.Error())
+			return
+		}
+		payload["model"] = encodedModel
+		body, err = json.Marshal(payload)
+		if err != nil {
+			writeResponsesError(w, http.StatusInternalServerError, "api_error", "encode request: "+err.Error())
+			return
+		}
+	}
+
+	upstreamReq, err := http.NewRequestWithContext(r.Context(), http.MethodPost,
+		strings.TrimRight(provider.BaseURL, "/")+"/responses", bytes.NewReader(body))
+	if err != nil {
+		writeResponsesError(w, http.StatusBadGateway, "api_error", "create upstream request: "+err.Error())
+		return
+	}
+	for name, values := range r.Header {
+		if isHopByHopHeader(name) || strings.EqualFold(name, "Authorization") || strings.EqualFold(name, "X-API-Key") || strings.EqualFold(name, "Content-Length") {
+			continue
+		}
+		for _, value := range values {
+			upstreamReq.Header.Add(name, value)
+		}
+	}
+	upstreamReq.Header.Set("Content-Type", "application/json")
+	if r.Header.Get("Accept") == "" {
+		if stream {
+			upstreamReq.Header.Set("Accept", "text/event-stream")
+		} else {
+			upstreamReq.Header.Set("Accept", "application/json")
+		}
+	}
+	if provider.APIKey != "" {
+		upstreamReq.Header.Set("Authorization", "Bearer "+provider.APIKey)
+	}
+
+	upstreamResp, err := (&http.Client{Timeout: 30 * time.Minute}).Do(upstreamReq)
+	if err != nil {
+		rec.update(func(e *LogEntry) { e.Error = err.Error() })
+		writeResponsesError(w, http.StatusBadGateway, "api_error", fmt.Sprintf("upstream request failed: %v", err))
+		return
+	}
+	defer upstreamResp.Body.Close()
+
+	for name, values := range upstreamResp.Header {
+		if isHopByHopHeader(name) {
+			continue
+		}
+		for _, value := range values {
+			w.Header().Add(name, value)
+		}
+	}
+
+	if stream {
+		w.WriteHeader(upstreamResp.StatusCode)
+		if err := copyStreamingResponse(w, upstreamResp.Body); err != nil {
+			rec.update(func(e *LogEntry) { e.Error = err.Error() })
+		}
+		return
+	}
+
+	responseBody, err := io.ReadAll(io.LimitReader(upstreamResp.Body, 64<<20))
+	if err != nil {
+		rec.update(func(e *LogEntry) { e.Error = err.Error() })
+		writeResponsesError(w, http.StatusBadGateway, "api_error", "read upstream response: "+err.Error())
+		return
+	}
+	if upstreamResp.StatusCode >= http.StatusBadRequest {
+		rec.update(func(e *LogEntry) {
+			e.Error = fmt.Sprintf("upstream %d: %s", upstreamResp.StatusCode, strings.TrimSpace(string(responseBody)))
+		})
+	} else {
+		var result struct {
+			Usage *responses.Usage `json:"usage"`
+		}
+		if json.Unmarshal(responseBody, &result) == nil && result.Usage != nil {
+			rec.update(func(e *LogEntry) {
+				e.InputTokens = result.Usage.InputTokens
+				e.OutputTokens = result.Usage.OutputTokens
+			})
+		}
+	}
+	w.WriteHeader(upstreamResp.StatusCode)
+	_, _ = w.Write(responseBody)
+}
+
+func copyStreamingResponse(w http.ResponseWriter, body io.Reader) error {
+	buf := make([]byte, 32<<10)
+	for {
+		n, readErr := body.Read(buf)
+		if n > 0 {
+			if _, err := w.Write(buf[:n]); err != nil {
+				return err
+			}
+			if flusher, ok := w.(http.Flusher); ok {
+				flusher.Flush()
+			}
+		}
+		if errors.Is(readErr, io.EOF) {
+			return nil
+		}
+		if readErr != nil {
+			return readErr
+		}
+	}
+}
+
+func isHopByHopHeader(name string) bool {
+	switch strings.ToLower(name) {
+	case "connection", "keep-alive", "proxy-authenticate", "proxy-authorization", "proxy-connection",
+		"te", "trailer", "transfer-encoding", "upgrade":
+		return true
+	default:
+		return false
+	}
+}
+
 // handleResponsesStream relays a Chat Completions SSE stream as
 // Responses events.
 func (s *Server) handleResponsesStream(w http.ResponseWriter, r *http.Request, client *chatcmpl.Client,