diff --git a/internal/gateway/gateway_test.go b/internal/gateway/gateway_test.go index 310e017..b56c3e9 100644 --- a/internal/gateway/gateway_test.go +++ b/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") diff --git a/internal/gateway/responses.go b/internal/gateway/responses.go index 9279f06..2eb62bc 100644 --- a/internal/gateway/responses.go +++ b/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,