diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 000000000..1dfb4e506 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,100 @@ +name: ci + +on: + pull_request: + push: + branches: + - main + +permissions: + contents: read + +concurrency: + group: ${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: true + +env: + CARGO_TERM_COLOR: always + +jobs: + rust_workspace: + name: Rust workspace and WebSocket pin validation + runs-on: ubuntu-22.04 + timeout-minutes: 20 + steps: + - name: Checkout + uses: actions/checkout@v5 + + - name: Setup Rust + uses: dtolnay/rust-toolchain@stable + with: + toolchain: stable + + - name: Cache Rust dependencies + uses: Swatinem/rust-cache@v2 + + - name: Check WebSocket dependency pins + run: bash scripts/ci/check-websocket-pins.sh + + - name: Check formatting + run: cargo fmt --all -- --check + + - name: Check workspace targets + run: cargo check --workspace --all-targets + + - name: Test Responses WebSocket regressions + run: | + cargo test -p codexmanager-service --lib official_responses_websocket_ --no-fail-fast + cargo test -p codexmanager-service --lib send_websocket_upstream_request_ --no-fail-fast + + frontend_and_web: + name: Frontend build and Web service tests + runs-on: ubuntu-22.04 + steps: + - name: Checkout + uses: actions/checkout@v5 + + - name: Setup release environment + uses: ./.github/actions/setup-release-env + + - name: Install frontend dependencies + working-directory: apps + run: pnpm install --frozen-lockfile + + - name: Build frontend static output + working-directory: apps + run: pnpm run build:desktop + + - name: Test Web service + run: cargo test -p codexmanager-web --no-fail-fast + + tauri_macos: + name: Tauri macOS arm64 bundle + runs-on: macos-14 + steps: + - name: Checkout + uses: actions/checkout@v5 + + - name: Setup release environment + uses: ./.github/actions/setup-release-env + with: + rust-target: aarch64-apple-darwin + enable-pnpm-cache: "false" + + - name: Install frontend dependencies + working-directory: apps + run: pnpm install --frozen-lockfile + + - name: Build frontend static output + working-directory: apps + run: pnpm run build:desktop + + - name: Build Tauri app bundle + uses: ./.github/actions/build-tauri-with-retry + with: + working-directory: apps + tauri-cli-version: 2.10.1 + bundles: app + target: aarch64-apple-darwin + shell-type: bash + max-attempts: "1" diff --git a/Cargo.lock b/Cargo.lock index 0d4347861..5b8d04b22 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -394,6 +394,7 @@ dependencies = [ "tokio", "tokio-tungstenite", "toml_edit", + "tungstenite", "url", "urlencoding", "webbrowser", @@ -430,6 +431,7 @@ dependencies = [ "tokio", "tokio-tungstenite", "tower-http", + "tungstenite", "webbrowser", "winres", ] @@ -947,6 +949,7 @@ checksum = "843fba2746e448b37e26a819579957415c8cef339bf08564fe8b7ddbd959573c" dependencies = [ "crc32fast", "miniz_oxide", + "zlib-rs", ] [[package]] @@ -1169,6 +1172,30 @@ dependencies = [ "hashbrown 0.15.5", ] +[[package]] +name = "headers" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b3314d5adb5d94bcdf56771f2e50dbbc80bb4bdf88967526706205ac9eff24eb" +dependencies = [ + "base64", + "bytes", + "headers-core", + "http", + "httpdate", + "mime", + "sha1", +] + +[[package]] +name = "headers-core" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "54b4a22553d4242c49fddb9ba998a99962b5cc6f22cb5a3482bec22522403ce4" +dependencies = [ + "http", +] + [[package]] name = "heck" version = "0.5.0" @@ -3172,8 +3199,7 @@ dependencies = [ [[package]] name = "tokio-tungstenite" version = "0.28.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d25a406cddcc431a75d3d9afc6a7c0f7428d4891dd973e4d54c56b46127bf857" +source = "git+https://github.com/openai-oss-forks/tokio-tungstenite?rev=0e5b2d73aa18dd9f0a50ee9ff199d5aef7594186#0e5b2d73aa18dd9f0a50ee9ff199d5aef7594186" dependencies = [ "futures-util", "log", @@ -3327,12 +3353,13 @@ checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" [[package]] name = "tungstenite" -version = "0.28.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8628dcc84e5a09eb3d8423d6cb682965dea9133204e8fb3efee74c2a0c259442" +version = "0.27.0" +source = "git+https://github.com/openai-oss-forks/tungstenite-rs?rev=4fffad30fe373adbdcffab9545e9e9bf4f2fc19f#4fffad30fe373adbdcffab9545e9e9bf4f2fc19f" dependencies = [ "bytes", "data-encoding", + "flate2", + "headers", "http", "httparse", "log", @@ -4088,6 +4115,12 @@ dependencies = [ "zopfli", ] +[[package]] +name = "zlib-rs" +version = "0.6.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "34b31d188d9d685a4f9c7b46d6e36631b07058d2cfe190267adce54dc230bf12" + [[package]] name = "zmij" version = "1.0.17" diff --git a/Cargo.toml b/Cargo.toml index 70de0ec02..591e31ee7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -11,5 +11,12 @@ exclude = [ ] resolver = "2" +[patch.crates-io] +tokio-tungstenite = { git = "https://github.com/openai-oss-forks/tokio-tungstenite", rev = "0e5b2d73aa18dd9f0a50ee9ff199d5aef7594186" } +tungstenite = { git = "https://github.com/openai-oss-forks/tungstenite-rs", rev = "4fffad30fe373adbdcffab9545e9e9bf4f2fc19f" } + +[patch."ssh://git@github.com/openai-oss-forks/tungstenite-rs.git"] +tungstenite = { git = "https://github.com/openai-oss-forks/tungstenite-rs", rev = "4fffad30fe373adbdcffab9545e9e9bf4f2fc19f" } + [workspace.package] version = "0.5.3" diff --git a/PR-websocket-preamble-reconnect.md b/PR-websocket-preamble-reconnect.md new file mode 100644 index 000000000..a27db7605 --- /dev/null +++ b/PR-websocket-preamble-reconnect.md @@ -0,0 +1,126 @@ +# fix(gateway): 对齐 Responses WebSocket 压缩协商与首帧恢复 + +## 摘要 + +本 PR 在 PR #430 的心跳、连接上限、大图像帧和有界恢复基础上，继续修复 Responses WebSocket 在大上下文首帧阶段断开、失败账号重复恢复，以及部分上游拒绝压缩协商的问题。 + +本次改动覆盖： + +- 使用官方 Codex 当前固定的 `tokio-tungstenite` / `tungstenite` fork revision； +- 默认按官方客户端协商 `permessage-deflate`，并保留 256 MiB message/frame 上限； +- 上游明确以握手 `400`/`426` 或扩展拒绝信息拒绝压缩时，只重新建立一次不带压缩扩展的连接； +- 首帧发送失败后的恢复排除已经失败的账号，优先尝试其他仍符合线程感知、优先级、禁用和限流过滤的账号； +- 只有完整收到 `response.completed` 才确认 WebSocket 成功，恢复预算耗尽后继续由客户端进入 HTTP fallback； +- 增加根 workspace、Web 测试、前端构建和 Tauri 目标构建的 CI 验证，以及 pinned fork 的维护文档。 + +## 官方行为基线 + +实现以官方 [Responses WebSocket Mode](https://developers.openai.com/api/docs/guides/websocket-mode) 和官方 [Codex Responses WebSocket 客户端](https://github.com/openai/codex/blob/main/codex-rs/codex-api/src/endpoint/responses_websocket.rs) 为基准： + +- 每一轮通过一个 `response.create` 消息开始； +- 图像上下文随 Responses create 请求发送； +- WebSocket 连接在达到服务端时限、断开或不可用后重新建立； +- 连接内的请求按顺序处理； +- 只有 `response.completed` 才确认该轮完成； +- 已经产生实质输出后不透明重放，避免重复输出或工具副作用； +- WebSocket 恢复失败后继续使用官方客户端的 HTTPS/SSE fallback，不强制已经回退的下游 session 再次升级 WebSocket。 + +## 问题与根因 + +### 1. 大图像上下文首帧发送阶段的兼容性不足 + +包含多张内联图像时，完整 `response.create` 文本帧可能达到数十 MiB。此前传输层没有完整对齐官方 fork 和扩展协商配置，首帧发送阶段更容易出现： + +```text +IO error: Broken pipe (os error 32) +``` + +此前的账号恢复逻辑已经处理了大帧发送失败，但如果上游连接策略本身拒绝 `permessage-deflate`，仍会在连接阶段失败。 + +### 2. 首帧失败后的恢复可能重复选择原账号 + +首帧发送失败后，conversation-bound 候选列表可能再次把原账号放在头部，导致同一失败 socket 被重复使用，其他可选账号无法接收首帧。 + +### 3. 依赖与独立 Tauri workspace 容易漂移 + +根 workspace 与 `apps/src-tauri` 是两个 Cargo workspace。只在其中一个 workspace 固定 fork，或只生成其中一个 lockfile，会使桌面构建和服务构建采用不同的 WebSocket 行为。 + +## 修改内容 + +### 官方 WebSocket 传输对齐 + +- 根 workspace 与 `apps/src-tauri` 同步固定： + - `tokio-tungstenite` fork revision `0e5b2d73aa18dd9f0a50ee9ff199d5aef7594186`； + - `tungstenite` fork revision `4fffad30fe373adbdcffab9545e9e9bf4f2fc19f`。 +- 启用 `deflate` / `proxy` feature，并保持官方压缩配置； +- message/frame 上限保持 256 MiB，不开放无限制消息； +- 上游不返回扩展时，正常按未压缩 WebSocket 继续工作； +- 如果上游以 `400`/`426` 或明确的 `permessage-deflate` / WebSocket 扩展拒绝信息拒绝压缩，只重新握手一次且不发送扩展； +- 非压缩协商错误不触发该回退，避免把认证、代理和普通网关错误误判为压缩问题。 + +### 有界首帧恢复与账号轮换 + +- 为本轮恢复维护已失败账号集合； +- 首次恢复排除导致首帧发送失败的账号； +- 每次恢复发送失败后将该账号加入排除集合； +- 有其他候选时保持现有线程感知、会话、手动优先级、禁用、冷却和限流过滤； +- 只有候选全部尝试过时，才在既有有限预算内复用候选池； +- 账号切换时继续清理跨账号 session affinity，并按当前候选重建请求上下文。 + +### CI 与依赖维护 + +- 新增 `scripts/ci/check-websocket-pins.sh`，校验两个 workspace 的 manifest、lockfile 和 fork revision 同步； +- 新增 `docs/zh-CN/WEBSOCKET_DEPENDENCY_MAINTENANCE.md`，记录 fork 的来源、升级步骤、压缩回退条件和验证要求； +- 新增 `.github/workflows/ci.yml`： + - 根 workspace 格式检查与 `cargo check --workspace --all-targets`； + - service WebSocket 回归测试，避免现有跨平台 `codex_skills` 测试失败阻塞本 PR 的传输验证； + - 先构建 `apps/out`，再运行 `codexmanager-web` 测试； + - macOS arm64 Tauri app bundle 构建； + - WebSocket pinned fork 同步检查。 + +## 回归覆盖 + +新增 `official_responses_websocket_retries_without_compression_after_upstream_rejection`： + +1. mock upstream 首次握手确认收到 `permessage-deflate`； +2. 返回 `400 unsupported extension: permessage-deflate`； +3. 验证第二次握手不再携带 `Sec-WebSocket-Extensions`； +4. 验证原始 `response.create` 完整转发； +5. 验证下游收到 `response.completed`。 + +同时保留并继续验证： + +- 约 34 MiB 图像上下文单帧； +- 首帧 socket reset 后的有界恢复； +- 重连 socket 在发送前再次断开的恢复； +- 首帧失败后的账号切换； +- 前导事件阶段重放； +- 已有实质输出后的不重放策略； +- 连接上限、心跳、follow-up 和账号绑定语义。 + +## 验证结果 + +已通过： + +- `bash scripts/ci/check-websocket-pins.sh`； +- `cargo fmt --all -- --check`； +- `git diff --check`； +- `cargo test -p codexmanager-service --lib official_responses_websocket_ --no-fail-fast` — 17 passed； +- `cargo test -p codexmanager-service --lib send_websocket_upstream_request_ --no-fail-fast` — 5 passed； +- WebSocket 连接错误分类测试 — 5 passed； +- `pnpm -C apps run build:desktop`； +- `cargo test -p codexmanager-web --no-fail-fast` — 26 passed； +- `cargo tauri build --bundles app`； +- 生成的 macOS app 通过 ad-hoc code-sign verification。 + +完整 service lib 串行验证结果为 `1421 passed / 1 failed / 3 ignored`。唯一失败是现有 `codex_skills::tests::directory_import_detects_same_size_file_replacement_and_fifo_entries` 的跨平台文件替换断言（Linux CI 中表现为返回了仍可读的文件句柄），单独运行也可复现，与本 PR 修改文件无关；因此 CI 保留 workspace target 编译检查，并将服务测试聚焦于本 PR 的 WebSocket 回归集合。 + +## 兼容性与边界 + +- 不改变公开 `/v1/responses` endpoint 形状； +- 不改变同一 session 已进入 HTTP fallback 后的粘性； +- 不强制 HTTP session 再次升级 WebSocket； +- 不在已转发实质模型/工具内容后静默复制请求； +- 不新增配置项，不覆盖线程感知账号分配或禁用/冷却/限流过滤； +- 心跳仍只发送 WebSocket 协议层 Ping； +- 上游仍不可用或恢复预算耗尽时，仍按官方策略返回失败并允许客户端降级 HTTP。 diff --git a/apps/src-tauri/Cargo.lock b/apps/src-tauri/Cargo.lock index a5244b241..28597bd09 100644 --- a/apps/src-tauri/Cargo.lock +++ b/apps/src-tauri/Cargo.lock @@ -927,6 +927,7 @@ dependencies = [ "tokio", "tokio-tungstenite", "toml_edit 0.22.27", + "tungstenite", "url", "urlencoding", "webbrowser", @@ -1839,6 +1840,7 @@ checksum = "843fba2746e448b37e26a819579957415c8cef339bf08564fe8b7ddbd959573c" dependencies = [ "crc32fast", "miniz_oxide", + "zlib-rs", ] [[package]] @@ -2411,6 +2413,30 @@ dependencies = [ "hashbrown 0.15.5", ] +[[package]] +name = "headers" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b3314d5adb5d94bcdf56771f2e50dbbc80bb4bdf88967526706205ac9eff24eb" +dependencies = [ + "base64 0.22.1", + "bytes", + "headers-core", + "http", + "httpdate", + "mime", + "sha1", +] + +[[package]] +name = "headers-core" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "54b4a22553d4242c49fddb9ba998a99962b5cc6f22cb5a3482bec22522403ce4" +dependencies = [ + "http", +] + [[package]] name = "heck" version = "0.4.1" @@ -6193,8 +6219,7 @@ dependencies = [ [[package]] name = "tokio-tungstenite" version = "0.28.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d25a406cddcc431a75d3d9afc6a7c0f7428d4891dd973e4d54c56b46127bf857" +source = "git+https://github.com/openai-oss-forks/tokio-tungstenite?rev=0e5b2d73aa18dd9f0a50ee9ff199d5aef7594186#0e5b2d73aa18dd9f0a50ee9ff199d5aef7594186" dependencies = [ "futures-util", "log", @@ -6459,12 +6484,13 @@ checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" [[package]] name = "tungstenite" -version = "0.28.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8628dcc84e5a09eb3d8423d6cb682965dea9133204e8fb3efee74c2a0c259442" +version = "0.27.0" +source = "git+https://github.com/openai-oss-forks/tungstenite-rs?rev=4fffad30fe373adbdcffab9545e9e9bf4f2fc19f#4fffad30fe373adbdcffab9545e9e9bf4f2fc19f" dependencies = [ "bytes", "data-encoding", + "flate2", + "headers", "http", "httparse", "log", @@ -7928,6 +7954,12 @@ dependencies = [ "zopfli", ] +[[package]] +name = "zlib-rs" +version = "0.6.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "34b31d188d9d685a4f9c7b46d6e36631b07058d2cfe190267adce54dc230bf12" + [[package]] name = "zmij" version = "1.0.21" diff --git a/apps/src-tauri/Cargo.toml b/apps/src-tauri/Cargo.toml index bf1eddc9a..5508b0138 100644 --- a/apps/src-tauri/Cargo.toml +++ b/apps/src-tauri/Cargo.toml @@ -39,3 +39,10 @@ glib = "0.18" [target.'cfg(windows)'.dependencies] windows-sys = { version = "0.61", features = ["Win32_Foundation", "Win32_UI_Shell", "Win32_UI_WindowsAndMessaging"] } + +[patch.crates-io] +tokio-tungstenite = { git = "https://github.com/openai-oss-forks/tokio-tungstenite", rev = "0e5b2d73aa18dd9f0a50ee9ff199d5aef7594186" } +tungstenite = { git = "https://github.com/openai-oss-forks/tungstenite-rs", rev = "4fffad30fe373adbdcffab9545e9e9bf4f2fc19f" } + +[patch."ssh://git@github.com/openai-oss-forks/tungstenite-rs.git"] +tungstenite = { git = "https://github.com/openai-oss-forks/tungstenite-rs", rev = "4fffad30fe373adbdcffab9545e9e9bf4f2fc19f" } diff --git a/crates/service/Cargo.toml b/crates/service/Cargo.toml index 3d09815c1..36cb38b94 100644 --- a/crates/service/Cargo.toml +++ b/crates/service/Cargo.toml @@ -20,7 +20,8 @@ axum = { version = "0.8", features = ["ws"] } tokio = { version = "1", features = ["rt-multi-thread", "net", "time"] } futures-util = "0.3" eventsource-stream = "0.2.3" -tokio-tungstenite = { version = "0.28", features = ["rustls-tls-webpki-roots"] } +tokio-tungstenite = { version = "0.28", features = ["proxy", "rustls-tls-webpki-roots"] } +tungstenite = { version = "0.27", features = ["deflate", "proxy"] } rustls = { version = "0.23", features = ["ring"] } url = "2" webbrowser = "0.8" diff --git a/crates/service/src/gateway/upstream/attempt_flow/transport_tests.rs b/crates/service/src/gateway/upstream/attempt_flow/transport_tests.rs index 6d8ffebfc..b6c8b5790 100644 --- a/crates/service/src/gateway/upstream/attempt_flow/transport_tests.rs +++ b/crates/service/src/gateway/upstream/attempt_flow/transport_tests.rs @@ -13,6 +13,10 @@ use std::sync::mpsc::{self, Receiver}; use std::thread; use std::time::{Duration, Instant}; use tokio::runtime::Builder; +use tokio_tungstenite::accept_hdr_async_with_config; +use tokio_tungstenite::tungstenite::extensions::compression::deflate::DeflateConfig; +use tokio_tungstenite::tungstenite::extensions::ExtensionsConfig; +use tokio_tungstenite::tungstenite::protocol::WebSocketConfig; type WsServerRequest = tokio_tungstenite::tungstenite::handshake::server::Request; type WsServerResponse = tokio_tungstenite::tungstenite::handshake::server::Response; @@ -55,6 +59,14 @@ fn header_value<'a>(headers: &'a [(String, String)], name: &str) -> Option<&'a s .map(|(_, value)| value.as_str()) } +fn mock_websocket_config() -> WebSocketConfig { + let mut config = WebSocketConfig::default(); + let mut extensions = ExtensionsConfig::default(); + extensions.permessage_deflate = Some(DeflateConfig::default()); + config.extensions = extensions; + config +} + #[test] fn explicit_stateless_mode_targets_only_session_scoped_headers() { for name in [ @@ -250,7 +262,7 @@ fn spawn_mock_websocket_upstream( let listener = tokio::net::TcpListener::from_std(listener).expect("convert websocket listener"); let (stream, _) = listener.accept().await.expect("accept websocket client"); - let mut websocket = tokio_tungstenite::accept_hdr_async( + let mut websocket = accept_hdr_async_with_config( stream, |request: &WsServerRequest, response: WsServerResponse| { let headers = request @@ -266,6 +278,7 @@ fn spawn_mock_websocket_upstream( let _ = headers_tx.send(headers); Ok(response) }, + Some(mock_websocket_config()), ) .await .expect("accept websocket handshake"); diff --git a/crates/service/src/http/responses_websocket.rs b/crates/service/src/http/responses_websocket.rs index 74ca6bc15..c39fc5a5a 100644 --- a/crates/service/src/http/responses_websocket.rs +++ b/crates/service/src/http/responses_websocket.rs @@ -13,6 +13,8 @@ use std::time::{Duration, Instant}; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::TcpStream; use tokio_tungstenite::tungstenite::client::IntoClientRequest; +use tokio_tungstenite::tungstenite::extensions::compression::deflate::DeflateConfig; +use tokio_tungstenite::tungstenite::extensions::ExtensionsConfig; use tokio_tungstenite::tungstenite::handshake::client::{ Request as WsClientRequest, Response as WsClientResponse, }; @@ -180,6 +182,19 @@ impl WsConnectError { body.contains(WEBSOCKET_CONNECTION_LIMIT_REACHED_CODE) || body.contains(&WEBSOCKET_CONNECTION_LIMIT_REACHED_MESSAGE.to_ascii_lowercase()) } + + fn is_compression_negotiation_rejection(&self) -> bool { + let message = self.message.to_ascii_lowercase(); + let body = String::from_utf8_lossy(&self.response_body).to_ascii_lowercase(); + let extension_signal = message.contains("sec-websocket-extensions") + || message.contains("permessage-deflate") + || body.contains("sec-websocket-extensions") + || body.contains("permessage-deflate") + || body.contains("unsupported extension") + || body.contains("compression extension"); + let rejected_handshake = matches!(self.status_code, Some(400 | 426)); + extension_signal && (rejected_handshake || self.status_code.is_none()) + } } impl fmt::Display for WsConnectError { @@ -1861,10 +1876,11 @@ fn ws_account_is_unavailable( .any(|(account, _)| account.id == upstream.account_id)) } -async fn connect_upstream_websocket( +async fn connect_upstream_websocket_excluding_accounts( context: &WsRequestContext, model: Option<&str>, previous_account_id: Option<&str>, + excluded_account_ids: &HashSet, ) -> Result { let storage = open_storage().ok_or_else(|| { WsSessionError::service_unavailable_bilingual("存储不可用", "storage unavailable") @@ -1888,11 +1904,22 @@ async fn connect_upstream_websocket( .map(|(account, _)| account.id.clone()) .collect::>(); let conversation_routing = routed.conversation_routing.clone(); + let has_unexcluded_candidate = routed + .candidates + .iter() + .any(|(account, _)| !excluded_account_ids.contains(account.id.as_str())); + let candidates = routed + .candidates + .into_iter() + .filter(|(account, _)| { + !has_unexcluded_candidate || !excluded_account_ids.contains(account.id.as_str()) + }) + .collect::>(); drop(storage); let ws_url = build_upstream_websocket_url(&context.effective_upstream_base)?; let mut last_error = None; - for (account, token) in routed.candidates { + for (account, token) in candidates { match connect_account_upstream_websocket( context, &account, @@ -1933,12 +1960,32 @@ async fn connect_upstream_websocket_with_timeout( context: &WsRequestContext, model: Option<&str>, previous_account_id: Option<&str>, +) -> Result { + connect_upstream_websocket_with_timeout_excluding_accounts( + context, + model, + previous_account_id, + &HashSet::new(), + ) + .await +} + +async fn connect_upstream_websocket_with_timeout_excluding_accounts( + context: &WsRequestContext, + model: Option<&str>, + previous_account_id: Option<&str>, + excluded_account_ids: &HashSet, ) -> Result { let connect_timeout = crate::gateway::current_upstream_connect_timeout().max(std::time::Duration::from_secs(1)); match tokio::time::timeout( connect_timeout, - connect_upstream_websocket(context, model, previous_account_id), + connect_upstream_websocket_excluding_accounts( + context, + model, + previous_account_id, + excluded_account_ids, + ), ) .await { @@ -1965,13 +2012,18 @@ async fn reconnect_upstream_for_pending_request( completed_tool_calls: &CompletedWsToolCallCache, ) -> Result { let mut previous_account_id = previous_account_id.map(str::to_owned); + let mut excluded_account_ids = HashSet::new(); + if let Some(account_id) = previous_account_id.as_deref() { + excluded_account_ids.insert(account_id.to_string()); + } let mut last_send_error = None; for attempt in 1..=RESPONSES_WS_MAX_PENDING_FRAME_SEND_ATTEMPTS { - let mut replacement = connect_upstream_websocket_with_timeout( + let mut replacement = connect_upstream_websocket_with_timeout_excluding_accounts( context, pending.prepared.model.as_deref(), previous_account_id.as_deref(), + &excluded_account_ids, ) .await?; let account_changed = previous_account_id @@ -2027,6 +2079,7 @@ async fn reconnect_upstream_for_pending_request( last_send_error = Some(format!( "send upstream websocket frame after reconnect failed for account {account_id}: {err}" )); + excluded_account_ids.insert(account_id.clone()); previous_account_id = Some(account_id); } } @@ -2465,29 +2518,69 @@ pub(crate) async fn connect_upstream_websocket_request_detailed( WsConnectError, > { ensure_rustls_crypto_provider(); + let compressed_request = request.clone(); + let first_result = connect_upstream_websocket_request_with_config( + compressed_request, + ws_url, + proxy_url, + responses_ws_transport_config(true), + ) + .await; + match first_result { + Ok(result) => Ok(result), + Err(err) if err.is_compression_negotiation_rejection() => { + log::info!( + "event=responses_ws_compression_rejected retry=without_permessage_deflate ws_url={}", + ws_url + ); + connect_upstream_websocket_request_with_config( + request, + ws_url, + proxy_url, + responses_ws_transport_config(false), + ) + .await + } + Err(err) => Err(err), + } +} + +async fn connect_upstream_websocket_request_with_config( + request: WsClientRequest, + ws_url: &str, + proxy_url: Option<&str>, + config: WebSocketConfig, +) -> Result< + ( + tokio_tungstenite::WebSocketStream>, + WsClientResponse, + ), + WsConnectError, +> { let Some(proxy_url) = proxy_url.map(str::trim).filter(|value| !value.is_empty()) else { - return connect_async_tls_with_config( - request, - Some(responses_ws_transport_config()), - false, - None, - ) - .await - .map_err(WsConnectError::from_tungstenite); + return connect_async_tls_with_config(request, Some(config), false, None) + .await + .map_err(WsConnectError::from_tungstenite); }; let stream = connect_websocket_proxy_tcp(ws_url, proxy_url) .await .map_err(WsConnectError::from_message)?; - client_async_tls_with_config(request, stream, Some(responses_ws_transport_config()), None) + client_async_tls_with_config(request, stream, Some(config), None) .await .map_err(WsConnectError::from_tungstenite) } -fn responses_ws_transport_config() -> WebSocketConfig { - WebSocketConfig::default() +fn responses_ws_transport_config(enable_permessage_deflate: bool) -> WebSocketConfig { + let mut config = WebSocketConfig::default() .max_message_size(Some(RESPONSES_WS_MAX_MESSAGE_BYTES)) - .max_frame_size(Some(RESPONSES_WS_MAX_MESSAGE_BYTES)) + .max_frame_size(Some(RESPONSES_WS_MAX_MESSAGE_BYTES)); + if enable_permessage_deflate { + let mut extensions = ExtensionsConfig::default(); + extensions.permessage_deflate = Some(DeflateConfig::default()); + config.extensions = extensions; + } + config } async fn connect_websocket_proxy_tcp(ws_url: &str, proxy_url: &str) -> Result { diff --git a/crates/service/src/http/responses_websocket_tests.rs b/crates/service/src/http/responses_websocket_tests.rs index 9cd4549bd..2431aea3d 100644 --- a/crates/service/src/http/responses_websocket_tests.rs +++ b/crates/service/src/http/responses_websocket_tests.rs @@ -531,6 +531,30 @@ fn websocket_connect_error_detects_connection_limit_body() { assert!(err.is_websocket_connection_limit_reached()); } +#[test] +fn websocket_connect_error_detects_compression_negotiation_rejection() { + let mut response = super::WsClientResponse::new(Some( + br#"unsupported extension: permessage-deflate"#.to_vec(), + )); + *response.status_mut() = axum::http::StatusCode::BAD_REQUEST; + let err = super::WsConnectError::from_tungstenite(tokio_tungstenite::tungstenite::Error::Http( + Box::new(response), + )); + + assert!(err.is_compression_negotiation_rejection()); +} + +#[test] +fn websocket_connect_error_does_not_treat_unrelated_bad_request_as_compression_rejection() { + let mut response = super::WsClientResponse::new(Some(br#"invalid response.create"#.to_vec())); + *response.status_mut() = axum::http::StatusCode::BAD_REQUEST; + let err = super::WsConnectError::from_tungstenite(tokio_tungstenite::tungstenite::Error::Http( + Box::new(response), + )); + + assert!(!err.is_compression_negotiation_rejection()); +} + #[test] fn inspect_ws_terminal_event_infers_usage_limit_status_without_explicit_status() { let event = inspect_ws_terminal_event( diff --git a/crates/service/src/http/tests/proxy_runtime_tests.rs b/crates/service/src/http/tests/proxy_runtime_tests.rs index 02f53acd7..18f480ccc 100644 --- a/crates/service/src/http/tests/proxy_runtime_tests.rs +++ b/crates/service/src/http/tests/proxy_runtime_tests.rs @@ -17,12 +17,14 @@ use std::collections::HashMap; use std::path::PathBuf; use std::sync::atomic::{AtomicU64, Ordering}; use std::time::{Duration, SystemTime, UNIX_EPOCH}; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; use tokio::sync::oneshot; -use tokio_tungstenite::accept_hdr_async; use tokio_tungstenite::accept_hdr_async_with_config; use tokio_tungstenite::connect_async; use tokio_tungstenite::tungstenite::client::IntoClientRequest; -use tokio_tungstenite::tungstenite::handshake::server::{Request, Response}; +use tokio_tungstenite::tungstenite::extensions::compression::deflate::DeflateConfig; +use tokio_tungstenite::tungstenite::extensions::ExtensionsConfig; +use tokio_tungstenite::tungstenite::handshake::server::{Callback, Request, Response}; use tokio_tungstenite::tungstenite::protocol::WebSocketConfig; use tokio_tungstenite::tungstenite::Message; @@ -34,6 +36,32 @@ struct EnvGuard { static TEST_DB_COUNTER: AtomicU64 = AtomicU64::new(0); const TEST_ZSTD_MAX_BODY_BYTES: usize = 256 * 1024 * 1024; const TEST_LARGE_RESPONSES_WS_FRAME_BYTES: usize = 17 * 1024 * 1024; +const TEST_IMAGE_CONTEXT_RESPONSES_WS_FRAME_BYTES: usize = 34 * 1024 * 1024; + +fn test_upstream_ws_config() -> WebSocketConfig { + let mut config = WebSocketConfig::default() + .max_message_size(Some( + TEST_IMAGE_CONTEXT_RESPONSES_WS_FRAME_BYTES + 2 * 1024 * 1024, + )) + .max_frame_size(Some( + TEST_IMAGE_CONTEXT_RESPONSES_WS_FRAME_BYTES + 2 * 1024 * 1024, + )); + let mut extensions = ExtensionsConfig::default(); + extensions.permessage_deflate = Some(DeflateConfig::default()); + config.extensions = extensions; + config +} + +async fn accept_hdr_async( + stream: S, + callback: C, +) -> Result, tokio_tungstenite::tungstenite::Error> +where + S: AsyncRead + AsyncWrite + Unpin, + C: Callback + Unpin, +{ + accept_hdr_async_with_config(stream, callback, Some(test_upstream_ws_config())).await +} impl EnvGuard { /// 函数 `set` @@ -645,6 +673,142 @@ struct UpstreamWsCapture { frames: Vec, } +#[derive(Debug)] +struct UpstreamWsCompressionFallbackCapture { + first_headers: HashMap, + second_headers: HashMap, + frames: Vec, +} + +async fn read_raw_websocket_handshake_headers( + stream: &mut tokio::net::TcpStream, +) -> HashMap { + let mut request = Vec::new(); + let mut chunk = [0_u8; 4096]; + while !request.windows(4).any(|window| window == b"\r\n\r\n") { + let count = stream + .read(&mut chunk) + .await + .expect("read raw websocket handshake"); + assert!( + count > 0, + "upstream handshake ended before headers completed" + ); + request.extend_from_slice(&chunk[..count]); + assert!( + request.len() <= 64 * 1024, + "upstream websocket handshake exceeded test limit" + ); + } + + String::from_utf8_lossy(&request) + .split("\r\n") + .skip(1) + .take_while(|line| !line.is_empty()) + .filter_map(|line| { + let (name, value) = line.split_once(':')?; + Some((name.trim().to_ascii_lowercase(), value.trim().to_string())) + }) + .collect() +} + +async fn start_mock_upstream_ws_rejects_compression_then_accepts() -> ( + String, + oneshot::Receiver, + tokio::task::JoinHandle<()>, +) { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind compression-fallback mock upstream"); + let addr = listener + .local_addr() + .expect("compression-fallback mock upstream addr"); + let (capture_tx, capture_rx) = oneshot::channel(); + let handle = tokio::spawn(async move { + let (mut first_stream, _) = listener + .accept() + .await + .expect("accept compressed upstream handshake"); + let first_headers = read_raw_websocket_handshake_headers(&mut first_stream).await; + let body = b"unsupported extension: permessage-deflate"; + let response = format!( + "HTTP/1.1 400 Bad Request\r\nContent-Type: text/plain\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", + body.len(), + String::from_utf8_lossy(body) + ); + first_stream + .write_all(response.as_bytes()) + .await + .expect("send compression rejection"); + first_stream + .shutdown() + .await + .expect("close rejected compressed websocket"); + + let (second_stream, _) = listener + .accept() + .await + .expect("accept uncompressed upstream handshake"); + let captured_headers = std::sync::Arc::new(std::sync::Mutex::new(None)); + let captured_headers_clone = captured_headers.clone(); + let mut websocket = accept_hdr_async_with_config( + second_stream, + move |request: &Request, response: Response| { + let headers = request + .headers() + .iter() + .filter_map(|(name, value)| { + Some(( + name.as_str().to_ascii_lowercase(), + value.to_str().ok()?.to_string(), + )) + }) + .collect::>(); + let mut guard = captured_headers_clone + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + *guard = Some(headers); + Ok(response) + }, + Some(test_upstream_ws_config()), + ) + .await + .expect("accept uncompressed websocket handshake"); + + let frame = match websocket.next().await { + Some(Ok(Message::Text(text))) => text.to_string(), + other => panic!("expected response.create after compression fallback, got {other:?}"), + }; + for payload in [ + serde_json::json!({ + "type": "response.created", + "response": { "id": "resp_ws_compression_fallback" } + }), + serde_json::json!({ + "type": "response.completed", + "response": { "id": "resp_ws_compression_fallback" } + }), + ] { + websocket + .send(Message::Text(payload.to_string().into())) + .await + .expect("send compression fallback response"); + } + + let second_headers = captured_headers + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .take() + .expect("capture uncompressed handshake"); + let _ = capture_tx.send(UpstreamWsCompressionFallbackCapture { + first_headers, + second_headers, + frames: vec![frame], + }); + }); + (addr.to_string(), capture_rx, handle) +} + async fn start_mock_upstream_ws() -> ( String, tokio::sync::mpsc::UnboundedReceiver, @@ -663,9 +827,7 @@ async fn start_mock_upstream_ws() -> ( None::<(String, HashMap)>, )); let captured_headers_clone = captured_headers.clone(); - let upstream_config = WebSocketConfig::default() - .max_message_size(Some(TEST_LARGE_RESPONSES_WS_FRAME_BYTES * 2)) - .max_frame_size(Some(TEST_LARGE_RESPONSES_WS_FRAME_BYTES * 2)); + let upstream_config = test_upstream_ws_config(); let mut websocket = accept_hdr_async_with_config( stream, move |request: &Request, response: Response| { @@ -764,9 +926,7 @@ async fn start_mock_upstream_ws_resets_before_first_frame() -> ( .accept() .await .expect("accept replacement initial-send-reset upstream"); - let upstream_config = WebSocketConfig::default() - .max_message_size(Some(TEST_LARGE_RESPONSES_WS_FRAME_BYTES * 2)) - .max_frame_size(Some(TEST_LARGE_RESPONSES_WS_FRAME_BYTES * 2)); + let upstream_config = test_upstream_ws_config(); let mut replacement = accept_hdr_async_with_config( replacement_stream, |_: &Request, response: Response| Ok(response), @@ -816,9 +976,7 @@ async fn start_mock_upstream_ws_resets_twice_before_first_frame() -> ( .expect("double-initial-send-reset mock upstream addr"); let (event_tx, event_rx) = tokio::sync::mpsc::unbounded_channel(); let handle = tokio::spawn(async move { - let upstream_config = WebSocketConfig::default() - .max_message_size(Some(TEST_LARGE_RESPONSES_WS_FRAME_BYTES * 2)) - .max_frame_size(Some(TEST_LARGE_RESPONSES_WS_FRAME_BYTES * 2)); + let upstream_config = test_upstream_ws_config(); for round in 1..=2 { let (stream, _) = listener @@ -890,6 +1048,103 @@ async fn start_mock_upstream_ws_resets_twice_before_first_frame() -> ( (addr.to_string(), event_rx, handle) } +async fn start_mock_upstream_ws_switches_after_initial_reset() -> ( + String, + tokio::sync::mpsc::UnboundedReceiver<(String, String)>, + tokio::task::JoinHandle<()>, +) { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind account-switch initial-send-reset mock upstream"); + let addr = listener + .local_addr() + .expect("account-switch initial-send-reset mock upstream addr"); + let (event_tx, event_rx) = tokio::sync::mpsc::unbounded_channel(); + let handle = tokio::spawn(async move { + let upstream_config = test_upstream_ws_config(); + + for _ in 0..3 { + let (stream, _) = listener + .accept() + .await + .expect("accept account-switch upstream"); + let captured_account_id = std::sync::Arc::new(std::sync::Mutex::new(None::)); + let captured_account_id_clone = captured_account_id.clone(); + let mut websocket = accept_hdr_async_with_config( + stream, + move |request: &Request, response: Response| { + let account_id = request + .headers() + .get("chatgpt-account-id") + .and_then(|value| value.to_str().ok()) + .map(str::to_string); + let mut guard = captured_account_id_clone + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + *guard = account_id; + Ok(response) + }, + Some(upstream_config), + ) + .await + .expect("accept account-switch websocket handshake"); + let account_id = captured_account_id + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .clone() + .unwrap_or_else(|| "missing-account-id".to_string()); + + if account_id == "workspace-failed" { + let raw_stream = websocket + .into_inner() + .into_std() + .expect("convert account-switch reset stream"); + force_tcp_reset(&raw_stream); + drop(raw_stream); + event_tx + .send((account_id, String::new())) + .expect("record failed account reset"); + continue; + } + + let text = match websocket.next().await { + Some(Ok(Message::Text(text))) => text.to_string(), + other => panic!( + "expected account-switch response.create on replacement account, got {other:?}" + ), + }; + event_tx + .send((account_id, text)) + .expect("record successful replacement account frame"); + for payload in [ + serde_json::json!({ + "type": "response.created", + "response": { "id": "resp_ws_account_switch" } + }), + serde_json::json!({ + "type": "response.completed", + "response": { "id": "resp_ws_account_switch" } + }), + ] { + websocket + .send(Message::Text(payload.to_string().into())) + .await + .expect("send account-switch replacement response"); + } + let _ = websocket.next().await; + return; + } + + event_tx + .send(( + "no-replacement-account".to_string(), + "the failed account was reused for every bounded retry".to_string(), + )) + .expect("record missing account switch"); + }); + (addr.to_string(), event_rx, handle) +} + async fn start_mock_upstream_ws_holds_first_response() -> ( String, tokio::sync::mpsc::UnboundedReceiver, @@ -2200,6 +2455,13 @@ async fn official_responses_websocket_proxies_frames_and_headers() { capture.headers.get("openai-beta").map(String::as_str), Some("responses_websockets=2026-02-06") ); + assert!( + capture + .headers + .get("sec-websocket-extensions") + .is_some_and(|value| value.contains("permessage-deflate")), + "official-compatible upstream websocket transport must offer permessage-deflate" + ); assert_eq!(capture.headers.get("version").map(String::as_str), None); assert_eq!( capture @@ -2319,6 +2581,111 @@ async fn official_responses_websocket_proxies_frames_and_headers() { .expect("join mock upstream"); } +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn official_responses_websocket_retries_without_compression_after_upstream_rejection() { + let _guard = crate::test_env_guard(); + let _http_proxy = EnvGuard::clear("http_proxy"); + let _https_proxy = EnvGuard::clear("https_proxy"); + let _all_proxy = EnvGuard::clear("all_proxy"); + let _upper_http_proxy = EnvGuard::clear("HTTP_PROXY"); + let _upper_https_proxy = EnvGuard::clear("HTTPS_PROXY"); + let _upper_all_proxy = EnvGuard::clear("ALL_PROXY"); + let _no_proxy = EnvGuard::set("NO_PROXY", "127.0.0.1,localhost"); + let _lower_no_proxy = EnvGuard::clear("no_proxy"); + let db_path = new_test_db_path("codexmanager-proxy-runtime-ws-compression-fallback"); + let storage = init_test_storage(&db_path); + let _db_guard = EnvGuard::set("CODEXMANAGER_DB_PATH", db_path.to_string_lossy().as_ref()); + let (upstream_addr, capture_rx, upstream_handle) = + start_mock_upstream_ws_rejects_compression_then_accepts().await; + insert_api_key_record( + &storage, + "platform_key_ws_compression_fallback", + crate::apikey_profile::ROTATION_ACCOUNT, + Some(format!( + "http://{upstream_addr}/chatgpt.com/backend-api/codex" + )), + ); + insert_account_and_token(&storage); + tokio::task::spawn_blocking(|| { + crate::gateway::reload_runtime_config_from_env(); + let _ = crate::gateway::front_proxy_max_body_bytes(); + }) + .await + .expect("reload runtime config"); + + let state = ProxyState { + backend_base_url: "http://127.0.0.1:1".to_string(), + client: Client::new(), + }; + let (front_addr, shutdown_tx, server_handle) = start_front_proxy_test_server(state).await; + let request = build_ws_request( + &format!("ws://{front_addr}/v1/responses"), + "platform_key_ws_compression_fallback", + &[("OpenAI-Beta", "responses_websockets=2026-02-06")], + ); + let (mut client_ws, response) = connect_async(request).await.expect("websocket connects"); + assert_eq!(response.status(), StatusCode::SWITCHING_PROTOCOLS); + + client_ws + .send(Message::Text( + serde_json::json!({ + "type": "response.create", + "model": "gpt-5.6-sol", + "input": "retry without compression after an upstream rejection" + }) + .to_string() + .into(), + )) + .await + .expect("send compression fallback response.create"); + + loop { + let event = tokio::time::timeout(Duration::from_secs(5), client_ws.next()) + .await + .expect("compression fallback client event timeout") + .expect("compression fallback client event") + .expect("compression fallback client event result"); + match event { + Message::Text(text) if text.contains("\"response.completed\"") => break, + Message::Text(text) if text.contains("\"type\":\"error\"") => { + panic!("compression fallback error escaped to client: {text}"); + } + Message::Text(_) => {} + other => panic!("unexpected compression fallback event: {other:?}"), + } + } + + let capture = tokio::time::timeout(Duration::from_secs(5), capture_rx) + .await + .expect("compression fallback capture timeout") + .expect("compression fallback capture result"); + assert!( + capture + .first_headers + .get("sec-websocket-extensions") + .is_some_and(|value| value.contains("permessage-deflate")), + "the first upstream handshake must offer the official compression extension" + ); + assert_eq!( + capture.second_headers.get("sec-websocket-extensions"), + None, + "a compression rejection must retry with an uncompressed handshake" + ); + assert_eq!(capture.frames.len(), 1); + assert!(capture.frames[0].contains("retry without compression")); + + let _ = client_ws.close(None).await; + let _ = shutdown_tx.send(()); + tokio::time::timeout(Duration::from_secs(5), server_handle) + .await + .expect("front proxy compression fallback shutdown timeout") + .expect("join compression fallback front proxy"); + tokio::time::timeout(Duration::from_secs(5), upstream_handle) + .await + .expect("compression fallback upstream shutdown timeout") + .expect("join compression fallback mock upstream"); +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn official_responses_websocket_block_policy_rejects_initial_frame() { let _guard = crate::test_env_guard(); @@ -2594,7 +2961,7 @@ async fn official_responses_websocket_accepts_large_image_context_frame() { let (mut client_ws, response) = connect_async(request).await.expect("websocket connects"); assert_eq!(response.status(), StatusCode::SWITCHING_PROTOCOLS); - let image_data = "A".repeat(TEST_LARGE_RESPONSES_WS_FRAME_BYTES); + let image_data = "A".repeat(TEST_IMAGE_CONTEXT_RESPONSES_WS_FRAME_BYTES); let payload = serde_json::json!({ "type": "response.create", "model": "gpt-5.6-sol", @@ -2611,7 +2978,7 @@ async fn official_responses_websocket_accepts_large_image_context_frame() { }] }) .to_string(); - assert!(payload.len() > 16 * 1024 * 1024); + assert!(payload.len() > TEST_IMAGE_CONTEXT_RESPONSES_WS_FRAME_BYTES); client_ws .send(Message::Text(payload.into())) @@ -2903,6 +3270,151 @@ async fn official_responses_websocket_retries_when_reconnected_socket_breaks_bef .expect("join double-reset recovery mock upstream"); } +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn official_responses_websocket_switches_account_after_initial_send_reset() { + let _guard = crate::test_env_guard(); + let _http_proxy = EnvGuard::clear("http_proxy"); + let _https_proxy = EnvGuard::clear("https_proxy"); + let _all_proxy = EnvGuard::clear("all_proxy"); + let _upper_http_proxy = EnvGuard::clear("HTTP_PROXY"); + let _upper_https_proxy = EnvGuard::clear("HTTPS_PROXY"); + let _upper_all_proxy = EnvGuard::clear("ALL_PROXY"); + let _no_proxy = EnvGuard::set("NO_PROXY", "127.0.0.1,localhost"); + let _lower_no_proxy = EnvGuard::clear("no_proxy"); + let db_path = new_test_db_path("codexmanager-proxy-runtime-ws-initial-send-account-switch"); + let mut storage = init_test_storage(&db_path); + let _db_guard = EnvGuard::set("CODEXMANAGER_DB_PATH", db_path.to_string_lossy().as_ref()); + let (upstream_addr, mut upstream_events, upstream_handle) = + start_mock_upstream_ws_switches_after_initial_reset().await; + insert_api_key_record( + &storage, + "platform_key_ws_initial_send_account_switch", + crate::apikey_profile::ROTATION_ACCOUNT, + Some(format!( + "http://{upstream_addr}/chatgpt.com/backend-api/codex" + )), + ); + insert_account_and_token_with_id( + &storage, + "acc_ws_failed", + "failed-account", + "workspace-failed", + "failed-token", + 0, + ); + insert_account_and_token_with_id( + &storage, + "acc_ws_replacement", + "replacement-account", + "workspace-replacement", + "replacement-token", + 1, + ); + storage + .set_preferred_account(Some("acc_ws_failed")) + .expect("prefer failed account for initial websocket attempt"); + crate::gateway::invalidate_candidate_cache(); + tokio::task::spawn_blocking(|| { + crate::gateway::reload_runtime_config_from_env(); + let _ = crate::gateway::front_proxy_max_body_bytes(); + }) + .await + .expect("reload runtime config"); + + let routed = crate::gateway::gateway_collect_routed_candidates_for_ws( + &storage, + "gk_proxy_runtime_ws", + Some("gpt-5.6-sol"), + None, + None, + ) + .expect("collect account-switch websocket candidates"); + assert_eq!( + routed + .candidates + .first() + .map(|(account, _)| account.id.as_str()), + Some("acc_ws_failed") + ); + + let state = ProxyState { + backend_base_url: "http://127.0.0.1:1".to_string(), + client: Client::new(), + }; + let (front_addr, shutdown_tx, server_handle) = start_front_proxy_test_server(state).await; + let request = build_ws_request( + &format!("ws://{front_addr}/v1/responses"), + "platform_key_ws_initial_send_account_switch", + &[("OpenAI-Beta", "responses_websockets=2026-02-06")], + ); + let (mut client_ws, response) = connect_async(request).await.expect("websocket connects"); + assert_eq!(response.status(), StatusCode::SWITCHING_PROTOCOLS); + + client_ws + .send(Message::Text( + serde_json::json!({ + "type": "response.create", + "model": "gpt-5.6-sol", + "store": true, + "input": "switch accounts after the initial upstream socket resets" + }) + .to_string() + .into(), + )) + .await + .expect("send account-switch response.create"); + + let mut replacement_frame = None; + for _ in 0..4 { + let event = tokio::time::timeout(Duration::from_secs(10), upstream_events.recv()) + .await + .expect("account-switch upstream event timeout") + .expect("account-switch upstream event"); + if event.0 == "workspace-replacement" { + replacement_frame = Some(event.1); + break; + } + assert_eq!( + event.0, "workspace-failed", + "the initial failed account may be attempted only before failover" + ); + assert!(event.1.is_empty()); + } + let replacement_frame = replacement_frame + .expect("a bounded initial-send recovery must try the next eligible account"); + assert!(replacement_frame.contains("switch accounts after the initial upstream socket resets")); + + loop { + let event = tokio::time::timeout(Duration::from_secs(10), client_ws.next()) + .await + .expect("account-switch response timeout") + .expect("account-switch client event") + .expect("account-switch client event result"); + match event { + Message::Text(text) if text.contains("\"response.completed\"") => break, + Message::Text(text) if text.contains("\"type\":\"error\"") => { + panic!("account-switch recovery error escaped to client: {text}"); + } + Message::Text(_) => {} + other => panic!("unexpected account-switch event: {other:?}"), + } + } + + let _ = client_ws.close(None).await; + let _ = shutdown_tx.send(()); + tokio::time::timeout(Duration::from_secs(10), server_handle) + .await + .expect("front proxy account-switch shutdown timeout") + .expect("join account-switch front proxy"); + tokio::time::timeout(Duration::from_secs(10), upstream_handle) + .await + .expect("mock upstream account-switch shutdown timeout") + .expect("join account-switch mock upstream"); + storage + .set_preferred_account(None) + .expect("clear preferred account after account-switch test"); +} + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn official_responses_websocket_keeps_idle_session_alive_with_heartbeat() { let _guard = crate::test_env_guard(); diff --git a/crates/web/Cargo.toml b/crates/web/Cargo.toml index 479c473ea..d71cfd04b 100644 --- a/crates/web/Cargo.toml +++ b/crates/web/Cargo.toml @@ -20,7 +20,8 @@ reqwest = { version = "0.12", features = ["rustls-tls", "stream"] } serde = { version = "1", features = ["derive"] } serde_json = "1" tokio = { version = "1", features = ["rt-multi-thread", "net", "time", "macros"] } -tokio-tungstenite = { version = "0.28", features = ["rustls-tls-webpki-roots"] } +tokio-tungstenite = { version = "0.28", features = ["proxy", "rustls-tls-webpki-roots"] } +tungstenite = { version = "0.27", features = ["deflate", "proxy"] } tower-http = { version = "0.6", features = ["fs"] } webbrowser = "0.8" diff --git a/docs/zh-CN/WEBSOCKET_DEPENDENCY_MAINTENANCE.md b/docs/zh-CN/WEBSOCKET_DEPENDENCY_MAINTENANCE.md new file mode 100644 index 000000000..067c2d895 --- /dev/null +++ b/docs/zh-CN/WEBSOCKET_DEPENDENCY_MAINTENANCE.md @@ -0,0 +1,35 @@ +# Responses WebSocket 依赖维护说明 + +## 当前锁定 + +Responses WebSocket 由服务端工作区和 `apps/src-tauri` 独立工作区共同编译。两个工作区必须使用同一组依赖来源： + +- `tokio-tungstenite`：OpenAI fork revision `0e5b2d73aa18dd9f0a50ee9ff199d5aef7594186`（版本线 `0.28`）。 +- `tungstenite`：OpenAI fork revision `4fffad30fe373adbdcffab9545e9e9bf4f2fc19f`（版本线 `0.27`）。 +- `tungstenite` 的 `deflate` feature 保持开启；Responses WebSocket 客户端默认协商 `permessage-deflate`，并把消息/帧上限保持在 256 MiB。 + +这些 revision 与官方 Codex Responses WebSocket 客户端当前使用的压缩配置保持一致。仓库同时保留无压缩兼容路径：如果上游在握手阶段以 `400`/`426` 或明确的扩展拒绝信息拒绝 `permessage-deflate`，客户端只重新建立一次不带扩展的握手；其他握手错误不触发该回退。 + +## 更新流程 + +升级任一 fork 时，必须在同一个变更中完成以下步骤： + +1. 先核对官方 Codex `responses_websocket` 实现和其 lockfile 中的依赖 revision，以及 fork 的变更记录。 +2. 同步修改根 `Cargo.toml`、`apps/src-tauri/Cargo.toml`、`Cargo.lock` 和 `apps/src-tauri/Cargo.lock`。 +3. 执行 `bash scripts/ci/check-websocket-pins.sh`，确认两个工作区的 patch 与 lockfile 一致。 +4. 执行 `cargo fmt --all -- --check`、Responses WebSocket 定向测试、`cargo test --workspace --no-fail-fast`。 +5. 安装前端依赖后执行 `pnpm -C apps run build:desktop`、`cargo test -p codexmanager-web --no-fail-fast`，再执行至少一个 Tauri 目标的 bundle 构建。 +6. 检查压缩协商成功和“上游拒绝压缩后无压缩重试”两条回归路径；在所有支持的上游都确认兼容前，保留无压缩回退。 + +禁止只更新一个工作区、只更新 lockfile，或在未通过上述验证时改用浮动 git branch/tag。若官方 Codex 切换 fork revision、上游修复了握手兼容性，或本仓库升级 `tokio-tungstenite`/`tungstenite` 的主版本，应重新评估回退条件、帧大小限制和代理路径，并在 CI 中保留上述同步检查。 + +## CI 保障 + +`.github/workflows/ci.yml` 在 Pull Request 和 `main` push 上执行： + +- 根工作区格式检查、`cargo check --workspace --all-targets` 和 service WebSocket 回归测试； +- 前端静态构建与 `codexmanager-web` 测试； +- macOS arm64 Tauri bundle 构建； +- 两个 Cargo 工作区的 revision/lockfile 同步检查。 + +这样可以在依赖 revision、前端静态资源或 Tauri 独立工作区发生漂移时尽早发现，而不会把运行时兼容性依赖隐藏在本机环境中。 diff --git a/scripts/ci/check-websocket-pins.sh b/scripts/ci/check-websocket-pins.sh new file mode 100755 index 000000000..deec6baca --- /dev/null +++ b/scripts/ci/check-websocket-pins.sh @@ -0,0 +1,33 @@ +#!/usr/bin/env bash + +set -euo pipefail + +repo_root="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)" +root_manifest="$repo_root/Cargo.toml" +tauri_manifest="$repo_root/apps/src-tauri/Cargo.toml" +root_lock="$repo_root/Cargo.lock" +tauri_lock="$repo_root/apps/src-tauri/Cargo.lock" + +tokio_tungstenite_rev="0e5b2d73aa18dd9f0a50ee9ff199d5aef7594186" +tungstenite_rev="4fffad30fe373adbdcffab9545e9e9bf4f2fc19f" + +require_line() { + local file="$1" + local needle="$2" + if ! grep -Fq "$needle" "$file"; then + printf 'missing expected WebSocket pin in %s: %s\n' "$file" "$needle" >&2 + exit 1 + fi +} + +for manifest in "$root_manifest" "$tauri_manifest"; do + require_line "$manifest" "tokio-tungstenite = { git = \"https://github.com/openai-oss-forks/tokio-tungstenite\", rev = \"$tokio_tungstenite_rev\" }" + require_line "$manifest" "tungstenite = { git = \"https://github.com/openai-oss-forks/tungstenite-rs\", rev = \"$tungstenite_rev\" }" +done + +for lockfile in "$root_lock" "$tauri_lock"; do + require_line "$lockfile" "source = \"git+https://github.com/openai-oss-forks/tokio-tungstenite?rev=$tokio_tungstenite_rev#$tokio_tungstenite_rev\"" + require_line "$lockfile" "source = \"git+https://github.com/openai-oss-forks/tungstenite-rs?rev=$tungstenite_rev#$tungstenite_rev\"" +done + +printf 'WebSocket dependency pins are synchronized across the root and Tauri workspaces.\n'