fix(openai): preserve requested service tier

Restore 2.x semantics: a request-specified tier drives persisted display and pricing, while the upstream response tier remains separately observable. Cover HTTP SSE and WebSocket conflicts plus the response-only fallback.

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

backend/crates/providers/openai/src/provider.rs

@@ -68,6 +68,7 @@ use crate::transport::request::{
     scope_request_to_account,
 };
 use crate::transport::session::CodexSessionIdentity;
+use crate::transport::usage::normalize_service_tier;
 use crate::transport::websocket::{CodexWebSocketExchangeError, PreviousResponseUnavailableReason};
 use crate::transport::{
     CODEX_RESPONSES_PATH, CodexAccountSelectionTelemetry, CodexBackendClient,
@@ -822,8 +823,9 @@ fn cold_response_stream(response: ColdResponse) -> EventStream {
         // 下游按字节转发上游原文，避免 serde 往返改写数值/精度（大整数→f64、logprobs 等）。
         // WS 帧由 reducer 以 encode_sse_event(&event, raw) 逐字节内嵌上游原始 JSON
         // （transport/protocol/websocket.rs），push_frames 抽出的 data 即上游原文。
-        let mut decoder =
-            CodexCanonicalDecoder::new(upstream_model.as_str()).with_raw_sse_passthrough();
+        let mut decoder = CodexCanonicalDecoder::new(upstream_model.as_str())
+            .with_requested_service_tier(request.service_tier())
+            .with_raw_sse_passthrough();
         let mut pre_commit_events = PreCommitClientEvents::new(stream_commit_policy);
         loop {
             let Some(stream_deadline) = remaining(context.deadline()) else {
@@ -895,7 +897,7 @@ fn cold_response_stream(response: ColdResponse) -> EventStream {
             };
             pre_commit_events.observe_chunk(chunk_len);
             let service_tier_changed = observation_state
-                .observe_service_tier(decoder.response_service_tier());
+                .observe_upstream_service_tier(decoder.response_service_tier());
             let terminal_failure = canonical_failure.map(|(error, semantic_output_seen)| {
                 let atomic_upstream_failure = matches!(&error, CodexCanonicalError::Upstream(_));
                 (
@@ -990,8 +992,8 @@ fn cold_response_stream(response: ColdResponse) -> EventStream {
             )
         });
         let timing_signals = decoder.take_timing_signals();
-        let service_tier_changed =
-            observation_state.observe_service_tier(decoder.response_service_tier());
+        let service_tier_changed = observation_state
+            .observe_upstream_service_tier(decoder.response_service_tier());
         let timing_changed = observation_state
             .observe_timing_signals(timing_signals, context.timing_started_at());
         let updates = take_rate_limit_updates(rate_limit_updates.as_ref()).await;
@@ -1070,7 +1072,8 @@ struct OpenAiResponseObservationState {
     requested_model: String,
     stream: bool,
     compact: bool,
-    service_tier: Option<String>,
+    requested_service_tier: Option<String>,
+    upstream_service_tier: Option<String>,
     rate_limit_headers: Vec<(String, String)>,
     account_id: String,
     attempt_index: u32,
@@ -1102,8 +1105,8 @@ impl OpenAiResponseObservationState {
             requested_model: request.model().to_owned(),
             stream: request.stream(),
             compact: semantics.compact,
-            // 请求档位保留在 requestSummary；一等观测只接受响应确认的实际档位。
-            service_tier: None,
+            requested_service_tier: normalize_service_tier(request.service_tier()),
+            upstream_service_tier: None,
             rate_limit_headers: selected_observation_headers(&response.rate_limit_headers),
             account_id: account_id.to_owned(),
             attempt_index,
@@ -1127,7 +1130,7 @@ impl OpenAiResponseObservationState {
         if let Some(metadata) = self.provider_metadata() {
             observation = observation.with_provider_metadata(metadata);
         }
-        if let Some(service_tier) = self.service_tier.as_deref() {
+        if let Some(service_tier) = self.effective_service_tier() {
             observation = observation.with_service_tier_if_valid(service_tier.to_owned());
         }
         Some(observation)
@@ -1158,17 +1161,23 @@ impl OpenAiResponseObservationState {
         changed
     }
 
-    fn observe_service_tier(&mut self, service_tier: Option<&str>) -> bool {
-        let Some(service_tier) = service_tier else {
+    fn observe_upstream_service_tier(&mut self, service_tier: Option<&str>) -> bool {
+        let Some(service_tier) = normalize_service_tier(service_tier) else {
             return false;
         };
-        if self.service_tier.as_deref() == Some(service_tier) {
+        if self.upstream_service_tier.as_deref() == Some(service_tier.as_str()) {
             return false;
         }
-        self.service_tier = Some(service_tier.to_owned());
+        self.upstream_service_tier = Some(service_tier);
         true
     }
 
+    fn effective_service_tier(&self) -> Option<&str> {
+        self.requested_service_tier
+            .as_deref()
+            .or(self.upstream_service_tier.as_deref())
+    }
+
     fn merge_rate_limit_headers(&mut self, updates: &[(String, String)]) -> bool {
         let mut changed = false;
         for (name, value) in selected_observation_headers(updates) {
@@ -1245,9 +1254,21 @@ impl OpenAiResponseObservationState {
             )),
         );
         metadata.insert("requestSummary".to_owned(), self.request_summary.clone());
-        if let Some(service_tier) = &self.service_tier {
+        if let Some(service_tier) = self.effective_service_tier() {
             metadata.insert(
                 "serviceTier".to_owned(),
+                Value::String(service_tier.to_owned()),
+            );
+        }
+        if let Some(service_tier) = &self.requested_service_tier {
+            metadata.insert(
+                "requestedServiceTier".to_owned(),
+                Value::String(service_tier.clone()),
+            );
+        }
+        if let Some(service_tier) = &self.upstream_service_tier {
+            metadata.insert(
+                "upstreamServiceTier".to_owned(),
                 Value::String(service_tier.clone()),
             );
         }

backend/crates/providers/openai/src/transport/canonical.rs

@@ -21,10 +21,9 @@ use thiserror::Error;
 use super::protocol::responses::{
     ResponseEventSignals, ResponsesSseFailure, response_event_signals,
 };
-use super::usage::openai_billing_breakdown;
+use super::usage::{normalize_service_tier, openai_billing_breakdown};
 
 const CONTENTS_PER_OUTPUT: u32 = 1_024;
-const MAX_SERVICE_TIER_BYTES: usize = 64;
 
 /// 单 attempt 的增量 Responses decoder。
 ///
@@ -42,6 +41,7 @@ pub struct CodexCanonicalDecoder {
     reasoning_output_seen: BTreeSet<u32>,
     usage_emitted: bool,
     semantic_output_seen: bool,
+    requested_service_tier: Option<String>,
     response_service_tier: Option<String>,
     timing_signals: ResponseEventSignals,
     raw_sse_passthrough: bool,
@@ -137,6 +137,7 @@ impl CodexCanonicalDecoder {
             reasoning_output_seen: BTreeSet::new(),
             usage_emitted: false,
             semantic_output_seen: false,
+            requested_service_tier: None,
             response_service_tier: None,
             timing_signals: ResponseEventSignals::default(),
             raw_sse_passthrough: false,
@@ -152,6 +153,13 @@ impl CodexCanonicalDecoder {
         self
     }
 
+    /// 请求显式指定的档位优先用于计费；响应档位只在请求未指定时兜底。
+    #[must_use]
+    pub fn with_requested_service_tier(mut self, service_tier: Option<&str>) -> Self {
+        self.requested_service_tier = normalize_service_tier(service_tier);
+        self
+    }
+
     pub fn push(&mut self, chunk: &[u8]) -> CodexCanonicalOutcome {
         self.timing_signals = ResponseEventSignals::default();
         if self.raw_sse_passthrough {
@@ -308,19 +316,14 @@ impl CodexCanonicalDecoder {
     }
 
     fn observe_response_service_tier(&mut self, value: &Value) {
-        let Some(service_tier) = response_object(value)
-            .and_then(|response| response.get("service_tier"))
-            .and_then(Value::as_str)
-            .map(str::trim)
-            .filter(|value| {
-                !value.is_empty()
-                    && value.len() <= MAX_SERVICE_TIER_BYTES
-                    && !value.chars().any(char::is_control)
-            })
-        else {
+        let Some(service_tier) = normalize_service_tier(
+            response_object(value)
+                .and_then(|response| response.get("service_tier"))
+                .and_then(Value::as_str),
+        ) else {
             return;
         };
-        self.response_service_tier = Some(service_tier.to_ascii_lowercase());
+        self.response_service_tier = Some(service_tier);
     }
 
     /// 构造下发的 wire event。
@@ -836,9 +839,9 @@ impl CodexCanonicalDecoder {
             .filter(|model| !model.is_empty())
             .unwrap_or(&self.fallback_model)
             .to_owned();
-        let service_tier = response
-            .get("service_tier")
-            .and_then(Value::as_str)
+        let service_tier = self
+            .requested_service_tier
+            .as_deref()
             .or(self.response_service_tier.as_deref());
         if let Some(breakdown) = usage
             .filter(|usage| billable_usage_is_complete(response, *usage))

backend/crates/providers/openai/src/transport/usage.rs

@@ -298,10 +298,13 @@ fn model_matches_rule(model: &str, rule: &str) -> bool {
         .is_some_and(|suffix| matches!(suffix.as_bytes().first(), Some(b'-' | b'.' | b':')))
 }
 
-fn normalize_service_tier(service_tier: Option<&str>) -> Option<String> {
+/// 规范化请求或响应携带的服务档位，供观测与计费共用。
+pub(crate) fn normalize_service_tier(service_tier: Option<&str>) -> Option<String> {
     service_tier
         .map(str::trim)
-        .filter(|value| !value.is_empty())
+        .filter(|value| {
+            !value.is_empty() && value.len() <= 64 && !value.chars().any(char::is_control)
+        })
         .map(str::to_ascii_lowercase)
 }
 

backend/crates/providers/openai/tests/provider/contract.rs

@@ -29,7 +29,7 @@ use provider_openai::credential::{
 use provider_openai::transport::CodexWebSocketPool;
 use provider_openai::transport::profile::{CodexWireProfile, CodexWireProfileState};
 use provider_openai::{CodexProvider, OFFICIAL_CODEX_BASE_URL};
-use serde_json::{Map, json};
+use serde_json::{Map, Value, json};
 use tokio::{
     io::{AsyncReadExt, AsyncWriteExt},
     net::{TcpListener, TcpStream},
@@ -827,13 +827,23 @@ async fn completed_response_persists_session_affinity_before_stream_consumer_sto
             .expect("prepare provider stream");
 
     let mut observed_service_tier = None;
+    let mut upstream_service_tier = None;
     while let Some(event) = stream.next().await {
         let event = event.expect("provider event");
-        if let Some(service_tier) = event
-            .response_observation()
-            .and_then(|observation| observation.service_tier())
-        {
-            observed_service_tier = Some(service_tier.to_owned());
+        if let Some(observation) = event.response_observation() {
+            if let Some(service_tier) = observation.service_tier() {
+                observed_service_tier = Some(service_tier.to_owned());
+            }
+            upstream_service_tier = observation
+                .provider_metadata()
+                .and_then(|metadata| serde_json::from_str::<Value>(metadata.as_json()).ok())
+                .and_then(|metadata| {
+                    metadata
+                        .get("upstreamServiceTier")
+                        .and_then(Value::as_str)
+                        .map(str::to_owned)
+                })
+                .or(upstream_service_tier);
         }
         if event
             .canonical_facts()
@@ -846,7 +856,8 @@ async fn completed_response_persists_session_affinity_before_stream_consumer_sto
     drop(stream);
 
     assert_eq!(affinity.binding_count(), 1);
-    assert_eq!(observed_service_tier.as_deref(), Some("default"));
+    assert_eq!(observed_service_tier.as_deref(), Some("priority"));
+    assert_eq!(upstream_service_tier.as_deref(), Some("default"));
 }
 
 #[tokio::test]

backend/crates/providers/openai/tests/transport/canonical.rs

@@ -203,18 +203,40 @@ fn decoder_should_emit_calculated_cost_for_complete_known_model_usage() {
 }
 
 #[test]
-fn decoder_should_keep_response_service_tier_when_terminal_event_omits_it() {
+fn decoder_should_prefer_requested_service_tier_over_response_tier_for_billing() {
     let body = concat!(
         "event: response.created\n",
-        "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_fast_cost\",\"model\":\"gpt-5.4\",\"service_tier\":\"priority\"}}\n\n",
+        "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_fast_cost\",\"model\":\"gpt-5.4\",\"service_tier\":\"default\"}}\n\n",
         "event: response.completed\n",
         "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_fast_cost\",\"model\":\"gpt-5.4\",\"status\":\"completed\",\"usage\":{\"input_tokens\":100,\"output_tokens\":10,\"input_tokens_details\":{\"cached_tokens\":25,\"cache_write_tokens\":0},\"total_tokens\":110}}}\n\n",
     );
-    let mut decoder = CodexCanonicalDecoder::new("fallback");
+    let mut decoder =
+        CodexCanonicalDecoder::new("fallback").with_requested_service_tier(Some("priority"));
     let events = decoder
         .push(body.as_bytes())
         .expect("canonical priority response");
 
+    assert_eq!(decoder.response_service_tier(), Some("default"));
+    assert!(canonical_facts(&events).into_iter().any(|event| matches!(
+        event,
+        GatewayEvent::CalculatedCost(cost)
+            if cost.total().amount().scaled() == 6_875_000
+    )));
+}
+
+#[test]
+fn decoder_should_use_response_service_tier_when_request_omits_it() {
+    let body = concat!(
+        "event: response.created\n",
+        "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_response_tier_cost\",\"model\":\"gpt-5.4\",\"service_tier\":\"priority\"}}\n\n",
+        "event: response.completed\n",
+        "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_response_tier_cost\",\"model\":\"gpt-5.4\",\"status\":\"completed\",\"usage\":{\"input_tokens\":100,\"output_tokens\":10,\"input_tokens_details\":{\"cached_tokens\":25,\"cache_write_tokens\":0},\"total_tokens\":110}}}\n\n",
+    );
+    let mut decoder = CodexCanonicalDecoder::new("fallback");
+    let events = decoder
+        .push(body.as_bytes())
+        .expect("canonical response-tier priority response");
+
     assert_eq!(decoder.response_service_tier(), Some("priority"));
     assert!(canonical_facts(&events).into_iter().any(|event| matches!(
         event,
@@ -223,6 +245,31 @@ fn decoder_should_keep_response_service_tier_when_terminal_event_omits_it() {
     )));
 }
 
+#[test]
+fn websocket_decoder_should_prefer_requested_service_tier_over_response_tier_for_billing() {
+    let created = websocket_event_to_sse_frame(
+        r#"{"type":"response.created","response":{"id":"resp_ws_fast_cost","model":"gpt-5.4","service_tier":"default"}}"#,
+    )
+    .expect("created frame");
+    let completed = websocket_event_to_sse_frame(
+        r#"{"type":"response.completed","response":{"id":"resp_ws_fast_cost","model":"gpt-5.4","status":"completed","usage":{"input_tokens":100,"output_tokens":10,"input_tokens_details":{"cached_tokens":25,"cache_write_tokens":0},"total_tokens":110}}}"#,
+    )
+    .expect("completed frame");
+    let mut decoder = CodexCanonicalDecoder::new("fallback")
+        .with_requested_service_tier(Some("priority"))
+        .with_raw_sse_passthrough();
+    let events = decoder
+        .push(format!("{created}{completed}").as_bytes())
+        .expect("canonical priority WebSocket response");
+
+    assert_eq!(decoder.response_service_tier(), Some("default"));
+    assert!(canonical_facts(&events).into_iter().any(|event| matches!(
+        event,
+        GatewayEvent::CalculatedCost(cost)
+            if cost.total().amount().scaled() == 6_875_000
+    )));
+}
+
 #[test]
 fn decoder_should_normalize_text_usage_and_completion() {
     let body = concat!(