feat: add codex sse and usage helpers

Co-authored-by: Codex <noreply@openai.com>

docs/implementation-status.md

@@ -17,6 +17,6 @@ Update this file before each feature commit.
 | Translation | Completed | `eff182e` | `cargo test chat_completion_translates_to_codex_response_request`; full `fmt/test/clippy` before commit | Adds OpenAI Chat request types, Codex Responses request type, translation, and OpenAI-compatible error body helper. |
 | HTTP routes | Completed | `ab0ad8f` + `3830b21` | `cargo test v1_requires_client_api_key_not_admin_cookie`; `cargo test models_route_returns_openai_compatible_codex_model_list`; full `fmt/test/clippy` before commit | Adds Axum router, health route, `/v1/models`, `/v1` unauthorized OpenAI error body, global request-id response header, and binary listener. |
 | Admin/frontend response contract | Completed | `1c34ea7` + `b0c10f2` | `cargo test admin_envelope_serializes_lower_camel_case_request_id`; `cargo test admin_logs_reject_missing_admin_session_cookie`; `cargo test v1_response_has_request_id_header_without_admin_body_field`; full `fmt/test/clippy` before commit | Adds reusable admin envelopes with body codes, lower camelCase JSON, paginated list shape, requestId bodies, admin session cookie enforcement, and keeps `/v1/*` OpenAI-compatible. |
-| Upstream lifecycle | Completed | `88cad87` | `cargo test responses_route_rejects_non_codex_provider_models`; `cargo test v1_requires_client_api_key_not_admin_cookie`; full `fmt/test/clippy` before commit | `/v1` accepts only `Bearer cpr_...`, rejects non-Codex provider models, and returns OpenAI-compatible errors. |
+| Upstream lifecycle | Completed | `88cad87` + SSE/usage module commit | `cargo test responses_route_rejects_non_codex_provider_models`; `cargo test v1_requires_client_api_key_not_admin_cookie`; `cargo test --test codex_sse_usage_test`; full `fmt/test/clippy` before commit | `/v1` accepts only `Bearer cpr_...`, rejects non-Codex provider models, returns OpenAI-compatible errors, and includes reusable SSE frame parsing/encoding plus token usage extraction utilities. |
 | Fingerprint updates | Completed | `dd26d31` + `e32220a` | `cargo test update_manifest_updates_app_version_and_build_number`; `cargo test fingerprint_updater_fetches_manifest_and_persists_history`; full `fmt/test/clippy` before commit | Parses Codex Desktop update manifests, fetches update source payloads, and persists app version/build number history to SQLite. |
 | Runtime docs and packaging | Completed | `33b63f3` | `cargo fmt --check`; `cargo clippy --all-targets --all-features --locked -- -D warnings`; `cargo test`; `cargo run`; `curl -s -i http://127.0.0.1:8080/health` | README scope, `.env.example`, Dockerfile, compose file, and health smoke test are complete. |

docs/superpowers/plans/2026-06-11-codex-proxy-rs.md

@@ -2954,6 +2954,7 @@ git commit -m "docs: document codex-proxy-rs scope"
 
 - [x] Add `POST /admin/login` so admin login is a password-backed session flow, not a client API key flow. The route issues an HttpOnly `cpr_admin_session` cookie, returns the admin envelope with lower camelCase `expiresAt`, and rejects `Bearer cpr_...` as admin credentials.
 - [x] Replace the refresh policy shell with a native refresh scheduler. It refreshes before expiry, refreshes immediately after an upstream 401 trigger, preserves the existing refresh token when the refresh response omits one, maps refresh failures to account statuses, and enforces configured concurrency with a semaphore.
+- [x] Fill the previously empty `src/codex/sse.rs` and `src/codex/usage.rs` modules with tested SSE frame parsing/encoding and token usage extraction helpers.
 
 ---
 

src/auth/api_key.rs

@@ -34,7 +34,7 @@ impl ApiKeyHasher {
     pub fn generate_client_api_key(&self, _name: &str) -> GeneratedClientApiKey {
         let mut bytes = [0u8; 32];
         rand::rng().fill_bytes(&mut bytes);
-        // 中文注释：cpr_ 只用于客户端调用 /v1，不能复用成管理员登录密码。
+        // cpr_ 只用于客户端调用 /v1，不能复用成管理员登录密码。
         let plaintext = format!("cpr_{}", URL_SAFE_NO_PAD.encode(bytes));
         let prefix = plaintext.chars().take(12).collect::<String>();
         let key_hash = self.hash_client_api_key(&plaintext);

src/auth/refresh.rs

@@ -89,7 +89,7 @@ where
             .acquire()
             .await
             .map_err(|_| RefreshError::ConcurrencyClosed)?;
-        // 中文注释：刷新响应未返回 refresh_token 时必须保留旧值，避免把账号永久刷新能力清空。
+        // 刷新响应未返回 refresh_token 时必须保留旧值，避免把账号永久刷新能力清空。
         match self.client.refresh(refresh_token).await {
             Ok(token_pair) => Ok(apply_token_pair(account, token_pair)),
             Err(RefreshFailure::Transport) => Err(RefreshError::Transport),

src/codex/client.rs

@@ -1,7 +1,7 @@
 use reqwest::Client;
 
 pub fn build_reqwest_client(force_http11: bool) -> Result<Client, reqwest::Error> {
-    // 中文注释：Codex Desktop 指纹依赖 reqwest/rustls 组合，升级前必须重新验证 TLS 行为。
+    // Codex Desktop 指纹依赖 reqwest/rustls 组合，升级前必须重新验证 TLS 行为。
     let builder = Client::builder()
         .use_rustls_tls()
         .no_proxy()

src/codex/headers.rs

@@ -18,7 +18,7 @@ pub fn build_codex_headers(
     headers.insert("originator".to_string(), fp.originator.clone());
     headers.insert("user-agent".to_string(), fp.user_agent());
     headers.insert("x-client-request-id".to_string(), request_id.to_string());
-    // 中文注释：这些 Codex Desktop 私有头是上游识别链路的一部分，不能按普通 OpenAI API 精简。
+    // 这些 Codex Desktop 私有头是上游识别链路的一部分，不能按普通 OpenAI API 精简。
     headers.insert(
         "x-openai-internal-codex-residency".to_string(),
         "global".to_string(),

src/codex/sse.rs

@@ -1 +1,111 @@
+use thiserror::Error;
 
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct SseEvent {
+    pub event: Option<String>,
+    pub data: String,
+    pub id: Option<String>,
+    pub retry: Option<u64>,
+}
+
+#[derive(Debug, Error, PartialEq, Eq)]
+pub enum SseError {
+    #[error("invalid SSE retry value: {0}")]
+    InvalidRetry(String),
+}
+
+#[derive(Debug, Default)]
+struct EventBuilder {
+    event: Option<String>,
+    data: String,
+    has_data: bool,
+    id: Option<String>,
+    retry: Option<u64>,
+}
+
+impl EventBuilder {
+    fn push_data(&mut self, value: &str) {
+        if self.has_data {
+            self.data.push('\n');
+        }
+        self.data.push_str(value);
+        self.has_data = true;
+    }
+
+    fn finish(&mut self) -> Option<SseEvent> {
+        if !self.has_data {
+            self.event = None;
+            self.id = None;
+            self.retry = None;
+            return None;
+        }
+        self.has_data = false;
+        Some(SseEvent {
+            event: self.event.take(),
+            data: std::mem::take(&mut self.data),
+            id: self.id.take(),
+            retry: self.retry.take(),
+        })
+    }
+}
+
+pub fn parse_sse_events(input: &str) -> Result<Vec<SseEvent>, SseError> {
+    let mut events = Vec::new();
+    let mut builder = EventBuilder::default();
+
+    for raw_line in input.lines() {
+        let line = raw_line.strip_suffix('\r').unwrap_or(raw_line);
+        if line.is_empty() {
+            if let Some(event) = builder.finish() {
+                events.push(event);
+            }
+            continue;
+        }
+        if line.starts_with(':') {
+            continue;
+        }
+
+        let (field, value) = split_sse_field(line);
+        match field {
+            "event" => builder.event = Some(value.to_string()),
+            "data" => builder.push_data(value),
+            "id" if !value.contains('\0') => builder.id = Some(value.to_string()),
+            "retry" => {
+                builder.retry = Some(
+                    value
+                        .parse::<u64>()
+                        .map_err(|_| SseError::InvalidRetry(value.to_string()))?,
+                );
+            }
+            _ => {}
+        }
+    }
+
+    if let Some(event) = builder.finish() {
+        events.push(event);
+    }
+    Ok(events)
+}
+
+pub fn encode_sse_event(event: &str, data: &str) -> String {
+    let mut frame = String::new();
+    if !event.is_empty() {
+        frame.push_str("event: ");
+        frame.push_str(event);
+        frame.push('\n');
+    }
+    for line in data.split('\n') {
+        frame.push_str("data: ");
+        frame.push_str(line);
+        frame.push('\n');
+    }
+    frame.push('\n');
+    frame
+}
+
+fn split_sse_field(line: &str) -> (&str, &str) {
+    let Some((field, value)) = line.split_once(':') else {
+        return (line, "");
+    };
+    (field, value.strip_prefix(' ').unwrap_or(value))
+}

src/codex/usage.rs

@@ -1 +1,81 @@
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
 
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct TokenUsage {
+    pub input_tokens: u64,
+    pub output_tokens: u64,
+    pub cached_tokens: u64,
+    pub total_tokens: u64,
+}
+
+impl TokenUsage {
+    pub fn new(input_tokens: u64, output_tokens: u64, cached_tokens: u64) -> Self {
+        Self {
+            input_tokens,
+            output_tokens,
+            cached_tokens,
+            total_tokens: input_tokens + output_tokens,
+        }
+    }
+
+    pub fn merged(self, other: Self) -> Self {
+        Self {
+            input_tokens: self.input_tokens + other.input_tokens,
+            output_tokens: self.output_tokens + other.output_tokens,
+            cached_tokens: self.cached_tokens + other.cached_tokens,
+            total_tokens: self.total_tokens + other.total_tokens,
+        }
+    }
+}
+
+pub fn extract_usage(body: &Value) -> Option<TokenUsage> {
+    let usage = body.get("usage").unwrap_or(body);
+    if !usage.is_object() {
+        return None;
+    }
+
+    let input_tokens = number_field(usage, "input_tokens")
+        .or_else(|| number_field(usage, "prompt_tokens"))
+        .unwrap_or_default();
+    let output_tokens = number_field(usage, "output_tokens")
+        .or_else(|| number_field(usage, "completion_tokens"))
+        .unwrap_or_default();
+    let cached_tokens = nested_number_field(usage, &["input_tokens_details", "cached_tokens"])
+        .or_else(|| nested_number_field(usage, &["prompt_tokens_details", "cached_tokens"]))
+        .or_else(|| number_field(usage, "cached_tokens"))
+        .unwrap_or_default();
+    let total_tokens = number_field(usage, "total_tokens").unwrap_or(input_tokens + output_tokens);
+
+    let has_usage = [
+        "input_tokens",
+        "output_tokens",
+        "prompt_tokens",
+        "completion_tokens",
+        "cached_tokens",
+        "total_tokens",
+    ]
+    .iter()
+    .any(|field| usage.get(*field).is_some())
+        || usage.get("input_tokens_details").is_some()
+        || usage.get("prompt_tokens_details").is_some();
+    has_usage.then_some(TokenUsage {
+        input_tokens,
+        output_tokens,
+        cached_tokens,
+        total_tokens,
+    })
+}
+
+fn number_field(value: &Value, field: &str) -> Option<u64> {
+    value.get(field)?.as_u64()
+}
+
+fn nested_number_field(value: &Value, path: &[&str]) -> Option<u64> {
+    let mut current = value;
+    for segment in path {
+        current = current.get(*segment)?;
+    }
+    current.as_u64()
+}

src/cookies/jar.rs

@@ -38,7 +38,7 @@ impl CookieJar {
     }
 
     pub fn cookie_header(&self, account_id: &str, domain: &str) -> Option<String> {
-        // 中文注释：Cloudflare Cookie 必须按账号隔离，不能在不同 Codex 账号之间复用。
+        // Cloudflare Cookie 必须按账号隔离，不能在不同 Codex 账号之间复用。
         let cookies = self.by_account.get(account_id)?;
         let pairs = cookies
             .iter()

src/fingerprint/updater.rs

@@ -28,7 +28,7 @@ struct Manifest {
 }
 
 pub fn parse_update_manifest(input: &str) -> Result<FingerprintUpdate, FingerprintError> {
-    // 中文注释：自动更新只同步桌面端指纹字段，不把远端配置当作运行时业务配置执行。
+    // 自动更新只同步桌面端指纹字段，不把远端配置当作运行时业务配置执行。
     let manifest: Manifest = serde_json::from_str(input)?;
     Ok(FingerprintUpdate {
         app_version: manifest.version,

src/http/admin.rs

@@ -35,7 +35,7 @@ impl<T> AdminEnvelope<T> {
         data: T,
         request_id: impl Into<String>,
     ) -> Self {
-        // 中文注释：body code 给前端做业务分支，HTTP status 仍然是传输层真相。
+        // body code 给前端做业务分支，HTTP status 仍然是传输层真相。
         Self {
             code,
             message: message.into(),
@@ -161,7 +161,7 @@ pub async fn login(
         }
     };
 
-    // 中文注释：admin 登录只校验管理员密码；客户端 cpr_ API Key 不能参与后台登录。
+    // admin 登录只校验管理员密码；客户端 cpr_ API Key 不能参与后台登录。
     match verify_admin_password(&payload.password, &admin.password_hash) {
         Ok(true) => {}
         Ok(false) => {

src/http/middleware.rs

@@ -31,7 +31,7 @@ pub async fn attach_request_id(mut request: Request, next: Next) -> Response {
         .and_then(RequestId::from_header)
         .unwrap_or_else(RequestId::generate);
 
-    // 中文注释：同一个 requestId 同时进入日志、响应头和 admin body，方便跨层排查。
+    // 同一个 requestId 同时进入日志、响应头和 admin body，方便跨层排查。
     request.extensions_mut().insert(request_id.clone());
     let mut response = next.run(request).await;
     if let Ok(value) = HeaderValue::from_str(request_id.as_str()) {

src/logs/rotation.rs

@@ -149,7 +149,7 @@ impl RotatingLogInner {
 pub fn init_tracing(config: RotationConfig) -> Result<RotatingLogWriter, LogError> {
     let writer = RotatingLogWriter::new(config)?;
     let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
-    // 中文注释：日志必须写入可轮转文件，不能只依赖 stdout，否则线上问题无法回溯。
+    // 日志必须写入可轮转文件，不能只依赖 stdout，否则线上问题无法回溯。
     tracing_subscriber::fmt()
         .json()
         .with_env_filter(filter)

src/storage/db.rs

@@ -10,7 +10,7 @@ pub async fn connect_sqlite(database_url: &str) -> Result<SqlitePool, sqlx::Erro
         .foreign_keys(true)
         .busy_timeout(Duration::from_secs(5));
 
-    // 中文注释：WAL 和外键必须在连接层统一启用，避免后续仓储绕过一致性约束。
+    // WAL 和外键必须在连接层统一启用，避免后续仓储绕过一致性约束。
     let pool = SqlitePool::connect_with(options).await?;
     sqlx::migrate!("./migrations").run(&pool).await?;
     Ok(pool)

tests/codex_sse_usage_test.rs

@@ -0,0 +1,102 @@
+use serde_json::json;
+
+use codex_proxy_rs::codex::{
+    sse::{encode_sse_event, parse_sse_events},
+    usage::{extract_usage, TokenUsage},
+};
+
+#[test]
+fn parse_sse_events_combines_multiline_data_and_metadata() {
+    let input = concat!(
+        ": keepalive\n",
+        "id: evt_1\n",
+        "event: response.output_text.delta\n",
+        "data: {\"delta\":\"hel\"}\n",
+        "data: {\"delta\":\"lo\"}\n",
+        "retry: 5000\n",
+        "\n",
+    );
+
+    let events = parse_sse_events(input).unwrap();
+
+    assert_eq!(events.len(), 1);
+    assert_eq!(events[0].id.as_deref(), Some("evt_1"));
+    assert_eq!(
+        events[0].event.as_deref(),
+        Some("response.output_text.delta")
+    );
+    assert_eq!(events[0].data, "{\"delta\":\"hel\"}\n{\"delta\":\"lo\"}");
+    assert_eq!(events[0].retry, Some(5000));
+}
+
+#[test]
+fn parse_sse_events_accepts_done_sentinel_without_event_type() {
+    let events = parse_sse_events("data: [DONE]\r\n\r\n").unwrap();
+
+    assert_eq!(events[0].data, "[DONE]");
+}
+
+#[test]
+fn encode_sse_event_emits_event_and_data_frame() {
+    let encoded = encode_sse_event("error", r#"{"error":{"code":"upstream_error"}}"#);
+
+    assert_eq!(
+        encoded,
+        "event: error\ndata: {\"error\":{\"code\":\"upstream_error\"}}\n\n"
+    );
+}
+
+#[test]
+fn extract_usage_reads_codex_token_usage_shape() {
+    let body = json!({
+        "usage": {
+            "input_tokens": 12,
+            "output_tokens": 5,
+            "input_tokens_details": {
+                "cached_tokens": 3
+            }
+        }
+    });
+
+    let usage = extract_usage(&body).unwrap();
+
+    assert_eq!(
+        usage,
+        TokenUsage {
+            input_tokens: 12,
+            output_tokens: 5,
+            cached_tokens: 3,
+            total_tokens: 17,
+        }
+    );
+}
+
+#[test]
+fn extract_usage_reads_openai_token_usage_shape_and_merges() {
+    let first = extract_usage(&json!({
+        "usage": {
+            "prompt_tokens": 8,
+            "completion_tokens": 4,
+            "prompt_tokens_details": {
+                "cached_tokens": 2
+            }
+        }
+    }))
+    .unwrap();
+    let second = TokenUsage {
+        input_tokens: 1,
+        output_tokens: 2,
+        cached_tokens: 0,
+        total_tokens: 3,
+    };
+
+    assert_eq!(
+        first.merged(second),
+        TokenUsage {
+            input_tokens: 9,
+            output_tokens: 6,
+            cached_tokens: 2,
+            total_tokens: 15,
+        }
+    );
+}