mirror of
https://github.com/zyycn/codex-proxy-rs.git
synced 2026-11-06 08:06:19 +00:00
fix(openai): use outbound service tier for usage and billing
Keep response-reported tiers as diagnostic metadata without overriding outbound-tier usage or cost estimates. Cover HTTP and WebSocket behavior and document the updated accounting policy. Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
@@ -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<String>) -> Self {
|
||||
let service_tier = service_tier.into();
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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))
|
||||
|
||||
@@ -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());
|
||||
|
||||
@@ -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"})]
|
||||
),
|
||||
|
||||
+10
-7
@@ -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 仅表示发送档位,不能证明上游实际加速,本地费用
|
||||
估算也不能代替官方账单。该口径仅作用于新记录,不回填历史档位或重算已存储费用。
|
||||
|
||||
本地计价规则只保留尚在服务的型号;已过官方关闭日期的型号不再新增本地估价。清理计价规则不删除或
|
||||
重新计算已存储的历史费用;缺少当前计价规则时,历史总额仍保留,但无法再据此补充费用拆分。
|
||||
|
||||
@@ -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 统一监督:
|
||||
|
||||
|
||||
Reference in New Issue
Block a user