Merge pull request #3906 from Wei-Shaw/fix/compact-sse-raw-output-item-preservation

fix(compact): SSE→JSON 保留 raw output_item.done 并为 unary 等待补下游心跳（修复 #3887）

backend/internal/handler/openai_gateway_handler.go

@@ -205,6 +205,11 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) {
 	if !ok {
 		return
 	}
+	// body-signal compact：上游 unary 等待期间向下游发 SSE 注释行心跳，防止
+	// 反向代理空闲超时掐断长压缩连接（#3887）。首拍延迟一个心跳间隔，快速
+	// 失败仍走 JSON+状态码链路；未标记客户端流式或间隔为 0 时是 no-op。
+	stopCompactKeepalive := service.StartOpenAICompactSSEKeepalive(c, h.openAICompactKeepaliveInterval())
+	defer stopCompactKeepalive()
 
 	// 校验请求体 JSON 合法性
 	if !gjson.ValidBytes(body) {
@@ -402,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 {
@@ -436,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
 					}
@@ -637,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
@@ -1943,6 +1957,11 @@ func (h *OpenAIGatewayHandler) mapUpstreamError(statusCode int) (int, string, st
 
 // handleStreamingAwareError handles errors that may occur after streaming has started
 func (h *OpenAIGatewayHandler) handleStreamingAwareError(c *gin.Context, status int, errType, message string, streamStarted bool) {
+	// body-signal compact 心跳可能已把响应头提交为 200：先停心跳（建立
+	// happens-before，接管 ResponseWriter），并升级为流内错误处理。
+	if service.StopOpenAICompactSSEKeepaliveCommitted(c) {
+		streamStarted = true
+	}
 	if streamStarted {
 		// /v1/responses 的严格 SDK（Codex CLI）要求终止事件必须属于
 		// response.completed/failed/incomplete/cancelled 集合。
@@ -1975,6 +1994,10 @@ func (h *OpenAIGatewayHandler) ensureForwardErrorResponse(c *gin.Context, stream
 	if c == nil || c.Writer == nil {
 		return false
 	}
+	// 先停 compact 心跳再读 Writer 状态，避免与心跳 goroutine 竞争。
+	if service.StopOpenAICompactSSEKeepaliveCommitted(c) {
+		streamStarted = true
+	}
 	if service.IsResponseCommitted(c) {
 		return false
 	}
@@ -2010,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
 	}
 
@@ -2036,6 +2061,14 @@ func openAIForwardErrorAlreadyCommunicated(c *gin.Context, writerSizeBeforeForwa
 
 // errorResponse returns OpenAI API format error response
 func (h *OpenAIGatewayHandler) errorResponse(c *gin.Context, status int, errType, message string) {
+	// body-signal compact 心跳可能已把响应头提交为 200：JSON 错误体会与已
+	// 提交的 SSE 流交错，必须降级为 response.failed 终止事件（#3887）。
+	if service.StopOpenAICompactSSEKeepaliveCommitted(c) {
+		service.MarkOpsStreamError(c, errType, message, status)
+		if writeResponsesFailedSSE(c, errType, message) {
+			return
+		}
+	}
 	c.JSON(status, gin.H{
 		"error": gin.H{
 			"type":    errType,
@@ -2044,6 +2077,15 @@ func (h *OpenAIGatewayHandler) errorResponse(c *gin.Context, status int, errType
 	})
 }
 
+// openAICompactKeepaliveInterval 复用流式 keepalive 配置作为 compact 下游
+// 心跳间隔；0 表示禁用（与流式路径语义一致）。
+func (h *OpenAIGatewayHandler) openAICompactKeepaliveInterval() time.Duration {
+	if h.cfg == nil || h.cfg.Gateway.StreamKeepaliveInterval <= 0 {
+		return 0
+	}
+	return time.Duration(h.cfg.Gateway.StreamKeepaliveInterval) * time.Second
+}
+
 func setOpenAIClientTransportHTTP(c *gin.Context) {
 	service.SetOpenAIClientTransport(c, service.OpenAIClientTransportHTTP)
 }
@@ -2306,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

@@ -0,0 +1,233 @@
+package service
+
+import (
+	"net/http"
+	"sync"
+	"time"
+
+	"github.com/gin-gonic/gin"
+)
+
+// openAICompactSSEKeepaliveKey 存放 body-signal compact 请求的下游 SSE 心跳器。
+const openAICompactSSEKeepaliveKey = "openai_compact_sse_keepalive"
+
+// openAICompactSSEKeepalive 在 compact 上游 unary 等待期间向下游写 SSE 注释行
+// 心跳。上游 /responses/compact 在模型处理期间不发送任何字节（大上下文可长达
+// 数分钟），下游若经过反向代理（Nginx/Cloudflare Tunnel 等），零字节静默会触发
+// 代理的空闲/读超时并掐断连接，Codex 只会盲目重连并重复消耗上游 compact
+// 配额（#3887）。SSE 注释行在 eventsource 解析层被直接忽略，不会进入客户端
+// 事件流。
+//
+// 首拍延迟一个 interval：绝大多数硬错误（鉴权/参数/限流）在此之前返回，仍走
+// 原 JSON+状态码链路（Codex 按 HTTP 状态码重试）；首拍之后状态码固化为 200，
+// 后续错误由写回方降级为 response.failed 流内终止事件。
+type openAICompactSSEKeepalive struct {
+	mu      sync.Mutex
+	writer  gin.ResponseWriter
+	started bool
+	stopped bool
+	// 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() {}
+	}
+	k := &openAICompactSSEKeepalive{
+		writer: c.Writer,
+		stop:   make(chan struct{}),
+	}
+	c.Set(openAICompactSSEKeepaliveKey, k)
+	c.Writer = &openAICompactKeepaliveWriter{ResponseWriter: c.Writer, k: k}
+
+	var reqDone <-chan struct{}
+	if c.Request != nil {
+		reqDone = c.Request.Context().Done()
+	}
+	go func() {
+		timer := time.NewTimer(interval)
+		defer timer.Stop()
+		for {
+			select {
+			case <-k.stop:
+				return
+			case <-reqDone:
+				return
+			case <-timer.C:
+			}
+			if !k.beat() {
+				return
+			}
+			timer.Reset(interval)
+		}
+	}()
+	return k.Stop
+}
+
+// beat 在锁内提交（首次）响应头并写出一条 SSE 注释行；返回 false 表示心跳已
+// 停止或下游写入失败，goroutine 应退出。
+func (k *openAICompactSSEKeepalive) beat() bool {
+	k.mu.Lock()
+	defer k.mu.Unlock()
+	if k.stopped {
+		return false
+	}
+	if !k.started {
+		header := k.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")
+		k.writer.WriteHeader(http.StatusOK)
+		k.started = true
+	}
+	n, err := k.writer.Write([]byte(": keepalive\n\n"))
+	k.bytes += n
+	if err != nil {
+		k.stopped = true
+		return false
+	}
+	k.writer.Flush()
+	return true
+}
+
+// Stop 停止心跳；幂等，可与写回路径并发调用。
+func (k *openAICompactSSEKeepalive) Stop() {
+	k.mu.Lock()
+	k.markStoppedLocked()
+	k.mu.Unlock()
+}
+
+func (k *openAICompactSSEKeepalive) markStoppedLocked() {
+	if k.stopped {
+		return
+	}
+	k.stopped = true
+	close(k.stop)
+}
+
+// StopOpenAICompactSSEKeepaliveCommitted 停止当前请求的 compact 心跳（若有）
+// 并报告响应头是否已被心跳提交为 200。写回方以此决定继续走原 JSON/状态码
+// 链路，还是降级为流内终止事件。调用后不会再有心跳字节写出，且经由互斥锁
+// 与心跳 goroutine 建立 happens-before，调用方可安全接管 ResponseWriter。
+func StopOpenAICompactSSEKeepaliveCommitted(c *gin.Context) bool {
+	if c == nil {
+		return false
+	}
+	value, ok := c.Get(openAICompactSSEKeepaliveKey)
+	if !ok {
+		return false
+	}
+	k, ok := value.(*openAICompactSSEKeepalive)
+	if !ok || k == nil {
+		return false
+	}
+	k.mu.Lock()
+	k.markStoppedLocked()
+	committed := k.started
+	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

@@ -0,0 +1,192 @@
+package service
+
+import (
+	"net/http"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"github.com/tidwall/gjson"
+)
+
+const keepaliveTestInterval = 10 * time.Millisecond
+
+// waitForKeepaliveBeats 等待至少一次心跳写出。读取 recorder 前必须先经
+// StopOpenAICompactSSEKeepaliveCommitted 停拍建立 happens-before。
+func waitForKeepaliveBeats() {
+	time.Sleep(20 * keepaliveTestInterval)
+}
+
+// stripKeepaliveComments 去掉 SSE 注释块，返回真实事件文本。
+func stripKeepaliveComments(body string) string {
+	var blocks []string
+	for _, block := range strings.Split(strings.TrimSpace(body), "\n\n") {
+		if strings.HasPrefix(strings.TrimSpace(block), ":") {
+			continue
+		}
+		blocks = append(blocks, block)
+	}
+	return strings.Join(blocks, "\n\n")
+}
+
+func TestStartOpenAICompactSSEKeepalive_NoopWhenUnmarkedOrDisabled(t *testing.T) {
+	// 未标记 client stream：不启动。
+	c, rec := newCompactBridgeTestContext(t, false)
+	stop := StartOpenAICompactSSEKeepalive(c, keepaliveTestInterval)
+	waitForKeepaliveBeats()
+	stop()
+	require.Zero(t, rec.Body.Len())
+	require.False(t, StopOpenAICompactSSEKeepaliveCommitted(c))
+
+	// interval=0（配置禁用）：不启动。
+	c, rec = newCompactBridgeTestContext(t, true)
+	stop = StartOpenAICompactSSEKeepalive(c, 0)
+	waitForKeepaliveBeats()
+	stop()
+	require.Zero(t, rec.Body.Len())
+	require.False(t, StopOpenAICompactSSEKeepaliveCommitted(c))
+}
+
+func TestOpenAICompactSSEKeepalive_CommitsHeadersAndComments(t *testing.T) {
+	c, rec := newCompactBridgeTestContext(t, true)
+	stop := StartOpenAICompactSSEKeepalive(c, keepaliveTestInterval)
+	defer stop()
+	waitForKeepaliveBeats()
+
+	require.True(t, StopOpenAICompactSSEKeepaliveCommitted(c))
+	require.Equal(t, http.StatusOK, rec.Code)
+	require.Equal(t, "text/event-stream", rec.Header().Get("Content-Type"))
+	require.Equal(t, "no", rec.Header().Get("X-Accel-Buffering"))
+	require.Contains(t, rec.Body.String(), ": keepalive\n\n")
+}
+
+func TestOpenAICompactSSEKeepalive_StopBeforeFirstBeatKeepsWriterUntouched(t *testing.T) {
+	c, rec := newCompactBridgeTestContext(t, true)
+	stop := StartOpenAICompactSSEKeepalive(c, time.Hour)
+	stop()
+	waitForKeepaliveBeats()
+	require.Zero(t, rec.Body.Len())
+	require.False(t, StopOpenAICompactSSEKeepaliveCommitted(c))
+}
+
+// 心跳已提交后，2xx 桥接续写事件而不重复提交响应头。
+func TestWriteOpenAICompactSSEBridge_AfterKeepaliveCommitAppendsEvents(t *testing.T) {
+	c, rec := newCompactBridgeTestContext(t, true)
+	stop := StartOpenAICompactSSEKeepalive(c, keepaliveTestInterval)
+	defer stop()
+	waitForKeepaliveBeats()
+
+	finalResponse := []byte(`{"id":"resp_ka_1","output":[{"id":"cmp_ka","type":"compaction","encrypted_content":"x"}],"usage":{"input_tokens":1,"output_tokens":1,"total_tokens":2}}`)
+	require.True(t, writeOpenAICompactSSEBridge(c, http.StatusOK, finalResponse))
+
+	require.Equal(t, http.StatusOK, rec.Code)
+	events := parseCompactBridgeSSE(t, stripKeepaliveComments(rec.Body.String()))
+	require.Len(t, events, 2)
+	require.Equal(t, "response.output_item.done", events[0][0])
+	require.Equal(t, "compaction", gjson.Get(events[0][1], "item.type").String())
+	require.Equal(t, "response.completed", events[1][0])
+	require.Equal(t, "resp_ka_1", gjson.Get(events[1][1], "response.id").String())
+}
+
+// 心跳已提交后上游非 2xx：状态码无法回传，必须以 response.failed 终止事件
+// 收尾（Codex 将其作为终止事件处理），并标记流内错误供 ops 采集。
+func TestWriteOpenAICompactSSEBridge_AfterKeepaliveCommitFailureEmitsFailedEvent(t *testing.T) {
+	c, rec := newCompactBridgeTestContext(t, true)
+	stop := StartOpenAICompactSSEKeepalive(c, keepaliveTestInterval)
+	defer stop()
+	waitForKeepaliveBeats()
+
+	require.True(t, writeOpenAICompactSSEBridge(c, http.StatusBadGateway, []byte(`{"error":{"message":"upstream exploded"}}`)))
+
+	events := parseCompactBridgeSSE(t, stripKeepaliveComments(rec.Body.String()))
+	require.Len(t, events, 1)
+	require.Equal(t, "response.failed", events[0][0])
+	require.Equal(t, "failed", gjson.Get(events[0][1], "response.status").String())
+	require.Contains(t, gjson.Get(events[0][1], "response.error.message").String(), "upstream exploded")
+	require.NotEmpty(t, gjson.Get(events[0][1], "response.id").String())
+
+	streamErr, ok := GetOpsStreamError(c)
+	require.True(t, ok)
+	require.Equal(t, http.StatusBadGateway, streamErr.IntendedStatus)
+}
+
+// 心跳未提交时非 2xx 行为不变：返回 false，调用方按原 JSON+状态码写回。
+func TestWriteOpenAICompactSSEBridge_BeforeKeepaliveCommitFailureKeepsJSONPath(t *testing.T) {
+	c, rec := newCompactBridgeTestContext(t, true)
+	stop := StartOpenAICompactSSEKeepalive(c, time.Hour)
+	stop()
+
+	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

@@ -3,6 +3,8 @@ package service
 import (
 	"bytes"
 	"encoding/json"
+	"net/http"
+	"strconv"
 	"strings"
 
 	"github.com/gin-gonic/gin"
@@ -48,25 +50,90 @@ func openAICompactClientWantsStream(c *gin.Context) bool {
 // compact v2 的消费协议合成为最小 Responses SSE 流写回客户端。仅当请求被标记
 // 为 body-signal 客户端流式、状态码为 2xx 且 body 是合法 JSON 对象时生效；
 // 返回 false 表示未写出任何内容，调用方应按原路径写回。
+//
+// 若下游心跳已把响应头提交为 200（见 openAICompactSSEKeepalive），则本函数
+// 必须接管一切写回：非 2xx 或不可合成的响应降级为 response.failed 终止事件，
+// 不能再返回 false（否则调用方的 JSON 写回会与已提交的 SSE 流交错）。
 func writeOpenAICompactSSEBridge(c *gin.Context, statusCode int, finalResponse []byte) bool {
-	if c == nil || statusCode < 200 || statusCode >= 300 || !openAICompactClientWantsStream(c) {
+	if c == nil || !openAICompactClientWantsStream(c) {
+		return false
+	}
+	// 先停心跳再写回，避免注释行与最终事件交错；停止后经互斥锁与心跳
+	// goroutine 建立 happens-before，可安全接管 ResponseWriter。
+	committed := StopOpenAICompactSSEKeepaliveCommitted(c)
+	if statusCode < 200 || statusCode >= 300 {
+		if committed {
+			writeOpenAICompactSSEFailure(c, statusCode, finalResponse)
+			return true
+		}
 		return false
 	}
 	payload, ok := buildOpenAICompactSSEPayload(finalResponse)
 	if !ok {
+		if committed {
+			writeOpenAICompactSSEFailure(c, http.StatusBadGateway, finalResponse)
+			return true
+		}
 		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)
+	if !committed {
+		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
 }
 
+// writeOpenAICompactSSEFailure 从上游错误 body 提取错误消息后，以
+// response.failed 终止事件回传。仅用于心跳已提交 200、无法再按 HTTP 状态码
+// 回传错误的场景。
+func writeOpenAICompactSSEFailure(c *gin.Context, statusCode int, errorBody []byte) {
+	message := ""
+	if len(errorBody) > 0 {
+		message = sanitizeUpstreamErrorMessage(strings.TrimSpace(extractUpstreamErrorMessage(errorBody)))
+	}
+	if message == "" {
+		message = "Upstream compact request failed with HTTP " + strconv.Itoa(statusCode)
+	}
+	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, errType, message string) {
+	if c == nil {
+		return
+	}
+	MarkOpsStreamError(c, errType, message, statusCode)
+	payload, err := json.Marshal(map[string]any{
+		"type": "response.failed",
+		"response": map[string]any{
+			"id":     "resp_" + strings.ReplaceAll(uuid.NewString(), "-", ""),
+			"object": "response",
+			"status": "failed",
+			"output": []any{},
+			"error": map[string]any{
+				"code":    errType,
+				"message": message,
+			},
+		},
+	})
+	if err != nil {
+		return
+	}
+	_, _ = c.Writer.Write([]byte("event: response.failed\ndata: "))
+	_, _ = c.Writer.Write(payload)
+	_, _ = c.Writer.Write([]byte("\n\n"))
+	c.Writer.Flush()
+}
+
 // buildOpenAICompactSSEPayload 把 compact 的 Response JSON 转成 SSE 事件序列：
 // 每个 output[] item 一条 response.output_item.done，最后一条 response.completed
 // 携带完整 response 对象。Codex 的 SSE 解析只从 output_item.done 收集 item，

backend/internal/service/openai_compact_stream_bridge_test.go

@@ -258,6 +258,246 @@ func TestHandleSSEToJSON_CompactClientStreamBridgesToSSE(t *testing.T) {
 	require.Equal(t, "resp_compact_sse", gjson.Get(events[1][1], "response.id").String())
 }
 
+// 回归 #3887（#3777 问题 2）：上游对 compact 返回 SSE，compaction item 只在
+// raw output_item.done 中、终态 response.completed 的 output 为空。SSE→JSON
+// 提取必须保留 raw item 修补终态 output，否则桥接合成 0 个 output_item.done，
+// Codex 报 "expected exactly one compaction output item, got 0" 并盲目重试，
+// 每次重试都重新计费。fixture 取自 #3777 的上游实录形态。
+func TestHandleSSEToJSON_CompactRawOutputItemDoneRepairsEmptyTerminalOutput(t *testing.T) {
+	svc := newCompactBridgeTestService()
+	c, rec := newCompactBridgeTestContext(t, true)
+	upstreamSSE := strings.Join([]string{
+		`data: {"type":"response.output_item.done","output_index":0,"item":{"id":"cmp_1","type":"compaction_summary","status":"completed","summary":[{"type":"summary_text","text":"compact summary"}],"encrypted_content":"compact-payload","opaque":{"kept":true}}}`,
+		``,
+		`data: {"type":"response.completed","response":{"id":"resp_compact","object":"response","model":"gpt-5.1-codex","status":"completed","output":[],"usage":{"input_tokens":9,"output_tokens":4,"total_tokens":13}}}`,
+		``,
+	}, "\n")
+	resp := &http.Response{
+		StatusCode: http.StatusOK,
+		Header:     http.Header{"Content-Type": []string{"text/event-stream"}},
+		Body:       io.NopCloser(strings.NewReader(upstreamSSE)),
+	}
+
+	result, err := svc.handleNonStreamingResponse(context.Background(), resp, c, &Account{ID: 1, Type: AccountTypeOAuth}, "gpt-5.5", "gpt-5.5")
+	require.NoError(t, err)
+	require.NotNil(t, result)
+
+	require.Equal(t, "text/event-stream", rec.Header().Get("Content-Type"))
+	events := parseCompactBridgeSSE(t, rec.Body.String())
+	require.Len(t, events, 2)
+	require.Equal(t, "response.output_item.done", events[0][0])
+	item := gjson.Get(events[0][1], "item")
+	require.Equal(t, "compaction_summary", item.Get("type").String())
+	require.Equal(t, "cmp_1", item.Get("id").String())
+	require.Equal(t, "compact-payload", item.Get("encrypted_content").String())
+	require.Equal(t, "compact summary", item.Get("summary.0.text").String())
+	require.True(t, item.Get("opaque.kept").Bool(), "raw item 字段必须逐字节保留")
+	require.Equal(t, "response.completed", events[1][0])
+	require.Equal(t, "resp_compact", gjson.Get(events[1][1], "response.id").String())
+	require.Len(t, gjson.Get(events[1][1], "response.output").Array(), 1)
+	require.Equal(t, int64(13), gjson.Get(events[1][1], "response.usage.total_tokens").Int())
+
+	require.NotNil(t, result.usage)
+	require.Equal(t, 9, result.usage.InputTokens)
+	require.Equal(t, 4, result.usage.OutputTokens)
+}
+
+// 同一形态经透传分支（handlePassthroughSSEToJSON）也必须修补。
+func TestHandlePassthroughSSEToJSON_CompactRawOutputItemDoneRepairsEmptyTerminalOutput(t *testing.T) {
+	svc := newCompactBridgeTestService()
+	c, rec := newCompactBridgeTestContext(t, true)
+	upstreamSSE := strings.Join([]string{
+		`data: {"type":"response.output_item.done","output_index":0,"item":{"id":"cmp_pt_1","type":"compaction","status":"completed","encrypted_content":"compact-pt-raw"}}`,
+		``,
+		`data: {"type":"response.completed","response":{"id":"resp_compact_pt_raw","object":"response","status":"completed","output":[],"usage":{"input_tokens":6,"output_tokens":2,"total_tokens":8}}}`,
+		``,
+	}, "\n")
+	resp := &http.Response{
+		StatusCode: http.StatusOK,
+		Header:     http.Header{"Content-Type": []string{"text/event-stream"}},
+		Body:       io.NopCloser(strings.NewReader(upstreamSSE)),
+	}
+
+	result, err := svc.handleNonStreamingResponsePassthrough(context.Background(), resp, c, "gpt-5.5", "")
+	require.NoError(t, err)
+	require.NotNil(t, result)
+
+	require.Equal(t, "text/event-stream", rec.Header().Get("Content-Type"))
+	events := parseCompactBridgeSSE(t, rec.Body.String())
+	require.Len(t, events, 2)
+	require.Equal(t, "compaction", gjson.Get(events[0][1], "item.type").String())
+	require.Equal(t, "compact-pt-raw", gjson.Get(events[0][1], "item.encrypted_content").String())
+	require.Len(t, gjson.Get(events[1][1], "response.output").Array(), 1)
+}
+
+// path-based（Codex v1 unary、链式 sub2api）未标记 client stream：同一上游
+// 形态修补后仍按 JSON 写回，output 中必须包含 compaction item。
+func TestHandleSSEToJSON_PathBasedCompactRawOutputItemDoneRepairsJSON(t *testing.T) {
+	svc := newCompactBridgeTestService()
+	c, rec := newCompactBridgeTestContext(t, false)
+	upstreamSSE := strings.Join([]string{
+		`data: {"type":"response.output_item.done","output_index":0,"item":{"id":"cmp_v1","type":"compaction_summary","encrypted_content":"compact-v1-raw"}}`,
+		``,
+		`data: {"type":"response.completed","response":{"id":"resp_compact_v1","object":"response","status":"completed","output":[],"usage":{"input_tokens":5,"output_tokens":1,"total_tokens":6}}}`,
+		``,
+	}, "\n")
+	resp := &http.Response{
+		StatusCode: http.StatusOK,
+		Header:     http.Header{"Content-Type": []string{"text/event-stream"}},
+		Body:       io.NopCloser(strings.NewReader(upstreamSSE)),
+	}
+
+	result, err := svc.handleNonStreamingResponse(context.Background(), resp, c, &Account{ID: 1, Type: AccountTypeOAuth}, "gpt-5.5", "gpt-5.5")
+	require.NoError(t, err)
+	require.NotNil(t, result)
+
+	// 写回 body 必须是修补后的 JSON 文档（非 SSE 事件流）。
+	body := rec.Body.String()
+	require.NotContains(t, body, "event:")
+	require.NotContains(t, body, "data:")
+	require.Equal(t, "resp_compact_v1", gjson.Get(body, "id").String())
+	require.Equal(t, "compaction_summary", gjson.Get(body, "output.0.type").String())
+	require.Equal(t, "compact-v1-raw", gjson.Get(body, "output.0.encrypted_content").String())
+}
+
+// raw done item 是协议上的最终完整形态，优先于 delta 重建且不得重复计入。
+func TestReconstructResponseOutputFromSSE_PrefersRawDoneItems(t *testing.T) {
+	bodyText := strings.Join([]string{
+		`data: {"type":"response.output_text.delta","delta":"hel"}`,
+		`data: {"type":"response.output_text.delta","delta":"lo"}`,
+		`data: {"type":"response.output_item.done","output_index":0,"item":{"id":"msg_1","type":"message","role":"assistant","status":"completed","content":[{"type":"output_text","text":"hello"}]}}`,
+		`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, "raw done item 与 delta 重建不得重复")
+	require.Equal(t, "msg_1", items[0].Get("id").String())
+	require.Equal(t, "hello", items[0].Get("content.0.text").String())
+}
+
+// 无任何 done 事件时，退回收集 output_item.added 中的 compaction 类 item。
+func TestReconstructResponseOutputFromSSE_CompactionAddedFallback(t *testing.T) {
+	bodyText := strings.Join([]string{
+		`data: {"type":"response.output_item.added","output_index":0,"item":{"id":"cmp_add","type":"compaction","encrypted_content":"added-only"}}`,
+		`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, "compaction", items[0].Get("type").String())
+	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())
+}
+
+// 上游不一致形态：终态 output 非空（含 message）但 compaction 只在 raw
+// output_item.done 中。146 纯流式透传下 Codex 直接读事件流能拿到 compaction，
+// SSE→JSON 提取必须补入等价结果。
+func TestHandleSSEToJSON_CompactSupplementsMissingCompactionIntoNonEmptyOutput(t *testing.T) {
+	svc := newCompactBridgeTestService()
+	c, rec := newCompactBridgeTestContext(t, true)
+	upstreamSSE := strings.Join([]string{
+		`data: {"type":"response.output_item.done","output_index":0,"item":{"id":"cmp_sup","type":"compaction","encrypted_content":"supplement"}}`,
+		``,
+		`data: {"type":"response.completed","response":{"id":"resp_sup","object":"response","status":"completed","output":[{"id":"msg_sup","type":"message","role":"assistant","content":[{"type":"output_text","text":"note"}]}],"usage":{"input_tokens":2,"output_tokens":1,"total_tokens":3}}}`,
+		``,
+	}, "\n")
+	resp := &http.Response{
+		StatusCode: http.StatusOK,
+		Header:     http.Header{"Content-Type": []string{"text/event-stream"}},
+		Body:       io.NopCloser(strings.NewReader(upstreamSSE)),
+	}
+
+	result, err := svc.handleNonStreamingResponse(context.Background(), resp, c, &Account{ID: 1, Type: AccountTypeOAuth}, "gpt-5.5", "gpt-5.5")
+	require.NoError(t, err)
+	require.NotNil(t, result)
+
+	events := parseCompactBridgeSSE(t, rec.Body.String())
+	require.Len(t, events, 3)
+	itemTypes := []string{
+		gjson.Get(events[0][1], "item.type").String(),
+		gjson.Get(events[1][1], "item.type").String(),
+	}
+	require.Contains(t, itemTypes, "compaction")
+	require.Contains(t, itemTypes, "message")
+	require.Equal(t, "response.completed", events[2][0])
+	require.Len(t, gjson.Get(events[2][1], "response.output").Array(), 2)
+}
+
+// 补全逻辑的门控：非 compact 请求原样返回；终态已含 compaction 不重复补入。
+func TestSupplementCompactionItemFromSSE_Gating(t *testing.T) {
+	bodyText := `data: {"type":"response.output_item.done","item":{"id":"cmp_g","type":"compaction","encrypted_content":"g"}}` + "\n"
+
+	// 非 compact 路径：不补入。
+	gin.SetMode(gin.TestMode)
+	rec := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(rec)
+	c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil)
+	finalResponse := []byte(`{"id":"r1","output":[{"type":"message"}]}`)
+	require.Equal(t, string(finalResponse), string(supplementCompactionItemFromSSE(c, finalResponse, bodyText)))
+
+	// compact 路径 + 终态已含 compaction：不重复补入。
+	c2, _ := newCompactBridgeTestContext(t, false)
+	already := []byte(`{"id":"r2","output":[{"type":"compaction","encrypted_content":"x"}]}`)
+	require.Equal(t, string(already), string(supplementCompactionItemFromSSE(c2, already, bodyText)))
+
+	// compact 路径 + 终态非空缺 compaction：补入到末尾。
+	missing := []byte(`{"id":"r3","output":[{"type":"message"}]}`)
+	patched := supplementCompactionItemFromSSE(c2, missing, bodyText)
+	items := gjson.GetBytes(patched, "output").Array()
+	require.Len(t, items, 2)
+	require.Equal(t, "compaction", items[1].Get("type").String())
+	require.Equal(t, "g", items[1].Get("encrypted_content").String())
+}
+
+// 非 compaction 的 output_item.added 不参与回退收集（added 阶段的 message
+// 通常是空壳），仍走 delta 重建。
+func TestReconstructResponseOutputFromSSE_NonCompactionAddedStillUsesDeltas(t *testing.T) {
+	bodyText := strings.Join([]string{
+		`data: {"type":"response.output_item.added","output_index":0,"item":{"id":"msg_1","type":"message","content":[]}}`,
+		`data: {"type":"response.output_text.delta","delta":"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, 1)
+	require.Equal(t, "hi", items[0].Get("content.0.text").String())
+}
+
 // 透传分支（OAuth passthrough）同样命中桥接。
 func TestHandleNonStreamingResponsePassthrough_CompactClientStreamBridgesToSSE(t *testing.T) {
 	svc := newCompactBridgeTestService()

backend/internal/service/openai_gateway_forward.go

@@ -902,6 +902,10 @@ func (s *OpenAIGatewayService) buildUpstreamRequest(ctx context.Context, c *gin.
 				req.Header.Set("conversation_id", isolated)
 			}
 		}
+	} else if isOpenAIResponsesCompactPath(c) {
+		// compact 上游是 unary JSON 协议：API-key 账号也显式声明 Accept，
+		// 避免 OpenAI 兼容网关按 SSE 返回（#3777 期望行为 4）。
+		req.Header.Set("accept", "application/json")
 	}
 
 	// Apply custom User-Agent if configured

backend/internal/service/openai_gateway_passthrough.go

@@ -382,6 +382,11 @@ func (s *OpenAIGatewayService) buildUpstreamRequestOpenAIPassthrough(
 		if clientConversationID != "" {
 			req.Header.Set("conversation_id", isolateOpenAISessionID(apiKeyID, clientConversationID))
 		}
+	} else if isOpenAIResponsesCompactPath(c) {
+		// 透传白名单会放行客户端的 Accept: text/event-stream；compact 上游是
+		// unary JSON 协议，API-key 账号同样强制 Accept，避免上游按 SSE 返回
+		// （#3777 期望行为 4）。
+		req.Header.Set("accept", "application/json")
 	}
 
 	// 透传模式也支持账户自定义 User-Agent 与 ForceCodexCLI 兜底。
@@ -1122,6 +1127,7 @@ func (s *OpenAIGatewayService) handlePassthroughSSEToJSON(resp *http.Response, c
 				}
 			}
 		}
+		finalResponse = supplementCompactionItemFromSSE(c, finalResponse, bodyText)
 		body = finalResponse
 		if originalModel != "" && mappedModel != "" && originalModel != mappedModel {
 			body = s.replaceModelInResponseBody(body, mappedModel, originalModel)

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

@@ -878,6 +878,7 @@ func (s *OpenAIGatewayService) handleSSEToJSON(resp *http.Response, c *gin.Conte
 				}
 			}
 		}
+		finalResponse = supplementCompactionItemFromSSE(c, finalResponse, bodyText)
 		body = finalResponse
 		if originalModel != mappedModel {
 			body = s.replaceModelInResponseBody(body, mappedModel, originalModel)
@@ -1012,6 +1013,12 @@ func (s *OpenAIGatewayService) writeOpenAINonStreamingProtocolError(resp *http.R
 		message = "Upstream returned an invalid non-streaming response"
 	}
 	setOpsUpstreamError(c, http.StatusBadGateway, message, "")
+	// body-signal compact 心跳可能已把响应头提交为 200，此时只能以
+	// response.failed 终止事件回传错误，不能再写 JSON+状态码。
+	if openAICompactClientWantsStream(c) && StopOpenAICompactSSEKeepaliveCommitted(c) {
+		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)
 	c.Writer.Header().Set("Content-Type", "application/json; charset=utf-8")
 	c.JSON(http.StatusBadGateway, gin.H{
@@ -1081,10 +1088,152 @@ func responsesStreamEventMayContributeToOutput(eventType string) bool {
 	}
 }
 
-// reconstructResponseOutputFromSSE scans raw SSE body text for delta events and
-// returns a JSON-encoded output array reconstructed from accumulated deltas.
-// Returns (nil, false) if no content was found in deltas.
+// collectRawResponsesOutputItemsFromSSE 按到达顺序收集 SSE 流中
+// response.output_item.done 携带的原始 item。item 以 raw JSON 逐字节保留，
+// 避免经窄结构体重建时丢弃 encrypted_content/summary/opaque 等 compact
+// 专属或未来新增字段（#3777 问题 2）。若整条流没有任何 done 事件，退回
+// 收集 output_item.added 中的 compaction 类 item——compaction 结果没有
+// delta 事件，部分上游只在 added 事件中携带完整 item。
+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
+		}
+		key := strings.TrimSpace(item.Get("id").String())
+		if key == "" {
+			key = item.Raw
+		}
+		if _, dup := seen[key]; dup {
+			return
+		}
+		seen[key] = struct{}{}
+		if isResponsesCompactionItemType(item.Get("type").String()) {
+			hasCompactionItem = true
+		}
+		items = append(items, json.RawMessage(item.Raw))
+	}
+	forEachOpenAISSEDataPayload(bodyText, func(data []byte) {
+		if strings.TrimSpace(gjson.GetBytes(data, "type").String()) != "response.output_item.done" {
+			return
+		}
+		appendItem(gjson.GetBytes(data, "item"))
+	})
+	// 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
+			}
+			item := gjson.GetBytes(data, "item")
+			if !isResponsesCompactionItemType(item.Get("type").String()) {
+				return
+			}
+			appendItem(item)
+		})
+	}
+	if len(items) == 0 {
+		return nil, false
+	}
+	outputJSON, err := json.Marshal(items)
+	if err != nil {
+		return nil, false
+	}
+	return outputJSON, true
+}
+
+// isResponsesCompactionItemType reports whether the item type is the Codex
+// remote-compact result item ("compaction", upstream alias "compaction_summary").
+func isResponsesCompactionItemType(itemType string) bool {
+	switch strings.TrimSpace(itemType) {
+	case "compaction", "compaction_summary":
+		return true
+	default:
+		return false
+	}
+}
+
+// supplementCompactionItemFromSSE 保证 compact 请求的终态 output 携带
+// compaction item：终态 output 非空但缺失 compaction、而原始事件流的
+// output_item.done（或 added）中存在时（上游不一致形态），以 raw JSON 补入。
+// Codex remote compact v2 只从 output_item.done 收集 item 且要求恰好一个
+// compaction item——纯流式透传（v0.1.146）下客户端直接读事件流天然拿得到，
+// SSE→JSON 提取链路必须给出等价结果。非 compact 请求原样返回。
+func supplementCompactionItemFromSSE(c *gin.Context, finalResponse []byte, bodyText string) []byte {
+	if !isOpenAIResponsesCompactPath(c) {
+		return finalResponse
+	}
+	if len(gjson.GetBytes(finalResponse, "output").Array()) == 0 {
+		// 空 output 由 reconstructResponseOutputFromSSE 整体修补，不在此处理。
+		return finalResponse
+	}
+	if responsesOutputHasCompactionItem(finalResponse) {
+		return finalResponse
+	}
+	item, found := findRawCompactionItemFromSSE(bodyText)
+	if !found {
+		return finalResponse
+	}
+	patched, err := sjson.SetRawBytes(finalResponse, "output.-1", item)
+	if err != nil {
+		return finalResponse
+	}
+	return patched
+}
+
+// responsesOutputHasCompactionItem reports whether the response JSON already
+// carries a compaction item in its output array.
+func responsesOutputHasCompactionItem(response []byte) bool {
+	for _, item := range gjson.GetBytes(response, "output").Array() {
+		if isResponsesCompactionItemType(item.Get("type").String()) {
+			return true
+		}
+	}
+	return false
+}
+
+// findRawCompactionItemFromSSE 从原始 SSE 事件流中提取第一个 compaction 类
+// item 的 raw JSON：output_item.done 优先，output_item.added 兜底。
+func findRawCompactionItemFromSSE(bodyText string) (json.RawMessage, bool) {
+	var found json.RawMessage
+	pick := func(eventType string) {
+		forEachOpenAISSEDataPayload(bodyText, func(data []byte) {
+			if found != nil {
+				return
+			}
+			if strings.TrimSpace(gjson.GetBytes(data, "type").String()) != eventType {
+				return
+			}
+			item := gjson.GetBytes(data, "item")
+			if !item.IsObject() || !isResponsesCompactionItemType(item.Get("type").String()) {
+				return
+			}
+			found = json.RawMessage(item.Raw)
+		})
+	}
+	pick("response.output_item.done")
+	if found == nil {
+		pick("response.output_item.added")
+	}
+	return found, found != nil
+}
+
+// reconstructResponseOutputFromSSE scans raw SSE body text and returns a
+// JSON-encoded output array for a terminal event whose output is empty.
+// Raw output_item.done items are preferred: per the Responses protocol they
+// are the authoritative final form of each item. Delta accumulation only
+// covers text/function_call/reasoning content and silently drops unknown
+// item types such as compaction — Codex remote compact v2 then fails with
+// "expected exactly one compaction output item, got 0" (#3887).
+// Returns (nil, false) if nothing could be reconstructed.
 func reconstructResponseOutputFromSSE(bodyText string) ([]byte, bool) {
+	if outputJSON, ok := collectRawResponsesOutputItemsFromSSE(bodyText); ok {
+		return outputJSON, true
+	}
 	acc := apicompat.NewBufferedResponseAccumulator()
 	imageOutputs := make([]json.RawMessage, 0, 1)
 	seenImages := make(map[string]struct{})