Merge pull request #7129 from kk-418/fix/grok-build-responses-sequence-number

fix(responses): 始终在 Responses SSE 中写出 sequence_number 避免反序列化失败

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 形式回报错误；

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)

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(`{

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"`
 }
 
 // ---------------------------------------------------------------------------

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
 }
 

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)

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),

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
 }