From 3ed7b3f357204d4443012d07d468811bafd37f7f Mon Sep 17 00:00:00 2001 From: slZhong <1542123803@qq.com> Date: Tue, 11 Aug 2026 16:30:03 +0800 Subject: [PATCH 1/8] fix(relay): handle Codex response terminal events --- relay/channel/openai/relay_responses.go | 83 ++++++++++---- relay/channel/openai/relay_responses_test.go | 110 +++++++++++++++++++ 2 files changed, 172 insertions(+), 21 deletions(-) diff --git a/relay/channel/openai/relay_responses.go b/relay/channel/openai/relay_responses.go index 3a8f2128c8d..ee3d167b2a6 100644 --- a/relay/channel/openai/relay_responses.go +++ b/relay/channel/openai/relay_responses.go @@ -1,6 +1,7 @@ package openai import ( + "errors" "fmt" "io" "net/http" @@ -83,6 +84,7 @@ func OaiResponsesStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp var usage = &dto.Usage{} var responseTextBuilder strings.Builder + var terminalErr *types.NewAPIError helper.StreamScannerHandler(c, resp, info, func(data string, sr *helper.StreamResult) { @@ -96,30 +98,22 @@ func OaiResponsesStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp if streamResponse.Response != nil && streamResponse.Response.ID != "" { c.Set(common.UpstreamResponseIdKey, streamResponse.Response.ID) } + if info.ChannelType == constant.ChannelTypeCodex && streamResponse.Type == "response.failed" { + responseWritten := c.Writer.Written() + if responseWritten { + sendResponsesStreamData(c, streamResponse, data) + } + terminalErr = newCodexResponsesFailedError(streamResponse.Response, responseWritten) + sr.Stop(terminalErr) + return + } sendResponsesStreamData(c, streamResponse, data) switch streamResponse.Type { case "response.completed", "response.incomplete": - if streamResponse.Response != nil { - if streamResponse.Response.Usage != nil { - if streamResponse.Response.Usage.InputTokens != 0 { - usage.PromptTokens = streamResponse.Response.Usage.InputTokens - } - if streamResponse.Response.Usage.OutputTokens != 0 { - usage.CompletionTokens = streamResponse.Response.Usage.OutputTokens - } - if streamResponse.Response.Usage.TotalTokens != 0 { - usage.TotalTokens = streamResponse.Response.Usage.TotalTokens - } - if streamResponse.Response.Usage.InputTokensDetails != nil { - usage.PromptTokensDetails.CachedTokens = streamResponse.Response.Usage.InputTokensDetails.CachedTokens - usage.PromptTokensDetails.CacheWriteTokens = streamResponse.Response.Usage.InputTokensDetails.CacheWriteTokens - } - } - if streamResponse.Response.HasImageGenerationCall() { - c.Set("image_generation_call", true) - c.Set("image_generation_call_quality", streamResponse.Response.GetQuality()) - c.Set("image_generation_call_size", streamResponse.Response.GetSize()) - } + applyResponsesTerminalUsage(c, usage, streamResponse.Response) + case "response.done": + if info.ChannelType == constant.ChannelTypeCodex { + applyResponsesTerminalUsage(c, usage, streamResponse.Response) } case "response.output_text.delta": // 处理输出文本 @@ -138,6 +132,9 @@ func OaiResponsesStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp } } }) + if terminalErr != nil { + return usage, terminalErr + } // FRT watchdog: upstream accepted the request but never produced a data // event within constant.StreamingFirstResponseTimeout seconds. Surface as @@ -173,3 +170,47 @@ func OaiResponsesStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp return usage, nil } + +func applyResponsesTerminalUsage(c *gin.Context, usage *dto.Usage, response *dto.OpenAIResponsesResponse) { + if usage == nil || response == nil { + return + } + if response.Usage != nil { + if response.Usage.InputTokens != 0 { + usage.PromptTokens = response.Usage.InputTokens + } + if response.Usage.OutputTokens != 0 { + usage.CompletionTokens = response.Usage.OutputTokens + } + if response.Usage.TotalTokens != 0 { + usage.TotalTokens = response.Usage.TotalTokens + } + if response.Usage.InputTokensDetails != nil { + usage.PromptTokensDetails.CachedTokens = response.Usage.InputTokensDetails.CachedTokens + usage.PromptTokensDetails.CacheWriteTokens = response.Usage.InputTokensDetails.CacheWriteTokens + } + } + if response.HasImageGenerationCall() { + c.Set("image_generation_call", true) + c.Set("image_generation_call_quality", response.GetQuality()) + c.Set("image_generation_call_size", response.GetSize()) + } +} + +func newCodexResponsesFailedError(response *dto.OpenAIResponsesResponse, skipRetry bool) *types.NewAPIError { + options := make([]types.NewAPIErrorOptions, 0, 1) + if skipRetry { + options = append(options, types.ErrOptionWithSkipRetry()) + } + if response != nil { + if openAIError := response.GetOpenAIError(); openAIError != nil && openAIError.Message != "" { + return types.WithOpenAIError(*openAIError, http.StatusInternalServerError, options...) + } + } + return types.NewOpenAIError( + errors.New("codex upstream response failed"), + types.ErrorCodeBadResponse, + http.StatusInternalServerError, + options..., + ) +} diff --git a/relay/channel/openai/relay_responses_test.go b/relay/channel/openai/relay_responses_test.go index 7ee0ada65f7..8a50240fe8e 100644 --- a/relay/channel/openai/relay_responses_test.go +++ b/relay/channel/openai/relay_responses_test.go @@ -10,6 +10,7 @@ import ( "github.com/QuantumNous/new-api/constant" "github.com/QuantumNous/new-api/dto" relaycommon "github.com/QuantumNous/new-api/relay/common" + "github.com/QuantumNous/new-api/types" "github.com/gin-gonic/gin" ) @@ -64,3 +65,112 @@ func TestOaiResponsesStreamHandlerCapturesIncompleteUsage(t *testing.T) { t.Fatalf("incomplete event was not forwarded: %s", recorder.Body.String()) } } + +func TestOaiResponsesStreamHandlerCodexCapturesDoneUsage(t *testing.T) { + upstream := strings.Join([]string{ + "event: response.done", + `data: {"type":"response.done","response":{"id":"resp_done","status":"completed","usage":{"input_tokens":71,"output_tokens":19,"total_tokens":90,"input_tokens_details":{"cached_tokens":7,"cache_write_tokens":2}}}}`, + "", + }, "\n") + + recorder, ctx, info, resp := newResponsesStreamTest(t, upstream, constant.ChannelTypeCodex) + usage, apiErr := OaiResponsesStreamHandler(ctx, info, resp) + if apiErr != nil { + t.Fatalf("handle Codex response.done: %v", apiErr) + } + if usage.PromptTokens != 71 || usage.CompletionTokens != 19 || usage.TotalTokens != 90 { + t.Fatalf("done usage = %#v", *usage) + } + if usage.PromptTokensDetails.CachedTokens != 7 || usage.PromptTokensDetails.CacheWriteTokens != 2 { + t.Fatalf("done token details = %#v", usage.PromptTokensDetails) + } + if !strings.Contains(recorder.Body.String(), "response.done") { + t.Fatalf("done event was not forwarded: %s", recorder.Body.String()) + } +} + +func TestOaiResponsesStreamHandlerNonCodexKeepsDoneBehavior(t *testing.T) { + upstream := strings.Join([]string{ + "event: response.done", + `data: {"type":"response.done","response":{"id":"resp_done","status":"completed","usage":{"input_tokens":71,"output_tokens":19,"total_tokens":90}}}`, + "", + }, "\n") + + _, ctx, info, resp := newResponsesStreamTest(t, upstream, constant.ChannelTypeOpenAI) + usage, apiErr := OaiResponsesStreamHandler(ctx, info, resp) + if apiErr != nil { + t.Fatalf("handle non-Codex response.done: %v", apiErr) + } + if usage.PromptTokens != 0 || usage.CompletionTokens != 0 || usage.TotalTokens != 0 { + t.Fatalf("non-Codex behavior changed, usage = %#v", *usage) + } +} + +func TestOaiResponsesStreamHandlerCodexFailedBeforeCommitIsRetryable(t *testing.T) { + upstream := strings.Join([]string{ + "event: response.failed", + `data: {"type":"response.failed","response":{"id":"resp_failed","status":"failed","error":{"code":"server_error","message":"upstream blew up"}}}`, + "", + }, "\n") + + recorder, ctx, info, resp := newResponsesStreamTest(t, upstream, constant.ChannelTypeCodex) + _, apiErr := OaiResponsesStreamHandler(ctx, info, resp) + if apiErr == nil { + t.Fatal("expected Codex response.failed error") + } + if types.IsSkipRetryError(apiErr) { + t.Fatal("failure before response commit must remain retryable") + } + if recorder.Body.Len() != 0 || ctx.Writer.Written() { + t.Fatalf("failed event committed before retry: %s", recorder.Body.String()) + } +} + +func TestOaiResponsesStreamHandlerCodexFailedAfterCommitSkipsRetry(t *testing.T) { + upstream := strings.Join([]string{ + "event: response.created", + `data: {"type":"response.created","response":{"id":"resp_failed","status":"in_progress"}}`, + "", + "event: response.failed", + `data: {"type":"response.failed","response":{"id":"resp_failed","status":"failed","error":{"code":"server_error","message":"upstream blew up"}}}`, + "", + }, "\n") + + recorder, ctx, info, resp := newResponsesStreamTest(t, upstream, constant.ChannelTypeCodex) + _, apiErr := OaiResponsesStreamHandler(ctx, info, resp) + if apiErr == nil { + t.Fatal("expected Codex response.failed error") + } + if !types.IsSkipRetryError(apiErr) { + t.Fatal("failure after response commit must skip retry") + } + if !strings.Contains(recorder.Body.String(), "response.created") || !strings.Contains(recorder.Body.String(), "response.failed") { + t.Fatalf("committed Codex events were not forwarded: %s", recorder.Body.String()) + } +} + +func newResponsesStreamTest(t *testing.T, upstream string, channelType int) (*httptest.ResponseRecorder, *gin.Context, *relaycommon.RelayInfo, *http.Response) { + t.Helper() + previousTimeout := constant.StreamingTimeout + if previousTimeout <= 0 { + constant.StreamingTimeout = 30 + } + t.Cleanup(func() { constant.StreamingTimeout = previousTimeout }) + + recorder := httptest.NewRecorder() + ctx, _ := gin.CreateTestContext(recorder) + ctx.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil) + info := &relaycommon.RelayInfo{ + IsStream: true, + ChannelMeta: &relaycommon.ChannelMeta{ + ChannelType: channelType, + UpstreamModelName: "gpt-5.6-terra", + }, + } + resp := &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}}, + Body: io.NopCloser(strings.NewReader(upstream)), + } + return recorder, ctx, info, resp +} From 702918e4f9a909f450e83b23ea9deed9003cee07 Mon Sep 17 00:00:00 2001 From: slZhong <1542123803@qq.com> Date: Tue, 11 Aug 2026 17:09:36 +0800 Subject: [PATCH 2/8] fix(relay): reset Codex stream state before retry --- relay/channel/openai/relay_responses.go | 33 +++++++++ relay/channel/openai/relay_responses_test.go | 70 ++++++++++++++++++++ relay/common/relay_info.go | 12 ++++ relay/common/relay_info_test.go | 24 +++++++ 4 files changed, 139 insertions(+) diff --git a/relay/channel/openai/relay_responses.go b/relay/channel/openai/relay_responses.go index ee3d167b2a6..27feb683839 100644 --- a/relay/channel/openai/relay_responses.go +++ b/relay/channel/openai/relay_responses.go @@ -85,6 +85,15 @@ func OaiResponsesStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp var usage = &dto.Usage{} var responseTextBuilder strings.Builder var terminalErr *types.NewAPIError + var retryableTerminalFailure bool + + var responseHeaderSnapshot http.Header + var eventStreamHeadersValue any + var hadEventStreamHeaders bool + if info.ChannelType == constant.ChannelTypeCodex { + responseHeaderSnapshot = c.Writer.Header().Clone() + eventStreamHeadersValue, hadEventStreamHeaders = c.Get("event_stream_headers_set") + } helper.StreamScannerHandler(c, resp, info, func(data string, sr *helper.StreamResult) { @@ -103,6 +112,7 @@ func OaiResponsesStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp if responseWritten { sendResponsesStreamData(c, streamResponse, data) } + retryableTerminalFailure = !responseWritten terminalErr = newCodexResponsesFailedError(streamResponse.Response, responseWritten) sr.Stop(terminalErr) return @@ -133,6 +143,9 @@ func OaiResponsesStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp } }) if terminalErr != nil { + if retryableTerminalFailure { + restoreResponsesStreamAttemptState(c, info, responseHeaderSnapshot, eventStreamHeadersValue, hadEventStreamHeaders) + } return usage, terminalErr } @@ -171,6 +184,26 @@ func OaiResponsesStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp return usage, nil } +func restoreResponsesStreamAttemptState( + c *gin.Context, + info *relaycommon.RelayInfo, + headerSnapshot http.Header, + eventStreamHeadersValue any, + hadEventStreamHeaders bool, +) { + header := c.Writer.Header() + clear(header) + for key, values := range headerSnapshot { + header[key] = append([]string(nil), values...) + } + if hadEventStreamHeaders { + c.Set("event_stream_headers_set", eventStreamHeadersValue) + } else if c.Keys != nil { + delete(c.Keys, "event_stream_headers_set") + } + info.ResetStreamResponseStateForRetry() +} + func applyResponsesTerminalUsage(c *gin.Context, usage *dto.Usage, response *dto.OpenAIResponsesResponse) { if usage == nil || response == nil { return diff --git a/relay/channel/openai/relay_responses_test.go b/relay/channel/openai/relay_responses_test.go index 8a50240fe8e..566b67d81ef 100644 --- a/relay/channel/openai/relay_responses_test.go +++ b/relay/channel/openai/relay_responses_test.go @@ -114,6 +114,10 @@ func TestOaiResponsesStreamHandlerCodexFailedBeforeCommitIsRetryable(t *testing. }, "\n") recorder, ctx, info, resp := newResponsesStreamTest(t, upstream, constant.ChannelTypeCodex) + info.ApiType = constant.APITypeCodex + ctx.Header("X-Existing", "keep") + resp.Header.Set("X-Reasoning-Included", "true") + resp.Header.Set("X-Codex-Turn-State", "state-from-failed-attempt") _, apiErr := OaiResponsesStreamHandler(ctx, info, resp) if apiErr == nil { t.Fatal("expected Codex response.failed error") @@ -124,6 +128,32 @@ func TestOaiResponsesStreamHandlerCodexFailedBeforeCommitIsRetryable(t *testing. if recorder.Body.Len() != 0 || ctx.Writer.Written() { t.Fatalf("failed event committed before retry: %s", recorder.Body.String()) } + if got := recorder.Header().Get("Content-Type"); got != "" { + t.Fatalf("retryable failure retained SSE content type %q", got) + } + if got := recorder.Header().Get("Transfer-Encoding"); got != "" { + t.Fatalf("retryable failure retained transfer encoding %q", got) + } + if got := recorder.Header().Get("X-Reasoning-Included"); got != "" { + t.Fatalf("retryable failure retained Codex response header %q", got) + } + if got := recorder.Header().Get("X-Codex-Turn-State"); got != "" { + t.Fatalf("retryable failure retained Codex turn state %q", got) + } + if got := recorder.Header().Get("X-Existing"); got != "keep" { + t.Fatalf("pre-existing response header = %q", got) + } + if _, exists := ctx.Get("event_stream_headers_set"); exists { + t.Fatal("retryable failure retained event_stream_headers_set") + } + if info.HasSendResponse() || info.ReceivedResponseCount != 0 { + t.Fatalf("retryable failure retained first-response state: sent=%v received=%d", info.HasSendResponse(), info.ReceivedResponseCount) + } + + ctx.JSON(apiErr.StatusCode, gin.H{"error": apiErr.ToOpenAIError()}) + if got := recorder.Header().Get("Content-Type"); got != "application/json; charset=utf-8" { + t.Fatalf("final JSON content type = %q", got) + } } func TestOaiResponsesStreamHandlerCodexFailedAfterCommitSkipsRetry(t *testing.T) { @@ -149,6 +179,46 @@ func TestOaiResponsesStreamHandlerCodexFailedAfterCommitSkipsRetry(t *testing.T) } } +func TestOaiResponsesStreamHandlerCodexRetryRearmsFirstResponseTracking(t *testing.T) { + failedUpstream := strings.Join([]string{ + "event: response.failed", + `data: {"type":"response.failed","response":{"id":"resp_failed","status":"failed","error":{"code":"server_error","message":"upstream blew up"}}}`, + "", + }, "\n") + + recorder, ctx, info, failedResp := newResponsesStreamTest(t, failedUpstream, constant.ChannelTypeCodex) + info.ApiType = constant.APITypeCodex + _, apiErr := OaiResponsesStreamHandler(ctx, info, failedResp) + if apiErr == nil || types.IsSkipRetryError(apiErr) { + t.Fatalf("first attempt error = %#v", apiErr) + } + + doneUpstream := strings.Join([]string{ + "event: response.done", + `data: {"type":"response.done","response":{"id":"resp_done","status":"completed","usage":{"input_tokens":8,"output_tokens":2,"total_tokens":10}}}`, + "", + }, "\n") + doneResp := &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}}, + Body: io.NopCloser(strings.NewReader(doneUpstream)), + } + + usage, apiErr := OaiResponsesStreamHandler(ctx, info, doneResp) + if apiErr != nil { + t.Fatalf("retry attempt: %v", apiErr) + } + if !info.HasSendResponse() || info.ReceivedResponseCount != 1 { + t.Fatalf("retry first-response state: sent=%v received=%d", info.HasSendResponse(), info.ReceivedResponseCount) + } + if usage.PromptTokens != 8 || usage.CompletionTokens != 2 { + t.Fatalf("retry usage = %#v", *usage) + } + if got := recorder.Header().Get("Content-Type"); got != "text/event-stream" { + t.Fatalf("retry content type = %q", got) + } +} + func newResponsesStreamTest(t *testing.T, upstream string, channelType int) (*httptest.ResponseRecorder, *gin.Context, *relaycommon.RelayInfo, *http.Response) { t.Helper() previousTimeout := constant.StreamingTimeout diff --git a/relay/common/relay_info.go b/relay/common/relay_info.go index 3c05a468be2..09220f42758 100644 --- a/relay/common/relay_info.go +++ b/relay/common/relay_info.go @@ -671,6 +671,18 @@ func (info *RelayInfo) SetFirstResponseTime() { } } +// ResetStreamResponseStateForRetry restores the response-observation fields +// after an upstream attempt produced no downstream-visible response. Callers +// must only use it before retrying an uncommitted response. +func (info *RelayInfo) ResetStreamResponseStateForRetry() { + if info == nil { + return + } + info.FirstResponseTime = info.StartTime.Add(-time.Second) + info.isFirstResponse = true + info.ReceivedResponseCount = 0 +} + func (info *RelayInfo) HasSendResponse() bool { return info.FirstResponseTime.After(info.StartTime) } diff --git a/relay/common/relay_info_test.go b/relay/common/relay_info_test.go index e53ec804ca0..ed8242dbf3a 100644 --- a/relay/common/relay_info_test.go +++ b/relay/common/relay_info_test.go @@ -2,11 +2,35 @@ package common import ( "testing" + "time" "github.com/QuantumNous/new-api/types" "github.com/stretchr/testify/require" ) +func TestResetStreamResponseStateForRetry(t *testing.T) { + start := time.Now().Add(-time.Second) + info := &RelayInfo{ + StartTime: start, + FirstResponseTime: time.Now(), + isFirstResponse: false, + ReceivedResponseCount: 3, + } + + info.ResetStreamResponseStateForRetry() + + if info.HasSendResponse() { + t.Fatal("reset stream state still reports a sent response") + } + if info.ReceivedResponseCount != 0 { + t.Fatalf("received response count = %d", info.ReceivedResponseCount) + } + info.SetFirstResponseTime() + if !info.HasSendResponse() { + t.Fatal("reset stream state did not re-arm first response tracking") + } +} + func TestRelayInfoGetFinalRequestRelayFormatPrefersExplicitFinal(t *testing.T) { info := &RelayInfo{ RelayFormat: types.RelayFormatOpenAI, From 63f775a96bd422853574a27309076d1784110bb7 Mon Sep 17 00:00:00 2001 From: slZhong <1542123803@qq.com> Date: Tue, 11 Aug 2026 17:44:11 +0800 Subject: [PATCH 3/8] fix(relay): settle failed Codex partial responses --- ...ex-response-failed-error-mapping-design.md | 31 +++- relay/channel/openai/relay_responses.go | 98 ++++++++-- relay/channel/openai/relay_responses_test.go | 102 ++++++++++- relay/responses_handler.go | 16 ++ relay/responses_handler_test.go | 33 ++++ service/text_quota.go | 38 +++- service/text_quota_error_test.go | 169 ++++++++++++++++++ 7 files changed, 456 insertions(+), 31 deletions(-) create mode 100644 service/text_quota_error_test.go diff --git a/docs/superpowers/specs/2026-07-16-codex-response-failed-error-mapping-design.md b/docs/superpowers/specs/2026-07-16-codex-response-failed-error-mapping-design.md index ed3448a09b9..94f727d8779 100644 --- a/docs/superpowers/specs/2026-07-16-codex-response-failed-error-mapping-design.md +++ b/docs/superpowers/specs/2026-07-16-codex-response-failed-error-mapping-design.md @@ -28,13 +28,14 @@ LiteLLM recognizes an OpenAI-shaped first SSE error frame whose `error.code` is - Never emit a success finish chunk or `[DONE]` after a failed or truncated stream. - Never retry after downstream bytes have been written. - Preserve upstream HTTP non-2xx status mapping and successful stream behavior. -- Retain usage carried by failed terminal events in the adapter result while keeping the existing refund-on-error billing policy. +- Retain usage carried by failed terminal events in the adapter result. +- For Codex streams that fail after billable output has reached the client, settle the delivered usage exactly once while preserving the failed request outcome. ## Non-goals - Changing the public Responses API relay path. - Retrying a request after any downstream bytes were written. -- Charging users for failed requests; existing refund behavior remains unchanged. +- Charging pre-commit failures or failures without any billable delivered usage. - Refactoring unrelated streaming adapters or the controller retry policy beyond the response-commit invariant. ## Considered Approaches @@ -61,7 +62,7 @@ For `response.failed`: - capture `response.error.message`, falling back to `response.error.code`, then to a generic upstream-failure message; - capture `response.usage` when present; -- create a `types.NewAPIError` with HTTP 500 and the existing bad-response error category; +- map recognized Responses error codes to their HTTP semantics, with unknown/server failures falling back to HTTP 500; - do not pass the event through the normal success finalizer. An actual upstream HTTP non-2xx response continues to preserve its original status code. Transport/protocol termination without a successful terminal event uses HTTP 502. @@ -99,7 +100,7 @@ The bridge tracks whether a successful terminal event (`response.completed`, `re The upstream SSE stream is fully accumulated before any downstream body is written. - successful terminal event: build the current HTTP 200 Chat Completions JSON response; -- `response.failed`: return HTTP 500 with an OpenAI error object, never a Chat Completions assistant message; +- `response.failed`: return the mapped non-2xx status with an OpenAI error object, never a Chat Completions assistant message; - scanner error or EOF without a terminal event: return HTTP 502 with an OpenAI error object; - failed-event usage remains returned by the adapter alongside the error for accounting visibility, while the caller's existing error path refunds the request. @@ -112,12 +113,18 @@ The relay controller enforces a generic HTTP invariant: This guard is not Codex-specific: retrying or changing response format after bytes have reached a client is invalid for every streaming provider. The Codex adapter still marks late failures with `ErrOptionWithSkipRetry` as defense in depth. +### 6. Partial-failure billing + +The public Responses relay preserves the terminal error even when partial output was already written. For Codex only, if the adapter returns positive terminal or estimated usage after the downstream response is committed, `ResponsesHelper` runs the normal text quota calculation and settlement before returning the error. The success relay sample is suppressed because the controller records the failed sample. + +Pre-commit failures, non-Codex channels, and failures without positive usage retain the existing refund behavior. If actual-usage settlement fails, the billing session retains the already reserved quota so the controller's deferred refund cannot make delivered output free; the settlement failure is logged for reconciliation. The request-local `BillingSession` keeps settlement/refund idempotent, while quota mutations remain database-backed and safe across application nodes. + ## Data Flow | Upstream sequence | Downstream result | | --- | --- | | HTTP non-2xx | Same non-2xx status through existing error mapping | -| `created -> failed` | HTTP 500 OpenAI error JSON; no role/content chunk | +| `created -> failed` | Mapped non-2xx OpenAI error JSON; no role/content chunk | | `created -> text delta -> failed` | Existing role/text chunks, then SSE error frame; no finish or `[DONE]` | | `created -> EOF` | HTTP 502 before output, or SSE error after output | | `created -> completed` | Existing successful Chat Completions stream/JSON | @@ -149,18 +156,26 @@ This guard is not Codex-specific: retrying or changing response format after byt - retain role/content/finish/usage behavior. 7. Failed terminal event with usage: - returns the captured prompt/completion/total token values alongside the error. + - after committed Codex output, settles once, remains an error, and is not refunded; + - before commit, retains the existing refund path. +8. Failed terminal event without usage after text or tool-call deltas: + - estimates positive delivered output usage and follows the same post-commit settlement path. +9. Recognized `response.error.code` values: + - map to accurate 4xx/429/5xx status classes; + - unknown and transient server codes fall back to retryable 500 behavior; + - pre-commit retry remains controlled by the centralized status policy. ### API compatibility conversion -8. `response.failed` conversion: +10. `response.failed` conversion: - never emits upstream error text as `delta.content`; - retains usage in converter state when called directly. ### Relay controller -9. Retry decision after response commitment: +11. Retry decision after response commitment: - `shouldRetry` is false after any downstream write. -10. Deferred error rendering after response commitment: +12. Deferred error rendering after response commitment: - does not append a JSON error object to an existing SSE body. ## Validation diff --git a/relay/channel/openai/relay_responses.go b/relay/channel/openai/relay_responses.go index 27feb683839..c6015a94001 100644 --- a/relay/channel/openai/relay_responses.go +++ b/relay/channel/openai/relay_responses.go @@ -108,6 +108,7 @@ func OaiResponsesStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp c.Set(common.UpstreamResponseIdKey, streamResponse.Response.ID) } if info.ChannelType == constant.ChannelTypeCodex && streamResponse.Type == "response.failed" { + applyResponsesTerminalUsage(c, usage, streamResponse.Response) responseWritten := c.Writer.Written() if responseWritten { sendResponsesStreamData(c, streamResponse, data) @@ -125,8 +126,15 @@ func OaiResponsesStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp if info.ChannelType == constant.ChannelTypeCodex { applyResponsesTerminalUsage(c, usage, streamResponse.Response) } - case "response.output_text.delta": - // 处理输出文本 + case "response.output_text.delta", + "response.reasoning_summary_text.delta", + "response.reasoning_text.delta", + "response.function_call_arguments.delta", + "response.custom_tool_call_input.delta", + "response.mcp_call_arguments.delta", + "response.code_interpreter_call_code.delta": + // Preserve a text-equivalent fallback when a failed terminal event + // omits usage, including tool-only and reasoning-only output. responseTextBuilder.WriteString(streamResponse.Delta) case dto.ResponsesOutputTypeItemDone: // 函数调用处理 @@ -145,6 +153,8 @@ func OaiResponsesStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp if terminalErr != nil { if retryableTerminalFailure { restoreResponsesStreamAttemptState(c, info, responseHeaderSnapshot, eventStreamHeadersValue, hadEventStreamHeaders) + } else { + finalizeResponsesUsage(usage, &responseTextBuilder, info) } return usage, terminalErr } @@ -165,23 +175,19 @@ func OaiResponsesStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp ) } - if usage.CompletionTokens == 0 { - // 计算输出文本的 token 数量 - tempStr := responseTextBuilder.String() - if len(tempStr) > 0 { - // 非正常结束，使用输出文本的 token 数量 - completionTokens := service.CountTextToken(tempStr, info.UpstreamModelName) - usage.CompletionTokens = completionTokens - } - } + finalizeResponsesUsage(usage, &responseTextBuilder, info) + return usage, nil +} + +func finalizeResponsesUsage(usage *dto.Usage, responseTextBuilder *strings.Builder, info *relaycommon.RelayInfo) { + if usage.CompletionTokens == 0 && responseTextBuilder.Len() > 0 { + usage.CompletionTokens = service.CountTextToken(responseTextBuilder.String(), info.UpstreamModelName) + } if usage.PromptTokens == 0 && usage.CompletionTokens != 0 { usage.PromptTokens = info.GetEstimatePromptTokens() } - usage.TotalTokens = usage.PromptTokens + usage.CompletionTokens - - return usage, nil } func restoreResponsesStreamAttemptState( @@ -237,7 +243,7 @@ func newCodexResponsesFailedError(response *dto.OpenAIResponsesResponse, skipRet } if response != nil { if openAIError := response.GetOpenAIError(); openAIError != nil && openAIError.Message != "" { - return types.WithOpenAIError(*openAIError, http.StatusInternalServerError, options...) + return types.WithOpenAIError(*openAIError, codexResponsesFailedStatus(openAIError), options...) } } return types.NewOpenAIError( @@ -247,3 +253,65 @@ func newCodexResponsesFailedError(response *dto.OpenAIResponsesResponse, skipRet options..., ) } + +func codexResponsesFailedStatus(openAIError *types.OpenAIError) int { + if openAIError == nil { + return http.StatusInternalServerError + } + errorType := strings.ToLower(strings.TrimSpace(openAIError.Type)) + errorCode := strings.ToLower(strings.TrimSpace(common.Interface2String(openAIError.Code))) + if statusCode := codexResponsesFailedIdentifierStatus(errorCode); statusCode != 0 { + return statusCode + } + if statusCode := codexResponsesFailedIdentifierStatus(errorType); statusCode != 0 { + return statusCode + } + return http.StatusInternalServerError +} + +func codexResponsesFailedIdentifierStatus(identifier string) int { + switch identifier { + case + "insufficient_quota", + "credit_balance_exhausted", + "organization_spend_limit_exceeded", + "project_spend_limit_exceeded", + "organization_usage_limit_exceeded", + "project_usage_limit_exceeded", + "rate_limit_error", + "rate_limit_exceeded": + return http.StatusTooManyRequests + case "permission_error", "permission_denied", "unsupported_country_region_territory": + return http.StatusForbidden + case "authentication_error", "invalid_api_key": + return http.StatusUnauthorized + case "invalid_request_error", + "invalid_request", + "bad_request", + "invalid_prompt", + "content_policy_violation", + "data_residency_mismatch", + "bio_policy", + "invalid_image", + "invalid_image_format", + "invalid_base64_image", + "invalid_image_url", + "image_too_large", + "image_too_small", + "image_parse_error", + "image_content_policy_violation", + "invalid_image_mode", + "image_file_too_large", + "unsupported_image_media_type", + "empty_image_file", + "failed_to_download_image", + "image_file_not_found": + return http.StatusBadRequest + case "overloaded_error", "overloaded", "service_unavailable": + return http.StatusServiceUnavailable + case "server_error": + return http.StatusInternalServerError + default: + return 0 + } +} diff --git a/relay/channel/openai/relay_responses_test.go b/relay/channel/openai/relay_responses_test.go index 566b67d81ef..df3560f10d2 100644 --- a/relay/channel/openai/relay_responses_test.go +++ b/relay/channel/openai/relay_responses_test.go @@ -10,6 +10,7 @@ import ( "github.com/QuantumNous/new-api/constant" "github.com/QuantumNous/new-api/dto" relaycommon "github.com/QuantumNous/new-api/relay/common" + "github.com/QuantumNous/new-api/service" "github.com/QuantumNous/new-api/types" "github.com/gin-gonic/gin" @@ -161,13 +162,16 @@ func TestOaiResponsesStreamHandlerCodexFailedAfterCommitSkipsRetry(t *testing.T) "event: response.created", `data: {"type":"response.created","response":{"id":"resp_failed","status":"in_progress"}}`, "", + "event: response.output_text.delta", + `data: {"type":"response.output_text.delta","delta":"partial output"}`, + "", "event: response.failed", - `data: {"type":"response.failed","response":{"id":"resp_failed","status":"failed","error":{"code":"server_error","message":"upstream blew up"}}}`, + `data: {"type":"response.failed","response":{"id":"resp_failed","status":"failed","error":{"code":"server_error","message":"upstream blew up"},"usage":{"input_tokens":11,"output_tokens":3,"total_tokens":14}}}`, "", }, "\n") recorder, ctx, info, resp := newResponsesStreamTest(t, upstream, constant.ChannelTypeCodex) - _, apiErr := OaiResponsesStreamHandler(ctx, info, resp) + usage, apiErr := OaiResponsesStreamHandler(ctx, info, resp) if apiErr == nil { t.Fatal("expected Codex response.failed error") } @@ -177,6 +181,100 @@ func TestOaiResponsesStreamHandlerCodexFailedAfterCommitSkipsRetry(t *testing.T) if !strings.Contains(recorder.Body.String(), "response.created") || !strings.Contains(recorder.Body.String(), "response.failed") { t.Fatalf("committed Codex events were not forwarded: %s", recorder.Body.String()) } + if usage.PromptTokens != 11 || usage.CompletionTokens != 3 || usage.TotalTokens != 14 { + t.Fatalf("failed response usage = %#v", *usage) + } +} + +func TestOaiResponsesStreamHandlerCodexFailedAfterCommitEstimatesDeliveredUsage(t *testing.T) { + upstream := strings.Join([]string{ + "event: response.created", + `data: {"type":"response.created","response":{"id":"resp_failed","status":"in_progress"}}`, + "", + "event: response.output_text.delta", + `data: {"type":"response.output_text.delta","delta":"partial output without terminal usage"}`, + "", + "event: response.failed", + `data: {"type":"response.failed","response":{"id":"resp_failed","status":"failed","error":{"code":"server_error","message":"upstream blew up"}}}`, + "", + }, "\n") + + _, ctx, info, resp := newResponsesStreamTest(t, upstream, constant.ChannelTypeCodex) + info.SetEstimatePromptTokens(17) + service.InitTokenEncoders() + usage, apiErr := OaiResponsesStreamHandler(ctx, info, resp) + if apiErr == nil || !types.IsSkipRetryError(apiErr) { + t.Fatalf("failed response error = %#v", apiErr) + } + if usage.PromptTokens != 17 || usage.CompletionTokens <= 0 || usage.TotalTokens <= 17 { + t.Fatalf("estimated failed response usage = %#v", *usage) + } +} + +func TestOaiResponsesStreamHandlerCodexFailedAfterToolDeltaEstimatesDeliveredUsage(t *testing.T) { + upstream := strings.Join([]string{ + "event: response.created", + `data: {"type":"response.created","response":{"id":"resp_failed","status":"in_progress"}}`, + "", + "event: response.function_call_arguments.delta", + `data: {"type":"response.function_call_arguments.delta","item_id":"call_1","delta":"{\"city\":\"Shanghai\"}"}`, + "", + "event: response.failed", + `data: {"type":"response.failed","response":{"id":"resp_failed","status":"failed","error":{"code":"server_error","message":"upstream blew up"}}}`, + "", + }, "\n") + + _, ctx, info, resp := newResponsesStreamTest(t, upstream, constant.ChannelTypeCodex) + info.SetEstimatePromptTokens(19) + service.InitTokenEncoders() + usage, apiErr := OaiResponsesStreamHandler(ctx, info, resp) + if apiErr == nil || !types.IsSkipRetryError(apiErr) { + t.Fatalf("failed response error = %#v", apiErr) + } + if usage.PromptTokens != 19 || usage.CompletionTokens <= 0 || usage.TotalTokens <= 19 { + t.Fatalf("estimated tool-only failed response usage = %#v", *usage) + } +} + +func TestCodexResponsesFailedStatus(t *testing.T) { + tests := []struct { + name string + errorType string + errorCode any + statusCode int + }{ + {name: "invalid request type", errorType: "invalid_request_error", statusCode: http.StatusBadRequest}, + {name: "content policy code", errorCode: "content_policy_violation", statusCode: http.StatusBadRequest}, + {name: "invalid prompt", errorCode: "invalid_prompt", statusCode: http.StatusBadRequest}, + {name: "data residency mismatch", errorCode: "data_residency_mismatch", statusCode: http.StatusBadRequest}, + {name: "image validation", errorCode: "unsupported_image_media_type", statusCode: http.StatusBadRequest}, + {name: "authentication", errorType: "authentication_error", statusCode: http.StatusUnauthorized}, + {name: "specific code overrides generic type", errorType: "invalid_request_error", errorCode: "permission_denied", statusCode: http.StatusForbidden}, + {name: "server code overrides generic client type", errorType: "invalid_request_error", errorCode: "server_error", statusCode: http.StatusInternalServerError}, + {name: "permanent quota", errorCode: "insufficient_quota", statusCode: http.StatusTooManyRequests}, + {name: "transient rate limit", errorType: "rate_limit_error", statusCode: http.StatusTooManyRequests}, + {name: "overloaded", errorCode: "service_unavailable", statusCode: http.StatusServiceUnavailable}, + {name: "vector store timeout remains retryable fallback", errorCode: "vector_store_timeout", statusCode: http.StatusInternalServerError}, + {name: "server error", errorCode: "server_error", statusCode: http.StatusInternalServerError}, + {name: "unknown fallback", errorCode: "new_unrecognized_code", statusCode: http.StatusInternalServerError}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + response := &dto.OpenAIResponsesResponse{Error: types.OpenAIError{ + Type: test.errorType, + Code: test.errorCode, + Message: "upstream failure", + }} + apiErr := newCodexResponsesFailedError(response, false) + if apiErr.StatusCode != test.statusCode { + t.Fatalf("status = %d, want %d", apiErr.StatusCode, test.statusCode) + } + if types.IsSkipRetryError(apiErr) { + t.Fatal("pre-commit retry must remain controlled by the centralized status policy") + } + }) + } } func TestOaiResponsesStreamHandlerCodexRetryRearmsFirstResponseTracking(t *testing.T) { diff --git a/relay/responses_handler.go b/relay/responses_handler.go index cbe4f37297d..e369ab4b3f8 100644 --- a/relay/responses_handler.go +++ b/relay/responses_handler.go @@ -125,6 +125,11 @@ func ResponsesHelper(c *gin.Context, info *relaycommon.RelayInfo) (newAPIError * usage, newAPIError := adaptor.DoResponse(c, httpResp, info) if newAPIError != nil { + if usageDto := failedCodexResponsesUsageToSettle(c, info, usage); usageDto != nil { + if settleErr := service.PostTextConsumeQuotaOnError(c, info, usageDto, []string{"Codex 流式响应部分输出后失败，按已返回 usage 结算"}); settleErr != nil { + logger.LogError(c, "failed to settle delivered Codex response usage: "+settleErr.Error()) + } + } // reset status code 重置状态码 service.ResetStatusCode(newAPIError, statusCodeMappingStr) return newAPIError @@ -156,6 +161,17 @@ func ResponsesHelper(c *gin.Context, info *relaycommon.RelayInfo) (newAPIError * return nil } +func failedCodexResponsesUsageToSettle(c *gin.Context, info *relaycommon.RelayInfo, usage any) *dto.Usage { + if c == nil || info == nil || info.ChannelType != appconstant.ChannelTypeCodex || !c.Writer.Written() { + return nil + } + usageDto, ok := usage.(*dto.Usage) + if !ok || usageDto == nil || usageDto.TotalTokens <= 0 { + return nil + } + return usageDto +} + func normalizeOpenAIResponsesRequest(request any) (*dto.OpenAIResponsesRequest, error) { switch req := request.(type) { case *dto.OpenAIResponsesRequest: diff --git a/relay/responses_handler_test.go b/relay/responses_handler_test.go index 5727fd62acd..62e55259ba5 100644 --- a/relay/responses_handler_test.go +++ b/relay/responses_handler_test.go @@ -2,6 +2,7 @@ package relay import ( "encoding/json" + "net/http/httptest" "testing" "github.com/QuantumNous/new-api/constant" @@ -9,9 +10,41 @@ import ( relaycommon "github.com/QuantumNous/new-api/relay/common" relayconstant "github.com/QuantumNous/new-api/relay/constant" "github.com/QuantumNous/new-api/setting/model_setting" + "github.com/gin-gonic/gin" "github.com/stretchr/testify/require" ) +func TestFailedCodexResponsesUsageToSettle(t *testing.T) { + gin.SetMode(gin.TestMode) + usage := &dto.Usage{PromptTokens: 10, CompletionTokens: 2, TotalTokens: 12} + + tests := []struct { + name string + channelType int + committed bool + usage any + wantSettle bool + }{ + {name: "post commit Codex usage", channelType: constant.ChannelTypeCodex, committed: true, usage: usage, wantSettle: true}, + {name: "pre commit Codex failure", channelType: constant.ChannelTypeCodex, usage: usage}, + {name: "non Codex failure", channelType: constant.ChannelTypeOpenAI, committed: true, usage: usage}, + {name: "missing billable usage", channelType: constant.ChannelTypeCodex, committed: true, usage: &dto.Usage{}}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + recorder := httptest.NewRecorder() + ctx, _ := gin.CreateTestContext(recorder) + if test.committed { + ctx.Writer.WriteHeaderNow() + } + info := &relaycommon.RelayInfo{ChannelMeta: &relaycommon.ChannelMeta{ChannelType: test.channelType}} + got := failedCodexResponsesUsageToSettle(ctx, info, test.usage) + require.Equal(t, test.wantSettle, got != nil) + }) + } +} + func TestShouldPassThroughResponsesRequest_BlockRunBridgeForcesConvert(t *testing.T) { original := model_setting.GetGlobalSettings().PassThroughRequestEnabled defer func() { diff --git a/service/text_quota.go b/service/text_quota.go index a31d99cfd44..cbc9e1ad64f 100644 --- a/service/text_quota.go +++ b/service/text_quota.go @@ -326,6 +326,17 @@ func usageSemanticFromUsage(relayInfo *relaycommon.RelayInfo, usage *dto.Usage) } func PostTextConsumeQuota(ctx *gin.Context, relayInfo *relaycommon.RelayInfo, usage *dto.Usage, extraContent []string) { + _ = postTextConsumeQuota(ctx, relayInfo, usage, extraContent, true) +} + +// PostTextConsumeQuotaOnError settles usage that was already delivered before +// a terminal stream failure. The controller records the failed relay sample, so +// this path deliberately avoids recording a second sample as a success. +func PostTextConsumeQuotaOnError(ctx *gin.Context, relayInfo *relaycommon.RelayInfo, usage *dto.Usage, extraContent []string) error { + return postTextConsumeQuota(ctx, relayInfo, usage, extraContent, false) +} + +func postTextConsumeQuota(ctx *gin.Context, relayInfo *relaycommon.RelayInfo, usage *dto.Usage, extraContent []string, recordRelaySample bool) error { originUsage := usage if usage == nil { extraContent = append(extraContent, "上游无计费信息") @@ -371,13 +382,25 @@ func PostTextConsumeQuota(ctx *gin.Context, relayInfo *relaycommon.RelayInfo, us if summary.TotalTokens == 0 { extraContent = append(extraContent, "上游没有返回计费信息，无法扣费（可能是上游超时）") logger.LogError(ctx, fmt.Sprintf("total tokens is 0, cannot consume quota, userId %d, channelId %d, tokenId %d, model %s， pre-consumed quota %d", relayInfo.UserId, relayInfo.ChannelId, relayInfo.TokenId, summary.ModelName, relayInfo.FinalPreConsumedQuota)) - } else { - model.UpdateUserUsedQuotaAndRequestCount(relayInfo.UserId, summary.Quota) - model.UpdateChannelUsedQuota(relayInfo.ChannelId, summary.Quota) } - if err := SettleBilling(ctx, relayInfo, summary.Quota); err != nil { - logger.LogError(ctx, "error settling billing: "+err.Error()) + settleErr := SettleBilling(ctx, relayInfo, summary.Quota) + if settleErr != nil { + logger.LogError(ctx, "error settling billing: "+settleErr.Error()) + if !recordRelaySample && relayInfo.Billing != nil && relayInfo.Billing.NeedsRefund() { + retainedQuota := relayInfo.Billing.GetPreConsumedQuota() + if retainErr := relayInfo.Billing.Settle(retainedQuota); retainErr != nil { + logger.LogError(ctx, "error retaining pre-consumed quota after failed partial-response settlement: "+retainErr.Error()) + } else { + summary.Quota = retainedQuota + extraContent = append(extraContent, fmt.Sprintf("实际 usage 结算失败，暂按预扣额度 %s 保留", logger.FormatQuota(retainedQuota))) + } + } + } + + if summary.TotalTokens != 0 { + model.UpdateUserUsedQuotaAndRequestCount(relayInfo.UserId, summary.Quota) + model.UpdateChannelUsedQuota(relayInfo.ChannelId, summary.Quota) } logModel := summary.ModelName @@ -481,5 +504,8 @@ func PostTextConsumeQuota(ctx *gin.Context, relayInfo *relaycommon.RelayInfo, us Other: other, }) perfmetrics.RecordChannelTokens(relayInfo, int64(summary.PromptTokens), int64(summary.CompletionTokens)) - perfmetrics.RecordRelaySample(relayInfo, true, int64(summary.CompletionTokens), nil) + if recordRelaySample { + perfmetrics.RecordRelaySample(relayInfo, true, int64(summary.CompletionTokens), nil) + } + return settleErr } diff --git a/service/text_quota_error_test.go b/service/text_quota_error_test.go new file mode 100644 index 00000000000..f3a7d141bf5 --- /dev/null +++ b/service/text_quota_error_test.go @@ -0,0 +1,169 @@ +package service + +import ( + "errors" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/QuantumNous/new-api/common" + "github.com/QuantumNous/new-api/constant" + "github.com/QuantumNous/new-api/dto" + relaycommon "github.com/QuantumNous/new-api/relay/common" + "github.com/QuantumNous/new-api/types" + + "github.com/gin-gonic/gin" +) + +type recordingBillingSettler struct { + settledQuotas []int + failNext bool + commitOnFailure bool + refundCalls int +} + +func (s *recordingBillingSettler) Settle(actualQuota int) error { + if s.failNext { + s.failNext = false + if s.commitOnFailure { + s.settledQuotas = append(s.settledQuotas, actualQuota) + } + return errors.New("injected settlement failure") + } + s.settledQuotas = append(s.settledQuotas, actualQuota) + return nil +} + +func TestPostTextConsumeQuotaOnErrorDoesNotReplaceCommittedSettlementAfterLateError(t *testing.T) { + previousBatchUpdateEnabled := common.BatchUpdateEnabled + previousLogConsumeEnabled := common.LogConsumeEnabled + common.BatchUpdateEnabled = true + common.LogConsumeEnabled = false + t.Cleanup(func() { + common.BatchUpdateEnabled = previousBatchUpdateEnabled + common.LogConsumeEnabled = previousLogConsumeEnabled + }) + + gin.SetMode(gin.TestMode) + ctx, _ := gin.CreateTestContext(httptest.NewRecorder()) + ctx.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil) + billing := &recordingBillingSettler{failNext: true, commitOnFailure: true} + info := &relaycommon.RelayInfo{ + StartTime: time.Now(), + OriginModelName: "gpt-5.6-sol", + IsStream: true, + IsPlayground: true, + Billing: billing, + ChannelMeta: &relaycommon.ChannelMeta{ChannelType: constant.ChannelTypeCodex}, + PriceData: types.PriceData{ + ModelRatio: 1, + CompletionRatio: 1, + GroupRatioInfo: types.GroupRatioInfo{GroupRatio: 1}, + }, + } + + err := PostTextConsumeQuotaOnError(ctx, info, &dto.Usage{PromptTokens: 20, CompletionTokens: 5, TotalTokens: 25}, nil) + if err == nil { + t.Fatal("expected late settlement error") + } + if len(billing.settledQuotas) != 1 || billing.settledQuotas[0] == billing.GetPreConsumedQuota() { + t.Fatalf("settled quotas = %v, actual settlement must not be replaced by pre-consumption", billing.settledQuotas) + } +} + +func (s *recordingBillingSettler) Refund(*gin.Context) { + if s.NeedsRefund() { + s.refundCalls++ + } +} +func (s *recordingBillingSettler) NeedsRefund() bool { return len(s.settledQuotas) == 0 } +func (s *recordingBillingSettler) GetPreConsumedQuota() int { return 100 } +func (s *recordingBillingSettler) Reserve(int) error { return nil } + +func TestPostTextConsumeQuotaOnErrorSettlesDeliveredUsage(t *testing.T) { + previousBatchUpdateEnabled := common.BatchUpdateEnabled + previousLogConsumeEnabled := common.LogConsumeEnabled + common.BatchUpdateEnabled = true + common.LogConsumeEnabled = false + t.Cleanup(func() { + common.BatchUpdateEnabled = previousBatchUpdateEnabled + common.LogConsumeEnabled = previousLogConsumeEnabled + }) + + gin.SetMode(gin.TestMode) + ctx, _ := gin.CreateTestContext(httptest.NewRecorder()) + ctx.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil) + billing := &recordingBillingSettler{} + info := &relaycommon.RelayInfo{ + StartTime: time.Now(), + OriginModelName: "gpt-5.6-sol", + IsStream: true, + IsPlayground: true, + Billing: billing, + ChannelMeta: &relaycommon.ChannelMeta{ + ChannelType: constant.ChannelTypeCodex, + }, + PriceData: types.PriceData{ + ModelRatio: 1, + CompletionRatio: 1, + GroupRatioInfo: types.GroupRatioInfo{GroupRatio: 1}, + }, + } + usage := &dto.Usage{PromptTokens: 20, CompletionTokens: 5, TotalTokens: 25} + + if err := PostTextConsumeQuotaOnError(ctx, info, usage, []string{"partial stream failed"}); err != nil { + t.Fatalf("settle delivered usage: %v", err) + } + + if len(billing.settledQuotas) != 1 { + t.Fatalf("settle calls = %d, want 1", len(billing.settledQuotas)) + } + if billing.settledQuotas[0] <= 0 { + t.Fatalf("settled quota = %d, want positive delivered usage charge", billing.settledQuotas[0]) + } + if billing.NeedsRefund() { + t.Fatal("settled billing must not remain refundable") + } +} + +func TestPostTextConsumeQuotaOnErrorRetainsPreConsumptionWhenActualSettlementFails(t *testing.T) { + previousBatchUpdateEnabled := common.BatchUpdateEnabled + previousLogConsumeEnabled := common.LogConsumeEnabled + common.BatchUpdateEnabled = true + common.LogConsumeEnabled = false + t.Cleanup(func() { + common.BatchUpdateEnabled = previousBatchUpdateEnabled + common.LogConsumeEnabled = previousLogConsumeEnabled + }) + + gin.SetMode(gin.TestMode) + ctx, _ := gin.CreateTestContext(httptest.NewRecorder()) + ctx.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil) + billing := &recordingBillingSettler{failNext: true} + info := &relaycommon.RelayInfo{ + StartTime: time.Now(), + OriginModelName: "gpt-5.6-sol", + IsStream: true, + IsPlayground: true, + Billing: billing, + ChannelMeta: &relaycommon.ChannelMeta{ChannelType: constant.ChannelTypeCodex}, + PriceData: types.PriceData{ + ModelRatio: 1, + CompletionRatio: 1, + GroupRatioInfo: types.GroupRatioInfo{GroupRatio: 1}, + }, + } + + err := PostTextConsumeQuotaOnError(ctx, info, &dto.Usage{PromptTokens: 20, CompletionTokens: 5, TotalTokens: 25}, nil) + if err == nil { + t.Fatal("expected original settlement failure") + } + if len(billing.settledQuotas) != 1 || billing.settledQuotas[0] != billing.GetPreConsumedQuota() { + t.Fatalf("retained quotas = %v, want pre-consumed quota %d", billing.settledQuotas, billing.GetPreConsumedQuota()) + } + billing.Refund(ctx) + if billing.refundCalls != 0 || billing.NeedsRefund() { + t.Fatalf("refund lifecycle: calls=%d needsRefund=%v", billing.refundCalls, billing.NeedsRefund()) + } +} From 9456ec8c6bf13edea569249b2fd8848c20351c07 Mon Sep 17 00:00:00 2001 From: slZhong <1542123803@qq.com> Date: Tue, 11 Aug 2026 18:11:48 +0800 Subject: [PATCH 4/8] fix(relay): bound Codex fallback accounting --- ...ex-response-failed-error-mapping-design.md | 4 +- relay/channel/openai/relay_responses.go | 18 ++++++- relay/channel/openai/relay_responses_test.go | 12 +++++ service/billing_session.go | 9 ++++ service/text_quota.go | 23 ++++++--- service/text_quota_error_test.go | 50 +++++++++++++++++++ 6 files changed, 106 insertions(+), 10 deletions(-) diff --git a/docs/superpowers/specs/2026-07-16-codex-response-failed-error-mapping-design.md b/docs/superpowers/specs/2026-07-16-codex-response-failed-error-mapping-design.md index 94f727d8779..10dabbc749d 100644 --- a/docs/superpowers/specs/2026-07-16-codex-response-failed-error-mapping-design.md +++ b/docs/superpowers/specs/2026-07-16-codex-response-failed-error-mapping-design.md @@ -115,9 +115,9 @@ This guard is not Codex-specific: retrying or changing response format after byt ### 6. Partial-failure billing -The public Responses relay preserves the terminal error even when partial output was already written. For Codex only, if the adapter returns positive terminal or estimated usage after the downstream response is committed, `ResponsesHelper` runs the normal text quota calculation and settlement before returning the error. The success relay sample is suppressed because the controller records the failed sample. +The public Responses relay preserves the terminal error even when partial output was already written. For Codex only, if the adapter returns positive terminal or estimated usage after the downstream response is committed, `ResponsesHelper` runs the normal text quota calculation and settlement before returning the error. Fallback estimation retains at most 1 MiB of text-equivalent deltas per request so large tool arguments cannot cause unbounded memory or tokenizer work. The success relay sample is suppressed because the controller records the failed sample. -Pre-commit failures, non-Codex channels, and failures without positive usage retain the existing refund behavior. If actual-usage settlement fails, the billing session retains the already reserved quota so the controller's deferred refund cannot make delivered output free; the settlement failure is logged for reconciliation. The request-local `BillingSession` keeps settlement/refund idempotent, while quota mutations remain database-backed and safe across application nodes. +Pre-commit failures, non-Codex channels, and failures without positive usage retain the existing refund behavior. If actual-usage settlement fails, the billing session retains the already reserved quota so the controller's deferred refund cannot make delivered output free; the settlement failure is logged for reconciliation. User/channel usage and consume logs are written only after actual settlement or that pre-consumption retention has taken effect. The request-local `BillingSession` keeps settlement/refund idempotent, while quota mutations remain database-backed and safe across application nodes. ## Data Flow diff --git a/relay/channel/openai/relay_responses.go b/relay/channel/openai/relay_responses.go index c6015a94001..7f085dcc2eb 100644 --- a/relay/channel/openai/relay_responses.go +++ b/relay/channel/openai/relay_responses.go @@ -19,6 +19,11 @@ import ( "github.com/gin-gonic/gin" ) +// Fallback token estimation is used only when a failed stream omits terminal +// usage. Bound retained deltas so large tool arguments cannot grow per-request +// memory and tokenizer CPU without limit. +const maxResponsesFallbackUsageBytes = 1 << 20 + func OaiResponsesHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Response) (*dto.Usage, *types.NewAPIError) { defer service.CloseResponseBodyGracefully(resp) @@ -135,7 +140,7 @@ func OaiResponsesStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp "response.code_interpreter_call_code.delta": // Preserve a text-equivalent fallback when a failed terminal event // omits usage, including tool-only and reasoning-only output. - responseTextBuilder.WriteString(streamResponse.Delta) + appendResponsesFallbackUsage(&responseTextBuilder, streamResponse.Delta) case dto.ResponsesOutputTypeItemDone: // 函数调用处理 if streamResponse.Item != nil { @@ -180,6 +185,17 @@ func OaiResponsesStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp return usage, nil } +func appendResponsesFallbackUsage(builder *strings.Builder, delta string) { + if builder == nil || builder.Len() >= maxResponsesFallbackUsageBytes || delta == "" { + return + } + remaining := maxResponsesFallbackUsageBytes - builder.Len() + if len(delta) > remaining { + delta = delta[:remaining] + } + builder.WriteString(delta) +} + func finalizeResponsesUsage(usage *dto.Usage, responseTextBuilder *strings.Builder, info *relaycommon.RelayInfo) { if usage.CompletionTokens == 0 && responseTextBuilder.Len() > 0 { usage.CompletionTokens = service.CountTextToken(responseTextBuilder.String(), info.UpstreamModelName) diff --git a/relay/channel/openai/relay_responses_test.go b/relay/channel/openai/relay_responses_test.go index df3560f10d2..2ea29dc3cba 100644 --- a/relay/channel/openai/relay_responses_test.go +++ b/relay/channel/openai/relay_responses_test.go @@ -236,6 +236,18 @@ func TestOaiResponsesStreamHandlerCodexFailedAfterToolDeltaEstimatesDeliveredUsa } } +func TestAppendResponsesFallbackUsageCapsBufferedBytes(t *testing.T) { + var builder strings.Builder + delta := strings.Repeat("x", maxResponsesFallbackUsageBytes+1024) + + appendResponsesFallbackUsage(&builder, delta) + appendResponsesFallbackUsage(&builder, "ignored after cap") + + if builder.Len() != maxResponsesFallbackUsageBytes { + t.Fatalf("buffered bytes = %d, want %d", builder.Len(), maxResponsesFallbackUsageBytes) + } +} + func TestCodexResponsesFailedStatus(t *testing.T) { tests := []struct { name string diff --git a/service/billing_session.go b/service/billing_session.go index f4bef075715..bc222355cb8 100644 --- a/service/billing_session.go +++ b/service/billing_session.go @@ -134,6 +134,15 @@ func (s *BillingSession) NeedsRefund() bool { return s.needsRefundLocked() } +// settlementApplied reports whether settlement has committed far enough that +// the original pre-consumption must not be refunded. It intentionally remains +// a service-local optional capability rather than widening BillingSettler. +func (s *BillingSession) settlementApplied() bool { + s.mu.Lock() + defer s.mu.Unlock() + return s.settled || s.fundingSettled +} + func (s *BillingSession) needsRefundLocked() bool { if s.settled || s.refunded || s.fundingSettled { // fundingSettled 时资金来源已提交结算，不能再退预扣费 diff --git a/service/text_quota.go b/service/text_quota.go index cbc9e1ad64f..1079ac8498e 100644 --- a/service/text_quota.go +++ b/service/text_quota.go @@ -385,18 +385,27 @@ func postTextConsumeQuota(ctx *gin.Context, relayInfo *relaycommon.RelayInfo, us } settleErr := SettleBilling(ctx, relayInfo, summary.Quota) + settlementApplied := settleErr == nil if settleErr != nil { logger.LogError(ctx, "error settling billing: "+settleErr.Error()) - if !recordRelaySample && relayInfo.Billing != nil && relayInfo.Billing.NeedsRefund() { - retainedQuota := relayInfo.Billing.GetPreConsumedQuota() - if retainErr := relayInfo.Billing.Settle(retainedQuota); retainErr != nil { - logger.LogError(ctx, "error retaining pre-consumed quota after failed partial-response settlement: "+retainErr.Error()) - } else { - summary.Quota = retainedQuota - extraContent = append(extraContent, fmt.Sprintf("实际 usage 结算失败，暂按预扣额度 %s 保留", logger.FormatQuota(retainedQuota))) + if !recordRelaySample && relayInfo.Billing != nil { + if state, ok := relayInfo.Billing.(interface{ settlementApplied() bool }); ok && state.settlementApplied() { + settlementApplied = true + } else if relayInfo.Billing.NeedsRefund() { + retainedQuota := relayInfo.Billing.GetPreConsumedQuota() + if retainErr := relayInfo.Billing.Settle(retainedQuota); retainErr != nil { + logger.LogError(ctx, "error retaining pre-consumed quota after failed partial-response settlement: "+retainErr.Error()) + } else { + settlementApplied = true + summary.Quota = retainedQuota + extraContent = append(extraContent, fmt.Sprintf("实际 usage 结算失败，暂按预扣额度 %s 保留", logger.FormatQuota(retainedQuota))) + } } } } + if !recordRelaySample && !settlementApplied { + return settleErr + } if summary.TotalTokens != 0 { model.UpdateUserUsedQuotaAndRequestCount(relayInfo.UserId, summary.Quota) diff --git a/service/text_quota_error_test.go b/service/text_quota_error_test.go index f3a7d141bf5..8e47d5c5bd0 100644 --- a/service/text_quota_error_test.go +++ b/service/text_quota_error_test.go @@ -10,6 +10,7 @@ import ( "github.com/QuantumNous/new-api/common" "github.com/QuantumNous/new-api/constant" "github.com/QuantumNous/new-api/dto" + "github.com/QuantumNous/new-api/model" relaycommon "github.com/QuantumNous/new-api/relay/common" "github.com/QuantumNous/new-api/types" @@ -19,11 +20,15 @@ import ( type recordingBillingSettler struct { settledQuotas []int failNext bool + failAlways bool commitOnFailure bool refundCalls int } func (s *recordingBillingSettler) Settle(actualQuota int) error { + if s.failAlways { + return errors.New("injected persistent settlement failure") + } if s.failNext { s.failNext = false if s.commitOnFailure { @@ -80,6 +85,7 @@ func (s *recordingBillingSettler) Refund(*gin.Context) { func (s *recordingBillingSettler) NeedsRefund() bool { return len(s.settledQuotas) == 0 } func (s *recordingBillingSettler) GetPreConsumedQuota() int { return 100 } func (s *recordingBillingSettler) Reserve(int) error { return nil } +func (s *recordingBillingSettler) settlementApplied() bool { return len(s.settledQuotas) > 0 } func TestPostTextConsumeQuotaOnErrorSettlesDeliveredUsage(t *testing.T) { previousBatchUpdateEnabled := common.BatchUpdateEnabled @@ -167,3 +173,47 @@ func TestPostTextConsumeQuotaOnErrorRetainsPreConsumptionWhenActualSettlementFai t.Fatalf("refund lifecycle: calls=%d needsRefund=%v", billing.refundCalls, billing.NeedsRefund()) } } + +func TestPostTextConsumeQuotaOnErrorStopsWhenSettlementAndRetentionFail(t *testing.T) { + previousBatchUpdateEnabled := common.BatchUpdateEnabled + previousLogConsumeEnabled := common.LogConsumeEnabled + previousSpendHook := model.TemporaryChannelSpendHook + common.BatchUpdateEnabled = true + common.LogConsumeEnabled = true + consumeLogCalls := 0 + model.TemporaryChannelSpendHook = func(int, string, int) { consumeLogCalls++ } + t.Cleanup(func() { + common.BatchUpdateEnabled = previousBatchUpdateEnabled + common.LogConsumeEnabled = previousLogConsumeEnabled + model.TemporaryChannelSpendHook = previousSpendHook + }) + + gin.SetMode(gin.TestMode) + ctx, _ := gin.CreateTestContext(httptest.NewRecorder()) + ctx.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil) + billing := &recordingBillingSettler{failAlways: true} + info := &relaycommon.RelayInfo{ + StartTime: time.Now(), + OriginModelName: "gpt-5.6-sol", + IsStream: true, + IsPlayground: true, + Billing: billing, + ChannelMeta: &relaycommon.ChannelMeta{ChannelType: constant.ChannelTypeCodex}, + PriceData: types.PriceData{ + ModelRatio: 1, + CompletionRatio: 1, + GroupRatioInfo: types.GroupRatioInfo{GroupRatio: 1}, + }, + } + + err := PostTextConsumeQuotaOnError(ctx, info, &dto.Usage{PromptTokens: 20, CompletionTokens: 5, TotalTokens: 25}, nil) + if err == nil { + t.Fatal("expected persistent settlement failure") + } + if billing.settlementApplied() || !billing.NeedsRefund() { + t.Fatalf("settlement state: applied=%v needsRefund=%v", billing.settlementApplied(), billing.NeedsRefund()) + } + if consumeLogCalls != 0 { + t.Fatalf("consume log calls = %d, want 0 when no settlement was applied", consumeLogCalls) + } +} From 5c43b27d7ae189d7e1b76761d7b47b6fcd1c3546 Mon Sep 17 00:00:00 2001 From: slZhong <1542123803@qq.com> Date: Tue, 11 Aug 2026 18:29:09 +0800 Subject: [PATCH 5/8] refactor(relay): simplify Codex terminal coverage --- relay/channel/openai/relay_responses_test.go | 73 +++++------- relay/responses_handler.go | 19 +--- relay/responses_handler_test.go | 33 ------ service/text_quota_error_test.go | 114 ++++--------------- 4 files changed, 58 insertions(+), 181 deletions(-) diff --git a/relay/channel/openai/relay_responses_test.go b/relay/channel/openai/relay_responses_test.go index 2ea29dc3cba..b78eff1e030 100644 --- a/relay/channel/openai/relay_responses_test.go +++ b/relay/channel/openai/relay_responses_test.go @@ -1,6 +1,7 @@ package openai import ( + "fmt" "io" "net/http" "net/http/httptest" @@ -187,52 +188,40 @@ func TestOaiResponsesStreamHandlerCodexFailedAfterCommitSkipsRetry(t *testing.T) } func TestOaiResponsesStreamHandlerCodexFailedAfterCommitEstimatesDeliveredUsage(t *testing.T) { - upstream := strings.Join([]string{ - "event: response.created", - `data: {"type":"response.created","response":{"id":"resp_failed","status":"in_progress"}}`, - "", - "event: response.output_text.delta", - `data: {"type":"response.output_text.delta","delta":"partial output without terminal usage"}`, - "", - "event: response.failed", - `data: {"type":"response.failed","response":{"id":"resp_failed","status":"failed","error":{"code":"server_error","message":"upstream blew up"}}}`, - "", - }, "\n") - - _, ctx, info, resp := newResponsesStreamTest(t, upstream, constant.ChannelTypeCodex) - info.SetEstimatePromptTokens(17) service.InitTokenEncoders() - usage, apiErr := OaiResponsesStreamHandler(ctx, info, resp) - if apiErr == nil || !types.IsSkipRetryError(apiErr) { - t.Fatalf("failed response error = %#v", apiErr) - } - if usage.PromptTokens != 17 || usage.CompletionTokens <= 0 || usage.TotalTokens <= 17 { - t.Fatalf("estimated failed response usage = %#v", *usage) + tests := []struct { + name string + eventType string + delta string + }{ + {name: "text", eventType: "response.output_text.delta", delta: "partial output without terminal usage"}, + {name: "tool arguments", eventType: "response.function_call_arguments.delta", delta: `{"city":"Shanghai"}`}, } -} -func TestOaiResponsesStreamHandlerCodexFailedAfterToolDeltaEstimatesDeliveredUsage(t *testing.T) { - upstream := strings.Join([]string{ - "event: response.created", - `data: {"type":"response.created","response":{"id":"resp_failed","status":"in_progress"}}`, - "", - "event: response.function_call_arguments.delta", - `data: {"type":"response.function_call_arguments.delta","item_id":"call_1","delta":"{\"city\":\"Shanghai\"}"}`, - "", - "event: response.failed", - `data: {"type":"response.failed","response":{"id":"resp_failed","status":"failed","error":{"code":"server_error","message":"upstream blew up"}}}`, - "", - }, "\n") + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + upstream := strings.Join([]string{ + "event: response.created", + `data: {"type":"response.created","response":{"id":"resp_failed","status":"in_progress"}}`, + "", + "event: " + test.eventType, + fmt.Sprintf(`data: {"type":%q,"item_id":"call_1","delta":%q}`, test.eventType, test.delta), + "", + "event: response.failed", + `data: {"type":"response.failed","response":{"id":"resp_failed","status":"failed","error":{"code":"server_error","message":"upstream blew up"}}}`, + "", + }, "\n") - _, ctx, info, resp := newResponsesStreamTest(t, upstream, constant.ChannelTypeCodex) - info.SetEstimatePromptTokens(19) - service.InitTokenEncoders() - usage, apiErr := OaiResponsesStreamHandler(ctx, info, resp) - if apiErr == nil || !types.IsSkipRetryError(apiErr) { - t.Fatalf("failed response error = %#v", apiErr) - } - if usage.PromptTokens != 19 || usage.CompletionTokens <= 0 || usage.TotalTokens <= 19 { - t.Fatalf("estimated tool-only failed response usage = %#v", *usage) + _, ctx, info, resp := newResponsesStreamTest(t, upstream, constant.ChannelTypeCodex) + info.SetEstimatePromptTokens(17) + usage, apiErr := OaiResponsesStreamHandler(ctx, info, resp) + if apiErr == nil || !types.IsSkipRetryError(apiErr) { + t.Fatalf("failed response error = %#v", apiErr) + } + if usage.PromptTokens != 17 || usage.CompletionTokens <= 0 || usage.TotalTokens <= 17 { + t.Fatalf("estimated failed response usage = %#v", *usage) + } + }) } } diff --git a/relay/responses_handler.go b/relay/responses_handler.go index e369ab4b3f8..085584034c2 100644 --- a/relay/responses_handler.go +++ b/relay/responses_handler.go @@ -125,9 +125,11 @@ func ResponsesHelper(c *gin.Context, info *relaycommon.RelayInfo) (newAPIError * usage, newAPIError := adaptor.DoResponse(c, httpResp, info) if newAPIError != nil { - if usageDto := failedCodexResponsesUsageToSettle(c, info, usage); usageDto != nil { - if settleErr := service.PostTextConsumeQuotaOnError(c, info, usageDto, []string{"Codex 流式响应部分输出后失败，按已返回 usage 结算"}); settleErr != nil { - logger.LogError(c, "failed to settle delivered Codex response usage: "+settleErr.Error()) + if info.ChannelType == appconstant.ChannelTypeCodex && c.Writer.Written() { + if usageDto, ok := usage.(*dto.Usage); ok && usageDto != nil && usageDto.TotalTokens > 0 { + if settleErr := service.PostTextConsumeQuotaOnError(c, info, usageDto, []string{"Codex 流式响应部分输出后失败，按已返回 usage 结算"}); settleErr != nil { + logger.LogError(c, "failed to settle delivered Codex response usage: "+settleErr.Error()) + } } } // reset status code 重置状态码 @@ -161,17 +163,6 @@ func ResponsesHelper(c *gin.Context, info *relaycommon.RelayInfo) (newAPIError * return nil } -func failedCodexResponsesUsageToSettle(c *gin.Context, info *relaycommon.RelayInfo, usage any) *dto.Usage { - if c == nil || info == nil || info.ChannelType != appconstant.ChannelTypeCodex || !c.Writer.Written() { - return nil - } - usageDto, ok := usage.(*dto.Usage) - if !ok || usageDto == nil || usageDto.TotalTokens <= 0 { - return nil - } - return usageDto -} - func normalizeOpenAIResponsesRequest(request any) (*dto.OpenAIResponsesRequest, error) { switch req := request.(type) { case *dto.OpenAIResponsesRequest: diff --git a/relay/responses_handler_test.go b/relay/responses_handler_test.go index 62e55259ba5..5727fd62acd 100644 --- a/relay/responses_handler_test.go +++ b/relay/responses_handler_test.go @@ -2,7 +2,6 @@ package relay import ( "encoding/json" - "net/http/httptest" "testing" "github.com/QuantumNous/new-api/constant" @@ -10,41 +9,9 @@ import ( relaycommon "github.com/QuantumNous/new-api/relay/common" relayconstant "github.com/QuantumNous/new-api/relay/constant" "github.com/QuantumNous/new-api/setting/model_setting" - "github.com/gin-gonic/gin" "github.com/stretchr/testify/require" ) -func TestFailedCodexResponsesUsageToSettle(t *testing.T) { - gin.SetMode(gin.TestMode) - usage := &dto.Usage{PromptTokens: 10, CompletionTokens: 2, TotalTokens: 12} - - tests := []struct { - name string - channelType int - committed bool - usage any - wantSettle bool - }{ - {name: "post commit Codex usage", channelType: constant.ChannelTypeCodex, committed: true, usage: usage, wantSettle: true}, - {name: "pre commit Codex failure", channelType: constant.ChannelTypeCodex, usage: usage}, - {name: "non Codex failure", channelType: constant.ChannelTypeOpenAI, committed: true, usage: usage}, - {name: "missing billable usage", channelType: constant.ChannelTypeCodex, committed: true, usage: &dto.Usage{}}, - } - - for _, test := range tests { - t.Run(test.name, func(t *testing.T) { - recorder := httptest.NewRecorder() - ctx, _ := gin.CreateTestContext(recorder) - if test.committed { - ctx.Writer.WriteHeaderNow() - } - info := &relaycommon.RelayInfo{ChannelMeta: &relaycommon.ChannelMeta{ChannelType: test.channelType}} - got := failedCodexResponsesUsageToSettle(ctx, info, test.usage) - require.Equal(t, test.wantSettle, got != nil) - }) - } -} - func TestShouldPassThroughResponsesRequest_BlockRunBridgeForcesConvert(t *testing.T) { original := model_setting.GetGlobalSettings().PassThroughRequestEnabled defer func() { diff --git a/service/text_quota_error_test.go b/service/text_quota_error_test.go index 8e47d5c5bd0..ad9fa9a9f06 100644 --- a/service/text_quota_error_test.go +++ b/service/text_quota_error_test.go @@ -19,18 +19,14 @@ import ( type recordingBillingSettler struct { settledQuotas []int - failNext bool - failAlways bool + failures int commitOnFailure bool refundCalls int } func (s *recordingBillingSettler) Settle(actualQuota int) error { - if s.failAlways { - return errors.New("injected persistent settlement failure") - } - if s.failNext { - s.failNext = false + if s.failures > 0 { + s.failures-- if s.commitOnFailure { s.settledQuotas = append(s.settledQuotas, actualQuota) } @@ -41,34 +37,10 @@ func (s *recordingBillingSettler) Settle(actualQuota int) error { } func TestPostTextConsumeQuotaOnErrorDoesNotReplaceCommittedSettlementAfterLateError(t *testing.T) { - previousBatchUpdateEnabled := common.BatchUpdateEnabled - previousLogConsumeEnabled := common.LogConsumeEnabled - common.BatchUpdateEnabled = true - common.LogConsumeEnabled = false - t.Cleanup(func() { - common.BatchUpdateEnabled = previousBatchUpdateEnabled - common.LogConsumeEnabled = previousLogConsumeEnabled - }) + billing := &recordingBillingSettler{failures: 1, commitOnFailure: true} + ctx, info, usage := newTextQuotaErrorTest(t, billing, false) - gin.SetMode(gin.TestMode) - ctx, _ := gin.CreateTestContext(httptest.NewRecorder()) - ctx.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil) - billing := &recordingBillingSettler{failNext: true, commitOnFailure: true} - info := &relaycommon.RelayInfo{ - StartTime: time.Now(), - OriginModelName: "gpt-5.6-sol", - IsStream: true, - IsPlayground: true, - Billing: billing, - ChannelMeta: &relaycommon.ChannelMeta{ChannelType: constant.ChannelTypeCodex}, - PriceData: types.PriceData{ - ModelRatio: 1, - CompletionRatio: 1, - GroupRatioInfo: types.GroupRatioInfo{GroupRatio: 1}, - }, - } - - err := PostTextConsumeQuotaOnError(ctx, info, &dto.Usage{PromptTokens: 20, CompletionTokens: 5, TotalTokens: 25}, nil) + err := PostTextConsumeQuotaOnError(ctx, info, usage, nil) if err == nil { t.Fatal("expected late settlement error") } @@ -87,11 +59,12 @@ func (s *recordingBillingSettler) GetPreConsumedQuota() int { return 100 } func (s *recordingBillingSettler) Reserve(int) error { return nil } func (s *recordingBillingSettler) settlementApplied() bool { return len(s.settledQuotas) > 0 } -func TestPostTextConsumeQuotaOnErrorSettlesDeliveredUsage(t *testing.T) { +func newTextQuotaErrorTest(t *testing.T, billing *recordingBillingSettler, logConsume bool) (*gin.Context, *relaycommon.RelayInfo, *dto.Usage) { + t.Helper() previousBatchUpdateEnabled := common.BatchUpdateEnabled previousLogConsumeEnabled := common.LogConsumeEnabled common.BatchUpdateEnabled = true - common.LogConsumeEnabled = false + common.LogConsumeEnabled = logConsume t.Cleanup(func() { common.BatchUpdateEnabled = previousBatchUpdateEnabled common.LogConsumeEnabled = previousLogConsumeEnabled @@ -100,23 +73,25 @@ func TestPostTextConsumeQuotaOnErrorSettlesDeliveredUsage(t *testing.T) { gin.SetMode(gin.TestMode) ctx, _ := gin.CreateTestContext(httptest.NewRecorder()) ctx.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil) - billing := &recordingBillingSettler{} info := &relaycommon.RelayInfo{ StartTime: time.Now(), OriginModelName: "gpt-5.6-sol", IsStream: true, IsPlayground: true, Billing: billing, - ChannelMeta: &relaycommon.ChannelMeta{ - ChannelType: constant.ChannelTypeCodex, - }, + ChannelMeta: &relaycommon.ChannelMeta{ChannelType: constant.ChannelTypeCodex}, PriceData: types.PriceData{ ModelRatio: 1, CompletionRatio: 1, GroupRatioInfo: types.GroupRatioInfo{GroupRatio: 1}, }, } - usage := &dto.Usage{PromptTokens: 20, CompletionTokens: 5, TotalTokens: 25} + return ctx, info, &dto.Usage{PromptTokens: 20, CompletionTokens: 5, TotalTokens: 25} +} + +func TestPostTextConsumeQuotaOnErrorSettlesDeliveredUsage(t *testing.T) { + billing := &recordingBillingSettler{} + ctx, info, usage := newTextQuotaErrorTest(t, billing, false) if err := PostTextConsumeQuotaOnError(ctx, info, usage, []string{"partial stream failed"}); err != nil { t.Fatalf("settle delivered usage: %v", err) @@ -134,34 +109,10 @@ func TestPostTextConsumeQuotaOnErrorSettlesDeliveredUsage(t *testing.T) { } func TestPostTextConsumeQuotaOnErrorRetainsPreConsumptionWhenActualSettlementFails(t *testing.T) { - previousBatchUpdateEnabled := common.BatchUpdateEnabled - previousLogConsumeEnabled := common.LogConsumeEnabled - common.BatchUpdateEnabled = true - common.LogConsumeEnabled = false - t.Cleanup(func() { - common.BatchUpdateEnabled = previousBatchUpdateEnabled - common.LogConsumeEnabled = previousLogConsumeEnabled - }) - - gin.SetMode(gin.TestMode) - ctx, _ := gin.CreateTestContext(httptest.NewRecorder()) - ctx.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil) - billing := &recordingBillingSettler{failNext: true} - info := &relaycommon.RelayInfo{ - StartTime: time.Now(), - OriginModelName: "gpt-5.6-sol", - IsStream: true, - IsPlayground: true, - Billing: billing, - ChannelMeta: &relaycommon.ChannelMeta{ChannelType: constant.ChannelTypeCodex}, - PriceData: types.PriceData{ - ModelRatio: 1, - CompletionRatio: 1, - GroupRatioInfo: types.GroupRatioInfo{GroupRatio: 1}, - }, - } + billing := &recordingBillingSettler{failures: 1} + ctx, info, usage := newTextQuotaErrorTest(t, billing, false) - err := PostTextConsumeQuotaOnError(ctx, info, &dto.Usage{PromptTokens: 20, CompletionTokens: 5, TotalTokens: 25}, nil) + err := PostTextConsumeQuotaOnError(ctx, info, usage, nil) if err == nil { t.Fatal("expected original settlement failure") } @@ -175,38 +126,17 @@ func TestPostTextConsumeQuotaOnErrorRetainsPreConsumptionWhenActualSettlementFai } func TestPostTextConsumeQuotaOnErrorStopsWhenSettlementAndRetentionFail(t *testing.T) { - previousBatchUpdateEnabled := common.BatchUpdateEnabled - previousLogConsumeEnabled := common.LogConsumeEnabled previousSpendHook := model.TemporaryChannelSpendHook - common.BatchUpdateEnabled = true - common.LogConsumeEnabled = true consumeLogCalls := 0 model.TemporaryChannelSpendHook = func(int, string, int) { consumeLogCalls++ } t.Cleanup(func() { - common.BatchUpdateEnabled = previousBatchUpdateEnabled - common.LogConsumeEnabled = previousLogConsumeEnabled model.TemporaryChannelSpendHook = previousSpendHook }) - gin.SetMode(gin.TestMode) - ctx, _ := gin.CreateTestContext(httptest.NewRecorder()) - ctx.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil) - billing := &recordingBillingSettler{failAlways: true} - info := &relaycommon.RelayInfo{ - StartTime: time.Now(), - OriginModelName: "gpt-5.6-sol", - IsStream: true, - IsPlayground: true, - Billing: billing, - ChannelMeta: &relaycommon.ChannelMeta{ChannelType: constant.ChannelTypeCodex}, - PriceData: types.PriceData{ - ModelRatio: 1, - CompletionRatio: 1, - GroupRatioInfo: types.GroupRatioInfo{GroupRatio: 1}, - }, - } + billing := &recordingBillingSettler{failures: 2} + ctx, info, usage := newTextQuotaErrorTest(t, billing, true) - err := PostTextConsumeQuotaOnError(ctx, info, &dto.Usage{PromptTokens: 20, CompletionTokens: 5, TotalTokens: 25}, nil) + err := PostTextConsumeQuotaOnError(ctx, info, usage, nil) if err == nil { t.Fatal("expected persistent settlement failure") } From c73a4f13f127375ad7c675d8aecc0cc799fd135f Mon Sep 17 00:00:00 2001 From: slZhong <1542123803@qq.com> Date: Tue, 11 Aug 2026 18:43:19 +0800 Subject: [PATCH 6/8] fix(service): preserve legacy partial settlement audit --- service/billing.go | 19 ++++++++++++----- service/billing_status_test.go | 35 ++++++++++++++++++++++++++++++++ service/quota.go | 19 ++++++++++++----- service/text_quota.go | 21 ++++++++----------- service/text_quota_error_test.go | 10 ++++++++- 5 files changed, 80 insertions(+), 24 deletions(-) diff --git a/service/billing.go b/service/billing.go index 9c9fc48f222..c9cc1ed16d8 100644 --- a/service/billing.go +++ b/service/billing.go @@ -32,6 +32,11 @@ func PreConsumeBilling(c *gin.Context, preConsumedQuota int, relayInfo *relaycom // SettleBilling 执行计费结算。如果 RelayInfo 上有 BillingSession 则通过 session 结算， // 否则回退到旧的 PostConsumeQuota 路径（兼容按次计费等场景）。 func SettleBilling(ctx *gin.Context, relayInfo *relaycommon.RelayInfo, actualQuota int) error { + _, err := settleBillingWithStatus(ctx, relayInfo, actualQuota) + return err +} + +func settleBillingWithStatus(ctx *gin.Context, relayInfo *relaycommon.RelayInfo, actualQuota int) (bool, error) { if relayInfo.Billing != nil { preConsumed := relayInfo.Billing.GetPreConsumedQuota() delta := actualQuota - preConsumed @@ -55,7 +60,10 @@ func SettleBilling(ctx *gin.Context, relayInfo *relaycommon.RelayInfo, actualQuo } if err := relayInfo.Billing.Settle(actualQuota); err != nil { - return err + if state, ok := relayInfo.Billing.(interface{ settlementApplied() bool }); ok { + return state.settlementApplied(), err + } + return false, err } // 发送额度通知（订阅计费使用订阅剩余额度） @@ -66,16 +74,17 @@ func SettleBilling(ctx *gin.Context, relayInfo *relaycommon.RelayInfo, actualQuo checkAndSendWalletQuotaNonEmailNotify(relayInfo, delta, preConsumed) } } - return nil + return true, nil } // 回退：无 BillingSession 时使用旧路径 quotaDelta := actualQuota - relayInfo.FinalPreConsumedQuota if quotaDelta != 0 { - if err := PostConsumeQuota(relayInfo, quotaDelta, relayInfo.FinalPreConsumedQuota, true); err != nil { - return err + fundingApplied, err := postConsumeQuotaWithStatus(relayInfo, quotaDelta, relayInfo.FinalPreConsumedQuota, true) + if err != nil { + return fundingApplied, err } relayInfo.FinalPreConsumedQuota = actualQuota } - return nil + return true, nil } diff --git a/service/billing_status_test.go b/service/billing_status_test.go index 80309e97afa..d122c53f5db 100644 --- a/service/billing_status_test.go +++ b/service/billing_status_test.go @@ -342,6 +342,26 @@ func TestPostConsumeQuotaTracksUnlimitedTokenQuota(t *testing.T) { require.Equal(t, 920, userQuota) } +func TestSettleBillingLegacyReportsCommittedWalletDebit(t *testing.T) { + const ( + userID = 10122 + tokenID = 10222 + tokenKey = "billing-status-legacy-partial-settlement" + ) + resetBillingStatusTables(t) + seedUser(t, userID, 1000) + seedToken(t, tokenID, userID, tokenKey, 1000) + blockTokenDebitForTest(t, tokenID, "legacy_token_debit_blocked") + + relayInfo := newQuotaStatusRelayInfo(userID, tokenID, tokenKey) + applied, err := settleBillingWithStatus(nil, relayInfo, 80) + + require.ErrorContains(t, err, "legacy_token_debit_blocked") + require.True(t, applied) + require.Equal(t, 920, getUserQuota(t, userID)) + require.Equal(t, 1000, getTokenRemainQuota(t, tokenID)) +} + func TestBillingSessionSettleTracksUnlimitedTokenQuota(t *testing.T) { const ( userID = 10117 @@ -724,6 +744,21 @@ func blockTokenCreditForTest(t *testing.T, tokenID int, message string) { }) } +func blockTokenDebitForTest(t *testing.T, tokenID int, message string) { + t.Helper() + triggerName := "test_block_token_debit_" + strings.ReplaceAll(message, "-", "_") + require.NoError(t, model.DB.Exec("DROP TRIGGER IF EXISTS "+triggerName).Error) + require.NoError(t, model.DB.Exec(fmt.Sprintf( + "CREATE TRIGGER %s BEFORE UPDATE OF remain_quota ON tokens "+ + "WHEN OLD.id = %d AND NEW.remain_quota < OLD.remain_quota "+ + "BEGIN SELECT RAISE(ABORT, '%s'); END", + triggerName, tokenID, message, + )).Error) + t.Cleanup(func() { + require.NoError(t, model.DB.Exec("DROP TRIGGER IF EXISTS "+triggerName).Error) + }) +} + func drainWalletAfterTokenDebitForTest(t *testing.T, userID int, tokenID int) { t.Helper() const triggerName = "test_drain_wallet_after_token_debit" diff --git a/service/quota.go b/service/quota.go index 70e42da65ef..3c0887364ff 100644 --- a/service/quota.go +++ b/service/quota.go @@ -416,18 +416,26 @@ func PreConsumeTokenQuota(relayInfo *relaycommon.RelayInfo, quota int) error { } func PostConsumeQuota(relayInfo *relaycommon.RelayInfo, quota int, preConsumedQuota int, sendEmail bool) (err error) { + _, err = postConsumeQuotaWithStatus(relayInfo, quota, preConsumedQuota, sendEmail) + return err +} + +// postConsumeQuotaWithStatus reports whether the funding-side mutation was +// committed before a later token-quota update failed. +func postConsumeQuotaWithStatus(relayInfo *relaycommon.RelayInfo, quota int, preConsumedQuota int, sendEmail bool) (fundingApplied bool, err error) { // 1) Consume from wallet quota OR subscription item if relayInfo != nil && relayInfo.BillingSource == BillingSourceSubscription { if relayInfo.SubscriptionId == 0 { - return errors.New("subscription id is missing") + return false, errors.New("subscription id is missing") } delta := int64(quota) if delta != 0 { if err := model.PostConsumeUserSubscriptionDelta(relayInfo.SubscriptionId, delta); err != nil { - return err + return false, err } relayInfo.SubscriptionPostDelta += delta + fundingApplied = true } } else { // Wallet @@ -437,8 +445,9 @@ func PostConsumeQuota(relayInfo *relaycommon.RelayInfo, quota int, preConsumedQu err = model.IncreaseUserQuota(relayInfo.UserId, -quota, false) } if err != nil { - return err + return false, err } + fundingApplied = quota != 0 } if !relayInfo.IsPlayground { @@ -448,7 +457,7 @@ func PostConsumeQuota(relayInfo *relaycommon.RelayInfo, quota int, preConsumedQu err = model.IncreaseTokenQuota(relayInfo.TokenId, relayInfo.TokenKey, -quota) } if err != nil { - return err + return fundingApplied, err } } @@ -460,7 +469,7 @@ func PostConsumeQuota(relayInfo *relaycommon.RelayInfo, quota int, preConsumedQu } } - return nil + return fundingApplied, nil } // notifyLang resolves the language for a background notification (no gin diff --git a/service/text_quota.go b/service/text_quota.go index 1079ac8498e..f3867599a97 100644 --- a/service/text_quota.go +++ b/service/text_quota.go @@ -384,22 +384,17 @@ func postTextConsumeQuota(ctx *gin.Context, relayInfo *relaycommon.RelayInfo, us logger.LogError(ctx, fmt.Sprintf("total tokens is 0, cannot consume quota, userId %d, channelId %d, tokenId %d, model %s， pre-consumed quota %d", relayInfo.UserId, relayInfo.ChannelId, relayInfo.TokenId, summary.ModelName, relayInfo.FinalPreConsumedQuota)) } - settleErr := SettleBilling(ctx, relayInfo, summary.Quota) - settlementApplied := settleErr == nil + settlementApplied, settleErr := settleBillingWithStatus(ctx, relayInfo, summary.Quota) if settleErr != nil { logger.LogError(ctx, "error settling billing: "+settleErr.Error()) - if !recordRelaySample && relayInfo.Billing != nil { - if state, ok := relayInfo.Billing.(interface{ settlementApplied() bool }); ok && state.settlementApplied() { + if !recordRelaySample && !settlementApplied && relayInfo.Billing != nil && relayInfo.Billing.NeedsRefund() { + retainedQuota := relayInfo.Billing.GetPreConsumedQuota() + if retainErr := relayInfo.Billing.Settle(retainedQuota); retainErr != nil { + logger.LogError(ctx, "error retaining pre-consumed quota after failed partial-response settlement: "+retainErr.Error()) + } else { settlementApplied = true - } else if relayInfo.Billing.NeedsRefund() { - retainedQuota := relayInfo.Billing.GetPreConsumedQuota() - if retainErr := relayInfo.Billing.Settle(retainedQuota); retainErr != nil { - logger.LogError(ctx, "error retaining pre-consumed quota after failed partial-response settlement: "+retainErr.Error()) - } else { - settlementApplied = true - summary.Quota = retainedQuota - extraContent = append(extraContent, fmt.Sprintf("实际 usage 结算失败，暂按预扣额度 %s 保留", logger.FormatQuota(retainedQuota))) - } + summary.Quota = retainedQuota + extraContent = append(extraContent, fmt.Sprintf("实际 usage 结算失败，暂按预扣额度 %s 保留", logger.FormatQuota(retainedQuota))) } } } diff --git a/service/text_quota_error_test.go b/service/text_quota_error_test.go index ad9fa9a9f06..ec39217a666 100644 --- a/service/text_quota_error_test.go +++ b/service/text_quota_error_test.go @@ -37,8 +37,13 @@ func (s *recordingBillingSettler) Settle(actualQuota int) error { } func TestPostTextConsumeQuotaOnErrorDoesNotReplaceCommittedSettlementAfterLateError(t *testing.T) { + previousSpendHook := model.TemporaryChannelSpendHook + consumeLogCalls := 0 + model.TemporaryChannelSpendHook = func(int, string, int) { consumeLogCalls++ } + t.Cleanup(func() { model.TemporaryChannelSpendHook = previousSpendHook }) + billing := &recordingBillingSettler{failures: 1, commitOnFailure: true} - ctx, info, usage := newTextQuotaErrorTest(t, billing, false) + ctx, info, usage := newTextQuotaErrorTest(t, billing, true) err := PostTextConsumeQuotaOnError(ctx, info, usage, nil) if err == nil { @@ -47,6 +52,9 @@ func TestPostTextConsumeQuotaOnErrorDoesNotReplaceCommittedSettlementAfterLateEr if len(billing.settledQuotas) != 1 || billing.settledQuotas[0] == billing.GetPreConsumedQuota() { t.Fatalf("settled quotas = %v, actual settlement must not be replaced by pre-consumption", billing.settledQuotas) } + if consumeLogCalls != 1 { + t.Fatalf("consume log calls = %d, want 1 after settlement committed", consumeLogCalls) + } } func (s *recordingBillingSettler) Refund(*gin.Context) { From 3483b1a212856986822b8146bb6e752e5720e7d1 Mon Sep 17 00:00:00 2001 From: slZhong <1542123803@qq.com> Date: Wed, 12 Aug 2026 14:45:52 +0800 Subject: [PATCH 7/8] test(service): isolate recall and subscription fixtures --- service/recall_email_test.go | 1 + service/subscription_contract_test.go | 2 ++ service/subscription_migration_test.go | 2 ++ 3 files changed, 5 insertions(+) diff --git a/service/recall_email_test.go b/service/recall_email_test.go index 31750e893a2..1c06660613e 100644 --- a/service/recall_email_test.go +++ b/service/recall_email_test.go @@ -1832,6 +1832,7 @@ func TestRecallEmailWorkerActivitySMTPConfigIsFreshAndControlsMessageIDDomain(t require.Len(t, configs, 1) require.Equal(t, "smtp.first.example.com", configs[0].Server) require.Equal(t, "first-secret", configs[0].Token) + clearRecallEmailPacingForTest(t) second := model.RecallMessage{ RecipientId: fixture.recipient.Id, StageNo: 2, TemplateVersion: 11, TemplateSnapshot: fixture.message.TemplateSnapshot, diff --git a/service/subscription_contract_test.go b/service/subscription_contract_test.go index b51454dd36d..97e986fdc9a 100644 --- a/service/subscription_contract_test.go +++ b/service/subscription_contract_test.go @@ -63,6 +63,8 @@ func setupSubscriptionContractServiceTestDB(t *testing.T) { &model.SubscriptionProviderBinding{}, &model.UserSubscriptionContract{}, &model.SubscriptionChangeIntent{}, + &model.SubscriptionDiscountAccount{}, + &model.SubscriptionDiscountEntry{}, &model.RecallLifecycleEvent{}, &model.QuotaLifecycleState{}, )) diff --git a/service/subscription_migration_test.go b/service/subscription_migration_test.go index 5f766e16709..6472c2690d0 100644 --- a/service/subscription_migration_test.go +++ b/service/subscription_migration_test.go @@ -216,6 +216,8 @@ func setupSubscriptionMigrationServiceTestDB(t *testing.T) { &model.SubscriptionProviderBinding{}, &model.UserSubscriptionContract{}, &model.SubscriptionChangeIntent{}, + &model.RecallLifecycleEvent{}, + &model.QuotaLifecycleState{}, )) } From b700a6d988cdb9a4a228890ccb7ab2a315dc89da Mon Sep 17 00:00:00 2001 From: slZhong <1542123803@qq.com> Date: Wed, 12 Aug 2026 15:06:25 +0800 Subject: [PATCH 8/8] test: narrow codex terminal failure regression scope --- ...ex-response-failed-error-mapping-design.md | 31 ++-- relay/channel/openai/relay_responses_test.go | 134 ++++++------------ relay/common/relay_info_test.go | 24 ---- service/recall_email_test.go | 1 - service/subscription_contract_test.go | 2 - service/subscription_migration_test.go | 2 - service/text_quota_error_test.go | 120 ++++++---------- 7 files changed, 94 insertions(+), 220 deletions(-) diff --git a/docs/superpowers/specs/2026-07-16-codex-response-failed-error-mapping-design.md b/docs/superpowers/specs/2026-07-16-codex-response-failed-error-mapping-design.md index 10dabbc749d..ed3448a09b9 100644 --- a/docs/superpowers/specs/2026-07-16-codex-response-failed-error-mapping-design.md +++ b/docs/superpowers/specs/2026-07-16-codex-response-failed-error-mapping-design.md @@ -28,14 +28,13 @@ LiteLLM recognizes an OpenAI-shaped first SSE error frame whose `error.code` is - Never emit a success finish chunk or `[DONE]` after a failed or truncated stream. - Never retry after downstream bytes have been written. - Preserve upstream HTTP non-2xx status mapping and successful stream behavior. -- Retain usage carried by failed terminal events in the adapter result. -- For Codex streams that fail after billable output has reached the client, settle the delivered usage exactly once while preserving the failed request outcome. +- Retain usage carried by failed terminal events in the adapter result while keeping the existing refund-on-error billing policy. ## Non-goals - Changing the public Responses API relay path. - Retrying a request after any downstream bytes were written. -- Charging pre-commit failures or failures without any billable delivered usage. +- Charging users for failed requests; existing refund behavior remains unchanged. - Refactoring unrelated streaming adapters or the controller retry policy beyond the response-commit invariant. ## Considered Approaches @@ -62,7 +61,7 @@ For `response.failed`: - capture `response.error.message`, falling back to `response.error.code`, then to a generic upstream-failure message; - capture `response.usage` when present; -- map recognized Responses error codes to their HTTP semantics, with unknown/server failures falling back to HTTP 500; +- create a `types.NewAPIError` with HTTP 500 and the existing bad-response error category; - do not pass the event through the normal success finalizer. An actual upstream HTTP non-2xx response continues to preserve its original status code. Transport/protocol termination without a successful terminal event uses HTTP 502. @@ -100,7 +99,7 @@ The bridge tracks whether a successful terminal event (`response.completed`, `re The upstream SSE stream is fully accumulated before any downstream body is written. - successful terminal event: build the current HTTP 200 Chat Completions JSON response; -- `response.failed`: return the mapped non-2xx status with an OpenAI error object, never a Chat Completions assistant message; +- `response.failed`: return HTTP 500 with an OpenAI error object, never a Chat Completions assistant message; - scanner error or EOF without a terminal event: return HTTP 502 with an OpenAI error object; - failed-event usage remains returned by the adapter alongside the error for accounting visibility, while the caller's existing error path refunds the request. @@ -113,18 +112,12 @@ The relay controller enforces a generic HTTP invariant: This guard is not Codex-specific: retrying or changing response format after bytes have reached a client is invalid for every streaming provider. The Codex adapter still marks late failures with `ErrOptionWithSkipRetry` as defense in depth. -### 6. Partial-failure billing - -The public Responses relay preserves the terminal error even when partial output was already written. For Codex only, if the adapter returns positive terminal or estimated usage after the downstream response is committed, `ResponsesHelper` runs the normal text quota calculation and settlement before returning the error. Fallback estimation retains at most 1 MiB of text-equivalent deltas per request so large tool arguments cannot cause unbounded memory or tokenizer work. The success relay sample is suppressed because the controller records the failed sample. - -Pre-commit failures, non-Codex channels, and failures without positive usage retain the existing refund behavior. If actual-usage settlement fails, the billing session retains the already reserved quota so the controller's deferred refund cannot make delivered output free; the settlement failure is logged for reconciliation. User/channel usage and consume logs are written only after actual settlement or that pre-consumption retention has taken effect. The request-local `BillingSession` keeps settlement/refund idempotent, while quota mutations remain database-backed and safe across application nodes. - ## Data Flow | Upstream sequence | Downstream result | | --- | --- | | HTTP non-2xx | Same non-2xx status through existing error mapping | -| `created -> failed` | Mapped non-2xx OpenAI error JSON; no role/content chunk | +| `created -> failed` | HTTP 500 OpenAI error JSON; no role/content chunk | | `created -> text delta -> failed` | Existing role/text chunks, then SSE error frame; no finish or `[DONE]` | | `created -> EOF` | HTTP 502 before output, or SSE error after output | | `created -> completed` | Existing successful Chat Completions stream/JSON | @@ -156,26 +149,18 @@ Pre-commit failures, non-Codex channels, and failures without positive usage ret - retain role/content/finish/usage behavior. 7. Failed terminal event with usage: - returns the captured prompt/completion/total token values alongside the error. - - after committed Codex output, settles once, remains an error, and is not refunded; - - before commit, retains the existing refund path. -8. Failed terminal event without usage after text or tool-call deltas: - - estimates positive delivered output usage and follows the same post-commit settlement path. -9. Recognized `response.error.code` values: - - map to accurate 4xx/429/5xx status classes; - - unknown and transient server codes fall back to retryable 500 behavior; - - pre-commit retry remains controlled by the centralized status policy. ### API compatibility conversion -10. `response.failed` conversion: +8. `response.failed` conversion: - never emits upstream error text as `delta.content`; - retains usage in converter state when called directly. ### Relay controller -11. Retry decision after response commitment: +9. Retry decision after response commitment: - `shouldRetry` is false after any downstream write. -12. Deferred error rendering after response commitment: +10. Deferred error rendering after response commitment: - does not append a JSON error object to an existing SSE body. ## Validation diff --git a/relay/channel/openai/relay_responses_test.go b/relay/channel/openai/relay_responses_test.go index b78eff1e030..a5462b0ebeb 100644 --- a/relay/channel/openai/relay_responses_test.go +++ b/relay/channel/openai/relay_responses_test.go @@ -68,43 +68,33 @@ func TestOaiResponsesStreamHandlerCapturesIncompleteUsage(t *testing.T) { } } -func TestOaiResponsesStreamHandlerCodexCapturesDoneUsage(t *testing.T) { +func TestOaiResponsesStreamHandlerResponseDoneIsCodexOnly(t *testing.T) { upstream := strings.Join([]string{ "event: response.done", `data: {"type":"response.done","response":{"id":"resp_done","status":"completed","usage":{"input_tokens":71,"output_tokens":19,"total_tokens":90,"input_tokens_details":{"cached_tokens":7,"cache_write_tokens":2}}}}`, "", }, "\n") - recorder, ctx, info, resp := newResponsesStreamTest(t, upstream, constant.ChannelTypeCodex) - usage, apiErr := OaiResponsesStreamHandler(ctx, info, resp) - if apiErr != nil { - t.Fatalf("handle Codex response.done: %v", apiErr) - } - if usage.PromptTokens != 71 || usage.CompletionTokens != 19 || usage.TotalTokens != 90 { - t.Fatalf("done usage = %#v", *usage) - } - if usage.PromptTokensDetails.CachedTokens != 7 || usage.PromptTokensDetails.CacheWriteTokens != 2 { - t.Fatalf("done token details = %#v", usage.PromptTokensDetails) - } - if !strings.Contains(recorder.Body.String(), "response.done") { - t.Fatalf("done event was not forwarded: %s", recorder.Body.String()) - } -} - -func TestOaiResponsesStreamHandlerNonCodexKeepsDoneBehavior(t *testing.T) { - upstream := strings.Join([]string{ - "event: response.done", - `data: {"type":"response.done","response":{"id":"resp_done","status":"completed","usage":{"input_tokens":71,"output_tokens":19,"total_tokens":90}}}`, - "", - }, "\n") - - _, ctx, info, resp := newResponsesStreamTest(t, upstream, constant.ChannelTypeOpenAI) - usage, apiErr := OaiResponsesStreamHandler(ctx, info, resp) - if apiErr != nil { - t.Fatalf("handle non-Codex response.done: %v", apiErr) - } - if usage.PromptTokens != 0 || usage.CompletionTokens != 0 || usage.TotalTokens != 0 { - t.Fatalf("non-Codex behavior changed, usage = %#v", *usage) + for _, test := range []struct { + name, want string + channel int + }{ + {name: "Codex captures usage", channel: constant.ChannelTypeCodex, want: "71/19/90"}, + {name: "other channels stay unchanged", channel: constant.ChannelTypeOpenAI, want: "0/0/0"}, + } { + t.Run(test.name, func(t *testing.T) { + recorder, ctx, info, resp := newResponsesStreamTest(t, upstream, test.channel) + usage, apiErr := OaiResponsesStreamHandler(ctx, info, resp) + if apiErr != nil { + t.Fatal(apiErr) + } + if got := fmt.Sprintf("%d/%d/%d", usage.PromptTokens, usage.CompletionTokens, usage.TotalTokens); got != test.want { + t.Fatalf("usage = %s, want %s", got, test.want) + } + if !strings.Contains(recorder.Body.String(), "response.done") { + t.Fatalf("done event was not forwarded: %s", recorder.Body.String()) + } + }) } } @@ -130,17 +120,10 @@ func TestOaiResponsesStreamHandlerCodexFailedBeforeCommitIsRetryable(t *testing. if recorder.Body.Len() != 0 || ctx.Writer.Written() { t.Fatalf("failed event committed before retry: %s", recorder.Body.String()) } - if got := recorder.Header().Get("Content-Type"); got != "" { - t.Fatalf("retryable failure retained SSE content type %q", got) - } - if got := recorder.Header().Get("Transfer-Encoding"); got != "" { - t.Fatalf("retryable failure retained transfer encoding %q", got) - } - if got := recorder.Header().Get("X-Reasoning-Included"); got != "" { - t.Fatalf("retryable failure retained Codex response header %q", got) - } - if got := recorder.Header().Get("X-Codex-Turn-State"); got != "" { - t.Fatalf("retryable failure retained Codex turn state %q", got) + for _, header := range []string{"Content-Type", "Transfer-Encoding", "X-Reasoning-Included", "X-Codex-Turn-State"} { + if got := recorder.Header().Get(header); got != "" { + t.Fatalf("retryable failure retained %s %q", header, got) + } } if got := recorder.Header().Get("X-Existing"); got != "keep" { t.Fatalf("pre-existing response header = %q", got) @@ -152,9 +135,24 @@ func TestOaiResponsesStreamHandlerCodexFailedBeforeCommitIsRetryable(t *testing. t.Fatalf("retryable failure retained first-response state: sent=%v received=%d", info.HasSendResponse(), info.ReceivedResponseCount) } - ctx.JSON(apiErr.StatusCode, gin.H{"error": apiErr.ToOpenAIError()}) - if got := recorder.Header().Get("Content-Type"); got != "application/json; charset=utf-8" { - t.Fatalf("final JSON content type = %q", got) + // A real retry reuses RelayInfo and the downstream writer. Prove the reset + // above re-arms both response observation and SSE headers. + retry := strings.Join([]string{ + "event: response.done", + `data: {"type":"response.done","response":{"id":"resp_done","status":"completed","usage":{"input_tokens":8,"output_tokens":2,"total_tokens":10}}}`, + "", + }, "\n") + resp = &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}}, + Body: io.NopCloser(strings.NewReader(retry)), + } + usage, apiErr := OaiResponsesStreamHandler(ctx, info, resp) + if apiErr != nil || !info.HasSendResponse() || info.ReceivedResponseCount != 1 { + t.Fatalf("retry error=%v sent=%v received=%d", apiErr, info.HasSendResponse(), info.ReceivedResponseCount) + } + if usage.PromptTokens != 8 || usage.CompletionTokens != 2 || recorder.Header().Get("Content-Type") != "text/event-stream" { + t.Fatalf("retry usage=%#v content-type=%q", *usage, recorder.Header().Get("Content-Type")) } } @@ -245,18 +243,12 @@ func TestCodexResponsesFailedStatus(t *testing.T) { statusCode int }{ {name: "invalid request type", errorType: "invalid_request_error", statusCode: http.StatusBadRequest}, - {name: "content policy code", errorCode: "content_policy_violation", statusCode: http.StatusBadRequest}, - {name: "invalid prompt", errorCode: "invalid_prompt", statusCode: http.StatusBadRequest}, - {name: "data residency mismatch", errorCode: "data_residency_mismatch", statusCode: http.StatusBadRequest}, {name: "image validation", errorCode: "unsupported_image_media_type", statusCode: http.StatusBadRequest}, {name: "authentication", errorType: "authentication_error", statusCode: http.StatusUnauthorized}, {name: "specific code overrides generic type", errorType: "invalid_request_error", errorCode: "permission_denied", statusCode: http.StatusForbidden}, {name: "server code overrides generic client type", errorType: "invalid_request_error", errorCode: "server_error", statusCode: http.StatusInternalServerError}, {name: "permanent quota", errorCode: "insufficient_quota", statusCode: http.StatusTooManyRequests}, - {name: "transient rate limit", errorType: "rate_limit_error", statusCode: http.StatusTooManyRequests}, {name: "overloaded", errorCode: "service_unavailable", statusCode: http.StatusServiceUnavailable}, - {name: "vector store timeout remains retryable fallback", errorCode: "vector_store_timeout", statusCode: http.StatusInternalServerError}, - {name: "server error", errorCode: "server_error", statusCode: http.StatusInternalServerError}, {name: "unknown fallback", errorCode: "new_unrecognized_code", statusCode: http.StatusInternalServerError}, } @@ -278,46 +270,6 @@ func TestCodexResponsesFailedStatus(t *testing.T) { } } -func TestOaiResponsesStreamHandlerCodexRetryRearmsFirstResponseTracking(t *testing.T) { - failedUpstream := strings.Join([]string{ - "event: response.failed", - `data: {"type":"response.failed","response":{"id":"resp_failed","status":"failed","error":{"code":"server_error","message":"upstream blew up"}}}`, - "", - }, "\n") - - recorder, ctx, info, failedResp := newResponsesStreamTest(t, failedUpstream, constant.ChannelTypeCodex) - info.ApiType = constant.APITypeCodex - _, apiErr := OaiResponsesStreamHandler(ctx, info, failedResp) - if apiErr == nil || types.IsSkipRetryError(apiErr) { - t.Fatalf("first attempt error = %#v", apiErr) - } - - doneUpstream := strings.Join([]string{ - "event: response.done", - `data: {"type":"response.done","response":{"id":"resp_done","status":"completed","usage":{"input_tokens":8,"output_tokens":2,"total_tokens":10}}}`, - "", - }, "\n") - doneResp := &http.Response{ - StatusCode: http.StatusOK, - Header: http.Header{"Content-Type": []string{"text/event-stream"}}, - Body: io.NopCloser(strings.NewReader(doneUpstream)), - } - - usage, apiErr := OaiResponsesStreamHandler(ctx, info, doneResp) - if apiErr != nil { - t.Fatalf("retry attempt: %v", apiErr) - } - if !info.HasSendResponse() || info.ReceivedResponseCount != 1 { - t.Fatalf("retry first-response state: sent=%v received=%d", info.HasSendResponse(), info.ReceivedResponseCount) - } - if usage.PromptTokens != 8 || usage.CompletionTokens != 2 { - t.Fatalf("retry usage = %#v", *usage) - } - if got := recorder.Header().Get("Content-Type"); got != "text/event-stream" { - t.Fatalf("retry content type = %q", got) - } -} - func newResponsesStreamTest(t *testing.T, upstream string, channelType int) (*httptest.ResponseRecorder, *gin.Context, *relaycommon.RelayInfo, *http.Response) { t.Helper() previousTimeout := constant.StreamingTimeout diff --git a/relay/common/relay_info_test.go b/relay/common/relay_info_test.go index ed8242dbf3a..e53ec804ca0 100644 --- a/relay/common/relay_info_test.go +++ b/relay/common/relay_info_test.go @@ -2,35 +2,11 @@ package common import ( "testing" - "time" "github.com/QuantumNous/new-api/types" "github.com/stretchr/testify/require" ) -func TestResetStreamResponseStateForRetry(t *testing.T) { - start := time.Now().Add(-time.Second) - info := &RelayInfo{ - StartTime: start, - FirstResponseTime: time.Now(), - isFirstResponse: false, - ReceivedResponseCount: 3, - } - - info.ResetStreamResponseStateForRetry() - - if info.HasSendResponse() { - t.Fatal("reset stream state still reports a sent response") - } - if info.ReceivedResponseCount != 0 { - t.Fatalf("received response count = %d", info.ReceivedResponseCount) - } - info.SetFirstResponseTime() - if !info.HasSendResponse() { - t.Fatal("reset stream state did not re-arm first response tracking") - } -} - func TestRelayInfoGetFinalRequestRelayFormatPrefersExplicitFinal(t *testing.T) { info := &RelayInfo{ RelayFormat: types.RelayFormatOpenAI, diff --git a/service/recall_email_test.go b/service/recall_email_test.go index 1c06660613e..31750e893a2 100644 --- a/service/recall_email_test.go +++ b/service/recall_email_test.go @@ -1832,7 +1832,6 @@ func TestRecallEmailWorkerActivitySMTPConfigIsFreshAndControlsMessageIDDomain(t require.Len(t, configs, 1) require.Equal(t, "smtp.first.example.com", configs[0].Server) require.Equal(t, "first-secret", configs[0].Token) - clearRecallEmailPacingForTest(t) second := model.RecallMessage{ RecipientId: fixture.recipient.Id, StageNo: 2, TemplateVersion: 11, TemplateSnapshot: fixture.message.TemplateSnapshot, diff --git a/service/subscription_contract_test.go b/service/subscription_contract_test.go index 97e986fdc9a..b51454dd36d 100644 --- a/service/subscription_contract_test.go +++ b/service/subscription_contract_test.go @@ -63,8 +63,6 @@ func setupSubscriptionContractServiceTestDB(t *testing.T) { &model.SubscriptionProviderBinding{}, &model.UserSubscriptionContract{}, &model.SubscriptionChangeIntent{}, - &model.SubscriptionDiscountAccount{}, - &model.SubscriptionDiscountEntry{}, &model.RecallLifecycleEvent{}, &model.QuotaLifecycleState{}, )) diff --git a/service/subscription_migration_test.go b/service/subscription_migration_test.go index 6472c2690d0..5f766e16709 100644 --- a/service/subscription_migration_test.go +++ b/service/subscription_migration_test.go @@ -216,8 +216,6 @@ func setupSubscriptionMigrationServiceTestDB(t *testing.T) { &model.SubscriptionProviderBinding{}, &model.UserSubscriptionContract{}, &model.SubscriptionChangeIntent{}, - &model.RecallLifecycleEvent{}, - &model.QuotaLifecycleState{}, )) } diff --git a/service/text_quota_error_test.go b/service/text_quota_error_test.go index ec39217a666..117f32775d9 100644 --- a/service/text_quota_error_test.go +++ b/service/text_quota_error_test.go @@ -36,27 +36,6 @@ func (s *recordingBillingSettler) Settle(actualQuota int) error { return nil } -func TestPostTextConsumeQuotaOnErrorDoesNotReplaceCommittedSettlementAfterLateError(t *testing.T) { - previousSpendHook := model.TemporaryChannelSpendHook - consumeLogCalls := 0 - model.TemporaryChannelSpendHook = func(int, string, int) { consumeLogCalls++ } - t.Cleanup(func() { model.TemporaryChannelSpendHook = previousSpendHook }) - - billing := &recordingBillingSettler{failures: 1, commitOnFailure: true} - ctx, info, usage := newTextQuotaErrorTest(t, billing, true) - - err := PostTextConsumeQuotaOnError(ctx, info, usage, nil) - if err == nil { - t.Fatal("expected late settlement error") - } - if len(billing.settledQuotas) != 1 || billing.settledQuotas[0] == billing.GetPreConsumedQuota() { - t.Fatalf("settled quotas = %v, actual settlement must not be replaced by pre-consumption", billing.settledQuotas) - } - if consumeLogCalls != 1 { - t.Fatalf("consume log calls = %d, want 1 after settlement committed", consumeLogCalls) - } -} - func (s *recordingBillingSettler) Refund(*gin.Context) { if s.NeedsRefund() { s.refundCalls++ @@ -97,61 +76,48 @@ func newTextQuotaErrorTest(t *testing.T, billing *recordingBillingSettler, logCo return ctx, info, &dto.Usage{PromptTokens: 20, CompletionTokens: 5, TotalTokens: 25} } -func TestPostTextConsumeQuotaOnErrorSettlesDeliveredUsage(t *testing.T) { - billing := &recordingBillingSettler{} - ctx, info, usage := newTextQuotaErrorTest(t, billing, false) - - if err := PostTextConsumeQuotaOnError(ctx, info, usage, []string{"partial stream failed"}); err != nil { - t.Fatalf("settle delivered usage: %v", err) - } - - if len(billing.settledQuotas) != 1 { - t.Fatalf("settle calls = %d, want 1", len(billing.settledQuotas)) - } - if billing.settledQuotas[0] <= 0 { - t.Fatalf("settled quota = %d, want positive delivered usage charge", billing.settledQuotas[0]) - } - if billing.NeedsRefund() { - t.Fatal("settled billing must not remain refundable") - } -} - -func TestPostTextConsumeQuotaOnErrorRetainsPreConsumptionWhenActualSettlementFails(t *testing.T) { - billing := &recordingBillingSettler{failures: 1} - ctx, info, usage := newTextQuotaErrorTest(t, billing, false) - - err := PostTextConsumeQuotaOnError(ctx, info, usage, nil) - if err == nil { - t.Fatal("expected original settlement failure") - } - if len(billing.settledQuotas) != 1 || billing.settledQuotas[0] != billing.GetPreConsumedQuota() { - t.Fatalf("retained quotas = %v, want pre-consumed quota %d", billing.settledQuotas, billing.GetPreConsumedQuota()) - } - billing.Refund(ctx) - if billing.refundCalls != 0 || billing.NeedsRefund() { - t.Fatalf("refund lifecycle: calls=%d needsRefund=%v", billing.refundCalls, billing.NeedsRefund()) - } -} - -func TestPostTextConsumeQuotaOnErrorStopsWhenSettlementAndRetentionFail(t *testing.T) { - previousSpendHook := model.TemporaryChannelSpendHook - consumeLogCalls := 0 - model.TemporaryChannelSpendHook = func(int, string, int) { consumeLogCalls++ } - t.Cleanup(func() { - model.TemporaryChannelSpendHook = previousSpendHook - }) - - billing := &recordingBillingSettler{failures: 2} - ctx, info, usage := newTextQuotaErrorTest(t, billing, true) - - err := PostTextConsumeQuotaOnError(ctx, info, usage, nil) - if err == nil { - t.Fatal("expected persistent settlement failure") - } - if billing.settlementApplied() || !billing.NeedsRefund() { - t.Fatalf("settlement state: applied=%v needsRefund=%v", billing.settlementApplied(), billing.NeedsRefund()) - } - if consumeLogCalls != 0 { - t.Fatalf("consume log calls = %d, want 0 when no settlement was applied", consumeLogCalls) +func TestPostTextConsumeQuotaOnErrorSettlementOutcomes(t *testing.T) { + for _, test := range []struct { + name string + failures int + commitOnFailure bool + wantErr, wantApplied bool + wantRetained bool + wantConsumeLogs int + }{ + {name: "settles delivered usage", wantApplied: true, wantConsumeLogs: 1}, + {name: "retains pre-consumption after settlement failure", failures: 1, wantErr: true, wantApplied: true, wantRetained: true, wantConsumeLogs: 1}, + {name: "does not replace a committed settlement after a late error", failures: 1, commitOnFailure: true, wantErr: true, wantApplied: true, wantConsumeLogs: 1}, + {name: "does not log when settlement and retention fail", failures: 2, wantErr: true}, + } { + t.Run(test.name, func(t *testing.T) { + previousSpendHook := model.TemporaryChannelSpendHook + consumeLogCalls := 0 + model.TemporaryChannelSpendHook = func(int, string, int) { consumeLogCalls++ } + t.Cleanup(func() { model.TemporaryChannelSpendHook = previousSpendHook }) + + billing := &recordingBillingSettler{failures: test.failures, commitOnFailure: test.commitOnFailure} + ctx, info, usage := newTextQuotaErrorTest(t, billing, true) + err := PostTextConsumeQuotaOnError(ctx, info, usage, nil) + if (err != nil) != test.wantErr { + t.Fatalf("error = %v, wantErr %v", err, test.wantErr) + } + if billing.settlementApplied() != test.wantApplied || billing.NeedsRefund() == test.wantApplied { + t.Fatalf("settlement applied=%v needsRefund=%v", billing.settlementApplied(), billing.NeedsRefund()) + } + if consumeLogCalls != test.wantConsumeLogs { + t.Fatalf("consume logs = %d, want %d", consumeLogCalls, test.wantConsumeLogs) + } + if test.wantApplied { + gotRetained := billing.settledQuotas[0] == billing.GetPreConsumedQuota() + if gotRetained != test.wantRetained { + t.Fatalf("settled quotas = %v, retained=%v", billing.settledQuotas, gotRetained) + } + } + billing.Refund(ctx) + if test.wantApplied && billing.refundCalls != 0 { + t.Fatalf("refund calls = %d", billing.refundCalls) + } + }) } }