From 71e26f44b7016a91119b38c9079fe2bf9684f67b Mon Sep 17 00:00:00 2001 From: Zyy <39424122+zyycn@users.noreply.github.com> Date: Thu, 17 Sep 2026 14:55:54 +0800 Subject: [PATCH] fix(usage): bill sent models and show response model mismatches (#136) * feat(usage): show upstream response model mismatches Persist observed response models separately from routing and billing facts. Surface mismatches with the existing model arrow and native title tooltips. Keep response observations isolated across retries and WebSocket reuse. Co-authored-by: Codex * fix(usage): calculate costs from the outbound model Co-authored-by: Codex --------- Co-authored-by: Zyy Co-authored-by: Codex --- .../gateway-admin/src/model/observability.rs | 2 + .../tests/use_case/observability.rs | 1 + .../src/admin/observability/presenter.rs | 2 + .../src/admin/observability/wire.rs | 2 + .../tests/admin/observability/response.rs | 4 + .../gateway-api/tests/key_usage/fixtures.rs | 1 + .../gateway-core/src/engine/coordinator.rs | 12 ++ backend/crates/gateway-core/src/engine/mod.rs | 1 + backend/crates/gateway-core/src/event.rs | 17 ++ .../gateway-core/tests/engine/coordinator.rs | 9 ++ .../gateway-protocol/src/openai/events.rs | 50 ++++++ .../gateway-protocol/tests/openai/events.rs | 44 ++++++ .../gateway-store/src/postgres/execution.rs | 12 +- .../src/postgres/execution_buffer.rs | 1 + .../src/postgres/observability/mapping.rs | 4 + .../src/postgres/observability/model.rs | 2 + .../postgres/observability/queries/usage.rs | 4 +- .../gateway-store/tests/postgres/execution.rs | 35 +++++ .../openai/src/provider/execution.rs | 69 ++++++--- .../providers/openai/src/provider/mod.rs | 4 +- .../openai/src/provider/observation.rs | 16 ++ .../openai/src/transport/canonical.rs | 50 +++++- .../providers/openai/src/transport/client.rs | 10 +- .../openai/src/transport/client_sse.rs | 4 +- .../providers/openai/src/transport/mod.rs | 3 +- .../openai/src/transport/response_meta.rs | 26 +++- .../src/transport/websocket/exchange/mod.rs | 24 ++- .../transport/websocket/exchange/reducer.rs | 3 + .../transport/websocket/exchange/stream.rs | 29 ++-- .../openai/src/transport/websocket/mod.rs | 4 +- .../openai/tests/provider/contract.rs | 146 ++++++++++++++++++ .../openai/tests/transport/canonical.rs | 108 ++++++++++++- .../providers/openai/tests/transport/mod.rs | 11 +- .../tests/transport/websocket/metadata.rs | 1 + .../openai/tests/transport/websocket/mod.rs | 14 +- .../providers/xai/src/provider/stream.rs | 20 +++ .../providers/xai/src/transport/canonical.rs | 36 +++-- .../xai/tests/transport/canonical.rs | 76 +++++++-- backend/migrations/.frozen-sha256 | 1 + .../0012_upstream_response_model.sql | 10 ++ docs/api.md | 11 ++ docs/architecture.md | 1 + frontend/src/api/modules/usage.ts | 2 + .../src/views/theme/fixtures/dashboard.ts | 1 + .../views/usage/components/UsageModelCell.vue | 16 +- .../components/UsageRecordDetailModal.vue | 1 + frontend/src/views/usage/utils/records.ts | 24 ++- 47 files changed, 825 insertions(+), 99 deletions(-) create mode 100644 backend/migrations/0012_upstream_response_model.sql 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) {