diff --git a/backend/internal/handler/openai_gateway_handler.go b/backend/internal/handler/openai_gateway_handler.go index feba261103..83e644d857 100644 --- a/backend/internal/handler/openai_gateway_handler.go +++ b/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{ diff --git a/backend/internal/service/openai_compact_sse_keepalive.go b/backend/internal/service/openai_compact_sse_keepalive.go new file mode 100644 index 0000000000..70ef3fc01a --- /dev/null +++ b/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() +} diff --git a/backend/internal/service/openai_compact_sse_keepalive_test.go b/backend/internal/service/openai_compact_sse_keepalive_test.go new file mode 100644 index 0000000000..3b217a0718 --- /dev/null +++ b/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()) +} diff --git a/backend/internal/service/openai_compact_stream_bridge.go b/backend/internal/service/openai_compact_stream_bridge.go index 33ca543850..bcac8304c2 100644 --- a/backend/internal/service/openai_compact_stream_bridge.go +++ b/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, diff --git a/backend/internal/service/openai_compact_stream_bridge_test.go b/backend/internal/service/openai_compact_stream_bridge_test.go index 2eb5ba80ee..bf3416192e 100644 --- a/backend/internal/service/openai_compact_stream_bridge_test.go +++ b/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() diff --git a/backend/internal/service/openai_gateway_forward.go b/backend/internal/service/openai_gateway_forward.go index 6fdabeb90f..3bbe1393f9 100644 --- a/backend/internal/service/openai_gateway_forward.go +++ b/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 diff --git a/backend/internal/service/openai_gateway_passthrough.go b/backend/internal/service/openai_gateway_passthrough.go index 3a56ab5c61..12eb248fcf 100644 --- a/backend/internal/service/openai_gateway_passthrough.go +++ b/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) diff --git a/backend/internal/service/openai_gateway_request_body.go b/backend/internal/service/openai_gateway_request_body.go index b48b7510eb..4824ce11e2 100644 --- a/backend/internal/service/openai_gateway_request_body.go +++ b/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", diff --git a/backend/internal/service/openai_gateway_response_handling.go b/backend/internal/service/openai_gateway_response_handling.go index cb4b780cdf..45c948f56f 100644 --- a/backend/internal/service/openai_gateway_response_handling.go +++ b/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{})