From 3c5553e253acfd01a147e0cca48cbf0cf36de38a Mon Sep 17 00:00:00 2001 From: li Date: Fri, 14 Aug 2026 11:05:29 +0800 Subject: [PATCH] =?UTF-8?q?fix(responses):=20=E7=BD=91=E5=85=B3=E5=90=88?= =?UTF-8?q?=E6=88=90=E7=9A=84=20Responses=20=E5=AF=B9=E8=B1=A1=E8=A1=A5?= =?UTF-8?q?=E9=BD=90=20created=5Fat=EF=BC=8C=E4=BF=AE=E5=A4=8D=E4=B8=A5?= =?UTF-8?q?=E6=A0=BC=E5=AE=A2=E6=88=B7=E7=AB=AF=E5=8F=8D=E5=BA=8F=E5=88=97?= =?UTF-8?q?=E5=8C=96=E5=A4=B1=E8=B4=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 严格的 Responses 客户端（Rust serde 系，如 Codex / Grok CLI）把 created_at 声明为 必填字段，缺失即 `missing field 'created_at'` 直接读不出响应。 apicompat.ResponsesResponse 此前根本没有这个字段，六个构造点全部不写： Chat→Responses 与 Anthropic→Responses 两座桥的非流式结果、response.created、 终止事件，以及两处手写的 response.failed 帧。 两个流式 state 早就在 New*State() 里采集了 Created 时间戳，只是没有出口—— 反方向（Responses→Chat）每个 chunk 都老老实实发 created，正方向直接丢在地上。 原生 Responses 透传走 gjson/sjson 字节级改写，保留上游原样，不受影响； 本次只补合成路径。 Fixes #5601 --- .../internal/handler/stream_error_event.go | 27 +-- .../handler/stream_error_event_test.go | 16 ++ .../anthropic_to_responses_response.go | 21 ++- .../chatcompletions_responses_bridge.go | 31 +++- .../responses_created_at_wire_test.go | 166 ++++++++++++++++++ backend/internal/pkg/apicompat/types.go | 17 +- .../service/openai_compact_stream_bridge.go | 8 +- ..._compact_stream_failure_created_at_test.go | 49 ++++++ 8 files changed, 299 insertions(+), 36 deletions(-) create mode 100644 backend/internal/pkg/apicompat/responses_created_at_wire_test.go create mode 100644 backend/internal/service/openai_compact_stream_failure_created_at_test.go diff --git a/backend/internal/handler/stream_error_event.go b/backend/internal/handler/stream_error_event.go index 2af098dbed..8ba6095cfb 100644 --- a/backend/internal/handler/stream_error_event.go +++ b/backend/internal/handler/stream_error_event.go @@ -5,6 +5,7 @@ import ( "fmt" "net/http" "strings" + "time" "github.com/Wei-Shaw/sub2api/internal/pkg/ctxkey" "github.com/gin-gonic/gin" @@ -19,13 +20,16 @@ type responsesFailedError struct { // responsesFailedBody 对齐 apicompat.makeResponsesCompletedEvent 输出的 response 子对象字段集。 // Output 用空 slice（不是 nil）确保 marshal 为 `[]` 而非 `null`。 +// CreatedAt 不带 omitempty：严格客户端把它当必填字段，缺失会以 +// `missing field 'created_at'` 反序列化失败——那正是本文件要避免的"客户端读不懂终止事件"。 type responsesFailedBody struct { - ID string `json:"id"` - Object string `json:"object"` - Model string `json:"model,omitempty"` - Status string `json:"status"` - Output []any `json:"output"` - Error responsesFailedError `json:"error"` + ID string `json:"id"` + Object string `json:"object"` + CreatedAt int64 `json:"created_at"` + Model string `json:"model,omitempty"` + Status string `json:"status"` + Output []any `json:"output"` + Error responsesFailedError `json:"error"` } // responsesFailedEvent 是写入 SSE data 行的顶层结构。 @@ -61,11 +65,12 @@ func writeResponsesFailedSSE(c *gin.Context, errType, message string) bool { payload, err := json.Marshal(responsesFailedEvent{ Type: "response.failed", Response: responsesFailedBody{ - ID: synthesizeResponseID(c), - Object: "response", - Model: requestModel(c), - Status: "failed", - Output: []any{}, + ID: synthesizeResponseID(c), + Object: "response", + CreatedAt: time.Now().Unix(), + Model: requestModel(c), + Status: "failed", + Output: []any{}, Error: responsesFailedError{ Code: mapResponsesErrorCode(errType), Message: message, diff --git a/backend/internal/handler/stream_error_event_test.go b/backend/internal/handler/stream_error_event_test.go index f24cf97fff..187deda19d 100644 --- a/backend/internal/handler/stream_error_event_test.go +++ b/backend/internal/handler/stream_error_event_test.go @@ -227,6 +227,22 @@ func TestOpenAIHandleStreamingAwareError_BareResponsesRouteEmitsResponseFailed(t } // Synthesized response.failed id falls back to uuid when no request_id is present. +// issue #5601：严格的 Responses 客户端把 created_at 当必填字段，缺失即 +// `missing field 'created_at'`。合成的终止事件若解析不了，本文件存在的意义 +// （给客户端一个可识别的终止事件而不是盲重连）就落空了。 +func TestOpenAIHandleStreamingAwareError_ResponsesStreamingCarriesCreatedAt(t *testing.T) { + c, w := newGinContextForEndpoint(t, EndpointResponses) + h := &OpenAIGatewayHandler{} + h.handleStreamingAwareError(c, http.StatusBadGateway, "upstream_error", "boom", true) + + resp, _ := parseResponsesFailedSSE(t, w.Body.String()) + raw, ok := resp["created_at"] + assert.True(t, ok, "response.failed 必须带 created_at") + createdAt, ok := raw.(float64) + assert.True(t, ok, "created_at 必须是数字，得到 %T", raw) + assert.Greater(t, int64(createdAt), int64(0), "created_at 必须是有效的 unix 时间戳") +} + func TestSynthesizeResponseID_FallbackUUID(t *testing.T) { c, _ := newGinContextForEndpoint(t, EndpointResponses) id := synthesizeResponseID(c) diff --git a/backend/internal/pkg/apicompat/anthropic_to_responses_response.go b/backend/internal/pkg/apicompat/anthropic_to_responses_response.go index db0343cd73..d5a7057036 100644 --- a/backend/internal/pkg/apicompat/anthropic_to_responses_response.go +++ b/backend/internal/pkg/apicompat/anthropic_to_responses_response.go @@ -21,10 +21,13 @@ func AnthropicToResponsesResponse(resp *AnthropicResponse) *ResponsesResponse { id = generateResponsesID() } + // Anthropic responses carry no creation timestamp, so stamp now — the same + // synthesize-what-the-client-requires rule the generated id above follows. out := &ResponsesResponse{ - ID: id, - Object: "response", - Model: resp.Model, + ID: id, + Object: "response", + CreatedAt: time.Now().Unix(), + Model: resp.Model, } var outputs []ResponsesOutput @@ -551,11 +554,12 @@ func makeResponsesCreatedEvent(state *AnthropicEventToResponsesState) ResponsesS Type: "response.created", SequenceNumber: seq, Response: &ResponsesResponse{ - ID: state.ResponseID, - Object: "response", - Model: state.Model, - Status: "in_progress", - Output: []ResponsesOutput{}, + ID: state.ResponseID, + Object: "response", + CreatedAt: state.Created, + Model: state.Model, + Status: "in_progress", + Output: []ResponsesOutput{}, }, } } @@ -602,6 +606,7 @@ func makeResponsesCompletedEvent( Response: &ResponsesResponse{ ID: state.ResponseID, Object: "response", + CreatedAt: state.Created, Model: state.Model, Status: status, Output: outputs, diff --git a/backend/internal/pkg/apicompat/chatcompletions_responses_bridge.go b/backend/internal/pkg/apicompat/chatcompletions_responses_bridge.go index 70dee035d8..c58caa1a56 100644 --- a/backend/internal/pkg/apicompat/chatcompletions_responses_bridge.go +++ b/backend/internal/pkg/apicompat/chatcompletions_responses_bridge.go @@ -1042,11 +1042,22 @@ func ChatCompletionsResponseToResponses(resp *ChatCompletionsResponse, model str id = generateResponsesID() } + // Carry the upstream's own creation timestamp when it sent one; otherwise + // stamp now, same fallback shape as the generated id above. + createdAt := int64(0) + if resp != nil { + createdAt = resp.Created + } + if createdAt <= 0 { + createdAt = time.Now().Unix() + } + out := &ResponsesResponse{ - ID: id, - Object: "response", - Model: model, - Status: "completed", + ID: id, + Object: "response", + CreatedAt: createdAt, + Model: model, + Status: "completed", } if resp == nil { out.Output = []ResponsesOutput{emptyResponsesMessageOutput()} @@ -1493,6 +1504,7 @@ func FinalizeChatCompletionsResponsesStream(state *ChatCompletionsToResponsesStr Response: &ResponsesResponse{ ID: state.ResponseID, Object: "response", + CreatedAt: state.Created, Model: state.Model, Status: status, Output: state.chatOutput(), @@ -1510,11 +1522,12 @@ func ensureChatToResponsesCreated(state *ChatCompletionsToResponsesStreamState) state.CreatedSent = true return []ResponsesStreamEvent{chatToResponsesEvent(state, "response.created", &ResponsesStreamEvent{ Response: &ResponsesResponse{ - ID: state.ResponseID, - Object: "response", - Model: state.Model, - Status: "in_progress", - Output: []ResponsesOutput{}, + ID: state.ResponseID, + Object: "response", + CreatedAt: state.Created, + Model: state.Model, + Status: "in_progress", + Output: []ResponsesOutput{}, }, })} } diff --git a/backend/internal/pkg/apicompat/responses_created_at_wire_test.go b/backend/internal/pkg/apicompat/responses_created_at_wire_test.go new file mode 100644 index 0000000000..48468409d6 --- /dev/null +++ b/backend/internal/pkg/apicompat/responses_created_at_wire_test.go @@ -0,0 +1,166 @@ +package apicompat + +import ( + "encoding/json" + "testing" + + "github.com/stretchr/testify/require" +) + +// issue #5601：严格的 Responses 客户端（Rust serde 系，如 Codex / Grok CLI）把 +// created_at 声明为必填字段，缺失即 `missing field 'created_at'` 反序列化失败。 +// 网关合成的 Responses 对象（Chat→Responses、Anthropic→Responses 两座桥）此前从不 +// 写这个字段——尽管两个流式 state 早就采集好了 Created 时间戳，只是没有出口。 +// 原生 Responses 透传走 gjson/sjson 字节级改写，不受影响。 + +// responseObjectOf 取出事件里的 response 子对象（按线格式，而不是按 Go 结构体）。 +func responseObjectOf(t *testing.T, evt ResponsesStreamEvent) map[string]any { + t.Helper() + m := marshalEvent(t, evt) + resp, ok := m["response"].(map[string]any) + require.True(t, ok, "event must carry a response object: %v", m) + return resp +} + +func requireCreatedAt(t *testing.T, resp map[string]any) int64 { + t.Helper() + raw, ok := resp["created_at"] + require.True(t, ok, "response 对象必须带 created_at，否则严格客户端直接反序列化失败") + value, ok := raw.(float64) + require.True(t, ok, "created_at 必须是数字，得到 %T", raw) + require.Greater(t, int64(value), int64(0), "created_at 必须是有效的 unix 时间戳") + return int64(value) +} + +// omitempty 陷阱守卫：created_at 为 0 时也必须出现在线格式里， +// 否则「字段存在」这件事就依赖于运行时恰好非零。 +func TestWire_CreatedAtPresentEvenAtZero(t *testing.T) { + resp := responseObjectOf(t, ResponsesStreamEvent{ + Type: "response.created", + Response: &ResponsesResponse{ID: "resp_1", Object: "response", Status: "in_progress"}, + }) + require.Contains(t, resp, "created_at", "created_at 不得带 omitempty") + require.EqualValues(t, 0, resp["created_at"]) +} + +// --------------------------------------------------------------------------- +// Chat Completions → Responses +// --------------------------------------------------------------------------- + +func TestChatCompletionsResponseToResponses_CarriesCreatedAt(t *testing.T) { + t.Run("uses_upstream_created_when_present", func(t *testing.T) { + out := ChatCompletionsResponseToResponses(&ChatCompletionsResponse{ + ID: "chatcmpl_1", + Created: 1700000000, + Model: "deepseek-v4-flash", + Choices: []ChatChoice{{Message: ChatMessage{Role: "assistant", Content: json.RawMessage(`"hi"`)}}}, + }, "deepseek-v4-flash", nil, false, nil) + require.EqualValues(t, 1700000000, out.CreatedAt, "上游给了 created 就照搬，不要另起时间") + }) + + t.Run("stamps_now_when_upstream_omits_created", func(t *testing.T) { + out := ChatCompletionsResponseToResponses(&ChatCompletionsResponse{ + ID: "chatcmpl_2", + Model: "deepseek-v4-flash", + Choices: []ChatChoice{{Message: ChatMessage{Role: "assistant", Content: json.RawMessage(`"hi"`)}}}, + }, "deepseek-v4-flash", nil, false, nil) + require.Greater(t, out.CreatedAt, int64(0)) + }) + + t.Run("nil_upstream_response_still_stamps", func(t *testing.T) { + out := ChatCompletionsResponseToResponses(nil, "deepseek-v4-flash", nil, false, nil) + require.Greater(t, out.CreatedAt, int64(0), "空上游响应也必须产出可解析的对象") + }) +} + +// 同一条流里 response.created 与终止事件必须报同一个 created_at +// （官方语义：created_at 是这次 response 的创建时刻，不随事件变化）。 +func TestChatCompletionsToResponsesStream_CreatedAtStableAcrossEvents(t *testing.T) { + state := NewChatCompletionsToResponsesStreamState("deepseek-v4-flash") + require.Greater(t, state.Created, int64(0), "前提：state 早就采集了时间戳") + + var chunk ChatCompletionsChunk + require.NoError(t, json.Unmarshal( + []byte(`{"choices":[{"index":0,"delta":{"content":"hi"}}]}`), &chunk)) + + events := ChatCompletionsChunkToResponsesEvents(&chunk, state) + events = append(events, FinalizeChatCompletionsResponsesStream(state)...) + + seen := map[string]int64{} + for _, evt := range events { + if evt.Response == nil { + continue + } + seen[evt.Type] = requireCreatedAt(t, responseObjectOf(t, evt)) + } + + require.Contains(t, seen, "response.created") + require.Contains(t, seen, "response.completed") + require.Equal(t, state.Created, seen["response.created"]) + require.Equal(t, seen["response.created"], seen["response.completed"], + "同一条流的 created_at 必须恒定") +} + +// --------------------------------------------------------------------------- +// Anthropic → Responses +// --------------------------------------------------------------------------- + +func TestAnthropicToResponsesResponse_StampsCreatedAt(t *testing.T) { + out := AnthropicToResponsesResponse(&AnthropicResponse{ + ID: "msg_1", + Type: "message", + Role: "assistant", + Model: "claude-sonnet-4-20250514", + Content: []AnthropicContentBlock{{Type: "text", Text: "hi"}}, + }) + require.Greater(t, out.CreatedAt, int64(0), + "Anthropic 响应不带时间戳，网关必须自己盖一个") +} + +func TestAnthropicEventToResponsesStream_CreatedAtStableAcrossEvents(t *testing.T) { + state := NewAnthropicEventToResponsesState() + state.Model = "claude-sonnet-4-20250514" + require.Greater(t, state.Created, int64(0), "前提：state 早就采集了时间戳") + + var events []ResponsesStreamEvent + for _, raw := range []string{ + `{"type":"message_start","message":{"id":"msg_1","type":"message","role":"assistant","model":"claude-sonnet-4-20250514","content":[],"usage":{"input_tokens":3,"output_tokens":0}}}`, + `{"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}`, + `{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"hi"}}`, + `{"type":"content_block_stop","index":0}`, + `{"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":2}}`, + `{"type":"message_stop"}`, + } { + var evt AnthropicStreamEvent + require.NoError(t, json.Unmarshal([]byte(raw), &evt)) + events = append(events, AnthropicEventToResponsesEvents(&evt, state)...) + } + events = append(events, FinalizeAnthropicResponsesStream(state)...) + + seen := map[string]int64{} + for _, evt := range events { + if evt.Response == nil { + continue + } + seen[evt.Type] = requireCreatedAt(t, responseObjectOf(t, evt)) + } + + require.Contains(t, seen, "response.created") + require.Contains(t, seen, "response.completed") + require.Equal(t, state.Created, seen["response.created"]) + require.Equal(t, seen["response.created"], seen["response.completed"], + "同一条流的 created_at 必须恒定") +} + +// ResponsesClientToolStreamRestorer 对部分事件走 unmarshal→re-marshal。 +// 结构体没有该字段时，上游带来的 created_at 会在这一步被静默抹掉。 +func TestResponsesStreamEvent_CreatedAtSurvivesUnmarshalRemarshal(t *testing.T) { + upstream := []byte(`{"type":"response.completed","response":{"id":"resp_9","object":"response",` + + `"created_at":1700000123,"model":"gpt-5.5","status":"completed","output":[]}}`) + + var evt ResponsesStreamEvent + require.NoError(t, json.Unmarshal(upstream, &evt)) + require.EqualValues(t, 1700000123, evt.Response.CreatedAt) + + require.EqualValues(t, 1700000123, requireCreatedAt(t, responseObjectOf(t, evt))) +} diff --git a/backend/internal/pkg/apicompat/types.go b/backend/internal/pkg/apicompat/types.go index bcc8cd481e..84175f2a89 100644 --- a/backend/internal/pkg/apicompat/types.go +++ b/backend/internal/pkg/apicompat/types.go @@ -339,12 +339,17 @@ func (t *ResponsesTool) UnmarshalJSON(data []byte) error { // ResponsesResponse is the non-streaming response from POST /v1/responses. type ResponsesResponse struct { - ID string `json:"id"` - Object string `json:"object"` // "response" - Model string `json:"model"` - Status string `json:"status"` // "completed" | "incomplete" | "failed" - Output []ResponsesOutput `json:"output"` - Usage *ResponsesUsage `json:"usage,omitempty"` + ID string `json:"id"` + Object string `json:"object"` // "response" + // CreatedAt is the unix creation timestamp. Strict Responses clients declare + // it non-optional and abort with `missing field 'created_at'` when it is + // absent, so it is always emitted — no omitempty. Same rule as ID (see the + // "clients treat it as required" fallback in ChatCompletionsResponseToAnthropic). + CreatedAt int64 `json:"created_at"` + Model string `json:"model"` + Status string `json:"status"` // "completed" | "incomplete" | "failed" + Output []ResponsesOutput `json:"output"` + Usage *ResponsesUsage `json:"usage,omitempty"` // incomplete_details is present when status="incomplete" IncompleteDetails *ResponsesIncompleteDetails `json:"incomplete_details,omitempty"` diff --git a/backend/internal/service/openai_compact_stream_bridge.go b/backend/internal/service/openai_compact_stream_bridge.go index bcac8304c2..1c36f9f27c 100644 --- a/backend/internal/service/openai_compact_stream_bridge.go +++ b/backend/internal/service/openai_compact_stream_bridge.go @@ -6,6 +6,7 @@ import ( "net/http" "strconv" "strings" + "time" "github.com/gin-gonic/gin" "github.com/google/uuid" @@ -117,8 +118,11 @@ func writeOpenAICompactSSEFailureMessage(c *gin.Context, statusCode int, errType "response": map[string]any{ "id": "resp_" + strings.ReplaceAll(uuid.NewString(), "-", ""), "object": "response", - "status": "failed", - "output": []any{}, + // 严格客户端把 created_at 当必填字段，缺失会反序列化失败， + // 终止事件就白发了（退化成盲重连）。与 writeResponsesFailedSSE 对齐。 + "created_at": time.Now().Unix(), + "status": "failed", + "output": []any{}, "error": map[string]any{ "code": errType, "message": message, diff --git a/backend/internal/service/openai_compact_stream_failure_created_at_test.go b/backend/internal/service/openai_compact_stream_failure_created_at_test.go new file mode 100644 index 0000000000..a3b6f9bd78 --- /dev/null +++ b/backend/internal/service/openai_compact_stream_failure_created_at_test.go @@ -0,0 +1,49 @@ +//go:build unit + +package service + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" +) + +// issue #5601：严格的 Responses 客户端把 created_at 当必填字段，缺失即 +// `missing field 'created_at'`。writeOpenAICompactSSEFailureMessage 存在的理由就是 +// 让 Codex 能把这帧识别成合法终止事件；解析不了就退化回它想避免的盲重连。 +func TestWriteOpenAICompactSSEFailureMessage_CarriesCreatedAt(t *testing.T) { + gin.SetMode(gin.TestMode) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil) + + writeOpenAICompactSSEFailureMessage(c, http.StatusBadGateway, "upstream_error", "boom") + + body := rec.Body.String() + require.Contains(t, body, "event: response.failed") + + _, payload, found := strings.Cut(body, "data: ") + require.True(t, found, "SSE 帧必须带 data 行: %q", body) + + var event struct { + Type string `json:"type"` + Response struct { + ID string `json:"id"` + Object string `json:"object"` + CreatedAt int64 `json:"created_at"` + Status string `json:"status"` + } `json:"response"` + } + require.NoError(t, json.Unmarshal([]byte(strings.TrimSpace(payload)), &event)) + + require.Equal(t, "response.failed", event.Type) + require.Equal(t, "response", event.Response.Object) + require.Equal(t, "failed", event.Response.Status) + require.Greater(t, event.Response.CreatedAt, int64(0), + "response.failed 必须带有效的 created_at，否则严格客户端读不出这帧") +}