diff --git a/backend/crates/gateway-admin/src/model/observability.rs b/backend/crates/gateway-admin/src/model/observability.rs index b6706a290..8d8d3e804 100644 --- a/backend/crates/gateway-admin/src/model/observability.rs +++ b/backend/crates/gateway-admin/src/model/observability.rs @@ -544,6 +544,7 @@ pub struct UsageListRecord { pub provider_account_authentication_kind: Option, pub upstream_model_id: Option, pub upstream_transport: Option, + pub upstream_response_model: Option, pub service_tier: Option, pub input_tokens: Option, pub output_tokens: Option, @@ -602,6 +603,7 @@ pub struct UsageRecord { pub upstream_transport: Option, pub http_version: Option, pub websocket_pool: Option, + pub upstream_response_model: Option, pub service_tier: Option, /// Provider 已筛选的专有观测 JSON;管理领域保持不透明。 pub provider_metadata_json: Option, diff --git a/backend/crates/gateway-admin/tests/use_case/observability.rs b/backend/crates/gateway-admin/tests/use_case/observability.rs index bfe72b710..85b9e0a37 100644 --- a/backend/crates/gateway-admin/tests/use_case/observability.rs +++ b/backend/crates/gateway-admin/tests/use_case/observability.rs @@ -860,6 +860,7 @@ fn total_record( provider_account_authentication_kind: None, upstream_model_id: Some("gpt-5.5".to_owned()), upstream_transport: None, + upstream_response_model: None, service_tier: None, input_tokens: Some(800), output_tokens: Some(200), diff --git a/backend/crates/gateway-api/src/admin/observability/presenter.rs b/backend/crates/gateway-api/src/admin/observability/presenter.rs index e2b480fb6..1f6678e35 100644 --- a/backend/crates/gateway-api/src/admin/observability/presenter.rs +++ b/backend/crates/gateway-api/src/admin/observability/presenter.rs @@ -267,6 +267,7 @@ pub(crate) fn usage_list_record_view(record: domain::UsageListRecord) -> UsageLi model, requested_model: record.requested_model_id, upstream_model: record.upstream_model_id, + upstream_response_model: record.upstream_response_model, service_tier: record.service_tier, client_transport: record.client_transport, upstream_transport: record.upstream_transport, @@ -360,6 +361,7 @@ pub(crate) fn usage_record_view(record: domain::UsageRecord) -> UsageRecordView model, requested_model: record.requested_model_id, upstream_model: record.upstream_model_id, + upstream_response_model: record.upstream_response_model, service_tier: record.service_tier, status_code, client_transport: record.client_transport, diff --git a/backend/crates/gateway-api/src/admin/observability/wire.rs b/backend/crates/gateway-api/src/admin/observability/wire.rs index f325d1aaa..3b143c3be 100644 --- a/backend/crates/gateway-api/src/admin/observability/wire.rs +++ b/backend/crates/gateway-api/src/admin/observability/wire.rs @@ -84,6 +84,7 @@ pub struct UsageListRecordView { pub model: Option, pub requested_model: Option, pub upstream_model: Option, + pub upstream_response_model: Option, pub service_tier: Option, pub client_transport: String, pub upstream_transport: Option, @@ -122,6 +123,7 @@ pub struct UsageRecordView { pub model: Option, pub requested_model: Option, pub upstream_model: Option, + pub upstream_response_model: Option, pub service_tier: Option, pub status_code: Option, pub client_transport: String, diff --git a/backend/crates/gateway-api/tests/admin/observability/response.rs b/backend/crates/gateway-api/tests/admin/observability/response.rs index eb8659228..ff62b436e 100644 --- a/backend/crates/gateway-api/tests/admin/observability/response.rs +++ b/backend/crates/gateway-api/tests/admin/observability/response.rs @@ -840,6 +840,7 @@ fn usage_record_with_account( upstream_transport: Some("http_sse".to_owned()), http_version: Some("h2".to_owned()), websocket_pool: None, + upstream_response_model: None, service_tier: None, provider_metadata_json: None, attempt_count: 1, @@ -963,6 +964,7 @@ async fn usage_route_should_expose_table_facts_without_detail_payload() { provider_account_authentication_kind: Some("oauth".to_owned()), upstream_model_id: Some("grok-4.5".to_owned()), upstream_transport: Some("http_sse".to_owned()), + upstream_response_model: Some("grok-4.6-build".to_owned()), service_tier: Some("default".to_owned()), input_tokens: Some(1), output_tokens: Some(1), @@ -1075,6 +1077,7 @@ async fn usage_route_should_expose_table_facts_without_detail_payload() { "imageOutputTokens": value["data"]["items"][0]["tokenDetails"]["imageOutputTokens"], "requestedModel": value["data"]["items"][0]["requestedModel"], "upstreamModel": value["data"]["items"][0]["upstreamModel"], + "upstreamResponseModel": value["data"]["items"][0]["upstreamResponseModel"], "reasoningEffort": value["data"]["items"][0]["reasoningEffort"], "reasoningPreset": value["data"]["items"][0]["reasoningPreset"], "subagentKind": value["data"]["items"][0]["subagentKind"], @@ -1098,6 +1101,7 @@ async fn usage_route_should_expose_table_facts_without_detail_payload() { "imageOutputTokens": 9, "requestedModel": "grok-4.5", "upstreamModel": "grok-4.5", + "upstreamResponseModel": "grok-4.6-build", "reasoningEffort": "max", "reasoningPreset": "ultra", "subagentKind": "worker", diff --git a/backend/crates/gateway-api/tests/key_usage/fixtures.rs b/backend/crates/gateway-api/tests/key_usage/fixtures.rs index 187fcb30f..f821c2699 100644 --- a/backend/crates/gateway-api/tests/key_usage/fixtures.rs +++ b/backend/crates/gateway-api/tests/key_usage/fixtures.rs @@ -100,6 +100,7 @@ fn usage_record() -> UsageListRecord { provider_account_authentication_kind: Some("private-sentinel".to_owned()), upstream_model_id: Some("private-sentinel".to_owned()), upstream_transport: Some("websocket".to_owned()), + upstream_response_model: Some("private-sentinel".to_owned()), service_tier: Some("private-sentinel".to_owned()), input_tokens: Some(1000), output_tokens: Some(100), diff --git a/backend/crates/gateway-core/src/engine/coordinator.rs b/backend/crates/gateway-core/src/engine/coordinator.rs index eaed69161..5ddbbee49 100644 --- a/backend/crates/gateway-core/src/engine/coordinator.rs +++ b/backend/crates/gateway-core/src/engine/coordinator.rs @@ -1498,6 +1498,12 @@ where http_version, websocket_pool, service_tier, + upstream_response_model: self + .current + .as_ref() + .and_then(|current| current.response_observation.as_ref()) + .and_then(ProviderResponseObservation::upstream_response_model) + .map(str::to_owned), provider_metadata_json, diagnostic_trace_json: self.trace.snapshot().map(|value| value.to_string()), error: None, @@ -1659,6 +1665,12 @@ where http_version, websocket_pool, service_tier, + upstream_response_model: self + .current + .as_ref() + .and_then(|current| current.response_observation.as_ref()) + .and_then(ProviderResponseObservation::upstream_response_model) + .map(str::to_owned), provider_metadata_json, diagnostic_trace_json: self.trace.snapshot().map(|value| value.to_string()), error: Some(finalization.error), diff --git a/backend/crates/gateway-core/src/engine/mod.rs b/backend/crates/gateway-core/src/engine/mod.rs index e8de22a1e..89a2c2ba4 100644 --- a/backend/crates/gateway-core/src/engine/mod.rs +++ b/backend/crates/gateway-core/src/engine/mod.rs @@ -659,6 +659,7 @@ pub struct ModelRequestFinalization { pub http_version: Option, pub websocket_pool: Option, pub service_tier: Option, + pub upstream_response_model: Option, /// Provider 已筛选的专有观测 JSON;Core 不解释字段。 pub provider_metadata_json: Option, /// 请求全程的有界诊断快照,跨 Provider 与重试保留。 diff --git a/backend/crates/gateway-core/src/event.rs b/backend/crates/gateway-core/src/event.rs index aa6cb4159..2b98272e9 100644 --- a/backend/crates/gateway-core/src/event.rs +++ b/backend/crates/gateway-core/src/event.rs @@ -530,6 +530,7 @@ pub struct ProviderResponseObservation { status_code: Option, request_id: Option, service_tier: Option, + upstream_response_model: Option, timings: ProviderResponseTimings, client_headers: Vec, provider_metadata: Option, @@ -545,6 +546,7 @@ impl ProviderResponseObservation { status_code: None, request_id: None, service_tier: None, + upstream_response_model: None, timings: ProviderResponseTimings::default(), client_headers: Vec::new(), provider_metadata: None, @@ -600,6 +602,21 @@ impl ProviderResponseObservation { self } + /// 记录 Provider 从上游响应明确取得的模型,不改写发送模型或计费合同。 + #[must_use] + pub fn with_upstream_response_model_if_valid(mut self, model: &str) -> Self { + let model = model.trim(); + if validate_text(model, 256, false, None).is_ok() { + self.upstream_response_model = Some(model.to_owned()); + } + self + } + + #[must_use] + pub fn upstream_response_model(&self) -> Option<&str> { + self.upstream_response_model.as_deref() + } + #[must_use] pub const fn with_timings(mut self, timings: ProviderResponseTimings) -> Self { self.timings = timings; diff --git a/backend/crates/gateway-core/tests/engine/coordinator.rs b/backend/crates/gateway-core/tests/engine/coordinator.rs index 4dc565edb..2e789d0d5 100644 --- a/backend/crates/gateway-core/tests/engine/coordinator.rs +++ b/backend/crates/gateway-core/tests/engine/coordinator.rs @@ -73,6 +73,7 @@ struct FinalState { http_version: Option, websocket_pool: Option, service_tier: Option, + upstream_response_model: Option, upstream_request_id: Option, upstream_status_code: Option, transport_decision_wait_ms: Option, @@ -239,6 +240,7 @@ impl ExecutionStore for FakeStore { http_version: finalization.http_version, websocket_pool: finalization.websocket_pool, service_tier: finalization.service_tier, + upstream_response_model: finalization.upstream_response_model, upstream_request_id: finalization.upstream_request_id, upstream_status_code: finalization.upstream_status_code, transport_decision_wait_ms: finalization.timings.transport_decision_wait_ms, @@ -1093,6 +1095,7 @@ fn response_observation_is_persisted_but_never_delivered() { .with_websocket_pool(WebSocketPoolKind::New) .with_status_code(200) .with_request_id(OpaqueUpstreamValue::new("upstream-observed")) + .with_upstream_response_model_if_valid("gpt-returned") .try_with_service_tier("priority") .expect("service tier") .with_timings(ProviderResponseTimings { @@ -1132,6 +1135,10 @@ fn response_observation_is_persisted_but_never_delivered() { assert_eq!(finalization.http_version.as_deref(), Some("HTTP/2")); assert_eq!(finalization.websocket_pool.as_deref(), Some("new")); assert_eq!(finalization.service_tier.as_deref(), Some("priority")); + assert_eq!( + finalization.upstream_response_model.as_deref(), + Some("gpt-returned") + ); assert_eq!( finalization.upstream_request_id.as_deref(), Some("upstream-observed") @@ -1350,6 +1357,7 @@ fn discarded_attempt_observation_does_not_leak_into_retry_result() { .with_http_version(UpstreamHttpVersion::Http11) .with_status_code(503) .with_request_id(OpaqueUpstreamValue::new("discarded-request")) + .with_upstream_response_model_if_valid("discarded-model") .with_timings(ProviderResponseTimings { transport_decision_wait_ms: Some(7), connect_ms: Some(11), @@ -1409,6 +1417,7 @@ fn discarded_attempt_observation_does_not_leak_into_retry_result() { let state = store.state.lock().expect("store lock"); let finalization = &state.finalizations[0]; + assert_eq!(finalization.upstream_response_model, None); let trace: Value = serde_json::from_str(finalization.diagnostic_trace_json.as_deref().unwrap()).unwrap(); let events = trace["events"].as_array().unwrap(); diff --git a/backend/crates/gateway-protocol/src/openai/events.rs b/backend/crates/gateway-protocol/src/openai/events.rs index 1fd6dae46..7384dfed2 100644 --- a/backend/crates/gateway-protocol/src/openai/events.rs +++ b/backend/crates/gateway-protocol/src/openai/events.rs @@ -5,6 +5,56 @@ use serde_json::Value; use super::sse::{SseError, parse_sse_events}; +/// 单次 Responses 流明确声明的模型;终态优先,缺失时不使用请求模型补齐。 +#[derive(Debug, Default)] +pub struct ResponseModelObservation { + model: Option, + terminal: bool, +} + +impl ResponseModelObservation { + /// 从原始事件读取响应模型,避免工具转换或 canonical 兜底改写观测事实。 + pub fn observe(&mut self, event_type: Option<&str>, value: &Value) { + let terminal = matches!( + event_type, + Some( + "response.completed" + | "response.done" + | "response.incomplete" + | "response.failed" + | "response.cancelled" + | "response.canceled" + ) + ); + if self.terminal || (self.model.is_some() && !terminal) { + return; + } + let Some(model) = value + .pointer("/response/model") + .and_then(Value::as_str) + .and_then(observed_model_name) + else { + return; + }; + self.model = Some(model.to_owned()); + self.terminal = terminal; + } + + /// 返回原始响应声明,未声明时保持未知。 + #[must_use] + pub fn model(&self) -> Option<&str> { + self.model.as_deref() + } +} + +/// 规范化可展示的上游模型名,非法值不影响响应交付。 +#[must_use] +pub fn observed_model_name(value: &str) -> Option<&str> { + let value = value.trim(); + (!value.is_empty() && value.len() <= 256 && !value.chars().any(char::is_control)) + .then_some(value) +} + /// 从 Codex/OpenAI usage 结构中提取出的标准化 token 用量。 #[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] diff --git a/backend/crates/gateway-protocol/tests/openai/events.rs b/backend/crates/gateway-protocol/tests/openai/events.rs index 306cc9fbd..518f53fbe 100644 --- a/backend/crates/gateway-protocol/tests/openai/events.rs +++ b/backend/crates/gateway-protocol/tests/openai/events.rs @@ -5,6 +5,50 @@ use gateway_protocol::openai::events::{ }; use serde_json::json; +#[test] +fn response_model_observation_keeps_first_declaration_until_terminal() { + use gateway_protocol::openai::events::ResponseModelObservation; + + let mut observed = ResponseModelObservation::default(); + observed.observe( + Some("response.created"), + &json!({"response": {"model": " first "}}), + ); + observed.observe( + Some("response.in_progress"), + &json!({"response": {"model": "intermediate"}}), + ); + assert_eq!(observed.model(), Some("first")); + observed.observe( + Some("response.completed"), + &json!({"response": {"model": "final"}}), + ); + observed.observe( + Some("response.in_progress"), + &json!({"response": {"model": "late"}}), + ); + assert_eq!(observed.model(), Some("final")); + assert_eq!(ResponseModelObservation::default().model(), None); +} + +#[test] +fn response_model_observation_ignores_missing_and_invalid_values() { + use gateway_protocol::openai::events::ResponseModelObservation; + + for response in [ + json!({}), + json!({"model": null}), + json!({"model": 42}), + json!({"model": " "}), + json!({"model": "a\nb"}), + json!({"model": "a".repeat(257)}), + ] { + let mut observed = ResponseModelObservation::default(); + observed.observe(Some("response.completed"), &json!({"response": response})); + assert_eq!(observed.model(), None); + } +} + #[test] fn billable_usage_should_validate_cache_writes_totals_and_overflow() { for (raw, expected) in [ diff --git a/backend/crates/gateway-store/src/postgres/execution.rs b/backend/crates/gateway-store/src/postgres/execution.rs index e308cfef0..8c0cf33be 100644 --- a/backend/crates/gateway-store/src/postgres/execution.rs +++ b/backend/crates/gateway-store/src/postgres/execution.rs @@ -282,6 +282,7 @@ pub struct ModelRequestFinalization { pub http_version: Option, pub websocket_pool: Option, pub service_tier: Option, + pub upstream_response_model: Option, pub provider_metadata_json: Option, pub diagnostic_trace_json: Option, pub error_kind: Option, @@ -347,6 +348,12 @@ impl ModelRequestFinalization { return Err(invalid("service tier is invalid")); } } + if let Some(model) = self.upstream_response_model.as_deref() { + require_nonempty(ENTITY, "upstream_response_model", model)?; + if model.len() > 256 || model.chars().any(char::is_control) { + return Err(invalid("upstream response model is invalid")); + } + } if self .provider_metadata_json .as_ref() @@ -735,7 +742,8 @@ impl ModelRequestRepository for PgExecutionStore { upstream_connection_id = $44, upstream_connection_exit_reason = $45, upstream_connection_age_ms = $46, - upstream_connection_idle_ms = $47, diagnostic_trace_json = $48 + upstream_connection_idle_ms = $47, diagnostic_trace_json = $48, + upstream_response_model = $49 where id = $1 and outcome = 'running' returning id, client_api_key_ref, continuation_affinity_hash, continuation_requested, provider_kind, upstream_transport, @@ -927,6 +935,7 @@ impl ModelRequestRepository for PgExecutionStore { "upstream_connection_idle_ms", )?) .bind(finalization.diagnostic_trace_json.map(sqlx::types::Json)) + .bind(finalization.upstream_response_model) .fetch_one(&self.pool) .await .map_err(|_| postgres_unavailable("finalize model request"))?; @@ -1223,6 +1232,7 @@ impl ExecutionStore for PgExecutionStore { http_version: finalization.http_version, websocket_pool: finalization.websocket_pool, service_tier: finalization.service_tier, + upstream_response_model: finalization.upstream_response_model, provider_metadata_json, diagnostic_trace_json: finalization .diagnostic_trace_json diff --git a/backend/crates/gateway-store/src/postgres/execution_buffer.rs b/backend/crates/gateway-store/src/postgres/execution_buffer.rs index 9c2874b0f..bda0856e2 100644 --- a/backend/crates/gateway-store/src/postgres/execution_buffer.rs +++ b/backend/crates/gateway-store/src/postgres/execution_buffer.rs @@ -582,6 +582,7 @@ impl ExecutionObservationWrite { finalization.http_version.as_deref(), finalization.websocket_pool.as_deref(), finalization.service_tier.as_deref(), + finalization.upstream_response_model.as_deref(), finalization.provider_metadata_json.as_deref(), finalization.diagnostic_trace_json.as_deref(), finalization.provider_error_code.as_deref(), diff --git a/backend/crates/gateway-store/src/postgres/observability/mapping.rs b/backend/crates/gateway-store/src/postgres/observability/mapping.rs index 24ce1f191..b37e93ff3 100644 --- a/backend/crates/gateway-store/src/postgres/observability/mapping.rs +++ b/backend/crates/gateway-store/src/postgres/observability/mapping.rs @@ -439,6 +439,7 @@ pub(crate) fn admin_usage_list_record( provider_account_authentication_kind: record.provider_account_authentication_kind, upstream_model_id: record.upstream_model_id, upstream_transport: record.upstream_transport, + upstream_response_model: record.upstream_response_model, service_tier: record.service_tier, input_tokens: record.input_tokens, output_tokens: record.output_tokens, @@ -515,6 +516,7 @@ pub(crate) fn admin_usage_record( upstream_transport: record.upstream_transport, http_version: record.http_version, websocket_pool: record.websocket_pool, + upstream_response_model: record.upstream_response_model, service_tier: record.service_tier, provider_metadata_json: record.provider_metadata_json, attempt_count: record.attempt_count, @@ -764,6 +766,7 @@ pub(crate) fn usage_list_record_from_row( provider_account_authentication_kind: get(row, "provider_account_authentication_kind")?, upstream_model_id: get(row, "upstream_model_id")?, upstream_transport: get(row, "upstream_transport")?, + upstream_response_model: get(row, "upstream_response_model")?, service_tier: get(row, "service_tier")?, input_tokens: optional_unsigned(row, "input_tokens")?, output_tokens: optional_unsigned(row, "output_tokens")?, @@ -825,6 +828,7 @@ pub(crate) fn usage_record_from_row(row: &sqlx::postgres::PgRow) -> StoreResult< upstream_transport: get(row, "upstream_transport")?, http_version: get(row, "http_version")?, websocket_pool: get(row, "websocket_pool")?, + upstream_response_model: get(row, "upstream_response_model")?, service_tier: get(row, "service_tier")?, provider_metadata_json: get::>(row, "provider_observation_json")? .map(|value| serde_json::to_string(&value)) diff --git a/backend/crates/gateway-store/src/postgres/observability/model.rs b/backend/crates/gateway-store/src/postgres/observability/model.rs index 81206e403..5cdb2e4f8 100644 --- a/backend/crates/gateway-store/src/postgres/observability/model.rs +++ b/backend/crates/gateway-store/src/postgres/observability/model.rs @@ -490,6 +490,7 @@ pub struct UsageListRecord { pub provider_account_authentication_kind: Option, pub upstream_model_id: Option, pub upstream_transport: Option, + pub upstream_response_model: Option, pub service_tier: Option, pub input_tokens: Option, pub output_tokens: Option, @@ -546,6 +547,7 @@ pub struct UsageRecord { pub upstream_transport: Option, pub http_version: Option, pub websocket_pool: Option, + pub upstream_response_model: Option, pub service_tier: Option, pub provider_metadata_json: Option, pub attempt_count: u32, diff --git a/backend/crates/gateway-store/src/postgres/observability/queries/usage.rs b/backend/crates/gateway-store/src/postgres/observability/queries/usage.rs index 2367f783a..0101a137a 100644 --- a/backend/crates/gateway-store/src/postgres/observability/queries/usage.rs +++ b/backend/crates/gateway-store/src/postgres/observability/queries/usage.rs @@ -130,7 +130,7 @@ pub(crate) const USAGE_LIST_RECORD_SELECT: &str = mr.provider_account_email_snapshot as provider_account_email, mr.provider_account_authentication_kind_snapshot as provider_account_authentication_kind, - mr.upstream_model_id, mr.upstream_transport, mr.service_tier, + mr.upstream_model_id, mr.upstream_transport, mr.upstream_response_model, mr.service_tier, mr.input_tokens, mr.output_tokens, mr.cached_tokens, mr.cache_write_tokens, mr.reasoning_tokens, mr.image_input_tokens, mr.image_output_tokens, mr.total_tokens, mr.cost_source, mr.cost_amount::text, mr.cost_currency, @@ -154,7 +154,7 @@ pub(crate) const USAGE_RECORD_DETAIL_SELECT: &str = mr.provider_account_authentication_kind_snapshot as provider_account_authentication_kind, mr.upstream_model_id, mr.upstream_transport, mr.http_version, mr.websocket_pool, - mr.service_tier, mr.provider_observation_json, mr.diagnostic_trace_json, + mr.service_tier, mr.upstream_response_model, mr.provider_observation_json, mr.diagnostic_trace_json, mr.attempt_count, mr.upstream_send_state, mr.downstream_committed_at, mr.outcome, mr.client_status_code, mr.upstream_status_code, mr.client_response_id, mr.upstream_request_id, mr.upstream_response_id, diff --git a/backend/crates/gateway-store/tests/postgres/execution.rs b/backend/crates/gateway-store/tests/postgres/execution.rs index 2d31ba8b4..37ce5c421 100644 --- a/backend/crates/gateway-store/tests/postgres/execution.rs +++ b/backend/crates/gateway-store/tests/postgres/execution.rs @@ -356,8 +356,10 @@ async fn core_adapter_should_persist_calculated_cost_exactly() { let repository = PgExecutionStore::new(database.pool.clone()); let mut finalization = successful_core_finalization("req_calculated_cost"); + finalization.downstream_committed_at = Some(std::time::SystemTime::now()); finalization.websocket_pool = Some("reuse".to_owned()); finalization.service_tier = Some("priority".to_owned()); + finalization.upstream_response_model = Some("grok-4.6-build".to_owned()); finalization.cost = CalculatedCost::from_usd_ticks(12_345) .expect("calculated cost") .into_estimate(); @@ -393,6 +395,37 @@ async fn core_adapter_should_persist_calculated_cost_exactly() { Some("priority".to_owned()), ) ); + let observation = observability_repository(&database.pool); + let detail = observation + .usage_record_detail("req_calculated_cost") + .await + .expect("request detail"); + assert_eq!( + detail.request.upstream_model_id.as_deref(), + Some("grok-4.5") + ); + assert_eq!( + detail.request.upstream_response_model.as_deref(), + Some("grok-4.6-build") + ); + let page = observation + .list_usage_records(UsageRecordQuery { + range: ObservabilityRange::new( + Utc::now() - Duration::hours(1), + Utc::now() + Duration::hours(1), + ) + .unwrap(), + filter: UsageRecordFilter::default(), + current_page: 1, + page_size: ObservabilityPageSize::new(10).unwrap(), + }) + .await + .expect("request list"); + assert_eq!(page.items.len(), 1); + assert_eq!( + page.items[0].upstream_response_model.as_deref(), + Some("grok-4.6-build") + ); database.close().await; } @@ -494,6 +527,7 @@ fn successful_core_finalization(id: &str) -> CoreModelRequestFinalization { upstream_transport: Some("websocket".to_owned()), http_version: Some("HTTP/2".to_owned()), websocket_pool: None, + upstream_response_model: None, service_tier: None, provider_metadata_json: None, error: None, @@ -1230,6 +1264,7 @@ pub(super) fn early_failure(request: &CoreNewModelRequest) -> CoreModelRequestFi upstream_transport: None, http_version: None, websocket_pool: None, + upstream_response_model: None, service_tier: None, provider_metadata_json: None, diagnostic_trace_json: trace.snapshot().map(|value| value.to_string()), diff --git a/backend/crates/providers/openai/src/provider/execution.rs b/backend/crates/providers/openai/src/provider/execution.rs index 9c7fd73ef..977fcfc07 100644 --- a/backend/crates/providers/openai/src/provider/execution.rs +++ b/backend/crates/providers/openai/src/provider/execution.rs @@ -730,12 +730,13 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream { let mut passive_quota_observation = OpenAiPassiveQuotaObservation::new(response.rate_limit_headers); let rate_limit_updates = response.rate_limit_updates; - let turn_state_updates = response.turn_state_update; + let response_metadata_updates = response.response_metadata_updates; // OpenAI 线路为透明代理:HTTP SSE 与 WebSocket 两条上游均启用 raw 透传, // 下游按字节转发上游原文,避免 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_reported_model(response.response_metadata.effective_model.as_deref()) .with_requested_service_tier(request.service_tier()) .with_request_tool_pricing(upstream_model.as_str(), request.tools()) .with_raw_sse_passthrough(); @@ -798,13 +799,14 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream { let update_headers = rate_limit_update_headers(&updates); observation_state.merge_rate_limit_headers(&update_headers) }; - let turn_state_merge = merge_turn_state_update( - turn_state_updates.as_ref(), + let metadata_merge = merge_response_metadata_updates( + response_metadata_updates.as_ref(), &mut session_capture, &mut observation_state, + &mut decoder, ) .await; - let observation_event = if rate_limits_changed || turn_state_merge.is_some() { + let observation_event = if rate_limits_changed || metadata_merge.is_some() { observation_state.observation(None).map(ProviderEvent::observation) } else { None @@ -857,13 +859,6 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream { passive_quota_observation.observe(&updates); observation_state.merge_rate_limit_headers(&rate_limit_update_headers(&updates)) }; - let turn_state_merge = merge_turn_state_update( - turn_state_updates.as_ref(), - &mut session_capture, - &mut observation_state, - ) - .await; - let turn_state_changed = turn_state_merge.unwrap_or(false); let first_event_changed = observation_state.observe_stream_chunk(&chunk, output_started_at); let chunk_len = chunk.len(); @@ -874,7 +869,17 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream { (events, Some((error, semantic_output_seen))) } }; + let metadata_merge = merge_response_metadata_updates( + response_metadata_updates.as_ref(), + &mut session_capture, + &mut observation_state, + &mut decoder, + ) + .await; + let metadata_changed = metadata_merge.unwrap_or(false); pre_commit_events.observe_chunk(chunk_len); + let response_model_changed = observation_state + .observe_upstream_response_model(decoder.response_model()); let service_tier_changed = observation_state .observe_upstream_service_tier(decoder.response_service_tier()); let terminal_failure = canonical_failure.map(|(error, semantic_output_seen)| { @@ -945,9 +950,10 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream { .await; } if (rate_limits_changed + || response_model_changed || service_tier_changed || timing_changed - || turn_state_changed + || metadata_changed || terminal_changed || (response_transport == CodexBackendTransport::WebSocket && terminal_failure.is_some())) && let Some(observation) = observation_state.observation( @@ -1011,6 +1017,8 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream { ) }); let timing_signals = decoder.take_timing_signals(); + let response_model_changed = observation_state + .observe_upstream_response_model(decoder.response_model()); let service_tier_changed = observation_state .observe_upstream_service_tier(decoder.response_service_tier()); let timing_changed = observation_state @@ -1034,10 +1042,11 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream { apply_failure(&failure_context, &active_account, failure) .await; } - let turn_state_changed = merge_turn_state_update( - turn_state_updates.as_ref(), + let metadata_changed = merge_response_metadata_updates( + response_metadata_updates.as_ref(), &mut session_capture, &mut observation_state, + &mut decoder, ) .await .unwrap_or(false); @@ -1067,10 +1076,11 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream { .observe_cyber_policy_success(cyber_policy_scope.as_ref()) .await; } - if (service_tier_changed + if (response_model_changed + || service_tier_changed || timing_changed || rate_limits_changed - || turn_state_changed + || metadata_changed || terminal_changed || (response_transport == CodexBackendTransport::WebSocket && terminal_failure.is_some())) && let Some(observation) = observation_state.observation( @@ -1101,15 +1111,30 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream { }) } -async fn merge_turn_state_update( - updates: Option<&CodexTurnStateUpdate>, +async fn merge_response_metadata_updates( + updates: Option<&CodexResponseMetadataUpdates>, session_capture: &mut Option, observation_state: &mut OpenAiResponseObservationState, + decoder: &mut CodexCanonicalDecoder, ) -> Option { let updates = updates?; - let turn_state = updates.lock().await.take()?; - if let Some(capture) = session_capture.as_mut() { - capture.turn_state = Some(turn_state.clone()); + let mut pending = updates.lock().await; + let turn_state = pending.turn_state.take(); + let reported_model = pending.reported_model.clone(); + drop(pending); + if turn_state.is_none() && reported_model.is_none() { + return None; } - Some(observation_state.merge_client_header("x-codex-turn-state", &turn_state)) + let mut changed = false; + if let Some(turn_state) = turn_state { + if let Some(capture) = session_capture.as_mut() { + capture.turn_state = Some(turn_state.clone()); + } + changed |= observation_state.merge_client_header("x-codex-turn-state", &turn_state); + } + if let Some(model) = reported_model { + decoder.observe_reported_model(&model); + changed |= observation_state.observe_upstream_response_model(decoder.response_model()); + } + Some(changed) } diff --git a/backend/crates/providers/openai/src/provider/mod.rs b/backend/crates/providers/openai/src/provider/mod.rs index c9fb2a0b1..0ddf6490d 100644 --- a/backend/crates/providers/openai/src/provider/mod.rs +++ b/backend/crates/providers/openai/src/provider/mod.rs @@ -91,8 +91,8 @@ use crate::transport::{ CODEX_RESPONSES_PATH, CodexAccountSelectionTelemetry, CodexBackendClient, CodexBackendJsonResponse, CodexBackendStreamingResponse, CodexBackendTransport, CodexClientError, CodexRateLimitUpdates, CodexRequestContext, CodexResponseMetadata, - CodexTransportMetrics, CodexTurnStateUpdate, CodexUpstreamDiagnostics, CodexWebSocketPool, - endpoint_url, + CodexResponseMetadataUpdates, CodexTransportMetrics, CodexUpstreamDiagnostics, + CodexWebSocketPool, endpoint_url, }; mod execution; diff --git a/backend/crates/providers/openai/src/provider/observation.rs b/backend/crates/providers/openai/src/provider/observation.rs index fe5f384d7..0c515c360 100644 --- a/backend/crates/providers/openai/src/provider/observation.rs +++ b/backend/crates/providers/openai/src/provider/observation.rs @@ -33,6 +33,7 @@ pub(super) struct OpenAiResponseObservationState { stream: bool, requested_service_tier: Option, upstream_service_tier: Option, + upstream_response_model: Option, rate_limit_headers: Vec<(String, String)>, timings: ProviderResponseTimings, terminal: Option, @@ -79,6 +80,7 @@ impl OpenAiResponseObservationState { stream: request.stream(), requested_service_tier: normalize_service_tier(request.service_tier()), upstream_service_tier: None, + upstream_response_model: response.response_metadata.effective_model.clone(), rate_limit_headers: selected_observation_headers(&response.rate_limit_headers), timings: openai_response_timings( &response.transport_metrics, @@ -121,9 +123,23 @@ impl OpenAiResponseObservationState { if let Some(service_tier) = &self.requested_service_tier { observation = observation.with_service_tier_if_valid(service_tier.clone()); } + if let Some(model) = &self.upstream_response_model { + observation = observation.with_upstream_response_model_if_valid(model); + } Some(observation) } + pub(super) fn observe_upstream_response_model(&mut self, model: Option<&str>) -> bool { + let Some(model) = model else { + return false; + }; + if self.upstream_response_model.as_deref() == Some(model) { + return false; + } + self.upstream_response_model = Some(model.to_owned()); + true + } + pub(super) fn observe_stream_chunk(&mut self, chunk: &[u8], started_at: Instant) -> bool { if chunk.is_empty() { return false; diff --git a/backend/crates/providers/openai/src/transport/canonical.rs b/backend/crates/providers/openai/src/transport/canonical.rs index 77be5b8f1..d86b67bf9 100644 --- a/backend/crates/providers/openai/src/transport/canonical.rs +++ b/backend/crates/providers/openai/src/transport/canonical.rs @@ -13,7 +13,10 @@ use gateway_core::event::{ }; use gateway_core::metering::Usage; use gateway_core::upstream::UpstreamSendState; -use gateway_protocol::openai::events::{TokenUsage, billable_usage_is_complete, extract_usage}; +use gateway_protocol::openai::events::{ + ResponseModelObservation, TokenUsage, billable_usage_is_complete, extract_usage, + observed_model_name, +}; use gateway_protocol::openai::sse::{SseEvent, SseEventDecoder, SseFrame, sse_frame_is_done}; use serde_json::Value; use thiserror::Error; @@ -34,7 +37,7 @@ const CONTENTS_PER_OUTPUT: u32 = 1_024; /// 因而未知或形状变化的 JSON event 只能放弃 canonical 投影,不能截断 wire 流。 pub struct CodexCanonicalDecoder { decoder: SseEventDecoder, - fallback_model: String, + upstream_model: String, response_id: Option, started: bool, completed: bool, @@ -46,6 +49,8 @@ pub struct CodexCanonicalDecoder { semantic_output_seen: bool, requested_service_tier: Option, response_service_tier: Option, + response_model: ResponseModelObservation, + reported_model: Option, web_search_pricing: Option, timing_signals: ResponseEventSignals, raw_sse_passthrough: bool, @@ -128,10 +133,11 @@ impl CodexCanonicalFailure { } impl CodexCanonicalDecoder { - pub fn new(fallback_model: impl Into) -> Self { + /// 使用路由后最终发往上游的请求模型计价,并在响应缺少模型时用于 canonical 兜底。 + pub fn new(upstream_model: impl Into) -> Self { Self { decoder: SseEventDecoder::default(), - fallback_model: fallback_model.into(), + upstream_model: upstream_model.into(), response_id: None, started: false, completed: false, @@ -143,6 +149,8 @@ impl CodexCanonicalDecoder { semantic_output_seen: false, requested_service_tier: None, response_service_tier: None, + response_model: ResponseModelObservation::default(), + reported_model: None, web_search_pricing: None, timing_signals: ResponseEventSignals::default(), raw_sse_passthrough: false, @@ -223,6 +231,28 @@ impl CodexCanonicalDecoder { self.response_service_tier.as_deref() } + /// 真实 HTTP 响应头提供初始报告;流内请求级报告可覆盖它。 + #[must_use] + pub fn with_reported_model(mut self, model: Option<&str>) -> Self { + self.reported_model = model.and_then(observed_model_name).map(str::to_owned); + self + } + + /// 接收 transport 已解析的内部 metadata 报告,内部帧无需交付客户端。 + pub(crate) fn observe_reported_model(&mut self, model: &str) { + if let Some(model) = observed_model_name(model) { + self.reported_model = Some(model.to_owned()); + } + } + + /// 官方服务端报告优先;缺少报告时仅使用正文明确声明,不使用请求兜底值。 + #[must_use] + pub fn response_model(&self) -> Option<&str> { + self.reported_model + .as_deref() + .or_else(|| self.response_model.model()) + } + fn decode(&mut self, events: Vec) -> CodexCanonicalOutcome { let mut output = Vec::new(); for event in events { @@ -288,6 +318,10 @@ impl CodexCanonicalDecoder { return Ok(()); } self.observe_response_service_tier(&value); + self.response_model.observe(event_type, &value); + if let Some(model) = super::response_meta::reported_model_from_event(&value) { + self.reported_model = Some(model.to_owned()); + } let signals = response_event_signals(event_type, &value); self.merge_timing_signals(signals); if matches!(event_type, Some("response.failed" | "error")) { @@ -452,7 +486,7 @@ impl CodexCanonicalDecoder { .get("model") .and_then(Value::as_str) .filter(|model| !model.is_empty()) - .unwrap_or(&self.fallback_model) + .unwrap_or(&self.upstream_model) .to_owned(); self.response_id = Some(response_id.clone()); self.started = true; @@ -859,9 +893,9 @@ impl CodexCanonicalDecoder { .get("model") .and_then(Value::as_str) .filter(|model| !model.is_empty()) - .unwrap_or(&self.fallback_model) + .unwrap_or(&self.upstream_model) .to_owned(); - // 与用量统计统一按最终发送档位估算,响应回显不改变本地计价口径。 + // 按最终发送的模型与档位估算,响应回显仅作观测,不改变本地计价口径。 let service_tier = self.requested_service_tier.as_deref(); let tool_calls = billable_tool_calls(response); if let Some(breakdown) = usage @@ -869,7 +903,7 @@ impl CodexCanonicalDecoder { .and_then(|usage| { let (web_search_calls, file_search_calls) = tool_calls?; openai_billing_breakdown( - &model, + &self.upstream_model, OpenAiBillingUsage::from(usage) .with_web_search_calls(web_search_calls, self.web_search_pricing) .with_file_search_calls(file_search_calls), diff --git a/backend/crates/providers/openai/src/transport/client.rs b/backend/crates/providers/openai/src/transport/client.rs index 158d47519..f46d5f7ab 100644 --- a/backend/crates/providers/openai/src/transport/client.rs +++ b/backend/crates/providers/openai/src/transport/client.rs @@ -33,7 +33,7 @@ use super::response_meta::CodexResponseMetadata; use super::tls::{CustomCaError, build_reqwest_client_with_custom_ca, custom_ca_env_cache_key}; use super::websocket::{ CodexWebSocketExchangeError, CodexWebSocketPool, CodexWebSocketPoolKey, - CodexWebSocketRateLimitUpdates, CodexWebSocketRequest, CodexWebSocketTurnStateUpdate, + CodexWebSocketRateLimitUpdates, CodexWebSocketRequest, CodexWebSocketResponseMetadataUpdates, PreparedWebSocket, WebSocketOriginBreaker, WebSocketPoolDecision, }; @@ -589,8 +589,8 @@ pub struct CodexTransportMetrics { /// 响应头之后在 live 流中采集的结构化限流更新。 pub type CodexRateLimitUpdates = CodexWebSocketRateLimitUpdates; -/// 响应头之后在 live 流中采集的 turn state 更新。 -pub type CodexTurnStateUpdate = CodexWebSocketTurnStateUpdate; +/// 响应头之后在 live 流中采集的请求级 metadata 更新。 +pub type CodexResponseMetadataUpdates = CodexWebSocketResponseMetadataUpdates; /// Codex Responses 上游 live SSE 响应。 pub struct CodexBackendStreamingResponse { @@ -608,8 +608,8 @@ pub struct CodexBackendStreamingResponse { pub rate_limit_headers: Vec<(String, String)>, /// live stream 期间捕获的结构化限流更新。 pub rate_limit_updates: Option, - /// live stream 期间捕获的 turn-state 更新。 - pub turn_state_update: Option, + /// live stream 期间捕获的请求级 metadata 更新。 + pub response_metadata_updates: Option, /// WebSocket 连接池决策。 pub websocket_pool_decision: Option, /// 上游诊断元数据。 diff --git a/backend/crates/providers/openai/src/transport/client_sse.rs b/backend/crates/providers/openai/src/transport/client_sse.rs index 8471ebf2b..ca2516e30 100644 --- a/backend/crates/providers/openai/src/transport/client_sse.rs +++ b/backend/crates/providers/openai/src/transport/client_sse.rs @@ -197,7 +197,7 @@ impl CodexBackendClient { set_cookie_headers, rate_limit_headers, rate_limit_updates: Some(rate_limit_updates), - turn_state_update: None, + response_metadata_updates: None, websocket_pool_decision: None, diagnostics, response_metadata, @@ -464,7 +464,7 @@ impl CodexBackendClient { set_cookie_headers: exchange.set_cookie_headers, rate_limit_headers: exchange.rate_limit_headers, rate_limit_updates: Some(exchange.rate_limit_updates), - turn_state_update: Some(exchange.turn_state_update), + response_metadata_updates: Some(exchange.response_metadata_updates), websocket_pool_decision: exchange.pool_decision, diagnostics: exchange.diagnostics, response_metadata: exchange.response_metadata, diff --git a/backend/crates/providers/openai/src/transport/mod.rs b/backend/crates/providers/openai/src/transport/mod.rs index d5bcaddb3..6b2cf0785 100644 --- a/backend/crates/providers/openai/src/transport/mod.rs +++ b/backend/crates/providers/openai/src/transport/mod.rs @@ -35,7 +35,8 @@ pub use self::{ CodexAccountSelectionTelemetry, CodexBackendClient, CodexBackendJsonResponse, CodexBackendSseStream, CodexBackendStreamingResponse, CodexBackendTransport, CodexClientError, CodexClientResult, CodexRateLimitUpdates, CodexRequestContext, - CodexTransportDecision, CodexTransportMetrics, CodexTurnStateUpdate, build_reqwest_client, + CodexResponseMetadataUpdates, CodexTransportDecision, CodexTransportMetrics, + build_reqwest_client, }, diagnostics::{CodexUpstreamDiagnostics, CodexUpstreamSendPhase}, endpoints::{ diff --git a/backend/crates/providers/openai/src/transport/response_meta.rs b/backend/crates/providers/openai/src/transport/response_meta.rs index 304ec91e8..e1f773933 100644 --- a/backend/crates/providers/openai/src/transport/response_meta.rs +++ b/backend/crates/providers/openai/src/transport/response_meta.rs @@ -119,13 +119,37 @@ fn observe_typed_response_header(metadata: &mut CodexResponseMetadata, name: &st return; } match name.trim().to_ascii_lowercase().as_str() { - "openai-model" => metadata.effective_model = Some(value.to_string()), + "openai-model" | "x-openai-model" => { + metadata.effective_model = + gateway_protocol::openai::events::observed_model_name(value).map(str::to_owned) + } "x-models-etag" => metadata.models_etag = Some(value.to_string()), "x-reasoning-included" => metadata.reasoning_included = true, _ => {} } } +/// 与官方 Codex 一致,优先读取 response.headers,再读取 WS metadata 的顶层 headers。 +pub(super) fn reported_model_from_event(value: &serde_json::Value) -> Option<&str> { + [value.pointer("/response/headers"), value.get("headers")] + .into_iter() + .flatten() + .filter_map(serde_json::Value::as_object) + .find_map(|headers| { + headers.iter().find_map(|(name, value)| { + if !name.eq_ignore_ascii_case("openai-model") + && !name.eq_ignore_ascii_case("x-openai-model") + { + return None; + } + value + .as_str() + .or_else(|| value.as_array()?.first()?.as_str()) + .and_then(gateway_protocol::openai::events::observed_model_name) + }) + }) +} + fn filter_client_headers( headers: impl IntoIterator, ) -> Vec<(String, Bytes)> { diff --git a/backend/crates/providers/openai/src/transport/websocket/exchange/mod.rs b/backend/crates/providers/openai/src/transport/websocket/exchange/mod.rs index c37fa7f9a..2984ceb69 100644 --- a/backend/crates/providers/openai/src/transport/websocket/exchange/mod.rs +++ b/backend/crates/providers/openai/src/transport/websocket/exchange/mod.rs @@ -37,8 +37,8 @@ pub struct CodexWebSocketStreamingExchange { pub rate_limit_headers: Vec<(String, String)>, /// 上游内部 `codex.rate_limits` 事件里的结构化动态更新。 pub rate_limit_updates: CodexWebSocketRateLimitUpdates, - /// 上游内部 metadata 事件里的首个动态 turn state。 - pub turn_state_update: CodexWebSocketTurnStateUpdate, + /// 上游内部 metadata 事件里的请求级动态更新。 + pub response_metadata_updates: CodexWebSocketResponseMetadataUpdates, /// WebSocket 连接池决策。 pub pool_decision: Option, /// terminal completed 后该 socket 是否会保留 connection-local continuation。 @@ -54,17 +54,31 @@ pub type CodexWebSocketSseStream = Pin> + Send + 'static>>; /// live 流中的结构化限流动态更新。 pub type CodexWebSocketRateLimitUpdates = Arc>>; -/// live 流中的 turn state 动态更新。 -pub type CodexWebSocketTurnStateUpdate = Arc>>; +/// 单次响应的动态 metadata,与连接池保存的握手快照隔离。 +#[derive(Debug, Default)] +pub struct CodexWebSocketResponseMetadataUpdate { + /// 当前响应首次声明的会话续接状态。 + pub turn_state: Option, + /// 当前响应最新的服务端模型报告。 + pub reported_model: Option, +} + +pub type CodexWebSocketResponseMetadataUpdates = Arc>; pub(super) fn reusable_websocket_metadata( mut metadata: CodexWebSocketConnectionMetadata, ) -> CodexWebSocketConnectionMetadata { metadata.rate_limit_headers.clear(); metadata.turn_state = None; + // 模型报告属于上一轮请求,池中连接不能把它带到下一轮缺失报告的响应。 + metadata.response_metadata.effective_model = None; metadata .response_metadata .client_headers - .retain(|(name, _)| !name.eq_ignore_ascii_case("x-codex-turn-state")); + .retain(|(name, _)| { + !["x-codex-turn-state", "openai-model", "x-openai-model"] + .iter() + .any(|header| name.eq_ignore_ascii_case(header)) + }); metadata } diff --git a/backend/crates/providers/openai/src/transport/websocket/exchange/reducer.rs b/backend/crates/providers/openai/src/transport/websocket/exchange/reducer.rs index 2d0cb9a3f..28079ae0e 100644 --- a/backend/crates/providers/openai/src/transport/websocket/exchange/reducer.rs +++ b/backend/crates/providers/openai/src/transport/websocket/exchange/reducer.rs @@ -63,6 +63,9 @@ pub(super) fn reduce_websocket_event( &mut metadata.response_metadata, websocket_metadata_headers(&value), ); + if let Some(model) = response_meta::reported_model_from_event(&value) { + metadata.response_metadata.effective_model = Some(model.to_owned()); + } let turn_state_update = websocket_metadata_turn_state(&value).and_then(|turn_state| { if metadata.turn_state.is_some() { return None; diff --git a/backend/crates/providers/openai/src/transport/websocket/exchange/stream.rs b/backend/crates/providers/openai/src/transport/websocket/exchange/stream.rs index 379add7d2..a240db532 100644 --- a/backend/crates/providers/openai/src/transport/websocket/exchange/stream.rs +++ b/backend/crates/providers/openai/src/transport/websocket/exchange/stream.rs @@ -20,8 +20,9 @@ use super::super::{ use super::io::{next_websocket_message, reused_stream_receive_error}; use super::reducer::{ExchangeAction, WebSocketTerminalKind, reduce_websocket_event}; use super::{ - CodexWebSocketExchangeError, CodexWebSocketRateLimitUpdates, CodexWebSocketStreamingExchange, - CodexWebSocketTurnStateUpdate, WEBSOCKET_STREAM_BUFFER, reusable_websocket_metadata, + CodexWebSocketExchangeError, CodexWebSocketRateLimitUpdates, + CodexWebSocketResponseMetadataUpdate, CodexWebSocketResponseMetadataUpdates, + CodexWebSocketStreamingExchange, WEBSOCKET_STREAM_BUFFER, reusable_websocket_metadata, }; pub(in crate::transport::websocket) struct WebSocketStreamPoolReturn { @@ -69,8 +70,11 @@ pub(in crate::transport::websocket) fn stream_websocket_response( let response_metadata = metadata.clone(); let rate_limit_updates = Arc::new(Mutex::new(Vec::new())); let rate_limit_updates_for_task = Arc::clone(&rate_limit_updates); - let turn_state_update = Arc::new(Mutex::new(metadata.turn_state.clone())); - let turn_state_update_for_task = Arc::clone(&turn_state_update); + let response_metadata_updates = Arc::new(Mutex::new(CodexWebSocketResponseMetadataUpdate { + turn_state: metadata.turn_state.clone(), + reported_model: None, + })); + let response_metadata_updates_for_task = Arc::clone(&response_metadata_updates); let (tx, rx) = mpsc::channel(WEBSOCKET_STREAM_BUFFER); let (task_tracker, shutdown) = pool_return .as_ref() @@ -89,7 +93,7 @@ pub(in crate::transport::websocket) fn stream_websocket_response( trace, shutdown, rate_limit_updates: rate_limit_updates_for_task, - turn_state_update: turn_state_update_for_task, + response_metadata_updates: response_metadata_updates_for_task, tx, }) .await; @@ -111,7 +115,7 @@ pub(in crate::transport::websocket) fn stream_websocket_response( set_cookie_headers: response_metadata.set_cookie_headers, rate_limit_headers: response_metadata.rate_limit_headers, rate_limit_updates, - turn_state_update, + response_metadata_updates, pool_decision: None, connection_local_continuation: false, diagnostics: response_metadata.diagnostics, @@ -128,7 +132,7 @@ struct WebSocketStreamForwardState { stream_idle_timeout: Option, shutdown: CancellationToken, rate_limit_updates: CodexWebSocketRateLimitUpdates, - turn_state_update: CodexWebSocketTurnStateUpdate, + response_metadata_updates: CodexWebSocketResponseMetadataUpdates, tx: mpsc::Sender>, } @@ -142,7 +146,7 @@ async fn forward_websocket_response_stream(state: WebSocketStreamForwardState) { stream_idle_timeout, shutdown, rate_limit_updates, - turn_state_update, + response_metadata_updates, tx, } = state; let mut pool_return = pool_return; @@ -298,11 +302,14 @@ async fn forward_websocket_response_stream(state: WebSocketStreamForwardState) { last_event_type = Some(event_type); } if let Some(turn_state) = reduced.turn_state_update { - let mut pending = turn_state_update.lock().await; - if pending.is_none() { - *pending = Some(turn_state); + let mut pending = response_metadata_updates.lock().await; + if pending.turn_state.is_none() { + pending.turn_state = Some(turn_state); } } + if let Some(model) = metadata.response_metadata.effective_model.as_ref() { + response_metadata_updates.lock().await.reported_model = Some(model.clone()); + } let (frame, terminal) = match reduced.action { ExchangeAction::RateLimits(rate_limits) => { rate_limit_updates.lock().await.push(rate_limits); diff --git a/backend/crates/providers/openai/src/transport/websocket/mod.rs b/backend/crates/providers/openai/src/transport/websocket/mod.rs index b4f806f3b..a91bef760 100644 --- a/backend/crates/providers/openai/src/transport/websocket/mod.rs +++ b/backend/crates/providers/openai/src/transport/websocket/mod.rs @@ -21,8 +21,8 @@ pub use self::{ }, error::{CodexWebSocketCloseError, CodexWebSocketExchangeError, CodexWebSocketUpstreamError}, exchange::{ - CodexWebSocketRateLimitUpdates, CodexWebSocketSseStream, CodexWebSocketStreamingExchange, - CodexWebSocketTurnStateUpdate, + CodexWebSocketRateLimitUpdates, CodexWebSocketResponseMetadataUpdates, + CodexWebSocketSseStream, CodexWebSocketStreamingExchange, }, handshake::responses_websocket_endpoint, model::{ diff --git a/backend/crates/providers/openai/tests/provider/contract.rs b/backend/crates/providers/openai/tests/provider/contract.rs index 7acd29301..256ce7e54 100644 --- a/backend/crates/providers/openai/tests/provider/contract.rs +++ b/backend/crates/providers/openai/tests/provider/contract.rs @@ -66,6 +66,152 @@ use crate::support::{ }; use crate::transport::accept_codex_test_websocket; +#[tokio::test] +async fn responses_bill_sent_model_and_observe_unpriced_response_without_rewriting_it() { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_provider_contract").await; + let server = MockServer::start().await; + let body = concat!( + "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_model\",\"model\":\"gpt-created\"}}\n\n", + "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_model\",\"model\":\"gpt-6-sol\",\"status\":\"completed\",\"output\":[],\"usage\":{\"input_tokens\":100,\"output_tokens\":10,\"input_tokens_details\":{\"cached_tokens\":25},\"total_tokens\":110}}}\n\n", + ); + Mock::given(method("POST")) + .and(path("/codex/responses")) + .respond_with( + ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .set_body_string(body), + ) + .expect(1) + .mount(&server) + .await; + let provider = provider_with_base_url(&store, server.uri()); + let operation = Operation::Generate(GenerateRequest::from_protocol_payload( + ProtocolPayload::json_object( + "openai", + json!({"model":"gpt-5.6-sol","input":"hello"}) + .as_object() + .unwrap() + .clone(), + ) + .unwrap() + .with_context(Map::from_iter([("use_websocket".to_owned(), json!(false))])), + )); + let mut stream = provider + .execute( + planned_request("openai", operation), + context("req_model", CancellationToken::new()), + ) + .await + .unwrap(); + let mut observed = None; + let mut returned = None; + let mut costs = Vec::new(); + while let Some(event) = stream.next().await { + let event = event.expect("upstream response"); + for fact in event.canonical_facts() { + if let GatewayEvent::CalculatedCost(cost) = fact { + costs.push(cost.total().amount().scaled()); + } + } + if let Some(observation) = event.response_observation() { + observed = observation.upstream_response_model().map(str::to_owned); + } + if let Some(wire) = event.wire_event() + && wire.data().get("type").and_then(Value::as_str) == Some("response.completed") + { + returned = wire + .data() + .pointer("/response/model") + .and_then(Value::as_str) + .map(str::to_owned); + } + } + let requests = server.received_requests().await.expect("upstream requests"); + assert_eq!(captured_request_body(&requests[0])["model"], "gpt-5.4"); + assert_eq!(costs, vec![3_437_500]); + assert_eq!(observed.as_deref(), Some("gpt-6-sol")); + assert_eq!(returned, observed); +} + +#[tokio::test] +async fn websocket_bills_sent_model_and_observes_internal_model_report() { + for metadata_type in ["response.metadata", "codex.response.metadata"] { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_provider_contract").await; + let listener = TcpListener::bind("127.0.0.1:0").await.expect("listener"); + let base_url = format!("http://{}", listener.local_addr().expect("address")); + let server = tokio::spawn(async move { + let (socket, _) = listener.accept().await.expect("accept WebSocket"); + let mut websocket = accept_codex_test_websocket(socket).await; + let request = websocket.next().await.expect("request").expect("frame"); + let request: Value = + serde_json::from_str(request.to_text().expect("text")).expect("request JSON"); + assert_eq!(request["model"], "gpt-5.4"); + for event in [ + json!({"type":"response.created","response":{"id":"resp_model","model":"gpt-created"}}), + json!({"type":metadata_type,"headers":{"X-OpenAI-Model":["gpt-server-report"]}}), + json!({"type":"response.completed","response":{"id":"resp_model","model":"gpt-6-astra","status":"completed","output":[],"usage":{"input_tokens":100,"output_tokens":10,"input_tokens_details":{"cached_tokens":25},"total_tokens":110}}}), + ] { + websocket + .send(Message::Text(event.to_string().into())) + .await + .expect("response"); + } + }); + let payload = ProtocolPayload::json_object( + "openai", + Map::from_iter([ + ("model".to_owned(), json!("gpt-5.6-sol")), + ("input".to_owned(), json!("hello")), + ]), + ) + .expect("payload") + .with_context(Map::from_iter([("use_websocket".to_owned(), json!(true))])); + let mut stream = provider_with_base_url(&store, base_url) + .execute( + planned_request( + "openai", + Operation::Generate(GenerateRequest::from_protocol_payload(payload)), + ), + context("req_ws_model", CancellationToken::new()), + ) + .await + .expect("provider stream"); + let mut observed = None; + let mut returned = None; + let mut costs = Vec::new(); + while let Some(event) = stream.next().await { + let event = event.expect("provider event"); + for fact in event.canonical_facts() { + if let GatewayEvent::CalculatedCost(cost) = fact { + costs.push(cost.total().amount().scaled()); + } + } + if let Some(observation) = event.response_observation() { + observed = observation.upstream_response_model().map(str::to_owned); + } + if let Some(wire) = event.wire_event() + && wire.data().get("type").and_then(Value::as_str) == Some("response.completed") + { + returned = wire + .data() + .pointer("/response/model") + .and_then(Value::as_str) + .map(str::to_owned); + } + } + server.await.expect("server task"); + assert_eq!(costs, vec![3_437_500]); + assert_eq!( + observed.as_deref(), + Some("gpt-server-report"), + "{metadata_type}" + ); + assert_eq!(returned.as_deref(), Some("gpt-6-astra")); + } +} + #[tokio::test] async fn selected_proxy_location_overrides_global_and_reloads_without_mutating_client_payload() { use gateway_core::account::{OutboundProxy, RequestLocation}; diff --git a/backend/crates/providers/openai/tests/transport/canonical.rs b/backend/crates/providers/openai/tests/transport/canonical.rs index 3ddfa7f7a..8b0c4c033 100644 --- a/backend/crates/providers/openai/tests/transport/canonical.rs +++ b/backend/crates/providers/openai/tests/transport/canonical.rs @@ -6,6 +6,56 @@ use provider_openai::transport::canonical::{ use provider_openai::transport::protocol::websocket::websocket_event_to_sse_frame; use serde_json::json; +#[test] +fn response_model_observation_prefers_reported_headers_without_changing_wire() { + for websocket in [false, true] { + let mut decoder = CodexCanonicalDecoder::new("requested") + .with_reported_model(Some("opening-model")) + .with_raw_sse_passthrough(); + for (value, expected) in [ + ( + json!({"type":"response.created","response":{"id":"resp_model","model":"body-model"}}), + "opening-model", + ), + ( + json!({"type":"codex.response.metadata","headers":{"X-OpenAI-Model":["metadata-model"]}}), + "metadata-model", + ), + ( + json!({"type":"response.completed","headers":{"openai-model":"top-level"},"response":{"id":"resp_model","model":"final-body-model","headers":{"OpenAI-Model":"terminal-model"}}}), + "terminal-model", + ), + ] { + let raw = value.to_string(); + let frame = if websocket { + websocket_event_to_sse_frame(&raw).expect("WS event") + } else { + format!("data: {raw}\n\n") + }; + let events = decoder + .push(frame.as_bytes()) + .expect("decode model observation"); + assert_eq!(decoder.response_model(), Some(expected)); + assert!( + events + .iter() + .any(|event| event.wire_event().is_some_and(|wire| wire.data() == &value)) + ); + } + } +} + +#[test] +fn response_model_observation_uses_explicit_body_and_never_request_fallback() { + let mut decoder = CodexCanonicalDecoder::new("requested"); + decoder + .push(b"data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_model\"}}\n\n") + .expect("response model event"); + assert_eq!(decoder.response_model(), None); + decoder.push(b"data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_model\",\"model\":\"returned\"}}\n\n").expect("response model event"); + assert_eq!(decoder.response_model(), Some("returned")); +} + const METADATA_PREFIX_FIXTURE: &str = include_str!("fixtures/metadata_only_prefix.sse"); #[test] @@ -199,6 +249,48 @@ fn decoder_should_preserve_long_event_metadata_without_raw_frame() { assert_eq!(wire.data()["marker"], serde_json::json!(1)); } +#[test] +fn decoder_should_bill_the_sent_model_independently_of_the_response_model() { + for (sent, returned, expected_cost) in [ + ("gpt-5.6-sol", Some("gpt-6-sol"), Some(6_875_000)), + ("gpt-5.6-sol", Some("gpt-6-astra"), Some(6_875_000)), + ("gpt-5.6-sol", Some("gpt-5.6-sol"), Some(6_875_000)), + ("gpt-5.6-sol", None, Some(6_875_000)), + ("gpt-6-sol", Some("gpt-5.6-sol"), None), + ] { + let created = + json!({"type":"response.created","response":{"id":"resp_model_cost","model":returned}}); + let completed = json!({ + "type":"response.completed", + "response":{ + "id":"resp_model_cost","model":returned,"status":"completed","output":[], + "usage":{"input_tokens":100,"output_tokens":10,"input_tokens_details":{"cached_tokens":25},"total_tokens":110} + } + }); + let body = format!("data: {created}\n\ndata: {completed}\n\n"); + let mut decoder = CodexCanonicalDecoder::new(sent).with_raw_sse_passthrough(); + let events = decoder.push(body.as_bytes()).expect("model cost response"); + let cost = canonical_facts(&events) + .into_iter() + .find_map(|event| match event { + GatewayEvent::CalculatedCost(cost) => Some(cost.total().amount().scaled()), + _ => None, + }); + assert_eq!(cost, expected_cost, "{sent} -> {returned:?}"); + assert_eq!(decoder.response_model(), returned); + assert!(canonical_facts(&events).into_iter().any(|event| matches!( + event, + GatewayEvent::Completed(meta) if meta.model() == Some(returned.unwrap_or(sent)) + ))); + let raw: Vec = events + .iter() + .filter_map(|event| event.wire_event().and_then(|wire| wire.raw_sse_frame())) + .flat_map(|frame| frame.iter().copied()) + .collect(); + assert_eq!(raw, body.as_bytes()); + } +} + #[test] fn decoder_should_emit_calculated_cost_for_complete_known_model_usage() { let body = concat!( @@ -207,7 +299,7 @@ fn decoder_should_emit_calculated_cost_for_complete_known_model_usage() { "event: response.completed\n", "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_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 events = CodexCanonicalDecoder::new("fallback") + let events = CodexCanonicalDecoder::new("gpt-5.4") .push(body.as_bytes()) .expect("canonical priced response"); @@ -227,7 +319,7 @@ fn decoder_should_bill_requested_service_tier_despite_default_response() { "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").with_requested_service_tier(Some("priority")); + CodexCanonicalDecoder::new("gpt-5.4").with_requested_service_tier(Some("priority")); let events = decoder .push(body.as_bytes()) .expect("canonical priority response"); @@ -248,7 +340,7 @@ fn decoder_should_bill_standard_when_request_omits_service_tier() { "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 mut decoder = CodexCanonicalDecoder::new("gpt-5.4"); let events = decoder .push(body.as_bytes()) .expect("canonical response-tier priority response"); @@ -270,7 +362,7 @@ fn decoder_should_add_standard_web_search_call_cost() { "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_search_cost\",\"model\":\"gpt-5.4\",\"status\":\"completed\",\"output\":[{\"type\":\"web_search_call\",\"id\":\"ws_1\",\"status\":\"completed\",\"action\":{\"type\":\"search\"}}],\"usage\":{\"input_tokens\":0,\"output_tokens\":0,\"total_tokens\":0}}}\n\n", ); let tools = vec![json!({ "type": "web_search" })]; - let events = CodexCanonicalDecoder::new("fallback") + let events = CodexCanonicalDecoder::new("gpt-5.4") .with_request_tool_pricing("gpt-5.4", Some(&tools)) .push(body.as_bytes()) .expect("canonical web search response"); @@ -291,7 +383,7 @@ fn decoder_should_use_non_reasoning_preview_web_search_price() { "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_preview_search_cost\",\"model\":\"gpt-4o\",\"status\":\"completed\",\"output\":[{\"type\":\"web_search_call\",\"id\":\"ws_1\",\"status\":\"completed\",\"action\":{\"type\":\"search\"}}],\"usage\":{\"input_tokens\":0,\"output_tokens\":0,\"total_tokens\":0}}}\n\n", ); let tools = vec![json!({ "type": "web_search_preview" })]; - let events = CodexCanonicalDecoder::new("fallback") + let events = CodexCanonicalDecoder::new("gpt-4o") .with_request_tool_pricing("gpt-4o", Some(&tools)) .push(body.as_bytes()) .expect("canonical preview web search response"); @@ -312,7 +404,7 @@ fn astra_decoder_should_use_reasoning_preview_web_search_price() { "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_preview_search_cost\",\"model\":\"gpt-6-astra\",\"status\":\"completed\",\"output\":[{\"type\":\"web_search_call\",\"id\":\"ws_1\",\"status\":\"completed\",\"action\":{\"type\":\"search\"}}],\"usage\":{\"input_tokens\":0,\"output_tokens\":0,\"total_tokens\":0}}}\n\n", ); let tools = vec![json!({ "type": "web_search_preview" })]; - let events = CodexCanonicalDecoder::new("fallback") + let events = CodexCanonicalDecoder::new("gpt-6-astra") .with_request_tool_pricing("gpt-6-astra", Some(&tools)) .push(body.as_bytes()) .expect("canonical preview web search response"); @@ -333,7 +425,7 @@ fn decoder_should_fail_closed_for_fixed_block_web_search_content() { "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_fixed_search_cost\",\"model\":\"gpt-4o-mini\",\"status\":\"completed\",\"output\":[{\"type\":\"web_search_call\",\"id\":\"ws_1\",\"status\":\"completed\",\"action\":{\"type\":\"search\"}}],\"usage\":{\"input_tokens\":1,\"output_tokens\":1,\"total_tokens\":2}}}\n\n", ); let tools = vec![json!({ "type": "web_search" })]; - let events = CodexCanonicalDecoder::new("fallback") + let events = CodexCanonicalDecoder::new("gpt-4o-mini") .with_request_tool_pricing("gpt-4o-mini", Some(&tools)) .push(body.as_bytes()) .expect("canonical fixed-block search response"); @@ -355,7 +447,7 @@ fn websocket_decoder_should_bill_requested_service_tier_despite_default_response 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") + let mut decoder = CodexCanonicalDecoder::new("gpt-5.4") .with_requested_service_tier(Some("priority")) .with_raw_sse_passthrough(); let events = decoder diff --git a/backend/crates/providers/openai/tests/transport/mod.rs b/backend/crates/providers/openai/tests/transport/mod.rs index aff75a752..1d211dbd1 100644 --- a/backend/crates/providers/openai/tests/transport/mod.rs +++ b/backend/crates/providers/openai/tests/transport/mod.rs @@ -172,6 +172,7 @@ struct CollectedBackendResponse { rate_limit_headers: Vec<(String, String)>, websocket_pool_decision: Option, response_metadata: CodexResponseMetadata, + reported_model: Option, transport_metrics: CodexTransportMetrics, connection_local_continuation: bool, } @@ -246,7 +247,7 @@ async fn collect_backend_response( set_cookie_headers, mut rate_limit_headers, rate_limit_updates, - turn_state_update, + response_metadata_updates, websocket_pool_decision, diagnostics: _, response_metadata, @@ -268,8 +269,11 @@ async fn collect_backend_response( rate_limit_headers.extend(rate_limits_to_header_pairs(update)); } } - if let Some(update) = turn_state_update { - turn_state = update.lock().await.clone().or(turn_state); + let mut reported_model = response_metadata.effective_model.clone(); + if let Some(update) = response_metadata_updates { + let update = update.lock().await; + turn_state = update.turn_state.clone().or(turn_state); + reported_model = update.reported_model.clone().or(reported_model); } let body = String::from_utf8_lossy(&body_bytes).into_owned(); let usage = extract_sse_usage(&body).map_err(CodexClientError::InvalidSse)?; @@ -282,6 +286,7 @@ async fn collect_backend_response( rate_limit_headers, websocket_pool_decision, response_metadata, + reported_model, transport_metrics, connection_local_continuation, }) diff --git a/backend/crates/providers/openai/tests/transport/websocket/metadata.rs b/backend/crates/providers/openai/tests/transport/websocket/metadata.rs index d9d91239f..ea34dfd2c 100644 --- a/backend/crates/providers/openai/tests/transport/websocket/metadata.rs +++ b/backend/crates/providers/openai/tests/transport/websocket/metadata.rs @@ -78,6 +78,7 @@ async fn reused_websocket_should_keep_response_metadata_scoped_to_each_exchange( .expect("response within timeout") .expect("response should complete on the same socket"); assert_eq!(response.response_metadata, expected, "exchange {round}"); + assert_eq!(response.reported_model, Some(format!("model-{round}"))); assert!(response.body.contains(&format!("etag-{round}"))); assert_eq!(response.turn_state, Some(format!("turn-{round}"))); if round > 0 { diff --git a/backend/crates/providers/openai/tests/transport/websocket/mod.rs b/backend/crates/providers/openai/tests/transport/websocket/mod.rs index 8eb40bd10..89d7b10e0 100644 --- a/backend/crates/providers/openai/tests/transport/websocket/mod.rs +++ b/backend/crates/providers/openai/tests/transport/websocket/mod.rs @@ -931,7 +931,8 @@ async fn websocket_execute_response_create_request_should_capture_internal_metad json!({ "type": "response.metadata", "headers": { - "x-codex-turn-state": ["turn-from-metadata"] + "x-codex-turn-state": ["turn-from-metadata"], + "x-openai-model": ["gpt-internal-report"] } }) .to_string() @@ -969,6 +970,10 @@ async fn websocket_execute_response_create_request_should_capture_internal_metad server.await.unwrap(); assert_eq!(response.turn_state.as_deref(), Some("turn-from-metadata")); + assert_eq!( + response.reported_model.as_deref(), + Some("gpt-internal-report") + ); assert!(!response.body.contains("codex.rate_limits")); assert!(!response.body.contains("response.metadata")); assert!(response.body.contains("event: response.completed")); @@ -1048,7 +1053,7 @@ async fn reused_websocket_should_not_leak_turn_state_into_the_next_response() { .send(Message::Text( json!({ "type": "codex.response.metadata", - "headers": {"x-codex-turn-state": "turn-from-first-response"} + "headers": {"x-codex-turn-state": "turn-from-first-response", "openai-model": "first-response-model"} }) .to_string() .into(), @@ -1104,6 +1109,11 @@ async fn reused_websocket_should_not_leak_turn_state_into_the_next_response() { .websocket_pool_decision .is_some_and(WebSocketPoolDecision::is_reuse) ); + assert_eq!( + first.reported_model.as_deref(), + Some("first-response-model") + ); + assert_eq!(second.reported_model, None); assert_eq!(second.turn_state, None); assert!( second diff --git a/backend/crates/providers/xai/src/provider/stream.rs b/backend/crates/providers/xai/src/provider/stream.rs index ba8d81b5a..5d600a3ec 100644 --- a/backend/crates/providers/xai/src/provider/stream.rs +++ b/backend/crates/providers/xai/src/provider/stream.rs @@ -252,6 +252,10 @@ pub(super) fn cold_compaction_http_sse_stream( } } + if let Some(model) = canonical.response_model() { + observation = observation.with_upstream_response_model_if_valid(model); + yield ProviderEvent::observation(observation.clone()); + } if let Some(tier) = canonical.response_service_tier() { observation = observation.with_service_tier_if_valid(tier.to_owned()); yield ProviderEvent::observation(observation); @@ -534,6 +538,14 @@ pub(super) fn cold_http_sse_stream( }), ); } + if let Some(model) = decoder.response_model() + && observation.upstream_response_model() != Some(model) + { + observation = observation.with_upstream_response_model_if_valid(model); + yield ProviderEvent::observation(observation.clone().with_timings( + ProviderResponseTimings { first_token_ms, ..base_timings }, + )); + } if let Some(tier) = decoder.response_service_tier() && observation.service_tier() != Some(tier) { @@ -594,6 +606,14 @@ pub(super) fn cold_http_sse_stream( }), ); } + if let Some(model) = decoder.response_model() + && observation.upstream_response_model() != Some(model) + { + observation = observation.with_upstream_response_model_if_valid(model); + yield ProviderEvent::observation(observation.clone().with_timings( + ProviderResponseTimings { first_token_ms, ..base_timings }, + )); + } if let Some(tier) = decoder.response_service_tier() && observation.service_tier() != Some(tier) { diff --git a/backend/crates/providers/xai/src/transport/canonical.rs b/backend/crates/providers/xai/src/transport/canonical.rs index 8863b6aa5..fc07971c4 100644 --- a/backend/crates/providers/xai/src/transport/canonical.rs +++ b/backend/crates/providers/xai/src/transport/canonical.rs @@ -14,7 +14,9 @@ use gateway_core::metering::{ CurrencyCode, Decimal, Money, ProviderReportedCost, Usage, }; use gateway_core::upstream::UpstreamSendState; -use gateway_protocol::openai::events::{TokenUsage, billable_usage_is_complete, extract_usage}; +use gateway_protocol::openai::events::{ + ResponseModelObservation, TokenUsage, billable_usage_is_complete, extract_usage, +}; use gateway_protocol::openai::sse::{SseEvent, SseEventDecoder}; use serde_json::Value; @@ -167,7 +169,7 @@ fn usd_price_per_million(per_token_ticks: u128) -> Option { pub struct GrokCanonicalDecoder { decoder: SseEventDecoder, response_transform: GrokResponseTransform, - fallback_model: String, + upstream_model: String, response_id: Option, started: bool, completed: bool, @@ -176,15 +178,17 @@ pub struct GrokCanonicalDecoder { usage_emitted: bool, output_start_seen: bool, response_service_tier: Option, + response_model: ResponseModelObservation, requires_provider_cost: bool, } impl GrokCanonicalDecoder { - pub fn new(fallback_model: impl Into) -> Self { + /// 使用路由后最终发往上游的请求模型计价,并在响应缺少模型时用于 canonical 兜底。 + pub fn new(upstream_model: impl Into) -> Self { Self { decoder: SseEventDecoder::default(), response_transform: GrokResponseTransform::default(), - fallback_model: fallback_model.into(), + upstream_model: upstream_model.into(), response_id: None, started: false, completed: false, @@ -193,6 +197,7 @@ impl GrokCanonicalDecoder { usage_emitted: false, output_start_seen: false, response_service_tier: None, + response_model: ResponseModelObservation::default(), requires_provider_cost: false, } } @@ -200,7 +205,7 @@ impl GrokCanonicalDecoder { /// 创建在 canonical 与 wire 投影处理每个上游 event 前先还原请求级 tool 别名的 /// 解码器。 #[must_use] - pub fn for_request(fallback_model: impl Into, request: &GrokResponsesRequest) -> Self { + pub fn for_request(upstream_model: impl Into, request: &GrokResponsesRequest) -> Self { Self { response_transform: request.response_transform(), requires_provider_cost: request.body().get("tool_choice").and_then(Value::as_str) @@ -217,7 +222,7 @@ impl GrokCanonicalDecoder { ) }) }), - ..Self::new(fallback_model) + ..Self::new(upstream_model) } } @@ -227,6 +232,12 @@ impl GrokCanonicalDecoder { self.response_service_tier.as_deref() } + /// 返回原始上游响应声明的模型,缺失时不使用请求模型补齐。 + #[must_use] + pub fn response_model(&self) -> Option<&str> { + self.response_model.model() + } + pub fn push(&mut self, chunk: &[u8]) -> Result, ProviderError> { let events = self.decoder.push(chunk).map_err(protocol_error)?; self.decode(events) @@ -314,6 +325,7 @@ impl GrokCanonicalDecoder { continue; }; let event_type = event_type.to_owned(); + self.response_model.observe(Some(&event_type), &value); // 工具转换可能隐藏注入的调用,计费事实必须从转换前的上游事件读取。 if let Some(response) = value.get("response") { if let Some(tier) = response.get("service_tier").and_then(Value::as_str) { @@ -442,7 +454,7 @@ impl GrokCanonicalDecoder { .get("model") .and_then(Value::as_str) .filter(|model| !model.is_empty()) - .unwrap_or(&self.fallback_model) + .unwrap_or(&self.upstream_model) .to_owned(); self.response_id = Some(response_id.clone()); self.started = true; @@ -649,13 +661,19 @@ impl GrokCanonicalDecoder { .get("model") .and_then(Value::as_str) .filter(|model| !model.is_empty()) - .unwrap_or(&self.fallback_model) + .unwrap_or(&self.upstream_model) .to_owned(); if let Some(cost) = provider_reported_cost(response)? { output.push(GatewayEvent::ProviderCost(cost)); } else if !self.requires_provider_cost && let Some(cost) = usage.and_then(|usage| { - calculated_cost(response, &model, usage, self.response_service_tier()) + // 已报告金额仍优先;本地估价只使用实际发送模型,响应模型仅作观测。 + calculated_cost( + response, + &self.upstream_model, + usage, + self.response_service_tier(), + ) }) { output.push(GatewayEvent::CalculatedCost(cost)); diff --git a/backend/crates/providers/xai/tests/transport/canonical.rs b/backend/crates/providers/xai/tests/transport/canonical.rs index 4d30b013f..dae13d82a 100644 --- a/backend/crates/providers/xai/tests/transport/canonical.rs +++ b/backend/crates/providers/xai/tests/transport/canonical.rs @@ -10,6 +10,60 @@ use provider_xai::{ grok_billing_breakdown_with_tier, }; +#[test] +fn decoder_should_bill_the_sent_model_independently_of_the_response_model() { + for (sent, returned, reported_cost, expected_cost) in [ + ("grok-4.5", Some("grok-future"), None, Some(2_175_000)), + ("grok-4.5", Some("grok-4.3"), None, Some(2_175_000)), + ("grok-4.5", None, None, Some(2_175_000)), + ("grok-future", Some("grok-4.5"), None, None), + ("grok-4.5", Some("grok-future"), Some(123), None), + ("grok-4.5", Some("grok-future"), Some(0), None), + ] { + let created = serde_json::json!({"type":"response.created","response":{"id":"resp_model_cost","model":returned}}); + let mut completed = serde_json::json!({ + "type":"response.completed", + "response":{ + "id":"resp_model_cost","model":returned,"status":"completed","output":[], + "usage":{"input_tokens":100,"output_tokens":10,"input_tokens_details":{"cached_tokens":25},"total_tokens":110} + } + }); + if let Some(ticks) = reported_cost { + completed["response"]["usage"]["cost_in_usd_ticks"] = serde_json::json!(ticks); + } + let body = format!("data: {created}\n\ndata: {completed}\n\n"); + let mut decoder = GrokCanonicalDecoder::new(sent); + let events = decoder.push(body.as_bytes()).expect("model cost response"); + let facts: Vec<_> = events + .into_iter() + .flat_map(|event| event.into_parts().0) + .collect(); + assert_eq!( + calculated_cost_ticks(&facts), + expected_cost, + "{sent} -> {returned:?}" + ); + assert_eq!(provider_cost_ticks(&facts), reported_cost); + assert_eq!(decoder.response_model(), returned); + assert!(facts.iter().any(|event| matches!( + event, + GatewayEvent::Completed(meta) if meta.model() == Some(returned.unwrap_or(sent)) + ))); + } +} + +#[test] +fn response_model_observation_preserves_raw_model_and_missing_values() { + for model in [None, Some("grok-4.6-build")] { + let mut decoder = GrokCanonicalDecoder::new("grok-4.6"); + let body = serde_json::json!({"type":"response.created","response":{"id":"resp_model","model":model}}); + decoder + .push(format!("data: {body}\n\n").as_bytes()) + .expect("decode created"); + assert_eq!(decoder.response_model(), model); + } +} + fn terminal_cost_events( model: &str, input_tokens: u64, @@ -33,11 +87,11 @@ fn terminal_cost_events( cached_tokens = cached_tokens, provider_cost = provider_cost, ); - decode_canonical(body.as_bytes()).expect("canonical cost response") + decode_canonical(body.as_bytes(), model).expect("canonical cost response") } -fn decode_canonical(body: &[u8]) -> Result, ProviderError> { - GrokCanonicalDecoder::new("fallback") +fn decode_canonical(body: &[u8], upstream_model: &str) -> Result, ProviderError> { + GrokCanonicalDecoder::new(upstream_model) .push(body) .map(|events| { events @@ -97,7 +151,7 @@ fn decoder_should_normalize_text_usage_and_completion() { "event: response.completed\n", "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_1\",\"model\":\"grok-code-test\",\"status\":\"completed\",\"usage\":{\"input_tokens\":3,\"output_tokens\":2,\"total_tokens\":5,\"cost_in_usd_ticks\":37756000}}}\n\n", ); - let events = decode_canonical(body.as_bytes()).expect("canonical response"); + let events = decode_canonical(body.as_bytes(), "fallback").expect("canonical response"); assert!(matches!(events[0], GatewayEvent::Started(_))); assert!(matches!( @@ -188,7 +242,8 @@ fn decoder_should_preserve_image_tool_tokens_in_canonical_usage() { "event: response.completed\n", "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_image_usage\",\"model\":\"grok-4.5\",\"status\":\"completed\",\"usage\":{\"input_tokens\":12,\"output_tokens\":5,\"total_tokens\":17},\"tool_usage\":{\"image_gen\":{\"input_tokens\":31,\"output_tokens\":9}}}}\n\n", ); - let events = decode_canonical(body.as_bytes()).expect("canonical image usage response"); + let events = + decode_canonical(body.as_bytes(), "fallback").expect("canonical image usage response"); let image_usage = events.iter().find_map(|event| match event { GatewayEvent::Usage(usage) => Some((usage.image_input_tokens, usage.image_output_tokens)), _ => None, @@ -205,7 +260,7 @@ fn decoder_should_leave_cost_unavailable_when_upstream_omits_it() { "event: response.completed\n", "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_no_cost\",\"status\":\"completed\",\"usage\":{\"input_tokens\":1,\"output_tokens\":1,\"total_tokens\":2}}}\n\n", ); - let events = decode_canonical(body.as_bytes()).expect("canonical response"); + let events = decode_canonical(body.as_bytes(), "fallback").expect("canonical response"); assert!(!events.iter().any(|event| matches!( event, @@ -420,7 +475,8 @@ fn decoder_should_leave_incomplete_usage_pricing_unavailable() { "event: response.completed\n", "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_partial\",\"model\":\"grok-4.5\",\"status\":\"completed\",\"usage\":{\"total_tokens\":3}}}\n\n", ); - let events = decode_canonical(body.as_bytes()).expect("canonical partial usage response"); + let events = + decode_canonical(body.as_bytes(), "grok-4.5").expect("canonical partial usage response"); assert_eq!(calculated_cost_ticks(&events), None); } @@ -501,7 +557,7 @@ fn decoder_should_normalize_function_call_and_tool_finish() { "event: response.completed\n", "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_tool\",\"model\":\"grok-code-test\",\"status\":\"completed\"}}\n\n", ); - let events = decode_canonical(body.as_bytes()).expect("canonical function call"); + let events = decode_canonical(body.as_bytes(), "fallback").expect("canonical function call"); assert!(events.iter().any(|event| matches!( event, @@ -1076,7 +1132,7 @@ fn decoder_should_coalesce_reasoning_item_part_and_summary_index() { "event: response.completed\n", "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_reason\",\"status\":\"completed\"}}\n\n", ); - let events = decode_canonical(body.as_bytes()).expect("canonical reasoning"); + let events = decode_canonical(body.as_bytes(), "fallback").expect("canonical reasoning"); assert_eq!( events @@ -1303,7 +1359,7 @@ fn decoder_should_preserve_incomplete_length_reason() { "event: response.incomplete\n", "data: {\"type\":\"response.incomplete\",\"response\":{\"id\":\"resp_short\",\"status\":\"incomplete\",\"incomplete_details\":{\"reason\":\"max_output_tokens\"}}}\n\n", ); - let events = decode_canonical(body.as_bytes()).expect("incomplete response"); + let events = decode_canonical(body.as_bytes(), "fallback").expect("incomplete response"); assert!(matches!( events.last(), diff --git a/backend/migrations/.frozen-sha256 b/backend/migrations/.frozen-sha256 index 5cc72da1a..3e5c9bfd3 100644 --- a/backend/migrations/.frozen-sha256 +++ b/backend/migrations/.frozen-sha256 @@ -9,3 +9,4 @@ b7d3df7b48d835a00a3b74cbefc958a039d38fea5529cdad04bfbc703dca72aa 0001_initial.s 161df9ae43da34d51b340bbf95d71c61a8c1a2aa142382f693a9d127c834630d 0009_account_model_access.sql 3751a802e80b5d538abc5619a2397aa4bf13f52076068a489892f664907adf49 0010_account_auto_freeze.sql ae7f7acb3014f0a60d979dd9c04b3065322b206e42eb6256505baf78caf9dd22 0011_proxy_request_location.sql +b20e425c0b062b72d35fdebe294e023eb17ceb44ac93083ff3d48400715bf430 0012_upstream_response_model.sql diff --git a/backend/migrations/0012_upstream_response_model.sql b/backend/migrations/0012_upstream_response_model.sql new file mode 100644 index 000000000..b5b0e92db --- /dev/null +++ b/backend/migrations/0012_upstream_response_model.sql @@ -0,0 +1,10 @@ +-- 返回模型独立于发送模型,避免展示观测改写路由与统计口径。 +ALTER TABLE model_requests ADD COLUMN upstream_response_model text; + +-- 历史记录只回填已经保存的明确报告,不从请求模型或映射模型推断。 +UPDATE model_requests +SET upstream_response_model = btrim(provider_observation_json->>'upstreamReportedModel') +WHERE provider_kind = 'openai' + AND jsonb_typeof(provider_observation_json->'upstreamReportedModel') = 'string' + AND octet_length(btrim(provider_observation_json->>'upstreamReportedModel')) BETWEEN 1 AND 256 + AND provider_observation_json->>'upstreamReportedModel' !~ '[[:cntrl:]]'; diff --git a/docs/api.md b/docs/api.md index 8c06c9578..38a704326 100644 --- a/docs/api.md +++ b/docs/api.md @@ -1102,6 +1102,13 @@ errorCode, errorMessage, startedAt, completedAt, expiresAt, createdAt, updatedAt request/response/upstream ID、outcome 与搜索文本。诊断 `dimension` 可取 `model`、`account`、 `apiKey`、`provider`、`transport`、`failureClass`、`status`。 +管理端请求列表与详情分别保留 `requestedModel`(客户端请求)、`upstreamModel`(网关发送)与 +`upstreamResponseModel`(上游返回)。返回模型缺失时为 `null`,不使用请求或映射模型补齐。 +OpenAI 优先采用服务端 `openai-model` / `x-openai-model` 报告(流内报告可覆盖初始响应头), +没有报告时采用正文明确声明的 `response.model`;xAI 采用原始正文声明。正文模型以终态优先, +缺少终态声明时保留首次声明。这些值仅表示上游报告,不作为模型真实性证明,也不参与路由、 +聚合或本地计价模型选择。历史数据只回填此前已保存的 OpenAI 模型报告,其余保留未知。 + 请求记录列表的 `search` 使用字面量前缀匹配,支持请求 ID、Client Key ID / 名称、 账号 ID、账号邮箱与名称、请求 / 上游模型 ID、上游请求 ID。密钥名称不区分大小写,其他字段区分大小写。 密钥名称按当前密钥记录检索,改名后使用新名称,删除后仍可按 Client Key ID 查询历史记录。 @@ -1145,6 +1152,10 @@ Dashboard 的 `accountUsage[]` 由后端提供 `usageWindow`、`metricLabel`、` 和本地用量由 Provider/Admin 投影。前端不得从套餐缺失推断免费套餐,也不得从显示时舍入的百分比推断 触顶。滚动窗口使用相应时间范围的本地用量,独立于 Dashboard 的今日统计范围。 +OpenAI 与 xAI 的本地费用估算按实际发送给上游的请求模型(`upstreamModel`)查价,结合响应中的实际 +用量计算;客户端请求 A、路由后发送 B 时按 B 计价,响应返回 C 不改变计价模型。实际发送模型缺少 +定价时不估算,也不借用响应模型的价格。Provider 明确上报的已计费金额仍优先于本地估算。 + OpenAI Responses 用量记录的 `serviceTier` 与本地费用估算统一采用 Provider 最终发给上游的请求 `service_tier`,不使用响应档位覆盖或回退。例如发送 `priority`、响应回显 `default` 时,仍显示 `Fast` 并按 Priority 价格估算。未发送档位时,`serviceTier` 保持缺失,展示与估算按标准档处理。 diff --git a/docs/architecture.md b/docs/architecture.md index 176300234..03500aaf7 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -135,6 +135,7 @@ Provider 在选定账号后应用代理位置覆盖。请求期间不额外查 `engine::observation` 统一维护单次响应的用量、费用、时间和响应 ID,并负责重试前清理;协调器继续 独占发送、提交、重试和终结顺序。Provider 上报费用优先于本地估算,丢弃的 attempt 不得污染最终计量。 +Provider 本地估算按当前 attempt 实际发送的上游模型查价,响应声明的模型只作观测;费用明细复用同一口径。 Client Key 费用账本独立累计各次 attempt 的实际费用,不能因请求重试而清空已产生的费用或未知计费状态。 ## 4. 数据面请求生命周期 diff --git a/frontend/src/api/modules/usage.ts b/frontend/src/api/modules/usage.ts index a83b6fb61..82effb080 100644 --- a/frontend/src/api/modules/usage.ts +++ b/frontend/src/api/modules/usage.ts @@ -78,6 +78,7 @@ export interface UsageListRecord { model: string | null requestedModel: string | null upstreamModel: string | null + upstreamResponseModel: string | null serviceTier: string | null clientTransport: string upstreamTransport: string | null @@ -110,6 +111,7 @@ export interface UsageRecord { model: string | null requestedModel: string | null upstreamModel: string | null + upstreamResponseModel: string | null serviceTier: string | null statusCode: number | null clientTransport: string diff --git a/frontend/src/views/theme/fixtures/dashboard.ts b/frontend/src/views/theme/fixtures/dashboard.ts index b1a0db4db..d243daa6d 100644 --- a/frontend/src/views/theme/fixtures/dashboard.ts +++ b/frontend/src/views/theme/fixtures/dashboard.ts @@ -120,6 +120,7 @@ function previewUsageRecord(options: PreviewUsageRecordOptions): UsageListRecord model: options.model, requestedModel: options.model, upstreamModel: options.model, + upstreamResponseModel: null, serviceTier: options.provider === 'openai' ? 'priority' : 'standard', clientTransport: options.clientTransport, upstreamTransport: options.upstreamTransport, diff --git a/frontend/src/views/usage/components/UsageModelCell.vue b/frontend/src/views/usage/components/UsageModelCell.vue index 21a797f0b..610e00483 100644 --- a/frontend/src/views/usage/components/UsageModelCell.vue +++ b/frontend/src/views/usage/components/UsageModelCell.vue @@ -16,16 +16,24 @@ const modelDisplay = computed(() => usageModelDisplay(props.record))
{{ modelDisplay.primary }}
- +
diff --git a/frontend/src/views/usage/components/UsageRecordDetailModal.vue b/frontend/src/views/usage/components/UsageRecordDetailModal.vue index 04a2a63cb..30a749381 100644 --- a/frontend/src/views/usage/components/UsageRecordDetailModal.vue +++ b/frontend/src/views/usage/components/UsageRecordDetailModal.vue @@ -80,6 +80,7 @@ const modelRouteItems = computed(() => [ mono: true, }, { label: '存储模型', value: props.record?.model, mono: true }, + { label: '上游返回模型', value: props.record?.upstreamResponseModel, mono: true }, ]) const clientUpstreamItems = computed(() => [ diff --git a/frontend/src/views/usage/utils/records.ts b/frontend/src/views/usage/utils/records.ts index fa65c4c4f..1291eb073 100644 --- a/frontend/src/views/usage/utils/records.ts +++ b/frontend/src/views/usage/utils/records.ts @@ -29,6 +29,7 @@ export interface UsageViewModel { model: string | null requestedModel: string | null upstreamModel: string | null + upstreamResponseModel: string | null serviceTier: string | null statusCode: number | null clientTransport: string @@ -96,6 +97,7 @@ export function normalizeUsageRecord(record: UsageRecordDetail): UsageViewModel model: record.model, requestedModel: record.requestedModel, upstreamModel: record.upstreamModel, + upstreamResponseModel: record.upstreamResponseModel, serviceTier: record.serviceTier, statusCode: record.statusCode, clientTransport: record.clientTransport, @@ -217,7 +219,27 @@ export function usageModelDisplay(record: UsageCommonRecord) { ? storedModel : '' - return { primary, secondary } + const responseModel = record.upstreamResponseModel || '' + const returned = responseModel && responseModel !== primary ? responseModel : '' + const routes = [] + if (secondary && secondary !== returned) { + routes.push({ + model: secondary, + kind: 'mapped' as const, + description: `网关映射后发送给上游的模型:${secondary}`, + }) + } + if (returned) { + routes.push({ + model: returned, + kind: 'returned' as const, + description: returned === secondary + ? `上游返回模型:${returned}(与网关映射后发送的模型一致)` + : `上游返回模型:${returned}`, + }) + } + + return { primary, secondary, routes } } export function usageTokenDetails(record: Pick) {