diff --git a/backend/internal/handler/openai_gateway_handler.go b/backend/internal/handler/openai_gateway_handler.go index 714e7906f3..83e644d857 100644 --- a/backend/internal/handler/openai_gateway_handler.go +++ b/backend/internal/handler/openai_gateway_handler.go @@ -407,7 +407,9 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) { // Forward request service.SetOpsLatencyMs(c, service.OpsRoutingLatencyMsKey, time.Since(routingStart).Milliseconds()) forwardStart := time.Now() - writerSizeBeforeForward := c.Writer.Size() + // 用扣除 compact 心跳字节的口径快照：心跳注释不构成语义响应， + // 不能因心跳字节变化而放弃 failover 换号（#3887）。 + writerSizeBeforeForward := service.OpenAICompactKeepaliveAdjustedWrittenSize(c) result, err := func() (*service.OpenAIForwardResult, error) { defer func() { if accountReleaseFunc != nil { @@ -441,7 +443,7 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) { } else { var failoverErr *service.UpstreamFailoverError if errors.As(err, &failoverErr) { - if c.Writer.Size() != writerSizeBeforeForward { + if service.OpenAICompactKeepaliveAdjustedWrittenSize(c) != writerSizeBeforeForward { h.handleFailoverExhausted(c, failoverErr, true) return } @@ -642,6 +644,13 @@ func (h *OpenAIGatewayHandler) logOpenAIRemoteCompactOutcome(c *gin.Context, sta if status >= 200 && status < 300 { outcome = "succeeded" } + // compact 心跳提交后失败的 wire 状态码固化为 200，真实结局以流内错误 + // 标记为准（response.failed 降级路径会 MarkOpsStreamError）。 + if outcome == "succeeded" && c != nil { + if _, hasStreamErr := service.GetOpsStreamError(c); hasStreamErr { + outcome = "failed" + } + } latencyMs := time.Since(startedAt).Milliseconds() if latencyMs < 0 { latencyMs = 0 @@ -2024,7 +2033,9 @@ func openAIForwardErrorAlreadyCommunicated(c *gin.Context, writerSizeBeforeForwa if err == nil || c == nil || c.Writer == nil { return false } - if c.Writer.Size() == writerSizeBeforeForward { + // 与快照同口径：排除 compact 心跳字节，避免"仅心跳写出"被误判为 + // 响应已写出（#3887）。 + if service.OpenAICompactKeepaliveAdjustedWrittenSize(c) == writerSizeBeforeForward { return false } @@ -2337,6 +2348,16 @@ func (h *OpenAIGatewayHandler) rejectIfCyberSessionBlocked(c *gin.Context, apiKe if !h.gatewayService.IsCyberSessionBlocked(c.Request.Context(), key) { return false } + // body-signal compact 心跳可能已把响应头提交为 200（cyber 检查在用户槽位 + // 长等待之后执行）：以 response.failed 终止事件回传；未提交时停拍后照常 + // 写 JSON（#3887）。 + if service.StopOpenAICompactSSEKeepaliveCommitted(c) { + service.MarkOpsStreamError(c, "permission_error", cyberSessionBlockedClientMsg, http.StatusForbidden) + if writeResponsesFailedSSE(c, "permission_error", cyberSessionBlockedClientMsg) { + h.enqueueCyberSessionBlockedOpsEntry(c, apiKey, model, key) + return true + } + } switch format { case cyberBlockFormatAnthropic: c.JSON(http.StatusForbidden, gin.H{"type": "error", "error": gin.H{ diff --git a/backend/internal/service/openai_compact_sse_keepalive.go b/backend/internal/service/openai_compact_sse_keepalive.go index 0695d16fb3..70ef3fc01a 100644 --- a/backend/internal/service/openai_compact_sse_keepalive.go +++ b/backend/internal/service/openai_compact_sse_keepalive.go @@ -26,11 +26,19 @@ type openAICompactSSEKeepalive struct { writer gin.ResponseWriter started bool stopped bool - stop chan struct{} + // bytes 是心跳已写出的注释字节数。心跳不构成语义响应，handler 的 + // "Forward 期间是否已写响应"判定（failover 放弃换号的依据）必须扣除 + // 这部分字节，见 OpenAICompactKeepaliveAdjustedWrittenSize。 + bytes int + stop chan struct{} } // StartOpenAICompactSSEKeepalive 为已标记 body-signal 客户端流式的 compact // 请求启动下游心跳，返回幂等的停止函数。interval<=0 或请求未标记时为 no-op。 +// +// 同时把 c.Writer 替换为 openAICompactKeepaliveWriter：请求 goroutine 的任何 +// 响应构造都会先在心跳互斥锁下停拍，未被显式拦截的写回路径（如 Forward +// 内部的本地拒绝）也不会与心跳 goroutine 产生数据竞争或字节交错。 func StartOpenAICompactSSEKeepalive(c *gin.Context, interval time.Duration) func() { if c == nil || c.Writer == nil || interval <= 0 || !openAICompactClientWantsStream(c) { return func() {} @@ -40,6 +48,7 @@ func StartOpenAICompactSSEKeepalive(c *gin.Context, interval time.Duration) func stop: make(chan struct{}), } c.Set(openAICompactSSEKeepaliveKey, k) + c.Writer = &openAICompactKeepaliveWriter{ResponseWriter: c.Writer, k: k} var reqDone <-chan struct{} if c.Request != nil { @@ -82,7 +91,9 @@ func (k *openAICompactSSEKeepalive) beat() bool { k.writer.WriteHeader(http.StatusOK) k.started = true } - if _, err := k.writer.Write([]byte(": keepalive\n\n")); err != nil { + n, err := k.writer.Write([]byte(": keepalive\n\n")) + k.bytes += n + if err != nil { k.stopped = true return false } @@ -127,3 +138,96 @@ func StopOpenAICompactSSEKeepaliveCommitted(c *gin.Context) bool { k.mu.Unlock() return committed } + +// OpenAICompactKeepaliveAdjustedWrittenSize 返回排除 compact 心跳注释字节后 +// 的响应已写字节数；无心跳的请求等价于 c.Writer.Size()。心跳字节不构成语义 +// 响应——handler 以"Forward 前后 Size 是否变化"判定是否已向客户端写出响应 +// （变化则放弃 failover 换号），该判定不得被心跳污染，否则 compact 请求 +// 一旦在上游等待期间发过心跳，上游 429/5xx 就不再换号（#3887 加固审计）。 +// 仅心跳字节时归一化为 -1（gin 的"未写出"哨兵值），与提交前的快照可比。 +func OpenAICompactKeepaliveAdjustedWrittenSize(c *gin.Context) int { + if c == nil || c.Writer == nil { + return -1 + } + value, ok := c.Get(openAICompactSSEKeepaliveKey) + if !ok { + return c.Writer.Size() + } + k, ok := value.(*openAICompactSSEKeepalive) + if !ok || k == nil { + return c.Writer.Size() + } + k.mu.Lock() + defer k.mu.Unlock() + size := k.writer.Size() + if size < 0 { + return size + } + if real := size - k.bytes; real > 0 { + return real + } + return -1 +} + +// openAICompactKeepaliveWriter 包装 gin.ResponseWriter：写侧方法先停拍心跳 +// （互斥锁下建立 happens-before），读侧方法仅加锁不停拍——热路径的状态读取 +// （如 Forward 前的 Size 快照）不能误杀心跳。心跳 goroutine 直接写内层 +// writer（k.writer），不经过本包装器，不会递归。 +type openAICompactKeepaliveWriter struct { + gin.ResponseWriter + k *openAICompactSSEKeepalive +} + +// suspend 停拍心跳；幂等。任何响应构造（含 Header 访问——写响应必先操作 +// 响应头）都视为请求侧接管 ResponseWriter。 +func (w *openAICompactKeepaliveWriter) suspend() { + w.k.Stop() +} + +func (w *openAICompactKeepaliveWriter) Header() http.Header { + w.suspend() + return w.ResponseWriter.Header() +} + +func (w *openAICompactKeepaliveWriter) Write(data []byte) (int, error) { + w.suspend() + return w.ResponseWriter.Write(data) +} + +func (w *openAICompactKeepaliveWriter) WriteString(s string) (int, error) { + w.suspend() + return w.ResponseWriter.WriteString(s) +} + +func (w *openAICompactKeepaliveWriter) WriteHeader(code int) { + w.suspend() + w.ResponseWriter.WriteHeader(code) +} + +func (w *openAICompactKeepaliveWriter) WriteHeaderNow() { + w.suspend() + w.ResponseWriter.WriteHeaderNow() +} + +func (w *openAICompactKeepaliveWriter) Flush() { + w.suspend() + w.ResponseWriter.Flush() +} + +func (w *openAICompactKeepaliveWriter) Status() int { + w.k.mu.Lock() + defer w.k.mu.Unlock() + return w.ResponseWriter.Status() +} + +func (w *openAICompactKeepaliveWriter) Size() int { + w.k.mu.Lock() + defer w.k.mu.Unlock() + return w.ResponseWriter.Size() +} + +func (w *openAICompactKeepaliveWriter) Written() bool { + w.k.mu.Lock() + defer w.k.mu.Unlock() + return w.ResponseWriter.Written() +} diff --git a/backend/internal/service/openai_compact_sse_keepalive_test.go b/backend/internal/service/openai_compact_sse_keepalive_test.go index 4abff79fc1..3b217a0718 100644 --- a/backend/internal/service/openai_compact_sse_keepalive_test.go +++ b/backend/internal/service/openai_compact_sse_keepalive_test.go @@ -120,3 +120,73 @@ func TestWriteOpenAICompactSSEBridge_BeforeKeepaliveCommitFailureKeepsJSONPath(t require.False(t, writeOpenAICompactSSEBridge(c, http.StatusBadGateway, []byte(`{"error":{"message":"fast fail"}}`))) require.Zero(t, rec.Body.Len()) } + +// 未被显式拦截的写回路径（直接操作 c.Writer）也必须与心跳互斥：包装器在 +// 请求侧任何响应构造时停拍。-race 下验证无数据竞争，且停拍后不再有心跳 +// 字节写出。 +func TestOpenAICompactKeepaliveWriter_RequestSideWriteSuspendsBeats(t *testing.T) { + c, rec := newCompactBridgeTestContext(t, true) + stop := StartOpenAICompactSSEKeepalive(c, keepaliveTestInterval) + defer stop() + waitForKeepaliveBeats() + + // 模拟未拦截路径的直接写回（如 Forward 内部本地拒绝的 c.JSON）。 + _, err := c.Writer.Write([]byte(`{"error":"local reject"}`)) + require.NoError(t, err) + + lenAfterWrite := rec.Body.Len() + waitForKeepaliveBeats() + require.Equal(t, lenAfterWrite, rec.Body.Len(), "请求侧写回后心跳必须停止") + require.Contains(t, rec.Body.String(), ": keepalive\n\n") + require.Contains(t, rec.Body.String(), `{"error":"local reject"}`) +} + +// fast policy block 在心跳提交后必须降级为 response.failed 终止事件。 +func TestWriteOpenAIFastPolicyBlockedResponse_AfterKeepaliveCommit(t *testing.T) { + c, rec := newCompactBridgeTestContext(t, true) + stop := StartOpenAICompactSSEKeepalive(c, keepaliveTestInterval) + defer stop() + waitForKeepaliveBeats() + + writeOpenAIFastPolicyBlockedResponse(c, &OpenAIFastBlockedError{Message: "tier blocked"}) + + require.Equal(t, http.StatusOK, rec.Code) + events := parseCompactBridgeSSE(t, stripKeepaliveComments(rec.Body.String())) + require.Len(t, events, 1) + require.Equal(t, "response.failed", events[0][0]) + require.Equal(t, "permission_error", gjson.Get(events[0][1], "response.error.code").String()) + require.Contains(t, gjson.Get(events[0][1], "response.error.message").String(), "tier blocked") +} + +// failover"是否已写响应"判定的口径：心跳字节必须被排除，否则 compact 在 +// 上游等待期间发过心跳后，可换号的 failover 会被误判放弃；真实响应字节 +// 写出后口径必须变化。 +func TestOpenAICompactKeepaliveAdjustedWrittenSize_ExcludesHeartbeatBytes(t *testing.T) { + c, rec := newCompactBridgeTestContext(t, true) + // 无心跳的请求：等价于 c.Writer.Size()。 + require.Equal(t, c.Writer.Size(), OpenAICompactKeepaliveAdjustedWrittenSize(c)) + + stop := StartOpenAICompactSSEKeepalive(c, keepaliveTestInterval) + defer stop() + before := OpenAICompactKeepaliveAdjustedWrittenSize(c) + waitForKeepaliveBeats() + require.Equal(t, before, OpenAICompactKeepaliveAdjustedWrittenSize(c), "仅心跳字节不得改变判定口径") + + // 真实响应字节写出（经包装器，先停拍再写）后口径必须变化。 + _, err := c.Writer.Write([]byte("real-bytes")) + require.NoError(t, err) + require.Equal(t, len("real-bytes"), OpenAICompactKeepaliveAdjustedWrittenSize(c)) + require.Contains(t, rec.Body.String(), ": keepalive\n\n") +} + +// fast policy block 在心跳未提交时保持 403 JSON 原语义。 +func TestWriteOpenAIFastPolicyBlockedResponse_BeforeKeepaliveCommit(t *testing.T) { + c, rec := newCompactBridgeTestContext(t, true) + stop := StartOpenAICompactSSEKeepalive(c, time.Hour) + defer stop() + + writeOpenAIFastPolicyBlockedResponse(c, &OpenAIFastBlockedError{Message: "tier blocked"}) + + require.Equal(t, http.StatusForbidden, rec.Code) + require.Equal(t, "permission_error", gjson.Get(rec.Body.String(), "error.type").String()) +} diff --git a/backend/internal/service/openai_compact_stream_bridge.go b/backend/internal/service/openai_compact_stream_bridge.go index 98240f4229..bcac8304c2 100644 --- a/backend/internal/service/openai_compact_stream_bridge.go +++ b/backend/internal/service/openai_compact_stream_bridge.go @@ -100,18 +100,18 @@ func writeOpenAICompactSSEFailure(c *gin.Context, statusCode int, errorBody []by if message == "" { message = "Upstream compact request failed with HTTP " + strconv.Itoa(statusCode) } - writeOpenAICompactSSEFailureMessage(c, statusCode, message) + writeOpenAICompactSSEFailureMessage(c, statusCode, "upstream_error", message) } // writeOpenAICompactSSEFailureMessage 写出 response.failed 终止事件。Codex 对 // 流式 Responses 请求把 response.failed 作为合法终止事件处理（普通 error 帧 // 不被识别，会退化为 "stream closed before response.completed" 盲重连）。 // 同时标记流内错误，保证挂在 200 流上的失败仍进入 ops 错误看板。 -func writeOpenAICompactSSEFailureMessage(c *gin.Context, statusCode int, message string) { +func writeOpenAICompactSSEFailureMessage(c *gin.Context, statusCode int, errType, message string) { if c == nil { return } - MarkOpsStreamError(c, "upstream_error", message, statusCode) + MarkOpsStreamError(c, errType, message, statusCode) payload, err := json.Marshal(map[string]any{ "type": "response.failed", "response": map[string]any{ @@ -120,7 +120,7 @@ func writeOpenAICompactSSEFailureMessage(c *gin.Context, statusCode int, message "status": "failed", "output": []any{}, "error": map[string]any{ - "code": "upstream_error", + "code": errType, "message": message, }, }, diff --git a/backend/internal/service/openai_compact_stream_bridge_test.go b/backend/internal/service/openai_compact_stream_bridge_test.go index 1f15084fd9..32b86380a9 100644 --- a/backend/internal/service/openai_compact_stream_bridge_test.go +++ b/backend/internal/service/openai_compact_stream_bridge_test.go @@ -392,6 +392,36 @@ func TestReconstructResponseOutputFromSSE_CompactionAddedFallback(t *testing.T) require.Equal(t, "added-only", items[0].Get("encrypted_content").String()) } +// 混合形态：其他 item 有 done、compaction 只在 added 中——compaction 必须 +// 被补入；done 已含 compaction 时 added 不得重复计入。 +func TestReconstructResponseOutputFromSSE_MixedDoneAndCompactionAdded(t *testing.T) { + bodyText := strings.Join([]string{ + `data: {"type":"response.output_item.added","output_index":0,"item":{"id":"cmp_mixed","type":"compaction","encrypted_content":"mixed"}}`, + `data: {"type":"response.output_item.done","output_index":1,"item":{"id":"msg_1","type":"message","content":[{"type":"output_text","text":"hi"}]}}`, + `data: {"type":"response.completed","response":{"id":"resp_1","output":[]}}`, + }, "\n") + + outputJSON, ok := reconstructResponseOutputFromSSE(bodyText) + require.True(t, ok) + items := gjson.ParseBytes(outputJSON).Array() + require.Len(t, items, 2) + require.Equal(t, "msg_1", items[0].Get("id").String()) + require.Equal(t, "cmp_mixed", items[1].Get("id").String()) + + // done 已含 compaction：added 中的同一 item（无 id 可去重的最坏情况用 + // 不同 raw 表达）不得再收集，Codex 要求恰好一个 compaction item。 + bodyText = strings.Join([]string{ + `data: {"type":"response.output_item.added","output_index":0,"item":{"type":"compaction","status":"in_progress"}}`, + `data: {"type":"response.output_item.done","output_index":0,"item":{"type":"compaction","status":"completed","encrypted_content":"final"}}`, + `data: {"type":"response.completed","response":{"id":"resp_1","output":[]}}`, + }, "\n") + outputJSON, ok = reconstructResponseOutputFromSSE(bodyText) + require.True(t, ok) + items = gjson.ParseBytes(outputJSON).Array() + require.Len(t, items, 1) + require.Equal(t, "final", items[0].Get("encrypted_content").String()) +} + // 非 compaction 的 output_item.added 不参与回退收集（added 阶段的 message // 通常是空壳），仍走 delta 重建。 func TestReconstructResponseOutputFromSSE_NonCompactionAddedStillUsesDeltas(t *testing.T) { 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 b4acfbfb35..9c9ea0c1a3 100644 --- a/backend/internal/service/openai_gateway_response_handling.go +++ b/backend/internal/service/openai_gateway_response_handling.go @@ -1015,7 +1015,7 @@ func (s *OpenAIGatewayService) writeOpenAINonStreamingProtocolError(resp *http.R // body-signal compact 心跳可能已把响应头提交为 200，此时只能以 // response.failed 终止事件回传错误，不能再写 JSON+状态码。 if openAICompactClientWantsStream(c) && StopOpenAICompactSSEKeepaliveCommitted(c) { - writeOpenAICompactSSEFailureMessage(c, http.StatusBadGateway, message) + writeOpenAICompactSSEFailureMessage(c, http.StatusBadGateway, "upstream_error", message) return fmt.Errorf("non-streaming openai protocol error: %s", message) } responseheaders.WriteFilteredHeaders(c.Writer.Header(), resp.Header, s.responseHeaderFilter) @@ -1096,6 +1096,7 @@ func responsesStreamEventMayContributeToOutput(eventType string) bool { func collectRawResponsesOutputItemsFromSSE(bodyText string) ([]byte, bool) { var items []json.RawMessage seen := make(map[string]struct{}) + hasCompactionItem := false appendItem := func(item gjson.Result) { if !item.Exists() || !item.IsObject() { return @@ -1108,6 +1109,9 @@ func collectRawResponsesOutputItemsFromSSE(bodyText string) ([]byte, bool) { return } seen[key] = struct{}{} + if isResponsesCompactionItemType(item.Get("type").String()) { + hasCompactionItem = true + } items = append(items, json.RawMessage(item.Raw)) } forEachOpenAISSEDataPayload(bodyText, func(data []byte) { @@ -1116,7 +1120,10 @@ func collectRawResponsesOutputItemsFromSSE(bodyText string) ([]byte, bool) { } appendItem(gjson.GetBytes(data, "item")) }) - if len(items) == 0 { + // done 事件未携带 compaction item 时再看 added：覆盖"其他 item 有 done、 + // compaction 只在 added 中"的混合形态；done 已含 compaction 时跳过， + // 避免同一 item 在无 id 可去重时被收集两份（Codex 要求恰好一个）。 + if !hasCompactionItem { forEachOpenAISSEDataPayload(bodyText, func(data []byte) { if strings.TrimSpace(gjson.GetBytes(data, "type").String()) != "response.output_item.added" { return