feat(codex): zstd-compress Codex OAuth turn bodies and stop advertising gzip upstream

backend/internal/config/config.go

@@ -980,6 +980,11 @@ type GatewayConfig struct {
 	// ForceCodexCLI: 强制将 OpenAI `/v1/responses` 请求按 Codex CLI 处理。
 	// 用于网关未透传/改写 User-Agent 时的兼容兜底（默认关闭，避免影响其他客户端）。
 	ForceCodexCLI bool `mapstructure:"force_codex_cli"`
+	// OpenAICodexRequestZstd: 对 OpenAI OAuth-like 账号发往 ChatGPT 后端的流式 /responses 请求体
+	// 做 zstd（level 3）压缩并带 Content-Encoding: zstd，与官方 Codex 客户端一致；compact 等其他
+	// 请求不压缩。默认开启；出现兼容问题时置 false 回退为明文 JSON。零值为关闭，手工构造的
+	// Config 不会意外压缩。
+	OpenAICodexRequestZstd bool `mapstructure:"openai_codex_request_zstd"`
 	// DisableCodexIdentityEnforcement: 关闭「强制统一 Codex 出站身份」。上游 /backend-api/codex
 	// 在容量紧张时按客户端身份分优先级降载，被降载的请求会拿到 HTTP 200 + 流内
 	// server_is_overloaded，该次请求失败。默认强制统一出口：所有 OAuth 出站的
@@ -2383,6 +2388,7 @@ func setDefaults() {
 	viper.SetDefault("gateway.max_account_switches", 10)
 	viper.SetDefault("gateway.max_account_switches_gemini", 3)
 	viper.SetDefault("gateway.force_codex_cli", false)
+	viper.SetDefault("gateway.openai_codex_request_zstd", true)
 	viper.SetDefault("gateway.disable_codex_identity_enforcement", false)
 	viper.SetDefault("gateway.disable_codex_originator_normalization", false)
 	viper.SetDefault("gateway.codex_image_generation_bridge_enabled", false)

backend/internal/repository/http_upstream.go

@@ -1335,6 +1335,12 @@ func buildUpstreamTransport(settings poolSettings, proxyURL *url.URL, protocolMo
 		IdleConnTimeout:       settings.idleConnTimeout,
 		ResponseHeaderTimeout: settings.responseHeaderTimeout,
 	}
+	if isOpenAIUpstreamProtocolMode(protocolMode) {
+		// 真实 Codex 客户端（reqwest，未启用压缩特性）不发送 Accept-Encoding；关闭 Go 的自动
+		// gzip 协商，避免出站多出一条 `Accept-Encoding: gzip`。上游若仍返回 Content-Encoding，
+		// 由 decompressResponseBody 手动解压，业务层语义不变。
+		transport.DisableCompression = true
+	}
 	switch protocolMode {
 	case upstreamProtocolModeLongStreamH2, upstreamProtocolModeOpenAIH2:
 		transport.ForceAttemptHTTP2 = true
@@ -1357,6 +1363,15 @@ func buildUpstreamTransport(settings poolSettings, proxyURL *url.URL, protocolMo
 	return transport, nil
 }
 
+// isOpenAIUpstreamProtocolMode 判断协议模式是否属于 OpenAI 专用传输（H2 / H1 / H1 回退）。
+func isOpenAIUpstreamProtocolMode(mode string) bool {
+	switch mode {
+	case upstreamProtocolModeOpenAIH1, upstreamProtocolModeOpenAIH2, upstreamProtocolModeOpenAIH1Fallback:
+		return true
+	}
+	return false
+}
+
 // enableHTTP2KeepAlive 在 http.Transport 上显式配置 HTTP/2 并启用连接健康探测。
 // Go 默认惰性配置 http2 且 ReadIdleTimeout=0（不发健康 PING），无法检测被代理/NAT
 // 静默掐断的死连接。此处主动设置 ReadIdleTimeout/PingTimeout，让死连接被提前 PING

backend/internal/repository/http_upstream_compression_test.go

@@ -0,0 +1,19 @@
+package repository
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+// OpenAI 专用传输不再让 Go 自动附加 Accept-Encoding: gzip；默认传输保持原样。
+func TestBuildUpstreamTransportDisablesCompressionForOpenAIModes(t *testing.T) {
+	for _, mode := range []string{upstreamProtocolModeOpenAIH2, upstreamProtocolModeOpenAIH1, upstreamProtocolModeOpenAIH1Fallback} {
+		transport, err := buildUpstreamTransport(defaultPoolSettings(nil), nil, mode)
+		require.NoError(t, err, mode)
+		require.True(t, transport.DisableCompression, mode)
+	}
+	transport, err := buildUpstreamTransport(defaultPoolSettings(nil), nil, upstreamProtocolModeDefault)
+	require.NoError(t, err)
+	require.False(t, transport.DisableCompression)
+}

backend/internal/service/openai_alpha_search.go

@@ -98,7 +98,7 @@ func (s *OpenAIGatewayService) ForwardAlphaSearch(ctx context.Context, c *gin.Co
 	}
 
 	upstreamStart := time.Now()
-	resp, err := s.httpUpstream.Do(req, proxyURL, account.ID, account.Concurrency)
+	resp, err := s.doOpenAICodexUpstream(req, account, proxyURL)
 	SetOpsLatencyMs(c, OpsUpstreamLatencyMsKey, time.Since(upstreamStart).Milliseconds())
 	if err != nil {
 		return nil, s.handleOpenAIUpstreamTransportError(ctx, c, account, err, true)
@@ -200,7 +200,7 @@ func (s *OpenAIGatewayService) forwardCindyAlphaSearchViaNativeMessages(
 	SetActualOpenAIUpstreamEndpoint(c, "/v1/messages")
 
 	upstreamStart := time.Now()
-	resp, err := s.httpUpstream.Do(req, proxyURL, account.ID, account.Concurrency)
+	resp, err := s.doOpenAICodexUpstream(req, account, proxyURL)
 	SetOpsLatencyMs(c, OpsUpstreamLatencyMsKey, time.Since(upstreamStart).Milliseconds())
 	if err != nil {
 		return nil, s.handleOpenAIUpstreamTransportError(ctx, c, account, err, true)
@@ -489,7 +489,7 @@ func (s *OpenAIGatewayService) forwardAlphaSearchViaResponsesWebSearch(
 	SetActualOpenAIUpstreamEndpoint(c, "/v1/responses")
 
 	upstreamStart := time.Now()
-	resp, err := s.httpUpstream.Do(req, proxyURL, account.ID, account.Concurrency)
+	resp, err := s.doOpenAICodexUpstream(req, account, proxyURL)
 	SetOpsLatencyMs(c, OpsUpstreamLatencyMsKey, time.Since(upstreamStart).Milliseconds())
 	if err != nil {
 		return nil, s.handleOpenAIUpstreamTransportError(ctx, c, account, err, true)

backend/internal/service/openai_alpha_search_test.go

@@ -15,6 +15,7 @@ import (
 	"github.com/Wei-Shaw/sub2api/internal/config"
 	"github.com/Wei-Shaw/sub2api/internal/pkg/openai"
 	"github.com/gin-gonic/gin"
+	"github.com/klauspost/compress/zstd"
 	"github.com/stretchr/testify/require"
 	"github.com/tidwall/gjson"
 )
@@ -136,6 +137,7 @@ func TestForwardAlphaSearchPATUsesResponsesWebSearchFallback(t *testing.T) {
 		Body:       io.NopCloser(strings.NewReader(alphaSearchResponsesSSE("search result"))),
 	}}
 	service := &OpenAIGatewayService{cfg: &config.Config{}, httpUpstream: upstream}
+	service.cfg.Gateway.OpenAICodexRequestZstd = true
 	account := &Account{
 		ID:          43,
 		Platform:    PlatformOpenAI,
@@ -174,13 +176,21 @@ func TestForwardAlphaSearchPATUsesResponsesWebSearchFallback(t *testing.T) {
 	require.Empty(t, upstream.lastReq.Header.Get("X-Codex-Turn-State"))
 	require.Empty(t, upstream.lastReq.Header.Get(responsesLiteHeaderKey))
 	require.Empty(t, upstream.lastReq.Header.Get("Accept-Language"))
-	require.False(t, gjson.GetBytes(upstream.lastBody, "prompt_cache_key").Exists())
-	require.False(t, gjson.GetBytes(upstream.lastBody, "prompt_cache_retention").Exists())
-	require.Equal(t, "gpt-5.6-sol", gjson.GetBytes(upstream.lastBody, "model").String())
-	require.True(t, gjson.GetBytes(upstream.lastBody, "stream").Bool())
-	require.False(t, gjson.GetBytes(upstream.lastBody, "store").Bool())
-	require.Equal(t, "web_search", gjson.GetBytes(upstream.lastBody, "tools.0.type").String())
-	require.Contains(t, gjson.GetBytes(upstream.lastBody, "input.0.content.0.text").String(), `"search_query"`)
+	// PAT/OAuth 的 Responses web-search fallback 是真实 Codex turn：开关开启时同样 zstd 出站，
+	// 解压后语义不变。
+	require.Equal(t, "zstd", upstream.lastReq.Header.Get("Content-Encoding"))
+	zstdReader, err := zstd.NewReader(bytes.NewReader(upstream.lastBody))
+	require.NoError(t, err)
+	wireBody, err := io.ReadAll(zstdReader)
+	zstdReader.Close()
+	require.NoError(t, err)
+	require.False(t, gjson.GetBytes(wireBody, "prompt_cache_key").Exists())
+	require.False(t, gjson.GetBytes(wireBody, "prompt_cache_retention").Exists())
+	require.Equal(t, "gpt-5.6-sol", gjson.GetBytes(wireBody, "model").String())
+	require.True(t, gjson.GetBytes(wireBody, "stream").Bool())
+	require.False(t, gjson.GetBytes(wireBody, "store").Bool())
+	require.Equal(t, "web_search", gjson.GetBytes(wireBody, "tools.0.type").String())
+	require.Contains(t, gjson.GetBytes(wireBody, "input.0.content.0.text").String(), `"search_query"`)
 }
 
 func TestForwardAlphaSearchOrdinaryAPIKeyIgnoresLegacyBridgeSettings(t *testing.T) {

backend/internal/service/openai_codex_request_compression.go

@@ -0,0 +1,145 @@
+package service
+
+import (
+	"bytes"
+	"errors"
+	"fmt"
+	"io"
+	"log/slog"
+	"net/http"
+	"strings"
+	"sync"
+
+	"github.com/klauspost/compress/zstd"
+)
+
+// 真实 Codex 客户端（codex-rs http-client）只对 ChatGPT 后端的流式 /responses 请求体做
+// zstd level 3 压缩并带 Content-Encoding: zstd；compact、models 等请求保持明文。网关在最终
+// 发送前对 OAuth-like 账号的同类请求做同样处理，并且只改写发出的副本：调用方持有的请求
+// 对象保持明文，推理恢复、诊断快照和重试路径继续按 JSON 读取。
+
+// codexRequestZstdWindowSize 与 zstd level 3 的默认 windowLog=21 一致。
+const codexRequestZstdWindowSize = 1 << 21
+
+// codexRequestZstdEncoders 复用流式编码器，避免每次请求分配窗口缓冲。帧形态与官方客户端
+// （Rust `zstd::stream::encode_all(body, 3)`）的差异如实记录：两者都不写校验和、窗口 2MB；
+// klauspost 对能一次编码完的请求体（小于一个块、约 128KB 以内）会在帧头写入内容长度并使用
+// single-segment，Rust 流式编码器不写内容长度。这是合法的 zstd 帧差异，不影响解压。
+var codexRequestZstdEncoders = sync.Pool{New: func() any {
+	enc, err := zstd.NewWriter(nil,
+		zstd.WithEncoderLevel(zstd.SpeedDefault),
+		zstd.WithEncoderCRC(false),
+		zstd.WithWindowSize(codexRequestZstdWindowSize),
+		zstd.WithEncoderConcurrency(1),
+	)
+	if err != nil {
+		return nil
+	}
+	return enc
+}}
+
+func compressCodexRequestBodyZstd(src []byte) ([]byte, error) {
+	enc, _ := codexRequestZstdEncoders.Get().(*zstd.Encoder)
+	if enc == nil {
+		return nil, errors.New("zstd encoder unavailable")
+	}
+	defer codexRequestZstdEncoders.Put(enc)
+	var buf bytes.Buffer
+	buf.Grow(len(src)/3 + 64)
+	enc.Reset(&buf)
+	if _, err := enc.Write(src); err != nil {
+		return nil, err
+	}
+	if err := enc.Close(); err != nil {
+		return nil, err
+	}
+	return buf.Bytes(), nil
+}
+
+func (s *OpenAIGatewayService) codexRequestZstdEnabled() bool {
+	return s != nil && s.cfg != nil && s.cfg.Gateway.OpenAICodexRequestZstd
+}
+
+// isCodexStreamingResponsesRequest 判断请求是否是官方客户端会压缩的那一类：
+// POST 到 ChatGPT 后端 /backend-api/codex/responses（不含 /compact）。
+func isCodexStreamingResponsesRequest(req *http.Request) bool {
+	if req == nil || req.URL == nil || req.Method != http.MethodPost {
+		return false
+	}
+	path := strings.TrimSuffix(req.URL.Path, "/")
+	return strings.HasSuffix(path, "/codex/responses")
+}
+
+// prepareOpenAICodexWireRequest 返回真正发往上游的请求。满足条件时返回一个请求体已
+// zstd 压缩、Content-Encoding/Content-Length 已改写的克隆；其余情况原样返回。
+//
+// 明文来源：可重放请求（GetBody 非空）从 GetBody 的独立副本取明文，原请求 Body 完全不动；
+// 不可重放请求（GetBody 已被 PrepareRequest 清空）只能读取 Body，读取成功后用同一份明文
+// 重新填充。读取失败时绝不把已读出的前缀当作完整请求发送：可重放请求退回原请求明文发送，
+// 不可重放请求显式返回错误（该请求本就无法完整发出）。编码失败一律退回明文。
+func (s *OpenAIGatewayService) prepareOpenAICodexWireRequest(req *http.Request, account *Account) (*http.Request, error) {
+	if !s.codexRequestZstdEnabled() || account == nil || !account.IsOpenAIOAuthLike() {
+		return req, nil
+	}
+	if !isCodexStreamingResponsesRequest(req) || req.Body == nil || req.Body == http.NoBody {
+		return req, nil
+	}
+	if strings.TrimSpace(req.Header.Get("Content-Encoding")) != "" {
+		return req, nil
+	}
+	if contentType := strings.ToLower(req.Header.Get("Content-Type")); contentType != "" && !strings.Contains(contentType, "json") {
+		return req, nil
+	}
+	var raw []byte
+	if req.GetBody != nil {
+		snapshot, err := req.GetBody()
+		if err != nil {
+			slog.Debug("codex_request_zstd_skipped", "reason", "snapshot_body", "error", err)
+			return req, nil
+		}
+		raw, err = io.ReadAll(snapshot)
+		_ = snapshot.Close()
+		if err != nil {
+			slog.Debug("codex_request_zstd_skipped", "reason", "read_snapshot", "error", err)
+			return req, nil
+		}
+	} else {
+		var err error
+		raw, err = io.ReadAll(req.Body)
+		_ = req.Body.Close()
+		if err != nil {
+			return nil, fmt.Errorf("codex request zstd: read request body: %w", err)
+		}
+		req.Body = io.NopCloser(bytes.NewReader(raw))
+	}
+	if len(raw) == 0 {
+		return req, nil
+	}
+	compressed, err := compressCodexRequestBodyZstd(raw)
+	if err != nil {
+		slog.Debug("codex_request_zstd_skipped", "reason", "encode", "error", err)
+		return req, nil
+	}
+	wire := req.Clone(req.Context())
+	wire.Body = io.NopCloser(bytes.NewReader(compressed))
+	wire.ContentLength = int64(len(compressed))
+	wire.Header.Set("Content-Encoding", "zstd")
+	if req.GetBody != nil {
+		wire.GetBody = func() (io.ReadCloser, error) {
+			return io.NopCloser(bytes.NewReader(compressed)), nil
+		}
+	} else {
+		wire.GetBody = nil
+	}
+	return wire, nil
+}
+
+// doOpenAICodexUpstream 是 OpenAI 网关所有 Responses 端点 POST 的统一发送入口：先按上述
+// 规则生成线上请求（满足条件时压缩），再交给 httpUpstream。读取失败作为传输错误返回。
+func (s *OpenAIGatewayService) doOpenAICodexUpstream(req *http.Request, account *Account, proxyURL string) (*http.Response, error) {
+	wire, err := s.prepareOpenAICodexWireRequest(req, account)
+	if err != nil {
+		return nil, err
+	}
+	return s.httpUpstream.Do(wire, proxyURL, account.ID, account.Concurrency)
+}