feat: wire v1 responses to codex upstream

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

docs/api.md

@@ -39,6 +39,12 @@ data: {"error":{"message":"Upstream failed","type":"server_error","param":null,"
 
 Model name parsing supports configured aliases plus `-low`, `-medium`, `-high`, `-xhigh`, `-fast`, and `-flex` suffixes.
 
+### `POST /v1/responses`
+
+Uses imported Codex accounts to call `POST /codex/responses` on the configured Codex backend. The upstream request is sent with Codex Desktop headers, account bearer token, optional account id, request id, and encrypted account-scoped Cookie replay.
+
+The current Rust route collects HTTP SSE until `response.completed` and returns the completed OpenAI-compatible response JSON. `previous_response_id` and explicit WebSocket-only requests are rejected until the WebSocket transport is implemented and verified.
+
 ## `/admin/*`
 
 Admin endpoints are authenticated only by HttpOnly admin session cookies.

docs/implementation-status.md

@@ -13,14 +13,14 @@ Update this file before each feature commit.
 | Logging and pagination | Completed | `eb0341e` + `fc16c48` + `b0c10f2` | `cargo test event_logs_are_cursor_paginated`; `cargo test admin_logs_are_cursor_paginated_and_include_request_id`; `cargo test rotating_writer_splits_files_when_size_limit_is_reached`; full `fmt/test/clippy` before commit | Adds cursor pagination, `Page<T>` camelCase serialization, admin `/admin/logs` query API, event log repository, runtime tracing initialization, HTTP trace layer, and date/size-based file rotation. |
 | TLS headers and fingerprint | Completed | `1bdd08c` | `cargo test codex_headers_include_desktop_identity_and_turn_state`; `cargo tree \| rg 'reqwest\|rustls'`; full `fmt/test/clippy` before commit | Adds Codex Desktop fingerprint model, exact identity/request headers, and pinned reqwest/rustls client builder. |
 | Cookie persistence | Completed | `4710926` + `5d3cc6e` | `cargo test cookie_jar_captures_and_replays_account_scoped_cookies`; `cargo test cookie_repository_encrypts_and_replays_account_scoped_cookies`; full `fmt/test/clippy` before commit | Adds account-scoped in-memory Cookie jar and encrypted SQLite Cookie repository; Cookies are replayed only for the same account/domain. |
-| Account pool and refresh | Partial | `0f97c5c` + `2f25db8` + `4dc0c9a` | `cargo test account_pool_skips_expired_disabled_banned_and_quota_exhausted_accounts`; `cargo test --test account_pool_scheduling_test`; `cargo test --test account_refresh_test`; full `fmt/test/clippy` before commit | Implements base account status filtering, refresh scheduler semantics, max concurrent slots, stale slot cleanup, least-used/round-robin/sticky strategies, tier priority, quota-cache filtering, Cloudflare cooldown skip, request staggering metadata, preferred account selection, exclude IDs, and capacity summary. Still missing settings-backed scheduling config, SQLite repository integration for pool snapshots/release usage, upstream retry/fallback wiring, and admin account CRUD. |
+| Account pool and refresh | Partial | `0f97c5c` + `2f25db8` + `4dc0c9a` + `pending` | `cargo test account_pool_skips_expired_disabled_banned_and_quota_exhausted_accounts`; `cargo test --test account_pool_scheduling_test`; `cargo test --test account_refresh_test`; `cargo test --test v1_upstream_route_test`; full `fmt/test/clippy` before commit | Implements base account status filtering, refresh scheduler semantics, max concurrent slots, stale slot cleanup, least-used/round-robin/sticky strategies, tier priority, quota-cache filtering, Cloudflare cooldown skip, request staggering metadata, preferred account selection, exclude IDs, capacity summary, settings-backed pool construction, and same-process import-to-pool sync. Still missing startup pool snapshots from SQLite, durable release-time slot state, upstream retry/fallback wiring, and admin account CRUD. |
 | Account repository | Completed | `14fe706` | `cargo test --test account_repository_test`; `cargo fmt --check`; `cargo clippy --all-targets --all-features --locked -- -D warnings`; `cargo test` | Adds encrypted SQLite account token storage, decrypting reads, cursor pagination, status/label updates, release-time usage counter persistence, and refresh-token-preserving token updates. |
 | 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 | Partial | `ab0ad8f` + `3830b21` + `6179d44` + `cfec8ba` + `6341336` + `228febd` | `cargo test v1_requires_client_api_key_not_admin_cookie`; `cargo test models_route_returns_openai_compatible_codex_model_list`; `cargo test --test admin_settings_route_test`; `cargo test --test model_catalog_test`; `cargo test --test admin_accounts_route_test`; full `fmt/test/clippy` before commit | Adds Axum router, health route, `/v1/models`, `/v1/models/catalog`, `/v1/models/{id}`, `/v1/models/{id}/info`, `/debug/models`, `/admin/accounts`, `/admin/accounts/import`, `/v1` unauthorized OpenAI error body, global request-id response header, binary listener, `gpt-5.5` as the default/test model baseline, and read-only `/admin/settings` with admin-session auth and lower camelCase envelope. `/admin/accounts` reads metadata only and does not decrypt tokens for list views. Still missing OAuth/device/CLI account login flows, admin account mutation/health/quota/Cookie routes, admin API-key management routes, settings mutation/persistence, diagnostics, usage stats, backend model refresh, model cache persistence, and additional Responses routes. |
+| HTTP routes | Partial | `ab0ad8f` + `3830b21` + `6179d44` + `cfec8ba` + `6341336` + `228febd` + `pending` | `cargo test v1_requires_client_api_key_not_admin_cookie`; `cargo test models_route_returns_openai_compatible_codex_model_list`; `cargo test --test admin_settings_route_test`; `cargo test --test model_catalog_test`; `cargo test --test admin_accounts_route_test`; `cargo test --test v1_upstream_route_test`; full `fmt/test/clippy` before commit | Adds Axum router, health route, `/v1/models`, `/v1/models/catalog`, `/v1/models/{id}`, `/v1/models/{id}/info`, `/debug/models`, `/admin/accounts`, `/admin/accounts/import`, `/v1/responses` HTTP SSE collection through imported Codex accounts, `/v1` unauthorized OpenAI error body, global request-id response header, binary listener, `gpt-5.5` as the default/test model baseline, and read-only `/admin/settings` with admin-session auth and lower camelCase envelope. `/admin/accounts` reads metadata only and does not decrypt tokens for list views. Still missing OAuth/device/CLI account login flows, admin account mutation/health/quota/Cookie routes, admin API-key management routes, settings mutation/persistence, diagnostics, usage stats, backend model refresh, model cache persistence, streaming passthrough, and WebSocket Responses routes. |
 | 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. |
 | HTTP auth boundary | Completed | `cec4607` | `cargo test --test http_auth_test`; `cargo test --test routes_responses_test`; `cargo test --test admin_logs_route_test`; `cargo fmt --check`; `cargo clippy --all-targets --all-features --locked -- -D warnings`; `cargo test` | Adds a real `src/http/auth.rs` boundary for client API key and admin session cookie parsing. `/v1/*` accepts only `Bearer cpr_...`, admin session parsing ignores client API keys, and route handlers no longer own ad hoc credential parsing. |
 | Model catalog | Partial | `6341336` | `cargo test --test model_catalog_test`; `cargo test model_catalog_route_returns_codex_metadata_without_alias_entries`; full `fmt/test/clippy` before commit | Adds a native static Codex model catalog with `gpt-5.5` default, configured aliases, reasoning/service-tier suffix parsing, OpenAI-compatible `/v1/models`, extended catalog/detail/info routes, and debug summary. Still missing backend Codex model fetching per plan, cache persistence, `/admin/refresh-models`, plan allowlist integration from fetched models, and full model metadata parity. |
-| Upstream lifecycle | Partial | `88cad87` + `225fbff` | `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` currently enforces client API key shape, rejects non-Codex provider models through the model catalog, returns OpenAI-compatible errors, and includes reusable SSE/usage helpers. It does not yet acquire accounts, forward to Codex, replay/capture Cookies, stream/collect upstream responses, capture rate-limit headers, retry across accounts, record usage, or preserve previous-response/session affinity. |
+| Upstream lifecycle | Partial | `88cad87` + `225fbff` + `pending` | `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`; `cargo test --test codex_client_test`; `cargo test --test v1_upstream_route_test`; full `fmt/test/clippy` before commit | `/v1` enforces client API key shape, rejects non-Codex provider models through the model catalog, returns OpenAI-compatible errors, includes reusable SSE/usage helpers, sends Codex Desktop HTTP SSE requests through imported accounts, replays/captures encrypted Cookies, collects `response.completed`, and records token usage. Still missing startup pool snapshot loading, rate-limit header capture, retry/fallback across accounts, 401 refresh integration, 429 quota cache, 402/403 classification, streaming passthrough, and previous-response/session affinity. |
 | Codex WebSocket boundary | Completed | `cc0516d` | `cargo test --test codex_websocket_test`; `cargo fmt --check`; `cargo clippy --all-targets --all-features --locked -- -D warnings`; `cargo test` | Adds `src/codex/websocket.rs` with tested transport classification for HTTP SSE versus WebSocket-only requests. `previous_response_id` and explicit WebSocket requests now fail through a typed boundary instead of being silently downgraded; real WebSocket transport remains an upstream lifecycle follow-up. |
 | Fingerprint updates | Partial | `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 a JSON update manifest and persists app version/build number history to SQLite. Still missing original appcast XML polling, update-state persistence, runtime header fingerprint selection from the latest stored fingerprint, and optional local Codex.app Chromium extraction. |
 | 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

@@ -2963,6 +2963,7 @@ git commit -m "docs: document codex-proxy-rs scope"
 - [x] Add read-only `/admin/settings` for the retained Rust settings surface. The response is admin-session gated, lower camelCase, and excludes proxy URL, Ollama, third-party provider, OpenAI official API key, proxy API key, and Electron/self-update fields. Persistent settings mutation remains a separate task.
 - [x] Add a native static model catalog foundation. It exposes `gpt-5.5`, configured aliases, reasoning/service tier suffix parsing, `/v1/models`, `/v1/models/catalog`, `/v1/models/{id}`, `/v1/models/{id}/info`, and `/debug/models`. Backend model fetching/cache and `/admin/refresh-models` remain separate tasks.
 - [x] Add encrypted admin account import/list foundations. `/admin/accounts/import` stores access/refresh tokens through `SecretBox` and `/admin/accounts` returns paginated sanitized account metadata without tokens or token decryption. The runtime now loads or creates the master key file before constructing repository-backed routes.
+- [x] Add the first native Codex upstream lifecycle slice. Imported accounts are synchronized into the process account pool, `/v1/responses` acquires an account, sends Codex Desktop HTTP SSE requests with encrypted Cookie replay/capture, collects `response.completed`, records usage, and releases the slot. Startup pool snapshots, streaming passthrough, retry/fallback, 401 refresh, rate-limit capture, and WebSocket remain separate tasks.
 
 ## 2026-06-11 Original Project Gap Audit
 
@@ -2979,8 +2980,8 @@ The Rust project has no legacy compatibility requirement, but the in-scope behav
 - [ ] **Diagnostics:** expand health/admin diagnostics to include authenticated state, pool summary, capacity summary, transport/fingerprint status, paths, runtime metadata, and test-connection checks. Keep production-local gating for sensitive debug endpoints.
 - [ ] **Logs and usage stats:** add `/admin/logs/state`, clear/detail, error-log equivalents if retained, and `/admin/usage-stats/summary` plus paginated or bounded history queries. Keep log file rotation and SQLite event pagination.
 - [ ] **Model catalog:** replace the hard-coded `/v1/models` list with a model store that supports static defaults, configured aliases, `-low/-medium/-high/-xhigh/-fast/-flex` suffix parsing, backend model fetching per plan, cache persistence, `/v1/models/catalog`, `/v1/models/{id}`, `/v1/models/{id}/info`, `/admin/refresh-models`, and `/debug/models`. Static defaults, aliases, suffix parsing, read-only model routes, and debug summary are complete. Backend fetch/cache, plan snapshots, and admin refresh remain.
-- [ ] **Account scheduling parity:** extend account acquisition beyond `active + least last_used`. Completed in the in-memory pool: max concurrent slots per account, stale slot cleanup, least-used/round-robin/sticky strategies, quota-cache skip, tier priority, model-plan filtering, exclude IDs, preferred account from session affinity, Cloudflare cooldown skip, request-staggering metadata, and capacity summary. Remaining: settings-backed scheduling config, pool snapshots from SQLite, release-time usage accounting through the repository, and upstream retry/fallback wiring.
-- [ ] **Upstream Codex lifecycle:** implement account acquire -> Codex request -> SSE/collect -> rate-limit capture -> usage release -> retry/fallback. Include Cookie replay/capture, Codex Desktop headers, `x-codex-turn-state`, `x-client-request-id`, upstream error classification, 401-triggered refresh, 429 quota cache, 402 quota exhausted, 403 ban/Cloudflare distinction, path-block Cookie clearing, and OpenAI-compatible error mapping.
+- [ ] **Account scheduling parity:** extend account acquisition beyond `active + least last_used`. Completed in the in-memory pool: max concurrent slots per account, stale slot cleanup, least-used/round-robin/sticky strategies, quota-cache skip, tier priority, model-plan filtering, exclude IDs, preferred account from session affinity, Cloudflare cooldown skip, request-staggering metadata, capacity summary, settings-backed construction, and import-to-pool sync. Remaining: startup pool snapshots from SQLite, durable release-time slot state, and upstream retry/fallback wiring.
+- [ ] **Upstream Codex lifecycle:** implement account acquire -> Codex request -> SSE/collect -> rate-limit capture -> usage release -> retry/fallback. Initial HTTP SSE collection is complete: `/v1/responses` acquires imported accounts, sends Codex Desktop headers, replays/captures Cookies, collects `response.completed`, records usage, and returns OpenAI-compatible JSON. Remaining: rate-limit capture, retry/fallback, 401-triggered refresh, 429 quota cache, 402 quota exhausted, 403 ban/Cloudflare distinction, path-block Cookie clearing, streaming passthrough, and richer OpenAI-compatible error mapping.
 - [ ] **Session affinity and previous response:** persist an in-memory response-id to account mapping with TTL, prompt-cache/conversation identity, turn-state replay, and preferred account selection. `src/codex/websocket.rs` must either implement the WebSocket path with verified TLS/header behavior or the API must reject WebSocket-only cases explicitly with documented limitations until fingerprint parity is proven.
 - [ ] **Fingerprint auto-update parity:** poll the real Codex Desktop appcast/update source, parse app version/build number, persist update state, select the latest stored fingerprint for runtime headers, and optionally extract Chromium version from a local Codex.app path when configured.
 

src/codex/client.rs

@@ -1,4 +1,141 @@
-use reqwest::Client;
+use reqwest::{
+    header::{HeaderMap, HeaderName, HeaderValue, CONTENT_TYPE, COOKIE, SET_COOKIE},
+    Client, StatusCode,
+};
+use serde_json::Value;
+use thiserror::Error;
+
+use crate::{
+    codex::{
+        headers::build_codex_headers,
+        sse::{parse_sse_events, SseError},
+        types::CodexResponsesRequest,
+        usage::{extract_usage, TokenUsage},
+        websocket::{ensure_http_sse_supported, WebSocketSupportError},
+    },
+    fingerprint::model::Fingerprint,
+};
+
+#[derive(Debug, Error)]
+pub enum CodexClientError {
+    #[error("http transport error: {0}")]
+    Http(#[from] reqwest::Error),
+    #[error("invalid request header name: {0}")]
+    InvalidHeaderName(#[from] reqwest::header::InvalidHeaderName),
+    #[error("invalid request header value: {0}")]
+    InvalidHeaderValue(#[from] reqwest::header::InvalidHeaderValue),
+    #[error("unsupported transport: {0}")]
+    UnsupportedTransport(#[from] WebSocketSupportError),
+    #[error("invalid upstream SSE response: {0}")]
+    InvalidSse(#[from] SseError),
+    #[error("upstream returned status {status}: {body}")]
+    Upstream { status: StatusCode, body: String },
+}
+
+pub type CodexClientResult<T> = Result<T, CodexClientError>;
+
+#[derive(Debug, Clone, Copy)]
+pub struct CodexRequestContext<'a> {
+    pub access_token: &'a str,
+    pub account_id: Option<&'a str>,
+    pub request_id: &'a str,
+    pub turn_state: Option<&'a str>,
+    pub cookie_header: Option<&'a str>,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct CodexBackendResponse {
+    pub body: String,
+    pub usage: Option<TokenUsage>,
+    pub turn_state: Option<String>,
+    pub set_cookie_headers: Vec<String>,
+}
+
+#[derive(Clone)]
+pub struct CodexBackendClient {
+    client: Client,
+    base_url: String,
+    fingerprint: Fingerprint,
+}
+
+impl CodexBackendClient {
+    pub fn new(client: Client, base_url: impl Into<String>, fingerprint: Fingerprint) -> Self {
+        Self {
+            client,
+            base_url: base_url.into().trim_end_matches('/').to_string(),
+            fingerprint,
+        }
+    }
+
+    pub async fn create_response(
+        &self,
+        request: &CodexResponsesRequest,
+        context: CodexRequestContext<'_>,
+    ) -> CodexClientResult<CodexBackendResponse> {
+        ensure_http_sse_supported(request)?;
+        let url = format!("{}/codex/responses", self.base_url);
+        let response = self
+            .client
+            .post(url)
+            .headers(self.request_headers(context)?)
+            .json(request)
+            .send()
+            .await?;
+        let status = response.status();
+        let turn_state = response
+            .headers()
+            .get("x-codex-turn-state")
+            .and_then(|value| value.to_str().ok())
+            .map(ToString::to_string);
+        let set_cookie_headers = response
+            .headers()
+            .get_all(SET_COOKIE)
+            .iter()
+            .filter_map(|value| value.to_str().ok().map(ToString::to_string))
+            .collect::<Vec<_>>();
+        let body = response.text().await?;
+        if !status.is_success() {
+            return Err(CodexClientError::Upstream { status, body });
+        }
+        let usage = extract_sse_usage(&body)?;
+
+        Ok(CodexBackendResponse {
+            body,
+            usage,
+            turn_state,
+            set_cookie_headers,
+        })
+    }
+
+    fn request_headers(&self, context: CodexRequestContext<'_>) -> CodexClientResult<HeaderMap> {
+        let mut headers = HeaderMap::new();
+        for (name, value) in build_codex_headers(
+            &self.fingerprint,
+            context.access_token,
+            context.account_id,
+            context.turn_state,
+            context.request_id,
+        ) {
+            headers.insert(
+                HeaderName::from_bytes(name.as_bytes())?,
+                HeaderValue::from_str(&value)?,
+            );
+        }
+        headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
+        headers.insert(
+            HeaderName::from_static("openai-beta"),
+            HeaderValue::from_static("responses_websockets=2026-02-06"),
+        );
+        headers.insert(
+            HeaderName::from_static("x-openai-internal-codex-residency"),
+            HeaderValue::from_static("us"),
+        );
+        if let Some(cookie_header) = context.cookie_header {
+            headers.insert(COOKIE, HeaderValue::from_str(cookie_header)?);
+        }
+        Ok(headers)
+    }
+}
 
 pub fn build_reqwest_client(force_http11: bool) -> Result<Client, reqwest::Error> {
     // Codex Desktop 指纹依赖 reqwest/rustls 组合，升级前必须重新验证 TLS 行为。
@@ -16,3 +153,25 @@ pub fn build_reqwest_client(force_http11: bool) -> Result<Client, reqwest::Error
     };
     builder.build()
 }
+
+fn extract_sse_usage(body: &str) -> CodexClientResult<Option<TokenUsage>> {
+    let events = parse_sse_events(body)?;
+    let mut usage: Option<TokenUsage> = None;
+    for event in events {
+        if event.data == "[DONE]" {
+            continue;
+        }
+        let Ok(value) = serde_json::from_str::<Value>(&event.data) else {
+            continue;
+        };
+        let event_usage =
+            extract_usage(&value).or_else(|| value.get("response").and_then(extract_usage));
+        if let Some(event_usage) = event_usage {
+            usage = Some(match usage {
+                Some(current) => current.merged(event_usage),
+                None => event_usage,
+            });
+        }
+    }
+    Ok(usage)
+}

src/http/admin.rs

@@ -15,7 +15,7 @@ use uuid::Uuid;
 
 use crate::{
     accounts::{
-        model::AccountStatus,
+        model::{Account, AccountStatus},
         repository::{NewAccount, StoredAccountMetadata},
     },
     auth::admin_session::verify_admin_password,
@@ -541,16 +541,38 @@ pub async fn import_accounts(
                 .into_response();
             }
         };
+        let email = empty_to_none(entry.email);
+        let account_id = empty_to_none(entry.account_id);
+        let user_id = empty_to_none(entry.user_id);
+        let label = empty_to_none(entry.label);
+        let plan_type = empty_to_none(entry.plan_type);
+        let refresh_token = empty_to_none(entry.refresh_token);
+        let access_token = token.to_string();
+        let pool_account = Account {
+            id: id.clone(),
+            email: email.clone(),
+            account_id: account_id.clone(),
+            user_id: user_id.clone(),
+            label: label.clone(),
+            plan_type: plan_type.clone(),
+            access_token: access_token.clone(),
+            refresh_token: refresh_token.clone(),
+            access_token_expires_at: None,
+            status,
+            quota_limit_reached: false,
+            cloudflare_cooldown_until: None,
+            added_at: chrono::Utc::now().to_rfc3339(),
+            last_used_at: None,
+        };
         let account = NewAccount {
             id,
-            email: empty_to_none(entry.email),
-            account_id: empty_to_none(entry.account_id),
-            user_id: empty_to_none(entry.user_id),
-            label: empty_to_none(entry.label),
-            plan_type: empty_to_none(entry.plan_type),
-            access_token: SecretString::new(token.to_string().into()),
-            refresh_token: empty_to_none(entry.refresh_token)
-                .map(|token| SecretString::new(token.into())),
+            email,
+            account_id,
+            user_id,
+            label,
+            plan_type,
+            access_token: SecretString::new(access_token.into()),
+            refresh_token: refresh_token.map(|token| SecretString::new(token.into())),
             access_token_expires_at: None,
             status,
         };
@@ -561,6 +583,7 @@ pub async fn import_accounts(
             )
             .into_response();
         }
+        state.account_pool().lock().await.insert(pool_account);
         imported += 1;
     }
 

src/http/v1.rs

@@ -3,13 +3,24 @@ use axum::{
     extract::{Path, State},
     http::{HeaderMap, StatusCode},
     response::IntoResponse,
-    Json,
+    Extension, Json,
 };
+use chrono::Utc;
+use reqwest::Url;
 use serde::Deserialize;
 use serde_json::{json, Value};
 
 use crate::{
-    http::auth::client_api_key, models::catalog::ModelCatalog, state::AppState,
+    accounts::{pool::AccountAcquireRequest, repository::UsageDelta},
+    codex::{
+        client::{build_reqwest_client, CodexBackendClient, CodexClientError, CodexRequestContext},
+        sse::parse_sse_events,
+        types::CodexResponsesRequest,
+    },
+    fingerprint::model::Fingerprint,
+    http::{auth::client_api_key, middleware::RequestId},
+    models::catalog::ModelCatalog,
+    state::AppState,
     translation::codex_to_openai::openai_error,
 };
 
@@ -18,10 +29,16 @@ const MODEL_CREATED_TIMESTAMP: i64 = 1_700_000_000;
 #[derive(Deserialize)]
 struct ResponsesBody {
     model: Option<String>,
+    input: Option<Vec<Value>>,
+    instructions: Option<String>,
+    reasoning: Option<Value>,
+    tools: Option<Vec<Value>>,
+    previous_response_id: Option<String>,
 }
 
 pub async fn responses(
     State(state): State<AppState>,
+    Extension(request_id): Extension<RequestId>,
     headers: HeaderMap,
     body: Bytes,
 ) -> impl IntoResponse {
@@ -30,18 +47,106 @@ pub async fn responses(
     }
 
     let default_model = state.config().model.default_model.clone();
-    let body = serde_json::from_slice::<ResponsesBody>(&body).unwrap_or_else(|_| ResponsesBody {
-        model: Some(default_model.clone()),
-    });
-    let model = body.model.unwrap_or(default_model);
+    let body = serde_json::from_slice::<ResponsesBody>(&body)
+        .unwrap_or_else(|_| default_body(default_model.clone()));
+    let requested_model = body.model.clone().unwrap_or(default_model);
     let catalog = ModelCatalog::from_config(&state.config().model);
-    if !catalog.is_recognized_model_name(&model) {
+    if !catalog.is_recognized_model_name(&requested_model) {
         return (
             StatusCode::NOT_FOUND,
             Json(openai_error("Model not found", "model_not_found")),
         );
     }
+    let parsed_model = catalog.parse_model_name(&requested_model);
+    let codex_request = CodexResponsesRequest {
+        model: parsed_model.model_id.clone(),
+        instructions: body.instructions.unwrap_or_default(),
+        input: body.input.unwrap_or_default(),
+        stream: true,
+        store: false,
+        reasoning: body.reasoning,
+        tools: body.tools,
+        previous_response_id: body.previous_response_id,
+        use_websocket: false,
+    };
+    let acquired = {
+        state
+            .account_pool()
+            .lock()
+            .await
+            .acquire_with(AccountAcquireRequest::new(&codex_request.model, Utc::now()))
+    };
+    let Some(acquired) = acquired else {
+        return no_available_accounts_response();
+    };
+    let account = acquired.account;
+    let response = send_codex_request(&state, &codex_request, &account, request_id.as_str()).await;
+    state.account_pool().lock().await.release(&account.id);
 
+    let response = match response {
+        Ok(response) => response,
+        Err(error) => return codex_client_error_response(error),
+    };
+    if let Some(cookie_repo) = state.cookie_repository() {
+        for cookie in &response.set_cookie_headers {
+            if cookie_repo
+                .capture_set_cookie(&account.id, cookie)
+                .await
+                .is_err()
+            {
+                return (
+                    StatusCode::INTERNAL_SERVER_ERROR,
+                    Json(openai_error(
+                        "Failed to persist upstream cookies",
+                        "cookie_store_error",
+                    )),
+                );
+            }
+        }
+    }
+    if let (Some(repo), Some(usage)) = (state.account_repository(), response.usage) {
+        if repo
+            .record_usage(
+                &account.id,
+                UsageDelta {
+                    input_tokens: u64_to_i64_saturating(usage.input_tokens),
+                    output_tokens: u64_to_i64_saturating(usage.output_tokens),
+                    cached_tokens: u64_to_i64_saturating(usage.cached_tokens),
+                },
+            )
+            .await
+            .is_err()
+        {
+            return (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(openai_error(
+                    "Failed to record account usage",
+                    "usage_store_error",
+                )),
+            );
+        }
+    }
+
+    match completed_response_json(&response.body) {
+        Ok(Some(body)) => (StatusCode::OK, Json(body)),
+        Ok(None) => (
+            StatusCode::BAD_GATEWAY,
+            Json(openai_error(
+                "Codex response did not include response.completed",
+                "empty_upstream_response",
+            )),
+        ),
+        Err(_) => (
+            StatusCode::BAD_GATEWAY,
+            Json(openai_error(
+                "Invalid Codex SSE response",
+                "invalid_upstream_sse",
+            )),
+        ),
+    }
+}
+
+fn no_available_accounts_response() -> (StatusCode, Json<Value>) {
     (
         StatusCode::SERVICE_UNAVAILABLE,
         Json(openai_error(
@@ -51,6 +156,102 @@ pub async fn responses(
     )
 }
 
+fn default_body(default_model: String) -> ResponsesBody {
+    ResponsesBody {
+        model: Some(default_model),
+        input: Some(Vec::new()),
+        instructions: Some(String::new()),
+        reasoning: None,
+        tools: None,
+        previous_response_id: None,
+    }
+}
+
+async fn send_codex_request(
+    state: &AppState,
+    request: &CodexResponsesRequest,
+    account: &crate::accounts::model::Account,
+    request_id: &str,
+) -> Result<crate::codex::client::CodexBackendResponse, CodexClientError> {
+    let request_domain = request_domain(&state.config().api.base_url);
+    let cookie_header = match (state.cookie_repository(), request_domain.as_deref()) {
+        (Some(repo), Some(domain)) => repo.cookie_header(&account.id, domain).await.ok().flatten(),
+        _ => None,
+    };
+    let client = CodexBackendClient::new(
+        build_reqwest_client(state.config().tls.force_http11)?,
+        state.config().api.base_url.clone(),
+        Fingerprint::default_codex_desktop(),
+    );
+    client
+        .create_response(
+            request,
+            CodexRequestContext {
+                access_token: &account.access_token,
+                account_id: account.account_id.as_deref(),
+                request_id,
+                turn_state: None,
+                cookie_header: cookie_header.as_deref(),
+            },
+        )
+        .await
+}
+
+fn completed_response_json(body: &str) -> Result<Option<Value>, crate::codex::sse::SseError> {
+    let events = parse_sse_events(body)?;
+    for event in events {
+        if event.event.as_deref() != Some("response.completed") {
+            continue;
+        }
+        let Ok(value) = serde_json::from_str::<Value>(&event.data) else {
+            continue;
+        };
+        if let Some(response) = value.get("response") {
+            return Ok(Some(response.clone()));
+        }
+    }
+    Ok(None)
+}
+
+fn codex_client_error_response(error: CodexClientError) -> (StatusCode, Json<Value>) {
+    match error {
+        CodexClientError::UnsupportedTransport(_) => (
+            StatusCode::BAD_REQUEST,
+            Json(openai_error(
+                "previous_response_id requires Codex WebSocket transport",
+                "websocket_required",
+            )),
+        ),
+        CodexClientError::Upstream { status, body } => (
+            status,
+            Json(openai_error(
+                &format!(
+                    "Codex upstream error: {}",
+                    body.chars().take(300).collect::<String>()
+                ),
+                "upstream_error",
+            )),
+        ),
+        _ => (
+            StatusCode::BAD_GATEWAY,
+            Json(openai_error(
+                "Codex upstream request failed",
+                "upstream_error",
+            )),
+        ),
+    }
+}
+
+fn request_domain(base_url: &str) -> Option<String> {
+    Url::parse(base_url)
+        .ok()
+        .and_then(|url| url.host_str().map(ToString::to_string))
+}
+
+fn u64_to_i64_saturating(value: u64) -> i64 {
+    value.min(i64::MAX as u64) as i64
+}
+
 pub async fn models(State(state): State<AppState>, headers: HeaderMap) -> impl IntoResponse {
     if !has_client_api_key(&headers) {
         return missing_client_api_key_response();

src/state.rs

@@ -1,9 +1,14 @@
 use std::sync::Arc;
 
 use sqlx::SqlitePool;
+use tokio::sync::Mutex;
 
-use crate::accounts::repository::AccountRepository;
+use crate::accounts::{
+    pool::{AccountPool, AccountPoolOptions, RotationStrategy},
+    repository::AccountRepository,
+};
 use crate::config::AppConfig;
+use crate::cookies::repository::CookieRepository;
 use crate::crypto::SecretBox;
 use crate::logs::repository::EventLogRepository;
 
@@ -17,27 +22,32 @@ pub struct AppServices {
     pub db: Option<SqlitePool>,
     pub event_logs: Option<EventLogRepository>,
     pub secret_box: Option<SecretBox>,
+    pub account_pool: Arc<Mutex<AccountPool>>,
 }
 
 impl AppState {
     pub fn new(config: AppConfig) -> Self {
+        let account_pool = account_pool_from_config(&config);
         Self {
             services: Arc::new(AppServices {
                 config,
                 db: None,
                 event_logs: None,
                 secret_box: None,
+                account_pool,
             }),
         }
     }
 
     pub fn with_pool(config: AppConfig, pool: SqlitePool) -> Self {
+        let account_pool = account_pool_from_config(&config);
         Self {
             services: Arc::new(AppServices {
                 config,
                 db: Some(pool.clone()),
                 event_logs: Some(EventLogRepository::new(pool)),
                 secret_box: None,
+                account_pool,
             }),
         }
     }
@@ -47,12 +57,14 @@ impl AppState {
         pool: SqlitePool,
         secret_box: SecretBox,
     ) -> Self {
+        let account_pool = account_pool_from_config(&config);
         Self {
             services: Arc::new(AppServices {
                 config,
                 db: Some(pool.clone()),
                 event_logs: Some(EventLogRepository::new(pool)),
                 secret_box: Some(secret_box),
+                account_pool,
             }),
         }
     }
@@ -75,4 +87,30 @@ impl AppState {
             self.services.secret_box.clone()?,
         ))
     }
+
+    pub fn cookie_repository(&self) -> Option<CookieRepository> {
+        Some(CookieRepository::new(
+            self.db()?.clone(),
+            self.services.secret_box.clone()?,
+        ))
+    }
+
+    pub fn account_pool(&self) -> Arc<Mutex<AccountPool>> {
+        self.services.account_pool.clone()
+    }
+}
+
+fn account_pool_from_config(config: &AppConfig) -> Arc<Mutex<AccountPool>> {
+    Arc::new(Mutex::new(AccountPool::with_options(AccountPoolOptions {
+        max_concurrent_per_account: config.auth.max_concurrent_per_account,
+        stale_slot_ttl: chrono::Duration::minutes(5),
+        rotation_strategy: match config.auth.rotation_strategy.as_str() {
+            "round_robin" => RotationStrategy::RoundRobin,
+            "sticky" => RotationStrategy::Sticky,
+            _ => RotationStrategy::LeastUsed,
+        },
+        skip_quota_limited: config.quota.skip_exhausted,
+        tier_priority: config.auth.tier_priority.clone(),
+        model_plan_allowlist: std::collections::BTreeMap::new(),
+    })))
 }

tests/codex_client_test.rs

@@ -0,0 +1,95 @@
+use serde_json::json;
+use wiremock::{
+    matchers::{body_json, header, method, path},
+    Mock, MockServer, ResponseTemplate,
+};
+
+use codex_proxy_rs::{
+    codex::{
+        client::{build_reqwest_client, CodexBackendClient, CodexRequestContext},
+        types::CodexResponsesRequest,
+        usage::TokenUsage,
+    },
+    fingerprint::model::Fingerprint,
+};
+
+#[tokio::test]
+async fn codex_backend_client_should_send_desktop_headers_and_capture_response_metadata() {
+    let server = MockServer::start().await;
+    let sse_body = concat!(
+        "event: response.completed\n",
+        "data: {\"response\":{\"id\":\"resp_1\",\"usage\":{\"input_tokens\":2,\"output_tokens\":3,\"input_tokens_details\":{\"cached_tokens\":1}}}}\n",
+        "\n",
+    );
+    Mock::given(method("POST"))
+        .and(path("/codex/responses"))
+        .and(header("authorization", "Bearer access-token"))
+        .and(header("chatgpt-account-id", "chatgpt-account"))
+        .and(header("originator", "Codex Desktop"))
+        .and(header("x-client-request-id", "req_1"))
+        .and(header("x-codex-turn-state", "turn_1"))
+        .and(header("cookie", "cf_clearance=old"))
+        .and(body_json(json!({
+            "model": "gpt-5.5",
+            "instructions": "",
+            "input": [],
+            "stream": true,
+            "store": false
+        })))
+        .respond_with(
+            ResponseTemplate::new(200)
+                .insert_header("content-type", "text/event-stream")
+                .insert_header(
+                    "set-cookie",
+                    "cf_clearance=new; Domain=.chatgpt.com; Path=/",
+                )
+                .insert_header("x-codex-turn-state", "turn_2")
+                .set_body_string(sse_body),
+        )
+        .mount(&server)
+        .await;
+    let client = CodexBackendClient::new(
+        build_reqwest_client(false).unwrap(),
+        server.uri(),
+        Fingerprint::default_for_tests(),
+    );
+
+    let response = client
+        .create_response(
+            &CodexResponsesRequest {
+                model: "gpt-5.5".to_string(),
+                instructions: String::new(),
+                input: Vec::new(),
+                stream: true,
+                store: false,
+                reasoning: None,
+                tools: None,
+                previous_response_id: None,
+                use_websocket: false,
+            },
+            CodexRequestContext {
+                access_token: "access-token",
+                account_id: Some("chatgpt-account"),
+                request_id: "req_1",
+                turn_state: Some("turn_1"),
+                cookie_header: Some("cf_clearance=old"),
+            },
+        )
+        .await
+        .unwrap();
+
+    assert_eq!(
+        response.usage,
+        Some(TokenUsage {
+            input_tokens: 2,
+            output_tokens: 3,
+            cached_tokens: 1,
+            total_tokens: 5,
+        })
+    );
+    assert_eq!(response.turn_state.as_deref(), Some("turn_2"));
+    assert_eq!(
+        response.set_cookie_headers,
+        vec!["cf_clearance=new; Domain=.chatgpt.com; Path=/".to_string()]
+    );
+}

tests/v1_upstream_route_test.rs

@@ -0,0 +1,210 @@
+use std::collections::BTreeMap;
+
+use axum::{
+    body::{to_bytes, Body},
+    http::{Request, StatusCode},
+};
+use chrono::Utc;
+use serde_json::{json, Value};
+use tower::ServiceExt;
+use wiremock::{
+    matchers::{body_json, header, method, path},
+    Mock, MockServer, ResponseTemplate,
+};
+
+use codex_proxy_rs::{
+    app::build_router,
+    config::{
+        AdminConfig, ApiConfig, AppConfig, AuthConfig, DatabaseConfig, LoggingConfig, ModelConfig,
+        QuotaConfig, QuotaWarningThresholds, SecurityConfig, ServerConfig, TlsConfig,
+        UsageStatsConfig,
+    },
+    cookies::repository::CookieRepository,
+    crypto::SecretBox,
+    state::AppState,
+    storage::db::connect_sqlite,
+};
+
+fn test_config(database_url: String, base_url: String) -> AppConfig {
+    AppConfig {
+        server: ServerConfig {
+            host: "127.0.0.1".to_string(),
+            port: 0,
+        },
+        api: ApiConfig { base_url },
+        model: ModelConfig {
+            default_model: "gpt-5.5".to_string(),
+            default_reasoning_effort: None,
+            service_tier: None,
+            aliases: BTreeMap::new(),
+        },
+        auth: AuthConfig {
+            refresh_margin_seconds: 300,
+            refresh_enabled: true,
+            refresh_concurrency: 2,
+            max_concurrent_per_account: 3,
+            request_interval_ms: 50,
+            rotation_strategy: "least_used".to_string(),
+            tier_priority: Vec::new(),
+        },
+        quota: QuotaConfig {
+            refresh_interval_minutes: 5,
+            warning_thresholds: QuotaWarningThresholds {
+                primary: vec![80, 90],
+                secondary: vec![80, 90],
+            },
+            skip_exhausted: true,
+        },
+        usage_stats: UsageStatsConfig {
+            history_retention_days: None,
+        },
+        database: DatabaseConfig { url: database_url },
+        security: SecurityConfig {
+            master_key_file: "data/master.key".to_string(),
+            api_key_pepper_file: "data/api-key-pepper.key".to_string(),
+        },
+        tls: TlsConfig {
+            force_http11: false,
+        },
+        admin: AdminConfig {
+            session_ttl_minutes: 1440,
+        },
+        logging: LoggingConfig {
+            directory: "logs".to_string(),
+            max_file_bytes: 10_485_760,
+            retention_days: 14,
+            enabled: false,
+            capacity: 2_000,
+            capture_body: false,
+        },
+    }
+}
+
+#[tokio::test]
+async fn v1_responses_should_use_imported_account_and_record_usage() {
+    let server = MockServer::start().await;
+    let sse_body = concat!(
+        "event: response.completed\n",
+        "data: {\"response\":{\"id\":\"resp_1\",\"object\":\"response\",\"usage\":{\"input_tokens\":7,\"output_tokens\":4,\"input_tokens_details\":{\"cached_tokens\":2}}}}\n",
+        "\n",
+    );
+    Mock::given(method("POST"))
+        .and(path("/codex/responses"))
+        .and(header("authorization", "Bearer access-secret"))
+        .and(header("chatgpt-account-id", "chatgpt-account"))
+        .and(body_json(json!({
+            "model": "gpt-5.5",
+            "instructions": "",
+            "input": [],
+            "stream": true,
+            "store": false
+        })))
+        .respond_with(
+            ResponseTemplate::new(200)
+                .insert_header("content-type", "text/event-stream")
+                .insert_header(
+                    "set-cookie",
+                    "cf_clearance=new; Domain=.chatgpt.com; Path=/",
+                )
+                .set_body_string(sse_body),
+        )
+        .mount(&server)
+        .await;
+    let dir = tempfile::tempdir().unwrap();
+    let db = dir.path().join("v1-upstream.sqlite");
+    let url = format!("sqlite://{}", db.display());
+    let pool = connect_sqlite(&url).await.unwrap();
+    seed_admin_session(&pool, "session_1").await;
+    let secret_box = SecretBox::new([21u8; 32]);
+    let app = build_router(AppState::with_pool_and_secret_box(
+        test_config(url, server.uri()),
+        pool.clone(),
+        secret_box.clone(),
+    ));
+
+    let import_response = app
+        .clone()
+        .oneshot(
+            Request::builder()
+                .method("POST")
+                .uri("/admin/accounts/import")
+                .header("content-type", "application/json")
+                .header("cookie", "cpr_admin_session=session_1")
+                .body(Body::from(
+                    json!({
+                        "accounts": [{
+                            "id": "acct_imported",
+                            "accountId": "chatgpt-account",
+                            "token": "access-secret",
+                            "refreshToken": "refresh-secret",
+                            "status": "active"
+                        }]
+                    })
+                    .to_string(),
+                ))
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+    assert_eq!(import_response.status(), StatusCode::OK);
+
+    let response = app
+        .oneshot(
+            Request::builder()
+                .method("POST")
+                .uri("/v1/responses")
+                .header("authorization", "Bearer cpr_test")
+                .header("content-type", "application/json")
+                .body(Body::from(r#"{"model":"gpt-5.5","input":[]}"#))
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+
+    assert_eq!(response.status(), StatusCode::OK);
+    let body = response_json(response).await;
+    assert_eq!(body["id"], "resp_1");
+    assert_eq!(body["usage"]["input_tokens"], 7);
+    let usage: (i64, i64, i64, i64) = sqlx::query_as(
+        "select request_count, input_tokens, output_tokens, cached_tokens from account_usage where account_id = ?",
+    )
+    .bind("acct_imported")
+    .fetch_one(&pool)
+    .await
+    .unwrap();
+    assert_eq!(usage, (1, 7, 4, 2));
+    let cookie_header = CookieRepository::new(pool, secret_box)
+        .cookie_header("acct_imported", "chatgpt.com")
+        .await
+        .unwrap();
+    assert_eq!(cookie_header.as_deref(), Some("cf_clearance=new"));
+}
+
+async fn seed_admin_session(pool: &sqlx::SqlitePool, session_id: &str) {
+    let now = Utc::now().to_rfc3339();
+    sqlx::query(
+        "insert into admin_users (id, password_hash, created_at, updated_at) values (?, ?, ?, ?)",
+    )
+    .bind("admin_1")
+    .bind("hash")
+    .bind(&now)
+    .bind(&now)
+    .execute(pool)
+    .await
+    .unwrap();
+    sqlx::query(
+        "insert into admin_sessions (id, user_id, expires_at, created_at) values (?, ?, ?, ?)",
+    )
+    .bind(session_id)
+    .bind("admin_1")
+    .bind("2999-01-01T00:00:00Z")
+    .bind(now)
+    .execute(pool)
+    .await
+    .unwrap();
+}
+
+async fn response_json(response: axum::response::Response) -> Value {
+    let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap();
+    serde_json::from_slice(&bytes).unwrap()
+}