From 9c68f6c1a42f9d730c8f64dfb6473d64bc4d8f7f Mon Sep 17 00:00:00 2001 From: Codex Date: Sun, 11 Oct 2026 01:17:41 +0800 Subject: [PATCH 1/2] fix(codex): align ranxi state echo and verify native borrowing tasks --- .downstream/native-domains.md | 29 +- .downstream/ranxi-borrow-reference.json | 82 +++ .downstream/upstream-review-ranxi-v2.10.5.md | 48 ++ .../generate_ranxi_borrow_reference.py | 123 ++++ .github/workflows/downstream-verify.yml | 3 + .../internal/service/account_test_service.go | 10 +- .../service/account_test_sse_keepalive.go | 118 ++++ .../account_test_sse_keepalive_test.go | 192 ++++++ .../internal/service/codex_gateway_borrow.go | 46 +- .../service/codex_gateway_borrow_diagnose.go | 59 +- .../service/codex_gateway_borrow_probe.go | 7 +- ...dex_gateway_borrow_ranxi_generated_test.go | 644 ++++++++++++++++++ ...dex_gateway_borrow_ranxi_reference_test.go | 141 ++++ .../service/codex_gateway_borrow_replay.go | 94 +++ .../codex_gateway_borrow_replay_test.go | 98 +++ .../service/codex_gateway_borrow_transport.go | 91 +++ .../codex_gateway_borrow_transport_test.go | 75 ++ .../service/openai_codex_turn_state.go | 11 +- .../service/openai_codex_turn_state_test.go | 12 +- .../service/openai_compact_sse_keepalive.go | 6 + 20 files changed, 1837 insertions(+), 52 deletions(-) create mode 100644 .downstream/ranxi-borrow-reference.json create mode 100644 .downstream/upstream-review-ranxi-v2.10.5.md create mode 100644 .github/scripts/generate_ranxi_borrow_reference.py create mode 100644 backend/internal/service/account_test_sse_keepalive.go create mode 100644 backend/internal/service/account_test_sse_keepalive_test.go create mode 100644 backend/internal/service/codex_gateway_borrow_ranxi_generated_test.go create mode 100644 backend/internal/service/codex_gateway_borrow_ranxi_reference_test.go create mode 100644 backend/internal/service/codex_gateway_borrow_replay.go create mode 100644 backend/internal/service/codex_gateway_borrow_replay_test.go create mode 100644 backend/internal/service/codex_gateway_borrow_transport.go create mode 100644 backend/internal/service/codex_gateway_borrow_transport_test.go diff --git a/.downstream/native-domains.md b/.downstream/native-domains.md index 44fca4557c73..9cd36fe75ad9 100644 --- a/.downstream/native-domains.md +++ b/.downstream/native-domains.md @@ -93,7 +93,7 @@ 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. -The protocol reference is ranxi v2.10.3 (`fd1b5ee4`), non-rotating branch. +The protocol reference is ranxi v2.10.5 (`d977e7a5`), 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 @@ -111,6 +111,15 @@ headers, and generate a new legacy `session_id` per shot. The second shot carrie 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`. +Business STATE echo follows the original provenance rule: unknown or expired +process-local provenance preserves the caller's STATE; a live foreign-account +record strips it. This does not relax two-shot target qualification. + +Observations report their connect/TLS/write/header phase and attempt count. +They may retry once only with transport trace proving a connection failure before +HTTP request headers were written. Unobserved errors, responses, written headers, +cancellation or an exhausted diagnostic budget forbid retry. Ordinary business +generations retain their existing retry policy. The original cache identity includes route cookie, proxy, Authorization, ChatGPT-Account-ID, UA, Originator, Version, business STATE and TLS profile. @@ -213,10 +222,24 @@ candidate owner. `probe_context.subject_role` distinguishes source self-checks from target checks. Self-check results do not qualify business traffic or mutate the source's health/status. -Ranxi reference: v2.10.3 (`fd1b5ee4eeb20961fbb783fa6f136a1704271e90`). Its +Administrator `scenario: codex_replay` accepts explicit HTTP `mode: ordinary` or +`borrowed` and `replay: {body, headers}` for controlled task/tool comparisons. +The JSON body is limited to 1 MiB, must select the requested model, stream with +`store: false`, use complete history without `previous_response_id`, and carry a +`codex-acceptance-` session in both prompt-cache and native session/thread headers. +Only native client identity/metadata/STATE headers are accepted; auth, cookies, +URL overrides and duplicate fields are rejected. The ordinary source control +must use a saved source and configured model; borrowing still requires a saved +target. Each call retains the eight-physical-attempt cap and a five-minute +generation budget. Results return actual business `turn_state` for the next +isolated replay turn; probe STATE is never returned or copied. The controller +executes synthetic tools locally and preserves full-history items, rather than +inventing tool results or executing model code in the production service. + +Ranxi reference: v2.10.5 (`d977e7a5e24e5536db6ce8b8478743e0c4b11f77`). Its fingerprint, gateway-cookie, target-probe and automatic-setup core files are unchanged from v2.10.2. The scoped update and retained host adaptations are -listed in [the review](upstream-review-ranxi-v2.10.3.md). OAuth search-history compatibility +listed in [the review](upstream-review-ranxi-v2.10.5.md). OAuth search-history compatibility is supplied by the official v0.2.15 integration. Automatic group membership, Mihomo exit rotation and removal of harvesting attempt/concurrency bounds are not adopted; existing account mappings, network routes and protocol switches stay diff --git a/.downstream/ranxi-borrow-reference.json b/.downstream/ranxi-borrow-reference.json new file mode 100644 index 000000000000..a466d1d410ed --- /dev/null +++ b/.downstream/ranxi-borrow-reference.json @@ -0,0 +1,82 @@ +{ + "repository": "ranxi2001/sub2api", + "tag": "v2.10.5", + "commit": "d977e7a5e24e5536db6ce8b8478743e0c4b11f77", + "adaptations": [ + "Test-only identifier prefix; original function bodies.", + "Host HTTPUpstream, TLS profile, context markers, turn-state extractor and error display helpers.", + "Route container retains only cookie/expires; unused Mihomo/proxy fields are omitted.", + "Source payload calls the original Pelican builder with Astra, OAuth, fixed prompt and medium effort." + ], + "files": { + "backend/internal/service/openai_codex_turn_state.go": { + "sha256": "d5c7720ad556e68bd4505c6dcec009d8489141e5bd4b6c7984d45a38430c7440", + "declarations": { + "guardOpenAICodexTurnStateEcho": "5ce7aabcb3057409d9b746c333a14b70823e0847299f8836c6a9a5d342b938c2" + } + }, + "backend/internal/service/openai_codex_state_probe.go": { + "sha256": "ee7de916510cf6d48c94a542c731aa52fd7c4b0e8dc410b0d217229df05e99ba", + "declarations": { + "runOpenAICodexStateProbe": "eccfd327d172e198d947dcead28d77a5921e60a90c2e00e5660f56ded41a86c6", + "ProbeOpenAICodexStateRoute": "13e1cf1f0d8492d65aced50bec502350506da3c0986f56091bafda9162d15ec1", + "fail": "de12462962b42aab47fef6c33329d80648f6089388f1159d69ad5629964a1897", + "shotUsable": "9e74562f9555a5cfa40fd8c6237191ea0c2507577af5ebdf0a69f3c3507a04b4", + "openAICodexStateVerdictReason": "936f726961005cc672e1e30ab293c92f4f00f29b87c4bb103f1b6202c00f62f1", + "fireOpenAICodexStateShotRequest": "9b3b1c7cfd29e9274ed3e28034e2abd1e39ab6c7d93eec13eca2d31eb19b6256", + "openAICodexStateRouteCookies": "f1c37c5f2411b18b2e1ad72b8cf68fa5e6e25c31bbb3cbcd4f77ba633e7ae68b", + "openAICodexStateDetail": "b35df3e1ee98dfb8ad423a8f37826e5ee4ec971b3329f338d3b62617f9bd70c4", + "openAICodexStateStreamEvents": "734c354320e4a9fe9debb331cafb747040de4e7b9c53c113357eaf9a86036107", + "openAICodexStateStreamModel": "6e6e64daea561c6ef38f52472f105b4670caab7de3a2bfe039751d3126df4e44", + "openAICodexStateStreamErrorPayload": "76abf87c7a62cd22398fa7779e15bcf140a2850b26a5d507c92a07f43dea1496" + } + }, + "backend/internal/service/openai_codex_probe_response.go": { + "sha256": "cff3af2e6c01bbc7308ec8a4f9ea4596ebfd96689e38e03ef09761c9b2eb6a88", + "declarations": { + "validateCodexProbeResponse": "43b9a73bf19ca3fd04c42f85150459dfd5d3823b8f963463cdc06d1726c4daea" + } + }, + "backend/internal/service/openai_codex_ws_anchor.go": { + "sha256": "b5a3100719948d0f6c5f6a4052f7bbb70d668eaa3ce4e4480d2edd3706adabb5", + "declarations": { + "replaceCodexWSAnchorCookie": "ee9b8934f40b8a0d33b9ced3e716bd21d2cdb81bae4e345c69782231d0269f3f" + } + }, + "backend/internal/service/openai_codex_ticket.go": { + "sha256": "186ea9b390324b58613948a1371944a0eb6b6457c3aa52ff6f24fddb35d59831", + "declarations": { + "jsonString": "d595eca92f7086de35294d4fc27edf07df49d70b057642fe77c2510e909ee311", + "normalizeOpenAICodexTicketModel": "8b4bf5d56064064308040b1a1eda56b220e76f2cf078a59c6042b48a3b940b70" + } + }, + "backend/internal/repository/codex_gateway_pin.go": { + "sha256": "e9f6eb813a440282c9d39ab5d4e03a96f8862438bcfd143fa61c1ac23513bdfb", + "declarations": { + "codexGatewayCookiePathMatches": "91628f271432d6fb6426783b670e1ed08f7c72811565963d2f13255e840958bc", + "codexGatewayRouteFromResponse": "31a7c69a8a39c482f01be40b3b6909b07140400f5893cd990890346087e60d85" + } + }, + "backend/internal/service/account_test_service.go": { + "sha256": "6c90400653ac6ab22987cb717db235530a353af4dc0e00ed35fd77f876283fcd", + "declarations": { + "createOpenAITestPayload": "35451845d4ae6c6b373ec9afe050fd2606f73f7fecef795662eb3cba2e94fe86" + } + }, + "backend/internal/service/pelican_test_service.go": { + "sha256": "a601a52aae9ccef507d90b7faa1f1cc029e9e907af735e949a236883f87bf4ee", + "declarations": { + "createPelicanOpenAIPayload": "56e2c4d395beefb8f564d331ffe2afc98ed4018c1ca1ab33ab9917f07cdf0a55", + "promptOrDefault": "b807f11ad9bc4f9e6d90d6df9bb2b1a60024731553893041ad9b87f6866acdcd", + "normalizePelicanReasoningEffort": "50d726f8594b9f5ad0dcf95921b46b73ab4a8e7078873d3281352d46b5ce6934" + } + } + }, + "generated_sha256": "9ad61fb4b2a7d3230f374568ba6145635a8434579763492b292ee0ce345c1f48", + "reviewed_host_policies": { + "backend/internal/repository/astra_target_validation.go": "420b9a2f631ae382ee5455aa922b71bdde413bfbe3ea25b048d8fc0d55f9f1d5", + "backend/internal/repository/astra_routing_upstream.go": "3969048bfe953dbded5a686378eb88be52b685cfb208f9f0a97a10c52f337b13", + "backend/internal/service/astra_automatic_setup.go": "c60fc40730d667ae5b8f55d6084c9961073a107fe7ba40e68d577f722389874f", + "backend/internal/service/openai_codex_fingerprint.go": "c974262d3b0fff3c8d1eb8286078234be2905b2402245f4a46b4e9bced63eeb9" + } +} diff --git a/.downstream/upstream-review-ranxi-v2.10.5.md b/.downstream/upstream-review-ranxi-v2.10.5.md new file mode 100644 index 000000000000..d6973bbf4e5a --- /dev/null +++ b/.downstream/upstream-review-ranxi-v2.10.5.md @@ -0,0 +1,48 @@ +# ranxi v2.10.5 borrowing review + +Reference: `d977e7a5e24e5536db6ce8b8478743e0c4b11f77`, released v2.10.5. +Official Wei-Shaw v0.2.15 remains the downstream base. This is a scoped update, +not a merge of ranxi's database, BPS, scheduling, or exit-management features. + +## Protocol evidence + +`.github/scripts/generate_ranxi_borrow_reference.py` extracts original functions +from the pinned Git object, prefixes test-only identifiers, and records source +and generated hashes in `ranxi-borrow-reference.json`. It does not translate the +reference algorithm. `--check` verifies provenance offline when that Git object +is available. The generated Go file is test-only and needs no network in CI. + +The differential test covers the original source payload, two-shot target +requests and verdicts, cookie scope/deletion/expiry, and business STATE echo. +Host-dependent preparation, fingerprint and cache policies have source hashes +and retain the previously reviewed account coordination and observation adapters. +These hashes are a review boundary, not a claim that those whole subsystems were +executed as a second reference service. + +The source/pool/probe core is unchanged from v2.10.3. Sol remains an independent +downstream extension; an Astra result never qualifies Sol. + +## Changes + +- Restore original business STATE echo behavior: unknown or expired local + provenance is preserved; a live record identifying a different account still + strips it. Target two-shot qualification is unchanged. +- Observe connect/TLS/write/header phases. Retry an observation once only when + transport trace proves failure before any HTTP request headers were written. + No trace, any response, cancellation, exhausted diagnostic budget, or written + headers forbids replay. Every connection attempt counts against the budget. +- Port account-test SSE keepalive from #350, preserving the host parser's error + wording and observation cancellation. A heartbeat is not model output. +- Extend the existing administrator diagnostic with an explicitly selected, + bounded native replay for comparable fixture/tool sessions. Ordinary source + controls are limited to saved source accounts. No credentials, routing cookies, + destination override or previous-response anchor can be supplied by replay. + +The production account-proxy selection and live model results are deployment +evidence, not protocol unit-test results. No recurring ranxi update is installed. + +## Excluded release changes + +BPS credentials/Redis/tool routing, optional timezone rewriting, subscription +renewal, Grok video, automatic grouping and Mihomo rotation are outside this fix. +The existing WS switch and official upstream synchronization remain unchanged. diff --git a/.github/scripts/generate_ranxi_borrow_reference.py b/.github/scripts/generate_ranxi_borrow_reference.py new file mode 100644 index 000000000000..9e405ab67f14 --- /dev/null +++ b/.github/scripts/generate_ranxi_borrow_reference.py @@ -0,0 +1,123 @@ +#!/usr/bin/env python3 +"""Extract pinned upstream declarations; no hand-translated reference algorithm. + +Run from the repository root. --check validates the committed generated file and +manifest against Git objects, without network access or rewriting files. +""" +import argparse +import hashlib +import json +from pathlib import Path +import re +import subprocess + +REVISION = 'd977e7a5e24e5536db6ce8b8478743e0c4b11f77' +ROOT = Path(__file__).resolve().parents[2] +OUTPUT = ROOT / 'backend/internal/service/codex_gateway_borrow_ranxi_generated_test.go' +MANIFEST = ROOT / '.downstream/ranxi-borrow-reference.json' +FILES = { + 'backend/internal/service/openai_codex_turn_state.go': ['guardOpenAICodexTurnStateEcho'], + 'backend/internal/service/openai_codex_state_probe.go': [ + 'runOpenAICodexStateProbe', 'ProbeOpenAICodexStateRoute', 'fail', + 'shotUsable', 'openAICodexStateVerdictReason', + 'fireOpenAICodexStateShotRequest', 'openAICodexStateRouteCookies', + 'openAICodexStateDetail', 'openAICodexStateStreamEvents', + 'openAICodexStateStreamModel', 'openAICodexStateStreamErrorPayload'], + 'backend/internal/service/openai_codex_probe_response.go': ['validateCodexProbeResponse'], + 'backend/internal/service/openai_codex_ws_anchor.go': ['replaceCodexWSAnchorCookie'], + 'backend/internal/service/openai_codex_ticket.go': ['jsonString', 'normalizeOpenAICodexTicketModel'], + 'backend/internal/repository/codex_gateway_pin.go': ['codexGatewayCookiePathMatches', 'codexGatewayRouteFromResponse'], + 'backend/internal/service/account_test_service.go': ['createOpenAITestPayload'], + 'backend/internal/service/pelican_test_service.go': ['createPelicanOpenAIPayload', 'promptOrDefault', 'normalizePelicanReasoningEffort'], +} + +def read(path): + return subprocess.check_output(['git', 'show', f'{REVISION}:{path}'], cwd=ROOT) + +def extract(source, name): + pattern = r'(?m)^func (?:\([^\n]*\) )?' + re.escape(name) + r'\([^\n]*' + start = re.search(pattern, source) + if not start: + raise ValueError('missing declaration: ' + name) + end = re.search(r'(?m)^}', source[start.start():]) + if not end: + raise ValueError('unterminated declaration: ' + name) + return source[start.start():start.start() + end.end()] + '\n' + +def generate(): + manifest = {'repository': 'ranxi2001/sub2api', 'tag': 'v2.10.5', 'commit': REVISION, + 'adaptations': ['Test-only identifier prefix; original function bodies.', + 'Host HTTPUpstream, TLS profile, context markers, turn-state extractor and error display helpers.', + 'Route container retains only cookie/expires; unused Mihomo/proxy fields are omitted.', + 'Source payload calls the original Pelican builder with Astra, OAuth, fixed prompt and medium effort.'], 'files': {}} + parts = [] + names = set() + for path, functions in FILES.items(): + raw = read(path) + source = raw.decode('utf-8') + fragments = [extract(source, name) for name in functions] + names.update(functions) + if path.endswith('openai_codex_state_probe.go'): + type_names = ['OpenAICodexStateVerdict', 'OpenAICodexStateProbeResult', 'openAICodexStateShot', 'openAICodexStateProbeContextKey'] + for name in type_names: + # Multi-line structs, unlike the string alias and empty key. + start = source.index('type ' + name + ' ') + line_end = source.index('\n', start) + if source[start:line_end].endswith('{'): + end = re.search(r'(?m)^}', source[line_end:]) + fragment = source[start:line_end+end.end()]+'\n' + else: + fragment = source[start:line_end]+'\n' + fragments.append(fragment) + names.add(name) + for block in re.findall(r'(?m)^const \([\s\S]*?^\)', source): + fragments.append(block+'\n') + names.update(re.findall(r'(?m)^\t([A-Za-z]\w*)', block)) + for name, fragment in zip(functions, fragments): + manifest['files'].setdefault(path, {'sha256': hashlib.sha256(raw).hexdigest(), 'declarations': {}})['declarations'][name] = hashlib.sha256(fragment.encode()).hexdigest() + parts.extend(fragments) + names.add('codexGatewayRoute') + combined = '\n'.join(parts) + combined = re.sub(r'\b('+'|'.join(sorted(names, key=len, reverse=True))+r')\b', lambda m: 'ranxiRef'+m[0][0].upper()+m[0][1:], combined) + header = '''// Code generated by .github/scripts/generate_ranxi_borrow_reference.py; DO NOT EDIT. +// ranxi2001/sub2api v2.10.5, LGPL-3.0, commit '''+REVISION+'''. +package service +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "strings" + "time" + "github.com/google/uuid" + "github.com/gin-gonic/gin" + "github.com/Wei-Shaw/sub2api/internal/pkg/openai" + "github.com/Wei-Shaw/sub2api/internal/pkg/tlsfingerprint" +) +type ranxiRefCodexGatewayRoute struct { cookie http.Cookie; expires time.Time } +''' + data = subprocess.run(['gofmt'], input=(header+combined).encode(), capture_output=True, check=True).stdout.replace(b'\r\n', b'\n') + manifest['generated_sha256'] = hashlib.sha256(data).hexdigest() + # Adjacent host-dependent policies are reviewed by source hash, not replaced + # with a second hand-written implementation or claimed as executed functions. + manifest['reviewed_host_policies'] = {p: hashlib.sha256(read(p)).hexdigest() for p in [ + 'backend/internal/repository/astra_target_validation.go', + 'backend/internal/repository/astra_routing_upstream.go', + 'backend/internal/service/astra_automatic_setup.go', + 'backend/internal/service/openai_codex_fingerprint.go']} + return data, (json.dumps(manifest, indent=2)+'\n').encode() + +if __name__ == '__main__': + parser = argparse.ArgumentParser(); parser.add_argument('--check', action='store_true') + args = parser.parse_args() + generated, manifest = generate() + if args.check: + assert OUTPUT.read_bytes().replace(b'\r\n', b'\n') == generated + assert MANIFEST.read_bytes().replace(b'\r\n', b'\n') == manifest + else: + OUTPUT.write_bytes(generated); MANIFEST.write_bytes(manifest) + print('Pinned ranxi reference '+('verified' if args.check else 'generated')) diff --git a/.github/workflows/downstream-verify.yml b/.github/workflows/downstream-verify.yml index 2a0d1d1d4ac1..7514d3190bc6 100644 --- a/.github/workflows/downstream-verify.yml +++ b/.github/workflows/downstream-verify.yml @@ -51,6 +51,9 @@ jobs: - name: Codex preparation coordination race tests working-directory: backend run: go test -p=1 -race -tags unit ./internal/service -run '^TestCodexBorrow(Flight|DifferentTargets|CancelledParent|DiagnoseHTTPTerminalAndDispatchEvidence)|^TestCodexGatewayBorrowRunnerCancellationStopsQueuedAndInflightAndRetainsRawOutput$' -count=1 + - name: Account test keepalive lifecycle race tests + working-directory: backend + run: go test -p=1 -race -tags unit ./internal/service -run '^TestAccountTestKeepalive' -count=1 - name: Upstream automation classifier tests run: | python3 .github/scripts/test_upstream_risk.py diff --git a/backend/internal/service/account_test_service.go b/backend/internal/service/account_test_service.go index 64b65c839d5e..fd9f4011d172 100644 --- a/backend/internal/service/account_test_service.go +++ b/backend/internal/service/account_test_service.go @@ -429,12 +429,20 @@ func createTestPayload(modelID string, prompts ...string) (map[string]any, error // modelID is optional - if empty, defaults to claude.DefaultTestModel // mode is optional - "compact" routes OpenAI accounts to the /responses/compact probe path // opts is optional media (image/audio data URLs for real generation / STT). -func (s *AccountTestService) TestAccountConnection(c *gin.Context, accountID int64, modelID string, prompt string, mode string, opts ...AccountTestOptions) error { +func (s *AccountTestService) TestAccountConnection(c *gin.Context, accountID int64, modelID string, prompt string, mode string, opts ...AccountTestOptions) (testErr error) { if err := ValidateAccountTextTestPrompt(prompt, modelID, mode); err != nil { return s.sendErrorAndEnd(c, err.Error()) } c.Set(accountTestPromptContextKey, prompt) initAccountTestLogger(c, accountID, modelID, mode) + stopKeepalive := startAccountTestSSEKeepalive(c, 10*time.Second) + defer func() { + // A failed downstream write cancels the observation, never its owner. + if testErr == nil { + testErr = c.Request.Context().Err() + } + stopKeepalive() + }() ctx := c.Request.Context() testOpts := firstAccountTestOptions(opts) diff --git a/backend/internal/service/account_test_sse_keepalive.go b/backend/internal/service/account_test_sse_keepalive.go new file mode 100644 index 000000000000..e76372ea5d6c --- /dev/null +++ b/backend/internal/service/account_test_sse_keepalive.go @@ -0,0 +1,118 @@ +// Adapted from ranxi2001/sub2api d977e7a5e24e5536db6ce8b8478743e0c4b11f77 (LGPL-3.0). +package service + +import ( + "context" + "log" + "net/http" + "time" + + "github.com/gin-gonic/gin" +) + +// startAccountTestSSEKeepalive 在首包等待和正文静默期间保活；注释不构成模型输出。 +// 复用已有心跳器的互斥锁，但业务事件后仍继续保活，退出时等待协程释放 writer。 +func startAccountTestSSEKeepalive(c *gin.Context, interval time.Duration) func() { + originalWriter := c.Writer + originalContext := c.Request.Context() + ctx, cancel := context.WithCancel(originalContext) + c.Request = c.Request.WithContext(ctx) + k := &openAICompactSSEKeepalive{writer: originalWriter, stop: make(chan struct{})} + w := &accountTestKeepaliveWriter{ + openAICompactKeepaliveWriter: &openAICompactKeepaliveWriter{ResponseWriter: originalWriter, k: k}, + ctx: ctx, cancel: cancel, + } + c.Writer = w + requestPath := "background" + if c.Request.URL != nil { + requestPath = c.Request.URL.Path + } + // 立即提交真实的 SSE 字节，让代理在上游尚未返回 HTTP 响应时也能看到连接已建立。 + if ctx.Err() == nil && !k.beat() { + log.Printf("[WARN] account test SSE initial write failed path=%s; canceling upstream", requestPath) + cancel() + } + done := make(chan struct{}) + go func() { + defer close(done) + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-k.stop: + return + case <-ticker.C: + if ctx.Err() != nil { + return + } + if !k.beat() { + log.Printf("[WARN] account test SSE keepalive write failed path=%s; canceling upstream", requestPath) + cancel() + return + } + } + } + }() + return func() { + k.Stop() + cancel() + <-done + c.Writer = originalWriter + // 成功测试后的限流恢复仍需使用调用方的有效上下文。 + c.Request = c.Request.WithContext(originalContext) + } +} + +type accountTestKeepaliveWriter struct { + *openAICompactKeepaliveWriter + ctx context.Context + cancel context.CancelFunc +} + +func (w *accountTestKeepaliveWriter) Header() http.Header { + w.k.mu.Lock() + w.k.paused = true + w.k.mu.Unlock() + return w.ResponseWriter.Header() +} + +func (w *accountTestKeepaliveWriter) Write(data []byte) (int, error) { + w.k.mu.Lock() + defer w.k.mu.Unlock() + if err := w.ctx.Err(); err != nil { + return 0, err + } + w.k.paused = false + n, err := w.ResponseWriter.Write(data) + if err != nil { + w.cancel() + } + return n, err +} + +func (w *accountTestKeepaliveWriter) WriteString(value string) (int, error) { + return w.Write([]byte(value)) +} + +func (w *accountTestKeepaliveWriter) WriteHeader(code int) { + w.k.mu.Lock() + defer w.k.mu.Unlock() + w.ResponseWriter.WriteHeader(code) +} + +func (w *accountTestKeepaliveWriter) WriteHeaderNow() { + w.k.mu.Lock() + defer w.k.mu.Unlock() + w.ResponseWriter.WriteHeaderNow() +} + +func (w *accountTestKeepaliveWriter) Flush() { + w.k.mu.Lock() + defer w.k.mu.Unlock() + if w.ctx.Err() == nil { + w.k.paused = false + w.ResponseWriter.Flush() + } +} diff --git a/backend/internal/service/account_test_sse_keepalive_test.go b/backend/internal/service/account_test_sse_keepalive_test.go new file mode 100644 index 000000000000..4b4098a0bb88 --- /dev/null +++ b/backend/internal/service/account_test_sse_keepalive_test.go @@ -0,0 +1,192 @@ +// Adapted from ranxi2001/sub2api d977e7a5e24e5536db6ce8b8478743e0c4b11f77 (LGPL-3.0). +package service + +import ( + "context" + "errors" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" +) + +func TestAccountTestKeepaliveContinuesAfterEventsAndHeaders(t *testing.T) { + c, rec := newCompactBridgeTestContext(t, false) + original := c.Writer + stop := startAccountTestSSEKeepalive(c, keepaliveTestInterval) + defer stop() + s := &AccountTestService{} + s.sendEvent(c, TestEvent{Type: "test_start", Model: "gpt-6-astra"}) + c.Writer.Header().Set("Content-Type", "text/event-stream") + c.Writer.Flush() + time.Sleep(5 * keepaliveTestInterval) + s.sendEvent(c, TestEvent{Type: "content", Text: "ok"}) + time.Sleep(5 * keepaliveTestInterval) + s.sendEvent(c, TestEvent{Type: "test_complete", Success: true}) + stop() + stop() + + // 停拍后再读 recorder，避免测试自身与后台 writer 产生竞态。 + body := rec.Body.String() + require.Same(t, original, c.Writer) + require.NoError(t, c.Request.Context().Err()) + require.GreaterOrEqual(t, strings.Count(body, ": keepalive\n\n"), 3) + require.Contains(t, strings.Split(body, `"type":"content"`)[1], ": keepalive\n\n") + text, errMsg, _ := parseTestSSEOutput(body) + require.Empty(t, errMsg) + require.Equal(t, "ok", text) + require.Equal(t, 1, strings.Count(body, `"type":"test_complete"`)) + time.Sleep(3 * keepaliveTestInterval) + require.Equal(t, body, rec.Body.String()) +} + +func TestAccountTestKeepaliveAlreadyCanceledDoesNotWrite(t *testing.T) { + c, rec := newCompactBridgeTestContext(t, false) + ctx, cancel := context.WithCancel(c.Request.Context()) + c.Request = c.Request.WithContext(ctx) + cancel() + stop := startAccountTestSSEKeepalive(c, keepaliveTestInterval) + stop() + require.Empty(t, rec.Body.String()) +} + +func TestAccountTestKeepaliveStreamTerminalSemantics(t *testing.T) { + for _, tc := range []struct { + name, upstream, text, errorText string + success bool + }{ + {"completed", "data: {\"type\":\"response.output_text.delta\",\"delta\":\"ok\"}\n\ndata: {\"type\":\"response.completed\"}\n\n", "ok", "", true}, + {"failed", "data: {\"type\":\"response.failed\",\"response\":{\"error\":{\"message\":\"probe failed\"}}}\n\n", "", "probe failed", false}, + {"truncated", "data: {\"type\":\"response.created\"}\n\n", "", "stream ended before terminal (last event: response.created)", false}, + } { + t.Run(tc.name, func(t *testing.T) { + c, rec := newCompactBridgeTestContext(t, false) + stop := startAccountTestSSEKeepalive(c, keepaliveTestInterval) + defer stop() + reader, writer := io.Pipe() + producerDone := make(chan struct{}) + go func() { + defer close(producerDone) + time.Sleep(5 * keepaliveTestInterval) + _, _ = io.WriteString(writer, tc.upstream) + _ = writer.Close() + }() + err := (&AccountTestService{}).processOpenAIStream(c, &Account{ID: 1}, reader) + _ = reader.Close() + <-producerDone + stop() + require.Equal(t, tc.success, err == nil) + text, errMsg, _ := parseTestSSEOutput(rec.Body.String()) + require.Equal(t, tc.text, text) + require.Equal(t, tc.errorText, errMsg) + require.Equal(t, tc.success, strings.Contains(rec.Body.String(), `"type":"test_complete"`)) + require.GreaterOrEqual(t, strings.Count(rec.Body.String(), ": keepalive\n\n"), 2) + }) + } +} + +type failingAccountTestWriter struct { + gin.ResponseWriter +} + +func (w *failingAccountTestWriter) Write([]byte) (int, error) { + return 0, errors.New("client disconnected") +} + +func TestAccountTestKeepaliveCancelsOnWriteFailure(t *testing.T) { + for _, heartbeat := range []bool{false, true} { + t.Run(map[bool]string{false: "event", true: "heartbeat"}[heartbeat], func(t *testing.T) { + c, _ := newCompactBridgeTestContext(t, false) + stop := startAccountTestSSEKeepalive(c, keepaliveTestInterval) + defer stop() + w, ok := c.Writer.(*accountTestKeepaliveWriter) + require.True(t, ok, "account test writer must serialize keepalive and business events") + w.k.mu.Lock() + broken := &failingAccountTestWriter{ResponseWriter: w.ResponseWriter} + w.ResponseWriter = broken + w.k.writer = broken + w.k.mu.Unlock() + if !heartbeat { + _, err := w.WriteString("data: content\n\n") + require.Error(t, err) + } + select { + case <-c.Request.Context().Done(): + case <-time.After(time.Second): + t.Fatal("upstream context was not canceled") + } + }) + } +} + +func TestAccountTestKeepaliveCancellationReleasesReader(t *testing.T) { + c, rec := newCompactBridgeTestContext(t, false) + parent, cancel := context.WithCancel(c.Request.Context()) + c.Request = c.Request.WithContext(parent) + stop := startAccountTestSSEKeepalive(c, keepaliveTestInterval) + reader, writer := io.Pipe() + released := make(chan struct{}) + ctx := c.Request.Context() + go func() { + defer close(released) + <-ctx.Done() + _ = writer.CloseWithError(ctx.Err()) + }() + go cancel() + err := (&AccountTestService{}).processOpenAIStream(c, &Account{ID: 1}, reader) + _ = reader.Close() + <-released + stop() + require.Error(t, err) + require.NotContains(t, rec.Body.String(), `"type":"test_complete"`) +} + +// 真实 HTTP 首包等待：客户端必须先收到 SSE，再等上游响应，取消后两端都要退出。 +func TestAccountTestKeepaliveBeforeUpstreamHeaders(t *testing.T) { + upstreamStarted := make(chan struct{}) + upstreamDone := make(chan struct{}) + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + close(upstreamStarted) + defer close(upstreamDone) + <-r.Context().Done() + })) + defer upstream.Close() + handlerDone := make(chan struct{}) + downstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + defer close(handlerDone) + c, _ := gin.CreateTestContext(w) + c.Request = r + stop := startAccountTestSSEKeepalive(c, keepaliveTestInterval) + defer stop() + req, _ := http.NewRequestWithContext(c.Request.Context(), http.MethodGet, upstream.URL, nil) + resp, err := http.DefaultClient.Do(req) + if err == nil { + _ = resp.Body.Close() + } + })) + defer downstream.Close() + client := &http.Client{Timeout: 2 * time.Second} + resp, err := client.Get(downstream.URL) + require.NoError(t, err) + require.Equal(t, "text/event-stream", resp.Header.Get("Content-Type")) + first := make([]byte, len(": keepalive\n\n")) + _, err = io.ReadFull(resp.Body, first) + require.NoError(t, err) + require.Equal(t, ": keepalive\n\n", string(first)) + <-upstreamStarted + _, err = io.ReadFull(resp.Body, first) + require.NoError(t, err) + _ = resp.Body.Close() + for _, done := range []<-chan struct{}{handlerDone, upstreamDone} { + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("HTTP cancellation did not release handler/upstream") + } + } +} diff --git a/backend/internal/service/codex_gateway_borrow.go b/backend/internal/service/codex_gateway_borrow.go index 294d78e32c74..decee424920a 100644 --- a/backend/internal/service/codex_gateway_borrow.go +++ b/backend/internal/service/codex_gateway_borrow.go @@ -199,27 +199,31 @@ type CodexGatewayBorrowSourceStatus struct { } type CodexGatewayBorrowVerification struct { - MintRequestEncoding string `json:"mint_request_encoding,omitempty"` - ContinueRequestEncoding string `json:"continue_request_encoding,omitempty"` - RequestShape string `json:"request_shape,omitempty"` - ServiceTier string `json:"service_tier,omitempty"` - MintCompleted bool `json:"mint_completed"` - ContinueCompleted bool `json:"continue_completed"` - MintStateLength int `json:"mint_state_length"` - ContinueStateLength int `json:"continue_state_length"` - RouteChanged bool `json:"route_changed"` - AccountID int64 `json:"account_id"` - Model string `json:"model"` - Success bool `json:"success"` - Reason string `json:"reason"` - Error string `json:"error,omitempty"` - MintStatus int `json:"mint_status"` - ContinueStatus int `json:"continue_status"` - Minted bool `json:"minted"` - NewTicket bool `json:"new_ticket"` - ReportedModel string `json:"reported_model,omitempty"` - CheckedAt time.Time `json:"checked_at"` - ExpiresAt *time.Time `json:"expires_at,omitempty"` + MintTransportPhase string `json:"mint_transport_phase,omitempty"` + ContinueTransportPhase string `json:"continue_transport_phase,omitempty"` + MintTransportAttempts int `json:"mint_transport_attempts,omitempty"` + ContinueTransportAttempts int `json:"continue_transport_attempts,omitempty"` + MintRequestEncoding string `json:"mint_request_encoding,omitempty"` + ContinueRequestEncoding string `json:"continue_request_encoding,omitempty"` + RequestShape string `json:"request_shape,omitempty"` + ServiceTier string `json:"service_tier,omitempty"` + MintCompleted bool `json:"mint_completed"` + ContinueCompleted bool `json:"continue_completed"` + MintStateLength int `json:"mint_state_length"` + ContinueStateLength int `json:"continue_state_length"` + RouteChanged bool `json:"route_changed"` + AccountID int64 `json:"account_id"` + Model string `json:"model"` + Success bool `json:"success"` + Reason string `json:"reason"` + Error string `json:"error,omitempty"` + MintStatus int `json:"mint_status"` + ContinueStatus int `json:"continue_status"` + Minted bool `json:"minted"` + NewTicket bool `json:"new_ticket"` + ReportedModel string `json:"reported_model,omitempty"` + CheckedAt time.Time `json:"checked_at"` + ExpiresAt *time.Time `json:"expires_at,omitempty"` } type CodexGatewayBorrowTargetStatus struct { diff --git a/backend/internal/service/codex_gateway_borrow_diagnose.go b/backend/internal/service/codex_gateway_borrow_diagnose.go index cd5e980772c5..baa0fa271c20 100644 --- a/backend/internal/service/codex_gateway_borrow_diagnose.go +++ b/backend/internal/service/codex_gateway_borrow_diagnose.go @@ -23,6 +23,7 @@ type codexBorrowDiagnosticContextKey struct{} type codexBorrowAppliedContextKey struct{} type codexBorrowDiagnostic struct { + replay *CodexBorrowReplay scenario string serviceTier string history []any @@ -72,17 +73,19 @@ func consumeBorrowDiagnosticRequest(ctx context.Context) error { } type CodexBorrowDiagnosticRequest struct { - Comparison string `json:"comparison,omitempty"` - Scenario string `json:"scenario,omitempty"` - Mode string `json:"mode,omitempty"` - ServiceTier string `json:"service_tier,omitempty"` - RequestLimit int `json:"request_limit,omitempty"` - AccountID int64 `json:"account_id"` - Model string `json:"model"` - Transport string `json:"transport"` + Replay *CodexBorrowReplay `json:"replay,omitempty"` + Comparison string `json:"comparison,omitempty"` + Scenario string `json:"scenario,omitempty"` + Mode string `json:"mode,omitempty"` + ServiceTier string `json:"service_tier,omitempty"` + RequestLimit int `json:"request_limit,omitempty"` + AccountID int64 `json:"account_id"` + Model string `json:"model"` + Transport string `json:"transport"` } type CodexBorrowDiagnosticResult struct { + TurnState string `json:"turn_state,omitempty"` ProbeOnly bool `json:"probe_only,omitempty"` ProbeContext *CodexBorrowProbeContext `json:"probe_context,omitempty"` Scenario string `json:"scenario,omitempty"` @@ -117,8 +120,11 @@ type CodexBorrowDiagnosticEvent struct { // Diagnose runs fixed short questions through the normal account sender. The // internal observation purpose isolates health/billing; it is never user input. func (s *CodexGatewayBorrowService) Diagnose(ctx context.Context, request CodexBorrowDiagnosticRequest, emit func(CodexBorrowDiagnosticEvent)) error { - if request.Scenario != "" && request.Scenario != "codex_session" && request.Scenario != "probe_contract" { - return errors.New("scenario must be empty, codex_session or probe_contract") + if request.Scenario != "" && request.Scenario != "codex_session" && request.Scenario != "probe_contract" && request.Scenario != "codex_replay" { + return errors.New("scenario must be empty, codex_session, probe_contract or codex_replay") + } + if err := validateCodexBorrowReplay(request); err != nil { + return err } if request.Mode != "" && request.Mode != "ordinary" && request.Mode != "borrowed" { return errors.New("mode must be empty, ordinary or borrowed") @@ -146,11 +152,12 @@ func (s *CodexGatewayBorrowService) Diagnose(ctx context.Context, request CodexB return errors.New("request_limit must be between 1 and 8") } cfg := s.ConfigSnapshot() + sourceReplay := request.Scenario == "codex_replay" && request.Mode == "ordinary" && slices.Contains(cfg.SourceAccountIDs, request.AccountID) && slices.Contains(cfg.Models, request.Model) sourceProbe := request.Scenario == "probe_contract" && request.Model == codexGatewayBorrowSourceModel && slices.Contains(cfg.SourceAccountIDs, request.AccountID) if sourceProbe && (request.ServiceTier != "" || request.Comparison == "encoding") { return errors.New("source self-check uses the current plaintext Astra template without a service tier") } - if !cfg.Enabled || (!sourceProbe && (!slices.Contains(cfg.TargetAccountIDs, request.AccountID) || !slices.Contains(cfg.Models, request.Model))) { + if !cfg.Enabled || (!sourceProbe && !sourceReplay && (!slices.Contains(cfg.TargetAccountIDs, request.AccountID) || !slices.Contains(cfg.Models, request.Model))) { return errors.New("select a saved, enabled target and model") } account, err := s.accounts.GetByID(ctx, request.AccountID) @@ -189,7 +196,7 @@ func (s *CodexGatewayBorrowService) Diagnose(ctx context.Context, request CodexB } d := &codexBorrowDiagnostic{limit: int32(limit), onRequest: func(count int32) { emit(CodexBorrowDiagnosticEvent{Type: "request", Mode: mode, Requests: count, Limit: limit}) - }, requests: &requests, borrow: borrowed, transport: request.Transport, session: uuid.NewString(), scenario: request.Scenario, serviceTier: request.ServiceTier} + }, requests: &requests, borrow: borrowed, transport: request.Transport, session: uuid.NewString(), scenario: request.Scenario, serviceTier: request.ServiceTier, replay: request.Replay} operation := context.WithValue(ctx, codexBorrowDiagnosticContextKey{}, d) emit(CodexBorrowDiagnosticEvent{Type: "phase", Mode: mode, Requests: requests.Load(), Limit: limit}) previous := "" @@ -227,7 +234,11 @@ func (s *CodexGatewayBorrowService) Diagnose(ctx context.Context, request CodexB func (s *CodexGatewayBorrowService) diagnoseTurn(ctx context.Context, account *Account, model, previous string) CodexBorrowDiagnosticResult { start := time.Now() result := CodexBorrowDiagnosticResult{} - state := newPelicanExecutionState(ctx, sharedPelicanExecution, 90*time.Second, time.Now) + budget := 90 * time.Second + if diagnostic := borrowDiagnosticFromContext(ctx); diagnostic != nil && diagnostic.replay != nil { + budget = 5 * time.Minute + } + state := newPelicanExecutionState(ctx, sharedPelicanExecution, budget, time.Now) ctx = context.WithValue(state.parent, pelicanExecutionContextKey{}, state) defer state.finish() BeginPelicanPreparation(ctx) @@ -259,6 +270,11 @@ func (s *CodexGatewayBorrowService) diagnoseTurn(ctx context.Context, account *A d.prepareHTTPSessionTurn(c.Request.Header, payload) } body, _ := json.Marshal(payload) + if d.replay != nil { + body = append([]byte(nil), d.replay.Body...) + c.Request.Header.Del("session_id") + d.replay.applyHeaders(c.Request.Header) + } forward, sendErr := s.gateway.Forward(ctx, c, account, body) result.Scenario, result.Verification = d.scenario, d.verification.Load() if sendErr != nil { @@ -289,6 +305,23 @@ func (s *CodexGatewayBorrowService) diagnoseTurn(ctx context.Context, account *A result.ReadError = last.err.Error() } result.RawResponse = last.raw.String() + if d.scenario == "codex_replay" && last.dispatched { + state.mu.Lock() + result.Applied = state.invocation.BorrowApplied + state.mu.Unlock() + result.StateLength = len(extractOpenAICodexTurnState(w.Header())) + response, terminalErr := codexBorrowSessionTerminal(result.RawResponse) + result.Completed = sendErr == nil && terminalErr == nil && last.status >= 200 && last.status < 300 + result.ResponseID, result.ReportedModel = response.Get("id").String(), response.Get("model").String() + if sendErr != nil { + result.Error = sendErr.Error() + } else if terminalErr != nil { + result.Error = terminalErr.Error() + } + // The replay controller needs the actual business state, never probe state. + result.TurnState = extractOpenAICodexTurnState(w.Header()) + return result + } if d.scenario == "codex_session" && last.dispatched { state.mu.Lock() result.Applied = state.invocation.BorrowApplied diff --git a/backend/internal/service/codex_gateway_borrow_probe.go b/backend/internal/service/codex_gateway_borrow_probe.go index fb264449240b..910a58572d5f 100644 --- a/backend/internal/service/codex_gateway_borrow_probe.go +++ b/backend/internal/service/codex_gateway_borrow_probe.go @@ -29,6 +29,8 @@ type codexBorrowProbePinnedRouteKey struct{} type codexBorrowSourceObservationKey struct{} type codexGatewayBorrowObservation struct { + transportPhase string + transportAttempts int routeChanged bool requestEncoding string status int @@ -255,6 +257,7 @@ func (s *CodexGatewayBorrowService) probeTarget(ctx context.Context, template *h return s.fireObservation(shotCtx, template.Header, account, model, "Reply with OK.", "", proxy, profile, state, cookies, HTTPUpstreamProfileCodexBorrowTarget, codexGatewayBorrowProbeMaxBody) } mint, err := fire("", cookies) + result.MintTransportPhase, result.MintTransportAttempts = mint.transportPhase, mint.transportAttempts result.MintStatus, result.ReportedModel = mint.status, mint.model result.MintRequestEncoding = mint.requestEncoding result.MintCompleted, result.MintStateLength = mint.complete, len(mint.state) @@ -283,6 +286,7 @@ func (s *CodexGatewayBorrowService) probeTarget(ctx context.Context, template *h } } continued, err := fire(mint.state, cookies) + result.ContinueTransportPhase, result.ContinueTransportAttempts = continued.transportPhase, continued.transportAttempts result.ContinueStatus = continued.status result.ContinueRequestEncoding = continued.requestEncoding result.ContinueCompleted, result.ContinueStateLength = continued.complete, len(continued.state) @@ -448,7 +452,8 @@ func (s *CodexGatewayBorrowService) sendObservation(req *http.Request, account * if shot.requestEncoding == "" { shot.requestEncoding = "identity" } - resp, err := s.upstream.DoWithTLS(wire, proxy, account.ID, account.Concurrency, profile) + resp, phase, attempts, err := s.sendBorrowObservationHTTP(wire, proxy, account, profile) + shot.transportPhase, shot.transportAttempts = phase, attempts if err != nil { if resp != nil && resp.Body != nil { _ = resp.Body.Close() diff --git a/backend/internal/service/codex_gateway_borrow_ranxi_generated_test.go b/backend/internal/service/codex_gateway_borrow_ranxi_generated_test.go new file mode 100644 index 000000000000..c2b1fd191b15 --- /dev/null +++ b/backend/internal/service/codex_gateway_borrow_ranxi_generated_test.go @@ -0,0 +1,644 @@ +// Code generated by .github/scripts/generate_ranxi_borrow_reference.py; DO NOT EDIT. +// ranxi2001/sub2api v2.10.5, LGPL-3.0, commit d977e7a5e24e5536db6ce8b8478743e0c4b11f77. +package service + +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "github.com/Wei-Shaw/sub2api/internal/pkg/openai" + "github.com/Wei-Shaw/sub2api/internal/pkg/tlsfingerprint" + "github.com/gin-gonic/gin" + "github.com/google/uuid" + "io" + "net/http" + "strings" + "time" +) + +type ranxiRefCodexGatewayRoute struct { + cookie http.Cookie + expires time.Time +} + +func (s *OpenAIGatewayService) ranxiRefGuardOpenAICodexTurnStateEcho(c *gin.Context, account *Account, h http.Header) { + if s == nil || h == nil || account == nil { + return + } + if strings.TrimSpace(h.Get(openAICodexTurnStateHeader)) == "" { + return + } + seed := openAICodexTurnStateSeed(c) + if seed == "" { + return + } + raw, ok := s.openaiCodexTurnStateOrigins.Load(seed) + if !ok { + return + } + origin, ok := raw.(openAICodexTurnStateOrigin) + if !ok { + s.openaiCodexTurnStateOrigins.Delete(seed) + return + } + if !origin.expiresAt.IsZero() && time.Now().After(origin.expiresAt) { + s.openaiCodexTurnStateOrigins.Delete(seed) + return + } + if origin.accountID != account.ID { + h.Del(openAICodexTurnStateHeader) + } +} + +func ranxiRefRunOpenAICodexStateProbe(ctx context.Context, result *ranxiRefOpenAICodexStateProbeResult, seed string, fire func(string, string) (ranxiRefOpenAICodexStateShot, error)) { + mint, err := fire("", seed) + result.MintStatus = mint.status + result.TicketLength = len(mint.state) + if !result.ranxiRefShotUsable(ctx, "打票", mint, err) { + return + } + if mint.state == "" { + result.ranxiRefFail(ranxiRefOpenAICodexStateFailureNoTicket, "打票成功但未返回门票头（x-codex-turn-state），无法进行续接判据", "") + return + } + result.Minted = true + cookies := strings.Join(mint.cookies, "; ") + if seed != "" { + // A borrowed route remains pinned when mint only returns __cflb. + h := http.Header{"Cookie": []string{cookies}} + request := &http.Request{Header: http.Header{"Cookie": []string{seed}}} + if pinned, err := request.Cookie("__oailb"); err == nil { + ranxiRefReplaceCodexWSAnchorCookie(h, pinned.Value) + } + cookies = h.Get("Cookie") + } + cont, err := fire(mint.state, cookies) + result.ContinueStatus = cont.status + result.ContinueTicketLength = len(cont.state) + if !result.ranxiRefShotUsable(ctx, "门票续接", cont, err) { + return + } + result.NewTicket = cont.state != "" && cont.state != mint.state + result.ReportedModel = cont.model + if result.ReportedModel == "" { + result.ReportedModel = mint.model + } + if result.NewTicket { + result.Verdict = ranxiRefOpenAICodexStateDegraded + } else { + result.Verdict = ranxiRefOpenAICodexStateHealthy + } +} + +func ranxiRefProbeOpenAICodexStateRoute(ctx context.Context, upstream HTTPUpstream, template *http.Request, proxy string, accountID int64, concurrency int, profile *tlsfingerprint.Profile) *ranxiRefOpenAICodexStateProbeResult { + result := &ranxiRefOpenAICodexStateProbeResult{AccountID: accountID, Model: ranxiRefOpenAICodexStateProbeDefaultModel, Verdict: ranxiRefOpenAICodexStateInconclusive, StartedAt: time.Now()} + defer func() { + result.FinishedAt = time.Now() + result.LatencyMs = result.FinishedAt.Sub(result.StartedAt).Milliseconds() + if result.Reason == "" { + result.Reason = ranxiRefOpenAICodexStateVerdictReason(result.Verdict) + } + }() + if upstream == nil || template == nil { + result.ranxiRefFail(ranxiRefOpenAICodexStateFailureUnsupported, "缺少网关探针请求", "") + return result + } + pinned, err := template.Cookie("__oailb") + if err != nil || pinned.Value == "" { + result.ranxiRefFail("route_changed", "缺少借用路由", "") + return result + } + routeChanged := false + ranxiRefRunOpenAICodexStateProbe(ctx, result, pinned.String(), func(state, cookie string) (ranxiRefOpenAICodexStateShot, error) { + return ranxiRefFireOpenAICodexStateShotRequest(ctx, template.Header, result.Model, state, cookie, func(req *http.Request) (*http.Response, error) { + req = req.WithContext(WithHTTPUpstreamRedirectsDisabled(req.Context())) + resp, err := upstream.DoWithTLS(req, proxy, accountID, concurrency, profile) + if resp != nil { + for _, got := range resp.Cookies() { + if got.Name == "__oailb" && (got.Value != pinned.Value || got.MaxAge < 0 || (!got.Expires.IsZero() && !time.Now().Before(got.Expires))) { + routeChanged = true + } + } + } + if routeChanged { + if resp != nil && resp.Body != nil { + _ = resp.Body.Close() + } + return nil, errors.New("target_route_changed") + } + return resp, err + }) + }) + if routeChanged { + result.ranxiRefFail("route_changed", "探针期间网关路由已改变", "") + } + return result +} + +func (r *ranxiRefOpenAICodexStateProbeResult) ranxiRefFail(kind, reason, detail string) { + r.Verdict = ranxiRefOpenAICodexStateInconclusive + r.Failure = kind + r.Reason = reason + r.Detail = detail +} + +func (r *ranxiRefOpenAICodexStateProbeResult) ranxiRefShotUsable(ctx context.Context, step string, shot ranxiRefOpenAICodexStateShot, err error) bool { + if err != nil { + if ctx.Err() != nil { + r.ranxiRefFail(ranxiRefOpenAICodexStateFailureCancelled, step+"请求被取消或超时", "") + return false + } + if errors.Is(err, context.DeadlineExceeded) { + r.ranxiRefFail(ranxiRefOpenAICodexStateFailureNetworkError, fmt.Sprintf("%s请求超时（%d 秒内没有完成）", step, int(ranxiRefOpenAICodexStateProbeShotTimeout/time.Second)), "") + return false + } + r.ranxiRefFail(ranxiRefOpenAICodexStateFailureNetworkError, step+"请求失败（网络或代理异常）", SanitizeUpstreamErrorMessage(truncateString(err.Error(), ranxiRefOpenAICodexStateProbeMaxDetailByte))) + return false + } + switch { + case shot.status == http.StatusOK: + if shot.streamErr != nil { + r.ranxiRefFail(ranxiRefOpenAICodexStateFailureStreamError, step+"返回 200 但回复流没有正常结束（上游过载或额度用尽等），本次无法判断", shot.detail) + return false + } + return true + case shot.status == http.StatusUnauthorized || shot.status == http.StatusForbidden: + r.ranxiRefFail(ranxiRefOpenAICodexStateFailureAccountError, fmt.Sprintf("%s被上游拒绝（HTTP %d），账号令牌可能失效或被封", step, shot.status), shot.detail) + case shot.status == http.StatusTooManyRequests: + r.ranxiRefFail(ranxiRefOpenAICodexStateFailureRateLimited, step+"被上游限流（HTTP 429），请稍后再试", shot.detail) + case shot.status == http.StatusBadRequest && strings.Contains(strings.ToLower(shot.detail), "not supported"): + r.ranxiRefFail(ranxiRefOpenAICodexStateFailureModelUnsupported, "该账号的套餐不支持这个模型（HTTP 400），请换一个模型再测", shot.detail) + default: + r.ranxiRefFail(ranxiRefOpenAICodexStateFailureUpstreamError, fmt.Sprintf("%s返回异常状态（HTTP %d）", step, shot.status), shot.detail) + } + return false +} + +func ranxiRefOpenAICodexStateVerdictReason(verdict ranxiRefOpenAICodexStateVerdict) string { + switch verdict { + case ranxiRefOpenAICodexStateDegraded: + return "带原门票续接时上游回了新的门票（x-codex-turn-state），符合降智特征：账号很可能已被降到 gpt-5.6-luna 档，建议尽快换号。" + case ranxiRefOpenAICodexStateHealthy: + return "带原门票续接时上游没有回新门票，门票延续正常：账号仍处于满血（gpt-6-astra）状态。" + default: + return "未取得有效的门票续接探测结果，无法判断账号是否降智。" + } +} + +func ranxiRefFireOpenAICodexStateShotRequest(ctx context.Context, headers http.Header, model, turnState, cookie string, send func(*http.Request) (*http.Response, error)) (ranxiRefOpenAICodexStateShot, error) { + var out ranxiRefOpenAICodexStateShot + ctx = context.WithValue(ctx, ranxiRefOpenAICodexStateProbeContextKey{}, true) + shotCtx, cancel := context.WithTimeout(ctx, ranxiRefOpenAICodexStateProbeShotTimeout) + defer cancel() + body := []byte(`{"model":` + ranxiRefJsonString(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"]}`) + req, err := http.NewRequestWithContext(shotCtx, http.MethodPost, chatgptCodexURL, bytes.NewReader(body)) + if err != nil { + return out, err + } + req = req.WithContext(WithHTTPUpstreamProfile(req.Context(), HTTPUpstreamProfileOpenAIHarvest)) + req.Close = true + req.Host = "chatgpt.com" + req.Header = headers.Clone() + req.Header.Set("Accept", "text/event-stream") + req.Header.Set("Content-Type", "application/json") + req.Header.Set("OpenAI-Beta", "responses=experimental") + req.Header.Set("session_id", uuid.NewString()) + // Never carry a business turn, request size, or compressed/lite body into a probe. + for _, name := range []string{responsesLiteHeaderKey, openAICodexTurnStateHeader, "Cookie", "Content-Length", "Content-Encoding", "conversation_id", "x-codex-turn-metadata"} { + req.Header.Del(name) + } + if turnState = strings.TrimSpace(turnState); turnState != "" { + req.Header.Set(openAICodexTurnStateHeader, turnState) + } + if cookie = strings.TrimSpace(cookie); cookie != "" { + req.Header.Set("Cookie", cookie) + } + + resp, err := send(req) + if err != nil { + if resp != nil && resp.Body != nil { + _ = resp.Body.Close() + } + return out, err + } + if resp == nil { + return out, errors.New("nil upstream response") + } + out.status = resp.StatusCode + out.state = strings.TrimSpace(extractOpenAICodexTurnState(resp.Header)) + out.cookies = ranxiRefOpenAICodexStateRouteCookies(resp) + if resp.Body == nil { + out.streamErr = errors.New("probe response body missing") + return out, nil + } + defer func() { _ = resp.Body.Close() }() + data, err := io.ReadAll(io.LimitReader(resp.Body, ranxiRefOpenAICodexStateProbeMaxBody+1)) + if err != nil { + if shotCtx.Err() != nil { + return out, shotCtx.Err() + } + out.streamErr = errors.New("probe response incomplete") + return out, nil + } + if len(data) > ranxiRefOpenAICodexStateProbeMaxBody { + out.streamErr = errors.New("probe response too large") + return out, nil + } + if out.status != http.StatusOK { + out.detail = ranxiRefOpenAICodexStateDetail(data) + return out, nil + } + out.model = ranxiRefOpenAICodexStateStreamModel(data) + if out.streamErr = ranxiRefValidateCodexProbeResponse(data); out.streamErr != nil { + out.detail = ranxiRefOpenAICodexStateDetail(ranxiRefOpenAICodexStateStreamErrorPayload(data)) + } + return out, nil +} + +func ranxiRefOpenAICodexStateRouteCookies(resp *http.Response) []string { + var out []string + seen := map[string]bool{} + for _, cookie := range resp.Cookies() { + if cookie.Name != "__cflb" && cookie.Name != "__oailb" { + continue + } + if seen[cookie.Name] || cookie.Value == "" || cookie.MaxAge < 0 { + continue + } + seen[cookie.Name] = true + out = append(out, cookie.Name+"="+cookie.Value) + } + return out +} + +func ranxiRefOpenAICodexStateDetail(body []byte) string { + msg := strings.TrimSpace(extractUpstreamErrorMessage(body)) + if msg == "" { + msg = strings.TrimSpace(string(body)) + } + return SanitizeUpstreamErrorMessage(truncateString(msg, ranxiRefOpenAICodexStateProbeMaxDetailByte)) +} + +func ranxiRefOpenAICodexStateStreamEvents(data []byte) [][]byte { + var events [][]byte + scanner := bufio.NewScanner(bytes.NewReader(data)) + scanner.Buffer(make([]byte, 4096), ranxiRefOpenAICodexStateProbeMaxBody) + var lines []string + flush := func() { + if len(lines) == 0 { + return + } + payload := strings.TrimSpace(strings.Join(lines, "\n")) + lines = nil + if payload != "" && payload != "[DONE]" { + events = append(events, []byte(payload)) + } + } + for scanner.Scan() { + line := strings.TrimSuffix(scanner.Text(), "\r") + if line == "" { + flush() + continue + } + if strings.HasPrefix(line, "data:") { + lines = append(lines, strings.TrimPrefix(strings.TrimPrefix(line, "data:"), " ")) + } + } + flush() + return events +} + +func ranxiRefOpenAICodexStateStreamModel(data []byte) string { + completed := "" + for _, payload := range ranxiRefOpenAICodexStateStreamEvents(data) { + var event struct { + Type string `json:"type"` + Response struct { + Model string `json:"model"` + } `json:"response"` + } + if json.Unmarshal(payload, &event) != nil { + continue + } + model := strings.TrimSpace(event.Response.Model) + if model == "" { + continue + } + switch event.Type { + case "response.created": + return model + case "response.completed": + completed = model + } + } + return completed +} + +func ranxiRefOpenAICodexStateStreamErrorPayload(data []byte) []byte { + for _, payload := range ranxiRefOpenAICodexStateStreamEvents(data) { + var event struct { + Type string `json:"type"` + } + if json.Unmarshal(payload, &event) != nil { + continue + } + switch event.Type { + case "error", "response.failed", "response.incomplete": + var wrapped struct { + Response struct { + Error json.RawMessage `json:"error"` + } `json:"response"` + } + if json.Unmarshal(payload, &wrapped) == nil && len(wrapped.Response.Error) > 0 && string(wrapped.Response.Error) != "null" { + return []byte(`{"error":` + string(wrapped.Response.Error) + `}`) + } + return payload + } + } + return nil +} + +type ranxiRefOpenAICodexStateVerdict string + +type ranxiRefOpenAICodexStateProbeResult struct { + AccountID int64 `json:"account_id"` + Model string `json:"model"` + Verdict ranxiRefOpenAICodexStateVerdict `json:"verdict"` + Reason string `json:"reason"` + Failure string `json:"failure,omitempty"` + // Detail 是脱敏截断后的上游报错原文，便于看出「模型不支持」之类的具体原因。 + Detail string `json:"detail,omitempty"` + + MintStatus int `json:"mint_status"` + ContinueStatus int `json:"continue_status"` + Minted bool `json:"minted"` + NewTicket bool `json:"new_ticket"` + // TicketLength 是第一发打到的门票长度；判据只在 780 长度的完整请求上验证过。 + TicketLength int `json:"ticket_length"` + ContinueTicketLength int `json:"continue_ticket_length"` + ReportedModel string `json:"reported_model,omitempty"` + + LatencyMs int64 `json:"latency_ms"` + StartedAt time.Time `json:"started_at"` + FinishedAt time.Time `json:"finished_at"` +} + +type ranxiRefOpenAICodexStateShot struct { + status int + state string + cookies []string + model string + // streamErr 非空表示 200 但回复流没有完整跑完（上游过载、额度用尽等流内失败）。 + streamErr error + detail string +} + +type ranxiRefOpenAICodexStateProbeContextKey struct{} + +const ( + // ranxiRefOpenAICodexStateHealthy：带原门票续接时上游没回新门票，账号仍是满血（astra）。 + ranxiRefOpenAICodexStateHealthy ranxiRefOpenAICodexStateVerdict = "healthy" + // ranxiRefOpenAICodexStateDegraded：带原门票续接时上游回了新门票，账号大概率已被降到 luna 档。 + ranxiRefOpenAICodexStateDegraded ranxiRefOpenAICodexStateVerdict = "degraded" + // ranxiRefOpenAICodexStateInconclusive：两发没有都干净跑通，本次无法判断。 + ranxiRefOpenAICodexStateInconclusive ranxiRefOpenAICodexStateVerdict = "inconclusive" +) + +const ( + ranxiRefOpenAICodexStateFailureUnsupported = "unsupported" + ranxiRefOpenAICodexStateFailureAccountError = "account_error" + ranxiRefOpenAICodexStateFailureRateLimited = "rate_limited" + ranxiRefOpenAICodexStateFailureModelUnsupported = "model_unsupported" + ranxiRefOpenAICodexStateFailureUpstreamError = "upstream_error" + ranxiRefOpenAICodexStateFailureNetworkError = "network_error" + ranxiRefOpenAICodexStateFailureStreamError = "stream_error" + ranxiRefOpenAICodexStateFailureNoTicket = "no_ticket" + ranxiRefOpenAICodexStateFailureCancelled = "cancelled" +) + +const ( + ranxiRefOpenAICodexStateProbeDefaultModel = "gpt-6-astra" + + ranxiRefOpenAICodexStateProbeShotTimeout = 45 * time.Second + ranxiRefOpenAICodexStateProbeMaxBody = 1 << 20 + ranxiRefOpenAICodexStateProbeMaxDetailByte = 300 +) + +func ranxiRefValidateCodexProbeResponse(body []byte, expectedModel ...string) error { + completed := false + inspect := func(data []byte) error { + if bytes.Equal(bytes.TrimSpace(data), []byte("[DONE]")) { + return nil + } + var event struct { + Model string `json:"model"` + Type string `json:"type"` + Status string `json:"status"` + Error json.RawMessage `json:"error"` + Response *struct { + Model string `json:"model"` + Status string `json:"status"` + } `json:"response"` + } + if json.Unmarshal(data, &event) != nil { + return errors.New("invalid probe event") + } + if len(expectedModel) > 0 { + models := []string{event.Model} + if event.Response != nil { + models = append(models, event.Response.Model) + } + for _, served := range models { + if served != "" && ranxiRefNormalizeOpenAICodexTicketModel(served) != ranxiRefNormalizeOpenAICodexTicketModel(expectedModel[0]) { + return errors.New("probe response model mismatch") + } + } + } + if event.Type == "error" || event.Type == "response.failed" || event.Type == "response.incomplete" || event.Status == "failed" || event.Status == "incomplete" || (len(event.Error) > 0 && string(event.Error) != "null") { + return errors.New("probe response failed") + } + if event.Response != nil && (event.Response.Status == "failed" || event.Response.Status == "incomplete") { + return errors.New("probe response failed") + } + if event.Type == "response.completed" { + if event.Response == nil || event.Response.Status != "completed" { + return errors.New("invalid probe completion") + } + completed = true + } + if event.Type == "" && event.Status == "completed" { + completed = true + } + return nil + } + trimmed := bytes.TrimSpace(body) + if len(trimmed) > 0 && trimmed[0] == '{' { + if err := inspect(trimmed); err != nil { + return err + } + } else { + scanner := bufio.NewScanner(bytes.NewReader(body)) + scanner.Buffer(make([]byte, 4096), 1<<20) + var data []string + flush := func() error { + if len(data) == 0 { + return nil + } + err := inspect([]byte(strings.Join(data, "\n"))) + data = nil + return err + } + for scanner.Scan() { + line := strings.TrimSuffix(scanner.Text(), "\r") + if line == "" { + if err := flush(); err != nil { + return err + } + continue + } + if strings.HasPrefix(line, "data:") { + data = append(data, strings.TrimPrefix(strings.TrimPrefix(line, "data:"), " ")) + } + } + if scanner.Err() != nil { + return errors.New("invalid probe stream") + } + // A trailing unterminated event may be truncated, even if its JSON parses. + if len(data) > 0 { + return errors.New("unterminated probe event") + } + } + if !completed { + return errors.New("probe did not complete successfully") + } + return nil +} + +func ranxiRefReplaceCodexWSAnchorCookie(h http.Header, value string) { + var keep []string + for _, raw := range h.Values("Cookie") { + for _, part := range strings.Split(raw, ";") { + part = strings.TrimSpace(part) + name, _, _ := strings.Cut(part, "=") + if part != "" && strings.TrimSpace(name) != "__oailb" { + keep = append(keep, part) + } + } + } + keep = append(keep, (&http.Cookie{Name: "__oailb", Value: value}).String()) + h.Set("Cookie", strings.Join(keep, "; ")) +} + +func ranxiRefJsonString(v string) string { + b, err := json.Marshal(v) + if err != nil { + return `""` + } + return string(b) +} + +func ranxiRefNormalizeOpenAICodexTicketModel(model string) string { + return strings.TrimSpace(model) +} + +func ranxiRefCodexGatewayCookiePathMatches(path, scope string) bool { + return path == scope || (strings.HasPrefix(path, scope) && (strings.HasSuffix(scope, "/") || strings.HasPrefix(strings.TrimPrefix(path, scope), "/"))) +} + +func ranxiRefCodexGatewayRouteFromResponse(resp *http.Response, path string, now time.Time, configuredTTL ...int) *ranxiRefCodexGatewayRoute { + ttl := 230 + if len(configuredTTL) > 0 && configuredTTL[0] >= 30 && configuredTTL[0] <= 240 { + ttl = configuredTTL[0] + } + var route *ranxiRefCodexGatewayRoute + for _, cookie := range resp.Cookies() { + if cookie.Name != "__oailb" || !cookie.Secure || (cookie.Domain != "" && strings.TrimPrefix(cookie.Domain, ".") != "chatgpt.com") { + continue + } + if cookie.Path == "" { + cookie.Path = path[:strings.LastIndex(path, "/")] + } + if !strings.HasPrefix(cookie.Path, "/") || !ranxiRefCodexGatewayCookiePathMatches(path, cookie.Path) { + continue + } + // A deletion sentinel is applied even if the successful response has no new route. + if cookie.MaxAge < 0 || cookie.Value == "" || (cookie.MaxAge == 0 && !cookie.Expires.IsZero() && !now.Before(cookie.Expires)) { + return &ranxiRefCodexGatewayRoute{} + } + expires := now.Add(time.Duration(ttl) * time.Second) + if cookie.MaxAge > 0 && cookie.MaxAge < ttl { + expires = now.Add(time.Duration(cookie.MaxAge) * time.Second) + } else if cookie.MaxAge == 0 { + if cookie.Expires.IsZero() { + continue + } + if cookie.Expires.Before(expires) { + expires = cookie.Expires + } + } + route = &ranxiRefCodexGatewayRoute{cookie: *cookie, expires: expires} + } + return route +} + +func ranxiRefCreateOpenAITestPayload(modelID string, isOAuth bool) map[string]any { + payload := map[string]any{ + "model": modelID, + "input": []map[string]any{ + { + "role": "user", + "content": []map[string]any{ + { + "type": "input_text", + "text": "hi", + }, + }, + }, + }, + "stream": true, + } + + // OAuth accounts using ChatGPT internal API require store: false + if isOAuth { + payload["store"] = false + } + + // All accounts require instructions for Responses API + payload["instructions"] = openai.DefaultInstructions + + return payload +} + +func ranxiRefCreatePelicanOpenAIPayload(modelID string, isOAuth bool, prompt, reasoningEffort string) map[string]any { + payload := ranxiRefCreateOpenAITestPayload(modelID, isOAuth) + input, ok := payload["input"].([]map[string]any) + if ok && len(input) > 0 { + content, ok := input[0]["content"].([]map[string]any) + if ok && len(content) > 0 { + content[0]["text"] = ranxiRefPromptOrDefault(prompt) + } + } + if effort := ranxiRefNormalizePelicanReasoningEffort(reasoningEffort); effort != "" { + payload["reasoning"] = map[string]any{"effort": effort} + } + return payload +} + +func ranxiRefPromptOrDefault(prompt string) string { + if value := strings.TrimSpace(prompt); value != "" { + return value + } + return "hi" +} + +func ranxiRefNormalizePelicanReasoningEffort(value string) string { + switch strings.ToLower(strings.TrimSpace(value)) { + case "none", "minimal", "low", "medium", "high", "xhigh", "max", "ultra": + return strings.ToLower(strings.TrimSpace(value)) + default: + return "" + } +} diff --git a/backend/internal/service/codex_gateway_borrow_ranxi_reference_test.go b/backend/internal/service/codex_gateway_borrow_ranxi_reference_test.go new file mode 100644 index 000000000000..18846e6b24c4 --- /dev/null +++ b/backend/internal/service/codex_gateway_borrow_ranxi_reference_test.go @@ -0,0 +1,141 @@ +package service + +import ( + "context" + "encoding/json" + "io" + "net/http" + "strings" + "testing" + "time" + + "github.com/Wei-Shaw/sub2api/internal/pkg/tlsfingerprint" + "github.com/google/uuid" + "github.com/stretchr/testify/require" +) + +// Both implementations consume identical responses. The reference functions +// are extracted from the pinned Git commit, not restated as expected results. +func TestCodexGatewayBorrowRanxiReference(t *testing.T) { + t.Run("business STATE provenance", func(t *testing.T) { + for _, tc := range []struct { + name string + owner int64 + ttl time.Duration + }{ + {"unknown", 0, 0}, {"same account", 2, time.Minute}, {"foreign account", 1, time.Minute}, {"expired foreign record", 1, -time.Minute}, + } { + t.Run(tc.name, func(t *testing.T) { + gateway := &OpenAIGatewayService{} + c, _ := newTurnStateTestContext(t, 7, "native-session") + if tc.owner != 0 { + gateway.openaiCodexTurnStateOrigins.Store(openAICodexTurnStateSeed(c), openAICodexTurnStateOrigin{accountID: tc.owner, expiresAt: time.Now().Add(tc.ttl)}) + } + h := http.Header{"X-Codex-Turn-State": {"business-state"}} + original := h.Clone() + gateway.ranxiRefGuardOpenAICodexTurnStateEcho(c, &Account{ID: 2}, original) + gateway.guardOpenAICodexTurnStateEcho(c, &Account{ID: 2}, h) + require.Equal(t, original, h) + }) + } + }) + t.Run("source payload", func(t *testing.T) { + req, err := borrowSourceObservationRequest(context.Background(), http.Header{}) + require.NoError(t, err) + original := ranxiRefCreatePelicanOpenAIPayload("gpt-6-astra", true, "Reply with OK only.", "medium") + want, err := json.Marshal(original) + require.NoError(t, err) + require.Equal(t, string(want), string(borrowCoreBody(t, req))) + }) + for _, tc := range []struct { + name, first, second, firstBody, secondBody, route string + status int + }{ + {name: "same STATE", first: "mint", second: "mint"}, + {name: "omitted second STATE", first: "mint"}, + {name: "new second STATE", first: "mint", second: "replacement"}, + {name: "missing first STATE"}, + {name: "route replaced", first: "mint", route: "__oailb=replaced; Path=/; Secure"}, + {name: "route deleted", first: "mint", route: "__oailb=; Max-Age=0; Path=/; Secure"}, + {name: "stream failed", first: "mint", secondBody: "data: {\"type\":\"response.failed\",\"response\":{\"status\":\"failed\"}}\n\n"}, + {name: "unterminated completion", first: "mint", secondBody: strings.TrimSuffix(borrowCoreSSE("gpt-6-astra", "OK"), "\n\n")}, + {name: "HTTP failure", first: "mint", status: 503}, + } { + t.Run(tc.name, func(t *testing.T) { + type sent struct { + Body string + Header http.Header + Close bool + Profile HTTPUpstreamProfile + } + runs := make([][]sent, 2) + sender := func(run int) borrowCoreProbe { + return func(req *http.Request, proxy string, id int64, n int, p *tlsfingerprint.Profile) (*http.Response, error) { + require.Equal(t, "synthetic-exit", proxy) + require.Equal(t, int64(2), id) + header := req.Header.Clone() + _, err := uuid.Parse(header.Get("session_id")) + require.NoError(t, err) + header.Del("session_id") // Each original shot deliberately generates a UUID. + runs[run] = append(runs[run], sent{string(borrowCoreBody(t, req)), header, req.Close, HTTPUpstreamProfileFromContext(req.Context())}) + state, body := tc.first, tc.firstBody + if len(runs[run]) == 2 { + state, body = tc.second, tc.secondBody + } + resp := borrowCoreResponse("gpt-6-astra", "OK", state, "__cflb=target-affinity; Path=/; Secure") + if body != "" { + resp.Body = io.NopCloser(strings.NewReader(body)) + } + if tc.route != "" { + resp.Header.Add("Set-Cookie", tc.route) + } + if tc.status != 0 { + resp.StatusCode = tc.status + } + return resp, nil + } + } + headers := http.Header{} + for k, v := range map[string]string{"Authorization": "Bearer target-only", "ChatGPT-Account-ID": "target-only", "User-Agent": "current-client", "Originator": "Codex Desktop", "Version": "0.162.0-alpha.17.2", "session-id": "native-session", "thread-id": "native-thread", "x-codex-window-id": "native-window", "X-Codex-Turn-State": "business-state", "X-Codex-Turn-Metadata": "business-metadata", "Cookie": "__oailb=borrowed; business=untouched", "Content-Encoding": "zstd"} { + headers.Set(k, v) + } + originalHeaders := headers.Clone() + req, _ := http.NewRequest(http.MethodPost, chatgptCodexURL, nil) + req.Header = headers + reference := ranxiRefProbeOpenAICodexStateRoute(context.Background(), sender(0), req, "synthetic-exit", 2, 4, nil) + service := newBorrowCoreTest(t, sender(1), borrowCoreAccount(1), borrowCoreAccount(2)) + candidate := &codexGatewayBorrowCandidate{sourceID: 1, cookie: http.Cookie{Name: "__oailb", Value: "borrowed"}, expires: time.Now().Add(time.Minute)} + local := service.probeTarget(context.Background(), req, borrowCoreAccount(2), "gpt-6-astra", "synthetic-exit", nil, candidate) + require.Equal(t, reference.Verdict == ranxiRefOpenAICodexStateHealthy, local.Success) + require.Equal(t, reference.NewTicket, local.NewTicket) + require.Equal(t, runs[0], runs[1], "both implementations must send the same two-shot protocol") + require.Equal(t, originalHeaders, req.Header, "probes must not overwrite business history or cookies") + }) + } + t.Run("cookie scope and lifetime", func(t *testing.T) { + now := time.Date(2026, 10, 11, 0, 0, 0, 0, time.UTC) + for _, cookies := range [][]string{ + {"__oailb=route; Secure; Path=/; Max-Age=600"}, + {"__oailb=route; Secure; Path=/; Max-Age=30"}, + {"__oailb=session; Secure; Path=/"}, + {"__oailb=route; Secure; Path=/other; Max-Age=60"}, + {"__oailb=route; Secure; Domain=other.example; Max-Age=60"}, + {"__oailb=route; Path=/; Max-Age=60"}, + {"__oailb=route; Secure; Path=/; Max-Age=60", "__oailb=; Secure; Path=/; Max-Age=0"}, + } { + h := http.Header{} + for _, c := range cookies { + h.Add("Set-Cookie", c) + } + response := &http.Response{Header: h} + ref := ranxiRefCodexGatewayRouteFromResponse(response, "/backend-api/codex/responses", now) + local := borrowCandidateFromCookies(response.Cookies(), "/backend-api/codex/responses", now) + usable := ref != nil && ref.cookie.Value != "" && now.Before(ref.expires) + require.Equal(t, usable, local != nil, cookies) + if usable { + require.Equal(t, ref.cookie, local.cookie) + require.Equal(t, ref.expires, local.expires) + } + } + }) +} diff --git a/backend/internal/service/codex_gateway_borrow_replay.go b/backend/internal/service/codex_gateway_borrow_replay.go new file mode 100644 index 000000000000..f31d405ea571 --- /dev/null +++ b/backend/internal/service/codex_gateway_borrow_replay.go @@ -0,0 +1,94 @@ +package service + +import ( + "bytes" + "encoding/json" + "errors" + "net/http" + "strings" + + "github.com/tidwall/gjson" +) + +// Administrator-only replay shares the real sender, account allowlist and +// observation isolation. The caller executes local fixture tools and supplies +// complete history. It cannot inject credentials, cookies, destinations, or +// anchors belonging to a real user conversation. +type CodexBorrowReplay struct { + Body json.RawMessage `json:"body"` + Headers map[string]string `json:"headers"` +} + +var codexBorrowReplayHeaders = map[string]bool{ + "user-agent": true, "originator": true, "version": true, "session-id": true, "thread-id": true, + "x-client-request-id": true, "x-codex-window-id": true, "x-codex-installation-id": true, + "x-codex-turn-metadata": true, "x-codex-turn-state": true, +} + +func validateCodexBorrowReplay(request CodexBorrowDiagnosticRequest) error { + if request.Scenario != "codex_replay" { + if request.Replay != nil { + return errors.New("replay requires codex_replay scenario") + } + return nil + } + if request.Transport != "http" || (request.Mode != "ordinary" && request.Mode != "borrowed") || request.Replay == nil { + return errors.New("codex_replay requires HTTP, explicit mode and replay") + } + raw := request.Replay.Body + if len(raw) == 0 || len(raw) > 1<<20 || !gjson.ValidBytes(raw) || !gjson.ParseBytes(raw).IsObject() { + return errors.New("replay body must be a JSON object up to 1 MiB") + } + body := gjson.ParseBytes(raw) + if body.Get("model").String() != request.Model || body.Get("stream").Type != gjson.True || body.Get("store").Type != gjson.False || body.Get("previous_response_id").Exists() { + return errors.New("replay requires the selected model, streaming, store=false and complete history") + } + session := body.Get("prompt_cache_key").String() + if !strings.HasPrefix(session, "codex-acceptance-") || len(session) > 128 { + return errors.New("replay requires an isolated codex-acceptance session") + } + if request.ServiceTier != "" && body.Get("service_tier").String() != request.ServiceTier { + return errors.New("replay service tier must match its body") + } + seen := map[string]bool{} + for name, value := range request.Replay.Headers { + lower := strings.ToLower(name) + if !codexBorrowReplayHeaders[lower] || seen[lower] || strings.ContainsAny(value, "\r\n") || len(value) > 16384 { + return errors.New("unsupported or ambiguous replay header") + } + seen[lower] = true + if (lower == "session-id" || lower == "thread-id") && value != session { + return errors.New("replay session headers must match its isolated session") + } + } + if !seen["session-id"] || !seen["thread-id"] { + return errors.New("replay requires isolated session and thread headers") + } + // Reject trailing values and duplicate top-level keys, so validation and the + // gateway's JSON reader cannot select different models or session identifiers. + decoder := json.NewDecoder(bytes.NewReader(raw)) + _, _ = decoder.Token() + keys := map[string]bool{} + for decoder.More() { + token, err := decoder.Token() + if err != nil { + return err + } + key, ok := token.(string) + if !ok || keys[key] { + return errors.New("duplicate replay body key") + } + keys[key] = true + var value json.RawMessage + if err = decoder.Decode(&value); err != nil { + return err + } + } + return nil +} + +func (r *CodexBorrowReplay) applyHeaders(headers http.Header) { + for name, value := range r.Headers { + headers.Set(name, value) + } +} diff --git a/backend/internal/service/codex_gateway_borrow_replay_test.go b/backend/internal/service/codex_gateway_borrow_replay_test.go new file mode 100644 index 000000000000..9aef71e3b3c1 --- /dev/null +++ b/backend/internal/service/codex_gateway_borrow_replay_test.go @@ -0,0 +1,98 @@ +//go:build unit + +package service + +import ( + "context" + "encoding/json" + "io" + "net/http" + "strings" + "testing" + + "github.com/Wei-Shaw/sub2api/internal/pkg/tlsfingerprint" + "github.com/stretchr/testify/require" + "github.com/tidwall/gjson" +) + +func borrowReplayFixture() CodexBorrowDiagnosticRequest { + return CodexBorrowDiagnosticRequest{Scenario: "codex_replay", Mode: "ordinary", Transport: "http", AccountID: 2, Model: "gpt-6-astra", ServiceTier: "default", Replay: &CodexBorrowReplay{ + Body: json.RawMessage(`{"model":"gpt-6-astra","stream":true,"store":false,"service_tier":"default","reasoning":{"effort":"xhigh"},"prompt_cache_key":"codex-acceptance-synthetic","input":[{"role":"user","content":[{"type":"input_text","text":"synthetic task"}]}],"tools":[{"type":"function","name":"read_file","parameters":{"type":"object","properties":{}}}]}`), + Headers: map[string]string{"session-id": "codex-acceptance-synthetic", "thread-id": "codex-acceptance-synthetic", "User-Agent": "Codex Desktop/0.162.0-alpha.17.2", "x-codex-turn-state": "previous-business-state"}, + }} +} + +func TestCodexGatewayBorrowReplayBoundaries(t *testing.T) { + require.NoError(t, validateCodexBorrowReplay(borrowReplayFixture())) + for _, tc := range []struct { + name string + edit func(*CodexBorrowDiagnosticRequest) + }{ + {"credentials", func(r *CodexBorrowDiagnosticRequest) { r.Replay.Headers["Authorization"] = "Bearer other" }}, + {"cookies", func(r *CodexBorrowDiagnosticRequest) { r.Replay.Headers["Cookie"] = "__oailb=other" }}, + {"ambiguous header", func(r *CodexBorrowDiagnosticRequest) { r.Replay.Headers["Session-Id"] = "another" }}, + {"real session", func(r *CodexBorrowDiagnosticRequest) { + r.Replay.Body = json.RawMessage(strings.ReplaceAll(string(r.Replay.Body), "codex-acceptance-synthetic", "real-user-session")) + }}, + {"wrong model", func(r *CodexBorrowDiagnosticRequest) { r.Model = "gpt-6.1-sol" }}, + {"different history anchor", func(r *CodexBorrowDiagnosticRequest) { + r.Replay.Body = append([]byte(`{"previous_response_id":"real-user-response",`), r.Replay.Body[1:]...) + }}, + {"duplicate body key", func(r *CodexBorrowDiagnosticRequest) { + r.Replay.Body = append([]byte(`{"model":"gpt-6-astra",`), r.Replay.Body[1:]...) + }}, + {"implicit mode", func(r *CodexBorrowDiagnosticRequest) { r.Mode = "" }}, + {"wrong scenario", func(r *CodexBorrowDiagnosticRequest) { r.Scenario = "short" }}, + {"oversize", func(r *CodexBorrowDiagnosticRequest) { + r.Replay.Body = json.RawMessage(strings.Repeat(" ", 1<<20) + string(r.Replay.Body)) + }}, + } { + t.Run(tc.name, func(t *testing.T) { + request := borrowReplayFixture() + tc.edit(&request) + require.Error(t, validateCodexBorrowReplay(request)) + }) + } +} + +func TestCodexGatewayBorrowReplayPreservesTasksAndReturnsBusinessState(t *testing.T) { + for _, id := range []int64{1, 2} { + account := borrowCoreAccount(id) + generator, repo := newPelicanGeneratorForTest(account, &pelicanGeneratorUpstream{}) + gateway := generator.openaiGatewayService + var calls int + gateway.httpUpstream = borrowCoreProbe(func(req *http.Request, _ string, accountID int64, n int, _ *tlsfingerprint.Profile) (*http.Response, error) { + calls++ + require.Equal(t, id, accountID) + body := borrowCoreBody(t, req) + require.Equal(t, "xhigh", gjson.GetBytes(body, "reasoning.effort").String()) + require.Equal(t, "default", gjson.GetBytes(body, "service_tier").String()) + require.Equal(t, "synthetic task", gjson.GetBytes(body, "input.0.content.0.text").String()) + require.Equal(t, "read_file", gjson.GetBytes(body, "tools.0.name").String()) + require.Equal(t, "previous-business-state", req.Header.Get("x-codex-turn-state")) + _, finish, err := PreparePelicanHTTPRequest(req, accountID, n, "http") + require.NoError(t, err) + raw := `data: {"type":"response.completed","response":{"id":"synthetic-response","model":"gpt-6-astra","status":"completed","output":[{"type":"function_call","name":"read_file","call_id":"test-call","arguments":"{}"}]}}` + "\n\n" + return finish(&http.Response{StatusCode: 200, Header: http.Header{"Content-Type": {"text/event-stream"}, "X-Codex-Turn-State": {"actual-business-state"}}, Body: io.NopCloser(strings.NewReader(raw))}, nil) + }) + s := NewCodexGatewayBorrowService(nil, repo, gateway, nil, nil) + s.publishConfig(CodexGatewayBorrowConfig{Enabled: true, SourceAccountIDs: []int64{1}, TargetAccountIDs: []int64{2}, Models: []string{"gpt-6-astra"}}, false) + request := borrowReplayFixture() + request.AccountID = id + var result *CodexBorrowDiagnosticResult + err := s.Diagnose(context.Background(), request, func(e CodexBorrowDiagnosticEvent) { + if e.Result != nil { + result = e.Result + } + }) + s.Stop() + require.NoError(t, err) + require.Equal(t, 1, calls) + require.NotNil(t, result) + require.True(t, result.Dispatched) + require.True(t, result.Completed, result.Error) + require.False(t, result.Applied) + require.Equal(t, "actual-business-state", result.TurnState) + require.Contains(t, result.RawResponse, "function_call") + } +} diff --git a/backend/internal/service/codex_gateway_borrow_transport.go b/backend/internal/service/codex_gateway_borrow_transport.go new file mode 100644 index 000000000000..1588ab08da90 --- /dev/null +++ b/backend/internal/service/codex_gateway_borrow_transport.go @@ -0,0 +1,91 @@ +package service + +import ( + "crypto/tls" + "fmt" + "net/http" + "net/http/httptrace" + "sync/atomic" + + "github.com/Wei-Shaw/sub2api/internal/pkg/tlsfingerprint" +) + +// Trace evidence, rather than error text, decides whether another connection is +// safe. An uninstrumented transport or any written HTTP header forbids retry. +// This wrapper is used only by isolated source/target observations. +type codexBorrowConnectionTrace struct { + wrote atomic.Bool + failedBeforeHTTP atomic.Bool + phase atomic.Int32 +} + +func (t *codexBorrowConnectionTrace) trace() *httptrace.ClientTrace { + return &httptrace.ClientTrace{ + ConnectStart: func(_, _ string) { t.phase.Store(1); t.failedBeforeHTTP.Store(false) }, + ConnectDone: func(_, _ string, err error) { t.failedBeforeHTTP.Store(err != nil) }, + TLSHandshakeStart: func() { t.phase.Store(2); t.failedBeforeHTTP.Store(false) }, + TLSHandshakeDone: func(_ tls.ConnectionState, err error) { t.failedBeforeHTTP.Store(err != nil) }, + GotConn: func(httptrace.GotConnInfo) { t.phase.Store(3); t.failedBeforeHTTP.Store(false) }, + WroteHeaders: func() { t.wrote.Store(true); t.phase.Store(4) }, + WroteRequest: func(httptrace.WroteRequestInfo) { t.wrote.Store(true); t.phase.Store(4) }, + GotFirstResponseByte: func() { t.wrote.Store(true); t.phase.Store(5) }, + } +} + +func (t *codexBorrowConnectionTrace) name() string { + switch t.phase.Load() { + case 1: + return "connect" + case 2: + return "tls_handshake" + case 3: + return "connected" + case 4: + return "request_write" + case 5: + return "response_headers" + default: + return "unobserved" + } +} + +type codexBorrowTransportError struct { + phase string + attempts int + cause error +} + +func (e *codexBorrowTransportError) Error() string { + return fmt.Sprintf("borrow transport %s after %d attempt(s): %s", e.phase, e.attempts, e.cause) +} +func (e *codexBorrowTransportError) Unwrap() error { return e.cause } + +func (s *CodexGatewayBorrowService) sendBorrowObservationHTTP(req *http.Request, proxy string, account *Account, profile *tlsfingerprint.Profile) (*http.Response, string, int, error) { + current := req + for attempt := 1; ; attempt++ { + trace := &codexBorrowConnectionTrace{} + current = current.WithContext(httptrace.WithClientTrace(req.Context(), trace.trace())) + resp, err := s.upstream.DoWithTLS(current, proxy, account.ID, account.Concurrency, profile) + if err == nil { + return resp, trace.name(), attempt, nil + } + if resp != nil && resp.Body != nil { + _ = resp.Body.Close() + } + retry := attempt == 1 && resp == nil && trace.failedBeforeHTTP.Load() && !trace.wrote.Load() && req.GetBody != nil && req.Context().Err() == nil + if !retry { + return nil, trace.name(), attempt, &codexBorrowTransportError{trace.name(), attempt, err} + } + body, bodyErr := req.GetBody() + if bodyErr != nil { + return nil, trace.name(), attempt, &codexBorrowTransportError{trace.name(), attempt, err} + } + // Preparation and diagnostic accounting includes every physical attempt. + if budgetErr := consumeBorrowDiagnosticRequest(req.Context()); budgetErr != nil { + _ = body.Close() + return nil, trace.name(), attempt, budgetErr + } + current = req.Clone(req.Context()) + current.Body = body + } +} diff --git a/backend/internal/service/codex_gateway_borrow_transport_test.go b/backend/internal/service/codex_gateway_borrow_transport_test.go new file mode 100644 index 000000000000..5db42a0ed4da --- /dev/null +++ b/backend/internal/service/codex_gateway_borrow_transport_test.go @@ -0,0 +1,75 @@ +package service + +import ( + "context" + "crypto/tls" + "errors" + "net/http" + "net/http/httptrace" + "strings" + "sync/atomic" + "testing" + + "github.com/Wei-Shaw/sub2api/internal/pkg/tlsfingerprint" + "github.com/stretchr/testify/require" +) + +func TestCodexGatewayBorrowTransportRetryRequiresPreHTTPProof(t *testing.T) { + for _, kind := range []string{"tls", "connect", "unobserved", "headers_written", "response_received", "budget_exhausted", "cancelled"} { + t.Run(kind, func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + var count atomic.Int32 + count.Store(1) + limit := int32(2) + if kind == "budget_exhausted" { + limit = 1 + } + ctx = context.WithValue(ctx, codexBorrowDiagnosticContextKey{}, &codexBorrowDiagnostic{requests: &count, limit: limit}) + calls := 0 + service := newBorrowCoreTest(t, func(req *http.Request, _ string, _ int64, _ int, _ *tlsfingerprint.Profile) (*http.Response, error) { + calls++ + require.Equal(t, "identical payload", string(borrowCoreBody(t, req))) + if calls == 2 { + return borrowCoreResponse("gpt-6-astra", "OK", ""), nil + } + trace := httptrace.ContextClientTrace(req.Context()) + switch kind { + case "unobserved": + case "connect": + trace.ConnectStart("tcp", "synthetic") + trace.ConnectDone("tcp", "synthetic", errors.New("connect failed")) + default: + trace.TLSHandshakeStart() + trace.TLSHandshakeDone(tls.ConnectionState{}, errors.New("handshake failed")) + if kind == "headers_written" { + trace.WroteHeaders() + } + if kind == "cancelled" { + cancel() + } + } + if kind == "response_received" { + return borrowCoreResponse("gpt-6-astra", "OK", ""), errors.New("late error") + } + return nil, errors.New("synthetic transport failure") + }) + req, _ := http.NewRequestWithContext(ctx, http.MethodPost, chatgptCodexURL, strings.NewReader("identical payload")) + resp, phase, attempts, err := service.sendBorrowObservationHTTP(req, "", borrowCoreAccount(2), nil) + if kind == "tls" || kind == "connect" { + require.NoError(t, err) + require.Equal(t, 2, calls) + require.Equal(t, 2, attempts) + require.Equal(t, int32(2), count.Load()) + _ = resp.Body.Close() + } else { + require.Error(t, err) + require.Equal(t, 1, calls) + require.Equal(t, 1, attempts) + } + if kind == "headers_written" { + require.Equal(t, "request_write", phase) + } + }) + } +} diff --git a/backend/internal/service/openai_codex_turn_state.go b/backend/internal/service/openai_codex_turn_state.go index 4614df3c47ca..232a9f8cbffa 100644 --- a/backend/internal/service/openai_codex_turn_state.go +++ b/backend/internal/service/openai_codex_turn_state.go @@ -157,9 +157,10 @@ func (s *OpenAIGatewayService) commitOpenAIWSSessionTurnState( s.noteOpenAICodexTurnStateProvenance(c, account) } -// guardOpenAICodexTurnStateEcho 出站守卫：客户端回带的 turn-state 只有在 -// 能证明由当前账号铸造时才保留。来源未知、过期、缺少会话标识或来自其他 -// 账号的值均剥离。只剥离、不注入—— +// guardOpenAICodexTurnStateEcho matches ranxi v2.10.5: strip a client's state +// only when a live provenance record identifies another account. Process-local +// record loss after restart or expiry is not evidence of a foreign state. +// 只剥离已知跨账号值、不注入—— // /responses 路径的客户端是真实 Codex，会按自身回合语义自行回带；服务端 // 注入是 Claude 兼容桥（无法回带的客户端）的专属行为。 func (s *OpenAIGatewayService) guardOpenAICodexTurnStateEcho(c *gin.Context, account *Account, h http.Header) { @@ -175,23 +176,19 @@ func (s *OpenAIGatewayService) guardOpenAICodexTurnStateEcho(c *gin.Context, acc } seed := openAICodexTurnStateSeed(c) if seed == "" { - h.Del(openAICodexTurnStateHeader) return } raw, ok := s.openaiCodexTurnStateOrigins.Load(seed) if !ok { - h.Del(openAICodexTurnStateHeader) return } origin, ok := raw.(openAICodexTurnStateOrigin) if !ok { s.openaiCodexTurnStateOrigins.Delete(seed) - h.Del(openAICodexTurnStateHeader) return } if !origin.expiresAt.IsZero() && time.Now().After(origin.expiresAt) { s.openaiCodexTurnStateOrigins.Delete(seed) - h.Del(openAICodexTurnStateHeader) return } if origin.accountID != account.ID { diff --git a/backend/internal/service/openai_codex_turn_state_test.go b/backend/internal/service/openai_codex_turn_state_test.go index 18f5bd0de7d5..69fc1563354d 100644 --- a/backend/internal/service/openai_codex_turn_state_test.go +++ b/backend/internal/service/openai_codex_turn_state_test.go @@ -193,15 +193,15 @@ func TestGuardOpenAICodexTurnStateEcho(t *testing.T) { require.Empty(t, h.Get("x-codex-turn-state")) }) - t.Run("no_provenance_strips_echo", func(t *testing.T) { + t.Run("no_provenance_preserves_echo", func(t *testing.T) { svc := &OpenAIGatewayService{} c, _ := newTurnStateTestContext(t, 7, "sess-g3") h := newOutbound("blob-unknown") svc.guardOpenAICodexTurnStateEcho(c, &Account{ID: 43}, h) - require.Empty(t, h.Get("x-codex-turn-state")) + require.Equal(t, "blob-unknown", h.Get("x-codex-turn-state")) }) - t.Run("expired_provenance_strips_echo_and_prunes", func(t *testing.T) { + t.Run("expired_provenance_preserves_echo_and_prunes", func(t *testing.T) { svc := &OpenAIGatewayService{} c, _ := newTurnStateTestContext(t, 7, "sess-g4") svc.openaiCodexTurnStateOrigins.Store("7\x00sess-g4", openAICodexTurnStateOrigin{ @@ -210,17 +210,17 @@ func TestGuardOpenAICodexTurnStateEcho(t *testing.T) { }) h := newOutbound("blob-A") svc.guardOpenAICodexTurnStateEcho(c, &Account{ID: 43}, h) - require.Empty(t, h.Get("x-codex-turn-state")) _, ok := svc.openaiCodexTurnStateOrigins.Load("7\x00sess-g4") + require.Equal(t, "blob-A", h.Get("x-codex-turn-state")) require.False(t, ok) }) - t.Run("no_session_seed_strips_echo", func(t *testing.T) { + t.Run("no_session_seed_preserves_echo", func(t *testing.T) { svc := &OpenAIGatewayService{} c, _ := newTurnStateTestContext(t, 7, "") h := newOutbound("blob-A") svc.guardOpenAICodexTurnStateEcho(c, &Account{ID: 43}, h) - require.Empty(t, h.Get("x-codex-turn-state")) + require.Equal(t, "blob-A", h.Get("x-codex-turn-state")) }) t.Run("missing_account_strips_echo", func(t *testing.T) { diff --git a/backend/internal/service/openai_compact_sse_keepalive.go b/backend/internal/service/openai_compact_sse_keepalive.go index c0aba9108b5f..6b07c5a1bac8 100644 --- a/backend/internal/service/openai_compact_sse_keepalive.go +++ b/backend/internal/service/openai_compact_sse_keepalive.go @@ -29,6 +29,9 @@ type openAICompactSSEKeepalive struct { writer gin.ResponseWriter started bool stopped bool + // Account-test writers pause while the caller owns the header map, then + // resume on Write/Flush. Compact writers retain their stop-on-write behavior. + paused bool // bytes 是心跳已写出的注释字节数。心跳不构成语义响应，handler 的 // "Forward 期间是否已写响应"判定（failover 放弃换号的依据）必须扣除 // 这部分字节，见 OpenAICompactKeepaliveAdjustedWrittenSize。 @@ -107,6 +110,9 @@ func (k *openAICompactSSEKeepalive) beat() bool { if k.stopped { return false } + if k.paused { + return true + } if !k.started { header := k.writer.Header() header.Set("Content-Type", "text/event-stream") From 0c2d565014a85eadda1b0478bbd476a4d4456c57 Mon Sep 17 00:00:00 2001 From: Codex Date: Sun, 11 Oct 2026 01:31:25 +0800 Subject: [PATCH 2/2] fix(codex): preserve WS provenance and committed test completions --- .downstream/native-domains.md | 5 +++-- .downstream/upstream-review-ranxi-v2.10.5.md | 8 +++++--- backend/internal/service/account_test_service.go | 5 ++++- .../internal/service/openai_codex_turn_state.go | 16 ++++++++++++++++ 4 files changed, 28 insertions(+), 6 deletions(-) diff --git a/.downstream/native-domains.md b/.downstream/native-domains.md index 9cd36fe75ad9..3ce5db23942f 100644 --- a/.downstream/native-domains.md +++ b/.downstream/native-domains.md @@ -111,9 +111,10 @@ headers, and generate a new legacy `session_id` per shot. The second shot carrie 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`. -Business STATE echo follows the original provenance rule: unknown or expired +Native HTTP business STATE echo follows the original provenance rule: unknown or expired process-local provenance preserves the caller's STATE; a live foreign-account -record strips it. This does not relax two-shot target qualification. +record strips it. This does not relax two-shot target qualification or the +existing WS/WS-to-HTTP bridge provenance requirements. Observations report their connect/TLS/write/header phase and attempt count. They may retry once only with transport trace proving a connection failure before diff --git a/.downstream/upstream-review-ranxi-v2.10.5.md b/.downstream/upstream-review-ranxi-v2.10.5.md index d6973bbf4e5a..92cb56857b87 100644 --- a/.downstream/upstream-review-ranxi-v2.10.5.md +++ b/.downstream/upstream-review-ranxi-v2.10.5.md @@ -24,15 +24,17 @@ downstream extension; an Astra result never qualifies Sol. ## Changes -- Restore original business STATE echo behavior: unknown or expired local +- Restore original native HTTP business STATE echo behavior: unknown or expired local provenance is preserved; a live record identifying a different account still - strips it. Target two-shot qualification is unchanged. + strips it. Target two-shot qualification and WS provenance are unchanged. - Observe connect/TLS/write/header phases. Retry an observation once only when transport trace proves failure before any HTTP request headers were written. No trace, any response, cancellation, exhausted diagnostic budget, or written headers forbids replay. Every connection attempt counts against the budget. - Port account-test SSE keepalive from #350, preserving the host parser's error - wording and observation cancellation. A heartbeat is not model output. + wording, observation cancellation and already committed successful terminals. + Body-close cancellation after completion must not erase a completed test. + A heartbeat is not model output. - Extend the existing administrator diagnostic with an explicitly selected, bounded native replay for comparable fixture/tool sessions. Ordinary source controls are limited to saved source accounts. No credentials, routing cookies, diff --git a/backend/internal/service/account_test_service.go b/backend/internal/service/account_test_service.go index fd9f4011d172..052563c4fa52 100644 --- a/backend/internal/service/account_test_service.go +++ b/backend/internal/service/account_test_service.go @@ -438,7 +438,7 @@ func (s *AccountTestService) TestAccountConnection(c *gin.Context, accountID int stopKeepalive := startAccountTestSSEKeepalive(c, 10*time.Second) defer func() { // A failed downstream write cancels the observation, never its owner. - if testErr == nil { + if testErr == nil && !c.GetBool("account_test_completion_committed") { testErr = c.Request.Context().Err() } stopKeepalive() @@ -3195,6 +3195,9 @@ func (s *AccountTestService) sendEvent(c *gin.Context, event TestEvent) { log.Printf("failed to write SSE event: %v", err) return } + if event.Type == "test_complete" && event.Success { + c.Set("account_test_completion_committed", true) + } c.Writer.Flush() } diff --git a/backend/internal/service/openai_codex_turn_state.go b/backend/internal/service/openai_codex_turn_state.go index 232a9f8cbffa..40dd578488c2 100644 --- a/backend/internal/service/openai_codex_turn_state.go +++ b/backend/internal/service/openai_codex_turn_state.go @@ -174,21 +174,37 @@ func (s *OpenAIGatewayService) guardOpenAICodexTurnStateEcho(c *gin.Context, acc h.Del(openAICodexTurnStateHeader) return } + // Native HTTP clients own their opaque continuation state. WS and its HTTP + // bridge retain the existing stricter anchor/provenance contract. + preserveUnknown := GetOpenAIClientTransport(c) == OpenAIClientTransportHTTP || + (GetOpenAIClientTransport(c) == OpenAIClientTransportUnknown && c != nil && c.Request != nil && c.Request.Method == http.MethodPost) seed := openAICodexTurnStateSeed(c) if seed == "" { + if !preserveUnknown { + h.Del(openAICodexTurnStateHeader) + } return } raw, ok := s.openaiCodexTurnStateOrigins.Load(seed) if !ok { + if !preserveUnknown { + h.Del(openAICodexTurnStateHeader) + } return } origin, ok := raw.(openAICodexTurnStateOrigin) if !ok { s.openaiCodexTurnStateOrigins.Delete(seed) + if !preserveUnknown { + h.Del(openAICodexTurnStateHeader) + } return } if !origin.expiresAt.IsZero() && time.Now().After(origin.expiresAt) { s.openaiCodexTurnStateOrigins.Delete(seed) + if !preserveUnknown { + h.Del(openAICodexTurnStateHeader) + } return } if origin.accountID != account.ID {