diff --git a/.downstream/codex-quality-diagnostics.md b/.downstream/codex-quality-diagnostics.md index a6b0bb7e81cd..00f5077f7b73 100644 --- a/.downstream/codex-quality-diagnostics.md +++ b/.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 --source-sha --reason `。入口要求旧阶段无业务回执、无未结束操作，完整保留旧阶段及未确认结果，为旧未确认 trial 保守占用预算，再建立新 trial/turn 命名空间。它不发模型请求、不重置服务器账本；迟到的旧回执会阻止后续生成。普通 `baseline` 和 `verification` 不能隐式改变已有阶段的源码版本。 + ## 发布次序 诊断基础设施先独立发布并采集修前基线。协议与账号身份修复随后发布，在共同宿主基线、同一账号和同一预算下完成修后序列。不能先应用修复再把结果命名为修前基线。 diff --git a/backend/internal/service/codex_quality_observation.go b/backend/internal/service/codex_quality_observation.go index 335a258da007..2208dbf43c6d 100644 --- a/backend/internal/service/codex_quality_observation.go +++ b/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 { diff --git a/backend/internal/service/codex_quality_wire_test.go b/backend/internal/service/codex_quality_wire_test.go new file mode 100644 index 000000000000..2aac82e0e199 --- /dev/null +++ b/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())) +} diff --git a/plugins/account-tools/manifest.source.json b/plugins/account-tools/manifest.source.json index 4ffabca969cd..c08ccd611633 100644 --- a/plugins/account-tools/manifest.source.json +++ b/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 }, diff --git a/plugins/admin-observability/manifest.source.json b/plugins/admin-observability/manifest.source.json index d97350fda0b1..7662dd053f51 100644 --- a/plugins/admin-observability/manifest.source.json +++ b/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": { diff --git a/plugins/cindy-provider/manifest.source.json b/plugins/cindy-provider/manifest.source.json index bc0355450730..d447ef56b938 100644 --- a/plugins/cindy-provider/manifest.source.json +++ b/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": { diff --git a/plugins/codex-runtime/manifest.source.json b/plugins/codex-runtime/manifest.source.json index a99bbda67bb2..ff26649aa98e 100644 --- a/plugins/codex-runtime/manifest.source.json +++ b/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": { diff --git a/plugins/image-tools/manifest.source.json b/plugins/image-tools/manifest.source.json index 027ec34966e8..aa6581ef56e4 100644 --- a/plugins/image-tools/manifest.source.json +++ b/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 }, diff --git a/plugins/model-policy/manifest.source.json b/plugins/model-policy/manifest.source.json index d3c1416413b2..db6a0a686502 100644 --- a/plugins/model-policy/manifest.source.json +++ b/plugins/model-policy/manifest.source.json @@ -2,7 +2,7 @@ "schema_version": 1, "id": "codexrip.model-policy", "name": "模型目录与容量策略", - "version": "0.2.11", + "version": "0.2.12", "description": "精确模型别名、供应商范围与经过核对的容量参考", "author": "HTExplicit", "requires": { diff --git a/plugins/prompt-skills/manifest.source.json b/plugins/prompt-skills/manifest.source.json index d72f477d8e41..22d326434728 100644 --- a/plugins/prompt-skills/manifest.source.json +++ b/plugins/prompt-skills/manifest.source.json @@ -2,7 +2,7 @@ "schema_version": 1, "id": "codexrip.prompt-skills", "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 },