diff --git a/.downstream/gateway-borrow-notices.md b/.downstream/gateway-borrow-notices.md index a67a53d140bb..0147f7b12bd0 100644 --- a/.downstream/gateway-borrow-notices.md +++ b/.downstream/gateway-borrow-notices.md @@ -6,6 +6,9 @@ commit `5ca3cca21eeaf4ca8a694a7f2f8f0ecd9575c549`, under LGPL-3.0. The source files are `codex_gateway_pin.go`, `astra_routing_upstream.go`, `astra_target_validation.go`, `openai_codex_state_probe.go`, `openai_codex_ws_anchor.go`, `pelicanHtml.ts` and the pelican preview components. +The non-rotating source/target protocol was subsequently realigned with v2.10.3, +commit `fd1b5ee4eeb20961fbb783fa6f136a1704271e90`; retained host adaptations +are listed in [the scoped review](upstream-review-ranxi-v2.10.3.md). The repository [LICENSE](../LICENSE) retains the LGPL-3.0 terms. HTML fence selection also references `web/src/lib/render.ts` from diff --git a/.downstream/native-domains.md b/.downstream/native-domains.md index 7653f7ea8dd3..d900aab86864 100644 --- a/.downstream/native-domains.md +++ b/.downstream/native-domains.md @@ -93,49 +93,34 @@ OAuth-like accounts can be selected; the source and target lists are disjoint. The extension never rewrites account model mappings, proxies, WS switches, groups, quota or scheduling state. Normal credential refresh remains available. -Source acquisition shares one process-local `__oailb` candidate. Its maximum -lease is 230 seconds, capped by an earlier upstream expiry; receiving the same -value during a live lease does not extend it. Source preparation is sequential -and uses the ordinary OAuth account-test payload (default instructions, fixed -short user prompt, medium effort), with its own configured normal connection -policy. It does not inherit target-probe session/window headers, routing hints, -encrypted-output options or connection-close behavior. -Source acquisition and the target's two STATE shots preserve the pinned -plaintext-JSON wire protocol. They do not enter the business zstd encoder after -clearing old body headers. This leaves the saved compression setting and normal -generation unchanged. Candidate status records `source_request_encoding`, and -verification records the actual encoding of each shot. -Target probes use the pinned byte template with the actual selected model and -effective tier. Continuation keeps the target's `__cflb` before the fixed borrowed -`__oailb`, as the pinned replacement helper does. Ordinary business bodies and -cookie ordering are unchanged. -It has a 90-second budget. Each target/model is validated with two completed -HTTP 200 streams, at most 45 seconds each: the first must return STATE, and -the second may omit STATE or return the same value. Qualification includes -the target identity, actual model, exit, client headers and TLS profile. -The probe clears the business STATE and validates its own two-shot STATE, so -echoing the next business STATE does not invalidate an otherwise identical live -proof. The sender keeps that exact business STATE and its existing continuation -isolation. Native session/window identity, model, tier, exit and the original -candidate expiry still constrain reuse; no credential gains a longer lease. -Target probes preserve the template's native session/thread/window headers and -generate only a fresh legacy `session_id` for each shot, matching ranxi's full -probe contract. The effective service tier in the gateway-owned routing hint is -also included in the probe body. Proofs include native identity and routing -headers; per-request tracing IDs do not invalidate them. Manual templates use a -stable synthetic window per account/model so cached-only generation can reuse -their proof without changing actual client windows. The pinned upstream's -HTTP borrowing is Astra-only: Sol is a downstream extension, independently -validated against Sol rather than certified by an Astra pass. -When both completed target shots reject a candidate's STATE or the upstream -replaces its route, a request may acquire and validate one replacement through -the existing source/proxy configuration. Replacement is capped at 90 seconds and -never retires a candidate with another live successful proof or validation in -flight. Another rejection cools acquisition for 15 seconds. Receiving the same -rejected cookie cannot renew or qualify it. This remains demand-driven, with no -exit rotation, renewal timer, ordinary fallback or business replay. -Source and target probes use separate HTTP/TLS pools. Business WS connections -stay in the ordinary account pool, with fixed one-hour continuation anchors. +The protocol reference is ranxi v2.10.3 (`fd1b5ee4`), non-rotating branch. +Candidates are stored per source with an active-source preference. Their maximum +lease is 230 seconds, capped by earlier upstream expiry; seeing a value again +while its lease is live does not extend it. Source acquisition uses the normal +OAuth account-test payload and OpenAI transport. Target probes use the fixed +plaintext JSON template and the host's OpenAIHarvest policy, with no copied +transport configuration and no business service tier inserted into their body. +The actual selected model is preserved: Astra follows the original; Sol is a +separate downstream model extension with independent qualification. + +Source acquisition has a 90-second bound. The two target probes have 45 seconds +each and require completed HTTP 200 streams. The first must return STATE; the +second may omit it or return the same value. Probes clone the business headers, +clear business STATE and old body headers, preserve native session/thread/window +headers, and generate a new legacy `session_id` per shot. The second shot carries +only the first probe's STATE and target `__cflb` with the same borrowed `__oailb`. +Nothing is written back to the business session. The business request keeps its +own model, tier, cookies, STATE and history; borrowing replaces only `__oailb`. + +The original cache identity includes route cookie, proxy, Authorization, +ChatGPT-Account-ID, UA, Originator, Version, business STATE and TLS profile. +The host adds actual model and fingerprint policy revision. Trace IDs, native +session headers and business tier are not additional cache rules. STATE changes +therefore require fresh validation. A failed target cools that qualification for +15 seconds; it does not discard a source candidate or force a replacement attempt. +After cooldown the next request may validate again; missing or expired candidates +are acquired on demand. There is no renewal timer, exit rotation, ordinary fallback +or business replay. Business WS keeps its existing continuation anchors. Status/config/history reads do not call models. Saving an enabled config starts one finite preparation; business misses prepare on demand. There is no renewal @@ -152,10 +137,10 @@ partial readiness after rechecking expiry. Policy, credential/model-map and exit changes invalidate displayed evidence; actual dispatch still verifies the finalized request fingerprint. -The effective service tier travels from the finalized HTTP body or WS frame in -local request context into both qualification and its cache key. Explicit -`default`, `auto` and `scale` remain distinct even though their routing hints are -model-only. No internal tier header is sent upstream. +The finalized business service tier stays in request context for diagnostics. +Probe and business inputs are reported separately. `acquisition` reports the +source trigger, stage, start/end time, failure and cooldown; each target also +reports its latest validation interval. A source candidate is not target proof. Status adds `setup`, `observed_since` and `recent_usage`. The last actual dispatch and in-process counters are retained per configured target/model/protocol/purpose; diff --git a/.downstream/upstream-review-ranxi-v2.10.3.md b/.downstream/upstream-review-ranxi-v2.10.3.md new file mode 100644 index 000000000000..084ae4ad03bc --- /dev/null +++ b/.downstream/upstream-review-ranxi-v2.10.3.md @@ -0,0 +1,34 @@ +# ranxi v2.10.3 scoped review + +Reference: `fd1b5ee4eeb20961fbb783fa6f136a1704271e90`, released tag v2.10.3. +Official Wei-Shaw v0.2.15 is already integrated; do not merge it again. + +| Upstream change | Disposition | +| --- | --- | +| Borrow fingerprint, candidate, STATE probe and automatic preparation core | No change from v2.10.2; correct downstream deviations against the non-rotation branch. | +| Official v0.2.15 protocol compatibility, provider catalog and model directory | Reuse integrated code. `model_protocol_catalog.go` and the single-manifest preservation in `openai_codex_models_pinned.go` are already present. | +| Pelican showcase notice (#356) | Apply its factual interpretation to the downstream workbench: visual works are observations, not model identity or intelligence scores. | +| Excel/BPS OAuth, dynamic tools and model discovery | Excluded. | +| Worker/TOTP, 2FA, tickets, notifications, automatic group takeover, exit rotation | Excluded. | +| Unreleased production branch | Excluded; not the update baseline. | + +## Host adaptations retained + +- Parameterize target model for Sol, with qualification isolated from Astra. +- Use the host credential/identity builders and currently bound account proxy. +- Keep request-body model context so strict borrowing also applies to large business requests. +- Shared flights, per-waiter cancellation and revision cancellation replace the original global try-lock. +- Serialize complete two-shot probes by account within the existing execution capacity. +- Bounded save preparation continues after individual target failure and checks actual expiry at completion. +- Observation context prevents health, scheduling, quota and billing writes, while allowing normal credential refresh. +- Retain strict dispatch proofs, continuation isolation, diagnostic budget and raw administrator evidence. + +## Removed deviations + +- Ignoring business STATE in the v12 cache identity. +- Extra cache keys and business-tier injection into the fixed target probe. +- Target rejection forcing a fresh source candidate and retry in the non-rotation path. +- Separate source/target HTTP manager configuration copies and custom TLS policy. + +Source alignment and offline regression do not prove successful live borrowing. +Production and live acceptance are recorded separately in the deployment evidence. diff --git a/backend/cmd/server/wire_gen.go b/backend/cmd/server/wire_gen.go index 907c1f0804d2..98ba69699e45 100644 --- a/backend/cmd/server/wire_gen.go +++ b/backend/cmd/server/wire_gen.go @@ -299,7 +299,7 @@ func initializeApplication(buildInfo handler.BuildInfo) (*Application, error) { auditLogService := service.ProvideAuditLogService(auditLogRepository, settingService) auditLogHandler := admin.NewAuditLogHandler(auditLogService, totpService) systemPromptHandler := admin.NewSystemPromptHandler(systemPromptService) - codexGatewayBorrowProbeUpstream := repository.NewCodexGatewayBorrowProbeUpstream(configConfig) + codexGatewayBorrowProbeUpstream := repository.NewCodexGatewayBorrowProbeUpstream(httpUpstream) codexGatewayBorrowService, err := service.ProvideCodexGatewayBorrowService(settingRepository, accountRepository, openAIGatewayService, codexGatewayBorrowProbeUpstream, tlsFingerprintProfileService) if err != nil { return nil, err diff --git a/backend/internal/repository/codex_gateway_borrow_upstream.go b/backend/internal/repository/codex_gateway_borrow_upstream.go index ae469db94087..b8fd2c19d13b 100644 --- a/backend/internal/repository/codex_gateway_borrow_upstream.go +++ b/backend/internal/repository/codex_gateway_borrow_upstream.go @@ -1,171 +1,10 @@ package repository -import ( - "crypto/sha256" - "crypto/tls" - "encoding/json" - "errors" - "fmt" - "net/http" - "net/url" - "slices" - "time" +import "github.com/Wei-Shaw/sub2api/internal/service" - "github.com/Wei-Shaw/sub2api/internal/config" - "github.com/Wei-Shaw/sub2api/internal/pkg/tlsfingerprint" - "github.com/Wei-Shaw/sub2api/internal/service" -) - -const ( - upstreamProtocolModeCodexBorrowTargetH1 = "codex_borrow_target_h1_no_reuse" - codexBorrowProbeHeaderTimeout = 45 * time.Second -) - -// codexGatewayBorrowProbeUpstream owns two independent client managers. In -// particular, probes cannot consume the business client's eviction budget or -// change its HTTP/2 proxy fallback state, even for the same account and proxy. -type codexGatewayBorrowProbeUpstream struct { - source *httpUpstreamService - target *httpUpstreamService -} - -func NewCodexGatewayBorrowProbeUpstream(cfg *config.Config) service.CodexGatewayBorrowProbeUpstream { - newManager := func(purpose service.HTTPUpstreamProfile) *httpUpstreamService { - return &httpUpstreamService{ - cfg: cloneCodexBorrowProbeHTTPConfig(cfg), - clients: make(map[string]*upstreamClientEntry), - codexBorrowPurpose: purpose, - } - } - return &codexGatewayBorrowProbeUpstream{ - source: newManager(service.HTTPUpstreamProfileCodexBorrowSource), - target: newManager(service.HTTPUpstreamProfileCodexBorrowTarget), - } -} - -func (s *codexGatewayBorrowProbeUpstream) Do(req *http.Request, proxyURL string, accountID int64, _ int) (*http.Response, error) { - manager, req, err := s.requestManager(req) - if err != nil { - return nil, err - } - // Probe capacity follows the copied global transport settings, independently - // of the account's gateway concurrency slots and business pool dimensions. - return manager.Do(req, proxyURL, accountID, 0) -} - -func (s *codexGatewayBorrowProbeUpstream) DoWithTLS(req *http.Request, proxyURL string, accountID int64, _ int, profile *tlsfingerprint.Profile) (*http.Response, error) { - manager, req, err := s.requestManager(req) - if err != nil { - return nil, err - } - return manager.DoWithTLS(req, proxyURL, accountID, 0, profile) -} - -func (s *codexGatewayBorrowProbeUpstream) requestManager(req *http.Request) (*httpUpstreamService, *http.Request, error) { - if req == nil || req.URL == nil { - return nil, nil, errors.New("borrow probe request URL is nil") - } - switch service.HTTPUpstreamProfileFromContext(req.Context()) { - case service.HTTPUpstreamProfileCodexBorrowSource: - // Source sends the normal OpenAI transport policy through its own pool. - ctx := service.WithHTTPUpstreamProfile(req.Context(), service.HTTPUpstreamProfileOpenAI) - return s.source, req.Clone(ctx), nil - case service.HTTPUpstreamProfileCodexBorrowTarget: - clone := req.Clone(req.Context()) - clone.Close = true - return s.target, clone, nil - default: - return nil, nil, errors.New("borrow probe request purpose is missing") - } -} - -// Copy only the configuration read by the upstream transport. The managers -// neither mutate nor retain slices belonging to the business configuration. -func cloneCodexBorrowProbeHTTPConfig(cfg *config.Config) *config.Config { - if cfg == nil { - return nil - } - allowlist := cfg.Security.URLAllowlist - allowlist.UpstreamHosts = slices.Clone(allowlist.UpstreamHosts) - allowlist.PricingHosts = slices.Clone(allowlist.PricingHosts) - allowlist.CRSHosts = slices.Clone(allowlist.CRSHosts) - return &config.Config{ - Gateway: config.GatewayConfig{ - ConnectionPoolIsolation: cfg.Gateway.ConnectionPoolIsolation, - MaxIdleConns: cfg.Gateway.MaxIdleConns, - MaxIdleConnsPerHost: cfg.Gateway.MaxIdleConnsPerHost, - MaxConnsPerHost: cfg.Gateway.MaxConnsPerHost, - IdleConnTimeoutSeconds: cfg.Gateway.IdleConnTimeoutSeconds, - ResponseHeaderTimeout: cfg.Gateway.ResponseHeaderTimeout, - OpenAIResponseHeaderTimeout: cfg.Gateway.OpenAIResponseHeaderTimeout, - OpenAIHTTP2: cfg.Gateway.OpenAIHTTP2, - MaxUpstreamClients: cfg.Gateway.MaxUpstreamClients, - ClientIdleTTLSeconds: cfg.Gateway.ClientIdleTTLSeconds, - }, - Security: config.SecurityConfig{URLAllowlist: allowlist}, - } -} - -func (s *httpUpstreamService) codexBorrowClientCacheKey(key string) string { - if s.codexBorrowPurpose == service.HTTPUpstreamProfileDefault { - return key - } - return string(s.codexBorrowPurpose) + "|" + key -} - -// Fingerprinted probe clients must be keyed by the full resolved profile, not -// merely its display name. A bound profile can change without changing its name. -func codexBorrowTLSProfileKey(profile *tlsfingerprint.Profile) string { - encoded, _ := json.Marshal(profile) // Profile contains only JSON-safe values. - return fmt.Sprintf("%x", sha256.Sum256(encoded)) -} - -func cloneCodexBorrowTLSProfile(profile *tlsfingerprint.Profile) *tlsfingerprint.Profile { - if profile == nil { - return nil - } - clone := *profile - clone.CipherSuites = slices.Clone(profile.CipherSuites) - clone.Curves = slices.Clone(profile.Curves) - clone.PointFormats = slices.Clone(profile.PointFormats) - clone.SignatureAlgorithms = slices.Clone(profile.SignatureAlgorithms) - clone.ALPNProtocols = slices.Clone(profile.ALPNProtocols) - clone.SupportedVersions = slices.Clone(profile.SupportedVersions) - clone.KeyShareGroups = slices.Clone(profile.KeyShareGroups) - clone.PSKModes = slices.Clone(profile.PSKModes) - clone.Extensions = slices.Clone(profile.Extensions) - return &clone -} - -func (s *httpUpstreamService) buildCodexBorrowTLSFingerprintTransport(settings poolSettings, proxyURL *url.URL, profile *tlsfingerprint.Profile) (*http.Transport, error) { - if s.codexBorrowPurpose != service.HTTPUpstreamProfileCodexBorrowTarget { - return buildUpstreamTransportWithTLSFingerprint(settings, proxyURL, profile) - } - // Keep the selected cipher suites, extensions and other fingerprint fields, - // while excluding H2 from the actual ClientHello used by every target dialer. - profile = cloneCodexBorrowTLSProfile(profile) - if profile != nil { - profile.ALPNProtocols = []string{"http/1.1"} - } - transport, err := buildUpstreamTransportWithTLSFingerprint(settings, proxyURL, profile) - if err != nil { - return nil, err - } - configureCodexBorrowTargetTransport(transport) - return transport, nil -} - -func configureCodexBorrowTargetTransport(transport *http.Transport) { - transport.ForceAttemptHTTP2 = false - transport.DisableKeepAlives = true - transport.DisableCompression = true - transport.TLSNextProto = make(map[string]func(string, *tls.Conn) http.RoundTripper) - transport.Protocols = &http.Protocols{} - transport.Protocols.SetHTTP1(true) - if transport.TLSClientConfig == nil { - transport.TLSClientConfig = &tls.Config{MinVersion: tls.VersionTLS12} - } else { - transport.TLSClientConfig = transport.TLSClientConfig.Clone() - } - transport.TLSClientConfig.NextProtos = []string{"http/1.1"} +// Borrow probes use the host transport and its provider profiles, exactly as +// ranxi fd1b5ee4's non-rotating path. Observation isolation belongs to the +// request context; copying transport configuration here would drift again. +func NewCodexGatewayBorrowProbeUpstream(upstream service.HTTPUpstream) service.CodexGatewayBorrowProbeUpstream { + return upstream } diff --git a/backend/internal/repository/codex_gateway_borrow_upstream_test.go b/backend/internal/repository/codex_gateway_borrow_upstream_test.go index f8aae92cd36c..a41aab63c8bd 100644 --- a/backend/internal/repository/codex_gateway_borrow_upstream_test.go +++ b/backend/internal/repository/codex_gateway_borrow_upstream_test.go @@ -1,380 +1,29 @@ package repository import ( - "context" - "crypto/tls" - "crypto/x509" - "errors" - "io" - "net" - "net/http" - "net/http/httptest" - "net/url" - "slices" - "strings" - "sync" - "sync/atomic" - "testing" - "time" - "github.com/Wei-Shaw/sub2api/internal/config" - "github.com/Wei-Shaw/sub2api/internal/pkg/tlsfingerprint" "github.com/Wei-Shaw/sub2api/internal/service" "github.com/stretchr/testify/require" + "testing" ) -func codexBorrowProbeTestConfig(maxClients int) *config.Config { - return &config.Config{Gateway: config.GatewayConfig{ - ConnectionPoolIsolation: config.ConnectionPoolIsolationAccountProxy, - MaxIdleConns: 4, - MaxIdleConnsPerHost: 2, - MaxConnsPerHost: 2, - MaxUpstreamClients: maxClients, - ClientIdleTTLSeconds: 900, - OpenAIHTTP2: config.GatewayOpenAIHTTP2Config{ - Enabled: true, - AllowProxyFallbackToHTTP1: true, - FallbackErrorThreshold: 1, - }, - }} -} - -func codexBorrowProbeTestRequest(t *testing.T, purpose service.HTTPUpstreamProfile, rawURL string) *http.Request { - t.Helper() - ctx := service.WithHTTPUpstreamProfile(t.Context(), purpose) - ctx = service.WithHTTPUpstreamRedirectsDisabled(ctx) - req, err := http.NewRequestWithContext(ctx, http.MethodGet, rawURL, nil) - require.NoError(t, err) - return req -} - -func closeCodexBorrowProbeTestClients(t *testing.T, managers ...*httpUpstreamService) { - t.Helper() - t.Cleanup(func() { - for _, manager := range managers { - manager.mu.RLock() - clients := make([]*http.Client, 0, len(manager.clients)) - for _, entry := range manager.clients { - clients = append(clients, entry.client) - } - manager.mu.RUnlock() - for _, client := range clients { - client.CloseIdleConnections() - } - } - }) -} - -func TestCodexGatewayBorrowProbeUpstream_IsolatesClientBudgetsAndFallbacks(t *testing.T) { - cfg := codexBorrowProbeTestConfig(1) - business, ok := NewHTTPUpstream(cfg).(*httpUpstreamService) - require.True(t, ok) - probe, ok := NewCodexGatewayBorrowProbeUpstream(cfg).(*codexGatewayBorrowProbeUpstream) +func TestCodexBorrowPinnedTransportPolicy(t *testing.T) { + cfg := &config.Config{Gateway: config.GatewayConfig{OpenAIHTTP2: config.GatewayOpenAIHTTP2Config{Enabled: true}}} + upstream := NewHTTPUpstream(cfg) + require.Same(t, upstream, NewCodexGatewayBorrowProbeUpstream(upstream)) + manager, ok := upstream.(*httpUpstreamService) require.True(t, ok) - closeCodexBorrowProbeTestClients(t, business, probe.source, probe.target) - require.NotSame(t, cfg, probe.source.cfg) - require.NotSame(t, cfg, probe.target.cfg) - require.NotSame(t, probe.source.cfg, probe.target.cfg) - - businessEntry, err := business.acquireClientWithProfile("", 7, 1, service.HTTPUpstreamProfileOpenAI) + require.Equal(t, upstreamProtocolModeOpenAIH2, manager.resolveProtocolMode(service.HTTPUpstreamProfileCodexBorrowSource, "", nil)) + mode := manager.resolveProtocolMode(service.HTTPUpstreamProfileCodexBorrowTarget, "", nil) + require.Equal(t, upstreamProtocolModeOpenAIH1NoReuse, mode) + transport, err := buildUpstreamTransport(defaultPoolSettings(cfg), nil, mode) require.NoError(t, err) - sourceEntry, err := probe.source.acquireClientWithProfile("", 7, 0, service.HTTPUpstreamProfileOpenAI) - require.NoError(t, err) - targetEntry, err := probe.target.acquireClientWithProfile("", 7, 0, service.HTTPUpstreamProfileCodexBorrowTarget) - require.NoError(t, err) - require.NotSame(t, businessEntry.client, sourceEntry.client) - require.NotSame(t, businessEntry.client, targetEntry.client) - require.NotSame(t, sourceEntry.client, targetEntry.client) - - _, err = probe.source.acquireClientWithProfile("", 8, 0, service.HTTPUpstreamProfileOpenAI) - require.ErrorIs(t, err, errUpstreamClientLimitReached) - // Releasing one source client permits only that manager's eviction. - atomic.AddInt64(&sourceEntry.inFlight, -1) - replacement, err := probe.source.acquireClientWithProfile("", 8, 0, service.HTTPUpstreamProfileOpenAI) - require.NoError(t, err) - require.NotSame(t, sourceEntry, replacement) - require.Len(t, business.clients, 1) - require.Len(t, probe.source.clients, 1) - require.Len(t, probe.target.clients, 1) - for _, entry := range business.clients { - require.Same(t, businessEntry, entry) - } - for _, entry := range probe.target.clients { - require.Same(t, targetEntry, entry) - } - for key := range probe.source.clients { - require.True(t, strings.HasPrefix(key, string(service.HTTPUpstreamProfileCodexBorrowSource)+"|")) - } - for key := range probe.target.clients { - require.True(t, strings.HasPrefix(key, string(service.HTTPUpstreamProfileCodexBorrowTarget)+"|")) - } - - proxyKey := "http://probe-proxy.example:8080" - parsedProxy, err := url.Parse(proxyKey) - require.NoError(t, err) - probe.source.recordOpenAIHTTP2Failure(service.HTTPUpstreamProfileOpenAI, upstreamProtocolModeOpenAIH2, proxyKey, errors.New("GOAWAY")) - require.Equal(t, upstreamProtocolModeOpenAIH1Fallback, probe.source.resolveProtocolMode(service.HTTPUpstreamProfileOpenAI, proxyKey, parsedProxy)) - require.Equal(t, upstreamProtocolModeOpenAIH2, business.resolveProtocolMode(service.HTTPUpstreamProfileOpenAI, proxyKey, parsedProxy)) - require.False(t, probe.target.isOpenAIHTTP2FallbackActive(proxyKey)) - require.Equal(t, upstreamProtocolModeCodexBorrowTargetH1, probe.target.resolveProtocolMode(service.HTTPUpstreamProfileCodexBorrowTarget, proxyKey, parsedProxy)) - - // TLS clients consume the same bounded cache within their own purpose only. - _, err = probe.source.acquireClientWithTLS("", 9, 0, &tlsfingerprint.Profile{Name: "test"}, service.HTTPUpstreamProfileOpenAI) - require.ErrorIs(t, err, errUpstreamClientLimitReached) - require.Same(t, businessEntry, business.clients[buildCacheKey(config.ConnectionPoolIsolationAccountProxy, directProxyKey, 7, upstreamProtocolModeOpenAIH2)]) -} - -func TestCodexGatewayBorrowProbeUpstream_UsesGlobalProbeCapacity(t *testing.T) { - for _, purpose := range []service.HTTPUpstreamProfile{ - service.HTTPUpstreamProfileCodexBorrowSource, - service.HTTPUpstreamProfileCodexBorrowTarget, - } { - t.Run(string(purpose), func(t *testing.T) { - cfg := codexBorrowProbeTestConfig(2) - cfg.Security.URLAllowlist.UpstreamHosts = []string{"original.example"} - probe, ok := NewCodexGatewayBorrowProbeUpstream(cfg).(*codexGatewayBorrowProbeUpstream) - require.True(t, ok) - closeCodexBorrowProbeTestClients(t, probe.source, probe.target) - cfg.Gateway.MaxConnsPerHost = 1 - cfg.Security.URLAllowlist.UpstreamHosts[0] = "changed.example" - require.Equal(t, 2, probe.source.cfg.Gateway.MaxConnsPerHost) - require.Equal(t, "original.example", probe.target.cfg.Security.URLAllowlist.UpstreamHosts[0]) - - arrived := make(chan struct{}, 2) - release := make(chan struct{}) - var releaseOnce sync.Once - unblock := func() { releaseOnce.Do(func() { close(release) }) } - srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { - arrived <- struct{}{} - <-release - w.WriteHeader(http.StatusNoContent) - })) - t.Cleanup(srv.Close) - t.Cleanup(unblock) - ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) - defer cancel() - results := make(chan error, 2) - for range 2 { - req := codexBorrowProbeTestRequest(t, purpose, srv.URL) - req = req.WithContext(service.WithHTTPUpstreamProfile(ctx, purpose)) - go func() { - resp, err := probe.Do(req, "", 7, 1) - if err == nil { - err = resp.Body.Close() - } - results <- err - }() - } - for range 2 { - select { - case <-arrived: - case <-ctx.Done(): - unblock() - t.Fatal("probe requests inherited the account concurrency of one") - } - } - unblock() - for range 2 { - require.NoError(t, <-results) - } - }) - } -} - -func TestCodexGatewayBorrowProbeUpstream_TargetUsesFreshHTTP1Connections(t *testing.T) { - type observedRequest struct { - protocol int - remote string - close bool - cookie string - identity string - } - observed := make(chan observedRequest, 3) - srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - observed <- observedRequest{r.ProtoMajor, r.RemoteAddr, r.Close, r.Header.Get("Cookie"), r.Header.Get("User-Agent")} - http.SetCookie(w, &http.Cookie{Name: "__cflb", Value: "response-cookie"}) - _, _ = io.WriteString(w, "probe") - })) - srv.EnableHTTP2 = true - srv.StartTLS() - t.Cleanup(srv.Close) - probe, ok := NewCodexGatewayBorrowProbeUpstream(codexBorrowProbeTestConfig(4)).(*codexGatewayBorrowProbeUpstream) - require.True(t, ok) - closeCodexBorrowProbeTestClients(t, probe.source, probe.target) - roots := x509.NewCertPool() - roots.AddCert(srv.Certificate()) - for _, pair := range []struct { - manager *httpUpstreamService - profile service.HTTPUpstreamProfile - }{ - {probe.source, service.HTTPUpstreamProfileOpenAI}, - {probe.target, service.HTTPUpstreamProfileCodexBorrowTarget}, - } { - entry, err := pair.manager.getClientEntry("", 7, 0, pair.profile, false, true) - require.NoError(t, err) - require.Nil(t, entry.client.Jar) - transport, ok := entry.client.Transport.(*http.Transport) - require.True(t, ok) - if transport.TLSClientConfig == nil { - transport.TLSClientConfig = &tls.Config{MinVersion: tls.VersionTLS12} - } - transport.TLSClientConfig.RootCAs = roots - } - - for _, purpose := range []service.HTTPUpstreamProfile{ - service.HTTPUpstreamProfileCodexBorrowSource, - service.HTTPUpstreamProfileCodexBorrowTarget, - service.HTTPUpstreamProfileCodexBorrowTarget, - } { - req := codexBorrowProbeTestRequest(t, purpose, srv.URL) - req.Header.Set("Cookie", "__cflb=explicit-target-cookie") - req.Header.Set("User-Agent", "codex-probe-identity") - resp, err := probe.Do(req, "", 7, 1) - require.NoError(t, err) - _, err = io.Copy(io.Discard, resp.Body) - require.NoError(t, err) - require.NoError(t, resp.Body.Close()) - require.False(t, req.Close, "wrapper must not mutate the original request") - } - source, firstTarget, secondTarget := <-observed, <-observed, <-observed - require.Equal(t, 2, source.protocol, "source retains configured OpenAI protocol policy") - for _, target := range []observedRequest{firstTarget, secondTarget} { - require.Equal(t, 1, target.protocol) - require.True(t, target.close) - require.Equal(t, "__cflb=explicit-target-cookie", target.cookie) - require.Equal(t, "codex-probe-identity", target.identity) - } - require.NotEqual(t, firstTarget.remote, secondTarget.remote, "target attempts require new TCP connections") -} - -func TestCodexGatewayBorrowProbeUpstream_TLSProfileCacheAndTargetALPN(t *testing.T) { - probe, ok := NewCodexGatewayBorrowProbeUpstream(codexBorrowProbeTestConfig(8)).(*codexGatewayBorrowProbeUpstream) - require.True(t, ok) - closeCodexBorrowProbeTestClients(t, probe.source, probe.target) - profile := &tlsfingerprint.Profile{ - Name: "bound-profile", - CipherSuites: []uint16{tls.TLS_AES_128_GCM_SHA256, tls.TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256}, - ALPNProtocols: []string{"h2", "http/1.1"}, - } - first, err := probe.source.getClientEntryWithTLS("", 7, 0, profile, service.HTTPUpstreamProfileOpenAI, false, true) - require.NoError(t, err) - second, err := probe.source.getClientEntryWithTLS("", 7, 0, cloneCodexBorrowTLSProfile(profile), service.HTTPUpstreamProfileOpenAI, false, true) - require.NoError(t, err) - require.Same(t, first, second) - changed := cloneCodexBorrowTLSProfile(profile) - changed.CipherSuites[0] = tls.TLS_AES_256_GCM_SHA384 - third, err := probe.source.getClientEntryWithTLS("", 7, 0, changed, service.HTTPUpstreamProfileOpenAI, false, true) - require.NoError(t, err) - require.NotSame(t, first, third, "same-name profiles with different TLS values require separate clients") - - targetEntry, err := probe.target.getClientEntryWithTLS("", 7, 0, profile, service.HTTPUpstreamProfileCodexBorrowTarget, false, true) - require.NoError(t, err) - require.NotSame(t, first.client, targetEntry.client) - targetTransport, ok := targetEntry.client.Transport.(*http.Transport) - require.True(t, ok) - require.True(t, targetTransport.DisableKeepAlives) - require.False(t, targetTransport.ForceAttemptHTTP2) - require.False(t, targetTransport.Protocols.HTTP2()) - require.True(t, targetTransport.Protocols.HTTP1()) - require.NotNil(t, targetTransport.DialTLSContext) - require.Equal(t, codexBorrowProbeHeaderTimeout, targetTransport.ResponseHeaderTimeout) - require.Contains(t, targetEntry.poolKey, upstreamProtocolModeCodexBorrowTargetH1) - require.Nil(t, targetEntry.client.Jar) - for key := range probe.target.clients { - require.Contains(t, key, "|profile:"+codexBorrowTLSProfileKey(profile)) - require.Contains(t, key, upstreamProtocolModeCodexBorrowTargetH1) - } - - // Read the actual ClientHello, then reject it before certificate verification. - // This checks the configured dialers without an external TLS/model request. - type hello struct { - protocols []string - ciphers []uint16 - } - hellos := make(chan hello, 2) - listener, err := net.Listen("tcp", "127.0.0.1:0") - require.NoError(t, err) - t.Cleanup(func() { _ = listener.Close() }) - go func() { - for { - conn, err := listener.Accept() - if err != nil { - return - } - _ = conn.SetDeadline(time.Now().Add(5 * time.Second)) - tlsConn := tls.Server(conn, &tls.Config{GetConfigForClient: func(info *tls.ClientHelloInfo) (*tls.Config, error) { - hellos <- hello{slices.Clone(info.SupportedProtos), slices.Clone(info.CipherSuites)} - return nil, errors.New("local ClientHello capture complete") - }}) - _ = tlsConn.Handshake() - _ = conn.Close() - } - }() - for _, purpose := range []service.HTTPUpstreamProfile{service.HTTPUpstreamProfileCodexBorrowSource, service.HTTPUpstreamProfileCodexBorrowTarget} { - req := codexBorrowProbeTestRequest(t, purpose, "https://"+listener.Addr().String()) - resp, err := probe.DoWithTLS(req, "", 7, 1, profile) - require.Error(t, err, "local capture rejects the handshake after observing ClientHello") - require.Nil(t, resp) - } - readHello := func() hello { - select { - case captured := <-hellos: - return captured - case <-time.After(5 * time.Second): - t.Fatal("fingerprint request did not send a ClientHello") - return hello{} - } - } - sourceHello, targetHello := readHello(), readHello() - require.Equal(t, profile.ALPNProtocols, sourceHello.protocols) - require.Equal(t, []string{"http/1.1"}, targetHello.protocols) - require.Equal(t, profile.CipherSuites, sourceHello.ciphers) - require.Equal(t, profile.CipherSuites, targetHello.ciphers) - require.Equal(t, []string{"h2", "http/1.1"}, profile.ALPNProtocols, "target must not change the caller's profile") -} - -func TestCodexGatewayBorrowProbeUpstream_PreservesProxyAndURLPolicy(t *testing.T) { - for _, purpose := range []service.HTTPUpstreamProfile{service.HTTPUpstreamProfileCodexBorrowSource, service.HTTPUpstreamProfileCodexBorrowTarget} { - t.Run(string(purpose), func(t *testing.T) { - var calls atomic.Int64 - proxy := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { - calls.Add(1) - w.WriteHeader(http.StatusNoContent) - })) - t.Cleanup(proxy.Close) - probe, ok := NewCodexGatewayBorrowProbeUpstream(nil).(*codexGatewayBorrowProbeUpstream) - require.True(t, ok) - closeCodexBorrowProbeTestClients(t, probe.source, probe.target) - req := codexBorrowProbeTestRequest(t, purpose, "http://upstream.invalid/borrow") - resp, err := probe.DoWithTLS(req, proxy.URL, 7, 1, &tlsfingerprint.Profile{Name: "unused-over-http"}) - require.NoError(t, err) - require.Equal(t, http.StatusNoContent, resp.StatusCode) - require.NoError(t, resp.Body.Close()) - require.Equal(t, int64(1), calls.Load()) - - cfg := codexBorrowProbeTestConfig(2) - cfg.Security.URLAllowlist.Enabled = true - protected := NewCodexGatewayBorrowProbeUpstream(cfg) - req = codexBorrowProbeTestRequest(t, purpose, "https://127.0.0.1/borrow") - _, err = protected.Do(req, proxy.URL, 7, 1) - require.Error(t, err) - require.Equal(t, int64(1), calls.Load(), "blocked private URL must not reach the proxy") - }) - } - - probe, ok := NewCodexGatewayBorrowProbeUpstream(nil).(*codexGatewayBorrowProbeUpstream) - require.True(t, ok) - proxyURL, err := url.Parse("https://proxy.example:8443") - require.NoError(t, err) - transport, err := probe.target.buildCodexBorrowTLSFingerprintTransport(poolSettings{}, proxyURL, &tlsfingerprint.Profile{Name: "bound"}) - require.NoError(t, err) - // Preserve the existing HTTPS-proxy fingerprint fallback, still with H1 and - // no connection reuse for the target, rather than bypassing the proxy. - require.NotNil(t, transport.Proxy) - require.Nil(t, transport.DialTLSContext) + defer transport.CloseIdleConnections() + require.False(t, transport.ForceAttemptHTTP2) require.True(t, transport.DisableKeepAlives) - require.False(t, transport.Protocols.HTTP2()) - resolved, err := transport.Proxy(&http.Request{URL: &url.URL{Scheme: "https", Host: "upstream.example"}}) - require.NoError(t, err) - require.Equal(t, proxyURL.String(), resolved.String()) + require.Zero(t, transport.MaxIdleConns) + require.Zero(t, transport.MaxIdleConnsPerHost) + require.Empty(t, transport.TLSNextProto) + cfg.Gateway.OpenAIHTTP2.Enabled = false + require.Equal(t, upstreamProtocolModeOpenAIH1, manager.resolveProtocolMode(service.HTTPUpstreamProfileCodexBorrowSource, "", nil)) } diff --git a/backend/internal/repository/http_upstream.go b/backend/internal/repository/http_upstream.go index 57cb55441ce6..60331b36a53a 100644 --- a/backend/internal/repository/http_upstream.go +++ b/backend/internal/repository/http_upstream.go @@ -39,6 +39,7 @@ import ( // 默认配置常量 // 这些值在配置文件未指定时作为回退默认值使用 const ( + upstreamProtocolModeOpenAIH1NoReuse = "openai_h1_no_reuse" // directProxyKey: 无代理时的缓存键标识 directProxyKey = "direct" // defaultMaxIdleConns: 默认最大空闲连接总数 @@ -167,9 +168,6 @@ type httpUpstreamService struct { clients map[string]*upstreamClientEntry // 客户端缓存池，key 由隔离策略决定 // OpenAI 走 HTTP/HTTPS 代理时的 H2->H1 回退状态（key=标准化 proxyKey） openAIHTTP2Fallbacks sync.Map - // Set only on the dedicated borrow probe managers; ordinary gateway pools - // keep their existing keys, TLS policy and eviction budget. - codexBorrowPurpose service.HTTPUpstreamProfile } // NewHTTPUpstream 创建通用 HTTP 上游服务 @@ -580,18 +578,9 @@ func (s *httpUpstreamService) getClientEntryWithTLS(proxyURL string, accountID i settings := s.resolvePoolSettings(isolation, accountConcurrency) settings = s.applyProfilePoolSettings(settings, upstreamProfile) protocolMode := upstreamProtocolModeDefault - if s.codexBorrowPurpose == service.HTTPUpstreamProfileCodexBorrowTarget { - protocolMode = upstreamProtocolModeCodexBorrowTargetH1 - } // TLS 指纹客户端使用独立的缓存键，加 "tls:" 前缀 cacheKey := "tls:" + buildCacheKey(isolation, proxyKey, accountID, protocolMode) poolKey := buildPoolKey(settings, protocolMode) + ":tls" - if s.codexBorrowPurpose != service.HTTPUpstreamProfileDefault { - profile = cloneCodexBorrowTLSProfile(profile) - profileKey := "|profile:" + codexBorrowTLSProfileKey(profile) - cacheKey = s.codexBorrowClientCacheKey(cacheKey + profileKey) - poolKey += profileKey - } now := time.Now() nowUnix := now.UnixNano() @@ -642,7 +631,7 @@ func (s *httpUpstreamService) getClientEntryWithTLS(proxyURL string, accountID i // 创建带 TLS 指纹的 Transport slog.Debug("tls_fingerprint_creating_new_client", "account_id", accountID, "cache_key", cacheKey, "proxy", proxyKey) - transport, err := s.buildCodexBorrowTLSFingerprintTransport(settings, parsedProxy, profile) + transport, err := buildUpstreamTransportWithTLSFingerprint(settings, parsedProxy, profile) if err != nil { s.mu.Unlock() return nil, fmt.Errorf("build TLS fingerprint transport: %w", err) @@ -658,9 +647,6 @@ func (s *httpUpstreamService) getClientEntryWithTLS(proxyURL string, accountID i proxyKey: proxyKey, poolKey: poolKey, } - if s.codexBorrowPurpose != service.HTTPUpstreamProfileDefault { - entry.protocolMode = protocolMode - } atomic.StoreInt64(&entry.lastUsed, nowUnix) if markInFlight { atomic.StoreInt64(&entry.inFlight, 1) @@ -756,7 +742,7 @@ func (s *httpUpstreamService) getClientEntry(proxyURL string, accountID int64, a settings := s.resolvePoolSettings(isolation, accountConcurrency) settings = s.applyProfilePoolSettings(settings, profile) // 构建缓存键（根据隔离策略不同） - cacheKey := s.codexBorrowClientCacheKey(buildCacheKey(isolation, proxyKey, accountID, protocolMode)) + cacheKey := buildCacheKey(isolation, proxyKey, accountID, protocolMode) // 构建连接池配置键（用于检测配置变更） poolKey := buildPoolKey(settings, protocolMode) @@ -1000,8 +986,6 @@ func (s *httpUpstreamService) resolvePoolSettings(isolation string, accountConcu func (s *httpUpstreamService) applyProfilePoolSettings(settings poolSettings, profile service.HTTPUpstreamProfile) poolSettings { switch profile { - case service.HTTPUpstreamProfileCodexBorrowTarget: - settings.responseHeaderTimeout = codexBorrowProbeHeaderTimeout case service.HTTPUpstreamProfileOpenAI: settings.responseHeaderTimeout = 0 if s != nil && s.cfg != nil && s.cfg.Gateway.OpenAIResponseHeaderTimeout > 0 { @@ -1093,8 +1077,8 @@ func (s *httpUpstreamService) resolveOpenAIHTTP2Settings() openAIHTTP2Settings { } func (s *httpUpstreamService) resolveProtocolMode(profile service.HTTPUpstreamProfile, proxyKey string, parsedProxy *url.URL) string { - if profile == service.HTTPUpstreamProfileCodexBorrowTarget { - return upstreamProtocolModeCodexBorrowTargetH1 + if profile == service.HTTPUpstreamProfileOpenAIHarvest { + return upstreamProtocolModeOpenAIH1NoReuse } if profile == service.HTTPUpstreamProfileLongStream { return upstreamProtocolModeLongStreamH2 @@ -1430,8 +1414,13 @@ func buildUpstreamTransport(settings poolSettings, proxyURL *url.URL, protocolMo transport.DisableCompression = true } switch protocolMode { - case upstreamProtocolModeCodexBorrowTargetH1: - configureCodexBorrowTargetTransport(transport) + case upstreamProtocolModeOpenAIH1NoReuse: + // Fixed ranxi v2.10.3 harvest policy; each probe opens a fresh connection. + transport.ForceAttemptHTTP2 = false + transport.DisableKeepAlives = true + transport.MaxIdleConns = 0 + transport.MaxIdleConnsPerHost = 0 + transport.TLSNextProto = make(map[string]func(string, *tls.Conn) http.RoundTripper) case upstreamProtocolModeLongStreamH2, upstreamProtocolModeOpenAIH2: transport.ForceAttemptHTTP2 = true // 显式配置 http2 并启用 PING 健康探测，剔除代理/NAT 静默掐断的死连接， diff --git a/backend/internal/service/codex_gateway_borrow.go b/backend/internal/service/codex_gateway_borrow.go index 5d59ec86068c..5363b56f272a 100644 --- a/backend/internal/service/codex_gateway_borrow.go +++ b/backend/internal/service/codex_gateway_borrow.go @@ -109,8 +109,8 @@ func codexBorrowFailureStage(err error) (string, string) { } // CodexGatewayBorrowProbeUpstream is deliberately a distinct injection type. -// Its implementation must own separate clients, pools and TLS caches, including -// separation between source acquisition and target observation. +// It delegates to the host transport; profiles select the pinned upstream +// policy without maintaining a separate transport configuration. type CodexGatewayBorrowProbeUpstream interface{ HTTPUpstream } type CodexGatewayBorrowConfig struct { @@ -223,6 +223,7 @@ type CodexGatewayBorrowVerification struct { } type CodexGatewayBorrowTargetStatus struct { + Acquisition CodexGatewayBorrowAcquisition `json:"acquisition"` CodexGatewayBorrowVerification State string `json:"state"` CacheValid bool `json:"cache_valid"` @@ -231,6 +232,7 @@ type CodexGatewayBorrowTargetStatus struct { } type CodexGatewayBorrowStatus struct { + Acquisition CodexGatewayBorrowAcquisition `json:"acquisition"` ObservedSince time.Time `json:"observed_since"` RecentUsage []CodexGatewayBorrowUsage `json:"recent_usage"` Setup CodexGatewayBorrowSetup `json:"setup"` @@ -268,6 +270,7 @@ type codexGatewayBorrowTargetKey struct { model string } type codexGatewayBorrowTargetCheck struct { + acquisition CodexGatewayBorrowAcquisition validationID uint64 policyRevision uint64 identityRevision string @@ -280,10 +283,6 @@ type codexGatewayBorrowTargetCheck struct { } type CodexGatewayBorrowService struct { - rejectedCookie string - rejectedCause error - rejectedRoutes map[string]time.Time - validatingRoutes map[string]int validationSequence uint64 sourceSequence uint64 qualifications map[string]codexGatewayBorrowTargetCheck @@ -301,9 +300,11 @@ type CodexGatewayBorrowService struct { revision uint64 revisionCtx context.Context cancel context.CancelFunc + candidates map[int64]*codexGatewayBorrowCandidate candidate *codexGatewayBorrowCandidate sources map[int64]CodexGatewayBorrowSourceStatus targets map[codexGatewayBorrowTargetKey]codexGatewayBorrowTargetCheck + acquisition CodexGatewayBorrowAcquisition preparing bool prepareRuns int prepareFailedUntil time.Time @@ -420,9 +421,8 @@ func (s *CodexGatewayBorrowService) publishConfig(cfg CodexGatewayBorrowConfig, s.revision++ s.config, s.loaded = cloneCodexGatewayBorrowConfig(cfg), true s.candidate = nil - s.rejectedCookie, s.rejectedCause = "", nil - s.rejectedRoutes = nil - s.validatingRoutes = nil + s.candidates = nil + s.acquisition = CodexGatewayBorrowAcquisition{} s.setup = CodexGatewayBorrowSetup{} if saved && cfg.Enabled { s.setup = CodexGatewayBorrowSetup{State: "queued", Total: len(cfg.TargetAccountIDs) * len(cfg.Models), StartedAt: time.Now()} @@ -504,8 +504,9 @@ func (s *CodexGatewayBorrowService) Status() CodexGatewayBorrowStatus { s.mu.Lock() defer s.mu.Unlock() now := time.Now() + s.currentCandidateLocked() out := CodexGatewayBorrowStatus{Enabled: s.config.Enabled, Revision: s.revision, GeneratedAt: now, Preparing: s.preparing || s.prepareRuns > 0 || s.setup.State == "queued", - ObservedSince: s.observedSince, RecentUsage: []CodexGatewayBorrowUsage{}, Setup: s.setup, Config: cloneCodexGatewayBorrowConfig(s.config), Sources: []CodexGatewayBorrowSourceStatus{}, Targets: []CodexGatewayBorrowTargetStatus{}, ModelEfforts: codexGatewayBorrowModelEfforts()} + Acquisition: s.acquisition, ObservedSince: s.observedSince, RecentUsage: []CodexGatewayBorrowUsage{}, Setup: s.setup, Config: cloneCodexGatewayBorrowConfig(s.config), Sources: []CodexGatewayBorrowSourceStatus{}, Targets: []CodexGatewayBorrowTargetStatus{}, ModelEfforts: codexGatewayBorrowModelEfforts()} if c := s.candidate; c != nil { out.Candidate = &CodexGatewayBorrowCandidateStatus{SourceAccountID: c.sourceID, SourceRequestEncoding: c.sourceRequestEncoding, ExpiresAt: c.expires, RemainingSeconds: borrowRemaining(c.expires, now), CookieFingerprint: borrowHash(c.cookie.Value)} } @@ -527,9 +528,7 @@ func (s *CodexGatewayBorrowService) Status() CodexGatewayBorrowStatus { row.State, row.Reason = "expired", "route_expired" } } - if s.config.Enabled && s.candidate != nil && s.candidate.sourceID == id && s.candidateRejectedLocked() { - row.State, row.Reason = "candidate", "source_passed_target_failed" - } + out.Sources = append(out.Sources, row) } for _, id := range s.config.TargetAccountIDs { @@ -537,7 +536,8 @@ func (s *CodexGatewayBorrowService) Status() CodexGatewayBorrowStatus { row := CodexGatewayBorrowTargetStatus{CodexGatewayBorrowVerification: CodexGatewayBorrowVerification{AccountID: id, Model: model, Reason: "not_verified"}, State: "waiting"} if c, ok := s.targets[codexGatewayBorrowTargetKey{id, model}]; ok { row.CodexGatewayBorrowVerification = c.result - row.CacheValid = c.policyRevision == currentCodexFingerprintPolicyForAccount(&Account{ID: id}).revision && c.result.Success && s.candidate != nil && !s.candidateRejectedLocked() && c.cookieKey == borrowHash(s.candidate.cookie.Value) && now.Before(c.expires) && now.Before(s.candidate.expires) + row.Acquisition = c.acquisition + row.CacheValid = c.policyRevision == currentCodexFingerprintPolicyForAccount(&Account{ID: id}).revision && c.result.Success && s.candidate != nil && c.cookieKey == borrowHash(s.candidate.cookie.Value) && now.Before(c.expires) && now.Before(s.candidate.expires) expires := c.expires if s.candidate != nil && c.cookieKey == borrowHash(s.candidate.cookie.Value) && s.candidate.expires.Before(expires) { expires = s.candidate.expires @@ -568,10 +568,7 @@ func (s *CodexGatewayBorrowService) Status() CodexGatewayBorrowStatus { if !s.config.Enabled { row.State, row.Reason, row.CacheValid = "disabled", "disabled", false } - if s.config.Enabled && !row.CacheValid && s.candidateRejectedLocked() && now.Before(s.prepareFailedUntil) { - retry := s.prepareFailedUntil - row.RetryAfter = &retry - } + out.Targets = append(out.Targets, row) } } @@ -691,13 +688,17 @@ func (s *CodexGatewayBorrowService) prepareSource(ctx context.Context, rev uint6 if s == nil { return ErrCodexGatewayBorrowUnavailable } + trigger, _ := ctx.Value(borrowAcquisitionTriggerKey{}).(string) + if trigger == "" { + trigger = "save_or_manual" + } _, err := s.sourceFlight.do(ctx, revisionCtx, fmt.Sprint(rev), codexGatewayBorrowTimeout, func(ctx context.Context) (any, error) { s.mu.Lock() if s.revision != rev || !s.config.Enabled { s.mu.Unlock() return nil, ErrCodexGatewayBorrowChanged } - if s.candidate != nil && !s.candidateRejectedLocked() && time.Until(s.candidate.expires) > codexGatewayBorrowTimeout { + if c := s.currentCandidateLocked(); c != nil && time.Until(c.expires) > codexGatewayBorrowTimeout { s.mu.Unlock() return nil, nil } @@ -706,21 +707,16 @@ func (s *CodexGatewayBorrowService) prepareSource(ctx context.Context, rev uint6 return nil, ErrCodexGatewayBorrowUnavailable } if time.Now().Before(s.prepareFailedUntil) { - detail := s.prepareFailure - cause := s.rejectedCause + detail, retry := s.prepareFailure, s.prepareFailedUntil s.mu.Unlock() - if cause != nil { - return nil, cause - } - if detail != "" { - return nil, errors.New(detail) - } - return nil, ErrCodexGatewayBorrowUnavailable + return nil, &CodexGatewayBorrowFailure{Cause: ErrCodexGatewayBorrowUnavailable, Stage: "source_preparation", Reason: "source_prepare_failed", Detail: detail, RetryAfter: &retry} } ids := append([]int64{}, s.config.SourceAccountIDs...) if s.sources == nil { s.sources = make(map[int64]CodexGatewayBorrowSourceStatus) } + now := time.Now() + s.acquisition = CodexGatewayBorrowAcquisition{Trigger: trigger, Phase: "acquiring", StartedAt: &now} s.preparing = true s.sourceSequence++ sequence := s.sourceSequence @@ -731,16 +727,21 @@ func (s *CodexGatewayBorrowService) prepareSource(ctx context.Context, rev uint6 s.mu.Lock() if s.revision == rev && s.sourceSequence == sequence { s.preparing = false + finished := time.Now() + s.acquisition.FinishedAt = &finished + if s.acquisition.Phase == "acquiring" { + s.acquisition.Phase, s.acquisition.Reason = "cancelled", "cancelled" + } } s.mu.Unlock() }() var lastErr error = ErrCodexGatewayBorrowUnavailable for _, id := range ids { - if errors.Is(probeCtx.Err(), context.Canceled) { + if probeCtx.Err() != nil { lastErr = probeCtx.Err() break } - candidate, detail, err := s.acquireSource(probeCtx, id) + candidate, detail, acquireErr := s.acquireSource(probeCtx, id) now := time.Now() s.mu.Lock() if s.revision != rev || revisionCtx.Err() != nil || s.sourceSequence != sequence || errors.Is(probeCtx.Err(), context.Canceled) { @@ -748,54 +749,33 @@ func (s *CodexGatewayBorrowService) prepareSource(ctx context.Context, rev uint6 return nil, ErrCodexGatewayBorrowChanged } row := CodexGatewayBorrowSourceStatus{AccountID: id, State: "rejected", Reason: "source_probe_failed", CheckedAt: &now, Error: detail} - previouslyRejected := false - if candidate != nil { - key := borrowHash(candidate.cookie.Value) - if until, found := s.rejectedRoutes[key]; found { - previouslyRejected = now.Before(until) - if !previouslyRejected { - delete(s.rejectedRoutes, key) - } - } - } - if err == nil && candidate != nil && previouslyRejected { - // Keep the old candidate and expiry: reacquiring the identical - // rejected credential must not renew its lease or qualify it. - err = errors.New("source returned the same rejected route") - row.Reason, row.Error = "source_returned_rejected_route", err.Error() - } - if err == nil && candidate != nil && now.Before(candidate.expires) { - // Seeing the same value while its original lease is live never - // renews it, even when a different source returns that value. - if old := s.candidate; old != nil && now.Before(old.expires) && old.cookie.Value == candidate.cookie.Value && old.expires.Before(candidate.expires) { - candidate.expires = old.expires - } - if s.candidate == nil || s.candidate.cookie.Value != candidate.cookie.Value || s.candidate.cookie.Path != candidate.cookie.Path { - s.qualifications = map[string]codexGatewayBorrowTargetCheck{} - } - s.candidate = candidate - s.rejectedCookie, s.rejectedCause = "", nil + if acquireErr == nil && candidate != nil && now.Before(candidate.expires) { + s.storeCandidateLocked(candidate) row.State, row.Reason, row.Error, row.ExpiresAt = "ready", "source_qualified", "", &candidate.expires s.sources[id] = row - s.preparing = false + s.acquisition.Phase = "candidate_acquired" s.prepareFailedUntil, s.prepareFailure = time.Time{}, "" s.mu.Unlock() return nil, nil } - if err != nil { - lastErr = err + if acquireErr != nil { + lastErr = acquireErr if detail != "" { - lastErr = errors.New(detail) + lastErr = errors.Join(acquireErr, errors.New(detail)) } } s.sources[id] = row s.mu.Unlock() } s.mu.Lock() - if s.revision == rev && s.sourceSequence == sequence && !errors.Is(probeCtx.Err(), context.Canceled) && !errors.Is(lastErr, ErrCodexBorrowDiagnosticBudget) { - s.preparing = false - s.prepareFailedUntil = time.Now().Add(codexGatewayBorrowFailureWait) - s.prepareFailure = lastErr.Error() + if s.revision == rev && s.sourceSequence == sequence { + s.acquisition.Phase, s.acquisition.Reason, s.acquisition.Error = "failed", "source_prepare_failed", lastErr.Error() + if !errors.Is(probeCtx.Err(), context.Canceled) && !errors.Is(lastErr, context.Canceled) && !errors.Is(lastErr, ErrCodexBorrowDiagnosticBudget) { + s.prepareFailedUntil = time.Now().Add(codexGatewayBorrowFailureWait) + s.prepareFailure = lastErr.Error() + retry := s.prepareFailedUntil + s.acquisition.RetryAfter = &retry + } } s.mu.Unlock() return nil, lastErr @@ -805,18 +785,10 @@ func (s *CodexGatewayBorrowService) prepareSource(ctx context.Context, rev uint6 func borrowRequestFingerprint(req *http.Request, model, proxy string, profile *tlsfingerprint.Profile, cookie string) string { tlsJSON, _ := json.Marshal(profile) - values := []string{cookie, model, proxy, req.URL.EscapedPath(), req.Header.Get("Authorization"), req.Header.Get("ChatGPT-Account-ID"), req.Header.Get("User-Agent"), req.Header.Get("Originator"), req.Header.Get("Version"), string(tlsJSON)} - values = append(values, codexBorrowRequestServiceTier(req)) - // Probes clear business STATE and mint their own independent two-shot STATE. - // A client's next turn must not discard that live route proof merely because - // it echoes a new business STATE. Apply still preserves the client's exact - // STATE; the candidate lease and native session identity remain part of reuse. - // Include the native identity and routing policy preserved by the probe. - // Per-request tracing IDs and the probe's random legacy session_id are not - // qualification identities and must not defeat reuse or failure cooldowns. - for _, name := range []string{"session-id", "thread-id", "x-codex-window-id", "x-codex-installation-id", "x-codex-beta-features", openAICodexRoutingHintHeader} { - values = append(values, req.Header.Get(name)) - } + // Original non-rotation qualification identity, plus the actual model for + // the downstream Sol extension. Request tracing and business tiers do not + // add probe rules. Business STATE intentionally invalidates qualification. + values := []string{cookie, proxy, req.Header.Get("Authorization"), req.Header.Get("ChatGPT-Account-ID"), req.Header.Get("User-Agent"), req.Header.Get("Originator"), req.Header.Get("Version"), req.Header.Get("X-Codex-Turn-State"), string(tlsJSON), model} identity, _ := json.Marshal(values) return borrowHash(string(identity)) } @@ -829,7 +801,7 @@ func borrowTargetFingerprint(req *http.Request, account *Account, model, proxy s // __cflb, other cookies, body and proxy stay exactly as supplied by its caller. // cachedOnly is used by manual generation: it never acquires or validates. func (s *CodexGatewayBorrowService) Apply(req *http.Request, account *Account, model, proxy string, profile *tlsfingerprint.Profile, cachedOnly bool) (*http.Request, *CodexGatewayBorrowApplication, error) { - wire, application, err := s.applyWithCandidateRetry(req, account, model, proxy, profile, cachedOnly, false) + wire, application, err := s.apply(req, account, model, proxy, profile, cachedOnly, false) if err != nil && !IsCodexGatewayBorrowFailure(err) { err = &CodexGatewayBorrowFailure{Cause: err} } @@ -864,24 +836,23 @@ func (s *CodexGatewayBorrowService) apply(req *http.Request, account *Account, m return req, nil, nil } rev, revisionCtx := s.revision, s.revisionCtx - candidate := s.candidate - rejected := s.candidateRejectedLocked() + candidate := s.currentCandidateLocked() s.mu.Unlock() if !codexGatewayBorrowAccountSupported(account) { return req, nil, ErrCodexGatewayBorrowUnavailable } - if candidate == nil || rejected || !time.Now().Before(candidate.expires) { + if candidate == nil || !time.Now().Before(candidate.expires) { if cachedOnly { return req, nil, ErrCodexGatewayBorrowUnavailable } - if err := s.prepareSource(req.Context(), rev, revisionCtx); err != nil { + if err := s.prepareSource(withBorrowAcquisitionTrigger(req.Context(), candidate), rev, revisionCtx); err != nil { if IsCodexGatewayBorrowFailure(err) { return req, nil, err } return req, nil, &CodexGatewayBorrowFailure{Cause: err, Stage: "source_preparation", Reason: "source_prepare_failed"} } s.mu.Lock() - candidate = s.candidate + candidate = s.currentCandidateLocked() s.mu.Unlock() } if candidate == nil || !time.Now().Before(candidate.expires) || !borrowCookiePathMatches(req.URL.Path, candidate.cookie.Path) { @@ -919,14 +890,20 @@ func (s *CodexGatewayBorrowService) apply(req *http.Request, account *Account, m s.mu.Unlock() return nil, ErrCodexGatewayBorrowChanged } + started := time.Now() + trigger := "qualification_missing" + if force { + trigger = "manual_validation" + } + acquisition := CodexGatewayBorrowAcquisition{Trigger: trigger, Phase: "validating", StartedAt: &started} s.validationSequence++ validationID := s.validationSequence - validationDone := s.trackCandidateValidationLocked(rev, borrowHash(candidate.cookie.Value)) - s.targets[targetKey] = codexGatewayBorrowTargetCheck{validationID: validationID, policyRevision: policyRevision, key: key, cookieKey: borrowHash(candidate.cookie.Value), validating: true, + s.targets[targetKey] = codexGatewayBorrowTargetCheck{acquisition: acquisition, validationID: validationID, policyRevision: policyRevision, key: key, cookieKey: borrowHash(candidate.cookie.Value), validating: true, result: CodexGatewayBorrowVerification{AccountID: account.ID, Model: model, Reason: "validating", CheckedAt: time.Now()}} s.mu.Unlock() - defer validationDone() probeCtx, cancel := borrowRevisionContext(operation, revisionCtx, codexGatewayBorrowTimeout) + probeCtx, policyCancel := borrowPolicyContext(probeCtx, account.ID, policyRevision) + defer policyCancel() defer cancel() credential, credentialErr := resolveCredentialAccount(probeCtx, s.accounts, account) identityRevision := borrowAccountIdentityRevision(account, credential) @@ -937,16 +914,23 @@ func (s *CodexGatewayBorrowService) apply(req *http.Request, account *Account, m result = s.probeTarget(probeCtx, req, account, model, proxy, profile, candidate) } s.mu.Lock() - if errors.Is(operation.Err(), context.Canceled) || !s.candidateCurrentLocked(rev, candidate) { + if errors.Is(operation.Err(), context.Canceled) || currentCodexFingerprintPolicyForAccount(account).revision != policyRevision || !s.candidateCurrentLocked(rev, candidate) { if s.revision == rev && s.targets[targetKey].validationID == validationID { delete(s.targets, targetKey) } s.mu.Unlock() return nil, ErrCodexGatewayBorrowChanged } - check := codexGatewayBorrowTargetCheck{validationID: validationID, identityRevision: identityRevision, policyRevision: policyRevision, key: key, cookieKey: borrowHash(candidate.cookie.Value), expires: candidate.expires, result: result} + finished := time.Now() + acquisition.FinishedAt, acquisition.Reason, acquisition.Error = &finished, result.Reason, result.Error + acquisition.Phase = "ready" + if !result.Success { + acquisition.Phase = "failed" + } + check := codexGatewayBorrowTargetCheck{acquisition: acquisition, validationID: validationID, identityRevision: identityRevision, policyRevision: policyRevision, key: key, cookieKey: borrowHash(candidate.cookie.Value), expires: candidate.expires, result: result} if !result.Success { check.retryAfter = time.Now().Add(codexGatewayBorrowFailureWait) + check.acquisition.RetryAfter = &check.retryAfter } if s.targets[targetKey].validationID == validationID { s.targets[targetKey] = check @@ -987,10 +971,6 @@ func (s *CodexGatewayBorrowService) apply(req *http.Request, account *Account, m return borrowApplyCookie(req, candidate, check), borrowApplication(candidate, key, check.expires, check.result), nil } -func (s *CodexGatewayBorrowService) candidateCurrentLocked(rev uint64, candidate *codexGatewayBorrowCandidate) bool { - return s.revision == rev && s.config.Enabled && s.revisionCtx.Err() == nil && s.candidate != nil && !s.candidateRejectedLocked() && s.candidate.sourceID == candidate.sourceID && s.candidate.cookie.Value == candidate.cookie.Value && s.candidate.cookie.Path == candidate.cookie.Path && s.candidate.expires.Equal(candidate.expires) && time.Now().Before(s.candidate.expires) && time.Now().Before(candidate.expires) -} - func (s *CodexGatewayBorrowService) cachedTarget(rev uint64, key codexGatewayBorrowTargetKey, fingerprint string, candidate *codexGatewayBorrowCandidate) (codexGatewayBorrowTargetCheck, bool, error) { s.mu.Lock() defer s.mu.Unlock() @@ -1069,7 +1049,7 @@ func (s *CodexGatewayBorrowService) Verify(ctx context.Context, accountID int64, if err != nil { return result, err } - _, _, err = s.applyWithCandidateRetry(req, a, model, proxy, nil, false, true) + _, _, err = s.apply(req, a, model, proxy, nil, false, true) s.mu.Lock() if check, ok := s.targets[codexGatewayBorrowTargetKey{accountID, model}]; ok { result = check.result diff --git a/backend/internal/service/codex_gateway_borrow_candidate_retry.go b/backend/internal/service/codex_gateway_borrow_candidate_retry.go deleted file mode 100644 index 9b422db36b1f..000000000000 --- a/backend/internal/service/codex_gateway_borrow_candidate_retry.go +++ /dev/null @@ -1,117 +0,0 @@ -package service - -import ( - "context" - "errors" - "net/http" - "time" - - "github.com/Wei-Shaw/sub2api/internal/pkg/tlsfingerprint" -) - -func (s *CodexGatewayBorrowService) candidateRejectedLocked() bool { - return s.candidate != nil && s.rejectedCookie != "" && borrowHash(s.candidate.cookie.Value) == s.rejectedCookie -} - -func (s *CodexGatewayBorrowService) trackCandidateValidationLocked(revision uint64, cookie string) func() { - if s.validatingRoutes == nil { - s.validatingRoutes = make(map[string]int) - } - s.validatingRoutes[cookie]++ - return func() { - s.mu.Lock() - defer s.mu.Unlock() - if s.revision != revision { - return - } - s.validatingRoutes[cookie]-- - if s.validatingRoutes[cookie] == 0 { - delete(s.validatingRoutes, cookie) - } - } -} - -// Source completion produces a candidate, not proof that a target can use it. -// Retire a definitively rejected candidate only when it has no live successful -// proof and no other validation in flight. Other targets' usable routes survive. -func (s *CodexGatewayBorrowService) rejectCandidateForRetry(err error, cooling bool) bool { - var failure *CodexGatewayBorrowFailure - if s == nil || !errors.As(err, &failure) || failure.CookieFingerprint == "" || - (failure.Reason != "target_state_changed" && failure.Reason != "target_route_changed") { - return false - } - s.mu.Lock() - defer s.mu.Unlock() - if s.revision != failure.Revision || !s.config.Enabled || s.candidate == nil || s.revisionCtx.Err() != nil { - return false - } - if borrowHash(s.candidate.cookie.Value) != failure.CookieFingerprint { - // A concurrent waiter has already replaced this candidate. Retry the - // published replacement; never retire it using the older observation. - return !cooling - } - if failure.Verification != nil && failure.Verification.ExpiresAt != nil && !failure.Verification.ExpiresAt.Equal(s.candidate.expires) { - return !cooling // A fresh lease cannot be retired by its predecessor's result. - } - now := time.Now() - if now.Before(s.prepareFailedUntil) { - return false - } - for _, check := range s.qualifications { - if check.cookieKey == failure.CookieFingerprint && check.result.Success && now.Before(check.expires) { - return false - } - } - // The presentation index holds only the latest check per account/model; - // another request identity can still be validating the same candidate. - if s.validatingRoutes[failure.CookieFingerprint] > 0 { - return false - } - if s.rejectedRoutes == nil { - s.rejectedRoutes = make(map[string]time.Time) - } - // Remember retired credentials through their original lease, including - // A -> B -> A source responses. Never forget a live rejection to make room. - if len(s.rejectedRoutes) >= 1024 { - for key, expires := range s.rejectedRoutes { - if !now.Before(expires) { - delete(s.rejectedRoutes, key) - } - } - if _, known := s.rejectedRoutes[failure.CookieFingerprint]; !known && len(s.rejectedRoutes) >= 1024 { - return false - } - } - s.rejectedRoutes[failure.CookieFingerprint] = s.candidate.expires - s.rejectedCookie, s.rejectedCause = failure.CookieFingerprint, err - if cooling { - s.prepareFailedUntil = now.Add(codexGatewayBorrowFailureWait) - s.prepareFailure = err.Error() - } - return true -} - -// One request may try at most one replacement through the same configured -// sources/proxies. There is no background renewal or business replay. A second -// rejection cools acquisition for 15 seconds; per-shot and caller deadlines, -// shared-flight cancellation and the diagnostic's eight-request cap still apply. -func (s *CodexGatewayBorrowService) applyWithCandidateRetry(req *http.Request, account *Account, model, proxy string, profile *tlsfingerprint.Profile, cachedOnly, force bool) (*http.Request, *CodexGatewayBorrowApplication, error) { - wire, application, firstErr := s.apply(req, account, model, proxy, profile, cachedOnly, force) - if firstErr == nil || cachedOnly || req == nil || req.Context().Err() != nil || !s.rejectCandidateForRetry(firstErr, false) { - return wire, application, firstErr - } - retryCtx, cancel := context.WithTimeout(req.Context(), codexGatewayBorrowTimeout) - defer cancel() - wire, application, retryErr := s.apply(req.WithContext(retryCtx), account, model, proxy, profile, false, force) - if retryErr != nil { - s.rejectCandidateForRetry(retryErr, true) - // Preserve the original target verdict as well as the replacement's - // complete cause, including cancellation or an exhausted diagnostic cap. - return wire, nil, errors.Join(firstErr, retryErr) - } - if wire != nil && application != nil && application.Applied { - // The bounded preparation context must not cancel the business send. - wire = wire.WithContext(context.WithValue(req.Context(), codexBorrowAppliedContextKey{}, true)) - } - return wire, application, nil -} diff --git a/backend/internal/service/codex_gateway_borrow_candidate_retry_test.go b/backend/internal/service/codex_gateway_borrow_candidate_retry_test.go index 581754969c1a..997cf2bbc226 100644 --- a/backend/internal/service/codex_gateway_borrow_candidate_retry_test.go +++ b/backend/internal/service/codex_gateway_borrow_candidate_retry_test.go @@ -2,237 +2,89 @@ package service import ( "context" + "github.com/Wei-Shaw/sub2api/internal/pkg/tlsfingerprint" + "github.com/stretchr/testify/require" "net/http" - "sync/atomic" "testing" "time" - - "github.com/Wei-Shaw/sub2api/internal/pkg/tlsfingerprint" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" - "github.com/tidwall/gjson" ) -func TestCodexBorrowRejectedCandidateGetsOneReplacement(t *testing.T) { - for _, replacement := range []string{"usable", "rejected", "identical"} { - t.Run(replacement, func(t *testing.T) { - var sourceCalls, targetCalls int - a := borrowCoreAccount(2) - s := newBorrowCoreTest(t, func(req *http.Request, _ string, id int64, _ int, _ *tlsfingerprint.Profile) (*http.Response, error) { - if id == 1 { - sourceCalls++ - cookie := "replacement-cookie" - if replacement == "identical" { - cookie = "synthetic-borrowed-cookie" - } - return borrowCoreResponse("gpt-6-astra", "OK", "source-state", "__oailb="+cookie+"; Secure; Path=/; Max-Age=230"), nil - } - targetCalls++ - cookie, err := req.Cookie("__oailb") - assert.NoError(t, err) - model := gjson.GetBytes(borrowCoreBody(t, req), "model").String() - state := "first" - if req.Header.Get("X-Codex-Turn-State") != "" { - state = "changed" - if cookie != nil && cookie.Value == "replacement-cookie" && replacement == "usable" { - state = "first" - } - } - return borrowCoreResponse(model, "OK", state), nil - }, borrowCoreAccount(1), a) - originalExpiry := time.Now().Add(120 * time.Second) - borrowCoreCandidate(s, originalExpiry) - _, req, _, err := s.accountTemplate(context.Background(), 2, "gpt-6.1-sol") - require.NoError(t, err) - wire, applied, err := s.Apply(req, a, "gpt-6.1-sol", "", nil, false) - require.Equal(t, 1, sourceCalls, "at most one replacement source acquisition") - if replacement == "usable" { - require.NoError(t, err) - require.True(t, applied.Applied) - require.NoError(t, wire.Context().Err(), "retry cleanup must not cancel the business request") - require.Contains(t, wire.Header.Get("Cookie"), "__oailb=replacement-cookie") - require.Equal(t, 4, targetCalls) - return - } - require.Error(t, err) - require.Nil(t, applied) - before := targetCalls - _, _, err = s.Apply(req, a, "gpt-6.1-sol", "", nil, false) - require.Error(t, err) - require.Equal(t, 1, sourceCalls, "exhausted replacement cools acquisition") - require.Equal(t, before, targetCalls, "cooldown does not revalidate a known rejected candidate") - if replacement == "identical" { - require.Equal(t, originalExpiry, s.candidate.expires, "the same rejected credential never gains lifetime") - require.Equal(t, 2, targetCalls) - } else { - require.Equal(t, 4, targetCalls) - } - }) - } -} - -func TestCodexBorrowCandidateReplacementPreservesOtherProofsAndWaiters(t *testing.T) { - for _, other := range []string{"passed", "validating"} { - t.Run(other, func(t *testing.T) { - s := newBorrowCoreTest(t, nil) - borrowCoreCandidate(s, time.Now().Add(codexGatewayBorrowTTL)) - key := borrowHash(s.candidate.cookie.Value) - check := codexGatewayBorrowTargetCheck{cookieKey: key, expires: s.candidate.expires, validating: other == "validating", result: CodexGatewayBorrowVerification{AccountID: 3, Model: "gpt-6-astra", Success: other == "passed"}} - if other == "passed" { - s.qualifications["other"] = check - } else { - s.targets[codexGatewayBorrowTargetKey{3, "gpt-6-astra"}] = check - s.validatingRoutes = map[string]int{key: 1} - } - err := &CodexGatewayBorrowFailure{Revision: s.revision, Cause: ErrCodexGatewayBorrowUnavailable, Reason: "target_state_changed", CookieFingerprint: key} - require.False(t, s.rejectCandidateForRetry(err, false)) - require.Empty(t, s.rejectedCookie) - }) - } -} - -func TestCodexBorrowReplacementWaitsForHiddenRequestIdentity(t *testing.T) { - started, release := make(chan struct{}), make(chan struct{}) - var first atomic.Bool - a := borrowCoreAccount(2) - s := newBorrowCoreTest(t, func(req *http.Request, _ string, _ int64, _ int, _ *tlsfingerprint.Profile) (*http.Response, error) { - if first.CompareAndSwap(false, true) { - close(started) - select { - case <-release: - case <-req.Context().Done(): - return nil, req.Context().Err() - } - } - return borrowCoreResponse("gpt-6.1-sol", "OK", "stable"), nil - }, a) - borrowCoreCandidate(s, time.Now().Add(codexGatewayBorrowTTL)) - _, req, _, err := s.accountTemplate(context.Background(), 2, "gpt-6.1-sol") - require.NoError(t, err) - done := make(chan error, 1) - go func() { _, _, runErr := s.Apply(req, a, "gpt-6.1-sol", "", nil, false); done <- runErr }() - <-started - s.mu.Lock() - key := borrowHash(s.candidate.cookie.Value) - failure := &CodexGatewayBorrowFailure{Revision: s.revision, Cause: ErrCodexGatewayBorrowUnavailable, Reason: "target_state_changed", CookieFingerprint: key} - // Another identity completed later and owns the presentation row, while - // the first identity's real validation is still using this candidate. - s.targets[codexGatewayBorrowTargetKey{2, "gpt-6.1-sol"}] = codexGatewayBorrowTargetCheck{cookieKey: key, validationID: 999} - s.mu.Unlock() - retired := s.rejectCandidateForRetry(failure, false) - close(release) - require.False(t, retired) - require.NoError(t, <-done) - s.mu.Lock() - defer s.mu.Unlock() - require.Empty(t, s.validatingRoutes) -} - -func TestCodexBorrowReplacementCannotRetireNewConfiguration(t *testing.T) { - s := newBorrowCoreTest(t, nil) - borrowCoreCandidate(s, time.Now().Add(codexGatewayBorrowTTL)) - err := &CodexGatewayBorrowFailure{Revision: s.revision, Cause: ErrCodexGatewayBorrowUnavailable, Reason: "target_state_changed", CookieFingerprint: borrowHash(s.candidate.cookie.Value)} - s.publishConfig(CodexGatewayBorrowConfig{Enabled: true, SourceAccountIDs: []int64{4}, TargetAccountIDs: []int64{2}, Models: []string{"gpt-6-astra"}}, false) - borrowCoreCandidate(s, time.Now().Add(codexGatewayBorrowTTL)) - require.False(t, s.rejectCandidateForRetry(err, false), "even an identical cookie belongs to the new config revision") - require.Empty(t, s.rejectedCookie) -} - -func TestCodexBorrowReplacementBudgetDoesNotCoolOtherCallers(t *testing.T) { - var sourceCalls int - a := borrowCoreAccount(2) - s := newBorrowCoreTest(t, func(req *http.Request, _ string, id int64, _ int, _ *tlsfingerprint.Profile) (*http.Response, error) { - if id == 1 { - sourceCalls++ - return borrowCoreResponse("gpt-6-astra", "OK", "source", "__oailb=replacement-cookie; Secure; Path=/; Max-Age=230"), nil - } - state := "first" - cookie, cookieErr := req.Cookie("__oailb") - assert.NoError(t, cookieErr) - if req.Header.Get("X-Codex-Turn-State") != "" && cookie != nil && cookie.Value != "replacement-cookie" { - state = "changed" - } - return borrowCoreResponse("gpt-6.1-sol", "OK", state), nil - }, borrowCoreAccount(1), a) - borrowCoreCandidate(s, time.Now().Add(codexGatewayBorrowTTL)) - _, req, _, err := s.accountTemplate(context.Background(), 2, "gpt-6.1-sol") - require.NoError(t, err) - var used atomic.Int32 - limited := context.WithValue(req.Context(), codexBorrowDiagnosticContextKey{}, &codexBorrowDiagnostic{limit: 2, requests: &used, borrow: true}) - _, _, err = s.Apply(req.WithContext(limited), a, "gpt-6.1-sol", "", nil, false) - require.ErrorIs(t, err, ErrCodexBorrowDiagnosticBudget) - require.EqualValues(t, 2, used.Load()) - require.Zero(t, sourceCalls) - require.True(t, s.prepareFailedUntil.IsZero(), "a caller's cap is not a source failure") - _, applied, err := s.Apply(req, a, "gpt-6.1-sol", "", nil, false) - require.NoError(t, err) - require.True(t, applied.Applied) - require.Equal(t, 1, sourceCalls) -} - -func TestCodexBorrowReplacementCannotReplayEarlierRejectedCredential(t *testing.T) { +func TestCodexBorrowNonRotationRejectsWithoutReplacingSource(t *testing.T) { var sources, targets int - a := borrowCoreAccount(2) + account, other := borrowCoreAccount(2), borrowCoreAccount(3) + reject := true s := newBorrowCoreTest(t, func(req *http.Request, _ string, id int64, _ int, _ *tlsfingerprint.Profile) (*http.Response, error) { if id == 1 { sources++ - cookie := "replacement-cookie" - if sources == 2 { - cookie = "synthetic-borrowed-cookie" - } - return borrowCoreResponse("gpt-6-astra", "OK", "source", "__oailb="+cookie+"; Secure; Path=/; Max-Age=230"), nil + return borrowCoreResponse("gpt-6-astra", "OK", "source", "__oailb=fresh; Secure; Path=/; Max-Age=230"), nil } targets++ state := "first" - if req.Header.Get("X-Codex-Turn-State") != "" { + if id == 2 && reject && req.Header.Get("X-Codex-Turn-State") != "" { state = "changed" } return borrowCoreResponse("gpt-6.1-sol", "OK", state), nil - }, borrowCoreAccount(1), a) - borrowCoreCandidate(s, time.Now().Add(codexGatewayBorrowTTL)) + }, borrowCoreAccount(1), account, other) + borrowCoreCandidate(s, time.Now().Add(120*time.Second)) _, req, _, err := s.accountTemplate(context.Background(), 2, "gpt-6.1-sol") require.NoError(t, err) - _, _, err = s.Apply(req, a, "gpt-6.1-sol", "", nil, false) - require.Error(t, err) - require.Equal(t, 1, sources) - require.Equal(t, 4, targets) - expires := s.candidate.expires - s.prepareFailedUntil = time.Time{} - _, _, err = s.Apply(req, a, "gpt-6.1-sol", "", nil, false) - require.Error(t, err) - require.Equal(t, 2, sources) - require.Equal(t, 4, targets, "an earlier rejected cookie is not qualified again") - require.Equal(t, "replacement-cookie", s.candidate.cookie.Value) - require.Equal(t, expires, s.candidate.expires) -} - -func TestCodexBorrowExpiredRejectionRequiresFreshSourceAndProof(t *testing.T) { - var sources, targets int - a := borrowCoreAccount(2) - s := newBorrowCoreTest(t, func(_ *http.Request, _ string, id int64, _ int, _ *tlsfingerprint.Profile) (*http.Response, error) { - if id == 1 { - sources++ - return borrowCoreResponse("gpt-6-astra", "OK", "source", "__oailb=synthetic-borrowed-cookie; Secure; Path=/; Max-Age=230"), nil - } - targets++ - return borrowCoreResponse("gpt-6.1-sol", "OK", "fresh-stable-state"), nil - }, borrowCoreAccount(1), a) - oldExpiry := time.Now().Add(-time.Second) - borrowCoreCandidate(s, oldExpiry) - _, req, _, err := s.accountTemplate(context.Background(), 2, "gpt-6.1-sol") + original := s.candidate + for range 2 { + _, applied, err := s.Apply(req, account, "gpt-6.1-sol", "", nil, false) + require.Nil(t, applied) + var failure *CodexGatewayBorrowFailure + require.ErrorAs(t, err, &failure) + require.Equal(t, "target_state_changed", failure.Reason) + require.NotNil(t, failure.RetryAfter) + } + require.Zero(t, sources) + require.Equal(t, 2, targets, "cooldown retains verdict without another probe") + require.Same(t, original, s.candidate, "target failure never retires source") + _, otherReq, _, err := s.accountTemplate(context.Background(), 3, "gpt-6.1-sol") require.NoError(t, err) - key := borrowHash(s.candidate.cookie.Value) - s.rejectedCookie = key - s.rejectedRoutes = map[string]time.Time{key: oldExpiry} - s.targets[codexGatewayBorrowTargetKey{2, "gpt-6.1-sol"}] = codexGatewayBorrowTargetCheck{ - key: borrowTargetFingerprint(req, a, "gpt-6.1-sol", "", nil, s.candidate.cookie.Value), cookieKey: key, - expires: oldExpiry, retryAfter: time.Now().Add(time.Minute), - result: CodexGatewayBorrowVerification{Reason: "target_state_changed", ExpiresAt: &oldExpiry}, + _, app, err := s.Apply(otherReq, other, "gpt-6.1-sol", "", nil, false) + require.NoError(t, err) + require.True(t, app.Applied) + reject = false + for key, check := range s.qualifications { + check.retryAfter = time.Now().Add(-time.Second) + s.qualifications[key] = check } - _, applied, err := s.Apply(req, a, "gpt-6.1-sol", "", nil, false) + _, app, err = s.Apply(req, account, "gpt-6.1-sol", "", nil, false) + require.NoError(t, err) + require.True(t, app.Applied, "next request resumes after cooldown") + require.Equal(t, 6, targets) + require.Zero(t, sources) + original.expires = time.Now().Add(-time.Second) + _, app, err = s.Apply(req, account, "gpt-6.1-sol", "", nil, false) require.NoError(t, err) - require.True(t, applied.Applied) - require.Equal(t, 1, sources, "expiration requires a new source observation") - require.Equal(t, 2, targets, "an old failure cooldown cannot certify or reject the new lease") - require.Empty(t, s.rejectedRoutes) + require.True(t, app.Applied) + require.Equal(t, 1, sources) + require.Equal(t, 8, targets) + status := s.Status() + require.Equal(t, "expired_route", status.Acquisition.Trigger) + require.Equal(t, "candidate_acquired", status.Acquisition.Phase) + require.NotNil(t, status.Acquisition.FinishedAt) + require.False(t, status.Preparing) +} +func TestCodexBorrowSourcePoolPreservesOtherCandidates(t *testing.T) { + s := newBorrowCoreTest(t, nil) + borrowCoreCandidate(s, time.Now().Add(time.Minute)) + first := s.candidate + s.storeCandidateLocked(first) + second := *first + second.sourceID, second.cookie.Value = 4, "second" + s.config.SourceAccountIDs = []int64{1, 4} + s.storeCandidateLocked(&second) + require.True(t, s.candidateCurrentLocked(s.revision, first)) + repeated := *first + repeated.expires = time.Now().Add(230 * time.Second) + s.storeCandidateLocked(&repeated) + require.Equal(t, first.expires, repeated.expires) + repeated.expires = time.Now().Add(-time.Second) + require.Same(t, &second, s.currentCandidateLocked()) + oldRev := s.revision + s.publishConfig(CodexGatewayBorrowConfig{}, false) + require.False(t, s.candidateCurrentLocked(oldRev, &second)) + require.Empty(t, s.candidates) } diff --git a/backend/internal/service/codex_gateway_borrow_candidates.go b/backend/internal/service/codex_gateway_borrow_candidates.go new file mode 100644 index 000000000000..1c39ab665631 --- /dev/null +++ b/backend/internal/service/codex_gateway_borrow_candidates.go @@ -0,0 +1,84 @@ +package service + +import ( + "context" + "time" +) + +// Candidates belong to sources, never to a target verdict. The active source +// is preferred until expiry; a target rejection only cools that qualification. +func (s *CodexGatewayBorrowService) currentCandidateLocked() *codexGatewayBorrowCandidate { + if s.candidate != nil && time.Now().Before(s.candidate.expires) { + return s.candidate + } + for _, id := range s.config.SourceAccountIDs { + if c := s.candidates[id]; c != nil && time.Now().Before(c.expires) { + s.candidate = c + return c + } + } + return s.candidate // Retain the expired candidate for status and trigger evidence. +} + +func (s *CodexGatewayBorrowService) storeCandidateLocked(candidate *codexGatewayBorrowCandidate) { + if s.candidates == nil { + s.candidates = make(map[int64]*codexGatewayBorrowCandidate) + } + now := time.Now() + clamp := func(old *codexGatewayBorrowCandidate) { + if old != nil && now.Before(old.expires) && old.cookie.Value == candidate.cookie.Value && old.expires.Before(candidate.expires) { + candidate.expires = old.expires + } + } + clamp(s.candidate) + for _, old := range s.candidates { + clamp(old) + } + s.candidates[candidate.sourceID] = candidate + s.candidate = candidate + // Other source proofs remain usable. Cookie, model and identity are already + // in their keys, so no target can accidentally inherit another qualification. +} + +func (s *CodexGatewayBorrowService) liveCookieLocked(key string) bool { + if c := s.candidate; c != nil && borrowHash(c.cookie.Value) == key && time.Now().Before(c.expires) { + return true + } + for _, c := range s.candidates { + if borrowHash(c.cookie.Value) == key && time.Now().Before(c.expires) { + return true + } + } + return false +} + +func (s *CodexGatewayBorrowService) candidateCurrentLocked(rev uint64, candidate *codexGatewayBorrowCandidate) bool { + if candidate == nil || s.revision != rev || !s.config.Enabled || s.revisionCtx.Err() != nil || !time.Now().Before(candidate.expires) { + return false + } + current := s.candidates[candidate.sourceID] + if current == nil { + current = s.candidate + } + return current != nil && current.sourceID == candidate.sourceID && current.cookie.Value == candidate.cookie.Value && current.cookie.Path == candidate.cookie.Path && current.expires.Equal(candidate.expires) +} + +type CodexGatewayBorrowAcquisition struct { + Trigger string `json:"trigger,omitempty"` + Phase string `json:"phase,omitempty"` + StartedAt *time.Time `json:"started_at,omitempty"` + FinishedAt *time.Time `json:"finished_at,omitempty"` + Reason string `json:"reason,omitempty"` + Error string `json:"error,omitempty"` + RetryAfter *time.Time `json:"retry_after,omitempty"` +} + +type borrowAcquisitionTriggerKey struct{} + +func withBorrowAcquisitionTrigger(ctx context.Context, candidate *codexGatewayBorrowCandidate) context.Context { + trigger := "missing_route" + if candidate != nil { + trigger = "expired_route" + } + return context.WithValue(ctx, borrowAcquisitionTriggerKey{}, trigger) +} diff --git a/backend/internal/service/codex_gateway_borrow_core_test.go b/backend/internal/service/codex_gateway_borrow_core_test.go index 65eb2cf81476..488df01c05f7 100644 --- a/backend/internal/service/codex_gateway_borrow_core_test.go +++ b/backend/internal/service/codex_gateway_borrow_core_test.go @@ -262,12 +262,15 @@ func TestCodexGatewayBorrowTargetCacheModelAndFingerprintIsolation(t *testing.T) continuation := template.Clone(template.Context()) continuation.Header.Set("X-Codex-Turn-State", "next-business-state") continuation.Header.Set("X-Client-Request-ID", "next-trace") - wire, reused, err := s.Apply(continuation, a, "gpt-6.1-sol", "target-exit", nil, true) - require.NoError(t, err, "business STATE is cleared from the independent probe, so it cannot invalidate its proof") - require.Equal(t, app.ExpiresAt, reused.ExpiresAt, "reuse never extends the candidate lease") - require.Equal(t, "next-business-state", wire.Header.Get("X-Codex-Turn-State")) - require.Equal(t, "business-state", template.Header.Get("X-Codex-Turn-State"), "business histories remain isolated") - for _, header := range []string{"Authorization", "ChatGPT-Account-ID", "User-Agent", "Originator", "Version", "session-id", "thread-id", "x-codex-window-id", "x-codex-installation-id"} { + _, _, err = s.Apply(continuation, a, "gpt-6.1-sol", "target-exit", nil, true) + require.Error(t, err, "original cache identity includes business STATE") + require.Equal(t, "business-state", template.Header.Get("X-Codex-Turn-State")) + traceOnly := template.Clone(template.Context()) + traceOnly.Header.Set("X-Client-Request-ID", "another-trace") + _, reused, err := s.Apply(traceOnly, a, "gpt-6.1-sol", "target-exit", nil, true) + require.NoError(t, err) + require.Equal(t, app.ExpiresAt, reused.ExpiresAt) + for _, header := range []string{"Authorization", "ChatGPT-Account-ID", "User-Agent", "Originator", "Version"} { req := template.Clone(template.Context()) req.Header.Set(header, req.Header.Get(header)+"-changed") _, _, err := s.Apply(req, a, "gpt-6.1-sol", "target-exit", nil, true) diff --git a/backend/internal/service/codex_gateway_borrow_diagnose_contract.go b/backend/internal/service/codex_gateway_borrow_diagnose_contract.go index ea7ee4f12b42..7a0ccfffbb3a 100644 --- a/backend/internal/service/codex_gateway_borrow_diagnose_contract.go +++ b/backend/internal/service/codex_gateway_borrow_diagnose_contract.go @@ -74,7 +74,7 @@ func (s *CodexGatewayBorrowService) diagnoseProbeContract(ctx context.Context, r return errors.New("local tier policy changes the source self-check template; no model request was sent") } s.mu.Lock() - candidate := s.candidate + candidate := s.currentCandidateLocked() current := s.candidateCurrentLocked(revision, candidate) s.mu.Unlock() if sourceProbe && (!current || candidate.sourceID != account.ID) { @@ -85,7 +85,7 @@ func (s *CodexGatewayBorrowService) diagnoseProbeContract(ctx context.Context, r return err } s.mu.Lock() - candidate = s.candidate + candidate = s.currentCandidateLocked() current = s.candidateCurrentLocked(revision, candidate) s.mu.Unlock() } diff --git a/backend/internal/service/codex_gateway_borrow_diagnose_test.go b/backend/internal/service/codex_gateway_borrow_diagnose_test.go index fec3537237d8..70265d7bb2e2 100644 --- a/backend/internal/service/codex_gateway_borrow_diagnose_test.go +++ b/backend/internal/service/codex_gateway_borrow_diagnose_test.go @@ -216,7 +216,7 @@ func testCodexBorrowDiagnoseHTTPToolContinuation(t *testing.T, tier string) { }) s := NewCodexGatewayBorrowService(nil, repo, gateway, borrowCoreProbe(func(req *http.Request, _ string, _ int64, _ int, _ *tlsfingerprint.Profile) (*http.Response, error) { probes++ - assert.Equal(t, tier, gjson.GetBytes(borrowCoreBody(t, req), "service_tier").String()) + assert.False(t, gjson.GetBytes(borrowCoreBody(t, req), "service_tier").Exists()) return borrowCoreResponse("gpt-6.1-sol", "OK", "probe-state"), nil }), nil) s.publishConfig(CodexGatewayBorrowConfig{Enabled: true, SourceAccountIDs: []int64{1}, TargetAccountIDs: []int64{2}, Models: []string{"gpt-6.1-sol"}}, false) @@ -240,7 +240,7 @@ func testCodexBorrowDiagnoseHTTPToolContinuation(t *testing.T, tier string) { } require.True(t, results[1].ToolRoundTrip) require.Equal(t, 2, business) - require.Equal(t, 2, probes, "the returned business STATE must not revalidate the same live route") + require.Equal(t, 4, probes, "original qualification identity includes the returned business STATE") require.EqualValues(t, business+probes, count) require.LessOrEqual(t, count, int32(8)) usage := s.Status().RecentUsage[0] diff --git a/backend/internal/service/codex_gateway_borrow_pelican.go b/backend/internal/service/codex_gateway_borrow_pelican.go index ad198d270aa2..70d3c71652fb 100644 --- a/backend/internal/service/codex_gateway_borrow_pelican.go +++ b/backend/internal/service/codex_gateway_borrow_pelican.go @@ -57,7 +57,8 @@ func (s *CodexGatewayBorrowService) GeneratePelican(ctx context.Context, account rev, revisionCtx := s.revision, s.revisionCtx s.mu.Unlock() s.mu.Lock() - cached := s.candidate != nil && time.Now().Before(s.candidate.expires) + candidate := s.currentCandidateLocked() + cached := candidate != nil && time.Now().Before(candidate.expires) s.mu.Unlock() if !cached { result.Error = "qualified route cache is missing or expired" @@ -87,7 +88,7 @@ func (s *CodexGatewayBorrowService) GeneratePelican(ctx context.Context, account return result, nil } s.mu.Lock() - valid := s.revision == rev && revisionCtx.Err() == nil && s.candidate != nil && borrowHash(s.candidate.cookie.Value) == applied.CookieFingerprint && s.candidate.expires.After(time.Now()) + valid := s.revision == rev && revisionCtx.Err() == nil && s.liveCookieLocked(applied.CookieFingerprint) s.mu.Unlock() if !valid { result.Error = "gateway borrow configuration or route changed before dispatch" diff --git a/backend/internal/service/codex_gateway_borrow_policy.go b/backend/internal/service/codex_gateway_borrow_policy.go new file mode 100644 index 000000000000..495a8ed6c60e --- /dev/null +++ b/backend/internal/service/codex_gateway_borrow_policy.go @@ -0,0 +1,51 @@ +package service + +import ( + "context" + "sync" +) + +// Only active observations register here. Identity edits cancel their work, +// without changing or replaying a business request that has already been sent. +var borrowPolicyObservers = struct { + sync.Mutex + next uint64 + accounts map[int64]map[uint64]context.CancelFunc +}{accounts: make(map[int64]map[uint64]context.CancelFunc)} + +func borrowPolicyContext(parent context.Context, accountID int64, revision uint64) (context.Context, context.CancelFunc) { + ctx, cancel := context.WithCancel(parent) + borrowPolicyObservers.Lock() + borrowPolicyObservers.next++ + id := borrowPolicyObservers.next + if borrowPolicyObservers.accounts[accountID] == nil { + borrowPolicyObservers.accounts[accountID] = make(map[uint64]context.CancelFunc) + } + borrowPolicyObservers.accounts[accountID][id] = cancel + if currentCodexFingerprintPolicyForAccount(&Account{ID: accountID}).revision != revision { + cancel() + } + borrowPolicyObservers.Unlock() + return ctx, func() { + cancel() + borrowPolicyObservers.Lock() + delete(borrowPolicyObservers.accounts[accountID], id) + if len(borrowPolicyObservers.accounts[accountID]) == 0 { + delete(borrowPolicyObservers.accounts, accountID) + } + borrowPolicyObservers.Unlock() + } +} + +func cancelBorrowPolicyObservers(accountID int64) { + borrowPolicyObservers.Lock() + defer borrowPolicyObservers.Unlock() + for id, callbacks := range borrowPolicyObservers.accounts { + if accountID != 0 && id != accountID { + continue + } + for _, cancel := range callbacks { + cancel() + } + } +} diff --git a/backend/internal/service/codex_gateway_borrow_policy_test.go b/backend/internal/service/codex_gateway_borrow_policy_test.go new file mode 100644 index 000000000000..9605fda2d93b --- /dev/null +++ b/backend/internal/service/codex_gateway_borrow_policy_test.go @@ -0,0 +1,49 @@ +package service + +import ( + "context" + "github.com/Wei-Shaw/sub2api/internal/pkg/tlsfingerprint" + "github.com/stretchr/testify/require" + "net/http" + "testing" + "time" +) + +func TestCodexBorrowPolicyEditCancelsOldPreparation(t *testing.T) { + for _, id := range []int64{1, 2} { + t.Run(map[int64]string{1: "source", 2: "target"}[id], func(t *testing.T) { + entered := make(chan struct{}) + a := borrowCoreAccount(2) + s := newBorrowCoreTest(t, func(req *http.Request, _ string, _ int64, _ int, _ *tlsfingerprint.Profile) (*http.Response, error) { + close(entered) + <-req.Context().Done() + return nil, req.Context().Err() + }, borrowCoreAccount(1), a) + if id == 2 { + borrowCoreCandidate(s, time.Now().Add(codexGatewayBorrowTTL)) + } + _, req, _, err := s.accountTemplate(context.Background(), 2, "gpt-6-astra") + require.NoError(t, err) + done := make(chan error, 1) + go func() { _, _, err := s.Apply(req, a, "gpt-6-astra", "", nil, false); done <- err }() + select { + case <-entered: + case <-time.After(time.Second): + t.Fatal("probe did not start") + } + notifyCodexFingerprintAccountChanged(id) + select { + case err := <-done: + require.Error(t, err) + case <-time.After(time.Second): + t.Fatal("old policy continued probing") + } + status := s.Status() + require.False(t, status.Preparing) + for _, target := range status.Targets { + require.False(t, target.CacheValid) + require.NotEqual(t, "validating", target.State) + } + }) + } +} diff --git a/backend/internal/service/codex_gateway_borrow_probe.go b/backend/internal/service/codex_gateway_borrow_probe.go index 0e3cdcf3e1c0..b0fd702df58a 100644 --- a/backend/internal/service/codex_gateway_borrow_probe.go +++ b/backend/internal/service/codex_gateway_borrow_probe.go @@ -1,7 +1,7 @@ package service // Adapted from the scoped-cookie and STATE probe protocol in ranxi2001/sub2api, -// commit 5ca3cca21eeaf4ca8a694a7f2f8f0ecd9575c549 (LGPL-3.0), specifically +// commit fd1b5ee4eeb20961fbb783fa6f136a1704271e90 (LGPL-3.0), specifically // backend/internal/service/openai_codex_state_probe.go. These are observation // requests: no normal response handlers, quota writes, recovery or health hooks. @@ -102,6 +102,8 @@ func (s *CodexGatewayBorrowService) accountTemplate(ctx context.Context, id int6 } func (s *CodexGatewayBorrowService) acquireSource(ctx context.Context, id int64) (*codexGatewayBorrowCandidate, string, error) { + ctx, policyCancel := borrowPolicyContext(ctx, id, currentCodexFingerprintPolicyForAccount(&Account{ID: id}).revision) + defer policyCancel() a, template, proxy, err := s.accountTemplate(ctx, id, codexGatewayBorrowSourceModel) if err != nil { return nil, err.Error(), err @@ -211,13 +213,18 @@ func borrowCandidateFromCookies(cookies []*http.Cookie, path string, now time.Ti func (s *CodexGatewayBorrowService) probeTarget(ctx context.Context, template *http.Request, account *Account, model, proxy string, profile *tlsfingerprint.Profile, candidate *codexGatewayBorrowCandidate) CodexGatewayBorrowVerification { ctx = withCodexBorrowPlainProbe(ctx) expires := candidate.expires - result := CodexGatewayBorrowVerification{AccountID: account.ID, Model: model, CheckedAt: time.Now(), ExpiresAt: &expires, Reason: "target_probe_failed", RequestShape: "ranxi_full", ServiceTier: codexBorrowRequestServiceTier(template)} + result := CodexGatewayBorrowVerification{AccountID: account.ID, Model: model, CheckedAt: time.Now(), ExpiresAt: &expires, Reason: "target_probe_failed", RequestShape: codexBorrowProbeRanxiLiteral} if shape, _ := ctx.Value(codexBorrowProbeShapeContextKey{}).(string); shape != "" { result.RequestShape = shape } else { // The pinned verdict was measured with this fixed byte template, not a - // generic map serializer. Keep the selected model and final tier exact. - ctx = context.WithValue(ctx, codexBorrowProbeShapeContextKey{}, codexBorrowProbeRanxiTier) + // generic map serializer. Only the selected model is parameterized. + ctx = context.WithValue(ctx, codexBorrowProbeShapeContextKey{}, codexBorrowProbeRanxiLiteral) + } + // Business tier stays on the real request; the original probe has none. + shape, _ := ctx.Value(codexBorrowProbeShapeContextKey{}).(string) + if shape != codexBorrowProbeRanxiLiteral { + result.ServiceTier = codexBorrowRequestServiceTier(template) } ctx = withCodexBorrowServiceTier(ctx, result.ServiceTier) // A two-shot STATE verdict is one observation. Releasing account capacity diff --git a/backend/internal/service/codex_gateway_borrow_request_context_test.go b/backend/internal/service/codex_gateway_borrow_request_context_test.go index a9c4923fa8a2..e17f8284784f 100644 --- a/backend/internal/service/codex_gateway_borrow_request_context_test.go +++ b/backend/internal/service/codex_gateway_borrow_request_context_test.go @@ -12,7 +12,7 @@ import ( "github.com/tidwall/gjson" ) -func TestCodexBorrowQualificationSeparatesModelOnlyRoutingTiers(t *testing.T) { +func TestCodexBorrowPinnedProbeKeepsBusinessTierOutOfQualification(t *testing.T) { account := borrowCoreAccount(2) var received []string s := newBorrowCoreTest(t, func(req *http.Request, _ string, _ int64, _ int, _ *tlsfingerprint.Profile) (*http.Response, error) { @@ -26,14 +26,12 @@ func TestCodexBorrowQualificationSeparatesModelOnlyRoutingTiers(t *testing.T) { require.NoError(t, err) for _, tier := range []string{"default", "auto", "scale", ""} { wire := req.WithContext(withCodexBorrowServiceTier(req.Context(), tier)) - _, _, err := s.Apply(wire, account, "gpt-6.1-sol", "", nil, true) - require.Error(t, err, "identical routing hints must not share another tier's proof") _, applied, err := s.Apply(wire, account, "gpt-6.1-sol", "", nil, false) require.NoError(t, err) - require.Equal(t, tier, applied.Verification.ServiceTier) + require.Empty(t, applied.Verification.ServiceTier) } - require.Equal(t, []string{"default", "default", "auto", "auto", "scale", "scale", "", ""}, received) + require.Equal(t, []string{"", ""}, received) _, _, err = s.Apply(req.WithContext(withCodexBorrowServiceTier(req.Context(), "default")), account, "gpt-6.1-sol", "", nil, true) require.NoError(t, err, "the first tier's independent successful proof remains reusable") - require.Len(t, received, 8) + require.Len(t, received, 2) } diff --git a/backend/internal/service/codex_gateway_borrow_request_repair_test.go b/backend/internal/service/codex_gateway_borrow_request_repair_test.go index 101af8023f0f..38b40e3e2e58 100644 --- a/backend/internal/service/codex_gateway_borrow_request_repair_test.go +++ b/backend/internal/service/codex_gateway_borrow_request_repair_test.go @@ -25,7 +25,7 @@ func TestCodexBorrowNilRequestDoesNotCreateDiagnosticEvidence(t *testing.T) { // session/thread headers belong to the template and must survive both shots. // A synthetic upstream rotates STATE when those headers change, reproducing // the false rejection introduced by manufacturing a new native session per shot. -func TestCodexBorrowProbePreservesNativeSessionAndTier(t *testing.T) { +func TestCodexBorrowProbePreservesPinnedSessionContract(t *testing.T) { oldCompression := codexRequestZstd.Load() SetCodexRequestZstdEnabled(true) defer SetCodexRequestZstdEnabled(oldCompression) @@ -38,8 +38,8 @@ func TestCodexBorrowProbePreservesNativeSessionAndTier(t *testing.T) { assert.Equal(t, "target-exit", proxy) body := borrowCoreBody(t, req) assert.Equal(t, model, gjson.GetBytes(body, "model").String()) - assert.Equal(t, "priority", gjson.GetBytes(body, "service_tier").String()) - assert.Equal(t, string(borrowRanxiLiteralPayload(model, "priority")), string(body), "use the pinned probe byte template with the actual model and tier") + assert.False(t, gjson.GetBytes(body, "service_tier").Exists()) + assert.Equal(t, `{"model":"`+model+`","instructions":"Reply with OK.","input":[{"type":"message","role":"user","content":[{"type":"input_text","text":"Reply with OK."}]}],"stream":true,"store":false,"parallel_tool_calls":true,"include":["reasoning.encrypted_content"]}`, string(body), "use the fixed upstream byte template with only the actual model parameterized") assert.Equal(t, "model="+model+";tier=priority", req.Header.Get(openAICodexRoutingHintHeader)) assert.Empty(t, req.Header.Get("Content-Encoding")) assert.Empty(t, req.Header.Get("X-Codex-Turn-Metadata")) @@ -123,7 +123,7 @@ func TestCodexBorrowPreparationFailureRetainsReasonAndCounts(t *testing.T) { require.Equal(t, "target_state_changed", usage[0].Reason) } -func TestCodexBorrowProofIncludesNativeSessionAndRoutingPolicy(t *testing.T) { +func TestCodexBorrowProofUsesOriginalCacheFields(t *testing.T) { a := borrowCoreAccount(2) s := newBorrowCoreTest(t, func(*http.Request, string, int64, int, *tlsfingerprint.Profile) (*http.Response, error) { return borrowCoreResponse("gpt-6.1-sol", "OK", "stable"), nil @@ -137,7 +137,7 @@ func TestCodexBorrowProofIncludesNativeSessionAndRoutingPolicy(t *testing.T) { changed := req.Clone(req.Context()) changed.Header.Set(header, "changed") _, _, err = s.Apply(changed, a, "gpt-6.1-sol", "", nil, true) - require.Error(t, err, header) + require.NoError(t, err, header) } req.Header.Set("x-client-request-id", "another-trace") _, application, err := s.Apply(req, a, "gpt-6.1-sol", "", nil, true) diff --git a/backend/internal/service/http_upstream_profile.go b/backend/internal/service/http_upstream_profile.go index 00c815591002..b729ee3ffc15 100644 --- a/backend/internal/service/http_upstream_profile.go +++ b/backend/internal/service/http_upstream_profile.go @@ -11,10 +11,11 @@ const ( HTTPUpstreamProfileOpenAI HTTPUpstreamProfile = "openai" HTTPUpstreamProfileGrok HTTPUpstreamProfile = "grok" HTTPUpstreamProfileLongStream HTTPUpstreamProfile = "long_stream" - // Borrow purposes route through the dedicated probe upstream, whose clients - // and fallback state are separate from normal gateway traffic. - HTTPUpstreamProfileCodexBorrowSource HTTPUpstreamProfile = "codex_borrow_source" - HTTPUpstreamProfileCodexBorrowTarget HTTPUpstreamProfile = "codex_borrow_target" + // The pinned upstream uses ordinary OpenAI for source acquisition and + // OpenAIHarvest for the two target probes. Keep aliases for local callers. + HTTPUpstreamProfileOpenAIHarvest HTTPUpstreamProfile = "openai_harvest" + HTTPUpstreamProfileCodexBorrowSource = HTTPUpstreamProfileOpenAI + HTTPUpstreamProfileCodexBorrowTarget = HTTPUpstreamProfileOpenAIHarvest ) type httpUpstreamProfileContextKey struct{} @@ -43,7 +44,7 @@ func HTTPUpstreamProfileFromContext(ctx context.Context) HTTPUpstreamProfile { } switch profile { case HTTPUpstreamProfileOpenAI, HTTPUpstreamProfileGrok, HTTPUpstreamProfileLongStream, - HTTPUpstreamProfileCodexBorrowSource, HTTPUpstreamProfileCodexBorrowTarget: + HTTPUpstreamProfileOpenAIHarvest: return profile default: return HTTPUpstreamProfileDefault diff --git a/backend/internal/service/setting_codex_fingerprint.go b/backend/internal/service/setting_codex_fingerprint.go index bd0d62fd258c..1815b2196ddf 100644 --- a/backend/internal/service/setting_codex_fingerprint.go +++ b/backend/internal/service/setting_codex_fingerprint.go @@ -50,6 +50,7 @@ var codexFingerprintAccountEpoch atomic.Uint64 // Account mode edits invalidate only that account's subsequent WS handshakes. func notifyCodexFingerprintAccountChanged(accountID int64) { codexFingerprintAccountEpochs.Store(accountID, codexFingerprintAccountEpoch.Add(1)) + cancelBorrowPolicyObservers(accountID) } func currentCodexFingerprintPolicyForAccount(account *Account) *codexFingerprintPolicy { @@ -193,10 +194,14 @@ func (s *SettingService) codexFingerprintSettingsFromValues(values map[string]st } func publishCodexFingerprintSettings(view CodexFingerprintSettingsView) { + previousPolicy := currentCodexFingerprintPolicy() publishedCodexFingerprintPolicy.Store(&codexFingerprintPolicy{ enabled: view.Enabled, userAgent: view.EffectiveUserAgent, revision: codexFingerprintPolicyRevision(view.Enabled, view.EffectiveUserAgent), }) + if currentCodexFingerprintPolicy().revision != previousPolicy.revision { + cancelBorrowPolicyObservers(0) + } } func (s *SettingService) GetCodexFingerprintSettings(ctx context.Context) (CodexFingerprintSettingsView, error) { diff --git a/frontend/src/api/admin/codexGatewayBorrow.ts b/frontend/src/api/admin/codexGatewayBorrow.ts index d38a445d7dfb..e74c114730fe 100644 --- a/frontend/src/api/admin/codexGatewayBorrow.ts +++ b/frontend/src/api/admin/codexGatewayBorrow.ts @@ -21,7 +21,13 @@ export interface BorrowSourceStatus { error?: string } +export interface BorrowAcquisition { + trigger?: string; phase?: string; started_at?: string; finished_at?: string + reason?: string; error?: string; retry_after?: string +} + export interface BorrowTargetStatus extends BorrowSourceStatus { + acquisition?: BorrowAcquisition model: string cache_valid: boolean retry_after?: string @@ -61,6 +67,7 @@ export interface CodexBorrowDiagnosticEvent { export interface CodexBorrowDiagnosticRequest { account_id: number; model: string; transport: 'http' | 'ws'; request_limit?: number; scenario?: 'codex_session'; mode?: 'ordinary' | 'borrowed'; service_tier?: string } export interface CodexGatewayBorrowStatus { + acquisition?: BorrowAcquisition setup?: { state: string; phase: string; account_id: number; model: string; completed: number; total: number; failed: number; started_at: string; finished_at?: string; error?: string } recent_usage?: CodexGatewayBorrowUsage[] observed_since?: string diff --git a/frontend/src/i18n/locales/en/admin/codexGatewayBorrow.ts b/frontend/src/i18n/locales/en/admin/codexGatewayBorrow.ts index 14115d8fcc68..3661ccedcd2f 100644 --- a/frontend/src/i18n/locales/en/admin/codexGatewayBorrow.ts +++ b/frontend/src/i18n/locales/en/admin/codexGatewayBorrow.ts @@ -51,6 +51,9 @@ export default { saveFailed: 'Unable to save borrow configuration', refreshStatus: 'Refresh status', borrowStatus: 'Borrow status', + onDemandHint: 'Saving enabled settings prepares once. The next business request acquires and validates missing or expired routes. There is no renewal timer; viewing or refreshing never calls a model.', + onDemandExpired: 'Expired; waiting for a request to acquire a route', + modelScope: 'Astra follows ranxi v2.10.3. Sol is a downstream model extension and is validated independently. Cookie and two-shot STATE checks do not prove model identity or capability.', statusTitle: 'Codex Borrow Status', statusDescription: 'Inspect borrow routes for the saved configuration, then prepare or revalidate when needed.', preparationState: 'Preparation progress', @@ -152,7 +155,7 @@ export default { not_verified: 'This account and model have not completed route validation', not_configured: 'This account and model are absent from the saved configuration', disabled: 'Borrowing is currently disabled', - route_expired: 'The borrow route has expired; prepare or validate it again', + route_expired: 'The borrow route has expired; the next business request will acquire and validate it', source_passed_target_failed: 'The source returned a candidate but target validation failed. Replacement is bounded and demand-driven, with an acquisition cooldown.', source_returned_rejected_route: 'The source returned the same rejected route. Its lease was not extended; acquisition can resume after cooldown.', source_qualified: 'The source account can provide a borrow route', diff --git a/frontend/src/i18n/locales/zh/admin/codexGatewayBorrow.ts b/frontend/src/i18n/locales/zh/admin/codexGatewayBorrow.ts index 2b62348d6521..14e77b65a4d0 100644 --- a/frontend/src/i18n/locales/zh/admin/codexGatewayBorrow.ts +++ b/frontend/src/i18n/locales/zh/admin/codexGatewayBorrow.ts @@ -51,6 +51,9 @@ export default { saveFailed: '保存借用配置失败', refreshStatus: '刷新状态', borrowStatus: '借用状态', + onDemandHint: '保存启用配置后准备一次；线路缺失或过期时，由下一次业务请求自动获取并验证。没有定时续期。查看或刷新页面不会调用模型。', + onDemandExpired: '已过期，等待请求自动获取', + modelScope: 'Astra 按 ranxi v2.10.3 原版验证；Sol 为本站模型扩展，单独验证。验证只检查路由 Cookie 与两轮 STATE，不证明模型身份或能力。', statusTitle: 'Codex 借用状态', statusDescription: '查看已保存配置的借用线路，按需准备或重新验证。', preparationState: '准备进度', @@ -152,7 +155,7 @@ export default { not_verified: '此账号与模型尚未完成线路验证', not_configured: '此账号与模型未包含在已保存配置中', disabled: '借用功能当前关闭', - route_expired: '借用线路已过期，请重新准备或验证', + route_expired: '借用线路已过期，下一次业务请求会自动获取并验证', source_passed_target_failed: '来源返回了候选，但目标验证失败；按需限次重新获取，冷却期间不重复采集。', source_returned_rejected_route: '来源再次返回同一条失败线路，未延长租期；冷却后再按需尝试。', source_qualified: '来源账号可以提供借用线路', diff --git a/frontend/src/views/admin/CodexGatewayBorrowStatusView.vue b/frontend/src/views/admin/CodexGatewayBorrowStatusView.vue index 8545eb6d898e..39d21f106db6 100644 --- a/frontend/src/views/admin/CodexGatewayBorrowStatusView.vue +++ b/frontend/src/views/admin/CodexGatewayBorrowStatusView.vue @@ -55,7 +55,21 @@

{{ status.setup.phase === 'source' ? text('正在获取来源路由', 'Acquiring a source route') : `${accountName(status.setup.account_id)} · ${status.setup.model}` }}

{{ status.setup.error }}
+
+

{{ t('admin.codexGatewayBorrow.onDemandHint') }}

+

{{ acquisitionLabel }} · {{ formatTime(status.acquisition.started_at) }}

+
+ {{ t('admin.codexGatewayBorrow.technicalDetails') }} +
+
{{ text('触发原因', 'Trigger') }}: {{ status.acquisition.trigger }}
+
{{ text('完成时间', 'Finished') }}: {{ formatTime(status.acquisition.finished_at) }}
+
{{ t('admin.codexGatewayBorrow.retryAfter') }}: {{ formatTime(status.acquisition.retry_after) }}
+
+ 
{{ status.acquisition.error }}
+
+

{{ t('admin.codexGatewayBorrow.validationMeaning') }}

+

{{ t('admin.codexGatewayBorrow.modelScope') }}