mirror of
https://github.com/Wei-Shaw/sub2api.git
synced 2026-10-20 00:18:55 +00:00
fix(compact): SSE→JSON 保留 raw output_item.done 并为 unary 等待补下游心跳(修复 #3887)
#3887 报告 v0.1.149(已含 #3880 桥接修复)remote compact 仍失败且持续 计费。核实为 #3880 明示的两个遗留: 1. 上游对 unary compact 返回 SSE 且 compaction item 只出现在 raw response.output_item.done、终态 completed.response.output 为空 (#3777 实录形态)时,reconstructResponseOutputFromSSE 只累加 text/function_call/reasoning 三类 delta,compaction item 被丢弃, 桥接合成 0 个 output_item.done,Codex 报 "expected exactly one compaction output item, got 0" 后盲目重试,每次重试重新消耗上游 compact 配额。 2. 桥接为全缓冲写回:上游 unary 完成前(大上下文可达数分钟,且上游 处理期间不发送任何字节)下游连响应头都收不到,反向代理 (Nginx/Cloudflare Tunnel)空闲超时掐断连接同样触发盲重连 (同类问题见 #2243/#2976)。 修复: - reconstructResponseOutputFromSSE 优先以 raw JSON 逐字节收集 output_item.done item(协议上的最终完整形态),不经窄结构体, encrypted_content/summary/opaque 等 compact 专属字段全部保留; 无 done 事件时退回收集 output_item.added 中的 compaction 类 item; 两者皆无才回到原 delta 重建。path-based v1 JSON 写回同样受益。 - 新增 openAICompactSSEKeepalive:body-signal 客户端流式 compact 在 上游等待期间按 gateway.stream_keepalive_interval 向下游写 SSE 注释 行心跳(eventsource 解析层直接忽略)。首拍延迟一个间隔,快速失败 仍走 JSON+状态码;首拍后状态码固化为 200,桥接/错误链路 (writeOpenAICompactSSEBridge、errorResponse、 handleStreamingAwareError、ensureForwardErrorResponse、 writeOpenAINonStreamingProtocolError)统一降级为 response.failed 终止事件并标记 ops 流内错误。 - API-key 账号的 compact 上游请求也强制 accept: application/json (#3777 期望行为 4;透传白名单原会放行客户端的 text/event-stream)。 测试:#3777 实录形态经 handleSSEToJSON / 透传 / path-based 三条链路 的修补断言;raw-done 优先不与 delta 重复;added 回退门控;心跳提交、 提交后 2xx 续写、提交后失败降级、未提交行为不变;-race 通过。 Fixes #3887 Refs #3777 #3875 #3880
This commit is contained in:
@@ -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) {
|
||||
@@ -1943,6 +1948,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 +1985,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
|
||||
}
|
||||
@@ -2036,6 +2050,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 +2066,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)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,129 @@
|
||||
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
|
||||
stop chan struct{}
|
||||
}
|
||||
|
||||
// StartOpenAICompactSSEKeepalive 为已标记 body-signal 客户端流式的 compact
|
||||
// 请求启动下游心跳,返回幂等的停止函数。interval<=0 或请求未标记时为 no-op。
|
||||
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)
|
||||
|
||||
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
|
||||
}
|
||||
if _, err := k.writer.Write([]byte(": keepalive\n\n")); 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
|
||||
}
|
||||
@@ -0,0 +1,122 @@
|
||||
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())
|
||||
}
|
||||
@@ -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, 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) {
|
||||
if c == nil {
|
||||
return
|
||||
}
|
||||
MarkOpsStreamError(c, "upstream_error", 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": "upstream_error",
|
||||
"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,
|
||||
|
||||
@@ -258,6 +258,156 @@ 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())
|
||||
}
|
||||
|
||||
// 非 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()
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 兜底。
|
||||
|
||||
@@ -1012,6 +1012,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, 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 +1087,80 @@ 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{})
|
||||
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{}{}
|
||||
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"))
|
||||
})
|
||||
if len(items) == 0 {
|
||||
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
|
||||
}
|
||||
}
|
||||
|
||||
// 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{})
|
||||
|
||||
Reference in New Issue
Block a user