fix: preserve Astra reasoning mode for Codex OAuth

This commit is contained in:
alfadb
2026-09-07 17:58:20 +08:00
parent 909b8c7ef4
commit 0aaed397cc
5 changed files with 608 additions and 0 deletions
@@ -0,0 +1,244 @@
//go:build unit
package service
// Integration-level tests for GPT-6 Astra reasoning.mode preservation through
// OpenAIGatewayService.Forward.
//
// These exercise the real forward pipeline through the httpUpstreamRecorder mock
// (no real network / credentials / config), covering the OpenAI OAuth account
// (native Codex transform when passthrough is disabled, and the passthrough
// branch when enabled). Client stream=true and stream=false both receive an
// upstream SSE fixture, because Codex upstreams stream: the stream=false client
// path exercises the real SSE->JSON aggregation instead of a synthetic JSON body.
import (
"context"
"io"
"net/http"
"net/http/httptest"
"strings"
"testing"
"github.com/Wei-Shaw/sub2api/internal/config"
"github.com/gin-gonic/gin"
"github.com/stretchr/testify/require"
"github.com/tidwall/gjson"
)
const astraProCodexResponsesURL = "https://chatgpt.com/backend-api/codex/responses"
type astraForwardSetup struct {
upstream *httpUpstreamRecorder
svc *OpenAIGatewayService
c *gin.Context
rec *httptest.ResponseRecorder
account *Account
}
// newAstraOAuthSetup builds an OpenAI OAuth account using the minimal
// svc+account harness pattern (mirrors openai_oauth_passthrough_test.go). When
// passthrough is true the account routes through forwardOpenAIPassthrough.
func newAstraOAuthSetup(t *testing.T, passthrough bool) *astraForwardSetup {
t.Helper()
gin.SetMode(gin.TestMode)
rec := httptest.NewRecorder()
c, _ := gin.CreateTestContext(rec)
c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil)
c.Request.Header.Set("User-Agent", "codex_cli_rs/0.144.1")
upstream := &httpUpstreamRecorder{}
svc := &OpenAIGatewayService{
cfg: &config.Config{Gateway: config.GatewayConfig{ForceCodexCLI: false}},
httpUpstream: upstream,
}
account := &Account{
ID: 888,
Name: "oauth-astra",
Platform: PlatformOpenAI,
Type: AccountTypeOAuth,
Concurrency: 1,
Credentials: map[string]any{"access_token": "oauth-token", "chatgpt_account_id": "chatgpt-acc"},
Extra: map[string]any{"openai_passthrough": passthrough},
Status: StatusActive,
Schedulable: true,
RateMultiplier: f64p(1),
}
return &astraForwardSetup{upstream: upstream, svc: svc, c: c, rec: rec, account: account}
}
func codexCompletedSSE(inner string) string {
return "data: {\"type\":\"response.completed\",\"response\":" + inner + "}\n\ndata: [DONE]\n\n"
}
func astraRequestBody(model string, stream bool, mode, effort string) []byte {
var reasoningParts []string
if mode != "" {
reasoningParts = append(reasoningParts, `"mode":"`+mode+`"`)
}
if effort != "" {
reasoningParts = append(reasoningParts, `"effort":"`+effort+`"`)
}
reasoning := ""
if len(reasoningParts) > 0 {
reasoning = `,"reasoning":{` + strings.Join(reasoningParts, ",") + `}`
}
streamJSON := "false"
if stream {
streamJSON = "true"
}
return []byte(`{"model":"` + model + `","stream":` + streamJSON + `,` +
`"instructions":"test","input":[{"type":"message","role":"user","content":[{"type":"input_text","text":"hi"}]}]` +
reasoning + `}`)
}
// TestForward_AstraOAuth_NonAstraLegacyStillStrips asserts the historical strip
// behavior is preserved for a non-Astra model on an OAuth account.
func TestForward_AstraOAuth_NonAstraLegacyStillStrips(t *testing.T) {
gin.SetMode(gin.TestMode)
s := newAstraOAuthSetup(t, false)
inner := `{"id":"resp_test","model":"gpt-5.6-sol","output":[],"usage":{"input_tokens":1,"output_tokens":1,"total_tokens":2}}`
s.upstream.resp = &http.Response{
StatusCode: http.StatusOK,
Header: http.Header{"Content-Type": []string{"text/event-stream"}},
Body: io.NopCloser(strings.NewReader(codexCompletedSSE(inner))),
}
body := astraRequestBody("gpt-5.6-sol", true, "pro", "")
result, err := s.svc.Forward(context.Background(), s.c, s.account, body)
require.NoError(t, err)
require.NotNil(t, result)
require.NotNil(t, s.upstream.lastReq)
require.Equal(t, astraProCodexResponsesURL, s.upstream.lastReq.URL.String())
forwarded := s.upstream.lastBody
require.False(t, gjson.GetBytes(forwarded, "reasoning.mode").Exists(),
"non-Astra legacy must strip reasoning.mode")
require.Equal(t, "max", gjson.GetBytes(forwarded, "reasoning.effort").String(),
"non-Astra legacy mode=pro without effort injects max")
}
// TestForward_AstraOAuth_ModeMatrix_Preserved runs the pro/standard/missing-mode
// x effort=max matrix across both forward branches and both client stream modes.
// Every attempt receives an upstream SSE fixture; assertions check the final
// upstream request keeps mode/effort and the URL is the Codex responses endpoint.
func TestForward_AstraOAuth_ModeMatrix_Preserved(t *testing.T) {
cases := []struct {
name string
mode string
effort string
wantMod string // gjson string value; "" == absent
wantEff string
}{
{name: "pro+max", mode: "pro", effort: "max", wantMod: "pro", wantEff: "max"},
{name: "standard+max", mode: "standard", effort: "max", wantMod: "standard", wantEff: "max"},
{name: "missing mode + max", mode: "", effort: "max", wantMod: "", wantEff: "max"},
{name: "pro+high", mode: "pro", effort: "high", wantMod: "pro", wantEff: "high"},
{name: "pro no effort", mode: "pro", effort: "", wantMod: "pro", wantEff: ""},
}
for _, passthrough := range []bool{false, true} {
branch := "native-codex"
if passthrough {
branch = "passthrough"
}
for _, stream := range []bool{true, false} {
for _, tt := range cases {
t.Run(tt.name+"/"+branch+"/stream="+map[bool]string{true: "true", false: "false"}[stream], func(t *testing.T) {
s := newAstraOAuthSetup(t, passthrough)
inner := `{"id":"resp_test","model":"gpt-6-astra","output":[],"usage":{"input_tokens":1,"output_tokens":1,"total_tokens":2}}`
s.upstream.resp = &http.Response{
StatusCode: http.StatusOK,
Header: http.Header{"Content-Type": []string{"text/event-stream"}},
Body: io.NopCloser(strings.NewReader(codexCompletedSSE(inner))),
}
body := astraRequestBody("gpt-6-astra", stream, tt.mode, tt.effort)
result, err := s.svc.Forward(context.Background(), s.c, s.account, body)
require.NoError(t, err)
require.NotNil(t, result)
require.NotNil(t, s.upstream.lastReq)
require.Equal(t, astraProCodexResponsesURL, s.upstream.lastReq.URL.String(),
"OAuth account must hit the Codex responses endpoint")
forwarded := s.upstream.lastBody
require.Equal(t, "gpt-6-astra", gjson.GetBytes(forwarded, "model").String())
mode := gjson.GetBytes(forwarded, "reasoning.mode")
if tt.wantMod == "" {
require.False(t, mode.Exists(), "mode should be absent for %q", tt.name)
} else {
require.True(t, mode.Exists(), "Astra mode must be preserved for %q", tt.name)
require.Equal(t, tt.wantMod, mode.String())
}
eff := gjson.GetBytes(forwarded, "reasoning.effort")
if tt.wantEff == "" {
require.False(t, eff.Exists(), "no effort injected when omitted for %q", tt.name)
} else {
require.Equal(t, tt.wantEff, eff.String(), "Astra effort preserved for %q", tt.name)
}
})
}
}
}
}
// TestForward_AstraOAuth_NonStreamAggregation_PreservesResponseReasoningMode
// drives client stream=false (which is forced to an SSE upstream) and asserts the
// aggregated JSON response object still carries response.reasoning.mode after
// SSE->JSON aggregation.
func TestForward_AstraOAuth_NonStreamAggregation_PreservesResponseReasoningMode(t *testing.T) {
gin.SetMode(gin.TestMode)
s := newAstraOAuthSetup(t, false)
// response.completed.response carries a top-level reasoning:{mode:pro,effort:max}.
inner := `{"id":"resp_astra","object":"response","created_at":0,"status":"completed","model":"gpt-6-astra","reasoning":{"mode":"pro","effort":"max"},"output":[{"type":"reasoning","id":"rs_1","summary":[{"type":"summary_text","text":"think"}]},{"type":"message","id":"msg_1","role":"assistant","content":[{"type":"output_text","text":"hi"}]}],"usage":{"input_tokens":1,"output_tokens":1,"total_tokens":2}}`
s.upstream.resp = &http.Response{
StatusCode: http.StatusOK,
Header: http.Header{"Content-Type": []string{"text/event-stream"}},
Body: io.NopCloser(strings.NewReader(codexCompletedSSE(inner))),
}
body := astraRequestBody("gpt-6-astra", false, "pro", "max")
result, err := s.svc.Forward(context.Background(), s.c, s.account, body)
require.NoError(t, err)
require.NotNil(t, result)
downstream := s.rec.Body.String()
// Downstream is JSON (SSE->JSON aggregation for the non-streaming client).
require.True(t, gjson.Valid(downstream), "downstream must be aggregated JSON, got: %s", downstream)
require.Equal(t, "pro", gjson.Get(downstream, "reasoning.mode").String(),
"aggregated response object must keep reasoning.mode")
require.Equal(t, "max", gjson.Get(downstream, "reasoning.effort").String())
}
// TestForward_AstraOAuth_Stream_PreservesResponseReasoningMode drives client
// stream=true and asserts the streamed response.completed event's response object
// still carries response.reasoning.mode.
func TestForward_AstraOAuth_Stream_PreservesResponseReasoningMode(t *testing.T) {
gin.SetMode(gin.TestMode)
s := newAstraOAuthSetup(t, false)
inner := `{"id":"resp_astra","object":"response","created_at":0,"status":"completed","model":"gpt-6-astra","reasoning":{"mode":"pro","effort":"max"},"output":[{"type":"message","id":"msg_1","role":"assistant","content":[{"type":"output_text","text":"hi"}]}],"usage":{"input_tokens":1,"output_tokens":1,"total_tokens":2}}`
s.upstream.resp = &http.Response{
StatusCode: http.StatusOK,
Header: http.Header{"Content-Type": []string{"text/event-stream"}},
Body: io.NopCloser(strings.NewReader(codexCompletedSSE(inner))),
}
body := astraRequestBody("gpt-6-astra", true, "pro", "max")
result, err := s.svc.Forward(context.Background(), s.c, s.account, body)
require.NoError(t, err)
require.NotNil(t, result)
downstream := s.rec.Body.String()
require.Contains(t, downstream, "response.completed", "streaming downstream must include completed event")
// Extract response.completed line and assert response.reasoning.mode inside it.
var completedJSON string
for _, line := range strings.Split(downstream, "\n") {
if strings.HasPrefix(line, "data: ") && strings.Contains(line, `"type":"response.completed"`) {
completedJSON = strings.TrimPrefix(line, "data: ")
break
}
}
require.NotEmpty(t, completedJSON, "no response.completed data line found in: %s", downstream)
require.Equal(t, "pro", gjson.Get(completedJSON, "response.reasoning.mode").String(),
"streamed completed response object must keep reasoning.mode")
require.Equal(t, "max", gjson.Get(completedJSON, "response.reasoning.effort").String())
}
@@ -0,0 +1,92 @@
package service
import (
"bytes"
"context"
"net/http"
"net/http/httptest"
"testing"
"github.com/gin-gonic/gin"
"github.com/stretchr/testify/require"
"github.com/tidwall/gjson"
)
const openAICodexResponsesTestURL = "https://chatgpt.com/backend-api/codex/responses"
// newOpenAIAstraProRetryContext returns a gin context plus its recorder so the
// caller can assert the service-layer client error shape (type/code/param/message).
func newOpenAIAstraProRetryContext(body []byte) (*gin.Context, *httptest.ResponseRecorder) {
gin.SetMode(gin.TestMode)
recorder := httptest.NewRecorder()
c, _ := gin.CreateTestContext(recorder)
c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", bytes.NewReader(body))
c.Request.Header.Set("Content-Type", "application/json")
return c, recorder
}
// TestOpenAIGatewayService_OAuthAstraProModeRejectionPassesThrough verifies that
// when the Codex OAuth upstream rejects an Astra pro+max request with a
// deterministic HTTP 400 on error.param=reasoning.mode, the gateway passes the
// client error through unchanged (upstream type/code/param/message, code not
// treated as official), makes exactly one upstream call, keeps reasoning.mode=pro
// and reasoning.effort=max in the sent body, and does not replay without mode.
func TestOpenAIGatewayService_OAuthAstraProModeRejectionPassesThrough(t *testing.T) {
body := []byte(`{"model":"gpt-6-astra","reasoning":{"mode":"pro","effort":"max"},"input":"hello"}`)
upstream := &httpUpstreamRecorder{resp: newOpenAIRejectedFieldTestResponse(
http.StatusBadRequest,
`{"error":{"type":"invalid_request_error","code":"model_specific_rejection","message":"reasoning.mode is not supported for this model","param":"reasoning.mode"}}`,
)}
svc := newOpenAIRejectedFieldTestService(upstream)
c, recorder := newOpenAIAstraProRetryContext(body)
_, err := svc.Forward(context.Background(), c, newOpenAIOAuthNamespaceTestAccount(), body)
require.Error(t, err)
require.Equal(t, http.StatusBadRequest, recorder.Code)
require.Len(t, upstream.bodies, 1, "deterministic 400 must not trigger a retry")
require.Len(t, upstream.requests, 1)
require.Equal(t, openAICodexResponsesTestURL, upstream.requests[0].URL.String())
sent := upstream.bodies[0]
require.Equal(t, "gpt-6-astra", gjson.GetBytes(sent, "model").String())
require.Equal(t, "pro", gjson.GetBytes(sent, "reasoning.mode").String())
require.Equal(t, "max", gjson.GetBytes(sent, "reasoning.effort").String())
respJSON := recorder.Body.String()
require.Equal(t, "invalid_request_error", gjson.Get(respJSON, "error.type").String())
require.Equal(t, "model_specific_rejection", gjson.Get(respJSON, "error.code").String())
require.Equal(t, "reasoning.mode", gjson.Get(respJSON, "error.param").String())
require.Equal(t, "reasoning.mode is not supported for this model", gjson.Get(respJSON, "error.message").String())
}
// TestOpenAIGatewayService_OAuthAstraProModeKeptAcrossRejectedFieldRetry reuses
// the existing OAuth rejected-field retry fixture (input[0].status is rejected,
// stripped, then replayed to success) to confirm Astra pro+max reasoning is kept
// across every in-service retry: each captured Codex URL request body still
// carries reasoning.mode=pro and reasoning.effort=max with model=gpt-6-astra,
// and only the second attempt drops the rejected status field.
func TestOpenAIGatewayService_OAuthAstraProModeKeptAcrossRejectedFieldRetry(t *testing.T) {
body := []byte(`{"model":"gpt-6-astra","stream":true,"instructions":"test","reasoning":{"mode":"pro","effort":"max"},"input":[{"type":"message","role":"user","status":"completed","content":"hello"}]}`)
upstream := &httpUpstreamRecorder{responses: []*http.Response{
newOpenAIRejectedFieldTestResponse(http.StatusBadRequest, `{"error":{"code":"unknown_parameter","message":"Unknown parameter: 'input[0].status'.","param":"input[0].status"}}`),
newOpenAIRejectedFieldTestResponse(http.StatusOK, "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_ok\",\"output\":[],\"usage\":{\"input_tokens\":1,\"output_tokens\":1,\"total_tokens\":2}}}\n\ndata: [DONE]\n\n"),
}}
upstream.responses[1].Header.Set("Content-Type", "text/event-stream")
svc := newOpenAIRejectedFieldTestService(upstream)
c, _ := newOpenAIAstraProRetryContext(body)
result, err := svc.Forward(context.Background(), c, newOpenAIOAuthNamespaceTestAccount(), body)
require.NoError(t, err)
require.NotNil(t, result)
require.Len(t, upstream.bodies, 2)
require.Len(t, upstream.requests, 2)
for i := range upstream.bodies {
require.Equal(t, openAICodexResponsesTestURL, upstream.requests[i].URL.String(), "attempt %d URL", i)
sent := upstream.bodies[i]
require.Equal(t, "gpt-6-astra", gjson.GetBytes(sent, "model").String(), "attempt %d model", i)
require.Equal(t, "pro", gjson.GetBytes(sent, "reasoning.mode").String(), "attempt %d reasoning.mode", i)
require.Equal(t, "max", gjson.GetBytes(sent, "reasoning.effort").String(), "attempt %d reasoning.effort", i)
}
require.Equal(t, "completed", gjson.GetBytes(upstream.bodies[0], "input.0.status").String())
require.False(t, gjson.GetBytes(upstream.bodies[1], "input.0.status").Exists())
}
@@ -1023,6 +1023,10 @@ func normalizeOpenAIResponsesReasoningMode(body []byte) ([]byte, bool, error) {
if len(body) == 0 {
return body, false, nil
}
// Astra 的 reasoning.mode 与 reasoning.effort 是独立参数,不做兼容替换;非 Astra 维持旧 strip-mode/pro->max 行为。
if isOpenAIGPT6AstraModel(gjson.GetBytes(body, "model").String()) {
return body, false, nil
}
mode := gjson.GetBytes(body, "reasoning.mode")
if !mode.Exists() || mode.Type != gjson.String {
return body, false, nil
@@ -138,6 +138,69 @@ func TestNormalizeOpenAIResponsesReasoningMode(t *testing.T) {
}
}
func TestNormalizeOpenAIResponsesReasoningMode_AstraPreservesBody(t *testing.T) {
// GPT-6 Astra 保留官方 reasoning.mode 与 reasoning.effort 各自原样:
// 不删 mode、缺失 effort 也不补 max。非 Astra 的旧兼容行为不受影响。
tests := []struct {
name string
body string
}{
{name: "pro + max preserved", body: `{"model":"gpt-6-astra","reasoning":{"mode":"pro","effort":"max"}}`},
{name: "standard + max preserved", body: `{"model":"gpt-6-astra","reasoning":{"mode":"standard","effort":"max"}}`},
{name: "missing mode + max preserved", body: `{"model":"gpt-6-astra","reasoning":{"effort":"max"}}`},
{name: "pro + explicit high preserved", body: `{"model":"gpt-6-astra","reasoning":{"mode":"pro","effort":"high"}}`},
{name: "pro without effort does not inject max", body: `{"model":"gpt-6-astra","reasoning":{"mode":"pro"}}`},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
normalized, changed, err := normalizeOpenAIResponsesReasoningMode([]byte(tt.body))
require.NoError(t, err)
require.False(t, changed)
require.JSONEq(t, tt.body, string(normalized))
})
}
}
func TestNormalizeOpenAIResponsesReasoningMode_NonAstraKeepsLegacyBehavior(t *testing.T) {
// 确保 guard 只放过 Astra,非 Astra model 字段不影响既有 strip/pro->max 语义。
tests := []struct {
name string
body string
wantEffort string
}{
{name: "gpt-5.6-sol pro maps to max", body: `{"model":"gpt-5.6-sol","reasoning":{"mode":"pro"}}`, wantEffort: "max"},
{name: "gpt-5.4 explicit effort wins", body: `{"model":"gpt-5.4","reasoning":{"mode":"pro","effort":"high"}}`, wantEffort: "high"},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
normalized, changed, err := normalizeOpenAIResponsesReasoningMode([]byte(tt.body))
require.NoError(t, err)
require.True(t, changed)
require.False(t, gjson.GetBytes(normalized, "reasoning.mode").Exists())
require.Equal(t, tt.wantEffort, gjson.GetBytes(normalized, "reasoning.effort").String())
})
}
}
func TestNormalizeOpenAIPassthroughOAuthBody_AstraPreservesReasoningMode(t *testing.T) {
body := []byte(`{"model":"gpt-6-astra","reasoning":{"mode":"pro","effort":"max"}}`)
normalized, _, err := normalizeOpenAIPassthroughOAuthBody(body, false)
require.NoError(t, err)
require.Equal(t, "pro", gjson.GetBytes(normalized, "reasoning.mode").String())
require.Equal(t, "max", gjson.GetBytes(normalized, "reasoning.effort").String())
}
func TestNormalizeOpenAIResponsesWebSocketCompatibilityBody_AstraPreservesReasoningMode(t *testing.T) {
body := []byte(`{"type":"response.create","model":"gpt-6-astra","reasoning":{"mode":"pro","effort":"max"}}`)
for _, accountType := range []string{AccountTypeOAuth, AccountTypeSetupToken} {
normalized, _, err := normalizeOpenAIResponsesWebSocketCompatibilityBody(body, &Account{Platform: PlatformOpenAI, Type: accountType}, false)
require.NoError(t, err)
require.Equal(t, "pro", gjson.GetBytes(normalized, "reasoning.mode").String())
require.Equal(t, "max", gjson.GetBytes(normalized, "reasoning.effort").String())
}
}
func TestNormalizeOpenAIResponsesWebSocketCompatibilityBody_ReasoningModeAccountScope(t *testing.T) {
body := []byte(`{"type":"response.create","reasoning":{"mode":"pro"}}`)
for _, accountType := range []string{AccountTypeOAuth, AccountTypeSetupToken} {