diff --git a/backend/src/api/admin/accounts_routes/query.rs b/backend/src/api/admin/accounts_routes/query.rs index b7afaaf49..2219d7215 100644 --- a/backend/src/api/admin/accounts_routes/query.rs +++ b/backend/src/api/admin/accounts_routes/query.rs @@ -353,10 +353,12 @@ pub(super) async fn list_current_window_model_usage( let input_tokens = row.input_tokens; let output_tokens = row.output_tokens; let cached_tokens = row.cached_tokens; + let cache_write_tokens = row.cache_write_tokens; let billing_amount_usd = billing::calculate_billing_amount( nonnegative_i64_to_u64(input_tokens), nonnegative_i64_to_u64(output_tokens), nonnegative_i64_to_u64(cached_tokens), + nonnegative_i64_to_u64(cache_write_tokens), &model, row.service_tier.as_deref(), ); diff --git a/backend/src/api/admin/usage_routes.rs b/backend/src/api/admin/usage_routes.rs index 859a028d4..8e3f81235 100644 --- a/backend/src/api/admin/usage_routes.rs +++ b/backend/src/api/admin/usage_routes.rs @@ -121,11 +121,13 @@ pub(crate) struct UsageRecordTokenDetailsData { input_tokens: u64, output_tokens: u64, cached_tokens: u64, + cache_write_tokens: u64, reasoning_tokens: u64, total_tokens: u64, input_tokens_display: String, output_tokens_display: String, cached_tokens_display: String, + cache_write_tokens_display: String, reasoning_tokens_display: String, total_tokens_display: String, } @@ -136,22 +138,26 @@ pub(crate) struct UsageRecordBillingData { input_amount: f64, output_amount: f64, cache_read_amount: f64, + cache_write_amount: f64, standard_amount: f64, total_amount: f64, input_price_per_mtoken: f64, output_price_per_mtoken: f64, cache_read_price_per_mtoken: f64, + cache_write_price_per_mtoken: f64, service_tier: Option, service_tier_display: String, multiplier: f64, input_amount_display: String, output_amount_display: String, cache_read_amount_display: String, + cache_write_amount_display: String, standard_amount_display: String, total_amount_display: String, input_price_display: String, output_price_display: String, cache_read_price_display: String, + cache_write_price_display: String, multiplier_display: String, } @@ -162,6 +168,7 @@ struct UsageRecordSummaryData { input_tokens: String, output_tokens: String, cached_tokens: String, + cache_write_tokens: String, total_tokens: String, average_latency_ms: String, } @@ -175,6 +182,7 @@ struct UsageRecordBreakdownData { input_tokens: String, output_tokens: String, cached_tokens: String, + cache_write_tokens: String, total_tokens: String, total_tokens_value: u64, total_tokens_total: String, @@ -195,6 +203,8 @@ struct UsageRecordTrendPointData { output_tokens_value: u64, cached_tokens: String, cached_tokens_value: u64, + cache_write_tokens: String, + cache_write_tokens_value: u64, total_tokens: String, total_tokens_value: u64, cache_hit_rate: String, @@ -574,6 +584,10 @@ pub(crate) fn usage_token_details(record: &UsageRecord) -> UsageRecordTokenDetai .cached_tokens .and_then(|value| value.try_into().ok()) .unwrap_or(0); + let cache_write_tokens: u64 = record + .cache_write_tokens + .and_then(|value| value.try_into().ok()) + .unwrap_or(0); let reasoning_tokens: u64 = record .reasoning_tokens .and_then(|value| value.try_into().ok()) @@ -584,11 +598,13 @@ pub(crate) fn usage_token_details(record: &UsageRecord) -> UsageRecordTokenDetai input_tokens, output_tokens, cached_tokens, + cache_write_tokens, reasoning_tokens, total_tokens, input_tokens_display: format_number(input_tokens), output_tokens_display: format_number(output_tokens), cached_tokens_display: format_compact_number(cached_tokens), + cache_write_tokens_display: format_compact_number(cache_write_tokens), reasoning_tokens_display: format_number(reasoning_tokens), total_tokens_display: format_number(total_tokens), } @@ -605,28 +621,33 @@ pub(crate) fn usage_billing( tokens.input_tokens, tokens.output_tokens, tokens.cached_tokens, + tokens.cache_write_tokens, )?; Some(UsageRecordBillingData { input_amount: billing.input_amount, output_amount: billing.output_amount, cache_read_amount: billing.cache_read_amount, + cache_write_amount: billing.cache_write_amount, standard_amount: billing.standard_amount, total_amount: billing.total_amount, input_price_per_mtoken: billing.input_price_per_mtoken, output_price_per_mtoken: billing.output_price_per_mtoken, cache_read_price_per_mtoken: billing.cache_read_price_per_mtoken, + cache_write_price_per_mtoken: billing.cache_write_price_per_mtoken, service_tier: billing.service_tier, service_tier_display: billing.service_tier_display, multiplier: billing.multiplier, input_amount_display: format_billing_amount(billing.input_amount), output_amount_display: format_billing_amount(billing.output_amount), cache_read_amount_display: format_billing_amount(billing.cache_read_amount), + cache_write_amount_display: format_billing_amount(billing.cache_write_amount), standard_amount_display: format_billing_amount(billing.standard_amount), total_amount_display: format_billing_amount(billing.total_amount), input_price_display: format_token_price(billing.input_price_per_mtoken), output_price_display: format_token_price(billing.output_price_per_mtoken), cache_read_price_display: format_token_price(billing.cache_read_price_per_mtoken), + cache_write_price_display: format_token_price(billing.cache_write_price_per_mtoken), multiplier_display: format_multiplier(billing.multiplier), }) } @@ -638,6 +659,7 @@ impl From for UsageRecordSummaryData { input_tokens: format_tokens(summary.input_tokens), output_tokens: format_tokens(summary.output_tokens), cached_tokens: format_tokens(summary.cached_tokens), + cache_write_tokens: format_tokens(summary.cache_write_tokens), total_tokens: format_tokens(summary.total_tokens), average_latency_ms: format_duration_ms_f64(summary.average_latency_ms), } @@ -659,6 +681,7 @@ impl UsageRecordBreakdownData { input_tokens: format_tokens(item.input_tokens), output_tokens: format_tokens(item.output_tokens), cached_tokens: format_tokens(item.cached_tokens), + cache_write_tokens: format_tokens(item.cache_write_tokens), total_tokens: format_tokens(item.total_tokens), total_tokens_value: item.total_tokens, total_tokens_total: format_tokens(total_tokens_total), @@ -698,6 +721,8 @@ impl From for UsageRecordTrendPointData { output_tokens_value: point.output_tokens, cached_tokens: format_tokens(point.cached_tokens), cached_tokens_value: point.cached_tokens, + cache_write_tokens: format_tokens(point.cache_write_tokens), + cache_write_tokens_value: point.cache_write_tokens, total_tokens: format_tokens(point.total_tokens), total_tokens_value: point.total_tokens, cache_hit_rate: format_percent(cache_hit_rate_value), diff --git a/backend/src/api/client/errors.rs b/backend/src/api/client/errors.rs index f455a061a..c699a8e10 100644 --- a/backend/src/api/client/errors.rs +++ b/backend/src/api/client/errors.rs @@ -83,7 +83,7 @@ const MODEL_NOT_FOUND_ERROR: OpenAiErrorKind = OpenAiErrorKind { }; const HISTORY_UNAVAILABLE_ERROR: OpenAiErrorKind = OpenAiErrorKind { error_type: "invalid_request_error", - code: "previous_response_unavailable", + code: "previous_response_not_found", }; const CONTINUATION_BUSY_ERROR: OpenAiErrorKind = OpenAiErrorKind { error_type: "server_error", diff --git a/backend/src/api/client/responses/mod.rs b/backend/src/api/client/responses/mod.rs index 11c696658..fb84d322e 100644 --- a/backend/src/api/client/responses/mod.rs +++ b/backend/src/api/client/responses/mod.rs @@ -19,7 +19,13 @@ use crate::{ api::AppState, api::middleware::request_id::{ClientIp, RequestId}, dispatch::{errors::ResponseDispatchError, service::ResponseDispatchStream}, - upstream::openai::protocol::{responses::CodexResponsesRequest, sse::sse_body_has_done}, + upstream::openai::protocol::{ + responses::{ + CodexResponsesRequest, WS_REQUEST_HEADER_RESPONSES_LITE_CLIENT_METADATA_KEY, + X_OPENAI_INTERNAL_CODEX_RESPONSES_LITE_HEADER, X_OPENAI_MEMGEN_REQUEST_HEADER, + }, + sse::sse_body_has_done, + }, }; use self::sse::{done_sse_frame, event_stream_response as sse_event_stream_response}; @@ -103,11 +109,29 @@ pub fn build_codex_request( None => {} } apply_context_header_fields(&mut request, headers); + apply_request_semantic_headers(&mut request, headers); apply_session_fields(&mut request, headers); request.client_user_agent = user_agent_from_headers(headers); request } +fn apply_request_semantic_headers(request: &mut CodexResponsesRequest, headers: &HeaderMap) { + request.responses_lite = header_string(headers, X_OPENAI_INTERNAL_CODEX_RESPONSES_LITE_HEADER) + .or_else(|| { + request + .client_metadata() + .and_then(Value::as_object) + .and_then(|metadata| { + metadata.get(WS_REQUEST_HEADER_RESPONSES_LITE_CLIENT_METADATA_KEY) + }) + .and_then(Value::as_str) + .map(str::trim) + .filter(|value| !value.is_empty()) + .map(ToString::to_string) + }); + request.memgen_request = header_string(headers, X_OPENAI_MEMGEN_REQUEST_HEADER); +} + fn apply_session_fields(request: &mut CodexResponsesRequest, headers: &HeaderMap) { request.client_conversation_id = session_context_string(request.body(), "conversation_id") .or_else(|| header_string(headers, "conversation-id")) diff --git a/backend/src/api/client/responses/websocket.rs b/backend/src/api/client/responses/websocket.rs index 31f26e92f..2db821bcb 100644 --- a/backend/src/api/client/responses/websocket.rs +++ b/backend/src/api/client/responses/websocket.rs @@ -545,7 +545,7 @@ fn parse_response_create(payload: &str) -> Result, ResponseCr Some(Value::String(message_type)) if message_type == "response.create" => {} _ => return Err(ResponseCreateFrameError::UnsupportedType), } - if body.get("stream").and_then(Value::as_bool) != Some(true) { + if matches!(body.get("stream"), Some(value) if value.as_bool() != Some(true)) { return Err(ResponseCreateFrameError::StreamingRequired); } Ok(body) diff --git a/backend/src/dispatch/lifecycle/request.rs b/backend/src/dispatch/lifecycle/request.rs index 6e9c97aa3..a5a1a9d48 100644 --- a/backend/src/dispatch/lifecycle/request.rs +++ b/backend/src/dispatch/lifecycle/request.rs @@ -60,7 +60,7 @@ pub(in crate::dispatch) async fn enter_request( let catalog = dependencies.models.catalog().await; let display_model = catalog.resolve_model_id(requested_model); request.set_model(display_model.clone()); - request.set_stream(mode.stream()); + debug_assert_eq!(request.stream(), mode.stream()); let compact = request.semantics().compact; let tuple_schema = request.tuple_schema.clone(); diff --git a/backend/src/infra/database.rs b/backend/src/infra/database.rs index e38f81c90..5e06b31f9 100644 --- a/backend/src/infra/database.rs +++ b/backend/src/infra/database.rs @@ -8,11 +8,18 @@ use tracing::info; const MIGRATION_LOCK_KEY: i64 = i64::from_be_bytes(*b"CPR-MIGR"); -const MIGRATIONS: &[Migration] = &[Migration { - version: 1, - name: "initial", - sql: include_str!("migrations/0001_initial.sql"), -}]; +const MIGRATIONS: &[Migration] = &[ + Migration { + version: 1, + name: "initial", + sql: include_str!("migrations/0001_initial.sql"), + }, + Migration { + version: 2, + name: "cache_write_tokens", + sql: include_str!("migrations/0002_cache_write_tokens.sql"), + }, +]; struct Migration { version: i64, diff --git a/backend/src/infra/migrations/0002_cache_write_tokens.sql b/backend/src/infra/migrations/0002_cache_write_tokens.sql new file mode 100644 index 000000000..da0f750ec --- /dev/null +++ b/backend/src/infra/migrations/0002_cache_write_tokens.sql @@ -0,0 +1,7 @@ +alter table usage_records + add column cache_write_tokens bigint + check (cache_write_tokens is null or cache_write_tokens >= 0); + +alter table request_time_buckets + add column cache_write_tokens bigint not null default 0 + check (cache_write_tokens >= 0); diff --git a/backend/src/telemetry/account_usage/query.rs b/backend/src/telemetry/account_usage/query.rs index e312a3181..ab655a9f6 100644 --- a/backend/src/telemetry/account_usage/query.rs +++ b/backend/src/telemetry/account_usage/query.rs @@ -55,6 +55,7 @@ impl AccountUsageQueryService { nonnegative_i64_to_u64(bucket.totals.input_tokens), nonnegative_i64_to_u64(bucket.totals.output_tokens), nonnegative_i64_to_u64(bucket.totals.cached_tokens), + nonnegative_i64_to_u64(bucket.totals.cache_write_tokens), &bucket.model, bucket.service_tier.as_deref(), ); @@ -159,6 +160,7 @@ pub struct AccountUsageTimeBucket { pub input_tokens: i64, pub output_tokens: i64, pub cached_tokens: i64, + pub cache_write_tokens: i64, pub first_token_latency_sum: i64, pub first_token_latency_count: i64, pub latency_sum: i64, @@ -178,6 +180,7 @@ impl From for AccountUsageTimeBucket { input_tokens: record.input_tokens, output_tokens: record.output_tokens, cached_tokens: record.cached_tokens, + cache_write_tokens: record.cache_write_tokens, first_token_latency_sum: record.first_token_latency_sum, first_token_latency_count: record.first_token_latency_count, latency_sum: record.latency_sum, diff --git a/backend/src/telemetry/billing.rs b/backend/src/telemetry/billing.rs index 34ba0c875..50c50da10 100644 --- a/backend/src/telemetry/billing.rs +++ b/backend/src/telemetry/billing.rs @@ -10,6 +10,7 @@ struct ModelPricing { output_price_per_mtoken_priority: f64, cache_read_price_per_mtoken: f64, cache_read_price_per_mtoken_priority: f64, + cache_write_multiplier: f64, long_input_price_per_mtoken: f64, long_input_price_per_mtoken_priority: f64, long_output_price_per_mtoken: f64, @@ -27,6 +28,7 @@ impl ModelPricing { output_price_per_mtoken_priority: 0.0, cache_read_price_per_mtoken: 0.0, cache_read_price_per_mtoken_priority: 0.0, + cache_write_multiplier: 0.0, long_input_price_per_mtoken: 0.0, long_input_price_per_mtoken_priority: 0.0, long_output_price_per_mtoken: 0.0, @@ -41,6 +43,11 @@ impl ModelPricing { self } + const fn with_cache_write(mut self, multiplier: f64) -> Self { + self.cache_write_multiplier = multiplier; + self + } + const fn with_priority(mut self, input: f64, output: f64, cache_read: f64) -> Self { self.input_price_per_mtoken_priority = input; self.output_price_per_mtoken_priority = output; @@ -78,24 +85,28 @@ const MODEL_PRICING_RULES: &[ModelPricingRule] = &[ model: "gpt-5.6-sol", pricing: ModelPricing::new(5.0, 30.0) .with_cache(0.5) + .with_cache_write(1.25) .with_priority(10.0, 60.0, 1.0), }, ModelPricingRule { model: "gpt-5.6-terra", pricing: ModelPricing::new(2.5, 15.0) .with_cache(0.25) + .with_cache_write(1.25) .with_priority(5.0, 30.0, 0.5), }, ModelPricingRule { model: "gpt-5.6-luna", pricing: ModelPricing::new(1.0, 6.0) .with_cache(0.1) + .with_cache_write(1.25) .with_priority(2.0, 12.0, 0.2), }, ModelPricingRule { model: "gpt-5.6", pricing: ModelPricing::new(5.0, 30.0) .with_cache(0.5) + .with_cache_write(1.25) .with_priority(10.0, 60.0, 1.0), }, ModelPricingRule { @@ -181,6 +192,8 @@ pub(crate) struct BillingBreakdown { pub output_amount: f64, /// 缓存读取金额。 pub cache_read_amount: f64, + /// 缓存写入金额。 + pub cache_write_amount: f64, /// 应用服务档位倍率前的标准金额。 pub standard_amount: f64, /// 最终计费金额。 @@ -191,6 +204,8 @@ pub(crate) struct BillingBreakdown { pub output_price_per_mtoken: f64, /// 缓存读取单价，美元 / 1M token。 pub cache_read_price_per_mtoken: f64, + /// 缓存写入单价，美元 / 1M token。 + pub cache_write_price_per_mtoken: f64, /// 服务档位。 pub service_tier: Option, /// 服务档位倍率。 @@ -202,16 +217,24 @@ pub(crate) fn calculate_billing( input_tokens: u64, output_tokens: u64, cached_tokens: u64, + cache_write_tokens: u64, model: &str, service_tier: Option<&str>, ) -> BillingBreakdown { let pricing = model_pricing(model).unwrap_or_else(|| { - highest_pricing_for_request(input_tokens, output_tokens, cached_tokens, service_tier) + highest_pricing_for_request( + input_tokens, + output_tokens, + cached_tokens, + cache_write_tokens, + service_tier, + ) }); calculate_billing_with_pricing( input_tokens, output_tokens, cached_tokens, + cache_write_tokens, pricing, service_tier, ) @@ -221,6 +244,7 @@ fn calculate_billing_with_pricing( input_tokens: u64, output_tokens: u64, cached_tokens: u64, + cache_write_tokens: u64, pricing: ModelPricing, service_tier: Option<&str>, ) -> BillingBreakdown { @@ -230,6 +254,7 @@ fn calculate_billing_with_pricing( input_tokens, output_tokens, cached_tokens, + cache_write_tokens, pricing, service_tier, is_long, @@ -240,6 +265,7 @@ fn calculate_billing_with_pricing_for_context( input_tokens: u64, output_tokens: u64, cached_tokens: u64, + cache_write_tokens: u64, pricing: ModelPricing, service_tier: Option<&str>, is_long: bool, @@ -277,26 +303,38 @@ fn calculate_billing_with_pricing_for_context( } let cached_tokens = cached_tokens.min(input_tokens); - let uncached_input_tokens = if cache_read_price > 0.0 { - input_tokens - cached_tokens + let billed_cache_read_tokens = if cache_read_price > 0.0 { + cached_tokens } else { - input_tokens + 0 }; + let cache_write_price = input_price * pricing.cache_write_multiplier; + let billed_cache_write_tokens = if cache_write_price > 0.0 { + cache_write_tokens.min(input_tokens - billed_cache_read_tokens) + } else { + 0 + }; + let uncached_input_tokens = input_tokens + .saturating_sub(billed_cache_read_tokens) + .saturating_sub(billed_cache_write_tokens); let input_amount = uncached_input_tokens as f64 / 1_000_000.0 * input_price; - let cache_read_amount = cached_tokens as f64 / 1_000_000.0 * cache_read_price; + let cache_read_amount = billed_cache_read_tokens as f64 / 1_000_000.0 * cache_read_price; + let cache_write_amount = billed_cache_write_tokens as f64 / 1_000_000.0 * cache_write_price; let output_amount = output_tokens as f64 / 1_000_000.0 * output_price; - let standard_amount = input_amount + cache_read_amount + output_amount; + let standard_amount = input_amount + cache_read_amount + cache_write_amount + output_amount; let total_amount = standard_amount * tier_multiplier; BillingBreakdown { input_amount, output_amount, cache_read_amount, + cache_write_amount, standard_amount, total_amount, input_price_per_mtoken: input_price, output_price_per_mtoken: output_price, cache_read_price_per_mtoken: cache_read_price, + cache_write_price_per_mtoken: cache_write_price, service_tier: normalize_service_tier(service_tier), tier_multiplier, } @@ -309,6 +347,7 @@ pub(crate) fn calculate_aggregate_billing( input_tokens: u64, output_tokens: u64, cached_tokens: u64, + cache_write_tokens: u64, model: &str, service_tier: Option<&str>, is_long: bool, @@ -318,6 +357,7 @@ pub(crate) fn calculate_aggregate_billing( input_tokens, output_tokens, cached_tokens, + cache_write_tokens, service_tier, is_long, ) @@ -326,6 +366,7 @@ pub(crate) fn calculate_aggregate_billing( input_tokens, output_tokens, cached_tokens, + cache_write_tokens, pricing, service_tier, is_long, @@ -337,6 +378,7 @@ pub(crate) fn calculate_billing_amount( input_tokens: u64, output_tokens: u64, cached_tokens: u64, + cache_write_tokens: u64, model: &str, service_tier: Option<&str>, ) -> f64 { @@ -344,6 +386,7 @@ pub(crate) fn calculate_billing_amount( input_tokens, output_tokens, cached_tokens, + cache_write_tokens, model, service_tier, ) @@ -365,6 +408,7 @@ fn highest_pricing_for_request( input_tokens: u64, output_tokens: u64, cached_tokens: u64, + cache_write_tokens: u64, service_tier: Option<&str>, ) -> ModelPricing { MODEL_PRICING_RULES @@ -374,6 +418,7 @@ fn highest_pricing_for_request( input_tokens, output_tokens, cached_tokens, + cache_write_tokens, highest, service_tier, ) @@ -382,6 +427,7 @@ fn highest_pricing_for_request( input_tokens, output_tokens, cached_tokens, + cache_write_tokens, rule.pricing, service_tier, ) @@ -398,6 +444,7 @@ fn highest_pricing_for_context( input_tokens: u64, output_tokens: u64, cached_tokens: u64, + cache_write_tokens: u64, service_tier: Option<&str>, is_long: bool, ) -> ModelPricing { @@ -408,6 +455,7 @@ fn highest_pricing_for_context( input_tokens, output_tokens, cached_tokens, + cache_write_tokens, highest, service_tier, is_long, @@ -417,6 +465,7 @@ fn highest_pricing_for_context( input_tokens, output_tokens, cached_tokens, + cache_write_tokens, rule.pricing, service_tier, is_long, diff --git a/backend/src/telemetry/buckets/query.rs b/backend/src/telemetry/buckets/query.rs index f9af462fb..074e66c62 100644 --- a/backend/src/telemetry/buckets/query.rs +++ b/backend/src/telemetry/buckets/query.rs @@ -18,6 +18,7 @@ select coalesce(sum(input_tokens), 0)::bigint as input_tokens, coalesce(sum(output_tokens), 0)::bigint as output_tokens, coalesce(sum(cached_tokens), 0)::bigint as cached_tokens, + coalesce(sum(cache_write_tokens), 0)::bigint as cache_write_tokens, coalesce(sum(first_token_latency_sum), 0)::bigint as first_token_latency_sum, coalesce(sum(first_token_latency_count), 0)::bigint as first_token_latency_count, coalesce(sum(latency_sum), 0)::bigint as latency_sum, @@ -36,7 +37,8 @@ select coalesce(sum(success_count + error_count), 0)::bigint as request_count, coalesce(sum(input_tokens), 0)::bigint as input_tokens, coalesce(sum(output_tokens), 0)::bigint as output_tokens, - coalesce(sum(cached_tokens), 0)::bigint as cached_tokens + coalesce(sum(cached_tokens), 0)::bigint as cached_tokens, + coalesce(sum(cache_write_tokens), 0)::bigint as cache_write_tokens from request_time_buckets group by bucket_start, model, service_tier"; @@ -59,6 +61,7 @@ pub struct UsageTimeBucketRecord { pub input_tokens: i64, pub output_tokens: i64, pub cached_tokens: i64, + pub cache_write_tokens: i64, pub first_token_latency_sum: i64, pub first_token_latency_count: i64, pub latency_sum: i64, @@ -81,6 +84,7 @@ pub struct UsageBucketTotals { pub input_tokens: i64, pub output_tokens: i64, pub cached_tokens: i64, + pub cache_write_tokens: i64, } /// 保留期内单个计费桶的用量事实。 @@ -97,6 +101,9 @@ impl UsageBucketTotals { self.input_tokens = self.input_tokens.saturating_add(other.input_tokens); self.output_tokens = self.output_tokens.saturating_add(other.output_tokens); self.cached_tokens = self.cached_tokens.saturating_add(other.cached_tokens); + self.cache_write_tokens = self + .cache_write_tokens + .saturating_add(other.cache_write_tokens); } } @@ -117,6 +124,7 @@ pub struct ModelBucketUsage { pub input_tokens: i64, pub output_tokens: i64, pub cached_tokens: i64, + pub cache_write_tokens: i64, pub last_used_at: Option>, } @@ -163,6 +171,7 @@ impl PgRequestBucketQuery { input_tokens: row.get("input_tokens"), output_tokens: row.get("output_tokens"), cached_tokens: row.get("cached_tokens"), + cache_write_tokens: row.get("cache_write_tokens"), }, }) .collect()) @@ -198,7 +207,8 @@ impl PgRequestBucketQuery { coalesce(sum(success_count + error_count), 0)::bigint as request_count, coalesce(sum(input_tokens), 0)::bigint as input_tokens, coalesce(sum(output_tokens), 0)::bigint as output_tokens, - coalesce(sum(cached_tokens), 0)::bigint as cached_tokens + coalesce(sum(cached_tokens), 0)::bigint as cached_tokens, + coalesce(sum(cache_write_tokens), 0)::bigint as cache_write_tokens from request_time_buckets where account_id in (", ); @@ -223,6 +233,7 @@ impl PgRequestBucketQuery { input_tokens: row.get("input_tokens"), output_tokens: row.get("output_tokens"), cached_tokens: row.get("cached_tokens"), + cache_write_tokens: row.get("cache_write_tokens"), }; let Some(account_windows) = windows_by_account.get(account_id.as_str()) else { continue; @@ -258,6 +269,7 @@ impl PgRequestBucketQuery { coalesce(sum(input_tokens), 0)::bigint as input_tokens, coalesce(sum(output_tokens), 0)::bigint as output_tokens, coalesce(sum(cached_tokens), 0)::bigint as cached_tokens, + coalesce(sum(cache_write_tokens), 0)::bigint as cache_write_tokens, max(bucket_start) as last_used_at from request_time_buckets where model != '__unknown__' and (", @@ -290,6 +302,7 @@ impl PgRequestBucketQuery { input_tokens: row.get("input_tokens"), output_tokens: row.get("output_tokens"), cached_tokens: row.get("cached_tokens"), + cache_write_tokens: row.get("cache_write_tokens"), last_used_at: row.get("last_used_at"), } }) @@ -307,6 +320,7 @@ fn usage_time_bucket_from_row(row: &sqlx::postgres::PgRow) -> UsageTimeBucketRec input_tokens: row.get("input_tokens"), output_tokens: row.get("output_tokens"), cached_tokens: row.get("cached_tokens"), + cache_write_tokens: row.get("cache_write_tokens"), first_token_latency_sum: row.get("first_token_latency_sum"), first_token_latency_count: row.get("first_token_latency_count"), latency_sum: row.get("latency_sum"), diff --git a/backend/src/telemetry/buckets/store.rs b/backend/src/telemetry/buckets/store.rs index 1031704b2..838859729 100644 --- a/backend/src/telemetry/buckets/store.rs +++ b/backend/src/telemetry/buckets/store.rs @@ -53,15 +53,16 @@ impl PgRequestBucketStore { r" insert into request_time_buckets ( bucket_start, provider, account_id, model, service_tier, success_count, - input_tokens, output_tokens, cached_tokens, first_token_latency_sum, + input_tokens, output_tokens, cached_tokens, cache_write_tokens, first_token_latency_sum, first_token_latency_count, latency_sum, latency_count, max_latency_ms, min_latency_ms, updated_at -) values ($1, $2, $3, $4, $5, 1, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15) +) values ($1, $2, $3, $4, $5, 1, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16) on conflict (bucket_start, provider, account_id, model, service_tier) do update set success_count = request_time_buckets.success_count + 1, input_tokens = request_time_buckets.input_tokens + excluded.input_tokens, output_tokens = request_time_buckets.output_tokens + excluded.output_tokens, cached_tokens = request_time_buckets.cached_tokens + excluded.cached_tokens, + cache_write_tokens = request_time_buckets.cache_write_tokens + excluded.cache_write_tokens, first_token_latency_sum = request_time_buckets.first_token_latency_sum + excluded.first_token_latency_sum, first_token_latency_count = request_time_buckets.first_token_latency_count + excluded.first_token_latency_count, latency_sum = request_time_buckets.latency_sum + excluded.latency_sum, @@ -82,6 +83,7 @@ on conflict (bucket_start, provider, account_id, model, service_tier) do update .bind(event.input_tokens.unwrap_or(0)) .bind(event.output_tokens.unwrap_or(0)) .bind(event.cached_tokens.unwrap_or(0)) + .bind(event.cache_write_tokens.unwrap_or(0)) .bind(first_token.unwrap_or(0)) .bind(i64::from(first_token.is_some())) .bind(latency.unwrap_or(0)) @@ -188,6 +190,7 @@ with facts as ( coalesce(input_tokens, 0)::bigint as input_tokens, coalesce(output_tokens, 0)::bigint as output_tokens, coalesce(cached_tokens, 0)::bigint as cached_tokens, + coalesce(cache_write_tokens, 0)::bigint as cache_write_tokens, coalesce(first_token_ms, 0)::bigint as first_token_latency_sum, (first_token_ms is not null)::integer::bigint as first_token_latency_count, coalesce(latency_ms, 0)::bigint as latency_sum, @@ -215,6 +218,7 @@ with facts as ( 0::bigint, 0::bigint, 0::bigint, + 0::bigint, null::bigint from ops_error_logs where created_at >= $1 @@ -226,6 +230,7 @@ with facts as ( sum(input_tokens) as input_tokens, sum(output_tokens) as output_tokens, sum(cached_tokens) as cached_tokens, + sum(cache_write_tokens) as cache_write_tokens, sum(first_token_latency_sum) as first_token_latency_sum, sum(first_token_latency_count) as first_token_latency_count, sum(latency_sum) as latency_sum, @@ -237,13 +242,13 @@ with facts as ( ) insert into request_time_buckets ( bucket_start, provider, account_id, model, service_tier, - success_count, error_count, input_tokens, output_tokens, cached_tokens, + success_count, error_count, input_tokens, output_tokens, cached_tokens, cache_write_tokens, first_token_latency_sum, first_token_latency_count, latency_sum, latency_count, max_latency_ms, min_latency_ms, updated_at ) select bucket_start, provider, account_id, model, service_tier, - success_count, error_count, input_tokens, output_tokens, cached_tokens, + success_count, error_count, input_tokens, output_tokens, cached_tokens, cache_write_tokens, first_token_latency_sum, first_token_latency_count, latency_sum, latency_count, max_latency_ms, min_latency_ms, now() from aggregated", diff --git a/backend/src/telemetry/recorder.rs b/backend/src/telemetry/recorder.rs index 349f5ae60..4ba6a6176 100644 --- a/backend/src/telemetry/recorder.rs +++ b/backend/src/telemetry/recorder.rs @@ -321,6 +321,7 @@ fn lift_success_fact_fields(event: &mut UsageRecord, metadata: &mut Value) { event.input_tokens = value_nonnegative_i64(usage, "inputTokens"); event.output_tokens = value_nonnegative_i64(usage, "outputTokens"); event.cached_tokens = value_nonnegative_i64(usage, "cachedTokens"); + event.cache_write_tokens = value_nonnegative_i64(usage, "cacheWriteTokens"); event.reasoning_tokens = value_nonnegative_i64(usage, "reasoningTokens"); } diff --git a/backend/src/telemetry/usage/insights.rs b/backend/src/telemetry/usage/insights.rs index b7f2330a9..69b3bec72 100644 --- a/backend/src/telemetry/usage/insights.rs +++ b/backend/src/telemetry/usage/insights.rs @@ -21,7 +21,7 @@ success_terminals as ( client_api_key_id, provider, account_id, transport, model_name, null::text as failure_class, false as is_client_cancelled, false as is_caller_error, latency_ms, first_token_ms, billing_model, - service_tier, input_tokens, output_tokens, cached_tokens + service_tier, input_tokens, output_tokens, cached_tokens, cache_write_tokens from ( select case when request_id is null then 'event:' || id else 'request:' || request_id end as request_key, @@ -31,7 +31,7 @@ success_terminals as ( coalesce(nullif(upstream_model, ''), model) as model_name, latency_ms, first_token_ms, coalesce(nullif(upstream_model, ''), model) as billing_model, - service_tier, input_tokens, output_tokens, cached_tokens + service_tier, input_tokens, output_tokens, cached_tokens, cache_write_tokens from usage_records where created_at >= $1 and created_at < $2 ) candidates @@ -45,7 +45,7 @@ error_terminals as ( latency_ms, null::bigint as first_token_ms, null::text as billing_model, null::text as service_tier, null::bigint as input_tokens, null::bigint as output_tokens, - null::bigint as cached_tokens + null::bigint as cached_tokens, null::bigint as cache_write_tokens from ( select case when request_id is null then 'event:' || id else 'request:' || request_id end as request_key, @@ -660,7 +660,8 @@ async fn load_cost( )::bigint as cache_hit_count,\n\ coalesce(sum(input_tokens), 0)::bigint as input_tokens,\n\ coalesce(sum(output_tokens), 0)::bigint as output_tokens,\n\ - coalesce(sum(least(coalesce(cached_tokens, 0), coalesce(input_tokens, 0))), 0)::bigint as cached_tokens\n\ + coalesce(sum(least(coalesce(cached_tokens, 0), coalesce(input_tokens, 0))), 0)::bigint as cached_tokens,\n\ + coalesce(sum(least(coalesce(cache_write_tokens, 0), coalesce(input_tokens, 0))), 0)::bigint as cache_write_tokens\n\ from usage_records\n\ where created_at >= $1 and created_at < $2\n\ group by bucket_start, billing_model, service_tier, is_long\n\ @@ -735,7 +736,8 @@ async fn load_diagnostic_costs( )::bigint as cache_hit_count,\n\ coalesce(sum(input_tokens), 0)::bigint as input_tokens,\n\ coalesce(sum(output_tokens), 0)::bigint as output_tokens,\n\ - coalesce(sum(least(coalesce(cached_tokens, 0), coalesce(input_tokens, 0))), 0)::bigint as cached_tokens\n\ + coalesce(sum(least(coalesce(cached_tokens, 0), coalesce(input_tokens, 0))), 0)::bigint as cached_tokens,\n\ + coalesce(sum(least(coalesce(cache_write_tokens, 0), coalesce(input_tokens, 0))), 0)::bigint as cache_write_tokens\n\ from {source}\n\ where terminal_requests.is_success\n\ group by name, billing_model, service_tier, is_long" @@ -764,6 +766,7 @@ struct CostSample { input_tokens: u64, output_tokens: u64, cached_tokens: u64, + cache_write_tokens: u64, estimated_cost: f64, standard_cost: f64, } @@ -776,6 +779,7 @@ struct CostAccumulator { input_tokens: u64, output_tokens: u64, cached_tokens: u64, + cache_write_tokens: u64, estimated_cost: f64, standard_cost: f64, } @@ -790,6 +794,9 @@ impl CostAccumulator { self.input_tokens = self.input_tokens.saturating_add(sample.input_tokens); self.output_tokens = self.output_tokens.saturating_add(sample.output_tokens); self.cached_tokens = self.cached_tokens.saturating_add(sample.cached_tokens); + self.cache_write_tokens = self + .cache_write_tokens + .saturating_add(sample.cache_write_tokens); self.estimated_cost += sample.estimated_cost; self.standard_cost += sample.standard_cost; } @@ -803,12 +810,14 @@ fn cost_sample(row: &sqlx::postgres::PgRow) -> CostSample { let input_tokens = nonnegative(row.get("input_tokens")); let output_tokens = nonnegative(row.get("output_tokens")); let cached_tokens = nonnegative(row.get("cached_tokens")).min(input_tokens); + let cache_write_tokens = nonnegative(row.get("cache_write_tokens")).min(input_tokens); let billing_model: String = row.get("billing_model"); let service_tier: Option = row.get("service_tier"); let billing = calculate_aggregate_billing( input_tokens, output_tokens, cached_tokens, + cache_write_tokens, &billing_model, service_tier.as_deref(), row.get("is_long"), @@ -820,6 +829,7 @@ fn cost_sample(row: &sqlx::postgres::PgRow) -> CostSample { input_tokens, output_tokens, cached_tokens, + cache_write_tokens, estimated_cost: billing.total_amount, standard_cost: billing.standard_amount, } diff --git a/backend/src/telemetry/usage/query.rs b/backend/src/telemetry/usage/query.rs index 1fcb2a23f..1e0c4602e 100644 --- a/backend/src/telemetry/usage/query.rs +++ b/backend/src/telemetry/usage/query.rs @@ -91,11 +91,13 @@ pub(crate) struct UsageRecordBilling { pub input_amount: f64, pub output_amount: f64, pub cache_read_amount: f64, + pub cache_write_amount: f64, pub standard_amount: f64, pub total_amount: f64, pub input_price_per_mtoken: f64, pub output_price_per_mtoken: f64, pub cache_read_price_per_mtoken: f64, + pub cache_write_price_per_mtoken: f64, pub service_tier: Option, pub service_tier_display: String, pub multiplier: f64, @@ -278,8 +280,9 @@ pub(crate) fn usage_record_billing( input_tokens: u64, output_tokens: u64, cached_tokens: u64, + cache_write_tokens: u64, ) -> Option { - if input_tokens == 0 && output_tokens == 0 && cached_tokens == 0 { + if input_tokens == 0 && output_tokens == 0 && cached_tokens == 0 && cache_write_tokens == 0 { return None; } @@ -293,6 +296,7 @@ pub(crate) fn usage_record_billing( input_tokens, output_tokens, cached_tokens, + cache_write_tokens, model, service_tier, ); @@ -301,11 +305,13 @@ pub(crate) fn usage_record_billing( input_amount: breakdown.input_amount, output_amount: breakdown.output_amount, cache_read_amount: breakdown.cache_read_amount, + cache_write_amount: breakdown.cache_write_amount, standard_amount: breakdown.standard_amount, total_amount: breakdown.total_amount, input_price_per_mtoken: breakdown.input_price_per_mtoken, output_price_per_mtoken: breakdown.output_price_per_mtoken, cache_read_price_per_mtoken: breakdown.cache_read_price_per_mtoken, + cache_write_price_per_mtoken: breakdown.cache_write_price_per_mtoken, service_tier: breakdown.service_tier.clone(), service_tier_display: breakdown .service_tier @@ -346,6 +352,7 @@ select coalesce(sum(input_tokens), 0)::bigint as input_tokens, coalesce(sum(output_tokens), 0)::bigint as output_tokens, coalesce(sum(cached_tokens), 0)::bigint as cached_tokens, + coalesce(sum(cache_write_tokens), 0)::bigint as cache_write_tokens, avg(latency_ms::double precision) as average_latency_ms from usage_records", ); @@ -354,11 +361,13 @@ from usage_records", let input_tokens = nonnegative(row.get("input_tokens")); let output_tokens = nonnegative(row.get("output_tokens")); let cached_tokens = nonnegative(row.get("cached_tokens")); + let cache_write_tokens = nonnegative(row.get("cache_write_tokens")); Ok(UsageRecordSummary { total_requests: nonnegative(row.get("total_requests")), input_tokens, output_tokens, cached_tokens, + cache_write_tokens, total_tokens: input_tokens + output_tokens, average_latency_ms: row.get("average_latency_ms"), }) @@ -448,6 +457,7 @@ pub(super) async fn usage_breakdown( coalesce(sum(input_tokens), 0)::bigint as input_tokens, coalesce(sum(output_tokens), 0)::bigint as output_tokens, coalesce(sum(cached_tokens), 0)::bigint as cached_tokens, + coalesce(sum(cache_write_tokens), 0)::bigint as cache_write_tokens, coalesce(sum(latency_ms), 0)::bigint as latency_sum, count(latency_ms) as latency_count from usage_records", @@ -462,6 +472,7 @@ from usage_records", let input_tokens = nonnegative(row.get("input_tokens")); let output_tokens = nonnegative(row.get("output_tokens")); let cached_tokens = nonnegative(row.get("cached_tokens")); + let cache_write_tokens = nonnegative(row.get("cache_write_tokens")); let request_count = nonnegative(row.get("request_count")); let latency_sum = nonnegative(row.get("latency_sum")); let latency_count = nonnegative(row.get("latency_count")); @@ -469,6 +480,7 @@ from usage_records", input_tokens, output_tokens, cached_tokens, + cache_write_tokens, &row.get::("billing_model"), row.get::, _>("service_tier").as_deref(), ); @@ -480,6 +492,7 @@ from usage_records", input_tokens, output_tokens, cached_tokens, + cache_write_tokens, billing_amount, latency_sum, latency_count, @@ -512,6 +525,7 @@ select coalesce(sum(input_tokens), 0)::bigint as input_tokens, coalesce(sum(output_tokens), 0)::bigint as output_tokens, coalesce(sum(cached_tokens), 0)::bigint as cached_tokens, + coalesce(sum(cache_write_tokens), 0)::bigint as cache_write_tokens, coalesce(sum(latency_ms), 0)::bigint as latency_sum, count(latency_ms) as latency_count from usage_records", @@ -525,23 +539,26 @@ from usage_records", let input_tokens = nonnegative(row.get("input_tokens")); let output_tokens = nonnegative(row.get("output_tokens")); let cached_tokens = nonnegative(row.get("cached_tokens")); + let cache_write_tokens = nonnegative(row.get("cache_write_tokens")); let billing_amount = usage_breakdown_billing_amount( input_tokens, output_tokens, cached_tokens, + cache_write_tokens, &row.get::("billing_model"), row.get::, _>("service_tier").as_deref(), ); days.entry(date.clone()) .or_insert_with(|| UsageTrendAccumulator::new(date)) - .push( + .push(UsageTrendSample { input_tokens, output_tokens, cached_tokens, + cache_write_tokens, billing_amount, - nonnegative(row.get("latency_sum")), - nonnegative(row.get("latency_count")), - ); + latency_sum: nonnegative(row.get("latency_sum")), + latency_count: nonnegative(row.get("latency_count")), + }); } let mut points = days .into_values() @@ -576,6 +593,7 @@ pub(super) fn usage_record_from_row(row: &sqlx::postgres::PgRow) -> UsageRecord input_tokens: row.get("input_tokens"), output_tokens: row.get("output_tokens"), cached_tokens: row.get("cached_tokens"), + cache_write_tokens: row.get("cache_write_tokens"), reasoning_tokens: row.get("reasoning_tokens"), message: row.get("message"), metadata: row @@ -593,6 +611,7 @@ fn usage_breakdown_billing_amount( input_tokens: u64, output_tokens: u64, cached_tokens: u64, + cache_write_tokens: u64, model: &str, service_tier: Option<&str>, ) -> f64 { @@ -600,6 +619,7 @@ fn usage_breakdown_billing_amount( input_tokens, output_tokens, cached_tokens, + cache_write_tokens, model, service_tier, ) @@ -616,6 +636,7 @@ struct BreakdownSample { input_tokens: u64, output_tokens: u64, cached_tokens: u64, + cache_write_tokens: u64, billing_amount: f64, latency_sum: u64, latency_count: u64, @@ -638,6 +659,7 @@ impl BreakdownAccumulator { self.item.input_tokens += sample.input_tokens; self.item.output_tokens += sample.output_tokens; self.item.cached_tokens += sample.cached_tokens; + self.item.cache_write_tokens += sample.cache_write_tokens; self.item.total_tokens += sample.input_tokens + sample.output_tokens; self.item.standard_billing_amount += sample.billing_amount; self.item.actual_billing_amount += sample.billing_amount; @@ -659,6 +681,16 @@ struct UsageTrendAccumulator { latency_count: u64, } +struct UsageTrendSample { + input_tokens: u64, + output_tokens: u64, + cached_tokens: u64, + cache_write_tokens: u64, + billing_amount: f64, + latency_sum: u64, + latency_count: u64, +} + impl UsageTrendAccumulator { fn new(date: String) -> Self { Self { @@ -671,23 +703,16 @@ impl UsageTrendAccumulator { } } - fn push( - &mut self, - input: u64, - output: u64, - cached: u64, - billing_amount: f64, - latency_sum: u64, - latency_count: u64, - ) { - self.point.input_tokens += input; - self.point.output_tokens += output; - self.point.cached_tokens += cached; - self.point.total_tokens += input + output; - self.point.standard_billing_amount += billing_amount; - self.point.actual_billing_amount += billing_amount; - self.latency_sum += latency_sum; - self.latency_count += latency_count; + fn push(&mut self, sample: UsageTrendSample) { + self.point.input_tokens += sample.input_tokens; + self.point.output_tokens += sample.output_tokens; + self.point.cached_tokens += sample.cached_tokens; + self.point.cache_write_tokens += sample.cache_write_tokens; + self.point.total_tokens += sample.input_tokens + sample.output_tokens; + self.point.standard_billing_amount += sample.billing_amount; + self.point.actual_billing_amount += sample.billing_amount; + self.latency_sum += sample.latency_sum; + self.latency_count += sample.latency_count; } fn finish(mut self) -> UsageRecordTrendPoint { @@ -703,6 +728,7 @@ pub struct UsageRecordSummary { pub input_tokens: u64, pub output_tokens: u64, pub cached_tokens: u64, + pub cache_write_tokens: u64, pub total_tokens: u64, pub average_latency_ms: Option, } @@ -730,6 +756,7 @@ pub struct UsageRecordBreakdown { pub input_tokens: u64, pub output_tokens: u64, pub cached_tokens: u64, + pub cache_write_tokens: u64, pub total_tokens: u64, pub standard_billing_amount: f64, pub actual_billing_amount: f64, @@ -743,6 +770,7 @@ pub struct UsageRecordTrendPoint { pub input_tokens: u64, pub output_tokens: u64, pub cached_tokens: u64, + pub cache_write_tokens: u64, pub total_tokens: u64, pub standard_billing_amount: f64, pub actual_billing_amount: f64, diff --git a/backend/src/telemetry/usage/store.rs b/backend/src/telemetry/usage/store.rs index fb5a25282..591834fdb 100644 --- a/backend/src/telemetry/usage/store.rs +++ b/backend/src/telemetry/usage/store.rs @@ -24,7 +24,8 @@ select id, request_id, client_api_key_id, kind, route, provider, account_id, model, requested_model, upstream_model, service_tier, status_code, transport, attempt_index, response_id, upstream_request_id, latency_ms, first_token_ms, input_tokens, - output_tokens, cached_tokens, reasoning_tokens, message, metadata_json, created_at + output_tokens, cached_tokens, cache_write_tokens, reasoning_tokens, message, metadata_json, + created_at from usage_records"; #[derive(Debug, Error)] @@ -82,10 +83,11 @@ insert into usage_records ( id, request_id, client_api_key_id, kind, route, provider, account_id, model, requested_model, upstream_model, service_tier, status_code, transport, attempt_index, response_id, upstream_request_id, latency_ms, first_token_ms, input_tokens, - output_tokens, cached_tokens, reasoning_tokens, message, metadata_json, created_at + output_tokens, cached_tokens, cache_write_tokens, reasoning_tokens, message, metadata_json, + created_at ) values ( $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, - $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25 + $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26 )", ) .bind(&event.id) @@ -109,6 +111,7 @@ insert into usage_records ( .bind(event.input_tokens) .bind(event.output_tokens) .bind(event.cached_tokens) + .bind(event.cache_write_tokens) .bind(event.reasoning_tokens) .bind(&event.message) .bind(sqlx::types::Json(&event.metadata)) diff --git a/backend/src/telemetry/usage/types.rs b/backend/src/telemetry/usage/types.rs index 85a2b3cac..1ae072a21 100644 --- a/backend/src/telemetry/usage/types.rs +++ b/backend/src/telemetry/usage/types.rs @@ -65,6 +65,8 @@ pub struct UsageRecord { pub output_tokens: Option, /// 缓存 token。 pub cached_tokens: Option, + /// 缓存写入 token。 + pub cache_write_tokens: Option, /// reasoning token。 pub reasoning_tokens: Option, /// 展示消息。 @@ -107,6 +109,7 @@ impl UsageRecord { input_tokens: None, output_tokens: None, cached_tokens: None, + cache_write_tokens: None, reasoning_tokens: None, message: message.into(), metadata: Value::Object(Map::default()), diff --git a/backend/src/upstream/openai/protocol/events.rs b/backend/src/upstream/openai/protocol/events.rs index 335bf4276..b2c231fec 100644 --- a/backend/src/upstream/openai/protocol/events.rs +++ b/backend/src/upstream/openai/protocol/events.rs @@ -15,6 +15,8 @@ pub struct TokenUsage { pub output_tokens: u64, /// 命中缓存的输入 token 数。 pub cached_tokens: u64, + /// 写入 prompt cache 的输入 token 数。 + pub cache_write_tokens: u64, /// 输出 token 中的 reasoning token 数。 pub reasoning_tokens: u64, /// 图片工具输入 token 数。 @@ -44,6 +46,13 @@ pub fn extract_usage(body: &Value) -> Option { .or_else(|| nested_number_field(usage, &["prompt_tokens_details", "cached_tokens"])) .or_else(|| number_field(usage, "cached_tokens")) .unwrap_or_default(); + let cache_write_tokens = + nested_number_field(usage, &["input_tokens_details", "cache_write_tokens"]) + .or_else(|| { + nested_number_field(usage, &["prompt_tokens_details", "cache_write_tokens"]) + }) + .or_else(|| number_field(usage, "cache_write_tokens")) + .unwrap_or_default(); let reasoning_tokens = nested_number_field(usage, &["output_tokens_details", "reasoning_tokens"]) .or_else(|| { @@ -64,6 +73,7 @@ pub fn extract_usage(body: &Value) -> Option { "prompt_tokens", "completion_tokens", "cached_tokens", + "cache_write_tokens", "reasoning_tokens", "total_tokens", ] @@ -80,6 +90,7 @@ pub fn extract_usage(body: &Value) -> Option { input_tokens, output_tokens, cached_tokens, + cache_write_tokens, reasoning_tokens, image_input_tokens, image_output_tokens, diff --git a/backend/src/upstream/openai/protocol/responses.rs b/backend/src/upstream/openai/protocol/responses.rs index 82388ad2a..ce04c03e3 100644 --- a/backend/src/upstream/openai/protocol/responses.rs +++ b/backend/src/upstream/openai/protocol/responses.rs @@ -13,6 +13,15 @@ const MULTI_AGENT_MODE_OPEN_TAG: &str = ""; const MULTI_AGENT_MODE_CLOSE_TAG: &str = ""; const PROACTIVE_MULTI_AGENT_MODE_PREFIX: &str = "Proactive multi-agent delegation is active."; +/// Codex Responses Lite 的 HTTP 兼容头。 +pub const X_OPENAI_INTERNAL_CODEX_RESPONSES_LITE_HEADER: &str = + "x-openai-internal-codex-responses-lite"; +/// Codex Responses Lite 在 WebSocket `client_metadata` 中的官方投影键。 +pub const WS_REQUEST_HEADER_RESPONSES_LITE_CLIENT_METADATA_KEY: &str = + "ws_request_header_x_openai_internal_codex_responses_lite"; +/// Codex memory consolidation 请求标记。 +pub const X_OPENAI_MEMGEN_REQUEST_HEADER: &str = "x-openai-memgen-request"; + /// Codex Responses 上游请求体。 /// /// 发往上游的 Responses 请求。`body` 持有客户端原始 JSON object，逐字段（含顺序、 @@ -64,6 +73,10 @@ pub struct CodexResponsesRequest { pub version: Option, /// timing metrics 透传头。 pub include_timing_metrics: Option, + /// Responses Lite 请求语义；HTTP 使用 header，WebSocket 使用 client metadata 投影。 + pub responses_lite: Option, + /// Memory consolidation 请求语义；HTTP 与 WebSocket opening 均使用 header。 + pub memgen_request: Option, /// codex window id。 pub codex_window_id: Option, /// 父线程 ID。 @@ -261,6 +274,14 @@ pub fn response_event_signals(event_type: Option<&str>, value: &Value) -> Respon signals.text_output = non_empty_string(value.get("text")); signals.semantic_output = signals.text_output; } + Some("response.refusal.delta") => { + signals.text_output = non_empty_string(value.get("delta")); + signals.semantic_output = signals.text_output; + } + Some("response.refusal.done") => { + signals.text_output = non_empty_string(value.get("refusal")); + signals.semantic_output = signals.text_output; + } Some("response.reasoning_summary_text.delta" | "response.reasoning_text.delta") => { signals.reasoning_output = non_empty_string(value.get("delta")); signals.semantic_output = signals.reasoning_output; @@ -295,6 +316,26 @@ pub fn response_event_signals(event_type: Option<&str>, value: &Value) -> Respon merge_output_signals(&mut signals, output_items_signals(items)); } } + Some(event_type) if event_type.ends_with(".delta") => { + signals.semantic_output = non_empty_semantic_value(value.get("delta")); + } + Some(event_type) if event_type.ends_with(".done") => { + signals.semantic_output = [ + "text", + "refusal", + "arguments", + "input", + "transcript", + "data", + "output", + "result", + ] + .into_iter() + .any(|field| non_empty_semantic_value(value.get(field))); + } + Some(event_type) if is_tool_execution_event(event_type) => { + signals.semantic_output = true; + } _ => {} } signals @@ -358,19 +399,18 @@ fn output_item_signals(item: &Value) -> ResponseEventSignals { signals.semantic_output = signals.text_output; } Some("reasoning") => { - signals.reasoning_output = - non_empty_string(item.get("text")) || non_empty_string(item.get("summary")); + signals.reasoning_output = non_empty_semantic_value(item.get("text")) + || non_empty_semantic_value(item.get("summary")); signals.semantic_output = signals.reasoning_output; } - Some("function_call") => { - signals.semantic_output = non_empty_string(item.get("arguments")); + Some("refusal") => { + signals.text_output = + non_empty_string(item.get("refusal")) || non_empty_string(item.get("text")); + signals.semantic_output = signals.text_output; } - Some("custom_tool_call") => { - signals.semantic_output = non_empty_string(item.get("input")); - } - Some("image_generation_call") => { - signals.semantic_output = - non_empty_string(item.get("result")) || non_empty_string(item.get("image_base64")); + Some(item_type) if item_type.ends_with("_call") => { + // done 的工具调用本身已进入不可安全重试的语义边界；不依赖每种工具的字段表。 + signals.semantic_output = true; } _ => { if let Some(items) = item.get("content").and_then(Value::as_array) { @@ -387,6 +427,30 @@ fn non_empty_string(value: Option<&Value>) -> bool { .is_some_and(|value| !value.is_empty()) } +fn non_empty_semantic_value(value: Option<&Value>) -> bool { + match value { + Some(Value::String(value)) => !value.is_empty(), + Some(Value::Array(value)) => !value.is_empty(), + Some(Value::Object(value)) => !value.is_empty(), + Some(Value::Number(_) | Value::Bool(_)) => true, + Some(Value::Null) | None => false, + } +} + +fn is_tool_execution_event(event_type: &str) -> bool { + let Some((call_type, phase)) = event_type + .strip_prefix("response.") + .and_then(|event| event.rsplit_once('.')) + else { + return false; + }; + call_type.ends_with("_call") + && matches!( + phase, + "in_progress" | "searching" | "interpreting" | "completed" | "failed" + ) +} + fn complete_sse_body_prefix(body: &str) -> Option<&str> { let lf_end = body.rfind("\n\n").map(|index| index + 2); let crlf_end = body.rfind("\r\n\r\n").map(|index| index + 4); @@ -669,6 +733,8 @@ impl CodexResponsesRequest { beta_features: None, version: None, include_timing_metrics: None, + responses_lite: None, + memgen_request: None, codex_window_id: None, parent_thread_id: None, previous_response_scope: None, diff --git a/backend/src/upstream/openai/protocol/websocket.rs b/backend/src/upstream/openai/protocol/websocket.rs index 50304d61f..c7e1fab91 100644 --- a/backend/src/upstream/openai/protocol/websocket.rs +++ b/backend/src/upstream/openai/protocol/websocket.rs @@ -76,6 +76,8 @@ struct ResponseCompletedUsage { struct ResponseCompletedInputTokensDetails { #[serde(rename = "cached_tokens")] _cached_tokens: i64, + #[serde(default, rename = "cache_write_tokens")] + _cache_write_tokens: i64, } #[derive(Debug, Deserialize)] diff --git a/backend/src/upstream/openai/transport/client.rs b/backend/src/upstream/openai/transport/client.rs index 614dc2531..c27b39fb5 100644 --- a/backend/src/upstream/openai/transport/client.rs +++ b/backend/src/upstream/openai/transport/client.rs @@ -24,7 +24,8 @@ use tokio_tungstenite::tungstenite::handshake::client::generate_key; use crate::upstream::openai::profile::CodexWireProfileState; use crate::upstream::openai::protocol::events::{extract_sse_usage, retry_after_seconds_from_body}; use crate::upstream::openai::protocol::responses::{ - CodexResponsesRequest, TransportRequirement, transport_requirement, + CodexResponsesRequest, TransportRequirement, + WS_REQUEST_HEADER_RESPONSES_LITE_CLIENT_METADATA_KEY, transport_requirement, }; use crate::upstream::openai::protocol::sse::SseError; use crate::upstream::openai::protocol::websocket::{ @@ -549,10 +550,26 @@ fn is_model_entry(value: &Value) -> bool { fn websocket_upstream_request(request: &CodexResponsesRequest) -> CodexResponsesRequest { let mut request = request.clone(); + project_responses_lite_to_ws_metadata(&mut request); stamp_ws_stream_request_start_ms(&mut request); request } +fn project_responses_lite_to_ws_metadata(request: &mut CodexResponsesRequest) { + let Some(responses_lite) = request.responses_lite.clone() else { + return; + }; + let mut metadata = match request.client_metadata() { + Some(Value::Object(metadata)) => metadata.clone(), + None => Map::new(), + Some(_) => return, + }; + metadata + .entry(WS_REQUEST_HEADER_RESPONSES_LITE_CLIENT_METADATA_KEY.to_string()) + .or_insert(Value::String(responses_lite)); + request.set_client_metadata(Some(Value::Object(metadata))); +} + fn stamp_ws_stream_request_start_ms(request: &mut CodexResponsesRequest) { let mut metadata = match request.client_metadata() { Some(Value::Object(metadata)) => metadata.clone(), diff --git a/backend/src/upstream/openai/transport/client_sse.rs b/backend/src/upstream/openai/transport/client_sse.rs index be28a4474..449eff6a6 100644 --- a/backend/src/upstream/openai/transport/client_sse.rs +++ b/backend/src/upstream/openai/transport/client_sse.rs @@ -268,7 +268,7 @@ impl CodexBackendClient { } let websocket_request = websocket_upstream_request(request); - let headers = self.request_headers_for_http_response(&websocket_request, context)?; + let headers = self.request_headers_for_websocket_response(&websocket_request, context)?; let websocket_create = CodexWebSocketConnection::responses_create_request( &self.base_url, &generate_key(), @@ -284,7 +284,9 @@ impl CodexBackendClient { if let Err(error) = write_websocket_audit_artifact_from_env(&artifact).await { tracing::warn!(error = %error, "Failed to write Codex WebSocket audit artifact"); } - let pool_key = self.websocket_pool_key(request, context, pool_account_id); + let connection_profile = websocket_connection_profile(&headers); + let pool_key = + self.websocket_pool_key(request, context, pool_account_id, &connection_profile); let pool_log_context = pool_key.as_ref().map(WebSocketPoolLogContext::from_key); let pool = self.websocket_pool.as_deref().zip(pool_key); let fast_path_budget = match requirement { @@ -498,17 +500,17 @@ impl CodexBackendClient { request: &CodexResponsesRequest, context: CodexRequestContext<'_>, pool_account_id: Option<&str>, + connection_profile: &str, ) -> Option { let account_id = pool_account_id.or(context.account_id)?; let conversation_id = request .local_conversation_id .as_deref() .or(request.previous_response_id())?; - Some(CodexWebSocketPoolKey::new( - &self.base_url, - account_id, - conversation_id, - )) + Some( + CodexWebSocketPoolKey::new(&self.base_url, account_id, conversation_id) + .with_connection_profile(connection_profile), + ) } /// 获取后端模型目录条目。 @@ -584,6 +586,28 @@ impl CodexBackendClient { HeaderValue::from_str(&subagent)?, ); } + insert_optional_header( + &mut headers, + crate::upstream::openai::protocol::responses::X_OPENAI_INTERNAL_CODEX_RESPONSES_LITE_HEADER, + request.responses_lite.as_deref(), + )?; + insert_optional_header( + &mut headers, + crate::upstream::openai::protocol::responses::X_OPENAI_MEMGEN_REQUEST_HEADER, + request.memgen_request.as_deref(), + )?; + Ok(headers) + } + + fn request_headers_for_websocket_response( + &self, + request: &CodexResponsesRequest, + context: CodexRequestContext<'_>, + ) -> CodexClientResult { + let mut headers = self.request_headers_for_http_response(request, context)?; + headers.remove( + crate::upstream::openai::protocol::responses::X_OPENAI_INTERNAL_CODEX_RESPONSES_LITE_HEADER, + ); Ok(headers) } @@ -678,6 +702,21 @@ impl CodexBackendClient { } } +fn websocket_connection_profile(headers: &HeaderMap) -> String { + [ + "originator", + "user-agent", + crate::upstream::openai::protocol::responses::X_OPENAI_MEMGEN_REQUEST_HEADER, + ] + .map(|name| { + headers + .get(name) + .and_then(|value| value.to_str().ok()) + .unwrap_or_default() + }) + .join("\0") +} + fn http_sse_stream(response: ReqwestResponse) -> CodexBackendSseStream { let stream: CodexBackendSseStream = Box::pin(response.bytes_stream().map_err(CodexClientError::Http)); diff --git a/backend/src/upstream/openai/transport/websocket.rs b/backend/src/upstream/openai/transport/websocket.rs index 7c05459f7..7ea862e57 100644 --- a/backend/src/upstream/openai/transport/websocket.rs +++ b/backend/src/upstream/openai/transport/websocket.rs @@ -484,7 +484,11 @@ pub(crate) async fn prepare_response_create_request_with_pool( .await; }; - match pool.acquire(&key).await { + let allow_profile_mismatch = matches!( + request.continuation(), + WebSocketContinuationRequirement::ConnectionLocal { .. } + ); + match pool.acquire(&key, allow_profile_mismatch).await { WebSocketPoolAcquire::Reused { connection, lease } => { if let WebSocketContinuationRequirement::ConnectionLocal { response_id } = request.continuation() diff --git a/backend/src/upstream/openai/transport/websocket_pool.rs b/backend/src/upstream/openai/transport/websocket_pool.rs index b60937913..c6b37151f 100644 --- a/backend/src/upstream/openai/transport/websocket_pool.rs +++ b/backend/src/upstream/openai/transport/websocket_pool.rs @@ -29,6 +29,7 @@ pub struct CodexWebSocketPoolKey { base_url: String, account_id: String, conversation_id: String, + connection_profile: String, } impl CodexWebSocketPoolKey { @@ -42,9 +43,16 @@ impl CodexWebSocketPoolKey { base_url: base_url.into().trim_end_matches('/').to_string(), account_id: account_id.into(), conversation_id: conversation_id.into(), + connection_profile: String::new(), } } + /// 区分实际 WebSocket opening 画像，防止复用旧 UA 或不同握手语义的连接。 + pub(crate) fn with_connection_profile(mut self, connection_profile: impl Into) -> Self { + self.connection_profile = connection_profile.into(); + self + } + pub(crate) fn account_id(&self) -> &str { &self.account_id } @@ -58,8 +66,15 @@ impl CodexWebSocketPoolKey { self.base_url.as_str(), self.account_id.as_str(), self.conversation_id.as_str(), + self.connection_profile.as_str(), ]) } + + fn same_logical_connection(&self, other: &Self) -> bool { + self.base_url == other.base_url + && self.account_id == other.account_id + && self.conversation_id == other.conversation_id + } } /// WebSocket 连接池。 @@ -166,14 +181,28 @@ impl CodexWebSocketPool { drop(self.tasks.spawn(future)); } - pub(crate) async fn acquire(&self, key: &CodexWebSocketPoolKey) -> WebSocketPoolAcquire { + pub(crate) async fn acquire( + &self, + key: &CodexWebSocketPoolKey, + allow_profile_mismatch: bool, + ) -> WebSocketPoolAcquire { let mut expired_connection = None; let acquire = { let mut state = self.inner.lock().await; if !self.config.enabled || state.shutting_down { return WebSocketPoolAcquire::Bypass(WebSocketPoolBypassReason::Disabled); } - match state.slots.get(key) { + let key = if allow_profile_mismatch && !state.slots.contains_key(key) { + state + .slots + .keys() + .find(|candidate| candidate.same_logical_connection(key)) + .cloned() + .unwrap_or_else(|| key.clone()) + } else { + key.clone() + }; + match state.slots.get(&key) { Some(WebSocketPoolSlot::Busy(_)) => { return WebSocketPoolAcquire::Bypass(WebSocketPoolBypassReason::Busy); } @@ -184,7 +213,7 @@ impl CodexWebSocketPool { }); } Some(WebSocketPoolSlot::Idle(_)) => { - let Some(WebSocketPoolSlot::Idle(connection)) = state.slots.remove(key) else { + let Some(WebSocketPoolSlot::Idle(connection)) = state.slots.remove(&key) else { return WebSocketPoolAcquire::Bypass(WebSocketPoolBypassReason::Busy); }; // 零成本探活：后台 pump 已实时感知连接死亡（RST/Close/EOF/失活）， @@ -210,7 +239,7 @@ impl CodexWebSocketPool { } else { let lease = WebSocketPoolConnectLease::reserve(self.clone(), key.clone()); state.slots.insert( - key.clone(), + key, WebSocketPoolSlot::Connecting(WebSocketPoolConnecting { id: lease.id, started_at: lease.started_at, diff --git a/backend/tests/api/admin/usage_routes.rs b/backend/tests/api/admin/usage_routes.rs index ace97890f..1523563a4 100644 --- a/backend/tests/api/admin/usage_routes.rs +++ b/backend/tests/api/admin/usage_routes.rs @@ -105,18 +105,21 @@ async fn usage_summary_and_trend_expose_only_real_success_metrics() { record.input_tokens = Some(100); record.output_tokens = Some(20); record.cached_tokens = Some(30); + record.cache_write_tokens = Some(10); record.latency_ms = Some(10); usage.append(&record).await.unwrap(); let summary = admin_get(&app, "/api/admin/usage/records/summary").await; assert_eq!(summary["data"]["totalRequests"], "1"); assert_eq!(summary["data"]["totalTokens"], "120"); + assert_eq!(summary["data"]["cacheWriteTokens"], "10"); assert!(summary["data"].get("errorRequests").is_none()); assert!(summary["data"].get("errorRate").is_none()); let trend = admin_get(&app, "/api/admin/usage/records/insights/token-trend").await; assert_eq!(trend["data"][0]["inputTokensValue"], 100); assert_eq!(trend["data"][0]["cacheHitRateValue"], 30.0); + assert_eq!(trend["data"][0]["cacheWriteTokensValue"], 10); assert!(trend["data"][0].get("cacheCreationTokens").is_none()); } @@ -252,6 +255,7 @@ async fn usage_route_applies_official_gpt_5_6_sol_pricing() { record.model = "gpt-5.6-sol-2026-07-01".to_string(); record.input_tokens = Some(1_000_000); record.cached_tokens = Some(200_000); + record.cache_write_tokens = Some(100_000); record.output_tokens = Some(1_000_000); usage.append(&record).await.unwrap(); @@ -260,8 +264,14 @@ async fn usage_route_applies_official_gpt_5_6_sol_pricing() { assert_eq!(billing["inputPricePerMtoken"], 5.0); assert_eq!(billing["cacheReadPricePerMtoken"], 0.5); + assert_eq!(billing["cacheWritePricePerMtoken"], 6.25); + assert_eq!(billing["cacheWriteAmount"], 0.625); assert_eq!(billing["outputPricePerMtoken"], 30.0); - assert_eq!(billing["totalAmount"], 34.1); + assert_eq!(billing["totalAmount"], 34.225); + assert_eq!( + body["data"]["items"][0]["tokenDetails"]["cacheWriteTokens"], + 100_000 + ); } #[tokio::test] diff --git a/backend/tests/api/client/responses/websocket.rs b/backend/tests/api/client/responses/websocket.rs index 3f927a69e..f435dd586 100644 --- a/backend/tests/api/client/responses/websocket.rs +++ b/backend/tests/api/client/responses/websocket.rs @@ -231,6 +231,36 @@ async fn responses_websocket_should_forward_multiple_response_create_requests_on assert_eq!(upstream_requests[1]["type"], "response.create"); } +#[tokio::test] +async fn responses_websocket_should_accept_omitted_stream_without_injecting_it() { + let upstream_listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let upstream_base_url = format!("http://{}", upstream_listener.local_addr().unwrap()); + let upstream = tokio::spawn(async move { + let (stream, _) = upstream_listener.accept().await.unwrap(); + let mut websocket = accept_upstream_websocket(stream).await.unwrap(); + let request = websocket.next().await.unwrap().unwrap(); + let request = serde_json::from_str::(&request.into_text().unwrap()).unwrap(); + send_upstream_response(&mut websocket, "resp_omitted_stream", "ok").await; + request + }); + let (app, api_key, _dir) = + crate::dispatch::service::test_app_with_account(upstream_base_url).await; + let server = spawn_app(app).await; + let mut websocket = connect_responses_websocket(server.address, &api_key).await; + let mut payload: Value = serde_json::from_str(&response_create_payload("hello")).unwrap(); + payload.as_object_mut().unwrap().remove("stream"); + + websocket + .send(Message::Text(payload.to_string().into())) + .await + .unwrap(); + let response = receive_response(&mut websocket).await; + let upstream_request = upstream.await.unwrap(); + + assert_eq!(response.last().unwrap()["type"], "response.completed"); + assert!(upstream_request.get("stream").is_none()); +} + #[tokio::test] async fn responses_websocket_cyber_terminal_should_change_the_immediate_next_request() { let upstream_listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); @@ -918,7 +948,7 @@ async fn responses_websocket_history_retry_should_not_fallback_when_owner_accoun assert_eq!(second.last().unwrap()["type"], "error"); assert_eq!( second.last().unwrap()["error"]["code"], - "previous_response_unavailable" + "previous_response_not_found" ); assert!( fallback.is_none(), diff --git a/backend/tests/bootstrap/tasks/desktop_release_update.rs b/backend/tests/bootstrap/tasks/desktop_release_update.rs index 55dd9a1dc..0758d5fdd 100644 --- a/backend/tests/bootstrap/tasks/desktop_release_update.rs +++ b/backend/tests/bootstrap/tasks/desktop_release_update.rs @@ -14,7 +14,7 @@ async fn desktop_release_update_task_should_start_background_checker() { } #[tokio::test] -async fn desktop_release_update_task_should_record_initial_appcast_observation() { +async fn desktop_release_update_task_should_apply_latest_release_to_wire_profile() { let server = MockServer::start().await; mount_appcast( &server, diff --git a/backend/tests/dispatch/service/responses_recovery.rs b/backend/tests/dispatch/service/responses_recovery.rs index 4c07627e0..181dbe615 100644 --- a/backend/tests/dispatch/service/responses_recovery.rs +++ b/backend/tests/dispatch/service/responses_recovery.rs @@ -1490,7 +1490,7 @@ async fn responses_explicit_previous_response_should_fail_without_transparent_re let captured = upstream.await.unwrap(); assert_eq!(status, StatusCode::BAD_REQUEST); - assert_eq!(body["error"]["code"], "previous_response_unavailable"); + assert_eq!(body["error"]["code"], "previous_response_not_found"); assert!( body["error"]["message"] .as_str() @@ -1714,7 +1714,7 @@ async fn responses_explicit_invalid_reasoning_replay_should_not_retry_without_fu let captured = upstream.await.unwrap(); assert_eq!(status, StatusCode::BAD_REQUEST); - assert_eq!(body["error"]["code"], "previous_response_unavailable"); + assert_eq!(body["error"]["code"], "previous_response_not_found"); assert_eq!(captured.payload["previous_response_id"], "resp_stale"); assert!(!captured.retry_attempted); } @@ -1777,7 +1777,7 @@ async fn responses_stream_structural_events_should_not_commit_before_history_fai let captured = upstream.await.unwrap(); assert_eq!(status, StatusCode::BAD_REQUEST); - assert!(body.contains("previous_response_unavailable")); + assert!(body.contains("previous_response_not_found")); assert!(!body.contains("response.created")); assert_eq!(captured.payload["previous_response_id"], "resp_stale"); assert!(!captured.retry_attempted); @@ -1826,7 +1826,7 @@ async fn responses_http_sse_structural_events_should_not_commit_before_history_f let body = response_text(response).await; assert_eq!(status, StatusCode::BAD_REQUEST); - assert!(body.contains("previous_response_unavailable")); + assert!(body.contains("previous_response_not_found")); assert!(!body.contains("response.created")); } diff --git a/backend/tests/infra/database.rs b/backend/tests/infra/database.rs index b45abda47..acf94db3b 100644 --- a/backend/tests/infra/database.rs +++ b/backend/tests/infra/database.rs @@ -5,7 +5,7 @@ use codex_proxy_rs::infra::database::migrate; use crate::support::storage::init_test_db; #[tokio::test] -async fn postgres_schema_is_terminal_v1_without_unused_storage() { +async fn postgres_schema_applies_versioned_migrations_without_unused_storage() { let (pool, _guard) = init_test_db("schema-terminal-v1").await; let tables = sqlx::query_scalar::<_, String>( "select table_name @@ -41,9 +41,11 @@ async fn postgres_schema_is_terminal_v1_without_unused_storage() { .fetch_all(&pool) .await .unwrap(); - assert_eq!(versions.len(), 1); + assert_eq!(versions.len(), 2); assert_eq!(versions[0].0, 1); assert_eq!(versions[0].1, "initial"); + assert_eq!(versions[1].0, 2); + assert_eq!(versions[1].1, "cache_write_tokens"); assert!(versions.iter().all(|migration| migration.2.len() == 64)); } @@ -69,7 +71,7 @@ async fn postgres_migrate_should_skip_existing_migration_table() { .fetch_one(&pool) .await .unwrap(); - assert_eq!(migration_count, 1); + assert_eq!(migration_count, 2); } #[tokio::test] @@ -110,6 +112,7 @@ async fn postgres_schema_separates_success_and_error_facts() { "input_tokens", "output_tokens", "cached_tokens", + "cache_write_tokens", "reasoning_tokens", ] { assert!(usage_columns.contains(column), "missing {column}"); @@ -117,6 +120,9 @@ async fn postgres_schema_separates_success_and_error_facts() { assert!(!usage_columns.contains("level")); assert!(!usage_columns.contains("failure_class")); + let bucket_columns = columns(&pool, "request_time_buckets").await; + assert!(bucket_columns.contains("cache_write_tokens")); + let error_columns = columns(&pool, "ops_error_logs").await; for column in [ "client_api_key_id", diff --git a/backend/tests/telemetry/rebuild.rs b/backend/tests/telemetry/rebuild.rs index 75e5caafc..cbcdcc498 100644 --- a/backend/tests/telemetry/rebuild.rs +++ b/backend/tests/telemetry/rebuild.rs @@ -20,6 +20,7 @@ async fn rebuild_buckets_should_replace_only_fully_reconstructible_range() { success.input_tokens = Some(12); success.output_tokens = Some(5); success.cached_tokens = Some(3); + success.cache_write_tokens = Some(2); PgUsageRecordStore::new(pool.clone()) .append(&success) .await @@ -60,15 +61,16 @@ async fn rebuild_buckets_should_replace_only_fully_reconstructible_range() { assert_eq!(report.deleted_rows, 1); assert_eq!(report.rebuilt_rows, 1); - let rebuilt: (i64, i64, i64, i64, i64, i64, i64, Option) = sqlx::query_as( + let rebuilt: (i64, i64, i64, i64, i64, i64, i64, i64, Option) = sqlx::query_as( "select success_count, error_count, input_tokens, output_tokens, cached_tokens, + cache_write_tokens, first_token_latency_count, latency_count, min_latency_ms from request_time_buckets where account_id = 'acct-1'", ) .fetch_one(&pool) .await .unwrap(); - assert_eq!(rebuilt, (1, 1, 12, 5, 3, 1, 1, Some(0))); + assert_eq!(rebuilt, (1, 1, 12, 5, 3, 2, 1, 1, Some(0))); let preserved: (i64, i64) = sqlx::query_as( "select success_count, error_count diff --git a/backend/tests/telemetry/usage/store.rs b/backend/tests/telemetry/usage/store.rs index 87bb94720..07cd29454 100644 --- a/backend/tests/telemetry/usage/store.rs +++ b/backend/tests/telemetry/usage/store.rs @@ -241,11 +241,12 @@ async fn usage_record_and_bucket_are_one_transaction_and_keep_zero_ms_samples() record.first_token_ms = Some(0); record.input_tokens = Some(11); record.output_tokens = Some(7); + record.cache_write_tokens = Some(3); record.service_tier = Some("priority".to_string()); store.append(&record).await.unwrap(); - let bucket: (i64, i64, i64, i64, i64, String) = sqlx::query_as( - "select success_count, input_tokens, output_tokens, latency_count, + let bucket: (i64, i64, i64, i64, i64, i64, String) = sqlx::query_as( + "select success_count, input_tokens, output_tokens, cache_write_tokens, latency_count, first_token_latency_count, service_tier from request_time_buckets where account_id = $1 and model = $2", ) @@ -254,7 +255,7 @@ async fn usage_record_and_bucket_are_one_transaction_and_keep_zero_ms_samples() .fetch_one(&pool) .await .unwrap(); - assert_eq!(bucket, (1, 11, 7, 1, 1, "priority".to_string())); + assert_eq!(bucket, (1, 11, 7, 3, 1, 1, "priority".to_string())); sqlx::raw_sql( "create function reject_test_bucket() returns trigger language plpgsql as $$ diff --git a/backend/tests/upstream/openai/protocol/events.rs b/backend/tests/upstream/openai/protocol/events.rs index 93206cc1d..f42dc9bcd 100644 --- a/backend/tests/upstream/openai/protocol/events.rs +++ b/backend/tests/upstream/openai/protocol/events.rs @@ -7,7 +7,8 @@ fn extract_usage_should_read_codex_usage_shape() { "input_tokens": 12, "output_tokens": 5, "input_tokens_details": { - "cached_tokens": 3 + "cached_tokens": 3, + "cache_write_tokens": 4 } } }); @@ -20,6 +21,7 @@ fn extract_usage_should_read_codex_usage_shape() { input_tokens: 12, output_tokens: 5, cached_tokens: 3, + cache_write_tokens: 4, reasoning_tokens: 0, image_input_tokens: 0, image_output_tokens: 0, @@ -49,6 +51,7 @@ fn extract_usage_should_read_reasoning_and_total_tokens() { input_tokens: 12, output_tokens: 8, cached_tokens: 0, + cache_write_tokens: 0, reasoning_tokens: 6, image_input_tokens: 0, image_output_tokens: 0, @@ -83,6 +86,7 @@ fn extract_usage_should_read_image_generation_tokens_separately() { input_tokens: 12, output_tokens: 5, cached_tokens: 3, + cache_write_tokens: 0, reasoning_tokens: 0, image_input_tokens: 31, image_output_tokens: 9, @@ -110,6 +114,7 @@ fn extract_usage_should_read_openai_usage_shape() { input_tokens: 8, output_tokens: 4, cached_tokens: 2, + cache_write_tokens: 0, reasoning_tokens: 0, image_input_tokens: 0, image_output_tokens: 0, @@ -132,6 +137,7 @@ fn extract_sse_usage_should_prefer_completed_response_usage() { input_tokens: 3, output_tokens: 5, cached_tokens: 1, + cache_write_tokens: 0, reasoning_tokens: 0, image_input_tokens: 0, image_output_tokens: 0, @@ -154,6 +160,7 @@ fn extract_sse_usage_should_read_completed_image_generation_tokens() { input_tokens: 12, output_tokens: 5, cached_tokens: 3, + cache_write_tokens: 0, reasoning_tokens: 0, image_input_tokens: 31, image_output_tokens: 9, diff --git a/backend/tests/upstream/openai/protocol/responses.rs b/backend/tests/upstream/openai/protocol/responses.rs index 0bab8d7fb..70a4263d8 100644 --- a/backend/tests/upstream/openai/protocol/responses.rs +++ b/backend/tests/upstream/openai/protocol/responses.rs @@ -170,6 +170,51 @@ fn build_codex_request_should_pass_through_input_and_preserve_context_fields() { assert!(!codex.force_http_sse); } +#[test] +fn build_codex_request_should_preserve_lite_and_memgen_transport_semantics() { + let body = json!({ + "model": "gpt-5.6-sol", + "input": [], + "client_metadata": { + "ws_request_header_x_openai_internal_codex_responses_lite": "true" + } + }); + let mut headers = HeaderMap::new(); + headers.insert("x-openai-memgen-request", "true".parse().unwrap()); + + let request = build_codex_request(body.as_object().unwrap().clone(), &headers, None); + + assert_eq!(request.responses_lite.as_deref(), Some("true")); + assert_eq!(request.memgen_request.as_deref(), Some("true")); + assert_eq!( + request.client_metadata().and_then( + |metadata| metadata.get("ws_request_header_x_openai_internal_codex_responses_lite") + ), + Some(&json!("true")) + ); +} + +#[test] +fn build_codex_request_should_prefer_http_lite_header_without_rewriting_metadata() { + let body = json!({ + "model": "gpt-5.6-sol", + "input": [], + "client_metadata": { + "ws_request_header_x_openai_internal_codex_responses_lite": "metadata-value" + } + }); + let mut headers = HeaderMap::new(); + headers.insert( + "x-openai-internal-codex-responses-lite", + "header-value".parse().unwrap(), + ); + + let request = build_codex_request(body.as_object().unwrap().clone(), &headers, None); + + assert_eq!(request.responses_lite.as_deref(), Some("header-value")); + assert_eq!(request.body(), body.as_object().unwrap()); +} + #[test] fn build_codex_request_should_pass_through_array_input_items_verbatim() { let input = json!([ diff --git a/backend/tests/upstream/openai/protocol/sse.rs b/backend/tests/upstream/openai/protocol/sse.rs index e35c876d1..dbcbebaa7 100644 --- a/backend/tests/upstream/openai/protocol/sse.rs +++ b/backend/tests/upstream/openai/protocol/sse.rs @@ -76,6 +76,50 @@ fn response_event_signals_should_count_reasoning_as_semantic_output() { assert!(!signals.text_output); } +#[test] +fn response_event_signals_should_count_refusal_as_visible_text() { + let signals = response_event_signals( + Some("response.refusal.delta"), + &json!({"delta": "I cannot help with that"}), + ); + + assert!(signals.semantic_output); + assert!(signals.text_output); + assert!(!signals.reasoning_output); +} + +#[test] +fn response_event_signals_should_count_future_nonempty_delta_without_event_table_entry() { + let signals = response_event_signals( + Some("response.future_tool_payload.delta"), + &json!({"delta": {"chunk": 1}}), + ); + + assert!(signals.semantic_output); + assert!(!signals.text_output); + assert!(!signals.reasoning_output); +} + +#[test] +fn response_event_signals_should_count_completed_hosted_tool_item_without_known_fields() { + let signals = response_event_signals( + Some("response.output_item.done"), + &json!({"item": {"type": "mcp_call", "status": "completed"}}), + ); + + assert!(signals.semantic_output); +} + +#[test] +fn response_event_signals_should_count_hosted_tool_execution_phase() { + let signals = response_event_signals( + Some("response.code_interpreter_call.in_progress"), + &json!({"type": "response.code_interpreter_call.in_progress"}), + ); + + assert!(signals.semantic_output); +} + #[test] fn first_token_detection_should_ignore_metadata_and_rate_limit_events() { let body = include_bytes!( diff --git a/backend/tests/upstream/openai/transport/headers.rs b/backend/tests/upstream/openai/transport/headers.rs index eb7054ce6..31d3f45a0 100644 --- a/backend/tests/upstream/openai/transport/headers.rs +++ b/backend/tests/upstream/openai/transport/headers.rs @@ -285,6 +285,8 @@ async fn codex_backend_client_websocket_should_forward_headers_and_preserve_payl Vec::new(), ); request.use_websocket = true; + request.responses_lite = Some("true".to_string()); + request.memgen_request = Some("true".to_string()); request.set_prompt_cache_key(Some("client-thread".to_string())); request.set_client_metadata(Some(json!({ "safe": "yes", @@ -351,7 +353,8 @@ async fn codex_backend_client_websocket_should_forward_headers_and_preserve_payl json!({ "safe": "yes", "x-openai-subagent": "review", - "ignored_non_string": 42 + "ignored_non_string": 42, + "ws_request_header_x_openai_internal_codex_responses_lite": "true" }) ); @@ -402,6 +405,16 @@ async fn codex_backend_client_websocket_should_forward_headers_and_preserve_payl .iter() .any(|(name, value)| name == "x-openai-subagent" && value == "review") ); + assert!( + headers + .iter() + .all(|(name, _)| name != "x-openai-internal-codex-responses-lite") + ); + assert!( + headers + .iter() + .any(|(name, value)| { name == "x-openai-memgen-request" && value == "true" }) + ); assert!( headers .iter() @@ -430,6 +443,14 @@ async fn codex_backend_client_should_send_desktop_headers_and_capture_response_m .and(wiremock::matchers::header("originator", "Codex Desktop")) .and(wiremock::matchers::header("x-client-request-id", "req_1")) .and(wiremock::matchers::header("x-codex-turn-state", "turn_1")) + .and(wiremock::matchers::header( + "x-openai-internal-codex-responses-lite", + "true", + )) + .and(wiremock::matchers::header( + "x-openai-memgen-request", + "true", + )) .and(wiremock::matchers::header("cookie", "cf_clearance=old")) .and(wiremock::matchers::body_json(json!({ "model": "gpt-5.5", @@ -460,6 +481,8 @@ async fn codex_backend_client_should_send_desktop_headers_and_capture_response_m Vec::new(), ); request.force_http_sse = true; + request.responses_lite = Some("true".to_string()); + request.memgen_request = Some("true".to_string()); let response = client .create_response( @@ -493,6 +516,7 @@ async fn codex_backend_client_should_send_desktop_headers_and_capture_response_m input_tokens: 2, output_tokens: 3, cached_tokens: 1, + cache_write_tokens: 0, reasoning_tokens: 0, image_input_tokens: 0, image_output_tokens: 0, @@ -579,6 +603,123 @@ async fn codex_backend_client_should_send_latest_wire_profile_user_agent() { assert_eq!(user_agents, vec![expected_user_agent]); } +#[tokio::test] +async fn codex_backend_client_websocket_should_preserve_exact_chain_then_rotate_profile() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + let (first_stream, _) = listener.accept().await.unwrap(); + let mut first_user_agent = String::new(); + let mut first = accept_codex_test_websocket_with(first_stream, |request, _response| { + first_user_agent = request + .headers() + .get("user-agent") + .and_then(|value| value.to_str().ok()) + .unwrap_or_default() + .to_string(); + }) + .await; + let _ = first.next().await.unwrap().unwrap(); + first + .send(Message::Text( + completed_websocket_response("resp_profile_first", 1, 1).into(), + )) + .await + .unwrap(); + + let _ = timeout(Duration::from_secs(2), first.next()) + .await + .expect("connection-local continuation should reuse the original websocket") + .unwrap() + .unwrap(); + first + .send(Message::Text( + completed_websocket_response("resp_profile_continued", 1, 1).into(), + )) + .await + .unwrap(); + + let (second_stream, _) = timeout(Duration::from_secs(2), listener.accept()) + .await + .expect("updated profile should open a new websocket") + .unwrap(); + let mut second_user_agent = String::new(); + let mut second = accept_codex_test_websocket_with(second_stream, |request, _response| { + second_user_agent = request + .headers() + .get("user-agent") + .and_then(|value| value.to_str().ok()) + .unwrap_or_default() + .to_string(); + }) + .await; + let _ = second.next().await.unwrap().unwrap(); + second + .send(Message::Text( + completed_websocket_response("resp_profile_second", 1, 1).into(), + )) + .await + .unwrap(); + + (first_user_agent, second_user_agent) + }); + + let profile = crate::support::wire_profile::test_wire_profile(); + let pool = Arc::new( + codex_proxy_rs::upstream::openai::transport::websocket_pool::CodexWebSocketPool::new( + 8, + Duration::from_mins(1), + ), + ); + let client = CodexBackendClient::new( + reqwest::Client::builder().no_proxy().build().unwrap(), + format!("http://{addr}"), + profile.clone(), + ) + .with_websocket_pool(pool); + let mut request = + codex_proxy_rs::upstream::openai::protocol::responses::CodexResponsesRequest::new_http_sse( + "gpt-5.5", + "", + Vec::new(), + ); + request.use_websocket = true; + request.local_conversation_id = Some("profile-rotation".to_string()); + + client + .create_response( + &request, + request_context("req_profile_first", Some("acct-profile")), + ) + .await + .unwrap(); + profile.update_desktop_release("26.900.1", "7001"); + let mut continuation = request.clone(); + continuation.set_previous_response_id(Some("resp_profile_first".to_string())); + continuation.previous_response_scope = Some( + codex_proxy_rs::upstream::openai::protocol::responses::PreviousResponseScope::ConnectionLocal, + ); + client + .create_response( + &continuation, + request_context("req_profile_continued", Some("acct-profile")), + ) + .await + .unwrap(); + client + .create_response( + &request, + request_context("req_profile_second", Some("acct-profile")), + ) + .await + .unwrap(); + + let (first_user_agent, second_user_agent) = server.await.unwrap(); + assert!(first_user_agent.contains("26.707.72221")); + assert!(second_user_agent.contains("26.900.1")); + assert_ne!(first_user_agent, second_user_agent); +} + #[tokio::test] async fn codex_backend_client_usage_should_use_wham_usage_headers() { let server = wiremock::MockServer::start().await; diff --git a/backend/tests/upstream/openai/transport/websocket_pool.rs b/backend/tests/upstream/openai/transport/websocket_pool.rs index cf1c5c057..5400456c8 100644 --- a/backend/tests/upstream/openai/transport/websocket_pool.rs +++ b/backend/tests/upstream/openai/transport/websocket_pool.rs @@ -1039,7 +1039,7 @@ async fn codex_backend_client_should_treat_active_business_frames_as_ping_livene let _request = websocket.next().await.unwrap().unwrap(); for index in 0..8 { - tokio::time::sleep(Duration::from_millis(20)).await; + tokio::time::sleep(Duration::from_secs(1)).await; websocket .send(Message::Text( json!({ @@ -1061,8 +1061,8 @@ async fn codex_backend_client_should_treat_active_business_frames_as_ping_livene websocket.close(None).await.unwrap(); }); let pool = Arc::new(CodexWebSocketPool::with_config(CodexWebSocketPoolConfig { - ping_interval: Some(Duration::from_millis(10)), - ping_timeout: Duration::from_millis(60), + ping_interval: Some(Duration::from_millis(100)), + ping_timeout: Duration::from_secs(5), liveness_timeout: None, ..websocket_pool_config_for_tests(None, None, None) })); diff --git a/docs/architecture.md b/docs/architecture.md index b3d5150d8..c91d42ce7 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -131,8 +131,8 @@ upstream/openai/ - `protocol` 定义 Responses body、canonical SSE/WS 事实和序列化，不包含账号策略。 - `failure.rs` 将各 transport 错误规范化为稳定的 `UpstreamFailureFacts`，不解释账号状态。 - `transport` 拥有 HTTP/SSE、WebSocket、TLS、header、连接池、熔断和 transport metrics。 -- `profile.rs` 定义配置驱动、启动后不可变的 Codex Desktop 上游请求画像。 -- `desktop_release.rs` 只观测官方 appcast；发布变化不会自动改写生效请求画像。 +- `profile.rs` 定义配置驱动的 Codex Desktop 上游请求画像；请求在发送前读取原子快照。 +- `desktop_release.rs` 定期读取官方 appcast，并自动把合法的 Desktop 版本与构建号应用到运行时画像。新 HTTP 请求立即使用新画像；WebSocket 连接池按实际 opening 画像分代，不复用旧 UA 连接，也不改写在途请求。 - `token_client.rs` 只实现 OAuth token 刷新协议。 ### fleet diff --git a/frontend/src/views/usage/components/UsageBillingCell.vue b/frontend/src/views/usage/components/UsageBillingCell.vue index bfe897028..075f53fde 100644 --- a/frontend/src/views/usage/components/UsageBillingCell.vue +++ b/frontend/src/views/usage/components/UsageBillingCell.vue @@ -20,6 +20,8 @@ const amountItems = computed(() => { { label: '输入单价', value: value.inputPriceDisplay, accent: true }, { label: '输出单价', value: value.outputPriceDisplay, accent: true }, { label: '缓存读取费用', value: value.cacheReadAmountDisplay, accent: false }, + { label: '缓存写入费用', value: value.cacheWriteAmountDisplay, accent: false }, + { label: '缓存写入单价', value: value.cacheWritePriceDisplay, accent: true }, ] }) const billingItems = computed(() => { diff --git a/frontend/src/views/usage/components/UsageRecordDetailModal.vue b/frontend/src/views/usage/components/UsageRecordDetailModal.vue index f95adc6c5..11cda9ed0 100644 --- a/frontend/src/views/usage/components/UsageRecordDetailModal.vue +++ b/frontend/src/views/usage/components/UsageRecordDetailModal.vue @@ -103,10 +103,12 @@ const billingItems = computed(() => { { label: '输入', value: value.inputAmountDisplay, mono: true }, { label: '输出', value: value.outputAmountDisplay, mono: true }, { label: '缓存读取', value: value.cacheReadAmountDisplay, mono: true }, + { label: '缓存写入', value: value.cacheWriteAmountDisplay, mono: true }, { label: '标准费用', value: value.standardAmountDisplay, mono: true }, { label: '输入单价', value: value.inputPriceDisplay, mono: true }, { label: '输出单价', value: value.outputPriceDisplay, mono: true }, { label: '缓存单价', value: value.cacheReadPriceDisplay, mono: true }, + { label: '缓存写入单价', value: value.cacheWritePriceDisplay, mono: true }, { label: '服务层级', value: value.serviceTierDisplay }, { label: '倍率', value: value.multiplierDisplay, mono: true }, ] @@ -131,6 +133,12 @@ const tokenChartItems = computed(() => [ display: tokenDetails.value.cachedTokensDisplay, color: themeColor('--cp-warning', '#D97706'), }, + { + label: '缓存写入', + value: Number(tokenDetails.value.cacheWriteTokens || 0), + display: tokenDetails.value.cacheWriteTokensDisplay, + color: themeColor('--cp-danger', '#DC2626'), + }, { label: '推理', value: Number(tokenDetails.value.reasoningTokens || 0), diff --git a/frontend/src/views/usage/components/UsageTokenCell.vue b/frontend/src/views/usage/components/UsageTokenCell.vue index c36cb69c5..b28344c25 100644 --- a/frontend/src/views/usage/components/UsageTokenCell.vue +++ b/frontend/src/views/usage/components/UsageTokenCell.vue @@ -14,6 +14,7 @@ const tokenItems = computed(() => [ { label: '输入 Token', value: tokenDetails.value.inputTokensDisplay }, { label: '输出 Token', value: tokenDetails.value.outputTokensDisplay }, { label: '缓存读取 Token', value: tokenDetails.value.cachedTokensDisplay }, + { label: '缓存写入 Token', value: tokenDetails.value.cacheWriteTokensDisplay }, { label: '推理 Token', value: tokenDetails.value.reasoningTokensDisplay }, ]) diff --git a/frontend/src/views/usage/constants.ts b/frontend/src/views/usage/constants.ts index 2027bad42..fe07ebee6 100644 --- a/frontend/src/views/usage/constants.ts +++ b/frontend/src/views/usage/constants.ts @@ -277,6 +277,7 @@ export function usageTokenDetails(record: any) { const inputTokens = numberValue(usage.inputTokens) const outputTokens = numberValue(usage.outputTokens) const cachedTokens = numberValue(usage.cachedTokens) + const cacheWriteTokens = numberValue(usage.cacheWriteTokens) const reasoningTokens = numberValue(usage.reasoningTokens) const totalTokens = numberValue(usage.totalTokens) || inputTokens + outputTokens @@ -284,11 +285,13 @@ export function usageTokenDetails(record: any) { inputTokens, outputTokens, cachedTokens, + cacheWriteTokens, reasoningTokens, totalTokens, inputTokensDisplay: formatTokenCount(inputTokens), outputTokensDisplay: formatTokenCount(outputTokens), cachedTokensDisplay: formatCompactTokenCount(cachedTokens), + cacheWriteTokensDisplay: formatCompactTokenCount(cacheWriteTokens), reasoningTokensDisplay: formatTokenCount(reasoningTokens), totalTokensDisplay: formatTokenCount(totalTokens), }