fix(compact): 二轮审计加固——心跳字节不得污染 failover 判定与并发写安全

对首轮修复的对抗式审计发现并修复以下问题：

1. failover 判定污染（真实回归风险）：handler 以 "Forward 前后
   c.Writer.Size() 是否变化" 判定响应是否已写出并据此放弃换号。心跳注释
   字节会使该判定恒真，compact 请求一旦在上游等待期间发过心跳，上游
   429/5xx 将不再 failover。新增
   OpenAICompactKeepaliveAdjustedWrittenSize（扣除心跳字节、互斥锁下
   一致读取、仅心跳字节归一化为未写哨兵 -1），快照、failover 比较与
   openAIForwardErrorAlreadyCommunicated 三处判定统一改用该口径；无心跳
   请求完全等价于原 c.Writer.Size()。

2. 并发写竞争：心跳 goroutine 与未被显式拦截的写回路径（Forward 内部
   本地拒绝等）存在 ResponseWriter 数据竞争。StartOpenAICompactSSE-
   Keepalive 现将 c.Writer 替换为 openAICompactKeepaliveWriter：写侧
   方法（Header/Write/WriteString/WriteHeader/WriteHeaderNow/Flush）
   先在互斥锁下停拍，读侧（Status/Size/Written）仅加锁不停拍——任何
   请求侧响应构造与心跳从构造上互斥，热路径状态读取不误杀心跳。

3. 语义拦截补齐：rejectIfCyberSessionBlocked（在用户槽位长等待之后
   执行的直接 c.JSON）与 writeOpenAIFastPolicyBlockedResponse 在心跳
   提交后降级为 response.failed 终止事件；未提交时先停拍再写 JSON，
   状态码语义不变。失败事件 errType 参数化（permission_error 等）。

4. reconstruct 混合形态：done 事件存在但 compaction 只在
   output_item.added 中时也要补入；done 已含 compaction 时跳过 added，
   避免无 id 可去重时收集两份（Codex 要求恰好一个）。

5. 观测性：logOpenAIRemoteCompactOutcome 对心跳提交后的失败（wire 200）
   以 GetOpsStreamError 纠正 outcome，不再误记 succeeded。

新增 5 个测试：failover 口径不变式、包装器停拍语义（-race）、fast
policy 提交前后两态、混合 done/added 形态（含去重）。

Refs #3887 #3777

backend/internal/handler/openai_gateway_handler.go

@@ -407,7 +407,9 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) {
 		// Forward request
 		service.SetOpsLatencyMs(c, service.OpsRoutingLatencyMsKey, time.Since(routingStart).Milliseconds())
 		forwardStart := time.Now()
-		writerSizeBeforeForward := c.Writer.Size()
+		// 用扣除 compact 心跳字节的口径快照：心跳注释不构成语义响应，
+		// 不能因心跳字节变化而放弃 failover 换号（#3887）。
+		writerSizeBeforeForward := service.OpenAICompactKeepaliveAdjustedWrittenSize(c)
 		result, err := func() (*service.OpenAIForwardResult, error) {
 			defer func() {
 				if accountReleaseFunc != nil {
@@ -441,7 +443,7 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) {
 			} else {
 				var failoverErr *service.UpstreamFailoverError
 				if errors.As(err, &failoverErr) {
-					if c.Writer.Size() != writerSizeBeforeForward {
+					if service.OpenAICompactKeepaliveAdjustedWrittenSize(c) != writerSizeBeforeForward {
 						h.handleFailoverExhausted(c, failoverErr, true)
 						return
 					}
@@ -642,6 +644,13 @@ func (h *OpenAIGatewayHandler) logOpenAIRemoteCompactOutcome(c *gin.Context, sta
 	if status >= 200 && status < 300 {
 		outcome = "succeeded"
 	}
+	// compact 心跳提交后失败的 wire 状态码固化为 200，真实结局以流内错误
+	// 标记为准（response.failed 降级路径会 MarkOpsStreamError）。
+	if outcome == "succeeded" && c != nil {
+		if _, hasStreamErr := service.GetOpsStreamError(c); hasStreamErr {
+			outcome = "failed"
+		}
+	}
 	latencyMs := time.Since(startedAt).Milliseconds()
 	if latencyMs < 0 {
 		latencyMs = 0
@@ -2024,7 +2033,9 @@ func openAIForwardErrorAlreadyCommunicated(c *gin.Context, writerSizeBeforeForwa
 	if err == nil || c == nil || c.Writer == nil {
 		return false
 	}
-	if c.Writer.Size() == writerSizeBeforeForward {
+	// 与快照同口径：排除 compact 心跳字节，避免"仅心跳写出"被误判为
+	// 响应已写出（#3887）。
+	if service.OpenAICompactKeepaliveAdjustedWrittenSize(c) == writerSizeBeforeForward {
 		return false
 	}
 
@@ -2337,6 +2348,16 @@ func (h *OpenAIGatewayHandler) rejectIfCyberSessionBlocked(c *gin.Context, apiKe
 	if !h.gatewayService.IsCyberSessionBlocked(c.Request.Context(), key) {
 		return false
 	}
+	// body-signal compact 心跳可能已把响应头提交为 200（cyber 检查在用户槽位
+	// 长等待之后执行）：以 response.failed 终止事件回传；未提交时停拍后照常
+	// 写 JSON（#3887）。
+	if service.StopOpenAICompactSSEKeepaliveCommitted(c) {
+		service.MarkOpsStreamError(c, "permission_error", cyberSessionBlockedClientMsg, http.StatusForbidden)
+		if writeResponsesFailedSSE(c, "permission_error", cyberSessionBlockedClientMsg) {
+			h.enqueueCyberSessionBlockedOpsEntry(c, apiKey, model, key)
+			return true
+		}
+	}
 	switch format {
 	case cyberBlockFormatAnthropic:
 		c.JSON(http.StatusForbidden, gin.H{"type": "error", "error": gin.H{

backend/internal/service/openai_compact_sse_keepalive.go

@@ -26,11 +26,19 @@ type openAICompactSSEKeepalive struct {
 	writer  gin.ResponseWriter
 	started bool
 	stopped bool
-	stop    chan struct{}
+	// bytes 是心跳已写出的注释字节数。心跳不构成语义响应，handler 的
+	// "Forward 期间是否已写响应"判定（failover 放弃换号的依据）必须扣除
+	// 这部分字节，见 OpenAICompactKeepaliveAdjustedWrittenSize。
+	bytes int
+	stop  chan struct{}
 }
 
 // StartOpenAICompactSSEKeepalive 为已标记 body-signal 客户端流式的 compact
 // 请求启动下游心跳，返回幂等的停止函数。interval<=0 或请求未标记时为 no-op。
+//
+// 同时把 c.Writer 替换为 openAICompactKeepaliveWriter：请求 goroutine 的任何
+// 响应构造都会先在心跳互斥锁下停拍，未被显式拦截的写回路径（如 Forward
+// 内部的本地拒绝）也不会与心跳 goroutine 产生数据竞争或字节交错。
 func StartOpenAICompactSSEKeepalive(c *gin.Context, interval time.Duration) func() {
 	if c == nil || c.Writer == nil || interval <= 0 || !openAICompactClientWantsStream(c) {
 		return func() {}
@@ -40,6 +48,7 @@ func StartOpenAICompactSSEKeepalive(c *gin.Context, interval time.Duration) func
 		stop:   make(chan struct{}),
 	}
 	c.Set(openAICompactSSEKeepaliveKey, k)
+	c.Writer = &openAICompactKeepaliveWriter{ResponseWriter: c.Writer, k: k}
 
 	var reqDone <-chan struct{}
 	if c.Request != nil {
@@ -82,7 +91,9 @@ func (k *openAICompactSSEKeepalive) beat() bool {
 		k.writer.WriteHeader(http.StatusOK)
 		k.started = true
 	}
-	if _, err := k.writer.Write([]byte(": keepalive\n\n")); err != nil {
+	n, err := k.writer.Write([]byte(": keepalive\n\n"))
+	k.bytes += n
+	if err != nil {
 		k.stopped = true
 		return false
 	}
@@ -127,3 +138,96 @@ func StopOpenAICompactSSEKeepaliveCommitted(c *gin.Context) bool {
 	k.mu.Unlock()
 	return committed
 }
+
+// OpenAICompactKeepaliveAdjustedWrittenSize 返回排除 compact 心跳注释字节后
+// 的响应已写字节数；无心跳的请求等价于 c.Writer.Size()。心跳字节不构成语义
+// 响应——handler 以"Forward 前后 Size 是否变化"判定是否已向客户端写出响应
+// （变化则放弃 failover 换号），该判定不得被心跳污染，否则 compact 请求
+// 一旦在上游等待期间发过心跳，上游 429/5xx 就不再换号（#3887 加固审计）。
+// 仅心跳字节时归一化为 -1（gin 的"未写出"哨兵值），与提交前的快照可比。
+func OpenAICompactKeepaliveAdjustedWrittenSize(c *gin.Context) int {
+	if c == nil || c.Writer == nil {
+		return -1
+	}
+	value, ok := c.Get(openAICompactSSEKeepaliveKey)
+	if !ok {
+		return c.Writer.Size()
+	}
+	k, ok := value.(*openAICompactSSEKeepalive)
+	if !ok || k == nil {
+		return c.Writer.Size()
+	}
+	k.mu.Lock()
+	defer k.mu.Unlock()
+	size := k.writer.Size()
+	if size < 0 {
+		return size
+	}
+	if real := size - k.bytes; real > 0 {
+		return real
+	}
+	return -1
+}
+
+// openAICompactKeepaliveWriter 包装 gin.ResponseWriter：写侧方法先停拍心跳
+// （互斥锁下建立 happens-before），读侧方法仅加锁不停拍——热路径的状态读取
+// （如 Forward 前的 Size 快照）不能误杀心跳。心跳 goroutine 直接写内层
+// writer（k.writer），不经过本包装器，不会递归。
+type openAICompactKeepaliveWriter struct {
+	gin.ResponseWriter
+	k *openAICompactSSEKeepalive
+}
+
+// suspend 停拍心跳；幂等。任何响应构造（含 Header 访问——写响应必先操作
+// 响应头）都视为请求侧接管 ResponseWriter。
+func (w *openAICompactKeepaliveWriter) suspend() {
+	w.k.Stop()
+}
+
+func (w *openAICompactKeepaliveWriter) Header() http.Header {
+	w.suspend()
+	return w.ResponseWriter.Header()
+}
+
+func (w *openAICompactKeepaliveWriter) Write(data []byte) (int, error) {
+	w.suspend()
+	return w.ResponseWriter.Write(data)
+}
+
+func (w *openAICompactKeepaliveWriter) WriteString(s string) (int, error) {
+	w.suspend()
+	return w.ResponseWriter.WriteString(s)
+}
+
+func (w *openAICompactKeepaliveWriter) WriteHeader(code int) {
+	w.suspend()
+	w.ResponseWriter.WriteHeader(code)
+}
+
+func (w *openAICompactKeepaliveWriter) WriteHeaderNow() {
+	w.suspend()
+	w.ResponseWriter.WriteHeaderNow()
+}
+
+func (w *openAICompactKeepaliveWriter) Flush() {
+	w.suspend()
+	w.ResponseWriter.Flush()
+}
+
+func (w *openAICompactKeepaliveWriter) Status() int {
+	w.k.mu.Lock()
+	defer w.k.mu.Unlock()
+	return w.ResponseWriter.Status()
+}
+
+func (w *openAICompactKeepaliveWriter) Size() int {
+	w.k.mu.Lock()
+	defer w.k.mu.Unlock()
+	return w.ResponseWriter.Size()
+}
+
+func (w *openAICompactKeepaliveWriter) Written() bool {
+	w.k.mu.Lock()
+	defer w.k.mu.Unlock()
+	return w.ResponseWriter.Written()
+}

backend/internal/service/openai_compact_sse_keepalive_test.go

@@ -120,3 +120,73 @@ func TestWriteOpenAICompactSSEBridge_BeforeKeepaliveCommitFailureKeepsJSONPath(t
 	require.False(t, writeOpenAICompactSSEBridge(c, http.StatusBadGateway, []byte(`{"error":{"message":"fast fail"}}`)))
 	require.Zero(t, rec.Body.Len())
 }
+
+// 未被显式拦截的写回路径（直接操作 c.Writer）也必须与心跳互斥：包装器在
+// 请求侧任何响应构造时停拍。-race 下验证无数据竞争，且停拍后不再有心跳
+// 字节写出。
+func TestOpenAICompactKeepaliveWriter_RequestSideWriteSuspendsBeats(t *testing.T) {
+	c, rec := newCompactBridgeTestContext(t, true)
+	stop := StartOpenAICompactSSEKeepalive(c, keepaliveTestInterval)
+	defer stop()
+	waitForKeepaliveBeats()
+
+	// 模拟未拦截路径的直接写回（如 Forward 内部本地拒绝的 c.JSON）。
+	_, err := c.Writer.Write([]byte(`{"error":"local reject"}`))
+	require.NoError(t, err)
+
+	lenAfterWrite := rec.Body.Len()
+	waitForKeepaliveBeats()
+	require.Equal(t, lenAfterWrite, rec.Body.Len(), "请求侧写回后心跳必须停止")
+	require.Contains(t, rec.Body.String(), ": keepalive\n\n")
+	require.Contains(t, rec.Body.String(), `{"error":"local reject"}`)
+}
+
+// fast policy block 在心跳提交后必须降级为 response.failed 终止事件。
+func TestWriteOpenAIFastPolicyBlockedResponse_AfterKeepaliveCommit(t *testing.T) {
+	c, rec := newCompactBridgeTestContext(t, true)
+	stop := StartOpenAICompactSSEKeepalive(c, keepaliveTestInterval)
+	defer stop()
+	waitForKeepaliveBeats()
+
+	writeOpenAIFastPolicyBlockedResponse(c, &OpenAIFastBlockedError{Message: "tier blocked"})
+
+	require.Equal(t, http.StatusOK, rec.Code)
+	events := parseCompactBridgeSSE(t, stripKeepaliveComments(rec.Body.String()))
+	require.Len(t, events, 1)
+	require.Equal(t, "response.failed", events[0][0])
+	require.Equal(t, "permission_error", gjson.Get(events[0][1], "response.error.code").String())
+	require.Contains(t, gjson.Get(events[0][1], "response.error.message").String(), "tier blocked")
+}
+
+// failover"是否已写响应"判定的口径：心跳字节必须被排除，否则 compact 在
+// 上游等待期间发过心跳后，可换号的 failover 会被误判放弃；真实响应字节
+// 写出后口径必须变化。
+func TestOpenAICompactKeepaliveAdjustedWrittenSize_ExcludesHeartbeatBytes(t *testing.T) {
+	c, rec := newCompactBridgeTestContext(t, true)
+	// 无心跳的请求：等价于 c.Writer.Size()。
+	require.Equal(t, c.Writer.Size(), OpenAICompactKeepaliveAdjustedWrittenSize(c))
+
+	stop := StartOpenAICompactSSEKeepalive(c, keepaliveTestInterval)
+	defer stop()
+	before := OpenAICompactKeepaliveAdjustedWrittenSize(c)
+	waitForKeepaliveBeats()
+	require.Equal(t, before, OpenAICompactKeepaliveAdjustedWrittenSize(c), "仅心跳字节不得改变判定口径")
+
+	// 真实响应字节写出（经包装器，先停拍再写）后口径必须变化。
+	_, err := c.Writer.Write([]byte("real-bytes"))
+	require.NoError(t, err)
+	require.Equal(t, len("real-bytes"), OpenAICompactKeepaliveAdjustedWrittenSize(c))
+	require.Contains(t, rec.Body.String(), ": keepalive\n\n")
+}
+
+// fast policy block 在心跳未提交时保持 403 JSON 原语义。
+func TestWriteOpenAIFastPolicyBlockedResponse_BeforeKeepaliveCommit(t *testing.T) {
+	c, rec := newCompactBridgeTestContext(t, true)
+	stop := StartOpenAICompactSSEKeepalive(c, time.Hour)
+	defer stop()
+
+	writeOpenAIFastPolicyBlockedResponse(c, &OpenAIFastBlockedError{Message: "tier blocked"})
+
+	require.Equal(t, http.StatusForbidden, rec.Code)
+	require.Equal(t, "permission_error", gjson.Get(rec.Body.String(), "error.type").String())
+}

backend/internal/service/openai_compact_stream_bridge.go

@@ -100,18 +100,18 @@ func writeOpenAICompactSSEFailure(c *gin.Context, statusCode int, errorBody []by
 	if message == "" {
 		message = "Upstream compact request failed with HTTP " + strconv.Itoa(statusCode)
 	}
-	writeOpenAICompactSSEFailureMessage(c, statusCode, message)
+	writeOpenAICompactSSEFailureMessage(c, statusCode, "upstream_error", message)
 }
 
 // writeOpenAICompactSSEFailureMessage 写出 response.failed 终止事件。Codex 对
 // 流式 Responses 请求把 response.failed 作为合法终止事件处理（普通 error 帧
 // 不被识别，会退化为 "stream closed before response.completed" 盲重连）。
 // 同时标记流内错误，保证挂在 200 流上的失败仍进入 ops 错误看板。
-func writeOpenAICompactSSEFailureMessage(c *gin.Context, statusCode int, message string) {
+func writeOpenAICompactSSEFailureMessage(c *gin.Context, statusCode int, errType, message string) {
 	if c == nil {
 		return
 	}
-	MarkOpsStreamError(c, "upstream_error", message, statusCode)
+	MarkOpsStreamError(c, errType, message, statusCode)
 	payload, err := json.Marshal(map[string]any{
 		"type": "response.failed",
 		"response": map[string]any{
@@ -120,7 +120,7 @@ func writeOpenAICompactSSEFailureMessage(c *gin.Context, statusCode int, message
 			"status": "failed",
 			"output": []any{},
 			"error": map[string]any{
-				"code":    "upstream_error",
+				"code":    errType,
 				"message": message,
 			},
 		},

backend/internal/service/openai_compact_stream_bridge_test.go

@@ -392,6 +392,36 @@ func TestReconstructResponseOutputFromSSE_CompactionAddedFallback(t *testing.T)
 	require.Equal(t, "added-only", items[0].Get("encrypted_content").String())
 }
 
+// 混合形态：其他 item 有 done、compaction 只在 added 中——compaction 必须
+// 被补入；done 已含 compaction 时 added 不得重复计入。
+func TestReconstructResponseOutputFromSSE_MixedDoneAndCompactionAdded(t *testing.T) {
+	bodyText := strings.Join([]string{
+		`data: {"type":"response.output_item.added","output_index":0,"item":{"id":"cmp_mixed","type":"compaction","encrypted_content":"mixed"}}`,
+		`data: {"type":"response.output_item.done","output_index":1,"item":{"id":"msg_1","type":"message","content":[{"type":"output_text","text":"hi"}]}}`,
+		`data: {"type":"response.completed","response":{"id":"resp_1","output":[]}}`,
+	}, "\n")
+
+	outputJSON, ok := reconstructResponseOutputFromSSE(bodyText)
+	require.True(t, ok)
+	items := gjson.ParseBytes(outputJSON).Array()
+	require.Len(t, items, 2)
+	require.Equal(t, "msg_1", items[0].Get("id").String())
+	require.Equal(t, "cmp_mixed", items[1].Get("id").String())
+
+	// done 已含 compaction：added 中的同一 item（无 id 可去重的最坏情况用
+	// 不同 raw 表达）不得再收集，Codex 要求恰好一个 compaction item。
+	bodyText = strings.Join([]string{
+		`data: {"type":"response.output_item.added","output_index":0,"item":{"type":"compaction","status":"in_progress"}}`,
+		`data: {"type":"response.output_item.done","output_index":0,"item":{"type":"compaction","status":"completed","encrypted_content":"final"}}`,
+		`data: {"type":"response.completed","response":{"id":"resp_1","output":[]}}`,
+	}, "\n")
+	outputJSON, ok = reconstructResponseOutputFromSSE(bodyText)
+	require.True(t, ok)
+	items = gjson.ParseBytes(outputJSON).Array()
+	require.Len(t, items, 1)
+	require.Equal(t, "final", items[0].Get("encrypted_content").String())
+}
+
 // 非 compaction 的 output_item.added 不参与回退收集（added 阶段的 message
 // 通常是空壳），仍走 delta 重建。
 func TestReconstructResponseOutputFromSSE_NonCompactionAddedStillUsesDeltas(t *testing.T) {

backend/internal/service/openai_gateway_request_body.go

@@ -772,6 +772,13 @@ func writeOpenAIFastPolicyBlockedResponse(c *gin.Context, err *OpenAIFastBlocked
 		return
 	}
 	MarkOpsClientBusinessLimited(c, OpsClientBusinessLimitedReasonLocalPolicyDenied)
+	// body-signal compact 心跳可能已把响应头提交为 200（长排队后才进入
+	// Forward），此时以 response.failed 终止事件回传；未提交时先停拍再写
+	// JSON，保持原状态码语义（#3887）。
+	if StopOpenAICompactSSEKeepaliveCommitted(c) {
+		writeOpenAICompactSSEFailureMessage(c, http.StatusForbidden, "permission_error", err.Message)
+		return
+	}
 	c.JSON(http.StatusForbidden, gin.H{
 		"error": gin.H{
 			"type":    "permission_error",

backend/internal/service/openai_gateway_response_handling.go

@@ -1015,7 +1015,7 @@ func (s *OpenAIGatewayService) writeOpenAINonStreamingProtocolError(resp *http.R
 	// body-signal compact 心跳可能已把响应头提交为 200，此时只能以
 	// response.failed 终止事件回传错误，不能再写 JSON+状态码。
 	if openAICompactClientWantsStream(c) && StopOpenAICompactSSEKeepaliveCommitted(c) {
-		writeOpenAICompactSSEFailureMessage(c, http.StatusBadGateway, message)
+		writeOpenAICompactSSEFailureMessage(c, http.StatusBadGateway, "upstream_error", message)
 		return fmt.Errorf("non-streaming openai protocol error: %s", message)
 	}
 	responseheaders.WriteFilteredHeaders(c.Writer.Header(), resp.Header, s.responseHeaderFilter)
@@ -1096,6 +1096,7 @@ func responsesStreamEventMayContributeToOutput(eventType string) bool {
 func collectRawResponsesOutputItemsFromSSE(bodyText string) ([]byte, bool) {
 	var items []json.RawMessage
 	seen := make(map[string]struct{})
+	hasCompactionItem := false
 	appendItem := func(item gjson.Result) {
 		if !item.Exists() || !item.IsObject() {
 			return
@@ -1108,6 +1109,9 @@ func collectRawResponsesOutputItemsFromSSE(bodyText string) ([]byte, bool) {
 			return
 		}
 		seen[key] = struct{}{}
+		if isResponsesCompactionItemType(item.Get("type").String()) {
+			hasCompactionItem = true
+		}
 		items = append(items, json.RawMessage(item.Raw))
 	}
 	forEachOpenAISSEDataPayload(bodyText, func(data []byte) {
@@ -1116,7 +1120,10 @@ func collectRawResponsesOutputItemsFromSSE(bodyText string) ([]byte, bool) {
 		}
 		appendItem(gjson.GetBytes(data, "item"))
 	})
-	if len(items) == 0 {
+	// done 事件未携带 compaction item 时再看 added：覆盖"其他 item 有 done、
+	// compaction 只在 added 中"的混合形态；done 已含 compaction 时跳过，
+	// 避免同一 item 在无 id 可去重时被收集两份（Codex 要求恰好一个）。
+	if !hasCompactionItem {
 		forEachOpenAISSEDataPayload(bodyText, func(data []byte) {
 			if strings.TrimSpace(gjson.GetBytes(data, "type").String()) != "response.output_item.added" {
 				return