opensource/sub2api
3
0
Fork 0
mirror of https://github.com/Wei-Shaw/sub2api.git synced 2026-09-24 16:05:44 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(messages): 修复 /v1/messages CC 回退的错误处理旁路并补齐流式测试

Browse Source 
针对 #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 账号门控回归、非流式转换
This commit is contained in:
shaw
2026-07-07 20:59:41 +08:00
parent 4bcb13f77f
commit dad92488e5
2 changed files with 417 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/internal/service/openai_gateway_messages_chat_fallback.go
+28 -16
View File
@@ -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
}
}
backend/internal/service/openai_gateway_messages_chat_fallback_test.go
+389
View File
@@ -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())
}