chore: bump version to 3.3.28

Co-authored-by: Codex <noreply@openai.com>
fix(openai): calculate standalone image generation costs
2026-11-07 00:05:49 +00:00 · 2026-09-08 02:42:09 +08:00 · 2026-09-08 02:41:14 +08:00 · 2026-09-08 02:23:00 +08:00 · 2026-09-08 01:51:00 +08:00 · 2026-09-08 01:46:08 +08:00
38 changed files with 1750 additions and 496 deletions
@@ -626,7 +626,7 @@ jobs:
          cosign sign --yes "${images[@]}"

      - name: Publish GitHub Release
-        uses: softprops/action-gh-release@3d0d9888cb7fd7b750713d6e236d1fcb99157228 # v3.0.2
+        uses: softprops/action-gh-release@efb35369e0ad2afab669f228072c1b0d510eae64 # v3.0.3
        with:
          tag_name: ${{ needs.release-info.outputs.tag }}
          generate_release_notes: true
@@ -154,9 +154,9 @@ dependencies = [

 [[package]]
 name = "aws-lc-rs"
-version = "1.16.2"
+version = "1.18.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a054912289d18629dc78375ba2c3726a3afe3ff71b4edba9dedfca0e3446d1fc"
+checksum = "b281d307588d634de920874890732659e2e7672f72b5e10e81badc1a8a83621e"
 dependencies = [
 "aws-lc-sys",
 "untrusted 0.7.1",
@@ -165,14 +165,15 @@ dependencies = [

 [[package]]
 name = "aws-lc-sys"
-version = "0.39.0"
+version = "0.45.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1fa7e52a4c5c547c741610a2c6f123f3881e409b714cd27e6798ef020c514f0a"
+checksum = "9bff6c3b54fad79a2e60b8102caf565819711497c1f5f092f49508e2f5c31b27"
 dependencies = [
 "cc",
 "cmake",
 "dunce",
 "fs_extra",
+ "pkg-config",
 ]

 [[package]]
@@ -400,9 +401,9 @@ dependencies = [

 [[package]]
 name = "aws-smithy-runtime-api"
-version = "1.15.0"
+version = "1.16.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "954c563ce84507722d2679f07a35d21b9c6466b3872d513020d0281fc8112ac9"
+checksum = "9c054752dd9e4dc73d0b75748c99ac2d0feafbf2f25c7b0516f03a3534161223"
 dependencies = [
 "aws-smithy-async",
 "aws-smithy-runtime-api-macros",
@@ -440,9 +441,9 @@ dependencies = [

 [[package]]
 name = "aws-smithy-types"
-version = "1.6.2"
+version = "1.6.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fce83ce9abbb198d25bc7131e468d0f9fe1257125e58c39f3f9fc9f5098c9647"
+checksum = "8f94d16e797ec62cd999fc9d5942b48fa7050c3093ddadff48e4d7528d16fcb9"
 dependencies = [
 "base64-simd",
 "bytes",
@@ -1257,9 +1258,9 @@ checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582"

 [[package]]
 name = "flate2"
-version = "1.1.9"
+version = "1.1.10"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "843fba2746e448b37e26a819579957415c8cef339bf08564fe8b7ddbd959573c"
+checksum = "6e634e2e0ebac1ee034020da1ca582e17ffe4e0f5e985823721e168928136dcb"
 dependencies = [
 "crc32fast",
 "miniz_oxide",
@@ -1466,7 +1467,7 @@ dependencies = [
 "tokio",
 "tokio-tungstenite 0.29.0",
 "tower",
- "tower-http 0.7.0",
+ "tower-http 0.7.1",
 "tracing",
 "url",
 "uuid",
@@ -1835,9 +1836,9 @@ dependencies = [

 [[package]]
 name = "hyper"
-version = "1.8.1"
+version = "1.11.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2ab2d4f250c3d7b1c9fcdff1cece94ea4e2dfbec68614f7b87cb205f24ca9d11"
+checksum = "27b501faa50e7a26c3d3560ca625132f4078a17771f4810baf70475ae48cbe43"
 dependencies = [
 "atomic-waker",
 "bytes",
@@ -1850,7 +1851,6 @@ dependencies = [
 "httpdate",
 "itoa",
 "pin-project-lite",
- "pin-utils",
 "smallvec",
 "tokio",
 "want",
@@ -2213,9 +2213,9 @@ dependencies = [

 [[package]]
 name = "miniz_oxide"
-version = "0.8.9"
+version = "0.9.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1fa76a2c86f704bdb222d66965fb3d63269ce38518b83cb0575fca855ebb6316"
+checksum = "b63fbc4a50860e98e7b2aa7804ded1db5cbc3aff9193adaff57a6931bf7c4b4c"
 dependencies = [
 "adler2",
 "simd-adler32",
@@ -3026,9 +3026,9 @@ dependencies = [

 [[package]]
 name = "rustls-native-certs"
-version = "0.8.3"
+version = "0.8.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "612460d5f7bea540c490b2b6395d8e34a953e52b491accd6c86c8164c5932a63"
+checksum = "dab5152771c58876a2146916e53e35057e1a4dfa2b9df0f0305b07f611fdea4d"
 dependencies = [
 "openssl-probe",
 "rustls-pki-types",
@@ -3038,9 +3038,9 @@ dependencies = [

 [[package]]
 name = "rustls-pki-types"
-version = "1.14.0"
+version = "1.15.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "be040f8b0a225e40375822a563fa9524378b9d63112f53e19ffff34df5d33fdd"
+checksum = "2f4925028c7eb5d1fcdaf196971378ed9d2c1c4efc7dc5d011256f76c99c0a96"
 dependencies = [
 "web-time",
 "zeroize",
@@ -3704,7 +3704,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd"
 dependencies = [
 "fastrand",
- "getrandom 0.3.4",
+ "getrandom 0.4.3",
 "once_cell",
 "rustix",
 "windows-sys 0.61.2",
@@ -3805,9 +3805,9 @@ checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20"

 [[package]]
 name = "tokio"
-version = "1.52.3"
+version = "1.53.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8fc7f01b389ac15039e4dc9531aa973a135d7a4135281b12d7c1bc79fd57fffe"
+checksum = "202caea871b69668250d242070849eb495be178ed697a3e98aebce5bc81a0bed"
 dependencies = [
 "bytes",
 "libc",
@@ -3928,9 +3928,9 @@ dependencies = [

 [[package]]
 name = "tower-http"
-version = "0.7.0"
+version = "0.7.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b11f75e912b0c2be01b63d8cf8057b8c3f97cf34abb3d431a3a4c8675498e233"
+checksum = "08a05a66a4fdd61cbbe0a1d755ffe0ca6aba159dd4820936a0ff8a8278245b9c"
 dependencies = [
 "bitflags",
 "bytes",
@@ -213,13 +213,17 @@ fn capitalize_first(value: &str) -> String {

 pub(crate) fn billing_view(billing: Option<&domain::UsageBilling>) -> Option<BillingView> {
    match billing? {
-        domain::UsageBilling::Total { total, .. } => Some(BillingView {
+        domain::UsageBilling::Total { source, total } => Some(BillingView {
            input_amount_display: "—".to_owned(),
            output_amount_display: "—".to_owned(),
            cache_read_amount_display: "—".to_owned(),
            cache_write_amount_display: "—".to_owned(),
            standard_amount_display: "—".to_owned(),
-            total_amount_display: format_money(total),
+            total_amount_display: if source == "calculated" {
+                format!("≈ {}", format_money(total))
+            } else {
+                format_money(total)
+            },
            input_price_display: "—".to_owned(),
            output_price_display: "—".to_owned(),
            cache_read_price_display: "—".to_owned(),
@@ -15,8 +15,8 @@ use serde_json::Value;
 use crate::ApiState;
 use crate::openai::{
    auth::{authenticate_client, client_access_error_response},
+    endpoint::collect_raw_json_response,
    error::gateway_error_response,
-    provider_endpoint::collect_raw_json_response,
    responses::{OpenAiRequestHeaders, request_client_context},
 };

@@ -1,10 +1,10 @@
 //! OpenAI 客户端协议 adapter。

 pub mod auth;
+mod endpoint;
 pub mod error;
 pub mod images;
 pub mod models;
-mod provider_endpoint;
 pub mod responses;
 pub mod router;
 pub mod search;
@@ -23,6 +23,7 @@ const OUTBOUND_COMMAND_BUFFER: usize = 32;
 const INBOUND_EVENT_BUFFER: usize = 32;
 const DOWNSTREAM_WRITE_TIMEOUT: Duration = Duration::from_secs(5 * 60);
 const CONNECTION_MAX_AGE: Duration = Duration::from_secs(60 * 60);
+const DOWNSTREAM_PING_INTERVAL: Duration = Duration::from_secs(25);

 /// WebSocket pump 的写入与生命周期预算。
 #[derive(Clone, Copy)]
@@ -171,9 +172,27 @@ struct ConnectionStats {
    command_backpressure_count: AtomicU64,
    ping_received_count: AtomicU64,
    pong_written_count: AtomicU64,
+    ping_written_count: AtomicU64,
+    pong_received_count: AtomicU64,
+    last_read_ms: AtomicU64,
+    last_write_ms: AtomicU64,
 }

 impl ConnectionStats {
+    fn record_read(&self, opened_at: Instant) {
+        self.last_read_ms.store(
+            u64::try_from(opened_at.elapsed().as_millis()).unwrap_or(u64::MAX),
+            Ordering::Relaxed,
+        );
+    }
+
+    fn record_write(&self, opened_at: Instant) {
+        self.last_write_ms.store(
+            u64::try_from(opened_at.elapsed().as_millis()).unwrap_or(u64::MAX),
+            Ordering::Relaxed,
+        );
+    }
+
    fn observe_command_queue(&self, sender: &mpsc::Sender<ConnectionCommand>) {
        let queued = OUTBOUND_COMMAND_BUFFER
            .saturating_sub(sender.capacity())
@@ -400,10 +419,15 @@ impl ResponsesWebSocketConnection {
    }

    pub(super) fn log_summary(&self, request_count: u64) {
+        let connection_age_ms = self.age().as_millis();
+        let read_idle_ms = connection_age_ms
+            .saturating_sub(u128::from(self.stats.last_read_ms.load(Ordering::Relaxed)));
+        let write_idle_ms = connection_age_ms
+            .saturating_sub(u128::from(self.stats.last_write_ms.load(Ordering::Relaxed)));
        tracing::info!(
            websocket_connection_id = %self.connection_id,
            request_count,
-            connection_age_ms = self.age().as_millis(),
+            connection_age_ms,
            pump_exit_reason = self
                .exit_reason
                .map_or("unknown", PumpExitReason::as_str),
@@ -417,6 +441,10 @@ impl ResponsesWebSocketConnection {
                .load(Ordering::Relaxed),
            ping_received_count = self.stats.ping_received_count.load(Ordering::Relaxed),
            pong_written_count = self.stats.pong_written_count.load(Ordering::Relaxed),
+            ping_written_count = self.stats.ping_written_count.load(Ordering::Relaxed),
+            pong_received_count = self.stats.pong_received_count.load(Ordering::Relaxed),
+            read_idle_ms,
+            write_idle_ms,
            "Responses WebSocket disconnected"
        );
    }
@@ -493,6 +521,12 @@ async fn run_pump<S, E>(
    let deadline = tokio::time::sleep_until(opened_at + config.max_age);
    tokio::pin!(deadline);
    let mut deadline_elapsed = false;
+    let mut heartbeat = tokio::time::interval_at(
+        opened_at + DOWNSTREAM_PING_INTERVAL,
+        DOWNSTREAM_PING_INTERVAL,
+    );
+    heartbeat.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+    let mut ping_sequence = 0_u64;
    let reason = loop {
        tokio::select! {
            biased;
@@ -531,6 +565,9 @@ async fn run_pump<S, E>(
                    ConnectionWriteError::Transport { message }
                });
                let failed = result.is_err();
+                if !failed {
+                    stats.record_write(opened_at);
+                }
                let _ = acknowledged.send(result);
                if failed {
                    break PumpExitReason::WriteError;
@@ -539,45 +576,79 @@ async fn run_pump<S, E>(
                    break PumpExitReason::ServerClose;
                }
            }
-            message = socket.next() => match message {
-                Some(Ok(Message::Text(payload))) => {
-                    if let Err(reason) = emit_incoming(&incoming, ConnectionEvent::Text(payload.to_string())) {
-                        break reason;
+            // 官方 Codex WsStream 在独立 pump 中回复 Pong，且不将控制帧交给业务流。
+            // 这里只为下游链路保活，不设置 Pong deadline，也不延长业务请求预算。
+            _ = heartbeat.tick() => {
+                ping_sequence = ping_sequence.wrapping_add(1);
+                let ping = Message::Ping(ping_sequence.to_be_bytes().to_vec().into());
+                let result = tokio::select! {
+                    biased;
+                    () = cancellation.cancelled() => break PumpExitReason::LifecycleShutdown,
+                    result = timeout(config.write_timeout, socket.send(ping)) => result,
+                };
+                match result {
+                    Ok(Ok(())) => {
+                        stats.ping_written_count.fetch_add(1, Ordering::Relaxed);
+                        stats.record_write(opened_at);
                    }
-                }
-                Some(Ok(Message::Binary(_))) => {
-                    if let Err(reason) = emit_incoming(&incoming, ConnectionEvent::Binary) {
-                        break reason;
+                    Ok(Err(error)) => {
+                        tracing::info!(
+                            websocket_connection_id = %connection_id,
+                            error = %error,
+                            "Responses WebSocket Ping write failed"
+                        );
+                        break PumpExitReason::WriteError;
                    }
+                    Err(_) => break PumpExitReason::WriteTimeout,
                }
-                Some(Ok(Message::Ping(payload))) => {
-                    stats.ping_received_count.fetch_add(1, Ordering::Relaxed);
-                    match timeout(config.write_timeout, socket.send(Message::Pong(payload))).await {
-                        Ok(Ok(())) => {
-                            stats.pong_written_count.fetch_add(1, Ordering::Relaxed);
+            }
+            message = socket.next() => {
+                if matches!(&message, Some(Ok(_))) {
+                    stats.record_read(opened_at);
+                }
+                match message {
+                    Some(Ok(Message::Text(payload))) => {
+                        if let Err(reason) = emit_incoming(&incoming, ConnectionEvent::Text(payload.to_string())) {
+                            break reason;
                        }
-                        Ok(Err(error)) => {
-                            tracing::info!(
-                                websocket_connection_id = %connection_id,
-                                error = %error,
-                                "Responses WebSocket Pong write failed"
-                            );
-                            break PumpExitReason::WriteError;
-                        }
-                        Err(_) => break PumpExitReason::WriteTimeout,
                    }
+                    Some(Ok(Message::Binary(_))) => {
+                        if let Err(reason) = emit_incoming(&incoming, ConnectionEvent::Binary) {
+                            break reason;
+                        }
+                    }
+                    Some(Ok(Message::Ping(payload))) => {
+                        stats.ping_received_count.fetch_add(1, Ordering::Relaxed);
+                        match timeout(config.write_timeout, socket.send(Message::Pong(payload))).await {
+                            Ok(Ok(())) => {
+                                stats.pong_written_count.fetch_add(1, Ordering::Relaxed);
+                                stats.record_write(opened_at);
+                            }
+                            Ok(Err(error)) => {
+                                tracing::info!(
+                                    websocket_connection_id = %connection_id,
+                                    error = %error,
+                                    "Responses WebSocket Pong write failed"
+                                );
+                                break PumpExitReason::WriteError;
+                            }
+                            Err(_) => break PumpExitReason::WriteTimeout,
+                        }
+                    }
+                    Some(Ok(Message::Pong(_))) => {
+                        stats.pong_received_count.fetch_add(1, Ordering::Relaxed);
+                    }
+                    Some(Ok(Message::Close(_))) => break PumpExitReason::ClientClose,
+                    Some(Err(error)) => {
+                        tracing::info!(
+                            websocket_connection_id = %connection_id,
+                            error = %error,
+                            "Responses WebSocket receive failed"
+                        );
+                        break PumpExitReason::ReadError;
+                    }
+                    None => break PumpExitReason::PeerEof,
                }
-                Some(Ok(Message::Pong(_))) => {}
-                Some(Ok(Message::Close(_))) => break PumpExitReason::ClientClose,
-                Some(Err(error)) => {
-                    tracing::info!(
-                        websocket_connection_id = %connection_id,
-                        error = %error,
-                        "Responses WebSocket receive failed"
-                    );
-                    break PumpExitReason::ReadError;
-                }
-                None => break PumpExitReason::PeerEof,
            },
        }
    };
@@ -247,7 +247,7 @@ async fn serve_responses_websocket(socket: WebSocket, session: ResponsesWebSocke
            }
        };
        started.session.trace().record("client.connection", serde_json::json!({
-            "transport": "websocket", "connectionId": connection.id(), "correlationId": correlation_id,
+            "transport": "websocket", "connectionId": connection.id(), "correlationId": correlation_id.as_ref(),
            "requestIndex": request_count,
        }));
        started
@@ -14,8 +14,8 @@ use gateway_core::operation::{Operation, RawJsonPayload, StandaloneSearchRequest
 use crate::ApiState;
 use crate::openai::{
    auth::{authenticate_client, client_access_error_response},
+    endpoint::collect_raw_json_response,
    error::gateway_error_response,
-    provider_endpoint::collect_raw_json_response,
    responses::{OpenAiRequestHeaders, request_client_context},
 };

@@ -919,6 +919,22 @@ async fn usage_route_should_expose_table_facts_without_detail_payload() {
            compact: true,
            started_at: Utc::now(),
        });
+    {
+        let mut records = fixture.usage_records.lock().expect("usage records");
+        for source in ["calculated", "provider_reported"] {
+            let mut image = records[0].clone();
+            image.id = format!("image_{source}");
+            image.endpoint = "/v1/images/generations".to_owned();
+            image.provider_kind = Some("openai".to_owned());
+            image.requested_model_id = Some("gpt-image-2".to_owned());
+            image.upstream_model_id = None;
+            image.billing = Some(UsageBilling::Total {
+                source: source.to_owned(),
+                total: usd("0.00696"),
+            });
+            records.push(image);
+        }
+    }
    let response = observability::router::<AdminTestState>()
        .with_state(fixture.state())
        .oneshot(
@@ -1016,4 +1032,12 @@ async fn usage_route_should_expose_table_facts_without_detail_payload() {
        })
    );
    assert!(value["data"]["items"][0].get("metadata").is_none());
+    assert_eq!(
+        value["data"]["items"][1]["billing"]["totalAmountDisplay"],
+        "≈ $0.007"
+    );
+    assert_eq!(
+        value["data"]["items"][2]["billing"]["totalAmountDisplay"],
+        "$0.007"
+    );
 }
@@ -61,7 +61,7 @@ fn source_tree_should_match_frozen_machine_manifest() {
        "src/openai/images/mod.rs",
        "src/openai/mod.rs",
        "src/openai/models.rs",
-        "src/openai/provider_endpoint.rs",
+        "src/openai/endpoint.rs",
        "src/openai/responses/error.rs",
        "src/openai/responses/http.rs",
        "src/openai/responses/mod.rs",
@@ -107,7 +107,7 @@ fn test_tree_should_match_frozen_rust_mirror() {
        "tests/openai/images/mod.rs",
        "tests/openai/mod.rs",
        "tests/openai/models.rs",
-        "tests/openai/provider_endpoint.rs",
+        "tests/openai/endpoint.rs",
        "tests/openai/responses/http.rs",
        "tests/openai/responses/mod.rs",
        "tests/openai/responses/websocket/connection.rs",
@@ -1,8 +1,8 @@
 mod auth;
+mod endpoint;
 mod error;
 mod images;
 mod models;
-mod provider_endpoint;
 mod responses;
 mod router;
 mod search;
@@ -444,21 +444,6 @@ async fn inbound_overload_should_close_without_executing_queued_frames() {
    ));
 }

-#[tokio::test(start_paused = true)]
-async fn pump_does_not_emit_an_active_ping() {
-    let PumpHarness {
-        connection,
-        incoming: _incoming,
-        mut written,
-        ..
-    } = test_connection(false);
-    tokio::time::advance(Duration::from_secs(5 * 60)).await;
-    tokio::task::yield_now().await;
-
-    assert!(matches!(written.try_recv(), Err(TryRecvError::Empty)));
-    drop(connection);
-}
-
 #[tokio::test]
 async fn outbound_commands_are_written_in_acknowledged_order() {
    let PumpHarness {
@@ -582,13 +567,17 @@ async fn idle_connection_reaches_the_official_limit_without_starting_an_executio

    tokio::time::advance(Duration::from_secs(60 * 60)).await;
    tokio::task::yield_now().await;
-    let message = tokio::time::timeout(Duration::from_secs(1), socket.next())
-        .await
-        .expect("connection limit response timeout")
-        .expect("connection remains available for the limit error")
-        .expect("valid WebSocket frame");
-    let ClientMessage::Text(text) = message else {
-        panic!("connection limit must be a text error event");
+    let text = loop {
+        let message = tokio::time::timeout(Duration::from_secs(1), socket.next())
+            .await
+            .expect("connection limit response timeout")
+            .expect("connection remains available for the limit error")
+            .expect("valid WebSocket frame");
+        match message {
+            ClientMessage::Text(text) => break text,
+            ClientMessage::Ping(_) => {}
+            other => panic!("unexpected frame before connection limit: {other:?}"),
+        }
    };
    let value = serde_json::from_str::<Value>(&text).expect("connection limit JSON");

@@ -606,7 +595,8 @@ async fn idle_connection_reaches_the_official_limit_without_starting_an_executio
    );
    assert_eq!(trace.starts.load(Ordering::Acquire), 0);

-    socket.close(None).await.expect("close WebSocket");
+    // 服务端已因生命周期到期关闭；不再向关闭中的连接写入。
+    drop(socket);
    server.abort();
 }

@@ -949,3 +939,134 @@ async fn client_close_should_cancel_active_execution_without_starting_queued_req
    assert!(trace.cancelled.load(Ordering::Acquire));
    server.abort();
 }
+
+#[tokio::test(start_paused = true)]
+async fn idle_pump_sends_heartbeat_without_bursting_after_a_delay() {
+    let PumpHarness {
+        connection,
+        incoming: _incoming,
+        mut written,
+        ..
+    } = test_connection(false);
+    tokio::task::yield_now().await;
+    assert!(matches!(written.try_recv(), Err(TryRecvError::Empty)));
+
+    tokio::time::advance(Duration::from_secs(25)).await;
+    let first = written.recv().await.expect("first heartbeat");
+    assert!(matches!(first, Message::Ping(_)));
+
+    tokio::time::advance(Duration::from_secs(120)).await;
+    let second = written.recv().await.expect("heartbeat after delay");
+    assert!(matches!(&second, Message::Ping(_)));
+    assert_ne!(first, second);
+    tokio::task::yield_now().await;
+    assert!(matches!(written.try_recv(), Err(TryRecvError::Empty)));
+    drop(connection);
+}
+
+#[tokio::test(start_paused = true)]
+async fn missing_pongs_do_not_cancel_the_connection_or_block_business_frames() {
+    let PumpHarness {
+        mut connection,
+        incoming,
+        mut written,
+        ..
+    } = test_connection(false);
+    for _ in 0..13 {
+        tokio::time::advance(Duration::from_secs(25)).await;
+        assert!(matches!(written.recv().await, Some(Message::Ping(_))));
+    }
+
+    incoming
+        .send(Ok(Message::Text("still connected".into())))
+        .expect("send business frame after 325 seconds without Pong");
+    assert!(matches!(
+        connection.next_event().await,
+        Some(ConnectionEvent::Text(text)) if text == "still connected"
+    ));
+    incoming.send(Ok(Message::Close(None))).expect("close peer");
+    assert_eq!(
+        connection.wait_for_exit().await,
+        PumpExitReason::ClientClose
+    );
+    tokio::time::advance(Duration::from_secs(25)).await;
+    assert!(written.recv().await.is_none());
+}
+
+#[tokio::test(start_paused = true)]
+async fn stalled_heartbeat_can_be_cancelled_without_waiting_for_write_timeout() {
+    let PumpHarness {
+        mut connection,
+        incoming: _incoming,
+        written: _written,
+        dropped,
+        cancellation,
+    } = test_connection(true);
+    tokio::time::advance(Duration::from_secs(25)).await;
+    tokio::task::yield_now().await;
+    cancellation.cancel();
+    let reason = tokio::time::timeout(Duration::from_secs(1), connection.wait_for_exit())
+        .await
+        .expect("cancel a blocked heartbeat promptly");
+    assert_eq!(reason, PumpExitReason::LifecycleShutdown);
+    assert!(dropped.load(Ordering::Acquire));
+}
+
+#[tokio::test(start_paused = true)]
+async fn stalled_heartbeat_obeys_the_existing_write_timeout() {
+    let PumpHarness {
+        mut connection,
+        incoming: _incoming,
+        written: _written,
+        dropped,
+        ..
+    } = test_connection(true);
+    let started = tokio::time::Instant::now();
+    assert_eq!(
+        connection.wait_for_exit().await,
+        PumpExitReason::WriteTimeout
+    );
+    assert_eq!(started.elapsed(), Duration::from_secs(25 + 300));
+    assert!(dropped.load(Ordering::Acquire));
+}
+
+#[tokio::test]
+async fn active_response_receives_heartbeats_during_two_hundred_seconds_of_silence() {
+    let (trace, mut socket, server) = start_active_response().await;
+    for _ in 0..8 {
+        tokio::time::pause();
+        tokio::time::advance(Duration::from_secs(25)).await;
+        tokio::task::yield_now().await;
+        tokio::time::resume();
+        let message = tokio::time::timeout(Duration::from_secs(2), socket.next())
+            .await
+            .expect("heartbeat while the upstream response is pending")
+            .expect("connection remains open")
+            .expect("valid heartbeat frame");
+        let ClientMessage::Ping(payload) = message else {
+            panic!("expected heartbeat while business stream is silent: {message:?}");
+        };
+        // 与官方 WsStream 一样，在业务响应未完成时回复同 payload 的 Pong。
+        socket
+            .send(ClientMessage::Pong(payload))
+            .await
+            .expect("reply to heartbeat");
+    }
+    assert_eq!(trace.starts.load(Ordering::Acquire), 1);
+    assert!(!trace.cancelled.load(Ordering::Acquire));
+
+    trace.release_terminal.notify_one();
+    let message = tokio::time::timeout(Duration::from_secs(2), socket.next())
+        .await
+        .expect("terminal response after heartbeats")
+        .expect("connection remains open")
+        .expect("valid terminal frame");
+    let ClientMessage::Text(text) = message else {
+        panic!("expected completed response: {message:?}");
+    };
+    let value: Value = serde_json::from_str(&text).expect("terminal JSON");
+    assert_eq!(value["type"], "response.completed");
+    assert_eq!(trace.starts.load(Ordering::Acquire), 1);
+    socket.close(None).await.expect("close WebSocket");
+    server.abort();
+}
@@ -22,7 +22,7 @@ use gateway_core::upstream::UpstreamSendState;
 use serde_json::{Value, json};
 use tower::ServiceExt;

-use super::provider_endpoint::BufferedJsonSession;
+use super::endpoint::BufferedJsonSession;
 use super::{api_router, authenticated_client};

 const SEARCH_RESPONSE: &[u8] = br#"{ "encrypted_output":"ciphertext", "output":"search result", "results":[{"type":"text_result","ref_id":"turn0search0","future":9007199254740993}] }"#;
@@ -524,7 +524,7 @@ impl DefaultExecutionService {
            operation: operation.kind(),
            endpoint: metadata.endpoint,
            client_transport: metadata.transport.as_str().to_owned(),
-            requested_model: target.into_public_model(),
+            requested_model: target.into_public_model().or(observation.requested_model),
            client_ip: metadata.client_ip,
            user_agent: metadata.user_agent,
            reasoning_effort: observation.reasoning_effort,
@@ -20,7 +20,9 @@ use crate::event::{EventSequenceValidator, ProviderEvent};
 use crate::identity::ProviderKind;
 use crate::operation::Operation;
 use crate::policy::ClientApiKeyId;
-use crate::routing::{ModelCapabilities, ModelPresentation, ProviderCandidate, UpstreamModelId};
+use crate::routing::{
+    ModelCapabilities, ModelPresentation, ProviderCandidate, PublicModelId, UpstreamModelId,
+};
 use crate::upstream::OpaqueUpstreamValue;
 use crate::upstream::{UpstreamSendState, UpstreamTransport};

@@ -416,6 +418,8 @@ pub struct ProviderRequest {
 /// Provider 对公共观测表可解释的请求语义；未知字段保持空值。
 #[derive(Debug, Clone, Default, PartialEq, Eq)]
 pub struct ProviderRequestObservation {
+    /// 独立 Provider 端点声明的请求模型，仅用于观测，不参与文本模型目录路由。
+    pub requested_model: Option<PublicModelId>,
    /// 客户端原始请求中的推理强度。
    pub reasoning_effort: Option<String>,
    pub reasoning_preset: Option<String>,
@@ -26,7 +26,7 @@ use gateway_core::engine::execution::{
 use gateway_core::engine::probe::{AccountProbe, AccountProbeErrorSource, AccountProbeRequest};
 use gateway_core::engine::provider::{
    Provider, ProviderCallMetadata, ProviderCatalogGeneration, ProviderModelCapabilities,
-    ProviderRegistry, ProviderRequest, ProviderStream,
+    ProviderRegistry, ProviderRequest, ProviderRequestObservation, ProviderStream,
 };
 use gateway_core::engine::{
    AttemptContext, AttemptRecord, ExecutionStore, IntermediateFailure, ModelRequestFinalization,
@@ -260,7 +260,9 @@ impl Provider for LocalFailingProvider {
    }
 }

-struct ColdFailingProvider;
+struct ColdFailingProvider {
+    requested_model: Option<PublicModelId>,
+}

 #[async_trait]
 impl Provider for ColdFailingProvider {
@@ -272,6 +274,13 @@ impl Provider for ColdFailingProvider {
        ProviderCatalogGeneration::default()
    }

+    fn request_observation(&self, _: &Operation, _: &ClientApiKeyId) -> ProviderRequestObservation {
+        ProviderRequestObservation {
+            requested_model: self.requested_model.clone(),
+            ..Default::default()
+        }
+    }
+
    async fn query_model_capabilities(
        &self,
    ) -> Result<Vec<ProviderModelCapabilities>, ProviderError> {
@@ -323,12 +332,19 @@ fn successful_authentication_should_record_client_key_usage() {

 #[test]
 fn provider_endpoint_should_persist_its_real_v1_endpoint() {
+    assert_provider_endpoint_observation(None);
+    assert_provider_endpoint_observation(Some("gpt-image-2"));
+}
+
+fn assert_provider_endpoint_observation(model: Option<&str>) {
    let store = Arc::new(TrackingExecutionStore::default());
    let service = DefaultExecutionService::new(
        RuntimeSnapshotHandle::new(client_snapshot()),
        store.clone(),
-        ProviderRegistry::new([Arc::new(ColdFailingProvider) as Arc<dyn Provider>])
-            .expect("provider registry"),
+        ProviderRegistry::new([Arc::new(ColdFailingProvider {
+            requested_model: model.map(|model| PublicModelId::new(model).expect("model")),
+        }) as Arc<dyn Provider>])
+        .expect("provider registry"),
        Arc::new(UnusedAdmissions),
        Arc::new(UnusedCircuits),
        Arc::new(UnusedContinuation),
@@ -369,7 +385,7 @@ fn provider_endpoint_should_persist_its_real_v1_endpoint() {
        vec!["/v1/images/generations".to_owned()],
        "execution error: {error:?}"
    );
-    assert_eq!(store.requested_models(), vec![None]);
+    assert_eq!(store.requested_models(), vec![model.map(str::to_owned)]);
    let upstream_models = store.upstream_models();
    assert!(!upstream_models.is_empty());
    assert!(upstream_models.iter().all(Option::is_none));
@@ -13,12 +13,12 @@ const REQUEST_ERROR_SELECT: &str = "select 'model_request'::text as source,
       mr.provider_account_authentication_kind_snapshot
         as provider_account_authentication_kind,
       mr.upstream_model_id, mr.upstream_transport,
-       coalesce(mr.error_kind, 'failed') as failure_kind,
+       mr.error_kind as failure_kind,
       mr.upstream_send_state,
       mr.client_status_code, mr.upstream_status_code,
       mr.provider_error_code, mr.client_response_id, mr.upstream_request_id,
       mr.latency_ms,
-       coalesce(mr.error_message, mr.error_kind, 'request failed') as message,
+       coalesce(mr.error_message, mr.error_kind) as message,
       mr.raw_upstream_error,
       host(mr.client_ip) as client_ip, mr.user_agent,
       mr.reasoning_effort, mr.reasoning_preset, mr.request_kind,
@@ -32,7 +32,7 @@ const REQUEST_ERROR_SELECT: &str = "select 'model_request'::text as source,
       mr.completed_at as occurred_at,
       'model_request:' || mr.id as stable_sort_id
 from model_requests mr
-where mr.outcome = 'failed'";
+where true";

 const OPS_EVENT_SELECT: &str = "select 'ops_event'::text as source,
       oe.id as event_id, oe.model_request_id as request_id, oe.attempt_index,
@@ -106,7 +106,7 @@ pub(crate) async fn count_ops_errors(
    filter: &OpsErrorFilter,
 ) -> StoreResult<u64> {
    let mut statement = QueryBuilder::<Postgres>::new(
-        "select coalesce(sum(source_count), 0)::bigint from (select count(*)::bigint as source_count from model_requests mr where mr.outcome = 'failed'",
+        "select coalesce(sum(source_count), 0)::bigint from (select count(*)::bigint as source_count from model_requests mr where true",
    );
    push_request_error_predicates(&mut statement, range, filter);
    statement.push(
@@ -127,6 +127,9 @@ fn push_request_error_predicates(
    range: ObservabilityRange,
    filter: &OpsErrorFilter,
 ) {
+    // 错误事实独立于请求结束状态；主动取消不属于需要排查的错误。
+    // 列表和总数共用此条件，避免流式响应中的错误因 outcome 被漏掉。
+    statement.push(" and mr.error_kind is not null and mr.error_kind <> 'cancelled'");
    push_range(statement, "mr.completed_at", range);
    for (column, value) in [
        ("mr.client_api_key_ref", &filter.client_api_key_ref),
@@ -258,6 +258,119 @@ async fn ops_search_should_treat_sql_wildcards_as_literals() {
    database.close().await;
 }

+#[tokio::test]
+async fn ops_should_include_incomplete_upstream_errors() {
+    let Some(database) = TestDatabase::create("ops_incomplete_error").await else {
+        return;
+    };
+    let now = Utc::now();
+    seed_observability_facts(&database.pool, now).await.unwrap();
+    let raw_error = r#"{"type":"error","status":400,"error":{"type":"invalid_request_error","message":"Invalid `previous_response_id`."}}"#;
+    sqlx::query(
+        "update model_requests
+            set outcome = 'incomplete', error_kind = 'invalid_request',
+                client_transport = 'websocket', upstream_transport = 'websocket',
+                downstream_committed_at = completed_at,
+                client_status_code = null, upstream_status_code = 400,
+                provider_error_code = null, raw_upstream_error = $1
+          where id = 'req_observe_failed'",
+    )
+    .bind(raw_error)
+    .execute(&database.pool)
+    .await
+    .unwrap();
+    let page = admin_observability_store(&database.pool)
+        .list_ops_errors(admin_observability::OpsErrorQuery {
+            range: admin_observability::TimeRange::new(
+                now - TimeDelta::hours(1),
+                now + TimeDelta::hours(1),
+            )
+            .unwrap(),
+            filter: admin_observability::OpsErrorFilter {
+                request_id: Some("req_observe_failed".to_owned()),
+                status_code: Some(400),
+                ..admin_observability::OpsErrorFilter::default()
+            },
+            current_page: 1,
+            page_size: PageSize::new(10).unwrap(),
+        })
+        .await
+        .expect("incomplete request must be available in admin error troubleshooting");
+    database.close().await;
+
+    assert_eq!(page.total, 1);
+    assert_eq!(page.items.len(), 1);
+    let error = &page.items[0];
+    assert_eq!(error.source, "model_request");
+    assert_eq!(error.failure_kind, "invalid_request");
+    assert_eq!(error.upstream_status_code, Some(400));
+    assert_eq!(error.client_status_code, None);
+    assert_eq!(error.raw_upstream_error.as_deref(), Some(raw_error));
+}
+
+#[tokio::test]
+async fn ops_should_select_errors_independently_of_request_outcome() {
+    let Some(database) = TestDatabase::create("ops_error_outcomes").await else {
+        return;
+    };
+    let now = Utc::now();
+    seed_observability_facts(&database.pool, now).await.unwrap();
+    let range =
+        ObservabilityRange::new(now - TimeDelta::hours(1), now + TimeDelta::hours(1)).unwrap();
+    let repository = observability_repository(&database.pool);
+    let mut results = Vec::new();
+    for (outcome, error_kind, expected_count) in [
+        ("incomplete", Some("upstream_unavailable"), 1),
+        ("succeeded", Some("upstream_unavailable"), 1),
+        ("failed", Some("upstream_unavailable"), 1),
+        ("failed", None, 0),
+        ("incomplete", None, 0),
+        ("cancelled", Some("cancelled"), 0),
+        ("incomplete", Some("cancelled"), 0),
+        ("failed", Some("cancelled"), 0),
+        ("succeeded", None, 0),
+    ] {
+        sqlx::query(
+            "update model_requests set outcome = $1, error_kind = $2
+              where id = 'req_observe_success'",
+        )
+        .bind(outcome)
+        .bind(error_kind)
+        .execute(&database.pool)
+        .await
+        .unwrap();
+        let page = repository
+            .list_ops_errors(OpsErrorQuery {
+                range,
+                filter: OpsErrorFilter {
+                    request_id: Some("req_observe_success".to_owned()),
+                    ..OpsErrorFilter::default()
+                },
+                current_page: 1,
+                page_size: ObservabilityPageSize::new(10).unwrap(),
+            })
+            .await
+            .unwrap();
+        results.push((outcome, error_kind, expected_count, page));
+    }
+    database.close().await;
+
+    for (outcome, error_kind, expected_count, page) in results {
+        assert_eq!(
+            page.total, expected_count,
+            "total for outcome={outcome}, error_kind={error_kind:?}"
+        );
+        assert_eq!(
+            page.items.len() as u64,
+            expected_count,
+            "items for outcome={outcome}, error_kind={error_kind:?}"
+        );
+        if let Some(error) = page.items.first() {
+            assert_eq!(Some(error.failure_kind.as_str()), error_kind);
+        }
+    }
+}
+
 #[tokio::test]
 async fn recovered_continuation_failure_should_be_visible_in_ops_but_hidden_from_business_metrics()
 {
@@ -23,8 +23,8 @@ getrandom = "0.4.3"
 hex = "0.4.3"
 # TLS 依赖按官方 Core 锁文件固定，包含用于约束传递依赖的直接声明。
 # 升级前须重复采样 ClientHello，核对算法、扩展集合及排序约束、key_share 与 ALPN。
-aws-lc-rs = "=1.16.2"
-hyper = { version = "=1.8.1", default-features = false }
+aws-lc-rs = "=1.18.1"
+hyper = { version = "=1.11.1", default-features = false }
 hyper-rustls = "=0.27.7"
 # 与官方 Codex Desktop 一致：reqwest 不启用压缩特性，上游请求不带 accept-encoding。
 reqwest = { version = "=0.12.28", default-features = false, features = ["json", "stream", "cookies", "rustls-tls-native-roots", "http2"] }
@@ -33,15 +33,15 @@ semver = "1.0.27"
 # 与官方 rustls 路径一致：aws-lc 支持 P-521，优先提供 X25519MLKEM768。
 # reqwest 的 roots feature 仍会编译 ring；实际 provider 由 TLS 初始化显式选择。
 rustls = { version = "=0.23.36", default-features = false, features = ["aws_lc_rs", "prefer-post-quantum", "std", "tls12"] }
-rustls-native-certs = "=0.8.3"
-rustls-pki-types = "=1.14.0"
+rustls-native-certs = "=0.8.4"
+rustls-pki-types = "=1.15.1"
 secrecy = "0.10.3"
 serde = { version = "1.0.228", features = ["derive"] }
 serde_json = { version = "1.0.150", features = ["arbitrary_precision", "preserve_order"] }
 sha2 = "0.11.0"
 subtle = "2.6.1"
 thiserror = "2.0.18"
-tokio = { version = "=1.52.3", default-features = false, features = ["fs", "macros", "rt", "sync", "time"] }
+tokio = { version = "=1.53.1", default-features = false, features = ["fs", "macros", "rt", "sync", "time"] }
 tokio-rustls = { version = "=0.26.4", default-features = false }
 tokio-util = { version = "0.7.18", features = ["rt"] }
 # 官方 Codex WebSocket 需要上游尚未发布的 proxy 与 permessage-deflate 实现，固定到已审计提交。
@@ -1,5 +1,7 @@
 //! OpenAI attempt 的选择、发送与响应流执行。

+use gateway_core::metering::{CalculatedCost, Usage};
+
 use super::*;

 impl CodexProvider {
@@ -490,6 +492,14 @@ pub(super) fn cold_json_response_stream(request: ColdJsonResponse) -> EventStrea
        let response_meta =
            ResponseMeta::for_provider_endpoint(request.context.request_id().as_str());
        yield ProviderEvent::canonical(GatewayEvent::Started(response_meta.clone()));
+        if matches!(request.endpoint_path, CODEX_IMAGE_GENERATIONS_PATH | CODEX_IMAGE_EDITS_PATH)
+            && let Some((usage, cost)) = image_response_metering(&request.body, &response.body)
+        {
+            yield ProviderEvent::canonical(GatewayEvent::Usage(usage));
+            if let Some(cost) = cost {
+                yield ProviderEvent::canonical(GatewayEvent::CalculatedCost(cost));
+            }
+        }
        let wire = ProtocolWireEvent::raw_json(PROVIDER_NAME, response.body).map_err(|_| {
            provider_error(ProviderErrorKind::Protocol, UpstreamSendState::Sent)
        })?;
@@ -500,6 +510,37 @@ pub(super) fn cold_json_response_stream(request: ColdJsonResponse) -> EventStrea
    })
 }

+fn image_response_metering(
+    request_body: &[u8],
+    body: &[u8],
+) -> Option<(Usage, Option<CalculatedCost>)> {
+    // 只保留 usage，跳过通常很大的 base64 图片；原始响应仍按字节透传。
+    #[derive(Deserialize)]
+    struct ImageUsageEnvelope {
+        usage: Option<Value>,
+    }
+
+    let raw = serde_json::from_slice::<ImageUsageEnvelope>(body)
+        .ok()?
+        .usage?;
+    let mut usage = Usage::new();
+    usage.input_tokens = raw.get("input_tokens").and_then(Value::as_u64);
+    usage.output_tokens = raw.get("output_tokens").and_then(Value::as_u64);
+    usage.cached_tokens = raw
+        .pointer("/input_tokens_details/cached_tokens")
+        .and_then(Value::as_u64);
+    usage.image_input_tokens = raw
+        .pointer("/input_tokens_details/image_tokens")
+        .and_then(Value::as_u64);
+    usage.image_output_tokens = raw
+        .pointer("/output_tokens_details/image_tokens")
+        .and_then(Value::as_u64);
+    // 总量是上游独立报告的事实；图片明细是总输入/输出的子集，不能再次相加。
+    usage.total_tokens = raw.get("total_tokens").and_then(Value::as_u64);
+    let cost = crate::transport::usage::image_calculated_cost(request_body, &raw);
+    (usage != Usage::default()).then_some((usage, cost))
+}
+
 pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream {
    let ColdResponse {
        client,
@@ -237,6 +237,15 @@ impl Provider for CodexProvider {
                _ => None,
            };
            return ProviderRequestObservation {
+                requested_model: match operation {
+                    Operation::Search(request) => {
+                        observation::endpoint_requested_model(request.payload())
+                    }
+                    Operation::GenerateImage(request) => {
+                        observation::endpoint_requested_model(request.payload())
+                    }
+                    _ => None,
+                },
                continuation: ContinuationRequestObservation {
                    affinity_hash: affinity.map(|affinity| affinity.persistence_hash().to_owned()),
                    ..Default::default()
@@ -259,6 +268,7 @@ impl Provider for CodexProvider {
            requested: previous_response_id.is_some(),
        };
        ProviderRequestObservation {
+            requested_model: None,
            reasoning_effort,
            reasoning_preset: semantics.reasoning_preset.map(str::to_owned),
            request_kind: semantics.request_kind,
@@ -2,6 +2,23 @@

 use super::*;

+pub(super) fn endpoint_requested_model(
+    payload: &gateway_core::operation::RawJsonPayload,
+) -> Option<gateway_core::routing::PublicModelId> {
+    if payload.protocol() != PROVIDER_NAME {
+        return None;
+    }
+    // 只读取 model，避免把编辑请求中的整张图片复制到观测数据。
+    #[derive(Deserialize)]
+    struct RequestModel {
+        model: Option<Value>,
+    }
+    let model = serde_json::from_slice::<RequestModel>(payload.body())
+        .ok()?
+        .model?;
+    gateway_core::routing::PublicModelId::new(model.as_str()?.to_owned()).ok()
+}
+
 /// OpenAI Responses 的观测状态完全归 Provider 所有。
 ///
 /// 每次原始 SSE/WS 事件推进状态后重新生成不可变 observation，Core 只负责携带、
@@ -851,7 +851,9 @@ pub(super) fn websocket_success_decision(
 ) -> CodexTransportDecision {
    match requirement {
        TransportRequirement::ExactWebSocketContinuation => CodexTransportDecision::ExactWebSocket,
-        TransportRequirement::ExplicitWebSocketWarmup => CodexTransportDecision::RequiredWebSocket,
+        TransportRequirement::ExplicitWebSocketWarmup | TransportRequirement::WebSocketNewChain => {
+            CodexTransportDecision::RequiredWebSocket
+        }
        _ if prepared.reused() => CodexTransportDecision::ReusedWebSocket,
        _ => CodexTransportDecision::ConnectedWebSocket,
    }
@@ -272,6 +272,7 @@ impl CodexBackendClient {
                Some(WEBSOCKET_FAST_PATH_BUDGET)
            }
            TransportRequirement::ExplicitWebSocketWarmup
+            | TransportRequirement::WebSocketNewChain
            | TransportRequirement::ExactWebSocketContinuation
            | TransportRequirement::ExternalUnknown => None,
            TransportRequirement::HttpRequired => None,
@@ -121,6 +121,8 @@ pub enum TransportRequirement {
    HttpRequired,
    /// `generate=false + store=false` 预热必须保留在同一条 WebSocket。
    ExplicitWebSocketWarmup,
+    /// 客户端 WebSocket 的非持久化新链，必须在池化连接上建立后续续接状态。
+    WebSocketNewChain,
    /// 只能使用持有指定 connection-local response 的精确 WebSocket。
    ExactWebSocketContinuation,
    /// previous response 已持久化，允许 WebSocket 或 HTTP/2。
@@ -136,7 +138,9 @@ impl TransportRequirement {
    pub fn requires_websocket(self) -> bool {
        matches!(
            self,
-            Self::ExplicitWebSocketWarmup | Self::ExactWebSocketContinuation
+            Self::ExplicitWebSocketWarmup
+                | Self::WebSocketNewChain
+                | Self::ExactWebSocketContinuation
        )
    }

@@ -150,7 +154,7 @@ impl TransportRequirement {

    /// 无续接依赖的新请求可以在明确的连接级拒绝后重新建连。
    pub fn allows_connection_restart(self) -> bool {
-        matches!(self, Self::NewChain)
+        matches!(self, Self::NewChain | Self::WebSocketNewChain)
    }

    /// 用于审计与遥测的稳定名称。
@@ -158,6 +162,7 @@ impl TransportRequirement {
        match self {
            Self::HttpRequired => "http_required",
            Self::ExplicitWebSocketWarmup => "explicit_websocket_warmup",
+            Self::WebSocketNewChain => "websocket_new_chain",
            Self::ExactWebSocketContinuation => "exact_websocket_continuation",
            Self::PersistedContinuation => "persisted_continuation",
            Self::ExternalUnknown => "external_unknown",
@@ -184,6 +189,11 @@ pub fn transport_requirement(request: &CodexResponsesRequest) -> TransportRequir
                TransportRequirement::ExternalUnknown
            }
        },
+        // 客户端会在下一轮提交 response ID；HTTP store=false 的成功响应无法
+        // 在池化 WebSocket 上续接，所以首轮也不能按普通快路径降级到 HTTP。
+        None if request.downstream_websocket_connection_id.is_some() && !request.store() => {
+            TransportRequirement::WebSocketNewChain
+        }
        None => TransportRequirement::NewChain,
    }
 }
@@ -1,6 +1,6 @@
 use gateway_core::metering::{
-    CalculatedCostAmounts, CalculatedCostBreakdown, CalculatedCostRates, CurrencyCode, Decimal,
-    Money,
+    CalculatedCost, CalculatedCostAmounts, CalculatedCostBreakdown, CalculatedCostRates,
+    CurrencyCode, Decimal, Money,
 };
 use gateway_protocol::openai::events::{TokenUsage, retry_after_seconds_from_body};
 use reqwest::StatusCode;
@@ -493,6 +493,76 @@ fn openai_billing_breakdown_with_context(
    ))
 }

+/// 独立 Images 端点按公开标准 API 单价估算；不是 ChatGPT 账号实际扣费。
+pub(crate) fn image_calculated_cost(request_body: &[u8], usage: &Value) -> Option<CalculatedCost> {
+    #[derive(serde::Deserialize)]
+    struct ImageModel {
+        model: String,
+    }
+    // 只读取模型，跳过编辑请求中可能很大的 base64 图片。
+    let request = serde_json::from_slice::<ImageModel>(request_body).ok()?;
+    if !matches!(
+        request.model.as_str(),
+        "gpt-image-2" | "gpt-image-2-2026-04-21"
+    ) {
+        return None;
+    }
+    let input = usage.get("input_tokens")?.as_u64()?;
+    let output = usage.get("output_tokens")?.as_u64()?;
+    let details = usage.get("input_tokens_details")?;
+    let text_input = details.get("text_tokens")?.as_u64()?;
+    let image_input = details.get("image_tokens")?.as_u64()?;
+    if text_input.checked_add(image_input)? != input {
+        return None;
+    }
+    if let Some(total) = usage.get("total_tokens")
+        && total.as_u64()? != input.checked_add(output)?
+    {
+        return None;
+    }
+    // 此模型仅输出图片；上游如报告其他输出模态，不能套用图片单价。
+    if let Some(details) = usage.get("output_tokens_details")
+        && (details.get("image_tokens")?.as_u64()? != output
+            || details.get("text_tokens")?.as_u64()? != 0)
+    {
+        return None;
+    }
+    let cached = match details.get("cached_tokens") {
+        None => 0,
+        Some(value) => value.as_u64()?,
+    };
+    if cached > input {
+        return None;
+    }
+    // 缓存只给总量时，混合输入无法判定应套用哪种缓存单价。
+    let (cached_text, cached_image) = match (text_input, image_input, cached) {
+        (_, _, 0) => (0, 0),
+        (_, 0, cached) => (cached, 0),
+        (0, _, cached) => (0, cached),
+        (_, _, cached) if cached == input => (text_input, image_input),
+        _ => return None,
+    };
+    // https://developers.openai.com/api/docs/pricing (2026-09-08)
+    // USD / 1M tokens: text 5 / cached 1.25; image 8 / cached 2 / output 30.
+    let text = token_amounts(
+        TokenRates::new(50_000, 0, 12_500),
+        0,
+        text_input,
+        0,
+        cached_text,
+        0,
+    )?;
+    let image = token_amounts(
+        TokenRates::new(80_000, 300_000, 20_000),
+        0,
+        image_input,
+        output,
+        cached_image,
+        0,
+    )?;
+    CalculatedCost::from_usd_ticks(text.total_ticks.checked_add(image.total_ticks)?).ok()
+}
+
 fn model_pricing(model: &str) -> Option<ModelPricing> {
    let normalized = normalize_model_name(model);
    if UNPRICED_MODELS
@@ -151,6 +151,9 @@ async fn forward_websocket_response_stream(state: WebSocketStreamForwardState) {
        .map(|pool_return| std::mem::take(&mut pool_return.continuation))
        .unwrap_or_default();
    let mut last_event_type = None;
+    // 官方 run_websocket_response_stream 按响应消费 metadata 事件；它们不属于握手头。
+    // 复用连接时恢复握手快照，避免上一轮模型信息及普通响应头随每次请求累积。
+    let opening_response_metadata = metadata.response_metadata.clone();
    loop {
        let message = tokio::select! {
            biased;
@@ -323,6 +326,7 @@ async fn forward_websocket_response_stream(state: WebSocketStreamForwardState) {
            );
            match terminal {
                WebSocketTerminalKind::Completed => {
+                    metadata.response_metadata = opening_response_metadata;
                    finish_stream_websocket(websocket, metadata, continuation, pool_return.take())
                        .await;
                }
@@ -26,6 +26,7 @@ use gateway_core::error::{
 };
 use gateway_core::event::{GatewayEvent, WebSocketPoolKind};
 use gateway_core::lifecycle::CancellationToken;
+use gateway_core::metering::Usage;
 use gateway_core::operation::{
    CapabilityRequirements, GenerateRequest, ImageRequest, ImageRequestKind, Operation,
    OperationKind, ProtocolPayload, ProviderSessionState, RawJsonPayload, StandaloneSearchRequest,
@@ -1051,6 +1052,13 @@ async fn image_endpoints_bypass_only_the_text_catalog_and_preserve_the_current_c
        let mut observed_http_json = false;
        while let Some(event) = stream.next().await {
            let event = event.expect("image provider event");
+            assert!(
+                event
+                    .canonical_facts()
+                    .iter()
+                    .all(|fact| !matches!(fact, GatewayEvent::Usage(_))),
+                "missing upstream usage must remain unknown"
+            );
            completed |= event
                .canonical_facts()
                .iter()
@@ -1075,6 +1083,289 @@ async fn image_endpoints_bypass_only_the_text_catalog_and_preserve_the_current_c
    server.verify().await;
 }

+#[tokio::test]
+async fn image_endpoints_should_report_usage_before_delivering_the_unchanged_body() {
+    let response = br#"{ "created":1778832973,"data":[{"b64_json":"AAEC"}],"quality":"medium","size":"1024x1536","usage":{"input_tokens":1474,"input_tokens_details":{"image_tokens":1457,"text_tokens":17,"cached_tokens":100},"output_tokens":1372,"output_tokens_details":{"image_tokens":1372,"text_tokens":0},"total_tokens":2846},"future":9007199254740993 }"#;
+    let expected = Usage {
+        input_tokens: Some(1474),
+        output_tokens: Some(1372),
+        cached_tokens: Some(100),
+        image_input_tokens: Some(1457),
+        image_output_tokens: Some(1372),
+        total_tokens: Some(2846),
+        ..Usage::default()
+    };
+    for kind in [ImageRequestKind::Generation, ImageRequestKind::Edit] {
+        let events = image_usage_events(kind, response).await;
+        let usage = events
+            .iter()
+            .flat_map(|event| event.canonical_facts())
+            .filter_map(|fact| match fact {
+                GatewayEvent::Usage(usage) => Some(usage.clone()),
+                _ => None,
+            })
+            .collect::<Vec<_>>();
+        assert_eq!(usage.as_slice(), std::slice::from_ref(&expected));
+        let usage_index = events
+            .iter()
+            .position(|event| {
+                event
+                    .canonical_facts()
+                    .iter()
+                    .any(|fact| matches!(fact, GatewayEvent::Usage(_)))
+            })
+            .expect("usage event");
+        let wire_index = events
+            .iter()
+            .position(|event| {
+                event
+                    .wire_event()
+                    .and_then(|wire| wire.raw_json_body())
+                    .is_some()
+            })
+            .expect("raw image response");
+        assert!(
+            usage_index < wire_index,
+            "usage must be observed before image delivery"
+        );
+        assert_eq!(
+            events[wire_index]
+                .wire_event()
+                .and_then(|wire| wire.raw_json_body())
+                .map(|body| body.as_ref()),
+            Some(response.as_slice())
+        );
+    }
+}
+
+#[tokio::test]
+async fn image_usage_should_preserve_unknown_fields_and_explicit_zero_counts() {
+    let cases = [
+        (json!(null), None),
+        (json!({}), None),
+        (json!([]), None),
+        (
+            json!({"input_tokens": -1, "output_tokens": "42", "total_tokens": 1.5}),
+            None,
+        ),
+        (
+            json!({"input_tokens": 0, "output_tokens": 0, "total_tokens": 0, "input_tokens_details": {"image_tokens": 0, "cached_tokens": 0}, "output_tokens_details": {"image_tokens": 0}}),
+            Some(Usage {
+                input_tokens: Some(0),
+                output_tokens: Some(0),
+                total_tokens: Some(0),
+                cached_tokens: Some(0),
+                image_input_tokens: Some(0),
+                image_output_tokens: Some(0),
+                ..Usage::default()
+            }),
+        ),
+        (
+            json!({"input_tokens": 17, "output_tokens": 1372, "input_tokens_details": {"image_tokens": "invalid"}, "output_tokens_details": {"image_tokens": 1372}}),
+            Some(Usage {
+                input_tokens: Some(17),
+                output_tokens: Some(1372),
+                image_output_tokens: Some(1372),
+                ..Usage::default()
+            }),
+        ),
+    ];
+    for (usage, expected) in cases {
+        let response = serde_json::to_vec(
+            &json!({"created":1778832973,"data":[{"b64_json":"AAEC"}],"usage":usage}),
+        )
+        .expect("image response JSON");
+        let events = image_usage_events(ImageRequestKind::Generation, &response).await;
+        let actual = events
+            .iter()
+            .flat_map(|event| event.canonical_facts())
+            .filter_map(|fact| match fact {
+                GatewayEvent::Usage(usage) => Some(usage.clone()),
+                _ => None,
+            })
+            .collect::<Vec<_>>();
+        assert_eq!(actual, expected.into_iter().collect::<Vec<_>>());
+        let raw = events
+            .iter()
+            .find_map(|event| event.wire_event().and_then(|wire| wire.raw_json_body()));
+        assert_eq!(raw.map(|body| body.as_ref()), Some(response.as_slice()));
+    }
+}
+
+#[tokio::test]
+async fn image_prices_should_use_modality_rates_and_precede_delivery() {
+    // First case is the real gpt-image-2 response verified on 2026-09-08.
+    let cases = [
+        (18, 0, 229, None, 69_600_000_u128),
+        (17, 1457, 1372, None, 529_010_000),
+        (100, 0, 10, Some(40), 6_500_000),
+        (0, 100, 10, Some(40), 8_600_000),
+        (20, 80, 10, Some(100), 4_850_000),
+        (0, 0, 0, Some(0), 0),
+    ];
+    for (text, image, output, cached, expected_ticks) in cases {
+        let mut usage = json!({
+            "input_tokens":text+image,
+            "input_tokens_details":{"text_tokens":text,"image_tokens":image},
+            "output_tokens":output,
+            "output_tokens_details":{"text_tokens":0,"image_tokens":output},
+            "total_tokens":text+image+output,
+        });
+        if let Some(cached) = cached {
+            usage["input_tokens_details"]["cached_tokens"] = json!(cached);
+        }
+        let response = serde_json::to_vec(&json!({"data":[{"b64_json":"AAEC"}],"usage":usage}))
+            .expect("response");
+        for (kind, model) in [
+            (ImageRequestKind::Generation, "gpt-image-2"),
+            (ImageRequestKind::Edit, "gpt-image-2-2026-04-21"),
+        ] {
+            let events = image_metering_events(kind, &response, model).await;
+            let costs = events
+                .iter()
+                .enumerate()
+                .flat_map(|(index, event)| {
+                    event
+                        .canonical_facts()
+                        .iter()
+                        .filter_map(move |fact| match fact {
+                            GatewayEvent::CalculatedCost(cost) => Some((index, *cost)),
+                            _ => None,
+                        })
+                })
+                .collect::<Vec<_>>();
+            assert_eq!(costs.len(), 1);
+            let (cost_index, cost) = costs[0];
+            assert_eq!(cost.total().amount().scaled(), expected_ticks);
+            assert_eq!(cost.into_estimate().source().as_str(), "calculated");
+            let wire_index = events
+                .iter()
+                .position(|event| event.wire_event().is_some())
+                .expect("wire event");
+            assert!(cost_index < wire_index);
+            assert_eq!(
+                events[wire_index]
+                    .wire_event()
+                    .and_then(|wire| wire.raw_json_body())
+                    .map(|body| body.as_ref()),
+                Some(response.as_slice())
+            );
+        }
+    }
+}
+
+#[tokio::test]
+async fn image_prices_should_remain_unknown_when_modality_or_model_is_uncertain() {
+    let valid = json!({
+        "input_tokens":100,
+        "input_tokens_details":{"text_tokens":20,"image_tokens":80},
+        "output_tokens":10,
+        "output_tokens_details":{"text_tokens":0,"image_tokens":10},
+        "total_tokens":110,
+    });
+    let mut cases = vec![
+        ("gpt-image-future", valid.clone()),
+        ("gpt-image-2-future", valid.clone()),
+        ("gpt-image-2", json!(null)),
+        (
+            "gpt-image-2",
+            json!({"input_tokens":100,"output_tokens":10}),
+        ),
+    ];
+    for (pointer, value) in [
+        ("/input_tokens", json!(101)),
+        ("/input_tokens_details/text_tokens", json!(null)),
+        ("/input_tokens_details/image_tokens", json!("80")),
+        ("/output_tokens_details/text_tokens", json!(1)),
+        ("/output_tokens_details/image_tokens", json!(9)),
+        ("/output_tokens", json!(-1)),
+        ("/total_tokens", json!(111)),
+    ] {
+        let mut usage = valid.clone();
+        *usage.pointer_mut(pointer).expect("existing field") = value;
+        cases.push(("gpt-image-2", usage));
+    }
+    for cached in [json!(40), json!(101), json!(null), json!("40")] {
+        let mut usage = valid.clone();
+        usage["input_tokens_details"]["cached_tokens"] = cached;
+        cases.push(("gpt-image-2", usage));
+    }
+    cases.push(("gpt-image-2", json!({
+        "input_tokens":u64::MAX,"input_tokens_details":{"text_tokens":u64::MAX,"image_tokens":0},
+        "output_tokens":10,
+    })));
+    for (model, usage) in cases {
+        let response = serde_json::to_vec(&json!({"data":[{"b64_json":"AAEC"}],"usage":usage}))
+            .expect("response");
+        let events = image_metering_events(ImageRequestKind::Generation, &response, model).await;
+        assert!(
+            events
+                .iter()
+                .flat_map(|event| event.canonical_facts())
+                .all(|fact| !matches!(fact, GatewayEvent::CalculatedCost(_)))
+        );
+        assert_eq!(
+            events
+                .iter()
+                .find_map(|event| event.wire_event().and_then(|wire| wire.raw_json_body()))
+                .map(|body| body.as_ref()),
+            Some(response.as_slice())
+        );
+    }
+}
+
+async fn image_usage_events(
+    kind: ImageRequestKind,
+    response: &[u8],
+) -> Vec<gateway_core::event::ProviderEvent> {
+    image_metering_events(kind, response, "gpt-image-2").await
+}
+
+async fn image_metering_events(
+    kind: ImageRequestKind,
+    response: &[u8],
+    model: &str,
+) -> Vec<gateway_core::event::ProviderEvent> {
+    let store = Arc::new(MemoryAccountStore::default());
+    create_account(&store, "acct_provider_contract").await;
+    let server = MockServer::start().await;
+    let endpoint = match kind {
+        ImageRequestKind::Generation => "/codex/images/generations",
+        ImageRequestKind::Edit => "/codex/images/edits",
+    };
+    Mock::given(method("POST"))
+        .and(path(endpoint))
+        .respond_with(
+            ResponseTemplate::new(200).set_body_raw(response.to_vec(), "application/json"),
+        )
+        .expect(1)
+        .mount(&server)
+        .await;
+    let provider = provider_with_base_url(&store, server.uri());
+    let payload = RawJsonPayload::new(
+        "openai",
+        Bytes::from(
+            serde_json::to_vec(&json!({"model":model,"prompt":"draw a tree"}))
+                .expect("request JSON"),
+        ),
+    )
+    .expect("image payload");
+    let operation = Operation::GenerateImage(ImageRequest::from_raw_json(kind, payload));
+    let mut stream = provider
+        .execute(
+            planned_provider_endpoint_request("openai", operation),
+            context("req_image_usage", CancellationToken::new()),
+        )
+        .await
+        .expect("prepare image stream");
+    let mut events = Vec::new();
+    while let Some(event) = stream.next().await {
+        events.push(event.expect("image provider event"));
+    }
+    server.verify().await;
+    events
+}
+
 #[tokio::test]
 async fn image_endpoint_returns_the_exact_upstream_error_response() {
    let store = Arc::new(MemoryAccountStore::default());
@@ -2521,6 +2812,127 @@ async fn fallback_attempt_transport_forces_http_sse_for_a_websocket_request() {
    }
 }

+#[tokio::test]
+async fn downstream_websocket_new_chain_should_override_session_and_attempt_http_fallback() {
+    const SESSION_ID: &str = "downstream-required-session";
+
+    let store = Arc::new(MemoryAccountStore::default());
+    create_account(&store, "acct_http_sse_exhausted").await;
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let base_url = format!("http://{}", listener.local_addr().unwrap());
+    let server = tokio::spawn(async move {
+        let (mut opening, _) = listener.accept().await.unwrap();
+        let request = capture_http_request(&mut opening).await;
+        assert!(String::from_utf8_lossy(&request).starts_with("GET /codex/responses"));
+        opening
+            .write_all(
+                b"HTTP/1.1 426 Upgrade Required\r\ncontent-length: 0\r\nconnection: close\r\n\r\n",
+            )
+            .await
+            .unwrap();
+
+        let (stream, _) = listener.accept().await.unwrap();
+        let mut websocket = accept_codex_test_websocket(stream).await;
+        for id in ["resp_required_session", "resp_required_attempt"] {
+            let request = websocket.next().await.unwrap().unwrap();
+            let payload: Value = serde_json::from_str(request.to_text().unwrap()).unwrap();
+            assert_eq!(payload["store"], false);
+            websocket
+                .send(Message::Text(
+                    json!({
+                        "type": "response.completed",
+                        "response": {
+                            "id": id,
+                            "model": "gpt-5.4",
+                            "status": "completed",
+                            "store": false,
+                            "output": [],
+                            "usage": {"input_tokens": 1, "output_tokens": 1, "total_tokens": 2}
+                        }
+                    })
+                    .to_string()
+                    .into(),
+                ))
+                .await
+                .unwrap();
+        }
+    });
+    let provider = provider_with_base_url(&store, base_url);
+    let mut seed = provider
+        .execute(
+            planned_request(
+                "openai",
+                Operation::Generate(generate_with_session_context(SESSION_ID, None, None)),
+            ),
+            context("req_disable_optional_websocket", CancellationToken::new()),
+        )
+        .await
+        .unwrap();
+    let error = loop {
+        match seed.next().await {
+            Some(Ok(_)) => {}
+            Some(Err(error)) => break error,
+            None => panic!("426 opening must enable session HTTP fallback"),
+        }
+    };
+    assert_eq!(
+        error.pre_delivery_retry(),
+        Some(PreDeliveryRetry::SameAccountTransportFallback)
+    );
+    drop(seed);
+
+    for attempt in [
+        context("req_required_session", CancellationToken::new()),
+        fallback_transport_context("req_required_attempt"),
+    ] {
+        let payload = ProtocolPayload::json_object(
+            "openai",
+            Map::from_iter([
+                ("model".to_owned(), json!("gpt-5.4")),
+                ("input".to_owned(), json!("hello")),
+                ("store".to_owned(), json!(false)),
+                ("session_id".to_owned(), json!(SESSION_ID)),
+            ]),
+        )
+        .unwrap()
+        .with_context(Map::from_iter([(
+            "downstream_websocket_connection_id".to_owned(),
+            json!("ws_required_session"),
+        )]));
+        let request = GenerateRequest::from_protocol_payload(payload).with_provider_session_state(
+            ProviderSessionState::new(
+                "openai",
+                Map::from_iter([
+                    ("account_id".to_owned(), json!("acct_http_sse_exhausted")),
+                    ("conversation_id".to_owned(), json!(SESSION_ID)),
+                    ("continuation_scope".to_owned(), json!("persisted")),
+                ]),
+            )
+            .unwrap(),
+        );
+        let mut stream = provider
+            .execute(
+                planned_request("openai", Operation::Generate(request)),
+                attempt,
+            )
+            .await
+            .unwrap();
+        assert_eq!(stream.metadata().transport().as_str(), "websocket");
+        let mut session = None;
+        while let Some(event) = stream.next().await {
+            let event = event.expect("required WebSocket response");
+            if let Some(update) = event.session_update() {
+                session = Some(update.clone());
+            }
+        }
+        assert_eq!(
+            session.unwrap().payload().get("continuation_scope"),
+            Some(&json!("connection_local"))
+        );
+    }
+    server.await.unwrap();
+}
+
 #[tokio::test]
 async fn websocket_turn_state_metadata_is_exposed_through_response_observation() {
    let store = Arc::new(MemoryAccountStore::default());
@@ -5510,6 +5922,52 @@ fn request_observation_reads_openai_metadata_without_changing_the_operation() {
    assert!(matches!(operation, Operation::Generate(_)));
 }

+#[test]
+fn endpoint_observation_should_read_models_without_rewriting_or_requiring_a_catalog() {
+    let store = Arc::new(MemoryAccountStore::default());
+    let provider = provider(&store);
+    let client_key = ClientApiKeyId::new("key_endpoint_model").expect("client key");
+    for model in [
+        json!("gpt-image-future"),
+        json!("gpt-5.6-sol"),
+        json!(null),
+        json!(42),
+        json!(""),
+    ] {
+        let body = serde_json::to_vec(&json!({"model":model,"future":9007199254740993_u64}))
+            .expect("body");
+        let payload =
+            RawJsonPayload::new("openai", Bytes::copy_from_slice(&body)).expect("payload");
+        let operations = [
+            Operation::GenerateImage(ImageRequest::from_raw_json(
+                ImageRequestKind::Generation,
+                payload.clone(),
+            )),
+            Operation::GenerateImage(ImageRequest::from_raw_json(
+                ImageRequestKind::Edit,
+                payload.clone(),
+            )),
+            Operation::Search(StandaloneSearchRequest::from_raw_json(payload)),
+        ];
+        for operation in operations {
+            let observation = provider.request_observation(&operation, &client_key);
+            assert_eq!(
+                observation
+                    .requested_model
+                    .as_ref()
+                    .map(PublicModelId::as_str),
+                model.as_str().filter(|model| !model.is_empty()),
+            );
+            let payload = match &operation {
+                Operation::GenerateImage(request) => request.payload(),
+                Operation::Search(request) => request.payload(),
+                _ => unreachable!(),
+            };
+            assert_eq!(payload.body().as_ref(), body.as_slice());
+        }
+    }
+}
+
 #[test]
 fn request_observation_preserves_the_raw_reasoning_effort() {
    let store = Arc::new(MemoryAccountStore::default());
@@ -115,6 +115,79 @@ async fn cold_websocket_should_fall_back_without_recording_a_successful_connect(
    pool.shutdown().await;
 }

+#[tokio::test(start_paused = true)]
+async fn downstream_websocket_new_chain_should_preserve_continuation_after_slow_opening() {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+    let server = tokio::spawn(async move {
+        let (stream, _) = listener.accept().await.unwrap();
+        tokio::select! {
+            () = tokio::time::sleep(Duration::from_secs(1)) => {}
+            accepted = listener.accept() => {
+                let (mut http, _) = accepted.unwrap();
+                let request = read_http_request(&mut http).await;
+                assert!(request.starts_with("POST /codex/responses HTTP/1.1"));
+                write_completed_sse_response(&mut http).await;
+                return;
+            }
+        }
+        let mut websocket = accept_codex_test_websocket(stream).await;
+        let _first = websocket.next().await.unwrap().unwrap();
+        websocket
+            .send(Message::Text(
+                completed_websocket_response("resp_slow_seed", 2, 1).into(),
+            ))
+            .await
+            .unwrap();
+
+        let second = websocket.next().await.unwrap().unwrap();
+        let second: Value = serde_json::from_str(second.to_text().unwrap()).unwrap();
+        assert_eq!(second["previous_response_id"], "resp_slow_seed");
+        websocket
+            .send(Message::Text(
+                completed_websocket_response("resp_slow_continuation", 2, 1).into(),
+            ))
+            .await
+            .unwrap();
+    });
+    let pool = Arc::new(CodexWebSocketPool::default());
+    let backend = CodexBackendClient::new(
+        reqwest::Client::builder().no_proxy().build().unwrap(),
+        format!("http://{addr}"),
+        test_wire_profile(),
+    )
+    .with_websocket_pool(Arc::clone(&pool));
+    let mut request = new_chain_request("conversation-downstream-slow");
+    request.downstream_websocket_connection_id = Some("ws_downstream_slow".to_owned());
+
+    let first = backend
+        .create_response(
+            &request,
+            request_context("req_slow_seed", Some("chatgpt-account")),
+        )
+        .await
+        .unwrap();
+    assert_eq!(first.transport, CodexBackendTransport::WebSocket);
+    assert!(first.connection_local_continuation);
+
+    request.set_previous_response_id(Some("resp_slow_seed".to_owned()));
+    request.previous_response_scope = Some(PreviousResponseScope::ConnectionLocal);
+    let second = backend
+        .create_response(
+            &request,
+            request_context("req_slow_continuation", Some("chatgpt-account")),
+        )
+        .await
+        .expect("the next turn must use the socket that generated the previous response");
+    assert_eq!(
+        second.transport_metrics.decision,
+        Some(CodexTransportDecision::ExactWebSocket)
+    );
+    assert!(second.body.contains("resp_slow_continuation"));
+    server.await.unwrap();
+    pool.shutdown().await;
+}
+
 #[tokio::test(start_paused = true)]
 async fn external_continuation_should_wait_for_a_cold_websocket() {
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
@@ -273,6 +273,34 @@ fn transport_should_prefer_websocket_when_requested_without_history() {
    assert!(transport_requirement(&request).allows_pre_send_http_fallback());
 }

+#[test]
+fn downstream_websocket_new_chain_should_require_pooled_websocket_without_storage() {
+    for store in [None, Some(false)] {
+        let mut body = serde_json::Map::new();
+        if let Some(store) = store {
+            body.insert("store".to_owned(), json!(store));
+        }
+        let mut request = CodexResponsesRequest::from_body(body);
+        request.downstream_websocket_connection_id = Some("ws_downstream".to_owned());
+
+        let requirement = transport_requirement(&request);
+        assert!(requirement.requires_websocket());
+        assert!(!requirement.allows_pre_send_http_fallback());
+        assert!(requirement.allows_connection_restart());
+    }
+}
+
+#[test]
+fn downstream_websocket_new_chain_should_allow_http_when_response_is_stored() {
+    let mut request = CodexResponsesRequest::from_body(serde_json::Map::from_iter([(
+        "store".to_owned(),
+        json!(true),
+    )]));
+    request.downstream_websocket_connection_id = Some("ws_downstream".to_owned());
+
+    assert!(transport_requirement(&request).allows_pre_send_http_fallback());
+}
+
 #[test]
 fn unspecified_transport_without_history_should_still_be_a_new_chain() {
    let request = codex_request("gpt-test", "be brief", Vec::new());
@@ -0,0 +1,92 @@
+use super::*;
+
+#[tokio::test]
+async fn reused_websocket_should_keep_response_metadata_scoped_to_each_exchange() {
+    const ROUNDS: usize = 40;
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+    let server = tokio::spawn(async move {
+        let (stream, _) = listener.accept().await.unwrap();
+        let mut websocket = accept_codex_test_websocket_with(stream, |_, response| {
+            for value in ["first", "second"] {
+                response
+                    .headers_mut()
+                    .append("x-opening-multi", value.parse().unwrap());
+            }
+        })
+        .await;
+        for round in 0..ROUNDS {
+            websocket.next().await.unwrap().unwrap();
+            websocket
+                .send(Message::Text(
+                    json!({
+                        "type": "codex.response.metadata",
+                        "headers": {
+                            "x-models-etag": format!("etag-{round}"),
+                            "openai-model": format!("model-{round}"),
+                            "x-reasoning-included": "true",
+                            "x-codex-safety-buffering-enabled": "true",
+                            "x-codex-safety-buffering-faster-model": "faster-model",
+                            "x-future-response-metadata": format!("future-{round}"),
+                            "x-codex-turn-state": format!("turn-{round}")
+                        }
+                    })
+                    .to_string()
+                    .into(),
+                ))
+                .await
+                .unwrap();
+            websocket
+                .send(Message::Text(
+                    completed_websocket_response(&format!("resp_metadata_{round}"), 1, 1).into(),
+                ))
+                .await
+                .unwrap();
+        }
+    });
+    let pool = Arc::new(CodexWebSocketPool::new(Duration::from_mins(1)));
+    let backend = CodexBackendClient::new(
+        reqwest::Client::builder().no_proxy().build().unwrap(),
+        format!("http://{addr}"),
+        test_wire_profile(),
+    )
+    .with_websocket_pool(pool);
+    let request = pooled_websocket_request("conversation-response-metadata");
+    let expected = CodexResponseMetadata {
+        client_headers: vec![
+            (
+                "x-opening-multi".to_owned(),
+                bytes::Bytes::from_static(b"first"),
+            ),
+            (
+                "x-opening-multi".to_owned(),
+                bytes::Bytes::from_static(b"second"),
+            ),
+        ],
+        ..CodexResponseMetadata::default()
+    };
+
+    for round in 0..ROUNDS {
+        let response = timeout(
+            Duration::from_secs(10),
+            backend.create_response(
+                &request,
+                request_context("req_response_metadata", Some("chatgpt-account")),
+            ),
+        )
+        .await
+        .expect("response within timeout")
+        .expect("response should complete on the same socket");
+        assert_eq!(response.response_metadata, expected, "exchange {round}");
+        assert!(response.body.contains(&format!("etag-{round}")));
+        assert_eq!(response.turn_state, Some(format!("turn-{round}")));
+        if round > 0 {
+            assert!(
+                response
+                    .websocket_pool_decision
+                    .is_some_and(WebSocketPoolDecision::is_reuse)
+            );
+        }
+    }
+    server.await.unwrap();
+}
@@ -4,6 +4,8 @@ use provider_openai::transport::{
    websocket::write_websocket_audit_artifact_for_dir,
 };

+mod metadata;
+
 fn rate_limit_event(used_percent: u64) -> String {
    json!({
        "type": "codex.rate_limits",
@@ -152,6 +152,7 @@ impl Provider for GrokBuildProvider {
        }
        let semantics = codex_responses_request_semantics(payload.body(), payload.context());
        ProviderRequestObservation {
+            requested_model: None,
            reasoning_effort: semantics.reasoning_effort,
            reasoning_preset: semantics.reasoning_preset.map(str::to_owned),
            request_kind: semantics.request_kind,
@@ -19,7 +19,7 @@
    "@boxicons/vue": "1.0.1",
    "@fontsource-variable/inter": "^5.3.0",
    "@fontsource-variable/jetbrains-mono": "^5.3.0",
-    "@lucide/vue": "^1.37.0",
+    "@lucide/vue": "^1.41.0",
    "@vueuse/core": "^14.4.0",
    "axios": "^1.20.0",
    "dayjs": "^1.11.23",
@@ -31,15 +31,15 @@
    "pinia": "^4.0.3",
    "pinia-plugin-persistedstate": "^4.7.1",
    "vue": "^3.5.42",
-    "vue-router": "^5.3.0"
+    "vue-router": "^5.3.1"
  },
  "devDependencies": {
-    "@antfu/eslint-config": "^9.3.0",
+    "@antfu/eslint-config": "^9.5.1",
    "@tailwindcss/vite": "^4.3.3",
-    "@types/node": "^26.4.0",
+    "@types/node": "^26.4.1",
    "@vitejs/plugin-vue": "^6.0.8",
    "@vue/tsconfig": "^0.9.1",
-    "eslint": "^10.9.1",
+    "eslint": "^10.10.0",
    "eslint-plugin-format": "^2.0.1",
    "eslint-plugin-vuejs-accessibility": "^2.6.0",
    "tailwindcss": "^4.3.3",
@@ -1 +1 @@
-version: "3.3.26"
+version: "3.3.28"
Author	SHA1	Message	Date
ZyyandCodex	f905e4b3fb	chore: bump version to 3.3.28 Co-authored-by: Codex <noreply@openai.com>	2026-09-08 02:42:09 +08:00
ZyyandCodex	ed3e23887b	fix(openai): calculate standalone image generation costs Estimate gpt-image-2 generation and edit costs from the official standard API rates for text input, image input, cached input and image output. Emit calculated costs before image delivery while preserving raw responses. Leave unsupported models, invalid usage and ambiguous mixed cache totals unpriced. Mark calculated total-only amounts with an approximation sign. Validation: 1,107 related backend tests, workspace rustfmt and strict OpenAI/gateway-api Clippy. Replaying the previously verified live usage of 18 text input and 229 image output tokens yields exactly USD 0.00696. Co-authored-by: Codex <noreply@openai.com>	2026-09-08 02:41:14 +08:00
ZyyandCodex	91f35d88cf	fix(openai): record image usage and endpoint request models Persist models declared by standalone image and search requests without routing them through the text model catalog. Extract image token usage before delivering the unchanged JSON response, preserving unknown counts and upstream totals without double-counting image token details. Validation: 1,474 related backend tests, workspace rustfmt and strict Clippy passed. One authorized live gpt-image-2 generation reported 18 input, 229 output and 247 total tokens, matching the canonical usage exactly. Search responses inspected in official source and production logs did not report usage, so their token counts remain unknown. Co-authored-by: Codex <noreply@openai.com>	2026-09-08 02:23:00 +08:00
ZyyandCodex	f465a337e8	refactor(openai): simplify connection and endpoint module files Restore connection.rs for production and integration tests by folding the heartbeat tests into the existing connection test module. Keep all heartbeat coverage while satisfying the workspace layout rules without a subdirectory. Rename provider_endpoint.rs to endpoint.rs in production and tests. Update module declarations, imports and frozen file manifests to match. Validation: gateway application tests (36 passed), all gateway-api tests (263 passed), workspace rustfmt, and strict gateway-api Clippy with all targets and features. Co-authored-by: Codex <noreply@openai.com>	2026-09-08 01:51:00 +08:00
ZyyandCodex	37465e2872	fix(tests): align WebSocket connection module layout Move production and test connection modules to connection/mod.rs so the heartbeat submodule follows the workspace directory module convention. Keep the production and integration test layouts mirrored, and update the frozen file manifests to include the heartbeat regression tests. Validation: gateway application tests (36 passed), all gateway-api tests (263 passed), workspace rustfmt, and strict gateway-api Clippy with all targets and features. Co-authored-by: Codex <noreply@openai.com>	2026-09-08 01:46:08 +08:00
ZyyandCodex	70c18dadd6	chore: bump version to 3.3.27 Co-authored-by: Codex <noreply@openai.com>	2026-09-08 01:37:11 +08:00
ZyyandCodex	881a0eaef6	fix(openai): keep downstream WebSocket connections alive Send a downstream Ping every 25 seconds while responses are pending or the connection is idle. Match the official Codex client's Pong handling without adding a Pong deadline or extending business request timeouts. Skip missed ticks and keep heartbeat writes bounded and cancellable. Record Ping/Pong counts and read/write idle durations in disconnect logs. Serialize correlation IDs as borrowed strings so gateway-api checks do not require the serde/rc feature. Cover silent active responses, delayed ticks, missing Pongs, blocked writes and lifecycle cancellation. Adapt connection expiry coverage for heartbeats. Validation: gateway-api WebSocket tests (38 passed), workspace rustfmt, and strict gateway-api Clippy with all targets and features, without serde/rc. Co-authored-by: Codex <noreply@openai.com>	2026-09-08 01:36:22 +08:00
ZyyandCodex	95f7a5a636	fix(openai): isolate response metadata across WebSocket reuse Restore opening response metadata before returning a completed WebSocket to its pool. Prevent per-response headers, model names and ETags from leaking into subsequent exchanges and growing downstream response headers. Match the official Codex response-scoped metadata handling while preserving opening headers, legitimate multi-value headers and turn-state updates. Cover 40 consecutive exchanges on one socket with a regression test that fails on the second exchange before the fix. Validation: provider-openai transport tests (292 passed), workspace rustfmt, and strict provider-openai Clippy with all targets and features. Co-authored-by: Codex <noreply@openai.com>	2026-09-08 01:10:44 +08:00
dependabot[bot]	871406ab24	chore(deps): bump the minor-and-patch group in /backend with 9 updates (#54 ) Bumps the minor-and-patch group in /backend with 9 updates: \| Package \| From \| To \| \| --- \| --- \| --- \| \| [tokio](https://github.com/tokio-rs/tokio) \| `1.52.3` \| `1.53.1` \| \| [tower-http](https://github.com/tower-rs/tower-http) \| `0.7.0` \| `0.7.1` \| \| [flate2](https://github.com/rust-lang/flate2-rs) \| `1.1.9` \| `1.1.10` \| \| [aws-smithy-runtime-api](https://github.com/smithy-lang/smithy-rs) \| `1.15.0` \| `1.16.0` \| \| [aws-smithy-types](https://github.com/smithy-lang/smithy-rs) \| `1.6.2` \| `1.6.3` \| \| [aws-lc-rs](https://github.com/aws/aws-lc-rs) \| `1.16.2` \| `1.18.1` \| \| [hyper](https://github.com/hyperium/hyper) \| `1.8.1` \| `1.11.1` \| \| [rustls-native-certs](https://github.com/rustls/rustls-native-certs) \| `0.8.3` \| `0.8.4` \| \| [rustls-pki-types](https://github.com/rustls/pki-types) \| `1.14.0` \| `1.15.1` \| Updates `tokio` from 1.52.3 to 1.53.1 - [Release notes](https://github.com/tokio-rs/tokio/releases) - [Commits](https://github.com/tokio-rs/tokio/compare/tokio-1.52.3...tokio-1.53.1) Updates `tower-http` from 0.7.0 to 0.7.1 - [Release notes](https://github.com/tower-rs/tower-http/releases) - [Commits](https://github.com/tower-rs/tower-http/compare/tower-http-0.7.0...tower-http-0.7.1) Updates `flate2` from 1.1.9 to 1.1.10 - [Release notes](https://github.com/rust-lang/flate2-rs/releases) - [Commits](https://github.com/rust-lang/flate2-rs/compare/1.1.9...1.1.10) Updates `aws-smithy-runtime-api` from 1.15.0 to 1.16.0 - [Release notes](https://github.com/smithy-lang/smithy-rs/releases) - [Changelog](https://github.com/smithy-lang/smithy-rs/blob/main/CHANGELOG.md) - [Commits](https://github.com/smithy-lang/smithy-rs/commits) Updates `aws-smithy-types` from 1.6.2 to 1.6.3 - [Release notes](https://github.com/smithy-lang/smithy-rs/releases) - [Changelog](https://github.com/smithy-lang/smithy-rs/blob/main/CHANGELOG.md) - [Commits](https://github.com/smithy-lang/smithy-rs/commits) Updates `aws-lc-rs` from 1.16.2 to 1.18.1 - [Release notes](https://github.com/aws/aws-lc-rs/releases) - [Commits](https://github.com/aws/aws-lc-rs/compare/v1.16.2...v1.18.1) Updates `hyper` from 1.8.1 to 1.11.1 - [Release notes](https://github.com/hyperium/hyper/releases) - [Changelog](https://github.com/hyperium/hyper/blob/master/CHANGELOG.md) - [Commits](https://github.com/hyperium/hyper/compare/v1.8.1...v1.11.1) Updates `rustls-native-certs` from 0.8.3 to 0.8.4 - [Release notes](https://github.com/rustls/rustls-native-certs/releases) - [Commits](https://github.com/rustls/rustls-native-certs/compare/v/0.8.3...v/0.8.4) Updates `rustls-pki-types` from 1.14.0 to 1.15.1 - [Release notes](https://github.com/rustls/pki-types/releases) - [Commits](https://github.com/rustls/pki-types/compare/v/1.14.0...v/1.15.1) --- updated-dependencies: - dependency-name: tokio dependency-version: 1.53.1 dependency-type: direct:production update-type: version-update:semver-minor dependency-group: minor-and-patch - dependency-name: tower-http dependency-version: 0.7.1 dependency-type: direct:production update-type: version-update:semver-patch dependency-group: minor-and-patch - dependency-name: flate2 dependency-version: 1.1.10 dependency-type: direct:production update-type: version-update:semver-patch dependency-group: minor-and-patch - dependency-name: aws-smithy-runtime-api dependency-version: 1.16.0 dependency-type: direct:production update-type: version-update:semver-minor dependency-group: minor-and-patch - dependency-name: aws-smithy-types dependency-version: 1.6.3 dependency-type: direct:production update-type: version-update:semver-patch dependency-group: minor-and-patch - dependency-name: aws-lc-rs dependency-version: 1.18.1 dependency-type: direct:production update-type: version-update:semver-minor dependency-group: minor-and-patch - dependency-name: hyper dependency-version: 1.11.1 dependency-type: direct:production update-type: version-update:semver-minor dependency-group: minor-and-patch - dependency-name: rustls-native-certs dependency-version: 0.8.4 dependency-type: direct:production update-type: version-update:semver-patch dependency-group: minor-and-patch - dependency-name: rustls-pki-types dependency-version: 1.15.1 dependency-type: direct:production update-type: version-update:semver-minor dependency-group: minor-and-patch ... Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2026-09-08 00:49:34 +08:00
dependabot[bot]	38a1cc6933	chore(deps): bump the minor-and-patch group in /frontend with 5 updates (#55 ) Bumps the minor-and-patch group in /frontend with 5 updates: \| Package \| From \| To \| \| --- \| --- \| --- \| \| [@lucide/vue](https://github.com/lucide-icons/lucide/tree/HEAD/packages/vue) \| `1.37.0` \| `1.41.0` \| \| [vue-router](https://github.com/vuejs/router) \| `5.3.0` \| `5.3.1` \| \| [@antfu/eslint-config](https://github.com/antfu/eslint-config) \| `9.3.0` \| `9.5.1` \| \| [@types/node](https://github.com/DefinitelyTyped/DefinitelyTyped/tree/HEAD/types/node) \| `26.4.0` \| `26.4.1` \| \| [eslint](https://github.com/eslint/eslint) \| `10.9.1` \| `10.10.0` \| Updates `@lucide/vue` from 1.37.0 to 1.41.0 - [Release notes](https://github.com/lucide-icons/lucide/releases) - [Commits](https://github.com/lucide-icons/lucide/commits/1.41.0/packages/vue) Updates `vue-router` from 5.3.0 to 5.3.1 - [Release notes](https://github.com/vuejs/router/releases) - [Commits](https://github.com/vuejs/router/compare/v5.3.0...v5.3.1) Updates `@antfu/eslint-config` from 9.3.0 to 9.5.1 - [Release notes](https://github.com/antfu/eslint-config/releases) - [Commits](https://github.com/antfu/eslint-config/compare/v9.3.0...v9.5.1) Updates `@types/node` from 26.4.0 to 26.4.1 - [Release notes](https://github.com/DefinitelyTyped/DefinitelyTyped/releases) - [Commits](https://github.com/DefinitelyTyped/DefinitelyTyped/commits/HEAD/types/node) Updates `eslint` from 10.9.1 to 10.10.0 - [Release notes](https://github.com/eslint/eslint/releases) - [Commits](https://github.com/eslint/eslint/compare/v10.9.1...v10.10.0) --- updated-dependencies: - dependency-name: "@lucide/vue" dependency-version: 1.41.0 dependency-type: direct:production update-type: version-update:semver-minor dependency-group: minor-and-patch - dependency-name: vue-router dependency-version: 5.3.1 dependency-type: direct:production update-type: version-update:semver-patch dependency-group: minor-and-patch - dependency-name: "@antfu/eslint-config" dependency-version: 9.5.1 dependency-type: direct:development update-type: version-update:semver-minor dependency-group: minor-and-patch - dependency-name: "@types/node" dependency-version: 26.4.1 dependency-type: direct:development update-type: version-update:semver-patch dependency-group: minor-and-patch - dependency-name: eslint dependency-version: 10.10.0 dependency-type: direct:development update-type: version-update:semver-minor dependency-group: minor-and-patch ... Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2026-09-08 00:42:23 +08:00
dependabot[bot]	d38e0d4296	chore(deps): bump softprops/action-gh-release (#53 ) Bumps the minor-and-patch group with 1 update: [softprops/action-gh-release](https://github.com/softprops/action-gh-release). Updates `softprops/action-gh-release` from 3.0.2 to 3.0.3 - [Release notes](https://github.com/softprops/action-gh-release/releases) - [Changelog](https://github.com/softprops/action-gh-release/blob/master/CHANGELOG.md) - [Commits](https://github.com/softprops/action-gh-release/compare/3d0d9888cb7fd7b750713d6e236d1fcb99157228...efb35369e0ad2afab669f228072c1b0d510eae64) --- updated-dependencies: - dependency-name: softprops/action-gh-release dependency-version: 3.0.3 dependency-type: direct:production update-type: version-update:semver-patch dependency-group: minor-and-patch ... Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2026-09-08 00:42:11 +08:00
ZyyandCodex	d12d75a037	fix(openai): preserve WebSocket continuation and surface streaming errors Keep non-persistent downstream WebSocket requests on pooled upstream sockets, even when opening exceeds the fast-path budget or HTTP fallback was selected. This preserves the response state needed by subsequent previous_response_id requests. Select troubleshooting records by non-cancellation error_kind instead of the request outcome. Share the predicate between the list and count queries, and remove missing-error fallbacks so incomplete streaming errors remain visible. Validation: provider-openai tests (532 passed), PostgreSQL observability tests (19 passed), workspace rustfmt, and strict workspace Clippy with all targets and features. Co-authored-by: Codex <noreply@openai.com>	2026-09-07 12:24:28 +08:00