From a4427b3e075d12d27069cc071028355d549ebf17 Mon Sep 17 00:00:00 2001 From: eric8810 Date: Sun, 13 Sep 2026 13:53:35 +0800 Subject: [PATCH 1/5] =?UTF-8?q?docs(rfc):=20add=20RFC-0034=20=E2=80=94=20r?= =?UTF-8?q?ealtime=20STT=20follow-ups=20(WS=20proxy,=20ElevenLabs/Cartesia?= =?UTF-8?q?=20do=5Fstream)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P1 WS proxy tunnel honoring the global ProxyConfig (CONNECT + no_proxy, SOCKS errors loudly); P2 ElevenLabs scribe_v2_realtime do_stream; P3 Cartesia ink-2 do_stream. No unified session abstraction (D1), minimal parameter surface (D5). Closes #178 (research record attached there). --- rfc/0034-realtime-stt-followups.md | 195 +++++++++++++++++++++++++++++ 1 file changed, 195 insertions(+) create mode 100644 rfc/0034-realtime-stt-followups.md diff --git a/rfc/0034-realtime-stt-followups.md b/rfc/0034-realtime-stt-followups.md new file mode 100644 index 00000000..de2a81a4 --- /dev/null +++ b/rfc/0034-realtime-stt-followups.md @@ -0,0 +1,195 @@ +# RFC-0034: 实时转写收尾 —— WS 代理、ElevenLabs/Cartesia 实现 + +> **Status**: DRAFT(设计稿,待评审) +> **Date**: 2026-09-13 +> **Scope**: 完成 RFC-0028 明确遗留的三件事:WS 代理隧道(全局 `ProxyConfig` 对 WS 生效)、ElevenLabs `scribe_v2_realtime` 与 Cartesia `ink-2` 的 `do_stream` 独立实现 +> **Related**: [RFC-0028](0028-transcription-streaming.md)(本 RFC 是其遗留项的收尾)、[#178](https://github.com/arcships/aimux/issues/178)(跟踪 issue,含研究记录)、[#157](https://github.com/arcships/aimux/pull/157)(Go 会话生命周期先例) +> **Closes**: #178 + +--- + +## 1. 背景 + +RFC-0028 落地了 WS 基础设施 + OpenAI realtime 转写 + FFI 会话 + 8 语言绑定,但留了三个尾巴(见 #178 研究记录): + +1. **WS 完全绕过代理**。`ProxyConfig`(全局 `OnceLock`,FFI/Node/Python 均暴露 `init_proxy`)只驱动 reqwest;`ws.rs` 用裸 `connect_async` 直连。对必须经代理访问 provider API 的用户,实时转写不可用。 +2. **只有 OpenAI 一家实现 `do_stream`**。Cartesia 最尴尬:代码已按 `ink-2*` 门控且 `do_generate` 明确返回"仅支持 WebSocket 端点"([cartesia.rs](../aimux-providers/src/cartesia.rs) L618/L710),但 `do_stream` 不存在——两条路径都失败。ElevenLabs `scribe_v2_realtime` 零代码。 +3. 上游 tokio-tungstenite **明确不做代理支持**(tungstenite-rs#177),只能自己实现 CONNECT 隧道。 + +### 1.1 关键前置决策:不做统一抽象 + +**每家独立实现,不抽公共"WS 转写会话"层。** 依据(#178 研究记录中的对照): + +| | OpenAI | ElevenLabs | Cartesia | +|---|---|---|---| +| 配置载体 | 连接后 `session.update` 消息 | URL query 参数 | 连接后 `config` 消息 | +| 音频编码 | base64 进 JSON | base64 进 JSON(每帧带 `sample_rate`) | **原始二进制帧** | +| 转录事件语义 | 增量(delta,追加) | **全量替换**(partial 为当前完整文本) | turn 级事件 | +| 结束方式 | 收 `completed`,客户端 close | **无结束事件**,末帧 `commit:true` | 发 `close` 命令 | +| 触发转录 | 自动 | `commit_strategy`(manual/vad) | manual 模式需 `finalize` 命令 | + +真正公共的部分——连接、背压、abort/超时竞争、close——**已经在 `ws.rs` 的 `WsConnection` 里**,三家共用;表内每一行差异都发生在 provider 层,统一层只会把每行变成一个参数或钩子,抽象本身比任何单家实现更复杂。 + +**复查条款**:Cartesia 落地后(第三家)若重复代码确实可观,再基于两个真实样本评估抽取;本 RFC 不做。 + +## 2. Phase 1 — WS 代理隧道(`ws.rs` + `http.rs` 小改) + +### 2.1 行为定义 + +`ws_connect` 在发起连接前查询全局代理配置(复用 `http.rs` 的 `GLOBAL_PROXY`,新增只读访问器): + +1. **选代理**:`wss://` → `https_url`(`all_url` 兜底);`ws://` → `http_url`(`all_url` 兜底)。均无 → 直连(现状路径,零行为变化)。 +2. **no_proxy 匹配**:自实现,语义对齐 reqwest `NoProxy::from_string`(逗号分隔;后缀匹配;`*` 全匹配;带端口的条目要求端口相等)。命中 → 直连。 +3. **SOCKS 明确报错**:代理 URL scheme 为 `socks5`/`socks5h` 时返回 `AiMuxError::UnsupportedFunctionality("WebSocket proxy tunneling supports http/https proxies only, got socks5")`。**绝不静默直连**——静默直连等于绕过用户的网络边界。 +4. **代理 URL 带 userinfo**(如 `http://user:pass@proxy:8080`)→ CONNECT 请求附 `Proxy-Authorization: Basic base64(user:pass)`。 + +### 2.2 隧道流程(全部 await 点在 `select!` 内与 abort + `first_chunk_ms` 竞争,沿用 RFC-0028 §3.1 强制模式) + +``` +1. TcpStream::connect(proxy_host:proxy_port) +2. 写 CONNECT target_host:target_port HTTP/1.1\r\n + Host: target_host:target_port\r\n + [Proxy-Authorization: ...]\r\n \r\n +3. 读至 \r\n\r\n,校验状态行 200(非 200 → ApiCall,报代理状态码与 reason, + 不重试 CONNECT——认证类错误重试无意义) +4. 将该 TcpStream 交给 tokio_tungstenite::client_async_tls_with_config( + request, stream, None, Connector::Rustls(Arc)) + ——与直连的 connect_async 不同点仅在 TLS 由我们自备: + ClientConfig 用 webpki-roots,与 reqwest 的 rustls-tls-webpki-roots 对齐。 +``` + +- `WsConnection.stream` 类型不变(`WebSocketStream>`,`client_async_tls_with_config` 返回同型),对上层零感知。 +- `ws.rs` 头部文档的"**No proxy support**"段删除,替换为本节指针。 +- 依赖变更:workspace 已有 tokio-tungstenite 0.24;`aimux-provider-utils` 需显式引入 `rustls` + `webpki-roots`(版本对齐 tokio-tungstenite 0.24 传递的 rustls 0.23 系)。 + +### 2.3 测试 + +- 本地假 CONNECT 代理(`TcpListener` 手写:校验 CONNECT 目标行 → 200 → 透传到真实本地 WS server):断言 CONNECT 目标、握手成功、事件往返。 +- no_proxy 命中 → 断言未经过代理;`*` 通配;带端口条目。 +- SOCKS scheme → 明确错误;代理回 407 → ApiCall 且不可重试;CONNECT 阶段 abort → `Aborted`;代理不通 → 超时归入 `first_chunk_ms` 语义。 + +### 2.4 范围外 + +- SOCKS 隧道、PAC/autoproxy、per-request 代理(与 HTTP 侧一致,代理是全局配置)。 + +## 3. Phase 2 — ElevenLabs `scribe_v2_realtime` + +### 3.1 门控(对称于 OpenAI) + +`elevenlabs.rs` 新增 `is_realtime_transcription_model_id`:`scribe_v2_realtime` 前缀 → `do_stream`;其余 → `do_generate`。**`do_generate` 对 realtime ID 返回 `UnsupportedFunctionality`**——现状是把 `scribe_v2_realtime` 当批处理模型发到 `/v1/speech-to-text`,错误来自服务端且信息误导。 + +### 3.2 协议序列 + +``` +1. connect wss://{base}/v1/speech-to-text/realtime + ?model_id=scribe_v2_realtime + &audio_format=pcm_{rate} ← options.input_audio_format + &commit_strategy=manual ← 固定,不暴露(§3.4 D3) + [&language_code=… ← providerOptions.elevenlabs.languageCode + &include_timestamps=true] ← providerOptions.elevenlabs.includeTimestamps + headers: xi-api-key(沿用 build_headers) +2. 等 session_started(配置回显)→ 发 StreamStart +3. loop select!(audio | ws.next() | abort | timeout): + a. audio chunk → {"message_type":"input_audio_chunk", + "audio_base_64":…, "commit":false, + "sample_rate": rate} + 音频流结束(FFI input_done / 流 None)→ 不再发 chunk; + 直接进入收尾(§3.3) + b. 事件映射: + partial_transcript → TranscriptPartial ← 全量文本,替换语义 + committed_transcript → TranscriptFinal ← 可多次(多段) + warning → 挂入 warnings / Raw(include_raw_chunks) + error 类事件 → ApiCall;retryable 一行规则: + 名称为 rate_limited / queue_overflow / + resource_exhausted → true,其余 false +``` + +**参数面(消融后)**:`providerOptions.elevenlabs` 仅 `languageCode` 与 `includeTimestamps` 两个命名参数;keyterms / secondary_languages / VAD 调优 / `previousText` 等均不做——无需求来源,待有人要再加(加 = 追加 query 参数,零结构改动)。 + +### 3.3 终止(与 OpenAI 的关键差异) + +服务端**没有"完成"事件**。`commit_strategy` 固定为 manual(D3): + +1. 音频流结束 → 末尾补发一条 `input_audio_chunk`(空音频或最后真实 chunk 带 `commit:true`); +2. 等待最后的 `committed_transcript` 到达 → 发 `Finish`(segments 由历次 committed 拼装)→ **客户端主动 close(1000)**; +3. 边界:commit 后若在 `chunk_ms` 窗口内无新事件,视为服务端静默,发 `Finish` 并 close(空 Finish 优于挂死——RFC-0028 的"终止保险丝"原则同样适用)。 + +流中途(音频未结束)收到的 `committed_transcript` 照常发 `TranscriptFinal`,不发 Finish——Finish 只属于收尾。 + +### 3.4 决策 + +- **`commit_strategy` 固定 manual,不作为选项暴露**:manual 模式下 `partial_transcript` 本来就持续流出,流式体验不损失;而 vad 模式需要另一条 Finish 语义分支。`vad` 模式待真实需求出现再加。 + +### 3.5 测试 + +本地 WS mock:`session_started` → `partial×2` → `committed` → (commit) → `committed` → close,逐字段断言 query 参数、base64、`sample_rate`、commit 时机、事件序列、close code 1000、abort 中途取消、retryable 分类规则(retryable 三名 / 其余 false,各抽一个错误事件断言)。**live smoke 一次**(几秒 PCM,真实 key)——RFC-0028 D4 的教训:OpenAI 当年没跑真 API,wire 形状只被 mock 验证过。 + +## 4. Phase 3 — Cartesia `ink-2` + +门控已存在(`is_streaming_transcription_model_id`,L618),补 `do_stream`。结构与 §3 同型,差异点: + +``` +1. connect wss://{base}/… ← 见 Open Question 1(路径锁定) + headers: Authorization Bearer + Cartesia-Version(沿用 build_headers) +2. 发 config 文本帧 {type:"config", model, encoding, sample_rate, + [language]} ← 参数面最小化,见下 +3. loop select!: + a. audio chunk → **Binary 帧**(不 base64、不 JSON) + 音频流结束 → 发 {"type":"close"} + b. 事件映射(turn 事件,auto-finalize 模式): + turn 中间态 → TranscriptPartial + turn 完成 → TranscriptFinal(带词级时间戳时填充 segments) + 收尾完成 → Finish → 客户端 close + error → ApiCall +``` + +**只做 auto-finalize 模式**(turn detection 自动分段,实时转写的默认形态)。manual finalize 是 push-to-talk 场景,aimux 当前没有这类调用方,不做;turn 阈值(end_threshold / end_timeout_ms 等)只在调优 turn 行为时有意义,不透传,用服务端默认值。两者待需求出现再加。 + +`do_generate` 中对 `providerOptions.cartesia.streaming` 的 Unsupported warning:ink-2 走 `do_stream` 后该选项无意义,`do_stream` 侧消费/忽略并在文档注明,warning 保留在批处理路径不动。 + +### 4.1 测试 + +同 §3.5 模式:mock 断言 Binary 帧、config/finalize/close 命令序列、turn 事件映射;live smoke 一次。 + +## 5. 实施计划 + +| 阶段 | 内容 | 依赖 | PR | +|------|------|------|----| +| P1 | WS 代理隧道 + 测试 | 无 | 独立(先行,三家受益) | +| P2 | ElevenLabs realtime 门控 + `do_stream` + mock 测试 + live smoke | 无(建议在 P1 后,便于 smoke 走代理验证) | 独立 | +| P3 | Cartesia `do_stream` + mock 测试 + live smoke | 无(同上) | 独立 | +| P4 | RFC-0028 文档更新:状态行加 follow-up 指针、§3.4"骨架同构,按需加"修正为"各家独立实现(本 RFC §1.1)"、§9.2/§9.4 关闭指向本 RFC、§9.5 挂 #167 | P1-P3 | 随 P3 或单独 docs PR | + +P2/P3 不依赖 P1,但排序在其后:live smoke 顺手验证代理路径。 + +## 6. Non-goals + +1. **不做统一 WS 转写会话抽象**(§1.1,P3 落地后复查)。 +2. **不做 xAI STT**——无公开端点;RFC-0028 §1.1 的提法来自 AI SDK 生态转述,待 API 公开再立项。 +3. **不做回调式 FFI**——拉取式是刻意设计(RFC-0028 §4.2);Node/Python 需要流式语法的在绑定层加 async-iterator 薄包装,零 wire 改动。 +4. **不做 WS 会话录制**(RFC-0023 覆盖)——若 #167 的传输层回放落地,WS 会话录制搭同一机制,单独实现不划算(交叉引用 #167)。 +5. **不做 WS 断线自动重连**——实时会话有状态(已发送的音频),静默重连会产生丢字或重复;断开即错误,由上层重建会话。 +6. **不做 SOCKS/PAC 代理**(§2.4)。 + +## 7. 风险 + +| 风险 | 等级 | 对策 | +|---|---|---| +| Cartesia turns API 文档在登录墙后,事件 schema 以 SDK 源码推断 | 中 | Open Question 1:实现时以官方 Python SDK 类型定义逐字段对齐,mock 测试断言 schema;schema 不符时报回本 RFC | +| ElevenLabs 无显式结束事件,Finish 时机是推断的协议边界 | 中 | §3.3 定死:commit → 最后 committed → Finish + close;`chunk_ms` 静默兜底;live smoke 重点验证此边界 | +| no_proxy 自实现与 reqwest 语义有细节差(端口/通配) | 低 | 单测直接对照 reqwest `NoProxy::from_string` 的行为用例;文档声明"语义对齐" | +| rustls ClientConfig 与 reqwest 侧 roots 不一致 | 低 | 锁同一 webpki-roots 版本;隧道内 TLS 由 tokio-tungstenite 握手 | +| 两家 API 均为新/实验性,事件形状可能变 | 中 | 事件映射集中在各自文件一处(OpenAI 先例);版本变化只动映射 | + +## 8. Open Questions + +1. **Cartesia turns WS 的准确路径与事件 schema**。旧 ink-whisper 时代为 `wss://api.cartesia.ai/stt/ws`;现 SDK 指向 turns 端点(docs 路径 `api-reference/stt/turns/websocket`,登录墙)。P3 动手前用官方 SDK 源码锁定 URL 与响应类型,结论记回本节。 +2. **WS connect 失败的重试**:不需要新设计——#164 后 Core 在 attempt 层重试 `do_stream`,连接失败大概率已被覆盖。P1 落一条测试断言验证(连接失败 → Core retry 重连),行为不符再补设计。 + +## 9. 决策记录 + +- **D1 不抽统一抽象**(§1.1):三家协议差异表为证;共享边界止于 `WsConnection`。 +- **D2 SOCKS 报错不直连**(§2.1.3):代理环境下静默直连 = 功能性错误(要么失败要么绕过网络边界),必须显式失败。 +- **D3 ElevenLabs 固定 manual commit**(§3.4):manual 下 partial 事件持续流出,流式体验无损;vad 是第二条 Finish 语义分支,无需求不做。 +- **D4 Cartesia 只做 auto-finalize**(§4):manual finalize 是 push-to-talk 场景,无调用方;turn 阈值不透传,用服务端默认。 +- **D5 参数面最小化**(§3.2/§4):ElevenLabs 仅 languageCode/includeTimestamps,Cartesia 仅 language;每个额外参数都要映射+文档+测试,没有需求来源的一律不加,追加成本为零结构改动。 From c058b84ac80039263a0b5ec1906c63f9233798ce Mon Sep 17 00:00:00 2001 From: eric8810 Date: Sun, 13 Sep 2026 14:07:33 +0800 Subject: [PATCH 2/5] feat(provider-utils): route WebSocket connections through the global proxy (RFC-0034 P1) ws_connect now consults the global ProxyConfig that already governs HTTP: wss uses https_url (all_url fallback), ws uses http_url (all_url fallback), no_proxy entries are honored with reqwest NoProxy semantics (comma-separated, suffix match, '*', port-specific entries), and matched targets keep the direct path unchanged. Tunneled connects: manual TCP to the proxy -> CONNECT (with Basic Proxy-Authorization from proxy-URL userinfo) -> 2xx validation -> WS handshake over the tunnel via client_async_tls_with_config (explicit ring provider + webpki-roots for wss targets, Plain for ws). SOCKS and https-scheme proxies fail loudly as UnsupportedFunctionality instead of silently bypassing the configured proxy; proxy CONNECT rejections surface as non-retryable ApiCall errors carrying the proxy status. The tunnel steps are bounded by first_chunk_ms and raced against abort by the existing ws_connect select. rustls/webpki-roots/base64 join the optional 'ws' feature only. --- aimux-provider-utils/Cargo.toml | 11 +- aimux-provider-utils/src/http.rs | 6 +- aimux-provider-utils/src/ws.rs | 291 ++++++++++++++- aimux-provider-utils/tests/ws_proxy_test.rs | 369 ++++++++++++++++++++ rfc/0034-realtime-stt-followups.md | 20 +- 5 files changed, 680 insertions(+), 17 deletions(-) create mode 100644 aimux-provider-utils/tests/ws_proxy_test.rs diff --git a/aimux-provider-utils/Cargo.toml b/aimux-provider-utils/Cargo.toml index 50f05eb8..29a6049f 100644 --- a/aimux-provider-utils/Cargo.toml +++ b/aimux-provider-utils/Cargo.toml @@ -27,14 +27,21 @@ chrono = "0.4" # WebSocket client (RFC-0028 realtime transcription). Optional: enabled via # the `ws` feature so consumers without realtime needs compile it out. +# rustls/webpki-roots build the TLS ClientConfig for the proxy-tunnel path +# (RFC-0034 §2); base64 encodes Proxy-Authorization credentials. tokio-tungstenite = { workspace = true, optional = true } +rustls = { version = "0.23", default-features = false, features = [ + "ring", +], optional = true } +webpki-roots = { version = "0.26", optional = true } +base64 = { version = "0.22", optional = true } [features] -ws = ["dep:tokio-tungstenite"] +ws = ["dep:tokio-tungstenite", "dep:rustls", "dep:webpki-roots", "dep:base64"] [dev-dependencies] aimux-core = { workspace = true } -tokio = { workspace = true, features = ["test-util", "macros", "rt", "time"] } +tokio = { workspace = true, features = ["test-util", "macros", "rt", "time", "net", "io-util"] } serial_test = { workspace = true } wiremock = "0.6" diff --git a/aimux-provider-utils/src/http.rs b/aimux-provider-utils/src/http.rs index ac419d78..76dc7550 100644 --- a/aimux-provider-utils/src/http.rs +++ b/aimux-provider-utils/src/http.rs @@ -55,7 +55,11 @@ pub fn init_proxy(config: ProxyConfig) -> bool { GLOBAL_PROXY.set(config).is_ok() } -fn global_proxy() -> ProxyConfig { +/// Return the process-wide proxy configuration (empty default when +/// `init_proxy` was never called). Read by both the HTTP client builder and +/// the WebSocket connect path (RFC-0034 §2), so `ws://`/`wss://` traffic +/// honors the same proxy/no_proxy settings as HTTP. +pub(crate) fn global_proxy() -> ProxyConfig { GLOBAL_PROXY.get().cloned().unwrap_or_default() } diff --git a/aimux-provider-utils/src/ws.rs b/aimux-provider-utils/src/ws.rs index c45d20ea..be5a4a93 100644 --- a/aimux-provider-utils/src/ws.rs +++ b/aimux-provider-utils/src/ws.rs @@ -5,7 +5,7 @@ //! realtime APIs are WebSocket-based (OpenAI `gpt-realtime-whisper` //! transcription today). //! -//! Design notes (RFC-0028 §3.1): +//! Design notes (RFC-0028 §3.1, RFC-0034 §2): //! - **Every await point is abort/timeout covered** — `connect`, `send`, and //! event receives all `select!` against the abort token and the timeout //! timers. This is the WS analogue of the HTTP API-call primitive and @@ -13,8 +13,13 @@ //! the loop alone does not cover the send path). //! - **Backpressure is socket-level**: tungstenite's `send().await` drives //! flush and pends while the socket write buffer is full. -//! - **No proxy support**: tokio-tungstenite has no proxy parameter; WS -//! connections are direct (see RFC-0028 §3.1 / Open Questions). +//! - **Proxy support (RFC-0034 §2)**: the global `ProxyConfig` that governs +//! HTTP also governs `ws://`/`wss://` — `wss` uses `https_url` (or +//! `all_url`), `ws` uses `http_url` (or `all_url`), `no_proxy` entries are +//! honored, and tunneled connections are established with a manual HTTP +//! CONNECT before the WebSocket handshake. tokio-tungstenite has no proxy +//! support (and will not add it), so the tunnel is ours. SOCKS proxies are +//! rejected loudly rather than silently bypassed. use std::future::pending; @@ -96,6 +101,13 @@ fn ws_error(url: &str, msg: impl std::fmt::Display) -> AiMuxError { enum ConnectError { Timeout, Tungstenite(tokio_tungstenite::tungstenite::Error), + /// The proxy rejected the CONNECT tunnel with a non-2xx status. + /// Carries the status and the proxy's response line so the surfaced + /// error says what the proxy said (407 auth, 403 policy, …). + ProxyRejected { + status: u16, + reason: String, + }, } // Rust 1.98 clippy: tungstenite::Error makes the Err variant ~136 bytes. @@ -104,6 +116,7 @@ enum ConnectError { #[allow(clippy::result_large_err)] async fn connect_with_timeout( request: tokio_tungstenite::tungstenite::http::Request<()>, + proxy: Option, timeout: Option, ) -> Result< ( @@ -112,14 +125,246 @@ async fn connect_with_timeout( ), ConnectError, > { - let fut = tokio_tungstenite::connect_async(request); + let fut = async { + let Some(proxy) = proxy else { + return tokio_tungstenite::connect_async(request) + .await + .map_err(ConnectError::Tungstenite); + }; + connect_through_proxy(request, &proxy).await + }; match timeout { Some(d) => match tokio::time::timeout(d, fut).await { - Ok(inner) => inner.map_err(ConnectError::Tungstenite), + Ok(inner) => inner, Err(_) => Err(ConnectError::Timeout), }, - None => fut.await.map_err(ConnectError::Tungstenite), + None => fut.await, + } +} + +/// A resolved proxy tunnel: where to TCP-connect and how to authenticate the +/// CONNECT request (RFC-0034 §2). Public fields are read by the proxy +/// integration tests only (`ws_proxy_test.rs`). +#[doc(hidden)] +#[derive(Debug)] +pub struct ProxyTunnel { + pub host: String, + pub port: u16, + /// `Proxy-Authorization` value (`Basic …`) when the proxy URL carried + /// userinfo. + pub authorization: Option, + /// The TARGET is `wss://` — the tunnel stream needs a TLS upgrade before + /// the WebSocket handshake. + pub target_tls: bool, + /// `host:port` of the WS target as it must appear in the CONNECT line. + pub target_authority: String, +} + +/// Test-only exposure of [`resolve_proxy`] (integration tests construct +/// arbitrary configs without the set-once global). +#[doc(hidden)] +#[allow(non_snake_case)] +pub fn ws__proxy_decision_for( + target: &url::Url, + config: &crate::http::ProxyConfig, +) -> Result, AiMuxError> { + resolve_proxy(target, config) +} + +/// Test-only exposure of [`no_proxy_matches`]. +#[doc(hidden)] +#[allow(non_snake_case)] +#[must_use] +pub fn ws__no_proxy_matches(no_proxy: &str, host: &str, port: u16) -> bool { + no_proxy_matches(no_proxy, host, port) +} + +/// Pick direct vs. tunneled for a WS target under the global proxy config. +fn resolve_proxy( + target: &url::Url, + config: &crate::http::ProxyConfig, +) -> Result, AiMuxError> { + let raw = match target.scheme() { + "wss" => config.https_url.clone().or_else(|| config.all_url.clone()), + "ws" => config.http_url.clone().or_else(|| config.all_url.clone()), + _ => None, + }; + let Some(raw) = raw else { + return Ok(None); + }; + let proxy = url::Url::parse(&raw) + .map_err(|e| AiMuxError::InvalidArgument(format!("invalid proxy URL {raw}: {e}")))?; + let scheme = proxy.scheme().to_ascii_lowercase(); + if scheme != "http" { + // SOCKS is untunnelable via CONNECT; `https` proxies need TLS to the + // proxy itself (TLS-in-TLS) which has no demonstrated need. Both fail + // loudly — never silently bypass a configured proxy (RFC-0034 D2). + return Err(AiMuxError::UnsupportedFunctionality(format!( + "WebSocket proxy tunneling supports http proxies only, got {scheme:?} ({raw}); \ + refusing to bypass the configured proxy with a direct connection" + ))); + } + let host = proxy + .host_str() + .ok_or_else(|| AiMuxError::InvalidArgument(format!("proxy URL has no host: {raw}")))? + .to_string(); + let port = proxy.port().unwrap_or(80); + let authorization = if proxy.username().is_empty() && proxy.password().is_none() { + None + } else { + let credentials = format!("{}:{}", proxy.username(), proxy.password().unwrap_or("")); + use base64::Engine as _; + Some(format!( + "Basic {}", + base64::engine::general_purpose::STANDARD.encode(credentials) + )) + }; + let target_host = target.host_str().unwrap_or_default(); + let target_port = target + .port_or_known_default() + .unwrap_or(if target.scheme() == "wss" { 443 } else { 80 }); + Ok(Some(ProxyTunnel { + host, + port, + authorization, + target_tls: target.scheme() == "wss", + target_authority: format!("{target_host}:{target_port}"), + })) +} + +/// `no_proxy` matching aligned with reqwest's `NoProxy::from_string` +/// semantics (RFC-0034 §2.1): comma-separated entries; `*` matches +/// everything; an entry matches by exact host or dot-suffix; an entry with +/// an explicit `:port` additionally requires the port to match. IPv6 +/// literals are matched as raw strings (provider WS hosts are domains). +fn no_proxy_matches(no_proxy: &str, host: &str, port: u16) -> bool { + for entry in no_proxy.split(',') { + let entry = entry.trim(); + if entry.is_empty() { + continue; + } + if entry == "*" { + return true; + } + let (name, port_entry) = match entry.rsplit_once(':') { + Some((name, digits)) + if !digits.is_empty() && digits.bytes().all(|b| b.is_ascii_digit()) => + { + (name, Some(digits)) + } + _ => (entry, None), + }; + let name = name.trim_start_matches('.').to_ascii_lowercase(); + let host = host.to_ascii_lowercase(); + if (host == name || host.ends_with(&format!(".{name}"))) + && port_entry.is_none_or(|p| p.parse::() == Ok(port)) + { + return true; + } + } + false +} + +/// TCP-connect to the proxy, issue CONNECT, validate the 2xx response, then +/// run the WebSocket handshake (with TLS for `wss://` targets) over the +/// tunnel stream. Every step is bounded by the caller's timeout; abort is +/// enforced by `ws_connect`'s select dropping this future. +// Same cold-path rationale as `connect_with_timeout` above. +#[allow(clippy::result_large_err)] +async fn connect_through_proxy( + request: tokio_tungstenite::tungstenite::http::Request<()>, + proxy: &ProxyTunnel, +) -> Result< + ( + WebSocketStream>, + tokio_tungstenite::tungstenite::http::Response>>, + ), + ConnectError, +> { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + let mut stream = tokio::net::TcpStream::connect((proxy.host.as_str(), proxy.port)) + .await + .map_err(|e| ConnectError::Tungstenite(tokio_tungstenite::tungstenite::Error::Io(e)))?; + + let mut connect_request = format!( + "CONNECT {} HTTP/1.1\r\nHost: {}\r\n", + proxy.target_authority, proxy.target_authority + ); + if let Some(authorization) = &proxy.authorization { + connect_request.push_str(&format!("Proxy-Authorization: {authorization}\r\n")); } + connect_request.push_str("\r\n"); + stream + .write_all(connect_request.as_bytes()) + .await + .map_err(|e| ConnectError::Tungstenite(tokio_tungstenite::tungstenite::Error::Io(e)))?; + + // Read until end of headers, bounded (a hostile proxy must not be able to + // feed us an unbounded "response"). + let mut response = Vec::with_capacity(256); + let mut chunk = [0u8; 512]; + while !response.windows(4).any(|w| w == b"\r\n\r\n") { + let n = stream + .read(&mut chunk) + .await + .map_err(|e| ConnectError::Tungstenite(tokio_tungstenite::tungstenite::Error::Io(e)))?; + if n == 0 { + return Err(ConnectError::ProxyRejected { + status: 0, + reason: "proxy closed the connection before responding to CONNECT".into(), + }); + } + response.extend_from_slice(&chunk[..n]); + if response.len() > 8 * 1024 { + return Err(ConnectError::ProxyRejected { + status: 0, + reason: "proxy CONNECT response headers exceed 8 KiB".into(), + }); + } + } + let response_text = String::from_utf8_lossy(&response); + let status_line = response_text.lines().next().unwrap_or_default(); + // "HTTP/1.1 200 Connection established" → parse the middle token. + let status = status_line + .split_whitespace() + .nth(1) + .and_then(|code| code.parse::().ok()) + .unwrap_or(0); + if !(200..300).contains(&status) { + return Err(ConnectError::ProxyRejected { + status, + reason: status_line.to_string(), + }); + } + + // TLS upgrade for wss targets: same roots as the reqwest client + // (webpki-roots), explicit ring provider so a build where multiple + // rustls CryptoProvider features unify cannot panic on the implicit + // default. + let connector = if proxy.target_tls { + let provider = std::sync::Arc::new(rustls::crypto::ring::default_provider()); + let mut roots = rustls::RootCertStore::empty(); + roots.extend(webpki_roots::TLS_SERVER_ROOTS.iter().cloned()); + let config = rustls::ClientConfig::builder_with_provider(provider) + .with_safe_default_protocol_versions() + .map_err(|e| { + ConnectError::Tungstenite(tokio_tungstenite::tungstenite::Error::Io( + std::io::Error::other(format!("building rustls client config: {e}")), + )) + })? + .with_root_certificates(roots) + .with_no_client_auth(); + tokio_tungstenite::Connector::Rustls(std::sync::Arc::new(config)) + } else { + tokio_tungstenite::Connector::Plain + }; + + let (websocket, response) = + tokio_tungstenite::client_async_tls_with_config(request, stream, None, Some(connector)) + .await + .map_err(ConnectError::Tungstenite)?; + Ok((websocket, response)) } /// Open a WebSocket connection. The connect phase races abort and the @@ -165,16 +410,48 @@ pub async fn ws_connect(req: &WebSocketRequest) -> Result { return Err(abort_error(&req.abort_signal)); } - res = connect_with_timeout(http_req, connect_deadline.map(|d| d - tokio::time::Instant::now())) => match res { + res = connect_with_timeout(http_req, proxy, connect_deadline.map(|d| d - tokio::time::Instant::now())) => match res { Ok(v) => v, Err(ConnectError::Timeout) => { return Err(AiMuxError::Timeout("websocket connect timed out".into())); } + Err(ConnectError::ProxyRejected { status, reason }) => { + // CONNECT rejections are auth/policy verdicts from the proxy: + // retrying the same tunnel cannot change the answer. + return Err(AiMuxError::ApiCall(Box::new(ApiCallError { + status_code: (status != 0).then_some(status), + is_retryable: false, + ..ApiCallError::new( + format!("proxy rejected websocket CONNECT: {reason}"), + crate::http::sanitized_request_url(&req.url), + serde_json::json!({}), + ) + }))); + } // An HTTP handshake rejection carries a real status: keep it and // classify retryability by the shared rule instead of the blanket // transport `is_retryable = true` (401/403 must not be retried). diff --git a/aimux-provider-utils/tests/ws_proxy_test.rs b/aimux-provider-utils/tests/ws_proxy_test.rs new file mode 100644 index 00000000..81f7d29e --- /dev/null +++ b/aimux-provider-utils/tests/ws_proxy_test.rs @@ -0,0 +1,369 @@ +//! RFC-0034 §2: the global `ProxyConfig` governs WebSocket connections. +//! +//! Integration tests run a real local WS server and a fake CONNECT proxy on +//! a dedicated runtime thread (the fixture must outlive every `#[tokio::test]` +//! runtime). Unit tests cover proxy selection and `no_proxy` matching. + +#![cfg(feature = "ws")] + +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex, OnceLock}; + +use aimux_core::options::TimeoutConfiguration; +use aimux_provider_utils::http::ProxyConfig; +use aimux_provider_utils::ws::{WebSocketRequest, ws_connect}; +use futures::{SinkExt, StreamExt}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::{TcpListener, TcpStream}; +use tokio_tungstenite::tungstenite::Message; + +// ── Fixture: one fake proxy + two real WS servers on a parked runtime ──────── + +struct Fixture { + /// WS server reached THROUGH the proxy (its port is not in no_proxy). + tunnel_port: u16, + /// WS server reached DIRECTLY (its port is a port-specific no_proxy entry). + direct_port: u16, + connect_authorities: Arc>>, + proxy_connections: Arc, +} + +static FIXTURE: OnceLock = OnceLock::new(); + +fn fixture() -> &'static Fixture { + FIXTURE.get_or_init(|| { + let connect_authorities: Arc>> = Arc::new(Mutex::new(Vec::new())); + let proxy_connections = Arc::new(AtomicUsize::new(0)); + + let (ready_tx, ready_rx) = std::sync::mpsc::channel::<(u16, u16, u16)>(); + let authorities_for_thread = Arc::clone(&connect_authorities); + let conns_for_thread = Arc::clone(&proxy_connections); + std::thread::spawn(move || { + let rt = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("fixture runtime"); + rt.block_on(async move { + let tunnel_port = spawn_echo_server().await; + let direct_port = spawn_echo_server().await; + let proxy_port = spawn_fake_proxy( + Arc::clone(&authorities_for_thread), + Arc::clone(&conns_for_thread), + ) + .await; + ready_tx + .send((proxy_port, tunnel_port, direct_port)) + .expect("test process gone"); + // Park this runtime forever: the listener tasks live on it. + std::future::pending::<()>().await; + }); + }); + let (proxy_port, tunnel_port, direct_port) = ready_rx.recv().expect("fixture thread died"); + + // One process-wide proxy config (init_proxy is a set-once global): + // everything tunnels through the fake proxy except the direct + // server's port, which is covered by a port-specific no_proxy entry. + let configured = aimux_provider_utils::http::init_proxy(ProxyConfig { + http_url: Some(format!("http://127.0.0.1:{proxy_port}")), + https_url: None, + all_url: None, + // Port-specific entry: also exercises the port-aware matching. + no_proxy: Some(format!("127.0.0.1:{direct_port}")), + }); + assert!(configured, "init_proxy must win (no earlier test set it)"); + + Fixture { + tunnel_port, + direct_port, + connect_authorities, + proxy_connections, + } + }) +} + +async fn spawn_echo_server() -> u16 { + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("bind echo server"); + let port = listener.local_addr().expect("local addr").port(); + tokio::spawn(async move { + loop { + let Ok((stream, _)) = listener.accept().await else { + continue; + }; + tokio::spawn(async move { + if let Ok(ws) = tokio_tungstenite::accept_async(stream).await { + let (mut sink, mut source) = ws.split(); + while let Some(Ok(Message::Text(text))) = source.next().await { + if sink.send(Message::Text(text)).await.is_err() { + break; + } + } + } + }); + } + }); + port +} + +/// Minimal CONNECT proxy: records the CONNECT authority, 407s hosts starting +/// with `reject.`, otherwise bridges to the real target. +async fn spawn_fake_proxy( + connect_authorities: Arc>>, + connections: Arc, +) -> u16 { + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("bind fake proxy"); + let port = listener.local_addr().expect("local addr").port(); + tokio::spawn(async move { + loop { + let Ok((mut stream, _)) = listener.accept().await else { + continue; + }; + let connect_authorities = Arc::clone(&connect_authorities); + let connections = Arc::clone(&connections); + tokio::spawn(async move { + connections.fetch_add(1, Ordering::SeqCst); + let mut buffer = Vec::new(); + let mut chunk = [0u8; 512]; + while !buffer.windows(4).any(|w| w == b"\r\n\r\n") { + match stream.read(&mut chunk).await { + Ok(0) | Err(_) => return, + Ok(n) => buffer.extend_from_slice(&chunk[..n]), + } + } + let request = String::from_utf8_lossy(&buffer); + let authority = request + .lines() + .next() + .and_then(|line| line.split_whitespace().nth(1)) + .unwrap_or_default() + .to_string(); + connect_authorities + .lock() + .expect("connect log mutex") + .push(authority.clone()); + if authority.starts_with("reject.") { + let _ = stream + .write_all(b"HTTP/1.1 407 Proxy Authentication Required\r\n\r\n") + .await; + return; + } + if stream + .write_all(b"HTTP/1.1 200 Connection established\r\n\r\n") + .await + .is_err() + { + return; + } + let Ok(mut upstream) = TcpStream::connect(authority.as_str()).await else { + return; + }; + let _ = tokio::io::copy_bidirectional(&mut stream, &mut upstream).await; + }); + } + }); + port +} + +fn ws_request(url: String) -> WebSocketRequest { + WebSocketRequest { + url, + headers: Vec::new(), + subprotocols: Vec::new(), + abort_signal: None, + timeout: Some(TimeoutConfiguration { + first_chunk_ms: Some(5_000), + chunk_ms: Some(5_000), + step_ms: None, + total_ms: Some(10_000), + }), + } +} + +// ── Integration ────────────────────────────────────────────────────────────── + +#[tokio::test] +#[serial_test::serial] +async fn ws_tunnels_through_connect_proxy() { + let fixture = fixture(); + let mut connection = ws_connect(&ws_request(format!( + "ws://127.0.0.1:{}", + fixture.tunnel_port + ))) + .await + .expect("tunneled connect"); + connection.send_text("ping").await.expect("send via tunnel"); + match connection.next().await { + Some(Ok(aimux_provider_utils::ws::WsMessage::Text(text))) => assert_eq!(text, "ping"), + other => panic!("expected echo through tunnel, got {other:?}"), + } + connection.close().await; + + let authorities = fixture + .connect_authorities + .lock() + .expect("connect log mutex") + .clone(); + assert!( + authorities.contains(&format!("127.0.0.1:{}", fixture.tunnel_port)), + "proxy must see the target authority, saw {authorities:?}" + ); + assert!(fixture.proxy_connections.load(Ordering::SeqCst) >= 1); +} + +#[tokio::test] +#[serial_test::serial] +async fn no_proxy_entry_connects_directly() { + let fixture = fixture(); + let before = fixture.proxy_connections.load(Ordering::SeqCst); + let mut connection = ws_connect(&ws_request(format!( + "ws://127.0.0.1:{}", + fixture.direct_port + ))) + .await + .expect("direct connect (port-specific no_proxy entry)"); + connection.send_text("direct").await.expect("send direct"); + match connection.next().await { + Some(Ok(aimux_provider_utils::ws::WsMessage::Text(text))) => assert_eq!(text, "direct"), + other => panic!("expected direct echo, got {other:?}"), + } + connection.close().await; + assert_eq!( + fixture.proxy_connections.load(Ordering::SeqCst), + before, + "no_proxy-matched target must not touch the proxy" + ); +} + +#[tokio::test] +#[serial_test::serial] +async fn proxy_rejection_surfaces_as_non_retryable_api_call() { + let fixture = fixture(); + let error = match ws_connect(&ws_request(format!( + "ws://reject.local:{}", + fixture.tunnel_port + ))) + .await + { + Err(error) => error, + Ok(_) => panic!("proxy answers 407, connect must fail"), + }; + match error { + aimux_core::AiMuxError::ApiCall(api_call) => { + assert_eq!(api_call.status_code, Some(407)); + assert!(!api_call.is_retryable, "CONNECT verdicts must not retry"); + } + other => panic!("expected ApiCall, got {other:?}"), + } +} + +// ── Unit: proxy selection ──────────────────────────────────────────────────── + +fn config_with( + https_url: Option<&str>, + http_url: Option<&str>, + all_url: Option<&str>, +) -> ProxyConfig { + ProxyConfig { + http_url: http_url.map(String::from), + https_url: https_url.map(String::from), + all_url: all_url.map(String::from), + no_proxy: None, + } +} + +#[test] +fn no_proxy_config_means_direct() { + let target = url::Url::parse("wss://api.example.test").unwrap(); + assert!( + aimux_provider_utils::ws::ws__proxy_decision_for(&target, &ProxyConfig::default()) + .expect("resolve") + .is_none() + ); +} + +#[test] +fn wss_uses_https_url_ws_uses_http_url() { + let config = config_with(Some("http://p:8080"), Some("http://h:3128"), None); + let wss = url::Url::parse("wss://api.example.test").unwrap(); + let tunnel = aimux_provider_utils::ws::ws__proxy_decision_for(&wss, &config) + .expect("resolve") + .expect("tunnel"); + assert_eq!(tunnel.host, "p"); + assert_eq!(tunnel.port, 8080); + assert!(tunnel.target_tls); + assert_eq!(tunnel.target_authority, "api.example.test:443"); + + let ws = url::Url::parse("ws://api.example.test:9000").unwrap(); + let tunnel = aimux_provider_utils::ws::ws__proxy_decision_for(&ws, &config) + .expect("resolve") + .expect("tunnel"); + assert_eq!(tunnel.host, "h"); + assert_eq!(tunnel.port, 3128); + assert!(!tunnel.target_tls); + assert_eq!(tunnel.target_authority, "api.example.test:9000"); +} + +#[test] +fn all_url_is_the_fallback_for_both_schemes() { + let config = config_with(None, None, Some("http://a:1")); + for scheme in ["wss", "ws"] { + let target = url::Url::parse(&format!("{scheme}://api.example.test")).unwrap(); + let tunnel = aimux_provider_utils::ws::ws__proxy_decision_for(&target, &config) + .expect("resolve") + .expect("tunnel"); + assert_eq!(tunnel.host, "a"); + } +} + +#[test] +fn proxy_port_defaults_to_80() { + let config = config_with(Some("http://p"), None, None); + let target = url::Url::parse("wss://api.example.test").unwrap(); + let tunnel = aimux_provider_utils::ws::ws__proxy_decision_for(&target, &config) + .expect("resolve") + .expect("tunnel"); + assert_eq!(tunnel.port, 80); +} + +#[test] +fn socks_and_https_proxies_fail_loudly() { + for scheme in ["socks5", "socks5h", "https"] { + let config = config_with(Some(&format!("{scheme}://p:1080")), None, None); + let target = url::Url::parse("wss://api.example.test").unwrap(); + let error = aimux_provider_utils::ws::ws__proxy_decision_for(&target, &config) + .expect_err("must refuse"); + assert!( + matches!(error, aimux_core::AiMuxError::UnsupportedFunctionality(_)), + "{scheme} must be refused, got {error:?}" + ); + } +} + +#[test] +fn proxy_userinfo_becomes_basic_authorization() { + let config = config_with(Some("http://user:pass@p:8080"), None, None); + let target = url::Url::parse("wss://api.example.test").unwrap(); + let tunnel = aimux_provider_utils::ws::ws__proxy_decision_for(&target, &config) + .expect("resolve") + .expect("tunnel"); + assert_eq!(tunnel.authorization.as_deref(), Some("Basic dXNlcjpwYXNz")); +} + +// ── Unit: no_proxy matching (reqwest NoProxy semantics) ───────────────────── + +#[test] +fn no_proxy_matching_rules() { + use aimux_provider_utils::ws::ws__no_proxy_matches as matches; + assert!(matches("*", "anything.test", 443)); + assert!(matches("api.example.test", "api.example.test", 443)); + assert!(matches("example.test", "api.example.test", 443)); + assert!(matches(".example.test", "api.example.test", 443)); + assert!(!matches("example.test", "notexample.test", 443)); + assert!(matches("api.example.test:9000", "api.example.test", 9000)); + assert!(!matches("api.example.test:9000", "api.example.test", 443)); + assert!(matches(" a.test , b.test", "b.test", 80)); + assert!(!matches("", "api.example.test", 443)); +} diff --git a/rfc/0034-realtime-stt-followups.md b/rfc/0034-realtime-stt-followups.md index de2a81a4..c0b7c9c0 100644 --- a/rfc/0034-realtime-stt-followups.md +++ b/rfc/0034-realtime-stt-followups.md @@ -40,7 +40,7 @@ RFC-0028 落地了 WS 基础设施 + OpenAI realtime 转写 + FFI 会话 + 8 语 1. **选代理**:`wss://` → `https_url`(`all_url` 兜底);`ws://` → `http_url`(`all_url` 兜底)。均无 → 直连(现状路径,零行为变化)。 2. **no_proxy 匹配**:自实现,语义对齐 reqwest `NoProxy::from_string`(逗号分隔;后缀匹配;`*` 全匹配;带端口的条目要求端口相等)。命中 → 直连。 -3. **SOCKS 明确报错**:代理 URL scheme 为 `socks5`/`socks5h` 时返回 `AiMuxError::UnsupportedFunctionality("WebSocket proxy tunneling supports http/https proxies only, got socks5")`。**绝不静默直连**——静默直连等于绕过用户的网络边界。 +3. **非 http scheme 一律明确报错**:代理 URL scheme 为 `socks5`/`socks5h`(CONNECT 隧道不通)或 `https`(需要 TLS-to-proxy,TLS 套 TLS,无需求来源)时,返回 `AiMuxError::UnsupportedFunctionality`,消息注明拒绝直连原因。**绝不静默直连**——静默直连等于绕过用户的网络边界。 4. **代理 URL 带 userinfo**(如 `http://user:pass@proxy:8080`)→ CONNECT 请求附 `Proxy-Authorization: Basic base64(user:pass)`。 ### 2.2 隧道流程(全部 await 点在 `select!` 内与 abort + `first_chunk_ms` 竞争,沿用 RFC-0028 §3.1 强制模式) @@ -50,12 +50,18 @@ RFC-0028 落地了 WS 基础设施 + OpenAI realtime 转写 + FFI 会话 + 8 语 2. 写 CONNECT target_host:target_port HTTP/1.1\r\n Host: target_host:target_port\r\n [Proxy-Authorization: ...]\r\n \r\n -3. 读至 \r\n\r\n,校验状态行 200(非 200 → ApiCall,报代理状态码与 reason, - 不重试 CONNECT——认证类错误重试无意义) +3. 读至 \r\n\r\n(上限 8 KiB),校验状态行 2xx(非 2xx → ApiCall,报代理状态码与 + status line,不可重试——认证/策略类判定重试无意义) 4. 将该 TcpStream 交给 tokio_tungstenite::client_async_tls_with_config( - request, stream, None, Connector::Rustls(Arc)) + request, stream, None, + Connector::Rustls(Arc) ← wss 目标 + Connector::Plain) ← ws 目标 ——与直连的 connect_async 不同点仅在 TLS 由我们自备: - ClientConfig 用 webpki-roots,与 reqwest 的 rustls-tls-webpki-roots 对齐。 + ClientConfig 显式 ring provider + webpki-roots(与 reqwest 侧 roots 对齐; + 显式 provider 避免多 CryptoProvider feature 合并时的隐式 default panic)。 + CONNECT 应答的残余字节:代理在客户端发出 WS 握手请求前没有任何合法的 + 下行数据,故读到应答头结束即把 socket 交给握手是安全的(残余只可能来自 + 不守规矩的代理,丢弃无害)。 ``` - `WsConnection.stream` 类型不变(`WebSocketStream>`,`client_async_tls_with_config` 返回同型),对上层零感知。 @@ -155,7 +161,7 @@ RFC-0028 落地了 WS 基础设施 + OpenAI realtime 转写 + FFI 会话 + 8 语 | 阶段 | 内容 | 依赖 | PR | |------|------|------|----| -| P1 | WS 代理隧道 + 测试 | 无 | 独立(先行,三家受益) | +| P1 | WS 代理隧道 + 测试 | 无 | ✅ #183(draft) | | P2 | ElevenLabs realtime 门控 + `do_stream` + mock 测试 + live smoke | 无(建议在 P1 后,便于 smoke 走代理验证) | 独立 | | P3 | Cartesia `do_stream` + mock 测试 + live smoke | 无(同上) | 独立 | | P4 | RFC-0028 文档更新:状态行加 follow-up 指针、§3.4"骨架同构,按需加"修正为"各家独立实现(本 RFC §1.1)"、§9.2/§9.4 关闭指向本 RFC、§9.5 挂 #167 | P1-P3 | 随 P3 或单独 docs PR | @@ -189,7 +195,7 @@ P2/P3 不依赖 P1,但排序在其后:live smoke 顺手验证代理路径。 ## 9. 决策记录 - **D1 不抽统一抽象**(§1.1):三家协议差异表为证;共享边界止于 `WsConnection`。 -- **D2 SOCKS 报错不直连**(§2.1.3):代理环境下静默直连 = 功能性错误(要么失败要么绕过网络边界),必须显式失败。 +- **D2 非 http 代理 scheme 报错不直连**(§2.1.3):SOCKS 隧道不通、https 代理需 TLS 套 TLS(无需求来源);代理环境下静默直连 = 功能性错误(要么失败要么绕过网络边界),必须显式失败。 - **D3 ElevenLabs 固定 manual commit**(§3.4):manual 下 partial 事件持续流出,流式体验无损;vad 是第二条 Finish 语义分支,无需求不做。 - **D4 Cartesia 只做 auto-finalize**(§4):manual finalize 是 push-to-talk 场景,无调用方;turn 阈值不透传,用服务端默认。 - **D5 参数面最小化**(§3.2/§4):ElevenLabs 仅 languageCode/includeTimestamps,Cartesia 仅 language;每个额外参数都要映射+文档+测试,没有需求来源的一律不加,追加成本为零结构改动。 From d593384b2180e35ea0e527c3a0716bd75f7acd3b Mon Sep 17 00:00:00 2001 From: eric8810 Date: Sun, 13 Sep 2026 14:29:06 +0800 Subject: [PATCH 3/5] fix(provider-utils): address WS proxy review findings (RFC-0034 P1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Independent review (PR #183) surfaced three defects and a test gap: - IPv6 proxy hosts never connected: url::Url::host_str returns bracketed literals and the socket resolver wants the bare address — strip them. - Proxy credentials leaked into error strings: all resolve_proxy error paths now mask userinfo (scheme://***@host), matching the existing request-URL sanitization. - CONNECT rejections were blanket non-retryable: classify by the shared is_retryable_status rule (407/403 stay terminal, 502/503/504 retry); EOF-before-response (status 0) is transient. - Missing promised tests added: abort during the CONNECT tunnel (against a black-hole proxy), unresponsive proxy tripping first_chunk_ms, 503 retryability, CONNECT wire shape (request line, Host, no stray Proxy-Authorization), IPv6 bracket stripping, credential masking. - no_proxy docstring now states the CIDR divergence; init_proxy documents that WS reads the config live. - RFC-0034: OQ2 verified and closed — stream_transcribe has no attempt retry and should not (audio input is consumed once; replayable sources would be new design). Status line and test list updated to match what landed. --- aimux-provider-utils/src/http.rs | 5 + aimux-provider-utils/src/ws.rs | 58 +++++-- aimux-provider-utils/tests/ws_proxy_test.rs | 172 +++++++++++++++++++- rfc/0034-realtime-stt-followups.md | 13 +- 4 files changed, 228 insertions(+), 20 deletions(-) diff --git a/aimux-provider-utils/src/http.rs b/aimux-provider-utils/src/http.rs index 76dc7550..397e78a7 100644 --- a/aimux-provider-utils/src/http.rs +++ b/aimux-provider-utils/src/http.rs @@ -51,6 +51,11 @@ static SHARED: OnceLock, Client>>> = On const SHARED_CLIENT_CAP: usize = 8; /// Set proxy configuration before the first HTTP operation. +/// +/// The HTTP client snapshots this config when it is first built; the +/// WebSocket connect path (RFC-0034 §2) re-reads it live on every connect, +/// so a late `init_proxy` can affect WS while HTTP keeps its first-built +/// client. Set it before any traffic either way. pub fn init_proxy(config: ProxyConfig) -> bool { GLOBAL_PROXY.set(config).is_ok() } diff --git a/aimux-provider-utils/src/ws.rs b/aimux-provider-utils/src/ws.rs index be5a4a93..2ad1c632 100644 --- a/aimux-provider-utils/src/ws.rs +++ b/aimux-provider-utils/src/ws.rs @@ -179,6 +179,19 @@ pub fn ws__no_proxy_matches(no_proxy: &str, host: &str, port: u16) -> bool { no_proxy_matches(no_proxy, host, port) } +/// Proxy URLs may carry credentials (`http://user:pass@proxy:8080`); error +/// strings must never echo them (they travel to FFI callers and logs). Mask +/// the userinfo portion the same way `sanitized_request_url` protects +/// request URLs. +fn sanitized_proxy_url(raw: &str) -> String { + if let Some((scheme, rest)) = raw.split_once("://") + && let Some((_userinfo, host_part)) = rest.split_once('@') + { + return format!("{scheme}://***@{host_part}"); + } + raw.to_string() +} + /// Pick direct vs. tunneled for a WS target under the global proxy config. fn resolve_proxy( target: &url::Url, @@ -192,21 +205,35 @@ fn resolve_proxy( let Some(raw) = raw else { return Ok(None); }; - let proxy = url::Url::parse(&raw) - .map_err(|e| AiMuxError::InvalidArgument(format!("invalid proxy URL {raw}: {e}")))?; + let proxy = url::Url::parse(&raw).map_err(|e| { + AiMuxError::InvalidArgument(format!( + "invalid proxy URL {}: {e}", + sanitized_proxy_url(&raw) + )) + })?; let scheme = proxy.scheme().to_ascii_lowercase(); if scheme != "http" { // SOCKS is untunnelable via CONNECT; `https` proxies need TLS to the // proxy itself (TLS-in-TLS) which has no demonstrated need. Both fail // loudly — never silently bypass a configured proxy (RFC-0034 D2). return Err(AiMuxError::UnsupportedFunctionality(format!( - "WebSocket proxy tunneling supports http proxies only, got {scheme:?} ({raw}); \ - refusing to bypass the configured proxy with a direct connection" + "WebSocket proxy tunneling supports http proxies only, got {scheme:?} ({}); \ + refusing to bypass the configured proxy with a direct connection", + sanitized_proxy_url(&raw) ))); } let host = proxy .host_str() - .ok_or_else(|| AiMuxError::InvalidArgument(format!("proxy URL has no host: {raw}")))? + .ok_or_else(|| { + AiMuxError::InvalidArgument(format!( + "proxy URL has no host: {}", + sanitized_proxy_url(&raw) + )) + })? + // `host_str` returns IPv6 literals WITH brackets; the socket resolver + // wants the bare address. + .trim_start_matches('[') + .trim_end_matches(']') .to_string(); let port = proxy.port().unwrap_or(80); let authorization = if proxy.username().is_empty() && proxy.password().is_none() { @@ -235,8 +262,12 @@ fn resolve_proxy( /// `no_proxy` matching aligned with reqwest's `NoProxy::from_string` /// semantics (RFC-0034 §2.1): comma-separated entries; `*` matches /// everything; an entry matches by exact host or dot-suffix; an entry with -/// an explicit `:port` additionally requires the port to match. IPv6 -/// literals are matched as raw strings (provider WS hosts are domains). +/// an explicit `:port` additionally requires the port to match. +/// +/// Known divergence: reqwest additionally supports IP/CIDR entries +/// (`10.0.0.0/8`); those are matched here as literal strings only (they will +/// not match a CIDR-style entry). Provider WS targets are DNS names, and a +/// non-matching entry means "tunnel through the proxy" — the safe direction. fn no_proxy_matches(no_proxy: &str, host: &str, port: u16) -> bool { for entry in no_proxy.split(',') { let entry = entry.trim(); @@ -440,11 +471,18 @@ pub async fn ws_connect(req: &WebSocketRequest) -> Result { - // CONNECT rejections are auth/policy verdicts from the proxy: - // retrying the same tunnel cannot change the answer. + // Classify by the shared rule like the handshake rejection + // below: 407/403 are auth/policy verdicts (never retried), + // but a proxy 502/503/504 is transient — same as HTTP. + // `status == 0` (EOF before responding / oversized headers) + // behaves like a dropped connection: transient. return Err(AiMuxError::ApiCall(Box::new(ApiCallError { status_code: (status != 0).then_some(status), - is_retryable: false, + is_retryable: if status == 0 { + true + } else { + aimux_core::error::is_retryable_status(status) + }, ..ApiCallError::new( format!("proxy rejected websocket CONNECT: {reason}"), crate::http::sanitized_request_url(&req.url), diff --git a/aimux-provider-utils/tests/ws_proxy_test.rs b/aimux-provider-utils/tests/ws_proxy_test.rs index 81f7d29e..e925c2a6 100644 --- a/aimux-provider-utils/tests/ws_proxy_test.rs +++ b/aimux-provider-utils/tests/ws_proxy_test.rs @@ -25,6 +25,8 @@ struct Fixture { /// WS server reached DIRECTLY (its port is a port-specific no_proxy entry). direct_port: u16, connect_authorities: Arc>>, + /// Full CONNECT request text (request line + headers), newest last. + connect_requests: Arc>>, proxy_connections: Arc, } @@ -33,10 +35,12 @@ static FIXTURE: OnceLock = OnceLock::new(); fn fixture() -> &'static Fixture { FIXTURE.get_or_init(|| { let connect_authorities: Arc>> = Arc::new(Mutex::new(Vec::new())); + let connect_requests: Arc>> = Arc::new(Mutex::new(Vec::new())); let proxy_connections = Arc::new(AtomicUsize::new(0)); let (ready_tx, ready_rx) = std::sync::mpsc::channel::<(u16, u16, u16)>(); let authorities_for_thread = Arc::clone(&connect_authorities); + let requests_for_thread = Arc::clone(&connect_requests); let conns_for_thread = Arc::clone(&proxy_connections); std::thread::spawn(move || { let rt = tokio::runtime::Builder::new_current_thread() @@ -48,6 +52,7 @@ fn fixture() -> &'static Fixture { let direct_port = spawn_echo_server().await; let proxy_port = spawn_fake_proxy( Arc::clone(&authorities_for_thread), + Arc::clone(&requests_for_thread), Arc::clone(&conns_for_thread), ) .await; @@ -76,6 +81,7 @@ fn fixture() -> &'static Fixture { tunnel_port, direct_port, connect_authorities, + connect_requests, proxy_connections, } }) @@ -106,10 +112,12 @@ async fn spawn_echo_server() -> u16 { port } -/// Minimal CONNECT proxy: records the CONNECT authority, 407s hosts starting -/// with `reject.`, otherwise bridges to the real target. +/// Minimal CONNECT proxy. Behavior by target-authority prefix: +/// `reject.` → 407, `busy.` → 503, `hang.` → accepts and never answers +/// (black hole); everything else → 200 + bridge to the real target. async fn spawn_fake_proxy( connect_authorities: Arc>>, + connect_requests: Arc>>, connections: Arc, ) -> u16 { let listener = TcpListener::bind("127.0.0.1:0") @@ -122,6 +130,7 @@ async fn spawn_fake_proxy( continue; }; let connect_authorities = Arc::clone(&connect_authorities); + let connect_requests = Arc::clone(&connect_requests); let connections = Arc::clone(&connections); tokio::spawn(async move { connections.fetch_add(1, Ordering::SeqCst); @@ -133,7 +142,7 @@ async fn spawn_fake_proxy( Ok(n) => buffer.extend_from_slice(&chunk[..n]), } } - let request = String::from_utf8_lossy(&buffer); + let request = String::from_utf8_lossy(&buffer).into_owned(); let authority = request .lines() .next() @@ -144,12 +153,26 @@ async fn spawn_fake_proxy( .lock() .expect("connect log mutex") .push(authority.clone()); + connect_requests + .lock() + .expect("connect request log mutex") + .push(request); if authority.starts_with("reject.") { let _ = stream .write_all(b"HTTP/1.1 407 Proxy Authentication Required\r\n\r\n") .await; return; } + if authority.starts_with("busy.") { + let _ = stream + .write_all(b"HTTP/1.1 503 Service Unavailable\r\n\r\n") + .await; + return; + } + if authority.starts_with("hang.") { + // Black hole: keep the connection open, never answer. + std::future::pending::<()>().await; + } if stream .write_all(b"HTTP/1.1 200 Connection established\r\n\r\n") .await @@ -253,12 +276,124 @@ async fn proxy_rejection_surfaces_as_non_retryable_api_call() { match error { aimux_core::AiMuxError::ApiCall(api_call) => { assert_eq!(api_call.status_code, Some(407)); - assert!(!api_call.is_retryable, "CONNECT verdicts must not retry"); + assert!( + !api_call.is_retryable, + "407 is an auth verdict, not transient" + ); } other => panic!("expected ApiCall, got {other:?}"), } } +#[tokio::test] +#[serial_test::serial] +async fn transient_proxy_503_is_retryable() { + let fixture = fixture(); + let error = match ws_connect(&ws_request(format!( + "ws://busy.local:{}", + fixture.tunnel_port + ))) + .await + { + Err(error) => error, + Ok(_) => panic!("proxy answers 503, connect must fail"), + }; + match error { + aimux_core::AiMuxError::ApiCall(api_call) => { + assert_eq!(api_call.status_code, Some(503)); + assert!( + api_call.is_retryable, + "proxy 503 is transient — same rule as HTTP" + ); + } + other => panic!("expected ApiCall, got {other:?}"), + } +} + +#[tokio::test] +#[serial_test::serial] +async fn unresponsive_proxy_times_out_under_first_chunk_budget() { + let fixture = fixture(); + let mut request = ws_request(format!("ws://hang.local:{}", fixture.tunnel_port)); + request.timeout = Some(TimeoutConfiguration { + first_chunk_ms: Some(300), + chunk_ms: None, + step_ms: None, + total_ms: None, + }); + let error = match ws_connect(&request).await { + Err(error) => error, + Ok(_) => panic!("proxy never answers, connect must time out"), + }; + assert!( + matches!(error, aimux_core::AiMuxError::Timeout(_)), + "expected Timeout, got {error:?}" + ); +} + +#[tokio::test] +#[serial_test::serial] +async fn abort_during_connect_tunnel_surfaces_as_aborted() { + let fixture = fixture(); + let abort = aimux_core::AbortSignal::new(); + let fire = abort.clone(); + tokio::spawn(async move { + tokio::time::sleep(std::time::Duration::from_millis(80)).await; + fire.abort(); + }); + let mut request = ws_request(format!("ws://hang.local:{}", fixture.tunnel_port)); + request.abort_signal = Some(abort); + // Generous timeout: abort must win the race, not the timer. + let error = match ws_connect(&request).await { + Err(error) => error, + Ok(_) => panic!("aborted connect must not succeed"), + }; + assert!( + matches!(error, aimux_core::AiMuxError::Aborted(_)), + "expected Aborted, got {error:?}" + ); +} + +#[tokio::test] +#[serial_test::serial] +async fn connect_request_wire_shape() { + let fixture = fixture(); + let before = fixture + .connect_requests + .lock() + .expect("connect request log mutex") + .len(); + let mut connection = ws_connect(&ws_request(format!( + "ws://127.0.0.1:{}", + fixture.tunnel_port + ))) + .await + .expect("tunneled connect"); + connection.close().await; + let requests = fixture + .connect_requests + .lock() + .expect("connect request log mutex") + .clone(); + let request = &requests[before..] + .last() + .expect("this test's CONNECT request must be recorded"); + let lines: Vec<&str> = request.lines().collect(); + assert_eq!( + lines[0], + format!("CONNECT 127.0.0.1:{} HTTP/1.1", fixture.tunnel_port), + "CONNECT request line" + ); + assert!( + lines.contains(&format!("Host: 127.0.0.1:{}", fixture.tunnel_port).as_str()), + "Host header must repeat the authority, got {request:?}" + ); + assert!( + !request.to_ascii_lowercase().contains("proxy-authorization"), + "no credentials configured: no Proxy-Authorization expected, got {request:?}" + ); +} + // ── Unit: proxy selection ──────────────────────────────────────────────────── fn config_with( @@ -352,6 +487,35 @@ fn proxy_userinfo_becomes_basic_authorization() { assert_eq!(tunnel.authorization.as_deref(), Some("Basic dXNlcjpwYXNz")); } +#[test] +fn ipv6_proxy_host_brackets_are_stripped_for_the_socket() { + // `url::Url::host_str` returns "[::1]"; the resolver needs "::1". + let config = config_with(Some("http://[::1]:8080"), None, None); + let target = url::Url::parse("wss://api.example.test").unwrap(); + let tunnel = aimux_provider_utils::ws::ws__proxy_decision_for(&target, &config) + .expect("resolve") + .expect("tunnel"); + assert_eq!(tunnel.host, "::1"); + assert!(tunnel.target_tls); +} + +#[test] +fn proxy_errors_do_not_leak_credentials() { + let config = config_with(Some("socks5://user:secret@p:1080"), None, None); + let target = url::Url::parse("wss://api.example.test").unwrap(); + let error = aimux_provider_utils::ws::ws__proxy_decision_for(&target, &config) + .expect_err("must refuse"); + let text = error.to_string(); + assert!( + !text.contains("secret") && !text.contains("user:secret"), + "proxy credentials must be masked in errors, got: {text}" + ); + assert!( + text.contains("socks5://***@p:1080"), + "masked URL expected, got: {text}" + ); +} + // ── Unit: no_proxy matching (reqwest NoProxy semantics) ───────────────────── #[test] diff --git a/rfc/0034-realtime-stt-followups.md b/rfc/0034-realtime-stt-followups.md index c0b7c9c0..47c028d8 100644 --- a/rfc/0034-realtime-stt-followups.md +++ b/rfc/0034-realtime-stt-followups.md @@ -1,6 +1,6 @@ # RFC-0034: 实时转写收尾 —— WS 代理、ElevenLabs/Cartesia 实现 -> **Status**: DRAFT(设计稿,待评审) +> **Status**: P1 已实现(#183);P2/P3/P4 待做。设计稿其余部分待评审 > **Date**: 2026-09-13 > **Scope**: 完成 RFC-0028 明确遗留的三件事:WS 代理隧道(全局 `ProxyConfig` 对 WS 生效)、ElevenLabs `scribe_v2_realtime` 与 Cartesia `ink-2` 的 `do_stream` 独立实现 > **Related**: [RFC-0028](0028-transcription-streaming.md)(本 RFC 是其遗留项的收尾)、[#178](https://github.com/arcships/aimux/issues/178)(跟踪 issue,含研究记录)、[#157](https://github.com/arcships/aimux/pull/157)(Go 会话生命周期先例) @@ -51,7 +51,8 @@ RFC-0028 落地了 WS 基础设施 + OpenAI realtime 转写 + FFI 会话 + 8 语 Host: target_host:target_port\r\n [Proxy-Authorization: ...]\r\n \r\n 3. 读至 \r\n\r\n(上限 8 KiB),校验状态行 2xx(非 2xx → ApiCall,报代理状态码与 - status line,不可重试——认证/策略类判定重试无意义) + status line,按共享 `is_retryable_status` 规则分类——407/403 等认证/策略判定 + 不可重试,502/503/504 等瞬态可重试;无应答(EOF/超长, status 0)视为瞬态可重试) 4. 将该 TcpStream 交给 tokio_tungstenite::client_async_tls_with_config( request, stream, None, Connector::Rustls(Arc) ← wss 目标 @@ -66,13 +67,13 @@ RFC-0028 落地了 WS 基础设施 + OpenAI realtime 转写 + FFI 会话 + 8 语 - `WsConnection.stream` 类型不变(`WebSocketStream>`,`client_async_tls_with_config` 返回同型),对上层零感知。 - `ws.rs` 头部文档的"**No proxy support**"段删除,替换为本节指针。 -- 依赖变更:workspace 已有 tokio-tungstenite 0.24;`aimux-provider-utils` 需显式引入 `rustls` + `webpki-roots`(版本对齐 tokio-tungstenite 0.24 传递的 rustls 0.23 系)。 +- 依赖变更:workspace 已有 tokio-tungstenite 0.24;`aimux-provider-utils` 需显式引入 `rustls` + `webpki-roots` + `base64`(Proxy-Authorization 编码;版本对齐 tokio-tungstenite 0.24 传递的 rustls 0.23 系)。 ### 2.3 测试 - 本地假 CONNECT 代理(`TcpListener` 手写:校验 CONNECT 目标行 → 200 → 透传到真实本地 WS server):断言 CONNECT 目标、握手成功、事件往返。 - no_proxy 命中 → 断言未经过代理;`*` 通配;带端口条目。 -- SOCKS scheme → 明确错误;代理回 407 → ApiCall 且不可重试;CONNECT 阶段 abort → `Aborted`;代理不通 → 超时归入 `first_chunk_ms` 语义。 +- SOCKS scheme → 明确错误;代理回 407 → ApiCall 且不可重试、503 → 可重试;CONNECT 阶段 abort → `Aborted`;代理不通 → 超时归入 `first_chunk_ms` 语义。(P1 全部落地于 `ws_proxy_test.rs`;另钉住 CONNECT 请求行/Host 头/无凭据时无 Proxy-Authorization 的 wire 形状、IPv6 代理 host 去方括号、错误信息脱敏。wss 隧道的 rustls 分支仅单元覆盖,端到端执行留待 P2 live smoke——本地无 TLS server 桩,注入根证书需要测试缝,不值当。) ### 2.4 范围外 @@ -183,14 +184,14 @@ P2/P3 不依赖 P1,但排序在其后:live smoke 顺手验证代理路径。 |---|---|---| | Cartesia turns API 文档在登录墙后,事件 schema 以 SDK 源码推断 | 中 | Open Question 1:实现时以官方 Python SDK 类型定义逐字段对齐,mock 测试断言 schema;schema 不符时报回本 RFC | | ElevenLabs 无显式结束事件,Finish 时机是推断的协议边界 | 中 | §3.3 定死:commit → 最后 committed → Finish + close;`chunk_ms` 静默兜底;live smoke 重点验证此边界 | -| no_proxy 自实现与 reqwest 语义有细节差(端口/通配) | 低 | 单测直接对照 reqwest `NoProxy::from_string` 的行为用例;文档声明"语义对齐" | +| no_proxy 自实现与 reqwest 语义有细节差 | 低 | reqwest 的匹配器非公开 API,无法程序化交叉验证——语义按其文档对齐并表格化单测;已知分歧(CIDR/IP 段条目仅按字面匹配,不展开网段)在 `ws.rs` docstring 与本表显式记录;非命中的安全方向是走代理 | | rustls ClientConfig 与 reqwest 侧 roots 不一致 | 低 | 锁同一 webpki-roots 版本;隧道内 TLS 由 tokio-tungstenite 握手 | | 两家 API 均为新/实验性,事件形状可能变 | 中 | 事件映射集中在各自文件一处(OpenAI 先例);版本变化只动映射 | ## 8. Open Questions 1. **Cartesia turns WS 的准确路径与事件 schema**。旧 ink-whisper 时代为 `wss://api.cartesia.ai/stt/ws`;现 SDK 指向 turns 端点(docs 路径 `api-reference/stt/turns/websocket`,登录墙)。P3 动手前用官方 SDK 源码锁定 URL 与响应类型,结论记回本节。 -2. **WS connect 失败的重试**:不需要新设计——#164 后 Core 在 attempt 层重试 `do_stream`,连接失败大概率已被覆盖。P1 落一条测试断言验证(连接失败 → Core retry 重连),行为不符再补设计。 +2. **WS connect 失败的重试**(P1 已验证并关闭):`stream_transcribe` 直通 `do_stream`,**没有任何 attempt 级重试**;且不重试是当前正确行为——音频输入流在首次尝试即被消费,重放需要可重播的音频源(HTTP 流可重试是因为请求体可克隆),非免费能力。连接失败的补救属于上层会话重建(Non-goal 5)。若未来要重试,需先设计可重播音频源,另立 RFC。 ## 9. 决策记录 From 04dcc0192858297bfeac44736205f5ad9d661d3a Mon Sep 17 00:00:00 2001 From: eric8810 Date: Sun, 13 Sep 2026 14:54:31 +0800 Subject: [PATCH 4/5] =?UTF-8?q?docs(rfc-0034):=20defer=20P3=20(Cartesia)?= =?UTF-8?q?=20=E2=80=94=20unverifiable=20wire,=20zero=20deferral=20cost?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Re-review after P1: the decisive difference vs P2 (ElevenLabs) is verifiability. ElevenLabs ships a public API reference, so mock tests pin documented facts; Cartesia's turns API is behind a login wall, its event schema is only inferable from the Python SDK, and the API is new enough to churn — on top of the D4 lesson that mock-only wire shapes can be wrong. Deferring loses nothing: ink-2 already fails honestly with UnsupportedFunctionality on both paths, and nothing depends on it. Restart triggers (any one): a user asks; the docs come out from behind the login wall; a key + smoke decision. Gate comment in cartesia.rs records the deferral. D6 added. --- rfc/0034-realtime-stt-followups.md | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/rfc/0034-realtime-stt-followups.md b/rfc/0034-realtime-stt-followups.md index 47c028d8..295aef5e 100644 --- a/rfc/0034-realtime-stt-followups.md +++ b/rfc/0034-realtime-stt-followups.md @@ -131,7 +131,9 @@ RFC-0028 落地了 WS 基础设施 + OpenAI realtime 转写 + FFI 会话 + 8 语 本地 WS mock:`session_started` → `partial×2` → `committed` → (commit) → `committed` → close,逐字段断言 query 参数、base64、`sample_rate`、commit 时机、事件序列、close code 1000、abort 中途取消、retryable 分类规则(retryable 三名 / 其余 false,各抽一个错误事件断言)。**live smoke 一次**(几秒 PCM,真实 key)——RFC-0028 D4 的教训:OpenAI 当年没跑真 API,wire 形状只被 mock 验证过。 -## 4. Phase 3 — Cartesia `ink-2` +## 4. Phase 3 — Cartesia `ink-2`(**暂缓**,D6) + +> 2026-09-13 复议后暂缓:官方文档在登录墙后、turns API 年轻易变、事件 schema 只能从 SDK 推断、无 key 可验证、无需求信号——三家里出错风险最高,而暂缓成本为零(现状对 ink-2 返回的 UnsupportedFunctionality 是诚实信息)。触发条件三选一即重启:有人提需求 / 文档公开出墙 / 拿到 key 决定跑 smoke。`cartesia.rs` 门控注释已标注 deferred。 门控已存在(`is_streaming_transcription_model_id`,L618),补 `do_stream`。结构与 §3 同型,差异点: @@ -164,7 +166,7 @@ RFC-0028 落地了 WS 基础设施 + OpenAI realtime 转写 + FFI 会话 + 8 语 |------|------|------|----| | P1 | WS 代理隧道 + 测试 | 无 | ✅ #183(draft) | | P2 | ElevenLabs realtime 门控 + `do_stream` + mock 测试 + live smoke | 无(建议在 P1 后,便于 smoke 走代理验证) | 独立 | -| P3 | Cartesia `do_stream` + mock 测试 + live smoke | 无(同上) | 独立 | +| P3 | Cartesia `do_stream` + mock 测试 + live smoke | 无(同上) | **暂缓**(2026-09-13 复议,触发条件见 §4;D6) | | P4 | RFC-0028 文档更新:状态行加 follow-up 指针、§3.4"骨架同构,按需加"修正为"各家独立实现(本 RFC §1.1)"、§9.2/§9.4 关闭指向本 RFC、§9.5 挂 #167 | P1-P3 | 随 P3 或单独 docs PR | P2/P3 不依赖 P1,但排序在其后:live smoke 顺手验证代理路径。 @@ -200,3 +202,4 @@ P2/P3 不依赖 P1,但排序在其后:live smoke 顺手验证代理路径。 - **D3 ElevenLabs 固定 manual commit**(§3.4):manual 下 partial 事件持续流出,流式体验无损;vad 是第二条 Finish 语义分支,无需求不做。 - **D4 Cartesia 只做 auto-finalize**(§4):manual finalize 是 push-to-talk 场景,无调用方;turn 阈值不透传,用服务端默认。 - **D5 参数面最小化**(§3.2/§4):ElevenLabs 仅 languageCode/includeTimestamps,Cartesia 仅 language;每个额外参数都要映射+文档+测试,没有需求来源的一律不加,追加成本为零结构改动。 +- **D6 Cartesia(P3)暂缓**(§4):与 P2 的决定性差异是可验证性——ElevenLabs 有公开 API 参考,mock 断言的是文档事实;Cartesia 文档在登录墙后、schema 靠 SDK 反推、API 新且易变,叠加 D4 教训(mock-only 验证的 wire 可能是错的),推断+无验证的组合风险不可接受。暂缓不损失任何东西:能力缺失的报错是诚实的,且无消费者。 From c8bfe99f5be72666cfa4b6fc2e901cd6df866e5d Mon Sep 17 00:00:00 2001 From: eric8810 Date: Sun, 13 Sep 2026 15:24:14 +0800 Subject: [PATCH 5/5] docs(rfc-0028): close out follow-ups; test(provider-utils): live wss smoke through the proxy tunnel MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P4 of RFC-0034 plus the executable closure of P1's untested wss branch: - RFC-0028: status line points at RFC-0034 for the follow-ups; §3.4's 'skeletons are isomorphic, add as needed' corrected to the actual outcome (per-provider implementations, no shared layer); Open Questions 2/4 resolved (WS proxy landed; ElevenLabs landed, Cartesia deferred with triggers, xAI dropped), 5 linked to #167's transport-level replay. - New #[ignore] smoke: wss://api.elevenlabs.io through the local CONNECT proxy — no key needed, the real server answering the handshake (HTTP rejection or in-session auth error) proves the tunnel + rustls webpki-roots path executed against a real public CA chain. Ran green; fixture now maps https_url to the fake proxy so wss targets tunnel. - Full transcript round-trip smoke still needs a provider key (D4 posture, flagged in code and RFC-0034). --- aimux-provider-utils/tests/ws_proxy_test.rs | 68 ++++++++++++++++++++- rfc/0028-transcription-streaming.md | 10 +-- 2 files changed, 72 insertions(+), 6 deletions(-) diff --git a/aimux-provider-utils/tests/ws_proxy_test.rs b/aimux-provider-utils/tests/ws_proxy_test.rs index e925c2a6..90e015d4 100644 --- a/aimux-provider-utils/tests/ws_proxy_test.rs +++ b/aimux-provider-utils/tests/ws_proxy_test.rs @@ -70,7 +70,10 @@ fn fixture() -> &'static Fixture { // server's port, which is covered by a port-specific no_proxy entry. let configured = aimux_provider_utils::http::init_proxy(ProxyConfig { http_url: Some(format!("http://127.0.0.1:{proxy_port}")), - https_url: None, + // wss targets tunnel too — the live smoke relies on it to drive + // the rustls branch through the CONNECT proxy against the real + // endpoint. + https_url: Some(format!("http://127.0.0.1:{proxy_port}")), all_url: None, // Port-specific entry: also exercises the port-aware matching. no_proxy: Some(format!("127.0.0.1:{direct_port}")), @@ -531,3 +534,66 @@ fn no_proxy_matching_rules() { assert!(matches(" a.test , b.test", "b.test", 80)); assert!(!matches("", "api.example.test", 443)); } + +// ── Live handshake smoke (manual: `cargo test -- --ignored`) ───────────────── +// +// No API key needed: the target is the REAL wss endpoint, so TLS through the +// CONNECT tunnel validates against a public CA chain and the server answers +// the WebSocket handshake itself (rejecting unauthenticated callers). Either +// outcome — an HTTP-status rejection from the handshake, or a connected +// session whose first event is an auth error/close — proves the full +// tunnel+rustls path executed against the real internet. A proxy/TLS/cert +// failure would surface as a transport error instead, which this test +// rejects. +#[tokio::test] +#[serial_test::serial] +#[ignore = "network-dependent manual smoke (RFC-0034 §2.3): exercises the real TLS+proxy path"] +async fn live_wss_handshake_through_connect_proxy() { + let fixture = fixture(); + let mut request = WebSocketRequest { + url: "wss://api.elevenlabs.io/v1/speech-to-text/realtime\ + ?model_id=scribe_v2_realtime&audio_format=pcm_16000&commit_strategy=manual" + .to_string(), + headers: Vec::new(), + subprotocols: Vec::new(), + abort_signal: None, + timeout: Some(TimeoutConfiguration { + first_chunk_ms: Some(10_000), + chunk_ms: Some(10_000), + step_ms: None, + total_ms: Some(20_000), + }), + }; + let _ = &mut request; + + let saw_server_response = match ws_connect(&request).await { + // Handshake rejected with a real HTTP status: tunnel + TLS + upgrade + // all executed; the server answered. + Err(aimux_core::AiMuxError::ApiCall(api_call)) => { + assert!( + api_call.status_code.is_some(), + "expected an HTTP status from the real endpoint, got {api_call:?}" + ); + true + } + // Connected: the server accepted the socket. Without a key the first + // inbound event must be an auth-flavored error or a close — anything + // the real server sent. A transport/TLS failure is Err here and fails + // the match. + Ok(mut connection) => matches!(connection.next().await, Some(Ok(_)) | Some(Err(_))), + Err(other) => panic!("tunnel/TLS failure against the real endpoint: {other:?}"), + }; + assert!( + saw_server_response, + "the real endpoint must answer the handshake" + ); + let authorities = fixture + .connect_authorities + .lock() + .expect("connect log mutex") + .clone(); + assert!( + authorities.contains(&"api.elevenlabs.io:443".to_string()), + "CONNECT must target the real host, saw {authorities:?}" + ); +} diff --git a/rfc/0028-transcription-streaming.md b/rfc/0028-transcription-streaming.md index adc3076b..4a29f96e 100644 --- a/rfc/0028-transcription-streaming.md +++ b/rfc/0028-transcription-streaming.md @@ -1,6 +1,6 @@ # RFC-0028: Transcription 流式支持(实时 STT) -> **Status**: 已实现(2026-08-14,三个 Phase 一次性落地;实现细节与本文差异见 §10) +> **Status**: 已实现(2026-08-14,三个 Phase 一次性落地;实现细节与本文差异见 §10)。遗留项(P1 期明确延后的部分)由 [RFC-0034](0034-realtime-stt-followups.md) 收尾:WS 代理已落地、ElevenLabs 已落地、Cartesia 暂缓(触发条件见其 §4/D6) > **Date**: 2026-08-14 > **Scope**: 为 `TranscriptionModel::do_stream` 落地完整实现 —— WebSocket 传输基础设施 + OpenAI realtime provider 实现(Rust 核心),以及后续 FFI 会话式 API + 8 语言绑定 > **Related**: [RFC-0008](0008-multimodal-bindings.md) §2.3 Mode C(当初 defer 的决策)、[#43](https://github.com/arcships/aimux/issues/43)、[RFC-0016](0016-align-with-aisdk.md) H1(abort 基础设施) @@ -158,7 +158,7 @@ pub async fn ws_connect(req: WebSocketRequest) -> Result