feat(gateway): complete OpenAI Responses WebSocket mode - #704
Open
AAEE86 wants to merge 15 commits into
Open
Conversation
在 /v1/responses 上支持 WebSocket 升级，把客户端帧中继到上游 Codex / OpenAI Responses WebSocket 端点，同时保持既有的路由、鉴权、配额与用量 语义： - 路由与准入：control/route/ai.rs 识别 WebSocket 升级请求； websocket/ingress.rs 复用 API Key 鉴权、IP 规则与并发许可，并引入 独立的 WebSocket 连接许可 - 中继：websocket/responses/* 按 connection / session / turn 分层， 帧解析归一化、socket 写入有界、continuation 保持调度亲和性 - 配额：orchestration/codex_quota_breaker.rs 在账号配额耗尽时熔断并 自动恢复，不再直接断开客户端连接 - 用量：每个 turn 的终态用量落库，request_metadata 记录 websocket_mode / websocket_transport，管理端与 usage 视图暴露 is_websocket - 管理端：provider 可配置 Responses WebSocket 开关
新增 aether-codex-ws-probe 与 aether-openai-responses-ws-probe 两个 二进制，用于在不暴露凭据的前提下验证上游 WebSocket 端点可用性：凭据 只从环境变量读取，不写入日志。公共流程放在 bin/support/responses_ws_probe.rs，各 profile 只负责自己的鉴权与 请求头要求。
补齐 aether-integration-tests 的 responses_websocket_e2e 集成测试， 并把 CI 的 scenario 任务从 --bins 改为 --bins --tests，否则该套件 不会被执行。
provider 表单支持开启 Responses WebSocket；用量列表、状态与详情 区分 WebSocket 请求。
评审第 2 条：BoundResponsesConnection 用 response_in_flight、active_turn、
active_response_create 三个可独立变化的字段编码同一件事，8 种组合里只有 3 种
合法，非法组合只能靠调用点的 if 和「记得同时改另外两个字段」来避免。
三字段合并为一个 ResponsesTurnState：
Idle 没有进行中的 logical turn
Responding { logical, attempt } logical 与 attempt 必须同时存在
Replanning { logical } attempt 已取走去结算/重绑，logical 仍在
Replanning 不是新概念：配额透明重试期间现状就处于这个状态，只是靠
Option::take 意外得到。转换只能走 begin / detach_attempt / resume / end，
response_in_flight 与「是否接受新 response.create」都由变体推导。
由此消除的运行时不变量（原来全靠调用点自觉）：
- 有 attempt 必有 logical turn
- response_in_flight 与 attempt 同生共死（原来 client 写失败后
active_turn=None 而 response_in_flight 仍为 true）
- logical turn 结束时必须清 attempt：原来 `active_response_create = None`
在 connection.rs 里手写 13 处，漏一处就残留；现在只有 end() 一个出口
- 上游绑定返回的连接不再自带 response_in_flight=true 的半成品状态
同时删除 update_response_in_flight：Started 帧把已经是 true 的字段再设一次，
Close 帧因为没有解析出的 frame 而根本不触发，是纯冗余写；它在 Idle 态收到
Started 帧时还会把 response_in_flight 置真，从而永久阻塞后续 response.create。
行为等价。ActiveResponsesWebSocketRequest 改名 LogicalTurn 并随状态机移入
新的 turn_state.rs；状态机对 attempt 类型泛型化，测试用轻量替身驱动同一套
转换逻辑，无需 AppState 或真实 socket。
评审第 5 条：一个 ResponsesWebSocketTurn 同时代表 logical turn 和 provider attempt，finalize() 又用 outcome.cancelled() 一个布尔驱动 billing、candidate 状态和供应商效果，于是 provider 终态已经到达、只是最后一跳写客户端失败时， 供应商事实会被 Cancelled 覆盖掉。 - ResponsesWebSocketTurn → ResponsesProviderAttempt， ActiveResponsesWebSocketTurn → ActiveProviderAttempt：类型名字明确它只代表 一次上游执行，logical turn 由 C1 落地的 LogicalTurn 承担。 - 新增 settlement.rs：AttemptProviderOutcome × AttemptClientDelivery 两个正交 事实，classify_attempt_settlement 一张表推出 status_code / billing / candidate 状态 / candidate 错误分类 / 供应商效果 / 是否提交 execution report。 - attempt 观察到 provider 终态即记录 provider_outcome。结算信号 ResponsesWebSocketTurnOutcome 只回答「为什么现在结算」：ProviderTerminal 与 Failure 对 provider 是权威的，Cancelled 只描述客户端/连接层面的停止，不再 覆盖已观察到的 provider 事实。 - candidate 状态与 candidate 错误分类分开输出：现状存在 「missing_terminal=true 而记账层判 Success」的组合（report kind 不要求观察到 终态事件时），会写出 status=Success + error_type=stream_missing_terminal_event， 这个组合必须原样保留。 classify_responses_websocket_turn_effect 的判定表原样搬入 settlement.rs，分支 和顺序均未改动，两个既有不变量测试随之迁移。 行为等价。结算表当前口径与拆分前完全一致：客户端投递失败仍与「供应商声明取消」 落在同一侧（作废账单、candidate 记 Cancelled、只释放 lease、不提交 execution report），即使 provider 终态已经到达——这一行由 settlement_table_row_client_delivery_failure_currently_voids_a_reached_terminal 锁住现状，修正它是下一步独立的行为修正。 新增 15 个测试：outcome → 双事实映射表逐行（含 stream_timeout 只在 504 失败一族 成立、provider 终态即使 504 也不投射流式超时）、结算表逐行、投递失败时 forced_error 必须为 None、已观察终态不被 Cancelled 覆盖、以及跨整张表的 「每个分支都释放 pool key lease」「作废账单一律不提交 report」不变量。
…ails
评审第 5 条后半：provider 终态已经到达、只是 gateway 写客户端 socket 失败时，
relay loop 用 client_disconnected() 覆盖了结算信号，于是一条供应商已经完成推理
并消耗了 token 的响应被记成 void billing、candidate 记 Cancelled、不投射供应商
效果、也不提交 execution report。上游成本凭空消失。
结算表只改一行：作废账单的条件从
provider.cancelled_by_provider() || delivery.is_aborted()
收紧为
provider.cancelled_by_provider() || (delivery.is_aborted() && !provider.is_terminal())
于是 Terminal{cancelled=false} + delivery Aborted 与 delivery Complete 落在同一侧：
Billed、candidate Success 或 Failed、投射供应商效果、提交 execution report。
状态码随之变成纯 provider 事实（不再把 200 改写成 499）；作废分支的 provider
状态码本身就是 499，取值不变。
依据：供应商已经完成推理并消耗 token，客户端还能用 previous_response_id 续取
这条响应。供应商没给出终态时（客户端先走了）仍然作废，这一侧未改。
配套改动：
- connection.rs 写客户端失败处改为 record_client_delivery_aborted(reason) +
settle_signal_for_client_delivery_failure(terminal_outcome)：provider 终态已到达
就用那条终态作结算信号，不再无条件覆盖。投递失败原因也不再谎称
「客户端在终态前断开」。
- 投递结果记在 attempt 上而非 logical turn 上：结算按 attempt 进行，且配额透明
重试时各 attempt 的投递结果彼此独立。
- report_context 新增 websocket_client_delivery="aborted" 与
websocket_client_delivery_reason，只增字段不改既有字段，便于事后区分
「客户端拿到了」和「客户端没拿到但已计费」。
- candidate error_type 新增 client_delivery_failed（原先这个场景写的是
websocket_cancelled）。它排在供应商侧分类之前：这条记录之所以特别正是因为
内容没送到客户端，供应商侧判定仍由 candidate_status 与 error_message 保留。
- finish_summary 改用作废判定而非「投递失败」判定：provider 终态已到达时摘要
必须保留真实的 finish_reason 与 usage，否则计费记录会被写坏。
e2e 期望值变化：client_disconnect_mid_turn_still_settles_the_usage_row 改名为
client_disconnect_before_any_provider_output_settles_a_void_row，并补上
「不计费 + status=cancelled + status_code=499」的断言。原用例的 mock 行为是
StallAfterCreated（只发 response.created 就静默），provider 从未给出终态，所以
它走的是未改动的作废一侧；原来的文档注释说「must still be billed」与实际语义
不符，一并纠正。真正被修正的那一行无法在 e2e 里确定性触发——它取决于 relay
loop 的 select! 先观察到上游终态帧还是先观察到已关闭的客户端 socket，是构造性
竞态——因此由 relay 级单测确定性覆盖，e2e 里以注释指向这两个单测。
新增 7 个测试：结算表修正行（并与「投递成功」逐字段对照，只有 candidate 错误
分类不同）、无终态时仍作废、供应商声明取消即使送达也不计费、结算信号选择、
已记录的投递失败不被结算信号覆盖、relay 级「终态到达 + 客户端已关闭 ⇒ Billed /
Success / ProviderSuccess / 已提交 report 且 usage 完整保留」及其镜像、
report_context 只增不改。
评审第 4 条：responses/turn.rs 实际复制了一整套 HTTP execution lifecycle——
usage 写入、candidate 状态流转、health/adaptive 效果投射、pool key lease 释放、
body capture、账单失败判定，与 HTTP 的顺序和超时语义只能靠人工对齐。
新增 execution_runtime/attempt_lifecycle.rs，把一次 provider attempt 的记账收成
transport 中立的三段：
ExecutionAttemptLifecycle::begin pending usage 行 + Pending candidate
ExecutionAttemptLifecycle::mark_started usage stream_started + Streaming candidate（幂等）
ExecutionAttemptLifecycle::settle 终态四段，顺序不可重排：
1 usage terminal（detachable，不可丢）
2 candidate terminal
3 provider 效果 + 超时兜底释放 lease
4 execution report（作废账单不提交）
顺序、5s 分段超时常量、detachable 语义、「每个效果分支都释放 lease」「作废账单
一律不提交 report」这些不变量全部保持原样。
一并上移的辅助设施：
- AttemptStageGuard 取代 await_websocket_lifecycle_stage /
await_detachable_lifecycle_stage，把「等多久」参数化：WS 用 Bounded(5s)，
HTTP 接线时用 Unbounded 即保持它现在的语义。
- AttemptBodyCapture 取代 append_capture / encode_stream_capture，把
「缓冲 + 截断标志」两个字段收成一个类型（WS 侧四个字段变两个）。捕获内容
仍然是 SSE 形状：usage runtime 按 data: 行解析被捕获的 body 来判定
StreamCapturedTerminalState，换成结构化 JSON 会让终态判定恒为 Missing。
- C2/C3 的结算表本来就不含任何 WS 类型，随之上移。效果表分支与注释逐字未改，
仅按新位置改名为 AttemptProviderEffect / classify_attempt_provider_effect。
responses/settlement.rs 只保留 WS 专属的一件事：把 relay loop 的结算信号
ResponsesWebSocketTurnOutcome 翻译成两个正交事实。
ResponsesProviderAttempt 现在只持有 WS 专有状态：lifecycle 句柄、deadline、
终态观测器、两侧 capture、准入、provider/delivery 事实。plan / trace_id /
report_kind / report_context / candidate 起始时间戳都归 lifecycle。
HTTP 侧不接线：execution_runtime/stream/execution.rs 的
DirectPassthroughFinalizerCore（38 字段）与 failover / oauth 重试 / prefetch 深度
纠缠，无法在「行为等价 + 单 commit 可验证」的前提下改动。逐调用点映射表写在
模块文档注释里作为后续 PR 的接线依据。验收：git diff 对
execution_runtime/stream/ 与 crates/aether-usage 均为零 diff。
新增 6 个测试：效果段超时后仍走兜底 lease 释放、Unbounded 会一直等、detachable
写入在调用方停止等待后仍跑完、settle 四段顺序（计数器替身）、body capture 的
SSE 形状与编码状态（并显式记录默认上限是 usize::MAX，截断分支不可达）、
candidate error_type 映射。
…trips
评审第 5 条：Responses WebSocket 收到的本来就是结构化协议事件，但为了复用面向
SSE 的 push_line，观测路径要先把每个事件序列化成 data: {json}\n\n，解析器再
decode 回 Value——一次纯粹的往返。这个「伪 SSE」形状是随手拼的，一旦拼装函数
以后被加上换行或分块逻辑，观测结果就会和真实事件悄悄分叉。
aether-ai-formats：
- OpenAIResponsesProviderState::push_line 机械拆成 decode + push_event，
push_line 现在只做解码。协议状态机一行未动，diff 里除函数签名外只有
&value → value（value 从拥有改成借用，持有结构化事件的传输不必为了调用它
先克隆一份）。
- StreamingStandardTerminalObserver::push_event 走 TerminalStreamParser::Standard，
service tier 的记录方式与 push_line 完全相同。openai:image 的终态状态机按 SSE
行做增量解析、没有结构化入口，返回 AiSurfaceFinalizeError 让调用方
disable_with_error 标记 parser_error，而不是静默丢事件、把摘要留成「未观察到
终态」。ProviderStreamParser 的其余三个格式同样返回 Err：机械拆分随时可做，
但不建无调用方的接口。
WS 侧：
- 新增 responses/observation.rs 的 ResponsesStructuredTerminalObserver，直接消费
frame.protocol_events() 借出的事件。包一层的意义是让「不再拼 SSE」成为类型层面
的事实——这个类型没有任何接受字节的方法，改回 push_line 不可能悄悄发生。
finish() 里的 Ok(None) / Err → disable_with_error 兜底也一并收进来。
- body capture 不动，仍然是 SSE 形状（data: 开头、\n\n 结尾）：
aether_usage_runtime::report 用 line.strip_prefix("data:") 解析被捕获的 body
判定 StreamCapturedTerminalState，而它是 stream_report_represents_failure 的一个
OR 项，换成结构化 JSON 会让终态判定恒为 Missing。capture_sse_event /
capture_client_frame / websocket_event_as_sse_line 全部保留，原因写在模块文档
注释里。这一层只换观测，不换捕获。
差分测试（8 个，aether-ai-formats）：同一组事件序列分别走 push_line 与
push_event，断言 ExecutionStreamTerminalSummary 完全相等——批量 delta 序列、
completed 带 usage、合法 incomplete、error、response.failed、未知事件、
service tier、缺终态；外加 openai:image 拒绝结构化入口。两条入口不可能有
过滤差异：任何 Value 序列化出来都不会命中 decode_json_data_line 的 empty /
":" / "event:" / [DONE] 四个过滤条件。
turn.rs 里三个既有的 WS 观测测试改走结构化入口；SSE 形状的断言留在 capture 一侧。
验收：crates/aether-usage 零 diff。
评审第 2 条。配额透明重试原来的顺序是「detach 旧 attempt → 规划并绑定新 attempt → 把旧 attempt 的结算排进队列」。规划因此读到的是旧 attempt 还没投射的 health / adaptive / pool 状态，而且旧 attempt 仍占着自己的 pool key lease——替代 key 的挑选看到的是一把仍被占用的 key，最坏情况下判成「无可用供应商」而放弃一次 本可以成功的重试。 普通的新 turn 早就挡住了这件事：client.rs 在处理 response.create 前调用 await_pending_turn_finalization，注释写的正是「不要让新 turn 基于陈旧的 health / adaptive / pool 状态规划」。透明重试是同一个问题的另一条入口，漏了这一步。 现在顺序是：detach → 释放准入 → 结算旧 attempt 并等它落地 → 规划/绑定新 attempt。 新增 lifecycle::settle_turn_finalization：与 queue_turn_finalization 的区别只在于 「等」。后者把 handle 挂在连接上让 relay loop 继续跑，用在结算之后不再读取共享 状态的出口；前者用在必须先看到结算结果才能继续的路径上。 顺序用类型固定，而不是靠注释：settle_turn_finalization 返回 PreviousAttemptSettled，retry_active_turn_after_quota_exhaustion 要求这个参数。 凭证只能由 lifecycle 颁发（结算完成，或明确「没有 attempt 要结算」），所以把顺序 写反连编译都过不了。 重试失败路径随之变化：旧 attempt 已经结算，不再 resume 回去。logical turn 仍停在 Replanning，后续分支的 end() / finalize_active_turn 只清 logical turn、不交出 attempt，因此不存在重复结算。结算 outcome 取值不变（两条路径用的都是 terminal_outcome.unwrap_or_else(upstream_closed)，而这条分支里 terminal_outcome 必为 Some——usage_limit_error 成立意味着有一个已解析的 error 终态帧）。 代价（都落在「重试失败」这一侧，且只影响已终态 attempt 的报告注解，不影响计费）： - 那条最终转发给客户端的 429 事件不再进旧 attempt 的 client capture； provider 侧 capture 早在 observe_upstream_frame 里就记下了。 - 如果转发 429 给客户端也失败，record_client_delivery_aborted 落在一个已经结算的 attempt 上，成为 no-op。 测试： - lifecycle：await_turn_finalization_handle 必须「等到落地」而不是「排进队列」 （C6 依赖的性质）；结算完成后规划才读状态的顺序型断言（计数器替身）；结算任务 panic 也必须放行调用方，不能卡死 relay loop。 - turn_state：Replanning 状态下 end() 不再交出第二个 attempt（无重复结算）。 - e2e 新增 provider_quota_exhaustion_transparently_retries_onto_another_key： mock 上游首轮只回 Codex 的 429 usage_limit_reached，网关换到第二把 key 重放同一个 response.create；断言客户端看不到 429、上游被连两次、两次用的不是同一把 key、两个 attempt 各留一条终态行（429 的那条 + 计费的那条）。已验证它在改动前后都通过—— 它覆盖的是整条路径可用，顺序由上面的单测确定性覆盖。 夹具随之参数化出 ProviderFixture::CodexKeyPair：透明重试只有 Codex adapter 会 开启，而 codex 候选要求 auth_type = oauth，所以这个夹具用未过期的 oauth 凭证。
Responses WebSocket 只实现了脱敏的一半：请求侧 mask 之后，provider 事件帧在推给
客户端之前没有还原，于是 session 映射内的占位符以 <AETHER:EMAIL:...> 的形式直接
透给客户端。这里补齐响应侧，语义与 HTTP 路径对齐。
- 还原点是 relay loop 的最后一跳（send_client_message 之前、capture_client_frame
之前），对应 HTTP 的 restore_sync_response_body / StreamingResponseRestorer 所在
位置。审计与终态观测继续消费脱敏态事件，只有发往客户端的那一份拷贝被还原。
- 复用 privacy::restore_json_strings（改为 pub(crate)）与
RedactionSession::restore_text，不复制任何还原逻辑：只还原本 session mask 过的
映射，未映射的占位符原样保留；type / model / id 等协议字段不可能命中 sentinel，
因此不受影响。批量 {"chunks":[...]} 帧一并递归还原。
- session 生命周期：mask 仍然是 per-turn（slot 依旧每轮新建），但 session 改由连接
持有，按有界 FIFO 留最近 8 轮。理由是 WS 的会话历史留在上游，continuation 只发
增量输入，per-turn 释放会漏还原后续响应里回显的更早轮次占位符；HTTP 不会漏，是
因为它每次重发整段历史、重新 mask 会派生出同一个 sentinel。被挤出窗口的轮次退回
「占位符原样透传」，不会错误还原成别的值。
- 未命中还原时不改写字节；连接上没有任何 mask session 时（未启用脱敏）连事件 clone
都不做。
测试：redaction.rs 新增 8 条单测（还原命中/批量帧/未映射占位符原样/未命中不改写/
无 session 不介入/空 session 不留存/审计侧入参不被改写/跨轮还原/窗口有界）；
responses_websocket_e2e 新增一条用例，mock 上游回显收到的 input，断言上游只看到
占位符而客户端拿到真实邮箱。
AAEE86
force-pushed
the
feat/openai-responses-websocket
branch
from
August 3, 2026 13:23
a5fbeff to
b60119b
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
关联 #686
背景
为 Codex 与 OpenAI 的 Responses WebSocket 请求提供持久化网关代理。原实现已具备代理模式、连通性探针、e2e 套件与前端配置/用量展示；本 PR 在此之上完成评审提出的全部架构整改（9 条评审意见逐一对应），并补齐 PII 脱敏的响应侧还原，以重构与行为修正两类 commit 分层落地。
功能范围
responses_websocket_e2e集成测试架构整改（评审意见 → 方案）
BoundResponsesConnection用response_in_flight/active_turn/active_response_create三字段编码同一状态，8 组合仅 3 合法，非法状态靠调用点 if 防御ResponsesTurnState状态机枚举（140f0059c）ExecutionAttemptLifecycle（15bfcb45b），HTTP 映射表见文件头注释RedactionSessionSlot，fail-closed 路由（d3cc068eb）；响应侧还原补齐（582df57a5）response.incomplete一律按失败处理，误伤max_output_tokens合法截断PII 脱敏（请求侧 + 响应侧）
response.create在进入任何下游用途前只脱敏一次（真实 PII →<AETHER:...>占位符），四个规划入口经build_planning_parts注入 per-turn slot；continuation 复用已绑定 upstream、不进 planner 的路径在 redaction 层统一拦截。脱敏命中却取不到 session 直接失败，不静默旁路。privacy::restore_json_strings/RedactionSession::restore_text，零逻辑复制；只还原本连接 mask 过的映射，未映射占位符原样透传；未命中时保持上游原字节（无 session 时连 clone 都不做）。终态观测与重试时序
data:文本回环解析；StreamingStandardTerminalObserver解码 JSON 后直接以结构化事件投递，观测与捕获路径分离，杜绝"改回文本解析"的回归可能await_pending_turn_finalization语义——旧 attempt 先结算落地（health/adaptive/pool 状态投射、pool key lease 释放）再规划新 attempt；顺序由零大小凭证PreviousAttemptSettled在签名层面强制行为修正（需评审重点确认）
client_delivery_failed+websocket_client_delivery=aborted），不影响计费判定。影响收入口径。resume()回去（原 debug_assert 已删）；logical turn 停在 Replanning，后续分支仅清 logical turn 不交出 attempt，无重复结算（turn_state 单测钉住）。Commit 清单（15 个，含功能底座）
验证
cargo clippy -p aether-gateway --lib --bins --examples -- -D warnings：0 告警RUST_MIN_STACK=33554432 cargo nextest run -p aether-gateway --lib：3876 通过cargo nextest run -p aether-integration-tests --test responses_websocket_e2e：5 通过crates/aether-usage、execution_runtime/stream均无 diff新增 e2e 用例：
provider_quota_exhaustion_transparently_retries_onto_another_key（mock 上游首轮仅回 usage_limit_reached，断言客户端无 429、上游被连两次、两次 key 不同、两个 attempt 各留终态行）redacted_pii_is_restored_before_the_client_sees_a_provider_frame（mock 上游回显收到的 input，断言上游只看到占位符、客户端拿到真实邮箱；反证：临时禁用还原该用例即失败复现原缺陷）已知后续事项（不在本 PR）
max_output_tokens等白名单 incomplete 目前在 usage runtime 记为 Failed（report.rs:611 / write.rs:2901），与 55e585e 的网关语义不一致；同样影响 SSE 路径，涉及收入口径，需独立 PR