diff --git a/.downstream/native-domains.md b/.downstream/native-domains.md index 44fca4557c73..3ce5db23942f 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,16 @@ 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`. +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 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 +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 +223,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..92cb56857b87 --- /dev/null +++ b/.downstream/upstream-review-ranxi-v2.10.5.md @@ -0,0 +1,50 @@ +# 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 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 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, 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, + 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..052563c4fa52 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 && !c.GetBool("account_test_completion_committed") { + testErr = c.Request.Context().Err() + } + stopKeepalive() + }() ctx := c.Request.Context() testOpts := firstAccountTestOptions(opts) @@ -3187,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/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..40dd578488c2 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) { @@ -173,25 +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 == "" { - h.Del(openAICodexTurnStateHeader) + if !preserveUnknown { + h.Del(openAICodexTurnStateHeader) + } return } raw, ok := s.openaiCodexTurnStateOrigins.Load(seed) if !ok { - h.Del(openAICodexTurnStateHeader) + if !preserveUnknown { + h.Del(openAICodexTurnStateHeader) + } return } origin, ok := raw.(openAICodexTurnStateOrigin) if !ok { s.openaiCodexTurnStateOrigins.Delete(seed) - h.Del(openAICodexTurnStateHeader) + if !preserveUnknown { + h.Del(openAICodexTurnStateHeader) + } return } if !origin.expiresAt.IsZero() && time.Now().After(origin.expiresAt) { s.openaiCodexTurnStateOrigins.Delete(seed) - h.Del(openAICodexTurnStateHeader) + if !preserveUnknown { + 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")