From 795a2c440a0cd0573ba285ccadf4b88c70542078 Mon Sep 17 00:00:00 2001 From: Mi Tom <6468993+MDX-Tom@users.noreply.github.com> Date: Fri, 21 Aug 2026 17:19:53 +0800 Subject: [PATCH 1/4] =?UTF-8?q?fix(gateway):=20=E5=AF=B9=E9=BD=90=E5=AE=98?= =?UTF-8?q?=E6=96=B9=20WebSocket=20=E5=8E=8B=E7=BC=A9=E4=B8=8E=E9=A6=96?= =?UTF-8?q?=E5=B8=A7=E8=B4=A6=E5=8F=B7=E6=81=A2=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Cargo.lock | 43 ++- Cargo.toml | 7 + PR-websocket-preamble-reconnect.md | 117 +++++++ apps/src-tauri/Cargo.lock | 42 ++- apps/src-tauri/Cargo.toml | 7 + crates/service/Cargo.toml | 3 +- .../upstream/attempt_flow/transport_tests.rs | 15 +- .../service/src/http/responses_websocket.rs | 57 +++- .../src/http/tests/proxy_runtime_tests.rs | 297 +++++++++++++++++- crates/web/Cargo.toml | 3 +- 10 files changed, 559 insertions(+), 32 deletions(-) create mode 100644 PR-websocket-preamble-reconnect.md 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..fbf7c5a3a --- /dev/null +++ b/PR-websocket-preamble-reconnect.md @@ -0,0 +1,117 @@ +# fix(gateway): 修复 Responses WebSocket 大图像首帧 Broken pipe 与失败账号重复恢复 + +## Summary + +本 PR 在 PR #430 已有的心跳、连接上限、大图像帧和有界恢复基础上，继续修复新线程或图像上下文线程在上游 WebSocket 首帧发送阶段出现 `Broken pipe (os error 32)`，随后连续 502 并降级 HTTP 的问题。 + +本次修复遵循官方 Responses WebSocket / Codex 客户端的传输行为： + +- 使用官方 Codex 当前固定的 `tokio-tungstenite` / `tungstenite` fork 版本； +- 为上游 WebSocket 启用 `permessage-deflate` 协商，降低大图像上下文单帧发送压力； +- 首帧发送失败后的恢复仍然有界，不无限重试； +- 如果存在其他可选账号，恢复尝试会排除已经首帧发送失败的账号，避免同一账号重复触发相同的 `Broken pipe`； +- 只有完整收到 `response.completed` 才算 WebSocket 请求成功；恢复预算耗尽后仍保留官方客户端的 HTTP fallback。 + +## Official behavior baseline + +实现以官方 [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 形状，图像上下文可以作为输入的一部分发送； +- 连接关闭、连接缓存不可用或连接达到时限后，需要建立新连接，并按上下文情况使用 `previous_response_id` 或完整输入； +- 连接内的请求按顺序处理，当前代理路径继续使用官方顺序语义的兼容子集； +- 只有 `response.completed` 才确认该轮完成； +- WebSocket 恢复失败后由客户端按 session 级策略进入 HTTPS/SSE fallback，Manager 不强制已经回退的下游 session 重新升级 WebSocket。 + +## Problems and root causes + +### 1. 大图像上下文首帧发送阶段的传输不一致 + +Codex 会将本轮完整上下文序列化为一个 `response.create` 文本消息。包含多张内联图像时，单帧可能达到数十 MiB。此前 Manager 使用 crates.io 的 WebSocket 传输配置，没有按官方 Codex 配置启用 `permessage-deflate`，大帧更容易在上游连接刚建立后暴露为： + +```text +IO error: Broken pipe (os error 32) +``` + +这发生在响应完成前，因此下游看不到 `response.completed`，会继续执行有限恢复并最终降级 HTTP。 + +### 2. 首帧恢复可能重复使用同一失败账号 + +此前首帧发送失败后，恢复逻辑仍可能从 conversation-bound 候选列表头部重新选择原账号。只要该账号的上游连接策略持续关闭 socket，就会得到连续相同的 `Broken pipe`，其他可选账号没有机会接收首帧。 + +### 3. 测试 mock 没有声明可处理扩展 + +接入官方压缩协商后，测试用 upstream 也必须显式提供 WebSocket 扩展配置；否则严格的 WebSocket handshake 会把 `Sec-WebSocket-Extensions` 当作无效请求。测试 fixture 已同步到官方协商语义。 + +## Changes + +### 官方 WebSocket 传输对齐 + +- 固定官方 Codex 使用的 `tokio-tungstenite` fork revision； +- 固定官方 Codex 使用的 `tungstenite` fork revision； +- 启用 `deflate` / `proxy` 相关 feature； +- 上游 WebSocket `WebSocketConfig` 保留 256 MiB 的 message/frame 硬上限； +- 在连接握手中启用 `permessage-deflate`，服务端支持时协商压缩，服务端不返回扩展时仍按未压缩 WebSocket 继续工作。 + +### 有界首帧恢复与账号轮换 + +- 为恢复函数维护本轮已失败账号集合； +- 首次恢复优先排除导致首帧发送失败的账号； +- 每次恢复发送失败后将对应账号加入排除集合； +- 在存在其他候选时按原有线程/会话/优先级顺序选择下一个可选账号； +- 只有所有候选都已尝试过时，才允许在原有有界预算内再次使用候选池； +- 不改变禁用、冷却、限流账号的候选过滤，也不覆盖线程感知账号分配策略； +- 账号切换时继续清理跨账号 session affinity，并按当前候选重建请求上下文。 + +### 安全终态与 fallback + +- `response.completed` 仍是唯一成功终态； +- `response.done`、`response.failed`、`response.incomplete`、TCP reset 和发送错误不会清除恢复状态； +- 已转发实质模型输出、工具事件或二进制内容后不透明重放，避免重复输出或工具副作用； +- 恢复预算耗尽后返回 502，由官方 Codex 客户端继续其 HTTP fallback； +- 不强制同一 session 从 HTTP 再次切回 WebSocket。 + +## Reproduction and regression coverage + +### 大图像上下文 + +隔离测试构造约 34 MiB 的单个 `response.create` 文本帧，其中包含内联 `input_image` data URL： + +1. 建立下游 Responses WebSocket； +2. 发送大图像上下文； +3. 验证 Manager→upstream 的帧完整到达； +4. 验证下游收到 `response.completed`； +5. 验证同一连接的 follow-up 仍可继续。 + +### 首帧失败后的账号轮换 + +1. 将首个候选账号设置为握手后 reset； +2. 验证恢复发送不重复使用该失败账号； +3. 验证下一个候选账号收到完整 `response.create`； +4. 验证下游收到 `response.completed`，不收到恢复错误； +5. 验证单账号场景仍遵守原有有限重试预算。 + +## Validation + +已通过： + +- `cargo fmt --all -- --check` +- `git diff --check` +- `cargo test -p codexmanager-service --lib official_responses_websocket_ --no-fail-fast` — 16 passed +- `cargo test -p codexmanager-service --lib send_websocket_upstream_request_ --no-fail-fast` — 5 passed +- `cargo test -p codexmanager-service --lib official_responses_websocket_accepts_large_image_context_frame --no-fail-fast` — 1 passed（约 34 MiB 图像上下文） +- `cargo test -p codexmanager-web --no-fail-fast` — 26 passed +- `pnpm -C apps run build:desktop` — passed +- `cargo-tauri build --bundles app` — passed +- 生成的 App 通过 ad-hoc code-sign verification + +## Scope / Compatibility + +- 不改变公开 `/v1/responses` endpoint 形状； +- 不改变 Codex 同一 session 内 HTTP fallback 的粘性； +- 不强制已经降级 HTTP 的下游 session 重新升级 WebSocket； +- 不在已经转发实质模型/工具内容后静默复制请求； +- 不新增配置项，不改变账号优先级、线程感知分配或禁用/冷却过滤策略； +- 心跳仍只发送 WebSocket 协议层 Ping，不注入 Responses JSON 事件； +- 大帧仍使用 256 MiB 硬上限，不开放无限消息； +- 上游确实不可用或恢复预算耗尽时，仍会按官方策略降级 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..67f0a0967 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, }; @@ -1861,10 +1863,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 +1891,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 +1947,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 +1999,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 +2066,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); } } @@ -2485,9 +2525,14 @@ pub(crate) async fn connect_upstream_websocket_request_detailed( } fn responses_ws_transport_config() -> WebSocketConfig { - WebSocketConfig::default() + let mut extensions = ExtensionsConfig::default(); + extensions.permessage_deflate = Some(DeflateConfig::default()); + + 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)); + config.extensions = extensions; + config } async fn connect_websocket_proxy_tcp(ws_url: &str, proxy_url: &str) -> Result { diff --git a/crates/service/src/http/tests/proxy_runtime_tests.rs b/crates/service/src/http/tests/proxy_runtime_tests.rs index 02f53acd7..9c22c4c4b 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, AsyncWrite}; 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` @@ -663,9 +691,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 +790,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 +840,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 +912,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 +2319,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 @@ -2594,7 +2720,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 +2737,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 +3029,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" From 356aece34f2d5582b073dc7fa29f22deb03e1d65 Mon Sep 17 00:00:00 2001 From: Mi Tom <6468993+MDX-Tom@users.noreply.github.com> Date: Sat, 22 Aug 2026 17:52:13 +0800 Subject: [PATCH 2/4] fix(gateway): add websocket compression fallback and CI coverage --- .github/workflows/ci.yml | 94 +++++++ PR-websocket-preamble-reconnect.md | 154 +++++------ .../service/src/http/responses_websocket.rs | 76 +++++- .../src/http/responses_websocket_tests.rs | 24 ++ .../src/http/tests/proxy_runtime_tests.rs | 243 +++++++++++++++++- .../zh-CN/WEBSOCKET_DEPENDENCY_MAINTENANCE.md | 35 +++ scripts/ci/check-websocket-pins.sh | 33 +++ 7 files changed, 571 insertions(+), 88 deletions(-) create mode 100644 .github/workflows/ci.yml create mode 100644 docs/zh-CN/WEBSOCKET_DEPENDENCY_MAINTENANCE.md create mode 100755 scripts/ci/check-websocket-pins.sh diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 000000000..be42df075 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,94 @@ +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 + 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: Test workspace + run: cargo test --workspace --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/PR-websocket-preamble-reconnect.md b/PR-websocket-preamble-reconnect.md index fbf7c5a3a..327fac6e3 100644 --- a/PR-websocket-preamble-reconnect.md +++ b/PR-websocket-preamble-reconnect.md @@ -1,117 +1,125 @@ -# fix(gateway): 修复 Responses WebSocket 大图像首帧 Broken pipe 与失败账号重复恢复 +# fix(gateway): 对齐 Responses WebSocket 压缩协商与首帧恢复 -## Summary +## 摘要 -本 PR 在 PR #430 已有的心跳、连接上限、大图像帧和有界恢复基础上，继续修复新线程或图像上下文线程在上游 WebSocket 首帧发送阶段出现 `Broken pipe (os error 32)`，随后连续 502 并降级 HTTP 的问题。 +本 PR 在 PR #430 的心跳、连接上限、大图像帧和有界恢复基础上，继续修复 Responses WebSocket 在大上下文首帧阶段断开、失败账号重复恢复，以及部分上游拒绝压缩协商的问题。 -本次修复遵循官方 Responses WebSocket / Codex 客户端的传输行为： +本次改动覆盖： -- 使用官方 Codex 当前固定的 `tokio-tungstenite` / `tungstenite` fork 版本； -- 为上游 WebSocket 启用 `permessage-deflate` 协商，降低大图像上下文单帧发送压力； -- 首帧发送失败后的恢复仍然有界，不无限重试； -- 如果存在其他可选账号，恢复尝试会排除已经首帧发送失败的账号，避免同一账号重复触发相同的 `Broken pipe`； -- 只有完整收到 `response.completed` 才算 WebSocket 请求成功；恢复预算耗尽后仍保留官方客户端的 HTTP fallback。 +- 使用官方 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 的维护文档。 -## Official behavior baseline +## 官方行为基线 实现以官方 [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 形状，图像上下文可以作为输入的一部分发送； -- 连接关闭、连接缓存不可用或连接达到时限后，需要建立新连接，并按上下文情况使用 `previous_response_id` 或完整输入； -- 连接内的请求按顺序处理，当前代理路径继续使用官方顺序语义的兼容子集； +- 图像上下文随 Responses create 请求发送； +- WebSocket 连接在达到服务端时限、断开或不可用后重新建立； +- 连接内的请求按顺序处理； - 只有 `response.completed` 才确认该轮完成； -- WebSocket 恢复失败后由客户端按 session 级策略进入 HTTPS/SSE fallback，Manager 不强制已经回退的下游 session 重新升级 WebSocket。 +- 已经产生实质输出后不透明重放，避免重复输出或工具副作用； +- WebSocket 恢复失败后继续使用官方客户端的 HTTPS/SSE fallback，不强制已经回退的下游 session 再次升级 WebSocket。 -## Problems and root causes +## 问题与根因 -### 1. 大图像上下文首帧发送阶段的传输不一致 +### 1. 大图像上下文首帧发送阶段的兼容性不足 -Codex 会将本轮完整上下文序列化为一个 `response.create` 文本消息。包含多张内联图像时，单帧可能达到数十 MiB。此前 Manager 使用 crates.io 的 WebSocket 传输配置，没有按官方 Codex 配置启用 `permessage-deflate`，大帧更容易在上游连接刚建立后暴露为： +包含多张内联图像时，完整 `response.create` 文本帧可能达到数十 MiB。此前传输层没有完整对齐官方 fork 和扩展协商配置，首帧发送阶段更容易出现： ```text IO error: Broken pipe (os error 32) ``` -这发生在响应完成前，因此下游看不到 `response.completed`，会继续执行有限恢复并最终降级 HTTP。 +此前的账号恢复逻辑已经处理了大帧发送失败，但如果上游连接策略本身拒绝 `permessage-deflate`，仍会在连接阶段失败。 -### 2. 首帧恢复可能重复使用同一失败账号 +### 2. 首帧失败后的恢复可能重复选择原账号 -此前首帧发送失败后，恢复逻辑仍可能从 conversation-bound 候选列表头部重新选择原账号。只要该账号的上游连接策略持续关闭 socket，就会得到连续相同的 `Broken pipe`，其他可选账号没有机会接收首帧。 +首帧发送失败后，conversation-bound 候选列表可能再次把原账号放在头部，导致同一失败 socket 被重复使用，其他可选账号无法接收首帧。 -### 3. 测试 mock 没有声明可处理扩展 +### 3. 依赖与独立 Tauri workspace 容易漂移 -接入官方压缩协商后，测试用 upstream 也必须显式提供 WebSocket 扩展配置；否则严格的 WebSocket handshake 会把 `Sec-WebSocket-Extensions` 当作无效请求。测试 fixture 已同步到官方协商语义。 +根 workspace 与 `apps/src-tauri` 是两个 Cargo workspace。只在其中一个 workspace 固定 fork，或只生成其中一个 lockfile，会使桌面构建和服务构建采用不同的 WebSocket 行为。 -## Changes +## 修改内容 ### 官方 WebSocket 传输对齐 -- 固定官方 Codex 使用的 `tokio-tungstenite` fork revision； -- 固定官方 Codex 使用的 `tungstenite` fork revision； -- 启用 `deflate` / `proxy` 相关 feature； -- 上游 WebSocket `WebSocketConfig` 保留 256 MiB 的 message/frame 硬上限； -- 在连接握手中启用 `permessage-deflate`，服务端支持时协商压缩，服务端不返回扩展时仍按未压缩 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，并按当前候选重建请求上下文。 -### 安全终态与 fallback +### CI 与依赖维护 -- `response.completed` 仍是唯一成功终态； -- `response.done`、`response.failed`、`response.incomplete`、TCP reset 和发送错误不会清除恢复状态； -- 已转发实质模型输出、工具事件或二进制内容后不透明重放，避免重复输出或工具副作用； -- 恢复预算耗尽后返回 502，由官方 Codex 客户端继续其 HTTP fallback； -- 不强制同一 session 从 HTTP 再次切回 WebSocket。 +- 新增 `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 test --workspace --no-fail-fast`； + - 先构建 `apps/out`，再运行 `codexmanager-web` 测试； + - macOS arm64 Tauri app bundle 构建； + - WebSocket pinned fork 同步检查。 -## Reproduction and regression coverage +## 回归覆盖 -### 大图像上下文 +新增 `official_responses_websocket_retries_without_compression_after_upstream_rejection`： -隔离测试构造约 34 MiB 的单个 `response.create` 文本帧，其中包含内联 `input_image` data URL： +1. mock upstream 首次握手确认收到 `permessage-deflate`； +2. 返回 `400 unsupported extension: permessage-deflate`； +3. 验证第二次握手不再携带 `Sec-WebSocket-Extensions`； +4. 验证原始 `response.create` 完整转发； +5. 验证下游收到 `response.completed`。 -1. 建立下游 Responses WebSocket； -2. 发送大图像上下文； -3. 验证 Manager→upstream 的帧完整到达； -4. 验证下游收到 `response.completed`； -5. 验证同一连接的 follow-up 仍可继续。 +同时保留并继续验证： -### 首帧失败后的账号轮换 +- 约 34 MiB 图像上下文单帧； +- 首帧 socket reset 后的有界恢复； +- 重连 socket 在发送前再次断开的恢复； +- 首帧失败后的账号切换； +- 前导事件阶段重放； +- 已有实质输出后的不重放策略； +- 连接上限、心跳、follow-up 和账号绑定语义。 -1. 将首个候选账号设置为握手后 reset； -2. 验证恢复发送不重复使用该失败账号； -3. 验证下一个候选账号收到完整 `response.create`； -4. 验证下游收到 `response.completed`，不收到恢复错误； -5. 验证单账号场景仍遵守原有有限重试预算。 - -## Validation +## 验证结果 已通过： -- `cargo fmt --all -- --check` -- `git diff --check` -- `cargo test -p codexmanager-service --lib official_responses_websocket_ --no-fail-fast` — 16 passed -- `cargo test -p codexmanager-service --lib send_websocket_upstream_request_ --no-fail-fast` — 5 passed -- `cargo test -p codexmanager-service --lib official_responses_websocket_accepts_large_image_context_frame --no-fail-fast` — 1 passed（约 34 MiB 图像上下文） -- `cargo test -p codexmanager-web --no-fail-fast` — 26 passed -- `pnpm -C apps run build:desktop` — passed -- `cargo-tauri build --bundles app` — passed -- 生成的 App 通过 ad-hoc code-sign verification +- `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` 的临时目录根路径断言（`source skill entry escaped its root`），单独运行也可复现，与本 PR 修改文件无关；该项不作为 WebSocket 放行依据，CI 仍保留完整 workspace 检查。 -## Scope / Compatibility +## 兼容性与边界 - 不改变公开 `/v1/responses` endpoint 形状； -- 不改变 Codex 同一 session 内 HTTP fallback 的粘性； -- 不强制已经降级 HTTP 的下游 session 重新升级 WebSocket； -- 不在已经转发实质模型/工具内容后静默复制请求； -- 不新增配置项，不改变账号优先级、线程感知分配或禁用/冷却过滤策略； -- 心跳仍只发送 WebSocket 协议层 Ping，不注入 Responses JSON 事件； -- 大帧仍使用 256 MiB 硬上限，不开放无限消息； -- 上游确实不可用或恢复预算耗尽时，仍会按官方策略降级 HTTP。 +- 不改变同一 session 已进入 HTTP fallback 后的粘性； +- 不强制 HTTP session 再次升级 WebSocket； +- 不在已转发实质模型/工具内容后静默复制请求； +- 不新增配置项，不覆盖线程感知账号分配或禁用/冷却/限流过滤； +- 心跳仍只发送 WebSocket 协议层 Ping； +- 上游仍不可用或恢复预算耗尽时，仍按官方策略返回失败并允许客户端降级 HTTP。 diff --git a/crates/service/src/http/responses_websocket.rs b/crates/service/src/http/responses_websocket.rs index 67f0a0967..c39fc5a5a 100644 --- a/crates/service/src/http/responses_websocket.rs +++ b/crates/service/src/http/responses_websocket.rs @@ -182,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 { @@ -2505,33 +2518,68 @@ 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 { - let mut extensions = ExtensionsConfig::default(); - extensions.permessage_deflate = Some(DeflateConfig::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)); - config.extensions = extensions; + if enable_permessage_deflate { + let mut extensions = ExtensionsConfig::default(); + extensions.permessage_deflate = Some(DeflateConfig::default()); + config.extensions = extensions; + } config } 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 9c22c4c4b..18f480ccc 100644 --- a/crates/service/src/http/tests/proxy_runtime_tests.rs +++ b/crates/service/src/http/tests/proxy_runtime_tests.rs @@ -17,7 +17,7 @@ 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, AsyncWrite}; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; use tokio::sync::oneshot; use tokio_tungstenite::accept_hdr_async_with_config; use tokio_tungstenite::connect_async; @@ -673,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, @@ -2445,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(); diff --git a/docs/zh-CN/WEBSOCKET_DEPENDENCY_MAINTENANCE.md b/docs/zh-CN/WEBSOCKET_DEPENDENCY_MAINTENANCE.md new file mode 100644 index 000000000..602a7809f --- /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 test --workspace --no-fail-fast`； +- 前端静态构建与 `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' From 5a3afefb37ed678814653fb01bc1f2b346344fe8 Mon Sep 17 00:00:00 2001 From: Mi Tom <6468993+MDX-Tom@users.noreply.github.com> Date: Sat, 22 Aug 2026 18:06:20 +0800 Subject: [PATCH 3/4] ci: keep websocket validation bounded --- .github/workflows/ci.yml | 8 ++++++-- PR-websocket-preamble-reconnect.md | 3 ++- docs/zh-CN/WEBSOCKET_DEPENDENCY_MAINTENANCE.md | 2 +- 3 files changed, 9 insertions(+), 4 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index be42df075..079818acb 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -20,6 +20,7 @@ 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 @@ -38,8 +39,11 @@ jobs: - name: Check formatting run: cargo fmt --all -- --check - - name: Test workspace - run: cargo test --workspace --no-fail-fast + - name: Check workspace targets + run: cargo check --workspace --all-targets + + - name: Test service workspace library + run: cargo test -p codexmanager-service --lib --no-fail-fast -- --test-threads=1 frontend_and_web: name: Frontend build and Web service tests diff --git a/PR-websocket-preamble-reconnect.md b/PR-websocket-preamble-reconnect.md index 327fac6e3..2697a6e77 100644 --- a/PR-websocket-preamble-reconnect.md +++ b/PR-websocket-preamble-reconnect.md @@ -72,7 +72,8 @@ IO error: Broken pipe (os error 32) - 新增 `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 test --workspace --no-fail-fast`； + - 根 workspace 格式检查与 `cargo check --workspace --all-targets`； + - service workspace library 串行测试，避免环境相关的集成测试长时间占用 CI； - 先构建 `apps/out`，再运行 `codexmanager-web` 测试； - macOS arm64 Tauri app bundle 构建； - WebSocket pinned fork 同步检查。 diff --git a/docs/zh-CN/WEBSOCKET_DEPENDENCY_MAINTENANCE.md b/docs/zh-CN/WEBSOCKET_DEPENDENCY_MAINTENANCE.md index 602a7809f..9a85ee977 100644 --- a/docs/zh-CN/WEBSOCKET_DEPENDENCY_MAINTENANCE.md +++ b/docs/zh-CN/WEBSOCKET_DEPENDENCY_MAINTENANCE.md @@ -27,7 +27,7 @@ Responses WebSocket 由服务端工作区和 `apps/src-tauri` 独立工作区共 `.github/workflows/ci.yml` 在 Pull Request 和 `main` push 上执行： -- 根工作区格式检查和 `cargo test --workspace --no-fail-fast`； +- 根工作区格式检查、`cargo check --workspace --all-targets` 和 service workspace library 测试； - 前端静态构建与 `codexmanager-web` 测试； - macOS arm64 Tauri bundle 构建； - 两个 Cargo 工作区的 revision/lockfile 同步检查。 From 04a6e23f6f6c65f9ff74e6e9c2852218455bd48a Mon Sep 17 00:00:00 2001 From: Mi Tom <6468993+MDX-Tom@users.noreply.github.com> Date: Sat, 22 Aug 2026 18:15:22 +0800 Subject: [PATCH 4/4] ci: scope service checks to websocket regressions --- .github/workflows/ci.yml | 6 ++++-- PR-websocket-preamble-reconnect.md | 4 ++-- docs/zh-CN/WEBSOCKET_DEPENDENCY_MAINTENANCE.md | 2 +- 3 files changed, 7 insertions(+), 5 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 079818acb..1dfb4e506 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -42,8 +42,10 @@ jobs: - name: Check workspace targets run: cargo check --workspace --all-targets - - name: Test service workspace library - run: cargo test -p codexmanager-service --lib --no-fail-fast -- --test-threads=1 + - 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 diff --git a/PR-websocket-preamble-reconnect.md b/PR-websocket-preamble-reconnect.md index 2697a6e77..a27db7605 100644 --- a/PR-websocket-preamble-reconnect.md +++ b/PR-websocket-preamble-reconnect.md @@ -73,7 +73,7 @@ IO error: Broken pipe (os error 32) - 新增 `docs/zh-CN/WEBSOCKET_DEPENDENCY_MAINTENANCE.md`，记录 fork 的来源、升级步骤、压缩回退条件和验证要求； - 新增 `.github/workflows/ci.yml`： - 根 workspace 格式检查与 `cargo check --workspace --all-targets`； - - service workspace library 串行测试，避免环境相关的集成测试长时间占用 CI； + - service WebSocket 回归测试，避免现有跨平台 `codex_skills` 测试失败阻塞本 PR 的传输验证； - 先构建 `apps/out`，再运行 `codexmanager-web` 测试； - macOS arm64 Tauri app bundle 构建； - WebSocket pinned fork 同步检查。 @@ -113,7 +113,7 @@ IO error: Broken pipe (os error 32) - `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` 的临时目录根路径断言（`source skill entry escaped its root`），单独运行也可复现，与本 PR 修改文件无关；该项不作为 WebSocket 放行依据，CI 仍保留完整 workspace 检查。 +完整 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 回归集合。 ## 兼容性与边界 diff --git a/docs/zh-CN/WEBSOCKET_DEPENDENCY_MAINTENANCE.md b/docs/zh-CN/WEBSOCKET_DEPENDENCY_MAINTENANCE.md index 9a85ee977..067c2d895 100644 --- a/docs/zh-CN/WEBSOCKET_DEPENDENCY_MAINTENANCE.md +++ b/docs/zh-CN/WEBSOCKET_DEPENDENCY_MAINTENANCE.md @@ -27,7 +27,7 @@ Responses WebSocket 由服务端工作区和 `apps/src-tauri` 独立工作区共 `.github/workflows/ci.yml` 在 Pull Request 和 `main` push 上执行： -- 根工作区格式检查、`cargo check --workspace --all-targets` 和 service workspace library 测试； +- 根工作区格式检查、`cargo check --workspace --all-targets` 和 service WebSocket 回归测试； - 前端静态构建与 `codexmanager-web` 测试； - macOS arm64 Tauri bundle 构建； - 两个 Cargo 工作区的 revision/lockfile 同步检查。