i/Aether
1
0
Fork 0
mirror of https://github.com/fawney19/Aether.git synced 2026-09-10 05:00:19 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(billing): preserve effective cache and tier facts

Browse Source
This commit is contained in:
MMEXA
2026-07-11 18:12:55 +08:00
parent 14c4e3a04e
commit 3014fd50c6
12 changed files with 568 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/aether-gateway/src/control/auth/gate.rs
+47 -4
View File
@@ -191,6 +191,13 @@ async fn estimate_execution_plan_cost_upper_bound_usd(
));
let model_id = report_context_string_field(report_context, "model_id");
let global_model_name = report_context_string_field(report_context, "global_model_name");
let cache_ttl_minutes =
aether_data_contracts::repository::usage::resolve_provider_cache_ttl_minutes(
Some(&api_format),
plan.model_name.as_deref(),
global_model_name,
Some(body_json),
);
if model_id.is_none() && global_model_name.is_none() {
return Ok(None);
}
@@ -202,6 +209,7 @@ async fn estimate_execution_plan_cost_upper_bound_usd(
input_tokens,
max_output_tokens,
requested_processing_tier.as_deref(),
cache_ttl_minutes,
);
let ttl = state.frontdoor_runtime_guards.auth_capacity_cache_ttl;
if ttl.is_zero() {
@@ -215,6 +223,7 @@ async fn estimate_execution_plan_cost_upper_bound_usd(
input_tokens,
max_output_tokens,
requested_processing_tier.as_deref(),
cache_ttl_minutes,
)
.await;
}
@@ -231,6 +240,7 @@ async fn estimate_execution_plan_cost_upper_bound_usd(
input_tokens,
max_output_tokens,
requested_processing_tier.as_deref(),
cache_ttl_minutes,
)
.await
})
@@ -248,6 +258,7 @@ async fn calculate_execution_plan_cost_upper_bound(
input_tokens: i64,
max_output_tokens: Option<i64>,
requested_processing_tier: Option<&str>,
cache_ttl_minutes: Option<i64>,
) -> Result<Option<f64>, GatewayError> {
let context = match model_id {
Some(model_id) => state
@@ -272,6 +283,7 @@ async fn calculate_execution_plan_cost_upper_bound(
aether_billing::BillingAuthorizationEstimateInput::new(task_type, input_tokens);
estimate.api_format = Some(api_format.to_string());
estimate.requested_processing_tier = requested_processing_tier.map(ToOwned::to_owned);
estimate.cache_ttl_minutes = cache_ttl_minutes;
estimate.max_output_tokens = max_output_tokens;
aether_billing::BillingService::new()
.estimate_authorization_cost_upper_bound(
@@ -289,9 +301,10 @@ fn execution_plan_cost_upper_bound_cache_key(
input_tokens: i64,
max_output_tokens: Option<i64>,
requested_processing_tier: Option<&str>,
cache_ttl_minutes: Option<i64>,
) -> String {
format!(
"{}\x1f{}\x1f{}\x1f{}\x1f{}\x1f{}\x1f{}\x1f{}",
"{}\x1f{}\x1f{}\x1f{}\x1f{}\x1f{}\x1f{}\x1f{}\x1f{}",
plan.provider_id,
plan.key_id,
model_id.unwrap_or(""),
@@ -302,6 +315,9 @@ fn execution_plan_cost_upper_bound_cache_key(
.map(|value| value.to_string())
.unwrap_or_else(|| "none".to_string()),
requested_processing_tier.unwrap_or("standard"),
cache_ttl_minutes
.map(|value| value.to_string())
.unwrap_or_else(|| "none".to_string()),
)
}
@@ -554,9 +570,9 @@ mod tests {
use serde_json::json;
use super::{
execution_plan_balance_capacity_rejection, max_output_tokens_from_request,
openai_request_input_is_self_contained, output_choice_count_upper_bound,
request_model_local_rejection, GatewayLocalAuthRejection,
execution_plan_balance_capacity_rejection, execution_plan_cost_upper_bound_cache_key,
max_output_tokens_from_request, openai_request_input_is_self_contained,
output_choice_count_upper_bound, request_model_local_rejection, GatewayLocalAuthRejection,
};
use crate::control::{GatewayControlAuthContext, GatewayControlDecision};
use crate::data::GatewayDataState;
@@ -1423,6 +1439,33 @@ mod tests {
);
}
#[test]
fn authorization_cache_key_includes_effective_cache_ttl() {
let plan = execution_plan(json!({"model": "gpt-5.6-sol"}), "openai:responses");
let without_ttl = execution_plan_cost_upper_bound_cache_key(
&plan,
Some("model-1"),
Some("gpt-5.6-sol"),
"openai:responses",
100,
Some(10),
Some("priority"),
None,
);
let with_ttl = execution_plan_cost_upper_bound_cache_key(
&plan,
Some("model-1"),
Some("gpt-5.6-sol"),
"openai:responses",
100,
Some(10),
Some("priority"),
Some(30),
);
assert_ne!(without_ttl, with_ttl);
}
#[test]
fn indirect_request_inputs_are_not_treated_as_body_bounded() {
let self_contained = json!({
crates/aether-ai-formats/src/api.rs
+1
View File
@@ -56,6 +56,7 @@ pub use crate::formats::openai::image::stream::{
maybe_build_openai_image_sync_finalize_product, OpenAiImageStreamState,
OpenAiImageSyncFinalizeProduct,
};
pub use crate::formats::openai::prompt_cache::resolve_openai_prompt_cache_ttl_minutes;
pub use crate::formats::openai::shared::{
copy_request_number_field, copy_request_number_field_as,
map_openai_reasoning_effort_to_claude_output, map_openai_reasoning_effort_to_gemini_budget,
crates/aether-ai-formats/src/formats/openai/prompt_cache.rs
+82 -1
View File
@@ -33,6 +33,31 @@ pub fn validate_openai_prompt_cache_request(
)
}
pub fn resolve_openai_prompt_cache_ttl_minutes(
provider_api_format: &str,
provider_model: &str,
source_model: &str,
body: &Value,
) -> Option<i64> {
OpenAiPromptCacheApi::parse(provider_api_format)?;
let request = body.as_object()?;
let explicit_ttl = request
.get("prompt_cache_options")
.and_then(Value::as_object)
.and_then(|options| options.get("ttl"))
.and_then(Value::as_str);
if explicit_ttl == Some("30m") {
return Some(30);
}
let capability_model =
crate::formats::shared::model_directives::openai_model_capability_identity(
provider_model,
source_model,
);
crate::openai_model_supports_prompt_cache_options(&capability_model).then_some(30)
}
pub(crate) fn validate_openai_prompt_cache_request_with_source_model(
source_api_format: &str,
provider_model: &str,
@@ -362,7 +387,63 @@ fn unsupported_for_model(field: &str, reason: &str) -> OpenAiPromptCacheContract
mod tests {
use serde_json::json;
use super::{validate_openai_prompt_cache_request, OpenAiPromptCacheViolationKind};
use super::{
resolve_openai_prompt_cache_ttl_minutes, validate_openai_prompt_cache_request,
OpenAiPromptCacheViolationKind,
};
#[test]
fn resolves_effective_prompt_cache_ttl_from_current_openai_contract() {
for body in [
json!({"model": "client-alias"}),
json!({
"model": "client-alias",
"prompt_cache_options": {"mode": "implicit"}
}),
json!({
"model": "client-alias",
"prompt_cache_options": {"mode": "explicit", "ttl": "30m"}
}),
] {
assert_eq!(
resolve_openai_prompt_cache_ttl_minutes(
"openai:responses",
"gpt-5.6-sol",
"client-alias",
&body,
),
Some(30)
);
}
assert_eq!(
resolve_openai_prompt_cache_ttl_minutes(
"openai:chat",
"deployment-alias",
"gpt-5.6-terra",
&json!({"model": "gpt-5.6-terra"}),
),
Some(30)
);
assert_eq!(
resolve_openai_prompt_cache_ttl_minutes(
"openai:chat",
"gpt-5.5",
"gpt-5.6-terra",
&json!({"model": "gpt-5.6-terra"}),
),
None
);
assert_eq!(
resolve_openai_prompt_cache_ttl_minutes(
"claude:messages",
"gpt-5.6-sol",
"gpt-5.6-sol",
&json!({"model": "gpt-5.6-sol"}),
),
None
);
}
#[test]
fn gpt_5_6_accepts_current_prompt_cache_options_and_breakpoints() {
crates/aether-ai-formats/src/lib.rs
+1
View File
@@ -22,6 +22,7 @@ pub use formats::matrix::{
sync_chat_response_conversion_kind, sync_cli_response_conversion_kind, RequestConversionKind,
SyncChatResponseConversionKind, SyncCliResponseConversionKind,
};
pub use formats::openai::prompt_cache::resolve_openai_prompt_cache_ttl_minutes;
pub use formats::openai::prompt_cache::{
validate_openai_prompt_cache_request, OpenAiPromptCacheContractViolation,
OpenAiPromptCacheViolationKind,
crates/aether-billing/src/event_enrichment.rs
+118 -4
View File
@@ -1,8 +1,9 @@
use aether_data_contracts::repository::billing::StoredBillingModelContext;
use aether_data_contracts::repository::usage::{
extract_provider_actual_service_tier_from_response, extract_provider_service_tier_from_body,
normalize_provider_service_tier, PROVIDER_ACTUAL_SERVICE_TIER_METADATA_KEY,
PROVIDER_SERVICE_TIER_METADATA_KEY,
extract_provider_actual_service_tier_from_response,
extract_provider_cache_ttl_minutes_from_metadata, extract_provider_service_tier_from_body,
normalize_provider_service_tier, resolve_provider_cache_ttl_minutes,
PROVIDER_ACTUAL_SERVICE_TIER_METADATA_KEY, PROVIDER_SERVICE_TIER_METADATA_KEY,
};
use aether_data_contracts::DataLayerError;
use aether_usage_runtime::{UsageEvent, UsageEventType};
@@ -177,7 +178,8 @@ fn calculate_billing_computation(
image_size: usage_event_dimension_string(&event.data, "image_size"),
image_quality: usage_event_dimension_string(&event.data, "image_quality"),
image_output_format: usage_event_dimension_string(&event.data, "image_output_format"),
cache_ttl_minutes: pricing.provider_api_key_cache_ttl_minutes,
cache_ttl_minutes: usage_event_provider_cache_ttl_minutes(&event.data)
.or(pricing.provider_api_key_cache_ttl_minutes),
};
BillingService::new()
@@ -215,6 +217,20 @@ fn usage_event_processing_tiers(
UsageEventProcessingTiers { requested, actual }
}
fn usage_event_provider_cache_ttl_minutes(
data: &aether_usage_runtime::UsageEventData,
) -> Option<i64> {
resolve_provider_cache_ttl_minutes(
data.endpoint_api_format
.as_deref()
.or(data.api_format.as_deref()),
data.target_model.as_deref().or(Some(data.model.as_str())),
Some(data.model.as_str()),
data.provider_request_body.as_ref(),
)
.or_else(|| extract_provider_cache_ttl_minutes_from_metadata(data.request_metadata.as_ref()))
}
fn usage_event_is_image_usage(data: &aether_usage_runtime::UsageEventData) -> bool {
data.request_type
.as_deref()
@@ -444,6 +460,104 @@ mod tests {
assert_eq!(tiers.actual.as_deref(), Some("default"));
}
#[tokio::test]
async fn settlement_uses_effective_gpt_5_6_cache_ttl_after_body_capture() {
let lookup = TestLookup {
name_context: Some(
StoredBillingModelContext::new(
"provider-1".to_string(),
Some("pay_as_you_go".to_string()),
Some("key-1".to_string()),
None,
Some(60),
"global-model-1".to_string(),
"gpt-5.6-sol".to_string(),
None,
None,
Some(json!({
"tiers": [{
"up_to": null,
"input_price_per_1m": 5.0,
"output_price_per_1m": 30.0,
"cache_creation_price_per_1m": 6.25,
"cache_read_price_per_1m": 0.5,
"cache_ttl_pricing": [{
"ttl_minutes": 60,
"cache_creation_price_per_1m": 100.0,
"cache_read_price_per_1m": 100.0
}]
}]
})),
Some("model-1".to_string()),
Some("gpt-5.6-sol".to_string()),
None,
None,
None,
)
.expect("billing context should build"),
),
model_id_context: None,
};
for (request_id, provider_request_body, request_metadata) in [
(
"req-cache-body",
Some(json!({"model": "gpt-5.6-sol"})),
None,
),
(
"req-cache-metadata",
None,
Some(json!({"provider_cache_ttl_minutes": 30})),
),
] {
let mut event = UsageEvent::new(
UsageEventType::Completed,
request_id,
UsageEventData {
provider_name: "OpenAI".to_string(),
model: "gpt-5.6-sol".to_string(),
target_model: Some("gpt-5.6-sol".to_string()),
provider_id: Some("provider-1".to_string()),
provider_api_key_id: Some("key-1".to_string()),
request_type: Some("chat".to_string()),
api_format: Some("openai:responses".to_string()),
endpoint_api_format: Some("openai:responses".to_string()),
provider_request_body,
request_metadata,
input_tokens: Some(1_000_000),
cache_creation_input_tokens: Some(1_000_000),
status_code: Some(200),
..UsageEventData::default()
},
);
enrich_usage_event_with_billing(&lookup, &mut event)
.await
.expect("billing should succeed");
let snapshot = event
.data
.request_metadata
.as_ref()
.and_then(|value| value.get("billing_snapshot"))
.expect("billing snapshot should exist");
assert_eq!(
snapshot
.get("resolved_dimensions")
.and_then(|value| value.get("cache_ttl_minutes")),
Some(&json!(30))
);
assert_eq!(
snapshot
.get("resolved_variables")
.and_then(|value| value.get("cache_creation_price_per_1m")),
Some(&json!(6.25))
);
assert_eq!(event.data.total_cost_usd, Some(6.25));
}
}
#[tokio::test]
async fn enriches_completed_usage_event_with_billing_snapshot() {
let lookup = TestLookup {
crates/aether-billing/src/pricing.rs
+12
View File
@@ -51,6 +51,15 @@ impl BillingPricingResolution {
.as_deref()
.is_some_and(processing_tier_is_standard)
}
pub fn bills_requested_processing_tier(&self) -> bool {
let requested = self
.requested_processing_tier
.as_deref()
.map(canonical_processing_tier)
.unwrap_or_else(|| "standard".to_string());
self.billing_processing_tier.as_deref() == Some(requested.as_str())
}
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
@@ -571,6 +580,8 @@ pub struct BillingAuthorizationEstimateInput {
pub task_type: String,
pub api_format: Option<String>,
pub requested_processing_tier: Option<String>,
#[serde(default)]
pub cache_ttl_minutes: Option<i64>,
pub input_tokens: i64,
pub max_output_tokens: Option<i64>,
}
@@ -581,6 +592,7 @@ impl BillingAuthorizationEstimateInput {
task_type: task_type.into(),
api_format: None,
requested_processing_tier: None,
cache_ttl_minutes: None,
input_tokens: input_tokens.max(0),
max_output_tokens: None,
}
crates/aether-billing/src/service.rs
+150 -12
View File
@@ -79,7 +79,9 @@ impl BillingService {
actual_processing_tier: None,
input_tokens,
output_tokens,
cache_ttl_minutes: pricing.provider_api_key_cache_ttl_minutes,
cache_ttl_minutes: estimate
.cache_ttl_minutes
.or(pricing.provider_api_key_cache_ttl_minutes),
..BillingUsageInput::new(estimate.task_type.clone())
};
let mut scenarios = vec![base_input.clone()];
@@ -92,17 +94,19 @@ impl BillingService {
cache_creation.cache_creation_tokens = input_tokens;
scenarios.push(cache_creation);
let mut cache_creation_5m = base_input.clone();
cache_creation_5m.cache_creation_tokens = input_tokens;
cache_creation_5m.cache_creation_ephemeral_5m_tokens = input_tokens;
cache_creation_5m.cache_ttl_minutes = Some(5);
scenarios.push(cache_creation_5m);
if estimate.cache_ttl_minutes.is_none() {
let mut cache_creation_5m = base_input.clone();
cache_creation_5m.cache_creation_tokens = input_tokens;
cache_creation_5m.cache_creation_ephemeral_5m_tokens = input_tokens;
cache_creation_5m.cache_ttl_minutes = Some(5);
scenarios.push(cache_creation_5m);
let mut cache_creation_1h = base_input.clone();
cache_creation_1h.cache_creation_tokens = input_tokens;
cache_creation_1h.cache_creation_ephemeral_1h_tokens = input_tokens;
cache_creation_1h.cache_ttl_minutes = Some(60);
scenarios.push(cache_creation_1h);
let mut cache_creation_1h = base_input.clone();
cache_creation_1h.cache_creation_tokens = input_tokens;
cache_creation_1h.cache_creation_ephemeral_1h_tokens = input_tokens;
cache_creation_1h.cache_ttl_minutes = Some(60);
scenarios.push(cache_creation_1h);
}
let mut cache_read = base_input;
cache_read.cache_read_tokens = input_tokens;
@@ -110,7 +114,8 @@ impl BillingService {
}
let mut upper_bound = 0.0_f64;
for pricing_resolution in pricing_resolutions {
'pricing_catalogs: for pricing_resolution in pricing_resolutions {
let is_requested_catalog = pricing_resolution.bills_requested_processing_tier();
for scenario in &scenarios {
let total_input_context = normalize_total_input_context_for_cache_hit_rate(
scenario.api_format.as_deref(),
@@ -129,6 +134,11 @@ impl BillingService {
let selected =
self.calculate_with_resolution(pricing, scenario, pricing_resolution.clone())?;
if !billing_computation_is_bounded(&selected) {
if !is_requested_catalog
&& billing_computation_is_outside_catalog_context(&selected)
{
continue 'pricing_catalogs;
}
return Ok(None);
}
@@ -300,6 +310,11 @@ fn billing_computation_is_bounded(computation: &BillingComputation) -> bool {
&& computation.actual_total_cost >= 0.0
}
fn billing_computation_is_outside_catalog_context(computation: &BillingComputation) -> bool {
computation.cost_result.status == BillingSnapshotStatus::NoRule
&& computation.cost_result.snapshot.missing_required == ["input_context_tier"]
}
fn authorization_pricing_candidates(
pricing: &BillingPricingResolution,
max_input_context: i64,
@@ -1176,6 +1191,129 @@ mod tests {
);
}
#[test]
fn processing_catalog_boundaries_match_context_and_priority_contracts() {
let cases = [
(
"default",
272_000,
BillingSnapshotStatus::Complete,
Some(5.0),
),
(
"default",
272_001,
BillingSnapshotStatus::Complete,
Some(10.0),
),
("flex", 272_000, BillingSnapshotStatus::Complete, Some(2.5)),
("flex", 272_001, BillingSnapshotStatus::Complete, Some(5.0)),
(
"priority",
272_000,
BillingSnapshotStatus::Complete,
Some(10.0),
),
("priority", 272_001, BillingSnapshotStatus::NoRule, None),
];
for (actual, input_tokens, status, input_price) in cases {
let result = BillingService::new()
.calculate(
&processing_pricing(),
&processing_usage(Some(actual), Some(actual), input_tokens),
)
.expect("processing boundary should resolve");
assert_eq!(
result.cost_result.status, status,
"{actual} at {input_tokens}"
);
if let Some(input_price) = input_price {
assert_eq!(
result.cost_result.snapshot.resolved_variables["input_price_per_1m"],
json!(input_price),
"{actual} at {input_tokens}"
);
} else {
assert_eq!(
result.cost_result.snapshot.missing_required,
vec!["input_context_tier"]
);
}
}
}
#[test]
fn authorization_estimate_uses_known_request_cache_ttl() {
let pricing = BillingModelPricingSnapshot {
provider_api_key_rate_multipliers: None,
default_price_per_request: None,
default_tiered_pricing: Some(json!({
"tiers": [{
"up_to": null,
"input_price_per_1m": 1.0,
"output_price_per_1m": 0.0,
"cache_creation_price_per_1m": 1.25,
"cache_read_price_per_1m": 0.1,
"cache_ttl_pricing": [{
"ttl_minutes": 60,
"cache_creation_price_per_1m": 100.0,
"cache_read_price_per_1m": 100.0
}]
}]
})),
..pricing()
};
let service = BillingService::new();
let mut estimate = BillingAuthorizationEstimateInput::new("chat", 1_000_000);
estimate.api_format = Some("openai:responses".to_string());
estimate.max_output_tokens = Some(0);
estimate.cache_ttl_minutes = Some(30);
assert_eq!(
service
.estimate_authorization_cost_upper_bound(&pricing, &estimate)
.expect("known TTL estimate should calculate"),
Some(1.25)
);
estimate.cache_ttl_minutes = None;
assert_eq!(
service
.estimate_authorization_cost_upper_bound(&pricing, &estimate)
.expect("unknown TTL estimate should calculate"),
Some(100.0)
);
}
#[test]
fn authorization_estimate_uses_only_processing_catalogs_eligible_for_context() {
let service = BillingService::new();
let mut estimate = BillingAuthorizationEstimateInput::new("chat", 300_000);
estimate.api_format = Some("openai:responses".to_string());
estimate.max_output_tokens = Some(0);
estimate.cache_ttl_minutes = Some(30);
for requested_processing_tier in [None, Some("standard"), Some("flex")] {
estimate.requested_processing_tier = requested_processing_tier.map(ToOwned::to_owned);
assert_eq!(
service
.estimate_authorization_cost_upper_bound(&processing_pricing(), &estimate)
.expect("eligible processing catalogs should calculate"),
Some(3.75),
"requested tier: {requested_processing_tier:?}"
);
}
estimate.requested_processing_tier = Some("priority".to_string());
assert_eq!(
service
.estimate_authorization_cost_upper_bound(&processing_pricing(), &estimate)
.expect("ineligible requested catalog should resolve"),
None
);
}
#[test]
fn unknown_actual_tier_cannot_fall_back_to_fixed_request_price() {
let pricing = BillingModelPricingSnapshot {
crates/aether-data-contracts/src/repository/usage/mod.rs
+14 -12
View File
@@ -2,17 +2,18 @@ mod types;
pub use types::{
extract_provider_actual_service_tier_from_response,
extract_provider_reasoning_effort_from_body, extract_provider_service_tier_from_body,
normalize_provider_service_tier, parse_usage_body_ref, usage_body_ref,
usage_request_metadata_client_family, ApiKeyLastUsedDelta, ManagementTokenCounterDelta,
PendingUsageCleanupSummary, ProviderApiKeyWindowUsageRequest, ProxyNodeCounterDelta,
StoredProviderApiKeyUsageSummary, StoredProviderApiKeyWindowUsageSummary,
StoredProviderUsageSummary, StoredProviderUsageWindow, StoredRequestUsageAudit,
StoredUsageAuditAggregation, StoredUsageAuditSummary, StoredUsageBreakdownSummaryRow,
StoredUsageCacheAffinityHitSummary, StoredUsageCacheAffinityIntervalRow,
StoredUsageCacheHitSummary, StoredUsageCostSavingsSummary, StoredUsageDailySummary,
StoredUsageDashboardDailyBreakdownRow, StoredUsageDashboardProviderCount,
StoredUsageDashboardSummary, StoredUsageErrorDistributionRow, StoredUsageLeaderboardSummary,
extract_provider_cache_ttl_minutes_from_metadata, extract_provider_reasoning_effort_from_body,
extract_provider_service_tier_from_body, normalize_provider_service_tier, parse_usage_body_ref,
resolve_provider_cache_ttl_minutes, usage_body_ref, usage_request_metadata_client_family,
ApiKeyLastUsedDelta, ManagementTokenCounterDelta, PendingUsageCleanupSummary,
ProviderApiKeyWindowUsageRequest, ProxyNodeCounterDelta, StoredProviderApiKeyUsageSummary,
StoredProviderApiKeyWindowUsageSummary, StoredProviderUsageSummary, StoredProviderUsageWindow,
StoredRequestUsageAudit, StoredUsageAuditAggregation, StoredUsageAuditSummary,
StoredUsageBreakdownSummaryRow, StoredUsageCacheAffinityHitSummary,
StoredUsageCacheAffinityIntervalRow, StoredUsageCacheHitSummary, StoredUsageCostSavingsSummary,
StoredUsageDailySummary, StoredUsageDashboardDailyBreakdownRow,
StoredUsageDashboardProviderCount, StoredUsageDashboardSummary,
StoredUsageErrorDistributionRow, StoredUsageLeaderboardSummary,
StoredUsagePerformancePercentilesRow, StoredUsageProviderPerformance,
StoredUsageProviderPerformanceProviderRow, StoredUsageProviderPerformanceSummary,
StoredUsageProviderPerformanceTimelineRow, StoredUsageSettledCostSummary,
@@ -30,5 +31,6 @@ pub use types::{
UsagePerformancePercentilesQuery, UsageProviderPerformanceQuery, UsageReadRepository,
UsageRepository, UsageSettledCostSummaryQuery, UsageTimeSeriesGranularity,
UsageTimeSeriesQuery, UsageWriteRepository, PROVIDER_ACTUAL_SERVICE_TIER_METADATA_KEY,
PROVIDER_REASONING_EFFORT_METADATA_KEY, PROVIDER_SERVICE_TIER_METADATA_KEY,
PROVIDER_CACHE_TTL_MINUTES_METADATA_KEY, PROVIDER_REASONING_EFFORT_METADATA_KEY,
PROVIDER_SERVICE_TIER_METADATA_KEY,
};
crates/aether-data-contracts/src/repository/usage/types.rs
+97 -2
View File
@@ -5,6 +5,7 @@ use serde_json::Value;
pub const PROVIDER_REASONING_EFFORT_METADATA_KEY: &str = "provider_reasoning_effort";
pub const PROVIDER_SERVICE_TIER_METADATA_KEY: &str = "provider_service_tier";
pub const PROVIDER_ACTUAL_SERVICE_TIER_METADATA_KEY: &str = "provider_actual_service_tier";
pub const PROVIDER_CACHE_TTL_MINUTES_METADATA_KEY: &str = "provider_cache_ttl_minutes";
pub fn extract_provider_reasoning_effort_from_body(value: Option<&Value>) -> Option<String> {
let object = value.and_then(Value::as_object)?;
@@ -71,6 +72,48 @@ pub fn normalize_provider_service_tier(value: &str) -> Option<String> {
Some(normalized)
}
pub fn resolve_provider_cache_ttl_minutes(
provider_api_format: Option<&str>,
provider_model: Option<&str>,
source_model: Option<&str>,
provider_request_body: Option<&Value>,
) -> Option<i64> {
let provider_api_format = provider_api_format?.trim();
let provider_request_body = provider_request_body?;
let provider_model = provider_request_body
.get("model")
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.or_else(|| {
provider_model
.map(str::trim)
.filter(|value| !value.is_empty())
})?;
let source_model = source_model
.map(str::trim)
.filter(|value| !value.is_empty())
.unwrap_or(provider_model);
aether_ai_formats::resolve_openai_prompt_cache_ttl_minutes(
provider_api_format,
provider_model,
source_model,
provider_request_body,
)
}
pub fn extract_provider_cache_ttl_minutes_from_metadata(value: Option<&Value>) -> Option<i64> {
value
.and_then(Value::as_object)
.and_then(|metadata| metadata.get(PROVIDER_CACHE_TTL_MINUTES_METADATA_KEY))
.and_then(|value| {
value
.as_i64()
.or_else(|| value.as_u64().and_then(|value| i64::try_from(value).ok()))
})
.filter(|value| *value > 0)
}
/// Joined usage read model assembled from the accounting row plus the newer audit/snapshot
/// satellite tables.
///
@@ -482,6 +525,20 @@ impl StoredRequestUsageAudit {
})
}
pub fn provider_cache_ttl_minutes(&self) -> Option<i64> {
resolve_provider_cache_ttl_minutes(
self.endpoint_api_format
.as_deref()
.or(self.api_format.as_deref()),
self.target_model.as_deref().or(Some(self.model.as_str())),
Some(self.model.as_str()),
self.provider_request_body.as_ref(),
)
.or_else(|| {
extract_provider_cache_ttl_minutes_from_metadata(self.request_metadata.as_ref())
})
}
pub fn body_ref(&self, field: UsageBodyField) -> Option<&str> {
match field {
UsageBodyField::RequestBody => self.request_body_ref.as_deref(),
@@ -2097,8 +2154,9 @@ fn parse_timestamp(value: i64, field_name: &str) -> Result<u64, crate::DataLayer
#[cfg(test)]
mod tests {
use super::{
extract_provider_actual_service_tier_from_response, StoredRequestUsageAudit,
UpsertUsageRecord, UsageBodyCaptureState, UsageBodyCaptureStorage, UsageBodyField,
extract_provider_actual_service_tier_from_response, resolve_provider_cache_ttl_minutes,
StoredRequestUsageAudit, UpsertUsageRecord, UsageBodyCaptureState, UsageBodyCaptureStorage,
UsageBodyField,
};
use serde_json::{json, Value};
@@ -2514,6 +2572,43 @@ mod tests {
assert_eq!(usage.provider_service_tier(), None);
}
#[test]
fn provider_cache_ttl_uses_final_openai_contract_then_preserved_metadata() {
let mut usage = sample_usage();
usage.model = "gpt-5.6-sol".to_string();
usage.target_model = Some("gpt-5.6-sol".to_string());
usage.provider_request_body = Some(json!({"model": "gpt-5.6-sol"}));
usage.request_metadata = Some(json!({"provider_cache_ttl_minutes": 60}));
assert_eq!(usage.provider_cache_ttl_minutes(), Some(30));
usage.provider_request_body = None;
usage.request_metadata = Some(json!({"provider_cache_ttl_minutes": 30}));
assert_eq!(usage.provider_cache_ttl_minutes(), Some(30));
}
#[test]
fn provider_cache_ttl_prefers_final_body_model() {
assert_eq!(
resolve_provider_cache_ttl_minutes(
Some("openai:responses"),
Some("gpt-5.5"),
Some("client-alias"),
Some(&json!({"model": "gpt-5.6-sol"})),
),
Some(30)
);
assert_eq!(
resolve_provider_cache_ttl_minutes(
Some("openai:responses"),
Some("gpt-5.6-sol"),
Some("client-alias"),
Some(&json!({"model": "gpt-5.5"})),
),
None
);
}
#[test]
fn actual_service_tier_is_independent_from_requested_tier() {
let mut usage = sample_usage();
crates/aether-usage-runtime/src/record.rs
+5
View File
@@ -40,6 +40,11 @@ pub fn build_upsert_usage_record_from_event(
let mut data = event.data.clone();
data.request_metadata = attach_provider_request_body_metadata(
data.request_metadata,
data.endpoint_api_format
.as_deref()
.or(data.api_format.as_deref()),
data.target_model.as_deref().or(Some(data.model.as_str())),
Some(data.model.as_str()),
data.provider_request_body.as_ref(),
);
let now_unix_secs = event.timestamp_ms / 1_000;
crates/aether-usage-runtime/src/request_metadata.rs
+34 -4
View File
@@ -6,7 +6,8 @@ use aether_contracts::ExecutionPlan;
use aether_data_contracts::repository::usage::{
extract_provider_actual_service_tier_from_response,
extract_provider_reasoning_effort_from_body, extract_provider_service_tier_from_body,
normalize_provider_service_tier, PROVIDER_ACTUAL_SERVICE_TIER_METADATA_KEY,
normalize_provider_service_tier, resolve_provider_cache_ttl_minutes,
PROVIDER_ACTUAL_SERVICE_TIER_METADATA_KEY, PROVIDER_CACHE_TTL_MINUTES_METADATA_KEY,
PROVIDER_REASONING_EFFORT_METADATA_KEY, PROVIDER_SERVICE_TIER_METADATA_KEY,
};
use serde_json::{json, Map, Value};
@@ -77,12 +78,25 @@ pub(crate) fn sanitize_usage_request_metadata_ref(value: Option<&Value>) -> Opti
pub(crate) fn attach_provider_request_body_metadata(
metadata: Option<Value>,
provider_api_format: Option<&str>,
provider_model: Option<&str>,
source_model: Option<&str>,
provider_request_body: Option<&Value>,
) -> Option<Value> {
let provider_body_is_object = provider_request_body.and_then(Value::as_object).is_some();
let reasoning_effort = extract_provider_reasoning_effort_from_body(provider_request_body);
let service_tier = extract_provider_service_tier_from_body(provider_request_body);
if !provider_body_is_object && reasoning_effort.is_none() && service_tier.is_none() {
let cache_ttl_minutes = resolve_provider_cache_ttl_minutes(
provider_api_format,
provider_model,
source_model,
provider_request_body,
);
if !provider_body_is_object
&& reasoning_effort.is_none()
&& service_tier.is_none()
&& cache_ttl_minutes.is_none()
{
return metadata;
}
let mut object = match metadata {
@@ -92,6 +106,7 @@ pub(crate) fn attach_provider_request_body_metadata(
if provider_body_is_object {
object.remove(PROVIDER_REASONING_EFFORT_METADATA_KEY);
object.remove(PROVIDER_SERVICE_TIER_METADATA_KEY);
object.remove(PROVIDER_CACHE_TTL_MINUTES_METADATA_KEY);
}
if let Some(reasoning_effort) = reasoning_effort {
object.insert(
@@ -105,6 +120,12 @@ pub(crate) fn attach_provider_request_body_metadata(
Value::String(service_tier),
);
}
if let Some(cache_ttl_minutes) = cache_ttl_minutes {
object.insert(
PROVIDER_CACHE_TTL_MINUTES_METADATA_KEY.to_string(),
Value::Number(cache_ttl_minutes.into()),
);
}
(!object.is_empty()).then_some(Value::Object(object))
}
@@ -161,6 +182,7 @@ fn copy_allowed_metadata_fields(source: &Map<String, Value>, target: &mut Map<St
copy_non_empty_string(source, target, PROVIDER_REASONING_EFFORT_METADATA_KEY);
copy_non_empty_string(source, target, PROVIDER_SERVICE_TIER_METADATA_KEY);
copy_non_empty_string(source, target, PROVIDER_ACTUAL_SERVICE_TIER_METADATA_KEY);
copy_number(source, target, PROVIDER_CACHE_TTL_MINUTES_METADATA_KEY);
copy_number(source, target, "provider_request_body_base64_bytes");
copy_number(source, target, "provider_response_body_base64_bytes");
copy_number(source, target, "client_response_body_base64_bytes");
@@ -211,6 +233,7 @@ fn move_allowed_metadata_fields(mut source: Map<String, Value>, target: &mut Map
target,
PROVIDER_ACTUAL_SERVICE_TIER_METADATA_KEY,
);
remove_number(&mut source, target, PROVIDER_CACHE_TTL_MINUTES_METADATA_KEY);
remove_number(&mut source, target, "provider_request_body_base64_bytes");
remove_number(&mut source, target, "provider_response_body_base64_bytes");
remove_number(&mut source, target, "client_response_body_base64_bytes");
@@ -817,8 +840,11 @@ mod tests {
let updated = attach_provider_request_body_metadata(
metadata.clone(),
Some("openai:responses"),
Some("gpt-5.6-sol"),
Some("gpt-5.6-sol"),
Some(&json!({
"model": "gpt-5",
"model": "gpt-5.6-sol",
"reasoning": { "effort": "low" },
"service_tier": "standard"
})),
@@ -830,12 +856,16 @@ mod tests {
json!({
"trace_id": "trace-1",
"provider_reasoning_effort": "low",
"provider_service_tier": "standard"
"provider_service_tier": "standard",
"provider_cache_ttl_minutes": 30
})
);
let cleared = attach_provider_request_body_metadata(
metadata,
Some("openai:responses"),
Some("gpt-5"),
Some("gpt-5"),
Some(&json!({
"model": "gpt-5"
})),
crates/aether-usage-runtime/src/write.rs
+7 -2
View File
@@ -691,8 +691,13 @@ fn build_terminal_usage_event_from_seed_impl(
} else {
merge_usage_request_metadata(request_metadata, audit_payload)
};
let request_metadata =
attach_provider_request_body_metadata(request_metadata, provider_request.as_ref());
let request_metadata = attach_provider_request_body_metadata(
request_metadata,
Some(provider_contract.as_str()),
target_model.as_deref().or(Some(model.as_str())),
Some(model.as_str()),
provider_request.as_ref(),
);
let mut data = UsageEventData {
user_id,