fix: inspect non-replayable Codex diagnostic bodies

.downstream/codex-quality-diagnostics.md

@@ -44,6 +44,10 @@ python scripts/invoke-codex-quality-acceptance.py status
 
 `reconcile` 不发送模型请求。只有收到应用明确的发送前拒绝、服务器账本也无该 trial，才允许保留原记录并安排新 trial；网络超时或代理错误不能当作“没发送”。
 
+诊断读取最终请求时兼容 reasoning recovery 清除 `GetBody` 的不可重放请求：有界读取实际编码字节后原样恢复 `Body`，保持 `GetBody=nil`、编码头和长度；编码体与解码体分别最多 2 MiB 和 1 MiB。读取或关闭失败时阻止发送，不预留生成预算，不重新启用透明 POST 重放。
+
+如果诊断本身在第一次业务发送前失败，发布者可在仅修诊断的新版本上线并完成旧进程排空后，显式执行 `restart-baseline --version <version> --source-sha <sha> --reason <reason>`。入口要求旧阶段无业务回执、无未结束操作，完整保留旧阶段及未确认结果，为旧未确认 trial 保守占用预算，再建立新 trial/turn 命名空间。它不发模型请求、不重置服务器账本；迟到的旧回执会阻止后续生成。普通 `baseline` 和 `verification` 不能隐式改变已有阶段的源码版本。
+
 ## 发布次序
 
 诊断基础设施先独立发布并采集修前基线。协议与账号身份修复随后发布，在共同宿主基线、同一账号和同一预算下完成修后序列。不能先应用修复再把结果命名为修前基线。

backend/internal/service/codex_quality_observation.go

@@ -16,17 +16,42 @@ import (
 )
 
 func codexQualityWireFields(req *http.Request) (string, string, error) {
-	if req == nil || req.GetBody == nil {
+	if req == nil {
 		return "", "", ErrCodexQualityUnavailable
 	}
-	body, err := req.GetBody()
-	if err != nil {
+	// Reasoning recovery deliberately clears GetBody to prevent transparent POST
+	// replay. Inspect the final encoded body without restoring that capability.
+	body := req.Body
+	independent := req.GetBody != nil
+	if independent {
+		var err error
+		body, err = req.GetBody()
+		if err != nil {
+			if body != nil {
+				_ = body.Close()
+			}
+			return "", "", ErrCodexQualityUnavailable
+		}
+	}
+	if body == nil {
 		return "", "", ErrCodexQualityUnavailable
 	}
-	defer func() { _ = body.Close() }()
-	var reader io.Reader = body
+	wire, readErr := io.ReadAll(io.LimitReader(body, (2<<20)+1))
+	closeErr := body.Close()
+	if readErr != nil || closeErr != nil || len(wire) > 2<<20 {
+		if !independent {
+			// The original close chain has run. Neither a consumed prefix nor an
+			// unread suffix may become a later request after a failed snapshot.
+			req.Body = io.NopCloser(codexQualityRejectedWireBody{})
+		}
+		return "", "", ErrCodexQualityUnavailable
+	}
+	if !independent {
+		req.Body = io.NopCloser(bytes.NewReader(wire))
+	}
+	var reader io.Reader = bytes.NewReader(wire)
 	if strings.EqualFold(req.Header.Get("Content-Encoding"), "zstd") {
-		decoder, err := zstd.NewReader(body, zstd.WithDecoderMaxMemory(8<<20))
+		decoder, err := zstd.NewReader(reader, zstd.WithDecoderMaxMemory(8<<20), zstd.WithDecoderConcurrency(1))
 		if err != nil {
 			return "", "", ErrCodexQualityUnavailable
 		}
@@ -40,6 +65,12 @@ func codexQualityWireFields(req *http.Request) (string, string, error) {
 	return gjson.GetBytes(raw, "model").String(), gjson.GetBytes(raw, "reasoning.effort").String(), nil
 }
 
+type codexQualityRejectedWireBody struct{}
+
+func (codexQualityRejectedWireBody) Read([]byte) (int, error) {
+	return 0, ErrCodexQualityUnavailable
+}
+
 func reserveCodexQualityAcquisition(req *http.Request, accountID int64) error {
 	e := codexQualityExecutionFromContext(req.Context())
 	if e == nil {

backend/internal/service/codex_quality_wire_test.go

@@ -0,0 +1,275 @@
+//go:build unit
+
+package service
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/Wei-Shaw/sub2api/internal/config"
+	"github.com/Wei-Shaw/sub2api/internal/pkg/tlsfingerprint"
+	extensionv1 "github.com/Wei-Shaw/sub2api/pkg/extensionapi/v1"
+	"github.com/gin-gonic/gin"
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/require"
+)
+
+const qualityWireFixtureBody = `{"model":"gpt-6-astra","reasoning":{"effort":"high"},"input":"question","stream":true,"store":false}`
+
+type qualityWireUpstream struct {
+	HTTPUpstream
+	accountID int64
+	scope     extensionv1.CodexRoutingScope
+	calls     int
+	request   *http.Request
+	body      []byte
+}
+
+func (u *qualityWireUpstream) HasCodexQualityConnection(id string, accountID int64, scope string) bool {
+	return id == u.scope.ConnectionLeaseID && accountID == u.accountID && scope == codexRoutingScopeKey(u.scope)
+}
+
+func (u *qualityWireUpstream) DoWithCodexConnectionLease(req *http.Request, _ string, accountID int64, scope, id string, _ time.Time, _ *tlsfingerprint.Profile) (*http.Response, string, error) {
+	if !u.HasCodexQualityConnection(id, accountID, scope) {
+		return nil, "", ErrCodexConnectionLeaseExpired
+	}
+	u.calls++
+	u.request = req
+	defer func() { _ = req.Body.Close() }()
+	var err error
+	u.body, err = io.ReadAll(req.Body)
+	if err != nil {
+		return nil, "", err
+	}
+	return &http.Response{StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"text/event-stream"}},
+		Body: io.NopCloser(strings.NewReader("data: {\"type\":\"response.completed\",\"response\":{\"status\":\"completed\",\"model\":\"gpt-6-astra\"}}\n\n"))}, id, nil
+}
+
+type qualityWireSource struct {
+	reader   io.Reader
+	readErr  error
+	closeErr error
+	read     int
+	closed   int
+}
+
+func (s *qualityWireSource) Read(p []byte) (int, error) {
+	n, err := s.reader.Read(p)
+	s.read += n
+	if err == io.EOF && s.readErr != nil {
+		return n, s.readErr
+	}
+	return n, err
+}
+
+func (s *qualityWireSource) Close() error {
+	s.closed++
+	return s.closeErr
+}
+
+// Use the real private transport and durable reservation boundary. Only the
+// database and network ports are in-memory fixtures; no model call is possible.
+func qualityWireServiceFixture(t *testing.T, compressed, recovery bool) (*OpenAIGatewayService, *Account, *routingMemoryStore, codexQualityRun, *qualityWireUpstream, context.Context, *http.Request) {
+	t.Helper()
+	group, proxy := int64(53), int64(34)
+	account := &Account{ID: 16380, Platform: PlatformOpenAI, Type: AccountTypeOAuth, Status: StatusActive,
+		Schedulable: false, Concurrency: 1, GroupIDs: []int64{group}, ProxyID: &proxy,
+		Credentials: map[string]any{"plan_type": "pro", "chatgpt_account_id": "quality-wire-fixture"}}
+	if !recovery {
+		account.Extra = map[string]any{OpenAIReasoningSignatureRecoveryEnabledExtraKey: false}
+	}
+	manager := ticketTestManager(t, config.OpenAICodexTicketConfig{Enabled: true, FailClosed: true, Models: []string{codexQualityModel}}, nil)
+	installation := manager.extensions.Load().installations[1]
+	installation.State = PluginStateEnabled
+	host, directory, store := routingHostFixture()
+	store.PluginRepository = &pluginTokenRepository{installation: installation}
+	manager.repo = store
+	upstream := &qualityWireUpstream{accountID: account.ID}
+	svc := &OpenAIGatewayService{cfg: &config.Config{}, pluginManager: manager, httpUpstream: upstream,
+		accountRepo: &routingAccountRepositoryFixture{account: account}}
+	ctx := withCodexTransportFixture(context.Background(), compressed)
+	scope, err := svc.PrepareCodexRoutingScope(ctx, account.ID, "http")
+	require.NoError(t, err)
+	directory.scope, directory.account = scope, *extensionAccount(account)
+	host.installation = installation
+	installation.Manifest.Capabilities = []PluginCapability{{ID: extensionv1.CapabilityCredentials, Platform: PlatformOpenAI, AccountType: AccountTypeOAuth}}
+	now := time.Now().UTC()
+	expiry := now.Add(time.Minute)
+	leaseScope := scope
+	leaseScope.ConnectionLeaseID, leaseScope.RouteEvidence = "quality-wire-connection", "connection"
+	upstream.scope = leaseScope
+	cookie := codexRoutingCookie{Name: "__cflb", Value: "quality-wire-route", Domain: "chatgpt.com", Path: "/", FirstSeen: now, ExpiresAt: expiry}
+	clock := codexRoutingCookieClock{Scope: scope}
+	clock.apply([]codexRoutingCookieChange{{Cookie: cookie}}, now)
+	clockKey := "clock.quality.wire-fixture"
+	put := func(key string, value any) {
+		raw, marshalErr := json.Marshal(value)
+		require.NoError(t, marshalErr)
+		_, writeErr := store.CompareSwapExtensionState(ctx, codexRuntimePluginKey, extensionv1.StateRequest{Namespace: codexRoutingPrivateNamespace, Key: key, Value: raw})
+		require.NoError(t, writeErr)
+	}
+	put(clockKey, clock)
+	put("bundle.quality.wire-fixture", codexRoutingPrivateBundle{Schema: extensionv1.CodexRoutingSchema, Scope: leaseScope,
+		Cookies: []codexRoutingCookie{cookie}, ClockKey: clockKey, ClockRevision: 1, Status: "qualified", Model: codexQualityModel, ExpiresAt: expiry})
+	qualification := &extensionv1.CodexRoutingQualification{Scope: leaseScope, Model: codexQualityModel, VerifiedAt: now, ExpiresAt: expiry,
+		Bundle: extensionv1.CodexRoutingBundleRef{Key: "bundle.quality.wire-fixture", Revision: 1, ExpiresAt: expiry, ConnectionLeaseID: leaseScope.ConnectionLeaseID}}
+	run := codexQualityRun{RunID: uuid.NewString(), ActorID: 9, APIKeyID: 101, AccountID: account.ID, GroupID: group, ProxyID: proxy,
+		Scope: scope, PromptSHA256: codexQualityHash("question"), MaxSends: 60, Status: "open", Qualification: qualification,
+		RouteGeneration: 1, RouteRuntimeGeneration: installation.RuntimeGeneration}
+	_, digest, err := newCodexQualityGrant(run.RunID)
+	require.NoError(t, err)
+	run, err = issueCodexQualityGrant(ctx, store, run, digest, now.Add(time.Hour))
+	require.NoError(t, err)
+	runtime := &codexQualityRuntime{s: svc, store: store, installation: installation, host: host}
+	key := &APIKey{ID: run.APIKeyID, UserID: run.ActorID, GroupID: &group, Status: StatusAPIKeyActive}
+	execution := &codexQualityExecution{runtime: runtime, runID: run.RunID, grantDigest: run.GrantDigest, trialID: uuid.NewString(),
+		stage: "business", accountID: account.ID, qualification: qualification,
+		keyLookup: func(context.Context, int64) (*APIKey, error) { return key, nil }}
+	ctx = context.WithValue(ctx, codexQualityExecutionKey{}, execution)
+	_, err = runtime.account(ctx, run)
+	require.NoError(t, err, "the fixture must pass account checks before exercising wire inspection")
+	_, err = runtime.qualification(ctx, run)
+	require.NoError(t, err, "the fixture must have a live private route before exercising wire inspection")
+	c, _ := gin.CreateTestContext(httptest.NewRecorder())
+	c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil).WithContext(ctx)
+	c.Set("api_key", key)
+	state := svc.newOpenAIReasoningRecoveryState(ctx, c, account, "synthetic-token")
+	require.Equal(t, recovery, state.enabled, "missing account setting must use the real enabled-by-default recovery policy")
+	t.Cleanup(state.Close)
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, "https://chatgpt.com/backend-api/codex/responses", strings.NewReader(qualityWireFixtureBody))
+	require.NoError(t, err)
+	req.Header.Set("Content-Type", "application/json")
+	req = withCodexRoutingModel(req, codexQualityModel)
+	req, _, err = state.PrepareRequest(req, []byte(qualityWireFixtureBody), "")
+	require.NoError(t, err)
+	return svc, account, store, run, upstream, ctx, req
+}
+
+func TestCodexQualityWireRecoveryPreservesBodyAndReplayPolicy(t *testing.T) {
+	for _, test := range []struct {
+		name                 string
+		compressed, recovery bool
+	}{
+		{"plain_nonreplayable", false, true}, {"zstd_nonreplayable", true, true},
+		{"plain_existing_snapshot", false, false}, {"zstd_existing_snapshot", true, false},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			svc, account, store, run, upstream, ctx, req := qualityWireServiceFixture(t, test.compressed, test.recovery)
+			require.Equal(t, test.recovery, req.GetBody == nil)
+			source := &qualityWireSource{reader: req.Body}
+			req.Body = source
+			expected := []byte(qualityWireFixtureBody)
+			if test.compressed {
+				var err error
+				expected, err = compressCodexRequestBodyZstd(expected)
+				require.NoError(t, err)
+			}
+			response, err := svc.doOpenAICodexUpstream(req, account, "")
+			require.NoError(t, err)
+			require.NotNil(t, response)
+			_, err = io.Copy(io.Discard, response.Body)
+			require.NoError(t, err)
+			require.NoError(t, response.Body.Close())
+			require.Equal(t, 1, upstream.calls)
+			require.Equal(t, expected, upstream.body, "the actual transport receives the exact final wire bytes")
+			require.Equal(t, int64(len(expected)), upstream.request.ContentLength)
+			require.Equal(t, test.recovery, upstream.request.GetBody == nil, "inspection must not enable transparent replay")
+			require.Equal(t, test.recovery, HTTPUpstreamRedirectsDisabled(upstream.request.Context()))
+			if !test.compressed || test.recovery {
+				require.Equal(t, 1, source.closed, "the consumed source keeps its close obligation")
+			}
+			if test.compressed {
+				require.Equal(t, "zstd", upstream.request.Header.Get("Content-Encoding"))
+			}
+			saved, _, err := readCodexQualityRun(ctx, store, run.RunID)
+			require.NoError(t, err)
+			require.Equal(t, 1, saved.UsedSends)
+			require.Len(t, saved.Attempts, 1)
+			require.Equal(t, codexQualityModel, saved.Attempts[0].RequestModel)
+			require.Equal(t, codexQualityEffort, saved.Attempts[0].ReasoningEffort)
+			require.False(t, account.Schedulable)
+		})
+	}
+}
+
+func TestCodexQualityWireFailuresDoNotReserveOrSend(t *testing.T) {
+	for _, test := range []struct {
+		name, body, encoding string
+		readErr, closeErr    error
+		snapshot             bool
+		getBodyErr           error
+	}{
+		{name: "read_error", body: qualityWireFixtureBody[:20], readErr: io.ErrUnexpectedEOF},
+		{name: "close_error", body: qualityWireFixtureBody, closeErr: errors.New("synthetic close failure")},
+		{name: "wire_limit", body: strings.Repeat("x", (2<<20)+2)},
+		{name: "invalid_json", body: qualityWireFixtureBody[:20]},
+		{name: "invalid_zstd", body: "invalid zstd frame", encoding: "zstd"},
+		{name: "decoded_limit", body: `{"model":"gpt-6-astra","reasoning":{"effort":"high"},"input":"` + strings.Repeat("x", 1<<20) + `"}`, encoding: "compress"},
+		{name: "snapshot_read_error", body: qualityWireFixtureBody[:20], readErr: io.ErrUnexpectedEOF, snapshot: true},
+		{name: "snapshot_close_error", body: qualityWireFixtureBody, closeErr: errors.New("synthetic close failure"), snapshot: true},
+		{name: "snapshot_wire_limit", body: strings.Repeat("x", (2<<20)+2), snapshot: true},
+		{name: "snapshot_getbody_error", body: qualityWireFixtureBody, getBodyErr: io.ErrUnexpectedEOF, snapshot: true},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			svc, account, store, run, upstream, ctx, req := qualityWireServiceFixture(t, false, !test.snapshot)
+			body := []byte(test.body)
+			encoding := test.encoding
+			if encoding == "compress" {
+				var err error
+				body, err = compressCodexRequestBodyZstd(body)
+				require.NoError(t, err)
+				encoding = "zstd"
+			}
+			source := &qualityWireSource{reader: bytes.NewReader(body), readErr: test.readErr, closeErr: test.closeErr}
+			if test.snapshot {
+				req.GetBody = func() (io.ReadCloser, error) { return source, test.getBodyErr }
+			} else {
+				req.Body, req.ContentLength = source, int64(len(body))
+			}
+			req.Header.Set("Content-Encoding", encoding)
+			_, revision, err := readCodexQualityRun(ctx, store, run.RunID)
+			require.NoError(t, err)
+			response, err := svc.doOpenAICodexUpstream(req, account, "")
+			require.ErrorIs(t, err, ErrCodexQualityUnavailable)
+			require.Nil(t, response)
+			require.Zero(t, upstream.calls)
+			require.Equal(t, !test.snapshot, req.GetBody == nil)
+			require.Equal(t, 1, source.closed)
+			require.LessOrEqual(t, source.read, (2<<20)+1)
+			saved, savedRevision, err := readCodexQualityRun(ctx, store, run.RunID)
+			require.NoError(t, err)
+			require.Equal(t, revision, savedRevision)
+			require.Zero(t, saved.UsedSends)
+			require.Empty(t, saved.Attempts)
+			require.False(t, account.Schedulable)
+		})
+	}
+}
+
+func TestCodexQualityWireFailedSnapshotCannotReplayPrefix(t *testing.T) {
+	ctx := WithHTTPUpstreamRedirectsDisabled(context.Background())
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, "https://chatgpt.com/backend-api/codex/responses", strings.NewReader(qualityWireFixtureBody))
+	require.NoError(t, err)
+	req.GetBody = nil
+	source := &qualityWireSource{reader: strings.NewReader(qualityWireFixtureBody[:20]), readErr: io.ErrUnexpectedEOF}
+	req.Body = source
+	length := req.ContentLength
+	_, _, err = codexQualityWireFields(req)
+	require.ErrorIs(t, err, ErrCodexQualityUnavailable)
+	remaining, err := io.ReadAll(req.Body)
+	require.ErrorIs(t, err, ErrCodexQualityUnavailable)
+	require.Empty(t, remaining)
+	require.NoError(t, req.Body.Close())
+	require.Equal(t, 1, source.closed, "snapshot failure closes the original body once")
+	require.Equal(t, length, req.ContentLength)
+	require.Nil(t, req.GetBody)
+	require.True(t, HTTPUpstreamRedirectsDisabled(req.Context()))
+}

plugins/account-tools/manifest.source.json

@@ -2,7 +2,7 @@
   "schema_version": 1,
   "id": "codexrip.account-tools",
   "name": "账号工具",
-  "version": "0.2.11",
+  "version": "0.2.12",
   "description": "账号分类操作策略与连接测试选项",
   "author": "HTExplicit",
   "requires": { "sub2api": ">=0.2.7-codexrip.2 <0.2.9", "tested_sub2api_versions": [], "plugin_protocol": 1, "transport_api": 1, "ui_bridge": 1, "extension_api": 1 },

plugins/admin-observability/manifest.source.json

@@ -2,7 +2,7 @@
   "schema_version": 1,
   "id": "codexrip.admin-observability",
   "name": "界面与观测",
-  "version": "0.2.11",
+  "version": "0.2.12",
   "description": "账号流量观测规则与指标展示",
   "author": "HTExplicit",
   "requires": {

plugins/cindy-provider/manifest.source.json

@@ -2,7 +2,7 @@
   "schema_version": 1,
   "id": "codexrip.cindy-provider",
   "name": "Cindy Provider",
-  "version": "0.2.11",
+  "version": "0.2.12",
   "description": "Cindy 模型目录、协议能力、价格参考及账号健康判定",
   "author": "HTExplicit",
   "requires": {

plugins/codex-runtime/manifest.source.json

@@ -2,7 +2,7 @@
   "schema_version": 1,
   "id": "codexrip.codex-runtime",
   "name": "Codex 运行扩展",
-  "version": "0.2.11",
+  "version": "0.2.12",
   "description": "Cookie 路由资格、按需有界续获与出站指纹",
   "author": "HTExplicit",
   "requires": {

plugins/image-tools/manifest.source.json

@@ -2,7 +2,7 @@
   "schema_version": 1,
   "id": "codexrip.image-tools",
   "name": "图像工具",
-  "version": "0.2.11",
+  "version": "0.2.12",
   "description": "Image Studio 请求计划、模型能力和 Responses 图像桥接策略",
   "author": "HTExplicit",
   "requires": { "sub2api": ">=0.2.7-codexrip.2 <0.2.9", "tested_sub2api_versions": [], "plugin_protocol": 1, "transport_api": 1, "ui_bridge": 1, "extension_api": 1 },