From 087987070b5a1413d698dbd9677986ba41a964e7 Mon Sep 17 00:00:00 2001 From: cikenerd Date: Mon, 14 Sep 2026 11:45:53 +0800 Subject: [PATCH] =?UTF-8?q?fix(responses):=20=E4=B8=BA=20grok-build=20?= =?UTF-8?q?=E5=A7=8B=E7=BB=88=E5=86=99=E5=87=BA=20sequence=5Fnumber?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 网关合成/重编的 Responses SSE 因 omitempty 或缺字段丢掉 sequence_number， grok-build 会整轮反序列化失败。合成失败帧、compact 桥接和转换 wire JSON 一律写出该字段，0 也保留。 Fixes #7128 Co-authored-by: Cursor --- .../internal/handler/stream_error_event.go | 12 +++++---- .../handler/stream_error_event_test.go | 25 ++++++++++++++++--- .../responses_stream_event_wire_test.go | 25 +++++++++++++++++++ backend/internal/pkg/apicompat/types.go | 7 ++++-- .../service/openai_compact_stream_bridge.go | 25 ++++++++++++++----- .../openai_compact_stream_bridge_test.go | 4 +++ ..._compact_stream_failure_created_at_test.go | 7 ++++-- .../internal/service/openai_ws_http_bridge.go | 11 ++++---- 8 files changed, 92 insertions(+), 24 deletions(-) diff --git a/backend/internal/handler/stream_error_event.go b/backend/internal/handler/stream_error_event.go index f60e6c7834..04cafd9bdd 100644 --- a/backend/internal/handler/stream_error_event.go +++ b/backend/internal/handler/stream_error_event.go @@ -33,10 +33,13 @@ type responsesFailedBody struct { } // responsesFailedEvent 是写入 SSE data 行的顶层结构。 -// 故意不带 sequence_number：spec 标记可选，且本函数被调用时无法可靠拿到 last seq。 +// SequenceNumber 不带 omitempty：grok-build 等 Rust serde 客户端把它当必填字段， +// 缺失即 `missing field 'sequence_number'`。拿不到上一帧序号时写 0——这是终止事件， +// 后面不会再续流。 type responsesFailedEvent struct { - Type string `json:"type"` - Response responsesFailedBody `json:"response"` + Type string `json:"type"` + SequenceNumber int `json:"sequence_number"` + Response responsesFailedBody `json:"response"` } // writeResponsesFailedSSE emits a `response.failed` SSE event in the OpenAI @@ -49,8 +52,7 @@ type responsesFailedEvent struct { // 而抛出 "stream closed before response.completed"。 // // 字段集对齐 apicompat.makeResponsesCompletedEvent：id/object/model/status/output/error。 -// 故意不写 sequence_number：本函数被调用时无法可靠拿到当前流的 last sequence， -// 而 OpenAI spec 将 sequence_number 设为可选；省略避免破坏单调性约束。 +// sequence_number 始终写出（未知时为 0）：OpenAI spec 标可选，但 grok-build 当必填。 // // 返回 true 表示已尝试 SSE 写出（不论 Write 是否成功，caller 都应直接 return）。 // 返回 false 表示 writer 不支持 Flusher，无法以 SSE 形式回报错误； diff --git a/backend/internal/handler/stream_error_event_test.go b/backend/internal/handler/stream_error_event_test.go index 94089e9fae..9ddf03fc57 100644 --- a/backend/internal/handler/stream_error_event_test.go +++ b/backend/internal/handler/stream_error_event_test.go @@ -48,9 +48,12 @@ func parseResponsesFailedSSE(t *testing.T, body string) (map[string]any, map[str require.NoError(t, json.Unmarshal([]byte(jsonStr), &parsed), "data must be valid JSON: %s", jsonStr) assert.Equal(t, "response.failed", parsed["type"]) - // 故意不发 sequence_number，避免与后续真实事件的序号冲突。 - _, hasSeq := parsed["sequence_number"] - assert.False(t, hasSeq, "synthetic event must not emit sequence_number") + // grok-build 把 sequence_number 当必填；合成终止事件未知上一帧时写 0。 + rawSeq, hasSeq := parsed["sequence_number"] + assert.True(t, hasSeq, "synthetic event must emit sequence_number") + seq, ok := rawSeq.(float64) + assert.True(t, ok, "sequence_number must be a number, got %T", rawSeq) + assert.GreaterOrEqual(t, seq, float64(0)) resp, ok := parsed["response"].(map[string]any) require.True(t, ok, "response object missing") @@ -316,7 +319,20 @@ func TestOpenAIHandleStreamingAwareError_BareResponsesRouteEmitsResponseFailed(t assert.Equal(t, "rate_limit_exceeded", errObj["code"]) } -// Synthesized response.failed id falls back to uuid when no request_id is present. +// issue #7128：grok-build 把顶层 sequence_number 当必填，合成的 response.failed +// 必须写出该字段，否则整轮反序列化失败。 +func TestOpenAIHandleStreamingAwareError_ResponsesStreamingCarriesSequenceNumber(t *testing.T) { + c, w := newGinContextForEndpoint(t, EndpointResponses) + h := &OpenAIGatewayHandler{} + h.handleStreamingAwareError(c, http.StatusBadGateway, "upstream_error", "boom", true) + + require.True(t, strings.HasPrefix(w.Body.String(), "event: response.failed\n")) + _, payload, found := strings.Cut(w.Body.String(), "data: ") + require.True(t, found) + require.True(t, gjson.Get(payload, "sequence_number").Exists(), "response.failed 必须带 sequence_number") + require.GreaterOrEqual(t, gjson.Get(payload, "sequence_number").Int(), int64(0)) +} + // issue #5601：严格的 Responses 客户端把 created_at 当必填字段，缺失即 // `missing field 'created_at'`。合成的终止事件若解析不了，本文件存在的意义 // （给客户端一个可识别的终止事件而不是盲重连）就落空了。 @@ -333,6 +349,7 @@ func TestOpenAIHandleStreamingAwareError_ResponsesStreamingCarriesCreatedAt(t *t assert.Greater(t, int64(createdAt), int64(0), "created_at 必须是有效的 unix 时间戳") } +// Synthesized response.failed id falls back to uuid when no request_id is present. func TestSynthesizeResponseID_FallbackUUID(t *testing.T) { c, _ := newGinContextForEndpoint(t, EndpointResponses) id := synthesizeResponseID(c) diff --git a/backend/internal/pkg/apicompat/responses_stream_event_wire_test.go b/backend/internal/pkg/apicompat/responses_stream_event_wire_test.go index fb138a1469..3d15da2afc 100644 --- a/backend/internal/pkg/apicompat/responses_stream_event_wire_test.go +++ b/backend/internal/pkg/apicompat/responses_stream_event_wire_test.go @@ -132,6 +132,31 @@ func TestWire_UnknownEventFallsBackToDefault(t *testing.T) { require.Contains(t, m, "response") } +// grok-build 把 sequence_number 当必填。response.created 从 0 起号， +// omitempty 会把 0 整段丢掉，第一帧就反序列化失败。 +func TestWire_SequenceNumberPresentAtZero(t *testing.T) { + created := marshalEvent(t, ResponsesStreamEvent{ + Type: "response.created", + Response: &ResponsesResponse{ID: "resp_1", Object: "response", Status: "in_progress"}, + }) + require.Contains(t, created, "sequence_number") + require.EqualValues(t, 0, created["sequence_number"]) + + completed := marshalEvent(t, ResponsesStreamEvent{ + Type: "response.completed", + SequenceNumber: 0, + Response: &ResponsesResponse{ID: "resp_1", Object: "response", Status: "completed"}, + }) + require.Contains(t, completed, "sequence_number") + require.EqualValues(t, 0, completed["sequence_number"]) + + delta := marshalEvent(t, ResponsesStreamEvent{ + Type: "response.output_text.delta", OutputIndex: 0, ContentIndex: 0, ItemID: "msg_1", Delta: "hi", + }) + require.Contains(t, delta, "sequence_number") + require.EqualValues(t, 0, delta["sequence_number"]) +} + func TestResponsesOutputUnmarshal_ToolSearchObjectArguments(t *testing.T) { var item ResponsesOutput require.NoError(t, json.Unmarshal([]byte(`{ diff --git a/backend/internal/pkg/apicompat/types.go b/backend/internal/pkg/apicompat/types.go index 3a9d0a8ff2..5d89769b15 100644 --- a/backend/internal/pkg/apicompat/types.go +++ b/backend/internal/pkg/apicompat/types.go @@ -642,8 +642,11 @@ type ResponsesStreamEvent struct { Code string `json:"code,omitempty"` Param string `json:"param,omitempty"` - // Sequence number for ordering events - SequenceNumber int `json:"sequence_number,omitempty"` + // SequenceNumber orders streamed events. Strict Responses clients (Grok Build, + // Codex CLI) declare it required and abort with `missing field 'sequence_number'` + // when it is absent, so it is always emitted — no omitempty. Same rule as + // ResponsesResponse.CreatedAt. Zero is a valid first-event value. + SequenceNumber int `json:"sequence_number"` } // --------------------------------------------------------------------------- diff --git a/backend/internal/service/openai_compact_stream_bridge.go b/backend/internal/service/openai_compact_stream_bridge.go index 1c36f9f27c..5ebab8c5fe 100644 --- a/backend/internal/service/openai_compact_stream_bridge.go +++ b/backend/internal/service/openai_compact_stream_bridge.go @@ -114,7 +114,8 @@ func writeOpenAICompactSSEFailureMessage(c *gin.Context, statusCode int, errType } MarkOpsStreamError(c, errType, message, statusCode) payload, err := json.Marshal(map[string]any{ - "type": "response.failed", + "type": "response.failed", + "sequence_number": 0, "response": map[string]any{ "id": "resp_" + strings.ReplaceAll(uuid.NewString(), "-", ""), "object": "response", @@ -143,7 +144,8 @@ func writeOpenAICompactSSEFailureMessage(c *gin.Context, statusCode int, errType // 携带完整 response 对象。Codex 的 SSE 解析只从 output_item.done 收集 item， // 并要求 response.completed 的 response.id 必填、usage（若存在）必须携带 // input_tokens/output_tokens/total_tokens 整数字段，否则整条 completed 事件 -// 解析失败，故此处做兜底修补。 +// 解析失败，故此处做兜底修补。每帧还要带单调的 sequence_number：grok-build +// 把它当必填，缺了整轮反序列化失败。 func buildOpenAICompactSSEPayload(finalResponse []byte) ([]byte, bool) { if len(finalResponse) == 0 || !gjson.ValidBytes(finalResponse) { return nil, false @@ -176,12 +178,19 @@ func buildOpenAICompactSSEPayload(finalResponse []byte) ([]byte, bool) { var buf bytes.Buffer outputIndex := 0 - appendEvent := func(eventType string, data []byte) { + sequenceNumber := 0 + appendEvent := func(eventType string, data []byte) bool { + numbered, err := sjson.SetBytes(data, "sequence_number", sequenceNumber) + if err != nil { + return false + } + sequenceNumber++ _, _ = buf.WriteString("event: ") _, _ = buf.WriteString(eventType) _, _ = buf.WriteString("\ndata: ") - _, _ = buf.Write(data) + _, _ = buf.Write(numbered) _, _ = buf.WriteString("\n\n") + return true } for _, item := range gjson.GetBytes(response, "output").Array() { if !item.IsObject() { @@ -195,7 +204,9 @@ func buildOpenAICompactSSEPayload(finalResponse []byte) ([]byte, bool) { if err != nil { return nil, false } - appendEvent("response.output_item.done", event) + if !appendEvent("response.output_item.done", event) { + return nil, false + } outputIndex++ } @@ -203,7 +214,9 @@ func buildOpenAICompactSSEPayload(finalResponse []byte) ([]byte, bool) { if err != nil { return nil, false } - appendEvent("response.completed", completed) + if !appendEvent("response.completed", completed) { + return nil, false + } return buf.Bytes(), true } diff --git a/backend/internal/service/openai_compact_stream_bridge_test.go b/backend/internal/service/openai_compact_stream_bridge_test.go index 3a095a6367..61c9febb8c 100644 --- a/backend/internal/service/openai_compact_stream_bridge_test.go +++ b/backend/internal/service/openai_compact_stream_bridge_test.go @@ -73,6 +73,8 @@ func TestBuildOpenAICompactSSEPayload_EmitsItemsAndCompleted(t *testing.T) { require.Equal(t, "response.output_item.done", events[0][0]) first := events[0][1] require.Equal(t, "response.output_item.done", gjson.Get(first, "type").String()) + require.True(t, gjson.Get(first, "sequence_number").Exists(), "compact 合成帧必须带 sequence_number") + require.Equal(t, int64(0), gjson.Get(first, "sequence_number").Int()) require.Equal(t, int64(0), gjson.Get(first, "output_index").Int()) require.Equal(t, "compaction", gjson.Get(first, "item.type").String()) require.Equal(t, "cmp_1", gjson.Get(first, "item.id").String()) @@ -81,12 +83,14 @@ func TestBuildOpenAICompactSSEPayload_EmitsItemsAndCompleted(t *testing.T) { require.True(t, gjson.Get(first, "item.opaque.kept").Bool(), "item 原始字段必须逐字节保留") require.Equal(t, "response.output_item.done", events[1][0]) + require.Equal(t, int64(1), gjson.Get(events[1][1], "sequence_number").Int()) require.Equal(t, int64(1), gjson.Get(events[1][1], "output_index").Int()) require.Equal(t, "message", gjson.Get(events[1][1], "item.type").String()) require.Equal(t, "response.completed", events[2][0]) completed := events[2][1] require.Equal(t, "response.completed", gjson.Get(completed, "type").String()) + require.Equal(t, int64(2), gjson.Get(completed, "sequence_number").Int()) require.Equal(t, "resp_compact_1", gjson.Get(completed, "response.id").String()) require.Equal(t, int64(13), gjson.Get(completed, "response.usage.total_tokens").Int()) require.Len(t, gjson.Get(completed, "response.output").Array(), 2) 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 index a3b6f9bd78..39d12ac23e 100644 --- a/backend/internal/service/openai_compact_stream_failure_created_at_test.go +++ b/backend/internal/service/openai_compact_stream_failure_created_at_test.go @@ -31,8 +31,9 @@ func TestWriteOpenAICompactSSEFailureMessage_CarriesCreatedAt(t *testing.T) { require.True(t, found, "SSE 帧必须带 data 行: %q", body) var event struct { - Type string `json:"type"` - Response struct { + Type string `json:"type"` + SequenceNumber *int `json:"sequence_number"` + Response struct { ID string `json:"id"` Object string `json:"object"` CreatedAt int64 `json:"created_at"` @@ -42,6 +43,8 @@ func TestWriteOpenAICompactSSEFailureMessage_CarriesCreatedAt(t *testing.T) { require.NoError(t, json.Unmarshal([]byte(strings.TrimSpace(payload)), &event)) require.Equal(t, "response.failed", event.Type) + require.NotNil(t, event.SequenceNumber, "response.failed 必须带 sequence_number，否则 grok-build 读不出这帧") + require.GreaterOrEqual(t, *event.SequenceNumber, 0) require.Equal(t, "response", event.Response.Object) require.Equal(t, "failed", event.Response.Status) require.Greater(t, event.Response.CreatedAt, int64(0), diff --git a/backend/internal/service/openai_ws_http_bridge.go b/backend/internal/service/openai_ws_http_bridge.go index 52420c584a..780e58c86d 100644 --- a/backend/internal/service/openai_ws_http_bridge.go +++ b/backend/internal/service/openai_ws_http_bridge.go @@ -354,8 +354,9 @@ func buildOpenAIWSHTTPBridgeErrorEvent(statusCode int, message string) []byte { message = "upstream request failed" } event := map[string]any{ - "type": "error", - "status": statusCode, + "type": "error", + "sequence_number": 0, + "status": statusCode, "error": map[string]any{ "type": "upstream_error", "message": message, @@ -363,7 +364,7 @@ func buildOpenAIWSHTTPBridgeErrorEvent(statusCode int, message string) []byte { } body, err := json.Marshal(event) if err != nil { - return []byte(`{"type":"error","error":{"type":"upstream_error","message":"upstream request failed"}}`) + return []byte(`{"type":"error","sequence_number":0,"error":{"type":"upstream_error","message":"upstream request failed"}}`) } return body } @@ -398,9 +399,9 @@ func buildOpenAIWSHTTPBridgeFailedEvent(responseID, model string, source []byte, if model = strings.TrimSpace(model); model != "" { response["model"] = model } - body, err := json.Marshal(map[string]any{"type": "response.failed", "response": response}) + body, err := json.Marshal(map[string]any{"type": "response.failed", "sequence_number": 0, "response": response}) if err != nil { - return []byte(`{"type":"response.failed","response":{"status":"failed","output":[],"error":{"code":"upstream_error","message":"Upstream response failed"}}}`) + return []byte(`{"type":"response.failed","sequence_number":0,"response":{"status":"failed","output":[],"error":{"code":"upstream_error","message":"Upstream response failed"}}}`) } return body }