From a4427b3e075d12d27069cc071028355d549ebf17 Mon Sep 17 00:00:00 2001 From: eric8810 Date: Sun, 13 Sep 2026 13:53:35 +0800 Subject: [PATCH 1/6] =?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/6] 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/6] 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/6] =?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 31fae9ae5ed1cba318c56f48f586ad641ca3ff03 Mon Sep 17 00:00:00 2001 From: eric8810 Date: Sun, 13 Sep 2026 14:56:31 +0800 Subject: [PATCH 5/6] feat(providers): ElevenLabs scribe_v2_realtime streaming transcription (RFC-0034 P2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ElevenLabsTranscriptionModel gains do_stream behind the existing 'realtime' feature, gated on scribe_v2_realtime* (inverse gating in do_generate rejects realtime IDs with a clear message instead of posting them to the batch endpoint). Wire shape per the public API reference (2026-09), pinned field-by-field by local mock-server tests: - config travels on the URL query (model_id, audio_format pcm_{rate}/ulaw_{rate}, commit_strategy=manual, optional language_code/include_timestamps from providerOptions.elevenlabs); xi-api-key rides the WS handshake headers - audio: base64 in input_audio_chunk JSON with sample_rate; a hold-one-chunk pipeline puts commit:true on the LAST real chunk (an empty stream commits a single empty chunk); audio waits for session_started - partial_transcript -> TranscriptPartial (replace semantics), committed_transcript[_with_timestamps] -> TranscriptFinal + Finish (segments from words, spacing entries dropped), then client close(1000) - 14 documented error event names classified: rate_limited / queue_overflow / resource_exhausted retry, the rest terminal - the server never closes after commit, so a settle window (chunk_ms, 5s default) bounds the wait with an empty Finish — the stream always terminates - non-pcm/ulaw input formats fail fast without connecting Live-API smoke pending (no key at implementation time) — same posture as RFC-0028 D4; the do_stream doc comment says so. Also: cartesia.rs gate comment now records the P3 deferral (RFC-0034 D6); RFC status/plan rows updated. --- aimux-providers/src/cartesia.rs | 5 + aimux-providers/src/elevenlabs.rs | 397 +++++++++++++ .../tests/elevenlabs_realtime_stream_test.rs | 536 ++++++++++++++++++ rfc/0034-realtime-stt-followups.md | 4 +- 4 files changed, 940 insertions(+), 2 deletions(-) create mode 100644 aimux-providers/tests/elevenlabs_realtime_stream_test.rs diff --git a/aimux-providers/src/cartesia.rs b/aimux-providers/src/cartesia.rs index c57d9613..cc0bb87e 100644 --- a/aimux-providers/src/cartesia.rs +++ b/aimux-providers/src/cartesia.rs @@ -617,6 +617,11 @@ use serde::Deserialize; /// Streaming transcription model IDs start with `ink-2` and only support the /// WebSocket streaming endpoint, not the REST batch endpoint. +/// +/// `do_stream` for these IDs is intentionally NOT implemented — deferred +/// with explicit triggers (RFC-0034 §4/D6: docs behind a login wall, young +/// turns API, schema inferable from the SDK only). Until it lands, both +/// paths return `UnsupportedFunctionality`. fn is_streaming_transcription_model_id(model_id: &str) -> bool { model_id == "ink-2" || model_id.starts_with("ink-2-") } diff --git a/aimux-providers/src/elevenlabs.rs b/aimux-providers/src/elevenlabs.rs index 46f7176b..8ab7f5e5 100644 --- a/aimux-providers/src/elevenlabs.rs +++ b/aimux-providers/src/elevenlabs.rs @@ -621,6 +621,13 @@ fn audio_input_to_bytes_stt(audio: &AudioInput) -> Result, AiMuxError> { } } +/// Realtime streaming model IDs (`scribe_v2_realtime*`) use the WebSocket +/// realtime endpoint; every other model uses the batch REST endpoint. +/// Wire shape per the public API reference (2026-09, RFC-0034 §3). +fn is_realtime_transcription_model_id(model_id: &str) -> bool { + model_id == "scribe_v2_realtime" || model_id.starts_with("scribe_v2_realtime-") +} + #[async_trait] impl TranscriptionModel for ElevenLabsTranscriptionModel { fn provider(&self) -> &str { @@ -635,6 +642,14 @@ impl TranscriptionModel for ElevenLabsTranscriptionModel { &self, options: &TranscriptionCallOptions, ) -> Result { + if is_realtime_transcription_model_id(&self.model_id) { + return Err(AiMuxError::UnsupportedFunctionality(format!( + "non-streaming transcription is not supported by `{}` \ + (realtime models stream over a WebSocket session)", + self.model_id + ))); + } + let warnings: Vec = Vec::new(); let audio_bytes = audio_input_to_bytes_stt(&options.audio)?; @@ -735,4 +750,386 @@ impl TranscriptionModel for ElevenLabsTranscriptionModel { provider_metadata: None, }) } + /// Streaming transcription over the ElevenLabs realtime WebSocket + /// (RFC-0034 §3, wire shape per the public API reference 2026-09). + /// + /// Config travels on the URL (no `session.update` message); audio rides + /// base64 in `input_audio_chunk` JSON frames; `commit_strategy` is fixed + /// to `manual` (D3) so the stream ends when the caller's audio ends — + /// the final chunk carries `commit: true`, the resulting + /// `committed_transcript` is THE final, then the client closes (1000). + /// A settle window (`chunk_ms`, default 5s) bounds the wait for that + /// final event so the stream always terminates. + /// + /// Live-API smoke pending (no key at implementation time); the wire + /// shape is pinned field-by-field against the documented reference by + /// the local mock-server tests — same posture as RFC-0028 D4. + #[cfg(feature = "realtime")] + async fn do_stream( + &self, + options: aimux_core::transcription_model::TranscriptionStreamOptions, + ) -> Result { + use aimux_core::transcription_model::{TranscriptionStreamPart, TranscriptionStreamResult}; + use aimux_provider_utils::ws::{WebSocketRequest, WsMessage, ws_connect}; + use futures::StreamExt; + + if !is_realtime_transcription_model_id(&self.model_id) { + return Err(AiMuxError::UnsupportedFunctionality(format!( + "streaming transcription is not supported by `{}` \ + (realtime models such as scribe_v2_realtime only)", + self.model_id + ))); + } + + // Parameter surface is deliberately minimal (RFC-0034 D5): + // languageCode + includeTimestamps. Anything else waits for a user. + let mut language_code: Option = None; + let mut include_timestamps = false; + if let Some(ref po) = options.provider_options + && let Some(el) = po.get("elevenlabs") + { + if let Some(v) = el.get("languageCode").and_then(serde_json::Value::as_str) { + language_code = Some(v.to_string()); + } + if let Some(v) = el + .get("includeTimestamps") + .and_then(serde_json::Value::as_bool) + { + include_timestamps = v; + } + } + + // Audio format: the realtime endpoint takes pcm_{rate} / ulaw_{rate}. + let default_rate = if options.input_audio_format.format_type == "audio/pcmu" { + 8_000 + } else { + 16_000 + }; + let sample_rate = options.input_audio_format.rate.unwrap_or(default_rate); + let audio_format = match options.input_audio_format.format_type.as_str() { + "audio/pcm" => format!("pcm_{sample_rate}"), + "audio/pcmu" => format!("ulaw_{sample_rate}"), + other => { + return Err(AiMuxError::UnsupportedFunctionality(format!( + "ElevenLabs realtime accepts audio/pcm or audio/pcmu input, got {other}" + ))); + } + }; + + let base = self.config.base_url.trim_end_matches('/'); + let (scheme, host) = if let Some(rest) = base.strip_prefix("https://") { + ("wss", rest) + } else if let Some(rest) = base.strip_prefix("http://") { + ("ws", rest) + } else { + ("wss", base) + }; + // Scoped: the serializer is not `Send`; keep it inside this block so + // nothing non-Send is alive across the connect await below. + let ws_url = { + let mut query = url::form_urlencoded::Serializer::new(String::new()); + query + .append_pair("model_id", &self.model_id) + .append_pair("audio_format", &audio_format) + .append_pair("commit_strategy", "manual"); + if let Some(code) = &language_code { + query.append_pair("language_code", code); + } + if include_timestamps { + query.append_pair("include_timestamps", "true"); + } + format!( + "{scheme}://{host}/v1/speech-to-text/realtime?{}", + query.finish() + ) + }; + + let header_list: Vec<(String, String)> = self + .build_headers(options.headers.as_ref()) + .into_iter() + .collect(); + + // Connect BEFORE the stream so connect failures surface from + // do_stream's Result (same contract as the OpenAI realtime path). + let req = WebSocketRequest { + url: ws_url.clone(), + headers: header_list, + subprotocols: Vec::new(), + abort_signal: options.abort_signal.clone(), + timeout: options.timeout, + }; + let mut ws = ws_connect(&req).await?; + + // Settle window for the post-commit final event: the server does not + // close the session (RFC-0034 §3.3), so an empty Finish must be + // possible. chunk_ms when configured, otherwise 5s. + let settle = options + .timeout + .as_ref() + .and_then(|t| t.chunk_ms) + .map(tokio::time::Duration::from_millis) + .unwrap_or(tokio::time::Duration::from_secs(5)); + + let include_raw = options.include_raw_chunks; + let model_id = self.model_id.clone(); + let error_url = ws_url.clone(); + let mut audio = options.audio; + + let stream = async_stream::stream! { + // One held chunk: the commit flag must ride the LAST real audio + // chunk, so each incoming chunk flushes the previous one and the + // stream-end flushes the held one with commit=true. An + // empty-audio stream commits via a single empty chunk. + let mut held: Option = None; + let mut audio_done = false; + let mut commit_deadline: Option = None; + // Audio is not sent before the session is established: the + // server confirms configuration with session_started first. + let mut session_started = false; + + loop { + let audio_next = async { + if audio_done || !session_started { + std::future::pending::<()>().await; + None + } else { + audio.next().await + } + }; + + tokio::select! { + biased; + + chunk = audio_next => { + // Hold-one-chunk pipeline: the commit flag must ride + // the LAST real chunk. A new chunk flushes the + // previously held one (commit=false); the stream end + // flushes the held one with commit=true; an + // empty-audio stream commits a single empty chunk. + let (payload, commit) = match chunk { + None => { + audio_done = true; + commit_deadline = Some(tokio::time::Instant::now() + settle); + (held.take().unwrap_or_default(), true) + } + Some(aimux_core::transcription_model::AudioChunk::Binary(bytes)) => { + use base64::Engine as _; + let b64 = base64::engine::general_purpose::STANDARD.encode(bytes); + match held.replace(b64) { + Some(previous) => (previous, false), + // First chunk: hold it, nothing to send yet. + None => continue, + } + } + Some(aimux_core::transcription_model::AudioChunk::Base64(b64)) => { + match held.replace(b64) { + Some(previous) => (previous, false), + None => continue, + } + } + }; + let message = serde_json::json!({ + "message_type": "input_audio_chunk", + "audio_base_64": payload, + "commit": commit, + "sample_rate": sample_rate, + }); + if let Err(e) = ws.send_text(&message.to_string()).await { + yield Err(e); + break; + } + } + + _ = async { + match commit_deadline { + Some(d) => tokio::time::sleep_until(d).await, + None => std::future::pending::<()>().await, + } + } => { + // Server went silent after commit: an empty Finish + // beats hanging (RFC-0034 §3.3.3). + yield Ok(TranscriptionStreamPart::Finish { + text: String::new(), + segments: vec![], + language: language_code.clone(), + duration_in_seconds: None, + provider_metadata: None, + }); + ws.close().await; + break; + } + + event = ws.next() => { + match event { + None => { + yield Err(AiMuxError::ApiCall(Box::new( + aimux_core::error::ApiCallError::new( + "realtime transcription socket closed before the committed transcript", + error_url.clone(), + serde_json::json!({}), + ), + ))); + break; + } + Some(Err(e)) => { + yield Err(e); + break; + } + Some(Ok(WsMessage::Binary(_))) => {} + Some(Ok(WsMessage::Text(text))) => { + let Ok(value) = serde_json::from_str::(&text) else { + continue; + }; + if include_raw { + yield Ok(TranscriptionStreamPart::Raw { + raw_value: value.clone(), + }); + } + let event_type = value.get("message_type") + .and_then(|t| t.as_str()).unwrap_or(""); + match event_type { + "session_started" => { + session_started = true; + yield Ok(TranscriptionStreamPart::StreamStart { + warnings: vec![], + }); + } + "partial_transcript" => { + yield Ok(TranscriptionStreamPart::TranscriptPartial { + id: None, + text: value.get("text") + .and_then(|v| v.as_str()).unwrap_or("").to_string(), + start_second: None, + duration_in_seconds: None, + channel_index: None, + provider_metadata: None, + }); + } + // include_timestamps=true swaps the event + // shape; both carry `text`, the timestamps + // variant adds `words[]`. + "committed_transcript" + | "committed_transcript_with_timestamps" => { + let text = value.get("text") + .and_then(|v| v.as_str()).unwrap_or("").to_string(); + let segments = value.get("words") + .and_then(|v| v.as_array()) + .map(|words| { + words.iter().filter_map(|w| { + let word = w.get("text") + .and_then(|v| v.as_str())?; + // Spacing-type entries are + // layout, not content. + if word.trim().is_empty() { + return None; + } + Some(TranscriptionSegment { + text: word.to_string(), + start_second: w.get("start") + .and_then(serde_json::Value::as_f64).unwrap_or(0.0), + end_second: w.get("end") + .and_then(serde_json::Value::as_f64).unwrap_or(0.0), + }) + }).collect::>() + }) + .unwrap_or_default(); + let duration = segments.last().map(|s| s.end_second); + yield Ok(TranscriptionStreamPart::TranscriptFinal { + id: None, + text: text.clone(), + start_second: segments.first() + .map(|s| s.start_second), + end_second: segments.last() + .map(|s| s.end_second), + channel_index: None, + provider_metadata: None, + }); + // Manual strategy commits exactly once + // (on our final flag): this committed + // transcript IS the finish. + yield Ok(TranscriptionStreamPart::Finish { + text, + segments, + language: language_code.clone(), + duration_in_seconds: duration, + provider_metadata: None, + }); + ws.close().await; + break; + } + "warning" => { + // No part mapping (StreamStart already + // went out); visible via Raw when + // include_raw_chunks is set. + } + other if matches!( + other, + "error" | "auth_error" | "quota_exceeded" + | "commit_throttled" | "unaccepted_terms" + | "rate_limited" | "queue_overflow" + | "resource_exhausted" + | "session_time_limit_exceeded" + | "input_error" | "invalid_request" + | "chunk_size_exceeded" + | "insufficient_audio_activity" + | "transcriber_error" + ) => { + let message = value.get("error") + .and_then(|v| v.as_str()) + .unwrap_or("realtime transcription error"); + // Retryable classification (RFC-0034 + // §3.2): three transient names, the + // rest are terminal verdicts. + let is_retryable = matches!( + other, + "rate_limited" | "queue_overflow" + | "resource_exhausted" + ); + yield Err(AiMuxError::ApiCall(Box::new( + aimux_core::error::ApiCallError { + is_retryable, + response_body: Some(value.to_string()), + ..aimux_core::error::ApiCallError::new( + format!("elevenlabs realtime: {message}"), + error_url.clone(), + serde_json::json!({}), + ) + }, + ))); + ws.close().await; + break; + } + _ => {} + } + } + } + } + } + } + }; + + Ok(TranscriptionStreamResult { + stream: Box::pin(stream), + request: Some(TranscriptionRequest { body: Some(ws_url) }), + response: Some(TranscriptionResponse { + timestamp: Some(chrono::Utc::now().to_rfc3339()), + model_id: Some(model_id), + headers: None, + body: None, + }), + }) + } + + /// Without the `realtime` feature the WebSocket path is compiled out. + #[cfg(not(feature = "realtime"))] + async fn do_stream( + &self, + _options: aimux_core::transcription_model::TranscriptionStreamOptions, + ) -> Result { + Err(AiMuxError::UnsupportedFunctionality(format!( + "streaming transcription with `{}` requires building aimux-providers \ + with the `realtime` feature", + self.model_id + ))) + } } diff --git a/aimux-providers/tests/elevenlabs_realtime_stream_test.rs b/aimux-providers/tests/elevenlabs_realtime_stream_test.rs new file mode 100644 index 00000000..b4fd3fc9 --- /dev/null +++ b/aimux-providers/tests/elevenlabs_realtime_stream_test.rs @@ -0,0 +1,536 @@ +//! ElevenLabs realtime streaming tests (RFC-0034 §3, P2). +//! +//! Each test runs a local WebSocket server playing the +//! `wss://…/v1/speech-to-text/realtime` role: handshake (capturing the +//! request URI + headers), `session_started`, audio chunks until the +//! committing `input_audio_chunk`, then `partial_transcript` / +//! `committed_transcript` events. +//! +//! Wire shape is pinned field-by-field against the public API reference +//! (2026-09). Live-API smoke pending (no key at implementation time) — same +//! posture as RFC-0028 D4. + +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use futures::{SinkExt, StreamExt}; +use tokio::net::{TcpListener, TcpStream}; +use tokio_tungstenite::tungstenite::Message; +use tokio_tungstenite::tungstenite::handshake::server::{Request, Response}; + +use aimux_core::AbortSignal; +use aimux_core::error::AiMuxError; +use aimux_core::options::TimeoutConfiguration; +use aimux_core::transcription_model::{ + AudioChunk, InputAudioFormat, TranscriptionModel, TranscriptionStreamOptions, + TranscriptionStreamPart, +}; +use aimux_providers::{ElevenLabsConfig, ElevenLabsProvider}; + +// ── helpers ───────────────────────────────────────────────────────────────── + +#[derive(Clone)] +struct Captured { + uri: Arc>>, + api_key_header: Arc>>, + /// All `input_audio_chunk` messages as (audio_base_64, commit, sample_rate). + chunks: Arc>>, +} + +impl Captured { + fn new() -> Self { + Self { + uri: Arc::new(Mutex::new(None)), + api_key_header: Arc::new(Mutex::new(None)), + chunks: Arc::new(Mutex::new(Vec::new())), + } + } + + fn uri(&self) -> String { + self.uri.lock().unwrap().clone().unwrap_or_default() + } + + fn chunks(&self) -> Vec<(String, bool, u64)> { + self.chunks.lock().unwrap().clone() + } +} + +/// Accept a connection, capture the handshake URI + `xi-api-key` header, +/// read `input_audio_chunk` frames until the committing one, then hand the +/// socket to the scripted `after_audio` behavior. +async fn serve( + stream: TcpStream, + captured: Captured, + after_audio: impl FnOnce( + &mut tokio_tungstenite::WebSocketStream, + ) -> futures::future::BoxFuture<'_, ()> + + Send + + 'static, +) { + let uri = Arc::clone(&captured.uri); + let api_key = Arc::clone(&captured.api_key_header); + #[allow(clippy::result_large_err)] + let handshake = move |req: &Request, resp: Response| { + *uri.lock().unwrap() = Some(req.uri().to_string()); + *api_key.lock().unwrap() = req + .headers() + .get("xi-api-key") + .and_then(|v| v.to_str().ok()) + .map(String::from); + Ok(resp) + }; + let ws = tokio_tungstenite::accept_hdr_async(stream, handshake) + .await + .expect("ws handshake"); + let mut ws = ws; + + // session_started first — the client holds audio until this arrives. + ws.send(Message::Text( + r#"{"message_type":"session_started","session_id":"s-1","sample_rate":24000}"#.into(), + )) + .await + .unwrap(); + + // Audio frames until the committing one. + loop { + let Some(Ok(Message::Text(text))) = ws.next().await else { + return; + }; + let v: serde_json::Value = serde_json::from_str(&text).unwrap(); + if v["message_type"] != "input_audio_chunk" { + panic!("expected input_audio_chunk, got {text}"); + } + let entry = ( + v["audio_base_64"].as_str().unwrap_or("").to_string(), + v["commit"].as_bool().unwrap_or(false), + v["sample_rate"].as_u64().unwrap_or(0), + ); + let committed = entry.1; + captured.chunks.lock().unwrap().push(entry); + if committed { + break; + } + } + + after_audio(&mut ws).await; +} + +/// Start the server; returns (base_url, captured). +async fn start( + after_audio: impl FnOnce( + &mut tokio_tungstenite::WebSocketStream, + ) -> futures::future::BoxFuture<'_, ()> + + Send + + 'static, +) -> (String, Captured) { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let captured = Captured::new(); + let captured_for_task = captured.clone(); + tokio::spawn(async move { + let (stream, _) = listener.accept().await.unwrap(); + serve(stream, captured_for_task, after_audio).await; + }); + (format!("http://127.0.0.1:{port}"), captured) +} + +fn realtime_model(base_url: &str, model_id: &str) -> aimux_providers::ElevenLabsTranscriptionModel { + let config = ElevenLabsConfig { + api_key: "test-api-key".to_string(), + base_url: base_url.to_string(), + headers: None, + }; + ElevenLabsProvider::new(config).transcription(model_id) +} + +fn stream_options( + chunks: Vec, + abort: Option, +) -> TranscriptionStreamOptions { + TranscriptionStreamOptions { + audio: Box::pin(futures::stream::iter(chunks)), + input_audio_format: InputAudioFormat { + format_type: "audio/pcm".to_string(), + rate: Some(24_000), + }, + provider_options: None, + abort_signal: abort, + headers: None, + include_raw_chunks: false, + timeout: None, + } +} + +async fn collect( + result: aimux_core::transcription_model::TranscriptionStreamResult, +) -> Vec> { + let mut out = Vec::new(); + let mut stream = result.stream; + while let Some(part) = stream.next().await { + out.push(part); + } + out +} + +// ── tests ─────────────────────────────────────────────────────────────────── + +/// Happy path: URL/config on the query string, audio as base64 JSON with the +/// commit flag on the LAST chunk, partial/final/finish parts, close(1000). +#[tokio::test] +async fn stream_realtime_happy_path() { + let (base_url, captured) = start(|ws| { + Box::pin(async move { + ws.send(Message::Text( + r#"{"message_type":"partial_transcript","text":"Hel"}"#.into(), + )) + .await + .unwrap(); + ws.send(Message::Text( + r#"{"message_type":"committed_transcript","text":"Hello world"}"#.into(), + )) + .await + .unwrap(); + match ws.next().await { + Some(Ok(Message::Close(frame))) => { + let code = frame.map(|f| u16::from(f.code)); + assert_eq!(code, Some(1000), "client must close(1000) after the final"); + } + other => panic!("expected close frame, got {other:?}"), + } + }) + }) + .await; + + let model = realtime_model(&base_url, "scribe_v2_realtime"); + let result = model + .do_stream(stream_options( + vec![ + AudioChunk::Binary(vec![1, 2, 3]), + AudioChunk::Binary(vec![4, 5, 6]), + ], + None, + )) + .await + .expect("do_stream should connect"); + let parts = collect(result).await; + + // Config travels on the URL, not a session message (wire fact). + let uri = captured.uri(); + assert!( + uri.starts_with("/v1/speech-to-text/realtime?"), + "path+query expected, got {uri}" + ); + assert!(uri.contains("model_id=scribe_v2_realtime"), "in {uri}"); + assert!(uri.contains("audio_format=pcm_24000"), "in {uri}"); + assert!(uri.contains("commit_strategy=manual"), "in {uri}"); + assert_eq!( + captured.api_key_header.lock().unwrap().as_deref(), + Some("test-api-key"), + "xi-api-key header must ride the WS handshake" + ); + + // Hold-one-chunk pipeline: exactly two frames, commit on the last. + assert_eq!( + captured.chunks(), + vec![ + ("AQID".to_string(), false, 24_000), + ("BAUG".to_string(), true, 24_000), + ], + "commit flag must ride the last real chunk" + ); + + assert_eq!(parts.len(), 4, "parts: {parts:?}"); + assert!(matches!( + &parts[0], + Ok(TranscriptionStreamPart::StreamStart { .. }) + )); + match &parts[1] { + Ok(TranscriptionStreamPart::TranscriptPartial { text, .. }) => assert_eq!(text, "Hel"), + other => panic!("expected partial, got {other:?}"), + } + match &parts[2] { + Ok(TranscriptionStreamPart::TranscriptFinal { text, .. }) => { + assert_eq!(text, "Hello world") + } + other => panic!("expected final, got {other:?}"), + } + match &parts[3] { + Ok(TranscriptionStreamPart::Finish { text, segments, .. }) => { + assert_eq!(text, "Hello world"); + assert!(segments.is_empty()); + } + other => panic!("expected finish, got {other:?}"), + } +} + +/// providerOptions: languageCode lands on the query; include_timestamps +/// switches the committed event to the words-carrying shape, which fills +/// segments (spacing entries dropped) and the finish duration. +#[tokio::test] +async fn stream_language_and_timestamps() { + let (base_url, captured) = start(|ws| { + Box::pin(async move { + ws.send(Message::Text( + r#"{"message_type":"committed_transcript_with_timestamps","text":"Hello world","language_code":"fr","words":[ + {"text":"Hello","start":0.0,"end":0.5,"type":"word"}, + {"text":" ","start":0.5,"end":0.5,"type":"spacing"}, + {"text":"world","start":0.5,"end":1.0,"type":"word"} + ]}"# + .replace('\n', ""), + )) + .await + .unwrap(); + let _ = ws.next().await; // close + }) + }) + .await; + + let model = realtime_model(&base_url, "scribe_v2_realtime"); + let mut options = stream_options(vec![AudioChunk::Binary(vec![1])], None); + let mut provider_options = std::collections::HashMap::new(); + provider_options.insert( + "elevenlabs".to_string(), + serde_json::json!({ "languageCode": "fr", "includeTimestamps": true }), + ); + options.provider_options = Some(provider_options); + let result = model.do_stream(options).await.unwrap(); + let parts = collect(result).await; + + let uri = captured.uri(); + assert!(uri.contains("language_code=fr"), "in {uri}"); + assert!(uri.contains("include_timestamps=true"), "in {uri}"); + + match &parts[parts.len() - 2] { + Ok(TranscriptionStreamPart::TranscriptFinal { + text, + start_second, + end_second, + .. + }) => { + assert_eq!(text, "Hello world"); + assert_eq!(*start_second, Some(0.0)); + assert_eq!(*end_second, Some(1.0)); + } + other => panic!("expected final with timestamps, got {other:?}"), + } + match parts.last().unwrap() { + Ok(TranscriptionStreamPart::Finish { + text, + segments, + language, + duration_in_seconds, + .. + }) => { + assert_eq!(text, "Hello world"); + assert_eq!(language.as_deref(), Some("fr")); + assert_eq!(*duration_in_seconds, Some(1.0)); + assert_eq!(segments.len(), 2, "spacing entry dropped: {segments:?}"); + assert_eq!(segments[0].text, "Hello"); + assert_eq!(segments[1].text, "world"); + } + other => panic!("expected finish, got {other:?}"), + } +} + +/// An empty audio stream commits via a single empty chunk. +#[tokio::test] +async fn stream_empty_audio_commits_empty_chunk() { + let (base_url, captured) = start(|ws| { + Box::pin(async move { + ws.send(Message::Text( + r#"{"message_type":"committed_transcript","text":""}"#.into(), + )) + .await + .unwrap(); + let _ = ws.next().await; // close + }) + }) + .await; + + let model = realtime_model(&base_url, "scribe_v2_realtime"); + let result = model + .do_stream(stream_options(vec![], None)) + .await + .expect("connect"); + let parts = collect(result).await; + + assert_eq!( + captured.chunks(), + vec![(String::new(), true, 24_000)], + "no audio → one empty committing chunk" + ); + assert!(matches!( + parts.last(), + Some(Ok(TranscriptionStreamPart::Finish { .. })) + )); +} + +/// Error classification (RFC-0034 §3.2): three transient names retry, the +/// rest are terminal verdicts. +#[tokio::test] +async fn stream_error_classification() { + for (message_type, expect_retryable) in [ + ("auth_error", false), + ("rate_limited", true), + ("queue_overflow", true), + ("quota_exceeded", false), + ] { + let (base_url, _captured) = start(move |ws| { + let message_type = message_type.to_string(); + Box::pin(async move { + ws.send(Message::Text( + serde_json::json!({ + "message_type": message_type, + "error": "boom", + }) + .to_string(), + )) + .await + .unwrap(); + }) + }) + .await; + + let model = realtime_model(&base_url, "scribe_v2_realtime"); + let result = model + .do_stream(stream_options(vec![AudioChunk::Binary(vec![1])], None)) + .await + .unwrap(); + let parts = collect(result).await; + // StreamStart precedes the error (session_started arrives first). + let error_part = parts.iter().find_map(|p| match p { + Err(e) => Some(e.clone()), + Ok(_) => None, + }); + match error_part { + Some(AiMuxError::ApiCall(api_call)) => { + assert_eq!( + api_call.is_retryable, expect_retryable, + "{message_type} retryability" + ); + assert!( + api_call + .response_body + .as_deref() + .unwrap_or("") + .contains(message_type), + "raw event preserved in response_body" + ); + } + Some(other) => panic!("{message_type}: expected ApiCall, got {other:?}"), + None => panic!("{message_type}: stream ended without an error part: {parts:?}"), + } + } +} + +/// A server that goes silent after the commit trips the settle window +/// (chunk_ms) → empty Finish, stream terminates (RFC-0034 §3.3.3). +#[tokio::test] +async fn stream_silent_server_finishes_via_settle_window() { + let (base_url, _captured) = start(|ws| { + Box::pin(async move { + // Silence: hold the socket open, never answer the commit. + let _ = ws.next().await; + }) + }) + .await; + + let model = realtime_model(&base_url, "scribe_v2_realtime"); + let mut options = stream_options(vec![AudioChunk::Binary(vec![1])], None); + options.timeout = Some(TimeoutConfiguration { + first_chunk_ms: Some(5_000), + chunk_ms: Some(100), + step_ms: None, + total_ms: Some(5_000), + }); + let result = model.do_stream(options).await.unwrap(); + let parts = collect(result).await; + match parts.last() { + Some(Ok(TranscriptionStreamPart::Finish { text, .. })) => { + assert_eq!(text, "", "silent server → empty finish, not a hang"); + } + other => panic!("expected empty finish, got {other:?}"), + } +} + +/// Abort mid-session surfaces `AiMuxError::Aborted`. +#[tokio::test] +async fn stream_abort_mid_session() { + let (base_url, _captured) = start(|ws| { + Box::pin(async move { + // Drain client messages without acting; the abort is what ends + // the session. + while let Some(_msg) = ws.next().await { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + }) + .await; + + let abort = AbortSignal::new(); + let abort_clone = abort.clone(); + tokio::spawn(async move { + tokio::time::sleep(Duration::from_millis(100)).await; + abort_clone.abort(); + }); + + let model = realtime_model(&base_url, "scribe_v2_realtime"); + let result = model + .do_stream(stream_options( + vec![AudioChunk::Binary(vec![1])], + Some(abort), + )) + .await + .unwrap(); + let parts = collect(result).await; + assert!( + parts + .iter() + .any(|p| matches!(p, Err(AiMuxError::Aborted(_)))), + "expected Aborted in {parts:?}" + ); +} + +/// Model gating: realtime IDs reject do_generate; batch IDs reject do_stream +/// (no server needed — nothing connects). +#[tokio::test] +async fn stream_model_gating_is_symmetric() { + let realtime = realtime_model("http://127.0.0.1:1", "scribe_v2_realtime"); + let error = realtime + .do_generate( + &aimux_core::transcription_model::TranscriptionCallOptions::new( + aimux_core::transcription_model::AudioInput::Binary(vec![1]), + "audio/pcm", + ), + ) + .await + .unwrap_err(); + assert!(matches!(error, AiMuxError::UnsupportedFunctionality(_))); + + let batch = realtime_model("http://127.0.0.1:1", "scribe_v1"); + let error = batch + .do_stream(stream_options(vec![], None)) + .await + .unwrap_err(); + assert!( + matches!(error, AiMuxError::UnsupportedFunctionality(ref e) if e.to_string().contains("scribe_v1")), + "got {error:?}" + ); +} + +/// The realtime endpoint accepts pcm/ulaw only; anything else fails fast +/// without connecting. +#[tokio::test] +async fn stream_rejects_non_pcm_formats() { + let model = realtime_model("http://127.0.0.1:1", "scribe_v2_realtime"); + let mut options = stream_options(vec![], None); + options.input_audio_format = InputAudioFormat { + format_type: "audio/mpeg".to_string(), + rate: None, + }; + let error = model.do_stream(options).await.unwrap_err(); + assert!( + matches!(error, AiMuxError::UnsupportedFunctionality(ref e) if e.to_string().contains("audio/mpeg")), + "got {error:?}" + ); +} diff --git a/rfc/0034-realtime-stt-followups.md b/rfc/0034-realtime-stt-followups.md index 295aef5e..7432c0bb 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**: P1 已实现(#183);P2/P3/P4 待做。设计稿其余部分待评审 +> **Status**: P1 已实现(#183);P2 已实现(RFC-0034/elevenlabs-realtime 分支,PR 见 #184);P3 暂缓(D6);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 会话生命周期先例) @@ -165,7 +165,7 @@ RFC-0028 落地了 WS 基础设施 + OpenAI realtime 转写 + FFI 会话 + 8 语 | 阶段 | 内容 | 依赖 | PR | |------|------|------|----| | P1 | WS 代理隧道 + 测试 | 无 | ✅ #183(draft) | -| P2 | ElevenLabs realtime 门控 + `do_stream` + mock 测试 + live smoke | 无(建议在 P1 后,便于 smoke 走代理验证) | 独立 | +| P2 | ElevenLabs realtime 门控 + `do_stream` + mock 测试 + live smoke | 无(建议在 P1 后,便于 smoke 走代理验证) | ✅ 实现+mock 测试落地(live smoke 待 key,见 §3.5 注) | | 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 | From 9097575d9d8c383cbdc648d9c27320af06c5f38c Mon Sep 17 00:00:00 2001 From: eric8810 Date: Sun, 13 Sep 2026 15:24:36 +0800 Subject: [PATCH 6/6] docs(rfc-0034): wss rustls branch closed by the handshake-level live smoke --- rfc/0034-realtime-stt-followups.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/rfc/0034-realtime-stt-followups.md b/rfc/0034-realtime-stt-followups.md index 7432c0bb..04531b17 100644 --- a/rfc/0034-realtime-stt-followups.md +++ b/rfc/0034-realtime-stt-followups.md @@ -73,7 +73,7 @@ RFC-0028 落地了 WS 基础设施 + OpenAI realtime 转写 + FFI 会话 + 8 语 - 本地假 CONNECT 代理(`TcpListener` 手写:校验 CONNECT 目标行 → 200 → 透传到真实本地 WS server):断言 CONNECT 目标、握手成功、事件往返。 - no_proxy 命中 → 断言未经过代理;`*` 通配;带端口条目。 -- 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 桩,注入根证书需要测试缝,不值当。) +- 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 分支已由握手级 smoke 闭环:经本地 CONNECT 代理连真实 `wss://api.elevenlabs.io`,真实服务器应答握手即证明隧道+证书链全链路执行(`#[ignore]` 手动跑,`live_wss_handshake_through_connect_proxy`);完整转写 round-trip 仍需 provider key,见 §3.5。) ### 2.4 范围外