Add native Responses API upstream proxy

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

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,