fix(compact): body-signal 客户端流式请求的响应合成回 SSE

v0.1.147 的 body-signal 提升（#3804）把 Codex remote compact v2 的流式
/responses 请求改写为上游 unary /responses/compact（JSON），但把 JSON
文档原样写回给了按 Responses SSE 协议消费的客户端。Codex 只从
response.output_item.done 收集 item 且必须收到 response.completed，
收到 JSON 后报 "stream disconnected before completion: stream closed
before response.completed" 无限重连，每次重连再白烧一次上游 compact
配额（#3875，v0.1.146 无提升逻辑故正常）。

修复保留提升的全部收益（requireCompact 调度过滤、compact 模型映射、
白名单归一化、上游 unary JSON），补上协议转换缺失的响应半程：

- handler 在 body-signal 提升时记录原始 body 的 stream:true 意图
  （MarkOpenAICompactClientStream）；path-based 请求不标记，Codex v1
  unary 协议与链式 sub2api 的 JSON 写回行为保持不变。
- 四个非流式写回点（handleNonStreamingResponse / handleSSEToJSON /
  handleNonStreamingResponsePassthrough / handlePassthroughSSEToJSON）
  对已标记的 2xx JSON 响应经 writeOpenAICompactSSEBridge 合成最小
  Responses SSE：每个 output[] item 一条 response.output_item.done
  （原始字段逐字节保留），最后一条 response.completed 携带完整
  response 对象。
- 按 codex-rs 解析器硬约束兜底：先 json.Compact 防 pretty JSON 换行
  破坏 SSE 帧；response.id 缺失时注入 resp_*（必填 string）；usage
  缺 input_tokens/output_tokens/total_tokens 整数字段时整体删除，
  避免整条 completed 事件解析失败。
- 非 2xx 一律保持 JSON 原样，Codex 依赖 HTTP 状态码走重试链路。

新增 9 个 service 测试 + 4 个 handler 测试覆盖合成形态、id/usage 兜底、
path-based 不受影响、主链路/透传链路/上游 SSE 提取再合成三条端到端。

Fixes #3875
Refs #3777

Author: Wei-Shaw

backend/internal/handler/openai_gateway_compact_body_signal_test.go

@@ -117,3 +117,46 @@ func TestNormalizeOpenAIResponsesCompactRequest_SubpathNotPromoted(t *testing.T)
 	require.Equal(t, "/v1/responses/resp_123/cancel", c.Request.URL.Path)
 	require.Equal(t, body, normalized)
 }
+
+// 回归 #3875：body-signal 原始请求 stream:true 时必须标记 client-stream，
+// 供响应写回阶段把上游 unary JSON 合成回 Codex remote compact v2 所需的 SSE。
+func TestNormalizeOpenAIResponsesCompactRequest_BodySignalStreamTrueMarksClientStream(t *testing.T) {
+	h := &OpenAIGatewayHandler{}
+	body := []byte(`{"model":"gpt-5.5","stream":true,"input":[{"type":"compaction_trigger"}]}`)
+	c := newCompactBodySignalTestContext(t, "/v1/responses", body)
+
+	_, ok := h.normalizeOpenAIResponsesCompactRequest(c, zap.NewNop(), body)
+	require.True(t, ok)
+
+	marked, exists := c.Get(service.OpenAICompactClientStreamKeyForTest())
+	require.True(t, exists)
+	require.Equal(t, true, marked)
+}
+
+func TestNormalizeOpenAIResponsesCompactRequest_BodySignalStreamFalseNotMarked(t *testing.T) {
+	h := &OpenAIGatewayHandler{}
+	for name, body := range map[string][]byte{
+		"stream_false":  []byte(`{"model":"gpt-5.5","stream":false,"input":[{"type":"compaction_trigger"}]}`),
+		"stream_absent": []byte(`{"model":"gpt-5.5","input":[{"type":"compaction_trigger"}]}`),
+	} {
+		c := newCompactBodySignalTestContext(t, "/v1/responses", body)
+		_, ok := h.normalizeOpenAIResponsesCompactRequest(c, zap.NewNop(), body)
+		require.True(t, ok, name)
+		require.Equal(t, "/v1/responses/compact", c.Request.URL.Path, name)
+		_, exists := c.Get(service.OpenAICompactClientStreamKeyForTest())
+		require.False(t, exists, "case %s 不应标记 client-stream", name)
+	}
+}
+
+// path-based compact（Codex v1 unary 协议）即使 body 带 stream:true 也不标记，
+// 保持 JSON 写回行为不变。
+func TestNormalizeOpenAIResponsesCompactRequest_PathBasedStreamTrueNotMarked(t *testing.T) {
+	h := &OpenAIGatewayHandler{}
+	body := []byte(`{"model":"gpt-5.5","stream":true,"input":[{"type":"message","role":"user","content":"hello"}]}`)
+	c := newCompactBodySignalTestContext(t, "/v1/responses/compact", body)
+
+	_, ok := h.normalizeOpenAIResponsesCompactRequest(c, zap.NewNop(), body)
+	require.True(t, ok)
+	_, exists := c.Get(service.OpenAICompactClientStreamKeyForTest())
+	require.False(t, exists)
+}

backend/internal/handler/openai_gateway_handler.go

@@ -585,7 +585,14 @@ func (h *OpenAIGatewayHandler) normalizeOpenAIResponsesCompactRequest(c *gin.Con
 	if !isCompactRequest && isBareOpenAIResponsesPath(c) && service.HasCompactionTriggerInInput(body) {
 		c.Request.URL.Path = strings.TrimRight(c.Request.URL.Path, "/") + "/compact"
 		isCompactRequest = true
-		reqLog.Info("codex.remote_compact.detected_body_signal")
+		// Codex remote compact v2 的原始请求是流式 /responses：白名单归一化会删除
+		// stream 并让上游走 unary JSON，但客户端仍按 SSE 消费响应。记录原始
+		// stream 意图，响应写回阶段据此把 JSON 合成回 SSE（#3875）。
+		clientStream := gjson.GetBytes(body, "stream").Bool()
+		if clientStream {
+			service.MarkOpenAICompactClientStream(c)
+		}
+		reqLog.Info("codex.remote_compact.detected_body_signal", zap.Bool("client_stream", clientStream))
 	}
 	if !isCompactRequest {
 		return body, true

backend/internal/service/openai_compact_stream_bridge.go

@@ -0,0 +1,149 @@
+package service
+
+import (
+	"bytes"
+	"encoding/json"
+	"strings"
+
+	"github.com/gin-gonic/gin"
+	"github.com/google/uuid"
+	"github.com/tidwall/gjson"
+	"github.com/tidwall/sjson"
+)
+
+// openAICompactClientStreamKey 标记 body-signal compact 请求（Codex remote
+// compact v2，见 #3777）的原始 body 携带 stream:true。白名单归一化会删除
+// stream 字段并让上游走 unary /responses/compact（JSON），但客户端仍按
+// Responses SSE 协议消费响应：它必须收到 response.output_item.done（其中恰好
+// 一个 type=compaction 的 item）和 response.completed，否则报
+// "stream closed before response.completed" 并无限重连（#3875）。
+const openAICompactClientStreamKey = "openai_compact_client_stream"
+
+// MarkOpenAICompactClientStream 由 handler 在 body-signal 提升时调用，记录
+// 客户端的原始 stream 意图，供响应写回阶段决定是否合成 SSE。
+func MarkOpenAICompactClientStream(c *gin.Context) {
+	if c == nil {
+		return
+	}
+	c.Set(openAICompactClientStreamKey, true)
+}
+
+func OpenAICompactClientStreamKeyForTest() string {
+	return openAICompactClientStreamKey
+}
+
+func openAICompactClientWantsStream(c *gin.Context) bool {
+	if c == nil {
+		return false
+	}
+	value, ok := c.Get(openAICompactClientStreamKey)
+	if !ok {
+		return false
+	}
+	wants, _ := value.(bool)
+	return wants
+}
+
+// writeOpenAICompactSSEBridge 将 unary compact 的最终 JSON 响应按 Codex remote
+// compact v2 的消费协议合成为最小 Responses SSE 流写回客户端。仅当请求被标记
+// 为 body-signal 客户端流式、状态码为 2xx 且 body 是合法 JSON 对象时生效；
+// 返回 false 表示未写出任何内容，调用方应按原路径写回。
+func writeOpenAICompactSSEBridge(c *gin.Context, statusCode int, finalResponse []byte) bool {
+	if c == nil || statusCode < 200 || statusCode >= 300 || !openAICompactClientWantsStream(c) {
+		return false
+	}
+	payload, ok := buildOpenAICompactSSEPayload(finalResponse)
+	if !ok {
+		return false
+	}
+	header := c.Writer.Header()
+	header.Set("Content-Type", "text/event-stream")
+	header.Set("Cache-Control", "no-cache")
+	header.Set("Connection", "keep-alive")
+	header.Set("X-Accel-Buffering", "no")
+	c.Writer.WriteHeader(statusCode)
+	_, _ = c.Writer.Write(payload)
+	c.Writer.Flush()
+	return true
+}
+
+// buildOpenAICompactSSEPayload 把 compact 的 Response JSON 转成 SSE 事件序列：
+// 每个 output[] item 一条 response.output_item.done，最后一条 response.completed
+// 携带完整 response 对象。Codex 的 SSE 解析只从 output_item.done 收集 item，
+// 并要求 response.completed 的 response.id 必填、usage（若存在）必须携带
+// input_tokens/output_tokens/total_tokens 整数字段，否则整条 completed 事件
+// 解析失败，故此处做兜底修补。
+func buildOpenAICompactSSEPayload(finalResponse []byte) ([]byte, bool) {
+	if len(finalResponse) == 0 || !gjson.ValidBytes(finalResponse) {
+		return nil, false
+	}
+	if !gjson.ParseBytes(finalResponse).IsObject() {
+		return nil, false
+	}
+	// SSE 的 data 行不允许出现裸换行：上游 JSON 可能是 pretty-printed 形态，
+	// 嵌入前必须压缩为单行。
+	var compacted bytes.Buffer
+	if err := json.Compact(&compacted, finalResponse); err != nil {
+		return nil, false
+	}
+	response := compacted.Bytes()
+	root := gjson.ParseBytes(response)
+	if strings.TrimSpace(root.Get("id").String()) == "" {
+		next, err := sjson.SetBytes(response, "id", "resp_"+strings.ReplaceAll(uuid.NewString(), "-", ""))
+		if err != nil {
+			return nil, false
+		}
+		response = next
+	}
+	if usage := gjson.GetBytes(response, "usage"); usage.Exists() && !openAICompactUsageParsableByCodex(usage) {
+		next, err := sjson.DeleteBytes(response, "usage")
+		if err != nil {
+			return nil, false
+		}
+		response = next
+	}
+
+	var buf bytes.Buffer
+	outputIndex := 0
+	appendEvent := func(eventType string, data []byte) {
+		buf.WriteString("event: ")
+		buf.WriteString(eventType)
+		buf.WriteString("\ndata: ")
+		buf.Write(data)
+		buf.WriteString("\n\n")
+	}
+	for _, item := range gjson.GetBytes(response, "output").Array() {
+		if !item.IsObject() {
+			continue
+		}
+		event, err := sjson.SetBytes([]byte(`{"type":"response.output_item.done"}`), "output_index", outputIndex)
+		if err != nil {
+			return nil, false
+		}
+		event, err = sjson.SetRawBytes(event, "item", []byte(item.Raw))
+		if err != nil {
+			return nil, false
+		}
+		appendEvent("response.output_item.done", event)
+		outputIndex++
+	}
+
+	completed, err := sjson.SetRawBytes([]byte(`{"type":"response.completed"}`), "response", response)
+	if err != nil {
+		return nil, false
+	}
+	appendEvent("response.completed", completed)
+	return buf.Bytes(), true
+}
+
+func openAICompactUsageParsableByCodex(usage gjson.Result) bool {
+	if !usage.IsObject() {
+		return false
+	}
+	for _, field := range []string{"input_tokens", "output_tokens", "total_tokens"} {
+		if usage.Get(field).Type != gjson.Number {
+			return false
+		}
+	}
+	return true
+}