From dad92488e57a1d9e1807d50f1b2fe08005a32184 Mon Sep 17 00:00:00 2001 From: shaw Date: Tue, 7 Jul 2026 20:59:41 +0800 Subject: [PATCH] =?UTF-8?q?fix(messages):=20=E4=BF=AE=E5=A4=8D=20/v1/messa?= =?UTF-8?q?ges=20CC=20=E5=9B=9E=E9=80=80=E7=9A=84=E9=94=99=E8=AF=AF?= =?UTF-8?q?=E5=A4=84=E7=90=86=E6=97=81=E8=B7=AF=E5=B9=B6=E8=A1=A5=E9=BD=90?= =?UTF-8?q?=E6=B5=81=E5=BC=8F=E6=B5=8B=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 针对 #3795 合并后审计发现的问题： - 非 failover 上游错误分支改走共享 handleAnthropicErrorResponse：恢复 ops 上游错误记录、错误透传规则、cyber_policy 检测与按状态码映射的 错误 type（400→invalid_request_error 等）；删除随之失效的 mapUpstreamStatusToAnthropicStatus - 流式读错误对齐 forwardResponsesViaRawChatCompletions：scanner 出错时 不再 finalize（不再补发 message_stop 把截断伪装成正常完成），返回 stream usage incomplete 错误 - sawDone 由死变量改为与兄弟一致的无 [DONE] 哨兵 debug 日志 - 补充 fast policy 在本路径有意省略的说明注释 - 新增 8 个单测：流式收尾闭块、tool_call 分片聚合、length→max_tokens、 空流补帧、非 failover 400 走共享处理器、读错误不伪造收尾、 Responses 账号门控回归、非流式转换 --- .../openai_gateway_messages_chat_fallback.go | 44 +- ...nai_gateway_messages_chat_fallback_test.go | 389 ++++++++++++++++++ 2 files changed, 417 insertions(+), 16 deletions(-) create mode 100644 backend/internal/service/openai_gateway_messages_chat_fallback_test.go diff --git a/backend/internal/service/openai_gateway_messages_chat_fallback.go b/backend/internal/service/openai_gateway_messages_chat_fallback.go index c9381130b7..c33aa911ad 100644 --- a/backend/internal/service/openai_gateway_messages_chat_fallback.go +++ b/backend/internal/service/openai_gateway_messages_chat_fallback.go @@ -87,6 +87,10 @@ func (s *OpenAIGatewayService) forwardAnthropicViaRawChatCompletions( if normalizedBody, normalized := NormalizeGLMOpenAIReasoningEffort(chatBody, upstreamModel); normalized { chatBody = normalizedBody } + // Unlike forwardResponsesViaRawChatCompletions, applyOpenAIFastPolicyToBody + // is intentionally skipped: Anthropic Messages bodies carry no service_tier, + // so the converted Chat Completions body never contains one and the policy + // would always be a no-op on this path. logger.L().Debug("openai messages: forwarding via raw chat completions", zap.Int64("account_id", account.ID), @@ -182,8 +186,9 @@ func (s *OpenAIGatewayService) forwardAnthropicViaRawChatCompletions( RetryableOnSameAccount: account.IsPoolMode() && (account.IsPoolModeRetryableStatus(resp.StatusCode) || isOpenAITransientProcessingError(resp.StatusCode, upstreamMsg, respBody)), } } - writeAnthropicError(c, mapUpstreamStatusToAnthropicStatus(resp.StatusCode), "api_error", "Upstream error: "+upstreamMsg) - return nil, fmt.Errorf("upstream %d: %s", resp.StatusCode, upstreamMsg) + // Non-failover error: return Anthropic-formatted error to client via the + // shared compat handler (passthrough rules, ops recording, cyber_policy). + return s.handleAnthropicErrorResponse(resp, c, account, billingModel) } // 5. Convert response @@ -349,6 +354,22 @@ func (s *OpenAIGatewayService) streamChatCompletionsAsAnthropic( zap.String("request_id", requestID), ) } + // Broken upstream read: skip finalization so no synthetic message_stop + // masks the truncation, and surface the error to flag usage incomplete + // (mirrors forwardResponsesViaRawChatCompletions). + return &OpenAIForwardResult{ + RequestID: requestID, + Usage: usage, + Model: originalModel, + BillingModel: billingModel, + UpstreamModel: upstreamModel, + ReasoningEffort: reasoningEffort, + ServiceTier: serviceTier, + Stream: true, + Duration: time.Since(startTime), + FirstTokenMs: firstTokenMs, + ClientDisconnect: clientDisconnected, + }, fmt.Errorf("stream usage incomplete: %w", err) } // Finalize CC→Responses stream (emit response.completed) @@ -376,7 +397,11 @@ func (s *OpenAIGatewayService) streamChatCompletionsAsAnthropic( if !clientDisconnected { c.Writer.Flush() } - _ = sawDone + if !sawDone { + logger.L().Debug("openai messages chat fallback: upstream stream ended without done sentinel", + zap.String("request_id", requestID), + ) + } return &OpenAIForwardResult{ RequestID: requestID, @@ -392,16 +417,3 @@ func (s *OpenAIGatewayService) streamChatCompletionsAsAnthropic( ClientDisconnect: clientDisconnected, }, nil } - -func mapUpstreamStatusToAnthropicStatus(status int) int { - switch { - case status == 401: - return http.StatusBadGateway - case status == 429: - return http.StatusTooManyRequests - case status >= 500: - return http.StatusBadGateway - default: - return status - } -} diff --git a/backend/internal/service/openai_gateway_messages_chat_fallback_test.go b/backend/internal/service/openai_gateway_messages_chat_fallback_test.go new file mode 100644 index 0000000000..8bb15c81aa --- /dev/null +++ b/backend/internal/service/openai_gateway_messages_chat_fallback_test.go @@ -0,0 +1,389 @@ +//go:build unit + +package service + +import ( + "bytes" + "context" + "errors" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/Wei-Shaw/sub2api/internal/pkg/openai_compat" + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" + "github.com/tidwall/gjson" +) + +func forceChatMessagesFallbackAccount() *Account { + account := rawChatCompletionsTestAccount() + account.Extra = map[string]any{ + openai_compat.ExtraKeyResponsesMode: string(openai_compat.ResponsesSupportModeForceChatCompletions), + } + return account +} + +// errTailReader yields the given data, then returns err instead of io.EOF, +// simulating an upstream connection that breaks mid-stream. +type errTailReader struct { + data []byte + off int + err error +} + +func (r *errTailReader) Read(p []byte) (int, error) { + if r.off < len(r.data) { + n := copy(p, r.data[r.off:]) + r.off += n + return n, nil + } + return 0, r.err +} + +func (r *errTailReader) Close() error { return nil } + +func TestForwardAsAnthropic_ForceChatCompletionsNonStreaming(t *testing.T) { + gin.SetMode(gin.TestMode) + + body := []byte(`{"model":"gpt-5.4","max_tokens":32,"messages":[{"role":"user","content":"hello"}],"stream":false}`) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", bytes.NewReader(body)) + c.Request.Header.Set("Content-Type", "application/json") + + upstream := &httpUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"application/json"}, "x-request-id": []string{"rid_msg_chat_json"}}, + Body: io.NopCloser(strings.NewReader( + `{"id":"chatcmpl_json","object":"chat.completion","model":"gpt-5.4","choices":[{"index":0,"message":{"role":"assistant","content":"ok"},"finish_reason":"stop"}],"usage":{"prompt_tokens":3,"completion_tokens":2,"total_tokens":5,"prompt_tokens_details":{"cached_tokens":1}}}`, + )), + }} + svc := &OpenAIGatewayService{ + cfg: rawChatCompletionsTestConfig(), + httpUpstream: upstream, + } + + result, err := svc.ForwardAsAnthropic(context.Background(), c, forceChatMessagesFallbackAccount(), body, "", "") + require.NoError(t, err) + require.NotNil(t, result) + require.Equal(t, "http://upstream.example/v1/chat/completions", upstream.lastReq.URL.String()) + require.Equal(t, "hello", gjson.GetBytes(upstream.lastBody, "messages.0.content").String()) + require.False(t, gjson.GetBytes(upstream.lastBody, "input").Exists()) + require.True(t, gjson.GetBytes(upstream.lastBody, "stream_options").Exists() == false) + + require.Equal(t, http.StatusOK, rec.Code) + require.Equal(t, "assistant", gjson.Get(rec.Body.String(), "role").String()) + require.Equal(t, "ok", gjson.Get(rec.Body.String(), "content.0.text").String()) + require.Equal(t, 3, result.Usage.InputTokens) + require.Equal(t, 2, result.Usage.OutputTokens) + require.Equal(t, 1, result.Usage.CacheReadInputTokens) + require.False(t, result.Stream) +} + +// Covers the fully-new streaming composition: text block is still open when +// [DONE] arrives, so finalization must close it (content_block_stop) before +// message_delta / message_stop. +func TestForwardAsAnthropic_ForceChatCompletionsStreamingClosesOpenBlockOnDone(t *testing.T) { + gin.SetMode(gin.TestMode) + + body := []byte(`{"model":"gpt-5.4","max_tokens":32,"messages":[{"role":"user","content":"hello"}],"stream":true}`) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", bytes.NewReader(body)) + c.Request.Header.Set("Content-Type", "application/json") + + upstreamBody := strings.Join([]string{ + `data: {"id":"chatcmpl_s","object":"chat.completion.chunk","model":"gpt-5.4","choices":[{"index":0,"delta":{"role":"assistant"},"finish_reason":null}]}`, + "", + `data: {"id":"chatcmpl_s","object":"chat.completion.chunk","model":"gpt-5.4","choices":[{"index":0,"delta":{"content":"he"},"finish_reason":null}]}`, + "", + `data: {"id":"chatcmpl_s","object":"chat.completion.chunk","model":"gpt-5.4","choices":[{"index":0,"delta":{"content":"llo"},"finish_reason":null}]}`, + "", + `data: {"id":"chatcmpl_s","object":"chat.completion.chunk","model":"gpt-5.4","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}`, + "", + `data: {"id":"chatcmpl_s","object":"chat.completion.chunk","model":"gpt-5.4","choices":[],"usage":{"prompt_tokens":4,"completion_tokens":3,"total_tokens":7}}`, + "", + "data: [DONE]", + "", + }, "\n") + upstream := &httpUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_msg_chat_stream"}}, + Body: io.NopCloser(strings.NewReader(upstreamBody)), + }} + svc := &OpenAIGatewayService{ + cfg: rawChatCompletionsTestConfig(), + httpUpstream: upstream, + } + + result, err := svc.ForwardAsAnthropic(context.Background(), c, forceChatMessagesFallbackAccount(), body, "", "") + require.NoError(t, err) + require.NotNil(t, result) + require.True(t, gjson.GetBytes(upstream.lastBody, "stream_options.include_usage").Bool()) + + out := rec.Body.String() + require.Contains(t, out, "event: message_start") + require.Contains(t, out, `"text":"he"`) + require.Contains(t, out, `"text":"llo"`) + require.Contains(t, out, "event: content_block_stop") + require.Contains(t, out, `"stop_reason":"end_turn"`) + require.Contains(t, out, "event: message_stop") + + blockStop := strings.Index(out, "event: content_block_stop") + msgDelta := strings.Index(out, `"stop_reason":"end_turn"`) + msgStop := strings.Index(out, "event: message_stop") + require.Greater(t, msgDelta, blockStop, "content_block_stop must precede message_delta") + require.Greater(t, msgStop, msgDelta, "message_delta must precede message_stop") + + require.Equal(t, 4, result.Usage.InputTokens) + require.Equal(t, 3, result.Usage.OutputTokens) + require.True(t, result.Stream) + require.NotNil(t, result.FirstTokenMs) +} + +// Covers multi-chunk tool_call fragments aggregated by index and finalized as +// an Anthropic tool_use block with stop_reason=tool_use. +func TestForwardAsAnthropic_ForceChatCompletionsStreamingToolCallAggregation(t *testing.T) { + gin.SetMode(gin.TestMode) + + body := []byte(`{"model":"gpt-5.4","max_tokens":32,"messages":[{"role":"user","content":"weather in sf?"}],"stream":true}`) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", bytes.NewReader(body)) + c.Request.Header.Set("Content-Type", "application/json") + + upstreamBody := strings.Join([]string{ + `data: {"id":"chatcmpl_t","object":"chat.completion.chunk","model":"gpt-5.4","choices":[{"index":0,"delta":{"role":"assistant","tool_calls":[{"index":0,"id":"call_1","type":"function","function":{"name":"get_weather","arguments":""}}]},"finish_reason":null}]}`, + "", + `data: {"id":"chatcmpl_t","object":"chat.completion.chunk","model":"gpt-5.4","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"function":{"arguments":"{\"city\":"}}]},"finish_reason":null}]}`, + "", + `data: {"id":"chatcmpl_t","object":"chat.completion.chunk","model":"gpt-5.4","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"function":{"arguments":"\"sf\"}"}}]},"finish_reason":null}]}`, + "", + `data: {"id":"chatcmpl_t","object":"chat.completion.chunk","model":"gpt-5.4","choices":[{"index":0,"delta":{},"finish_reason":"tool_calls"}]}`, + "", + `data: {"id":"chatcmpl_t","object":"chat.completion.chunk","model":"gpt-5.4","choices":[],"usage":{"prompt_tokens":6,"completion_tokens":5,"total_tokens":11}}`, + "", + "data: [DONE]", + "", + }, "\n") + upstream := &httpUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_msg_chat_tool"}}, + Body: io.NopCloser(strings.NewReader(upstreamBody)), + }} + svc := &OpenAIGatewayService{ + cfg: rawChatCompletionsTestConfig(), + httpUpstream: upstream, + } + + result, err := svc.ForwardAsAnthropic(context.Background(), c, forceChatMessagesFallbackAccount(), body, "", "") + require.NoError(t, err) + require.NotNil(t, result) + + out := rec.Body.String() + require.Contains(t, out, `"type":"tool_use"`) + require.Contains(t, out, `"name":"get_weather"`) + require.Contains(t, out, `"input_json_delta"`) + require.Contains(t, out, `"stop_reason":"tool_use"`) + require.Contains(t, out, "event: message_stop") + require.Equal(t, 6, result.Usage.InputTokens) + require.Equal(t, 5, result.Usage.OutputTokens) +} + +// finish_reason=length must survive the double conversion (CC → Responses → +// Anthropic) as stop_reason=max_tokens. +func TestForwardAsAnthropic_ForceChatCompletionsStreamingLengthMapsToMaxTokens(t *testing.T) { + gin.SetMode(gin.TestMode) + + body := []byte(`{"model":"gpt-5.4","max_tokens":8,"messages":[{"role":"user","content":"hello"}],"stream":true}`) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", bytes.NewReader(body)) + c.Request.Header.Set("Content-Type", "application/json") + + upstreamBody := strings.Join([]string{ + `data: {"id":"chatcmpl_l","object":"chat.completion.chunk","model":"gpt-5.4","choices":[{"index":0,"delta":{"role":"assistant","content":"truncat"},"finish_reason":null}]}`, + "", + `data: {"id":"chatcmpl_l","object":"chat.completion.chunk","model":"gpt-5.4","choices":[{"index":0,"delta":{},"finish_reason":"length"}],"usage":{"prompt_tokens":4,"completion_tokens":8,"total_tokens":12}}`, + "", + "data: [DONE]", + "", + }, "\n") + upstream := &httpUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_msg_chat_len"}}, + Body: io.NopCloser(strings.NewReader(upstreamBody)), + }} + svc := &OpenAIGatewayService{ + cfg: rawChatCompletionsTestConfig(), + httpUpstream: upstream, + } + + result, err := svc.ForwardAsAnthropic(context.Background(), c, forceChatMessagesFallbackAccount(), body, "", "") + require.NoError(t, err) + require.NotNil(t, result) + + out := rec.Body.String() + require.Contains(t, out, `"stop_reason":"max_tokens"`) + require.Contains(t, out, "event: message_stop") +} + +// An upstream that ends immediately with [DONE] must still produce a fully +// framed (message_start → message_delta → message_stop) Anthropic stream. +func TestForwardAsAnthropic_ForceChatCompletionsEmptyStreamStillFramesMessage(t *testing.T) { + gin.SetMode(gin.TestMode) + + body := []byte(`{"model":"gpt-5.4","max_tokens":8,"messages":[{"role":"user","content":"hello"}],"stream":true}`) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", bytes.NewReader(body)) + c.Request.Header.Set("Content-Type", "application/json") + + upstream := &httpUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_msg_chat_empty"}}, + Body: io.NopCloser(strings.NewReader("data: [DONE]\n\n")), + }} + svc := &OpenAIGatewayService{ + cfg: rawChatCompletionsTestConfig(), + httpUpstream: upstream, + } + + result, err := svc.ForwardAsAnthropic(context.Background(), c, forceChatMessagesFallbackAccount(), body, "", "") + require.NoError(t, err) + require.NotNil(t, result) + + out := rec.Body.String() + require.Contains(t, out, "event: message_start") + require.Contains(t, out, "event: message_delta") + require.Contains(t, out, "event: message_stop") +} + +// Non-failover 4xx responses must go through the shared compat error handler: +// status-specific Anthropic error type, upstream message preserved, and ops +// upstream-error events recorded (previously this branch bypassed all three). +func TestForwardAsAnthropic_ForceChatCompletionsNonFailover400UsesSharedErrorHandler(t *testing.T) { + gin.SetMode(gin.TestMode) + + body := []byte(`{"model":"gpt-5.4","max_tokens":8,"messages":[{"role":"user","content":"hello"}],"stream":false}`) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", bytes.NewReader(body)) + c.Request.Header.Set("Content-Type", "application/json") + + upstream := &httpUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusBadRequest, + Header: http.Header{"Content-Type": []string{"application/json"}, "x-request-id": []string{"rid_msg_chat_400"}}, + Body: io.NopCloser(strings.NewReader(`{"error":{"message":"invalid roles","type":"invalid_request_error"}}`)), + }} + svc := &OpenAIGatewayService{ + cfg: rawChatCompletionsTestConfig(), + httpUpstream: upstream, + } + + result, err := svc.ForwardAsAnthropic(context.Background(), c, forceChatMessagesFallbackAccount(), body, "", "") + require.Error(t, err) + require.Nil(t, result) + + require.Equal(t, http.StatusBadRequest, rec.Code) + require.Equal(t, "error", gjson.Get(rec.Body.String(), "type").String()) + require.Equal(t, "invalid_request_error", gjson.Get(rec.Body.String(), "error.type").String()) + require.Equal(t, "invalid roles", gjson.Get(rec.Body.String(), "error.message").String()) + + statusVal, ok := c.Get(OpsUpstreamStatusCodeKey) + require.True(t, ok, "shared handler must record the upstream status for ops") + require.Equal(t, http.StatusBadRequest, statusVal) + + eventsVal, ok := c.Get(OpsUpstreamErrorsKey) + require.True(t, ok, "shared handler must append an ops upstream error event") + events, castOK := eventsVal.([]*OpsUpstreamErrorEvent) + require.True(t, castOK) + require.Len(t, events, 1) + require.Equal(t, http.StatusBadRequest, events[0].UpstreamStatusCode) + require.Equal(t, "http_error", events[0].Kind) + require.Equal(t, "invalid roles", events[0].Message) +} + +// A broken upstream read mid-stream must surface an error and must NOT emit a +// synthetic message_stop that would disguise the truncation as a completion. +func TestForwardAsAnthropic_ForceChatCompletionsStreamReadErrorSkipsFinalize(t *testing.T) { + gin.SetMode(gin.TestMode) + + body := []byte(`{"model":"gpt-5.4","max_tokens":8,"messages":[{"role":"user","content":"hello"}],"stream":true}`) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", bytes.NewReader(body)) + c.Request.Header.Set("Content-Type", "application/json") + + partial := strings.Join([]string{ + `data: {"id":"chatcmpl_e","object":"chat.completion.chunk","model":"gpt-5.4","choices":[{"index":0,"delta":{"role":"assistant"},"finish_reason":null}]}`, + "", + `data: {"id":"chatcmpl_e","object":"chat.completion.chunk","model":"gpt-5.4","choices":[{"index":0,"delta":{"content":"he"},"finish_reason":null}]}`, + "", + "", + }, "\n") + upstream := &httpUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_msg_chat_err"}}, + Body: &errTailReader{data: []byte(partial), err: errors.New("simulated upstream read failure")}, + }} + svc := &OpenAIGatewayService{ + cfg: rawChatCompletionsTestConfig(), + httpUpstream: upstream, + } + + result, err := svc.ForwardAsAnthropic(context.Background(), c, forceChatMessagesFallbackAccount(), body, "", "") + require.Error(t, err) + require.Contains(t, err.Error(), "stream usage incomplete") + require.NotNil(t, result) + require.True(t, result.Stream) + + out := rec.Body.String() + require.Contains(t, out, `"text":"he"`, "delta emitted before the failure must reach the client") + require.NotContains(t, out, "event: message_stop", "no synthetic completion after a broken read") +} + +// Gate regression: an API-key account whose upstream is confirmed to support +// the Responses API must keep using /v1/responses, never the CC fallback. +func TestForwardAsAnthropic_ResponsesSupportedAccountStillUsesResponsesEndpoint(t *testing.T) { + gin.SetMode(gin.TestMode) + + body := []byte(`{"model":"gpt-5.4","max_tokens":16,"messages":[{"role":"user","content":"hello"}],"stream":false}`) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", bytes.NewReader(body)) + c.Request.Header.Set("Content-Type", "application/json") + + upstreamBody := strings.Join([]string{ + `data: {"type":"response.completed","response":{"id":"resp_native","object":"response","model":"gpt-5.4","status":"completed","output":[{"type":"message","id":"msg_1","role":"assistant","status":"completed","content":[{"type":"output_text","text":"ok"}]}],"usage":{"input_tokens":5,"output_tokens":2,"total_tokens":7}}}`, + "", + "data: [DONE]", + "", + }, "\n") + upstream := &httpUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_msg_native"}}, + Body: io.NopCloser(strings.NewReader(upstreamBody)), + }} + svc := &OpenAIGatewayService{ + cfg: rawChatCompletionsTestConfig(), + httpUpstream: upstream, + } + account := rawChatCompletionsTestAccount() + account.Extra = map[string]any{ + openai_compat.ExtraKeyResponsesMode: string(openai_compat.ResponsesSupportModeAuto), + openai_compat.ExtraKeyResponsesSupported: true, + } + + result, err := svc.ForwardAsAnthropic(context.Background(), c, account, body, "", "") + require.NoError(t, err) + require.NotNil(t, result) + require.True(t, strings.HasSuffix(upstream.lastReq.URL.Path, "/responses"), + "responses-capable account must stay on /v1/responses, got %s", upstream.lastReq.URL.String()) + require.True(t, gjson.GetBytes(upstream.lastBody, "input").Exists()) + require.False(t, gjson.GetBytes(upstream.lastBody, "messages").Exists()) + require.Equal(t, "ok", gjson.Get(rec.Body.String(), "content.0.text").String()) +}