diff --git a/backend/crates/gateway-core/src/event.rs b/backend/crates/gateway-core/src/event.rs index bcc5d5ea7..aa6cb4159 100644 --- a/backend/crates/gateway-core/src/event.rs +++ b/backend/crates/gateway-core/src/event.rs @@ -521,7 +521,7 @@ impl fmt::Debug for ProviderResponseHeader { } } -/// Core 消费的实际上游响应事实。 +/// Core 消费的 Provider 执行观测,不会原样进入客户端响应。 #[derive(Debug, Clone, PartialEq, Eq)] pub struct ProviderResponseObservation { transport: UpstreamTransport, @@ -575,7 +575,7 @@ impl ProviderResponseObservation { self } - /// 附加 Provider 在响应中确认的服务档位。 + /// 附加 Provider 确定的统计档位,Core 不重新解释其来源。 /// /// # Errors /// @@ -590,7 +590,7 @@ impl ProviderResponseObservation { Ok(self) } - /// 附加合法服务档位;不适合持久化的上游观测值会被忽略。 + /// 附加 Provider 确定的统计档位;不适合持久化的值会被忽略。 #[must_use] pub fn with_service_tier_if_valid(mut self, service_tier: impl Into) -> Self { let service_tier = service_tier.into(); diff --git a/backend/crates/providers/openai/src/provider/observation.rs b/backend/crates/providers/openai/src/provider/observation.rs index 1a399fe34..80a6e32ff 100644 --- a/backend/crates/providers/openai/src/provider/observation.rs +++ b/backend/crates/providers/openai/src/provider/observation.rs @@ -117,8 +117,9 @@ impl OpenAiResponseObservationState { if let Some(metadata) = self.provider_metadata() { observation = observation.with_provider_metadata(metadata); } - if let Some(service_tier) = self.effective_service_tier() { - observation = observation.with_service_tier_if_valid(service_tier.to_owned()); + // 统计与计费采用最终发送档位;响应回显只作为独立诊断事实保留。 + if let Some(service_tier) = &self.requested_service_tier { + observation = observation.with_service_tier_if_valid(service_tier.clone()); } Some(observation) } @@ -161,12 +162,6 @@ impl OpenAiResponseObservationState { true } - pub(super) fn effective_service_tier(&self) -> Option<&str> { - self.upstream_service_tier - .as_deref() - .or(self.requested_service_tier.as_deref()) - } - pub(super) fn merge_rate_limit_headers(&mut self, updates: &[(String, String)]) -> bool { let mut changed = false; for (name, value) in selected_observation_headers(updates) { diff --git a/backend/crates/providers/openai/src/transport/canonical.rs b/backend/crates/providers/openai/src/transport/canonical.rs index 7e542305f..77be5b8f1 100644 --- a/backend/crates/providers/openai/src/transport/canonical.rs +++ b/backend/crates/providers/openai/src/transport/canonical.rs @@ -158,7 +158,7 @@ impl CodexCanonicalDecoder { self } - /// 响应未报告实际档位时,才使用请求档位估算费用。 + /// 使用最终发送给上游的请求档位估算费用,不由响应回显覆盖。 #[must_use] pub fn with_requested_service_tier(mut self, service_tier: Option<&str>) -> Self { self.requested_service_tier = normalize_service_tier(service_tier); @@ -217,7 +217,7 @@ impl CodexCanonicalDecoder { std::mem::take(&mut self.timing_signals) } - /// 返回本 attempt 已从上游响应生命周期帧观察到的实际服务档位。 + /// 返回本 attempt 的上游响应回显档位,仅供诊断,不作为本地计费依据。 #[must_use] pub fn response_service_tier(&self) -> Option<&str> { self.response_service_tier.as_deref() @@ -861,10 +861,8 @@ impl CodexCanonicalDecoder { .filter(|model| !model.is_empty()) .unwrap_or(&self.fallback_model) .to_owned(); - let service_tier = self - .response_service_tier - .as_deref() - .or(self.requested_service_tier.as_deref()); + // 与用量统计统一按最终发送档位估算,响应回显不改变本地计价口径。 + let service_tier = self.requested_service_tier.as_deref(); let tool_calls = billable_tool_calls(response); if let Some(breakdown) = usage .filter(|usage| billable_usage_is_complete(response, *usage)) diff --git a/backend/crates/providers/openai/tests/provider/contract.rs b/backend/crates/providers/openai/tests/provider/contract.rs index f314b8508..62a0afc9a 100644 --- a/backend/crates/providers/openai/tests/provider/contract.rs +++ b/backend/crates/providers/openai/tests/provider/contract.rs @@ -4843,10 +4843,157 @@ async fn completed_response_persists_session_affinity_before_stream_consumer_sto drop(stream); assert_eq!(affinity.binding_count(), 1); - assert_eq!(observed_service_tier.as_deref(), Some("default")); + assert_eq!(observed_service_tier.as_deref(), Some("priority")); assert_eq!(upstream_service_tier.as_deref(), Some("default")); } +#[tokio::test] +async fn responses_should_observe_and_bill_the_outbound_service_tier_on_both_transports() { + for use_websocket in [false, true] { + for (requested, reported, expected_cost) in [ + (Some("priority"), Some("default"), 6_875_000), + (Some("priority"), None, 6_875_000), + (Some("default"), Some("priority"), 3_437_500), + (None, Some("priority"), 3_437_500), + ] { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_provider_contract").await; + let mut created = + json!({"type":"response.created","response":{"id":"resp_tier","model":"gpt-5.4"}}); + let mut completed = json!({ + "type":"response.completed", + "response":{ + "id":"resp_tier","model":"gpt-5.4","status":"completed","output":[], + "usage":{"input_tokens":100,"output_tokens":10,"input_tokens_details":{"cached_tokens":25,"cache_write_tokens":0},"total_tokens":110} + } + }); + if let Some(reported) = reported { + created["response"]["service_tier"] = json!("auto"); + completed["response"]["service_tier"] = json!(reported); + } + let response_frames = format!( + "event: response.created\ndata: {created}\n\nevent: response.completed\ndata: {completed}\n\n" + ); + let (base_url, http_server, websocket_server) = if use_websocket { + 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"); + for event in [created, completed] { + websocket + .send(Message::Text(event.to_string().into())) + .await + .expect("response"); + } + request + }); + (base_url, None, Some(server)) + } else { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/codex/responses")) + .respond_with( + ResponseTemplate::new(200) + .set_body_raw(response_frames.clone(), "text/event-stream"), + ) + .expect(1) + .mount(&server) + .await; + (server.uri(), Some(server), None) + }; + let mut body = Map::from_iter([ + ("model".to_owned(), json!("gpt-5.4")), + ("input".to_owned(), json!("hello")), + ]); + if let Some(requested) = requested { + body.insert("service_tier".to_owned(), json!(requested)); + } + let payload = ProtocolPayload::json_object("openai", body) + .expect("payload") + .with_context(Map::from_iter([( + "use_websocket".to_owned(), + json!(use_websocket), + )])); + let mut stream = provider_with_base_url(&store, base_url) + .execute( + planned_request( + "openai", + Operation::Generate(GenerateRequest::from_protocol_payload(payload)), + ), + context("req_service_tier", CancellationToken::new()), + ) + .await + .expect("provider stream"); + let mut observations = Vec::new(); + let mut costs = Vec::new(); + let mut raw_response = Vec::new(); + while let Some(event) = stream.next().await { + let event = event.expect("provider event"); + if let Some(observation) = event.response_observation() { + observations.push(observation.clone()); + } + for fact in event.canonical_facts() { + if let GatewayEvent::CalculatedCost(cost) = fact { + costs.push(cost.total().amount().scaled()); + } + } + if let Some(frame) = event.wire_event().and_then(|wire| wire.raw_sse_frame()) { + raw_response.extend_from_slice(frame); + } + } + let outbound = if let Some(server) = http_server { + server.verify().await; + let requests = server.received_requests().await.expect("upstream requests"); + captured_request_body(&requests[0]) + } else { + websocket_server + .expect("WebSocket server") + .await + .expect("server task") + }; + assert_eq!( + outbound.get("service_tier").and_then(Value::as_str), + requested + ); + assert!(!observations.is_empty()); + for observation in &observations { + assert_eq!( + observation.service_tier(), + requested, + "WebSocket={use_websocket}" + ); + } + let metadata: Value = serde_json::from_str( + observations + .last() + .expect("final observation") + .provider_metadata() + .expect("metadata") + .as_json(), + ) + .expect("metadata JSON"); + assert_eq!( + metadata.get("requestedServiceTier").and_then(Value::as_str), + requested + ); + assert_eq!( + metadata.get("upstreamServiceTier").and_then(Value::as_str), + reported + ); + assert_eq!( + costs, + vec![expected_cost], + "WebSocket={use_websocket}, requested={requested:?}, reported={reported:?}" + ); + assert_eq!(raw_response, response_frames.as_bytes()); + } + } +} + #[tokio::test] async fn response_failed_before_semantic_output_is_atomic_and_persists_quota_lock() { let store = Arc::new(MemoryAccountStore::default()); diff --git a/backend/crates/providers/openai/tests/transport/canonical.rs b/backend/crates/providers/openai/tests/transport/canonical.rs index 5e6775293..832e8c540 100644 --- a/backend/crates/providers/openai/tests/transport/canonical.rs +++ b/backend/crates/providers/openai/tests/transport/canonical.rs @@ -219,7 +219,7 @@ fn decoder_should_emit_calculated_cost_for_complete_known_model_usage() { } #[test] -fn decoder_should_prefer_response_service_tier_over_requested_tier_for_billing() { +fn decoder_should_bill_requested_service_tier_despite_default_response() { let body = concat!( "event: response.created\n", "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_fast_cost\",\"model\":\"gpt-5.4\",\"service_tier\":\"default\"}}\n\n", @@ -236,12 +236,12 @@ fn decoder_should_prefer_response_service_tier_over_requested_tier_for_billing() assert!(canonical_facts(&events).into_iter().any(|event| matches!( event, GatewayEvent::CalculatedCost(cost) - if cost.total().amount().scaled() == 3_437_500 + if cost.total().amount().scaled() == 6_875_000 ))); } #[test] -fn decoder_should_use_response_service_tier_when_request_omits_it() { +fn decoder_should_bill_standard_when_request_omits_service_tier() { let body = concat!( "event: response.created\n", "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_response_tier_cost\",\"model\":\"gpt-5.4\",\"service_tier\":\"priority\"}}\n\n", @@ -257,7 +257,7 @@ fn decoder_should_use_response_service_tier_when_request_omits_it() { assert!(canonical_facts(&events).into_iter().any(|event| matches!( event, GatewayEvent::CalculatedCost(cost) - if cost.total().amount().scaled() == 6_875_000 + if cost.total().amount().scaled() == 3_437_500 ))); } @@ -346,7 +346,7 @@ fn decoder_should_fail_closed_for_fixed_block_web_search_content() { } #[test] -fn websocket_decoder_should_prefer_response_service_tier_over_requested_tier_for_billing() { +fn websocket_decoder_should_bill_requested_service_tier_despite_default_response() { let created = websocket_event_to_sse_frame( r#"{"type":"response.created","response":{"id":"resp_ws_fast_cost","model":"gpt-5.4","service_tier":"default"}}"#, ) @@ -366,7 +366,7 @@ fn websocket_decoder_should_prefer_response_service_tier_over_requested_tier_for assert!(canonical_facts(&events).into_iter().any(|event| matches!( event, GatewayEvent::CalculatedCost(cost) - if cost.total().amount().scaled() == 3_437_500 + if cost.total().amount().scaled() == 6_875_000 ))); } @@ -975,14 +975,19 @@ fn pricing_response_cost( } #[test] -fn billing_should_follow_actual_tier_and_only_fall_back_when_it_is_absent() { +fn billing_should_follow_requested_tier_independently_of_response_tier() { for (requested, actual, expected) in [ - (Some("fast"), Some("default"), Some(1_000_000_000)), - (Some("auto"), Some("fast"), Some(2_000_000_000)), - (Some("default"), Some("flex"), Some(500_000_000)), + (Some("priority"), Some("default"), Some(2_000_000_000)), + (Some("fast"), Some("default"), Some(2_000_000_000)), + (Some("auto"), Some("fast"), None), + (Some("default"), Some("flex"), Some(1_000_000_000)), + (Some("flex"), Some("priority"), Some(500_000_000)), (Some("fast"), None, Some(2_000_000_000)), (Some("auto"), None, None), - (Some("default"), Some("future"), None), + (Some("default"), Some("future"), Some(1_000_000_000)), + (Some("future"), Some("default"), None), + (None, Some("priority"), Some(1_000_000_000)), + (None, None, Some(1_000_000_000)), ] { assert_eq!( pricing_response_cost(requested, actual, json!([]), &[]), @@ -1002,8 +1007,8 @@ fn billing_should_add_file_and_web_search_fees_without_tier_markup() { for (tier, expected) in [("default", 1_150_000_000), ("fast", 2_150_000_000)] { assert_eq!( pricing_response_cost( - None, Some(tier), + Some("default"), output.clone(), &[json!({"type": "web_search"})] ), diff --git a/docs/api.md b/docs/api.md index a345ea7f9..81d746164 100644 --- a/docs/api.md +++ b/docs/api.md @@ -884,13 +884,16 @@ Dashboard 的 `accountUsage[]` 由后端提供 `usageWindow`、`metricLabel`、` 和本地用量由 Provider/Admin 投影。前端不得从套餐缺失推断免费套餐,也不得从显示时舍入的百分比推断 触顶。滚动窗口使用相应时间范围的本地用量,独立于 Dashboard 的今日统计范围。 -OpenAI 的 `serviceTier` 只接受上游响应生命周期事件确认的实际 `response.service_tier`;请求里的 -期望档位只保留在 request summary,不能冒充响应事实。计费展示把 `priority`/`fast` 映射为 `Fast`, -`flex` 映射为 `Flex`,缺失或 `default` 映射为 `Default`;未知非空值原样展示。各模型、档位及长上下文 -区间使用明确登记的价格;缺少对应价格时不估算,不以固定倍数兜底。展示的倍率由所选档位与标准档位 -费用之比计算,托管工具调用费不随 Token 档位倍增。 -响应未报告实际档位时,本地费用估算回退到请求档位;该推测不写入响应 `serviceTier`,也不能确认 -上游最终采用了该档位,因此本地费用不能代替官方账单。 +OpenAI Responses 用量记录的 `serviceTier` 与本地费用估算统一采用 Provider 最终发给上游的请求 +`service_tier`,不使用响应档位覆盖或回退。例如发送 `priority`、响应回显 `default` 时,仍显示 +`Fast` 并按 Priority 价格估算。未发送档位时,`serviceTier` 保持缺失,展示与估算按标准档处理。 +计费展示把 `priority`/`fast` 映射为 `Fast`,`flex` 映射为 `Flex`,缺失或 `default`/`standard` +映射为 `Standard`;其他非空值以首字母大写展示。各模型、档位及长上下文区间使用明确登记的价格;缺少对应 +价格时不估算(包括 `auto` 和未知档位),不以固定倍数兜底。展示的倍率由所选档位与标准档位费用之比 +计算,托管工具调用费不随 Token 档位倍增。 +Provider metadata 分别保留 `requestedServiceTier` 与 `upstreamServiceTier` 供诊断;发送给客户端的 +原始 `response.service_tier` 不变。用量中的 Fast 仅表示发送档位,不能证明上游实际加速,本地费用 +估算也不能代替官方账单。该口径仅作用于新记录,不回填历史档位或重算已存储费用。 本地计价规则只保留尚在服务的型号;已过官方关闭日期的型号不再新增本地估价。清理计价规则不删除或 重新计算已存储的历史费用;缺少当前计价规则时,历史总额仍保留,但无法再据此补充费用拆分。 diff --git a/docs/architecture.md b/docs/architecture.md index 6f3ee9aa7..4d6fe74f9 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -427,7 +427,9 @@ observation、调用 metadata;最终失败的关联头不与 opening 身份混 `generate: false` 将连接与上下文准备归类为 `prewarm`,不信任客户端单独声明的同名 metadata。 Store 的共享用量口径排除这些预热记录,账号用量与额度预测复用同一规则;原始请求审计、响应额度 观测和费用事实仍保留,不将未知费用改写为零,也不影响 Client Key 结算账本。 -实际 `service_tier` 只接受上游响应事件确认,不能用请求期望值替代。 +OpenAI Responses 的统计档位与本地费用估算统一使用 Provider 最终发给上游的请求 `service_tier`, +不由响应回显覆盖;未发送档位时保留缺失值,展示与估算按标准档处理。上游响应档位独立保留在 +Provider metadata 的 `upstreamServiceTier`,不改写客户端收到的响应,也不据此断言实际加速效果。 Worker 由各 Bundle 贡献、由 Host 统一监督: