传输层抽象：HTTP/SSE/WebSocket/WebRTC 统一

传输层抽象：HTTP/SSE/WebSocket/WebRTC 统一

学习目标

• 理解 Codex 的 HttpTransport trait 统一抽象
• 掌握 Codex 的 SSE、WebSocket 和 WebRTC 三种传输实现
• 分析 Codex 的自定义 CA 证书注入和 OpenTelemetry 集成

---

项目实践

传输层架构

Codex 的传输层位于 codex-rs/codex-api/src/endpoint/：

codex-api/src/endpoint/
- responses.rs                # HTTP/SSE 传输
- responses_websocket.rs      # WebSocket 传输
- realtime_call.rs            # WebRTC 实时 API
- transport.rs                # HttpTransport trait
- ...

HttpTransport Trait

transport.rs 定义统一抽象：

#[async_trait]
pub trait HttpTransport: Send + Sync {
    async fn execute(
        &self,
        req: Request,
    ) -> Result<Response, TransportError>;

    async fn stream(
        &self,
        req: Request,
    ) -> Result<StreamResponse, TransportError>;
}

实现者：ReqwestTransport 基于 CodexHttpClient 实现，统一处理请求/响应日志、自定义 CA、超时配置。

HTTP/SSE 传输

responses.rs 实现标准 HTTP 流式请求：

impl ResponsesClient {
    pub async fn stream_request(
        &self,
        request: ResponsesRequest,
    ) -> Result<ResponseStream> {
        let req = self.build_request(request);
        let stream = self.transport.stream(req).await?;
        Ok(ResponseStream::new(stream))
    }
}

SSE 解析：codex-api/src/sse/responses.rs 解析 Server-Sent Events：

• response.output_item.done：输出项完成
• response.completed：整体完成
• response.failed：错误分类（context_window、quota、server_overloaded 等）

错误分类：Codex 对 response.failed 进行细粒度分类，不同错误类型触发不同的处理策略（如 server_overloaded 触发重试，context_window 触发压缩）。

WebSocket 传输

responses_websocket.rs 实现 WebSocket 升级：

特点：

• 支持自定义 CA（通过 rustls 配置）
• 事件结构与 SSE 对齐，上层无需区分传输方式
• 适合低延迟场景

WebRTC 实时 API

realtime_call.rs 实现 WebRTC 通信：

impl RealtimeCallClient {
    pub async fn create_call(
        &self,
        offer: SdpOffer,
    ) -> Result<SdpAnswer> {
        let req = self.build_multipart_request(offer);
        let resp = self.transport.execute(req).await?;
        Ok(SdpAnswer::from(resp))
    }
}

流程：

1. 客户端创建 SDP offer
2. POST /realtime/calls（multipart/form-data）
3. 服务端返回 SDP answer
4. 完成 P2P 媒体通道建立

自定义 CA 证书

Codex 支持通过环境变量注入自定义根证书：

export CODEX_CA_CERTIFICATE=/path/to/ca.crt
# 或
export SSL_CERT_FILE=/path/to/ca.crt

实现：CodexHttpClient 在创建 reqwest::Client 时读取环境变量，将自定义 CA 加入信任链。

OpenTelemetry 集成

Codex 在传输层集成 OpenTelemetry 追踪：

async fn run_with_request_telemetry<F, Fut, T>(
    &self,
    operation: F,
) -> Result<T>
where
    F: FnOnce() -> Fut,
    Fut: Future<Output = Result<T>>,
{
    let span = info_span!("http_request", url = %self.url);
    let _enter = span.enter();

    let start = Instant::now();
    let result = operation().await;
    let duration = start.elapsed();

    // 记录 metrics
    record_histogram!("http.request.duration", duration.as_secs_f64());

    result
}

---

问题与规避

长连接保活

Codex 对流式请求不设总超时（timeout(None)），但设置了读取超时。这避免了长回复被中断，同时可以检测网络断开。

压缩与流式传输

Codex 支持 zstd 响应压缩。在流式传输场景下，压缩是透明的——SSE 解析器在解压后处理事件。

---

设计取舍

SSE vs WebSocket vs WebRTC

传输	延迟	复杂度	适用场景
SSE	中	低	标准流式文本
WebSocket	低	中	低延迟交互
WebRTC	极低	高	实时语音/视频

Codex 同时支持三种传输，由配置或自动协商选择。SSE 是默认选项，WebSocket 和 WebRTC 用于特定场景。

统一 trait vs 特化实现

Codex 使用统一的 HttpTransport trait，但 WebRTC 的 API 差异较大（SDP 交换而非 HTTP 请求）。实际实现中，WebRTC 有独立的客户端，不直接实现 HttpTransport。

---

参考来源

• OpenAI API Reference
• WebRTC Specification