diff --git a/backend/crates/providers/openai/src/provider.rs b/backend/crates/providers/openai/src/provider.rs index cbcd59e6d..11339b753 100644 --- a/backend/crates/providers/openai/src/provider.rs +++ b/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, + requested_service_tier: Option, + upstream_service_tier: Option, 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()), ); } diff --git a/backend/crates/providers/openai/src/transport/canonical.rs b/backend/crates/providers/openai/src/transport/canonical.rs index 72484a4c6..4ffa4eda1 100644 --- a/backend/crates/providers/openai/src/transport/canonical.rs +++ b/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, usage_emitted: bool, semantic_output_seen: bool, + requested_service_tier: Option, response_service_tier: Option, 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)) diff --git a/backend/crates/providers/openai/src/transport/usage.rs b/backend/crates/providers/openai/src/transport/usage.rs index a518faacb..ae04a892f 100644 --- a/backend/crates/providers/openai/src/transport/usage.rs +++ b/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 { +/// 规范化请求或响应携带的服务档位,供观测与计费共用。 +pub(crate) fn normalize_service_tier(service_tier: Option<&str>) -> Option { 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) } diff --git a/backend/crates/providers/openai/tests/provider/contract.rs b/backend/crates/providers/openai/tests/provider/contract.rs index 07a6f5cc5..94a27e05b 100644 --- a/backend/crates/providers/openai/tests/provider/contract.rs +++ b/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::(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] diff --git a/backend/crates/providers/openai/tests/transport/canonical.rs b/backend/crates/providers/openai/tests/transport/canonical.rs index f438b6d80..f0b39402c 100644 --- a/backend/crates/providers/openai/tests/transport/canonical.rs +++ b/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!(