diff --git a/apps/aether-gateway/src/ai_serving/codex_context.rs b/apps/aether-gateway/src/ai_serving/codex_context.rs index ac45689ac..87cb13b99 100644 --- a/apps/aether-gateway/src/ai_serving/codex_context.rs +++ b/apps/aether-gateway/src/ai_serving/codex_context.rs @@ -1,15 +1,15 @@ use std::sync::{Arc, OnceLock}; use std::time::{SystemTime, UNIX_EPOCH}; -use aether_provider_transport::CodexFingerprintConvergenceContext; use http::{request::Parts, HeaderMap}; use serde_json::Value; use uuid::Uuid; +use crate::ai_serving::transport::ProviderOutboundRequestContext; use crate::client_session_affinity::codex_request_signals_from_request; #[derive(Debug, Clone)] -pub(crate) struct CodexFingerprintContextSlot(Arc>); +pub(crate) struct CodexFingerprintContextSlot(Arc>); impl Default for CodexFingerprintContextSlot { fn default() -> Self { @@ -18,11 +18,7 @@ impl Default for CodexFingerprintContextSlot { } impl CodexFingerprintContextSlot { - fn resolve( - &self, - headers: &HeaderMap, - body_json: &Value, - ) -> CodexFingerprintConvergenceContext { + fn resolve(&self, headers: &HeaderMap, body_json: &Value) -> ProviderOutboundRequestContext { self.0 .get_or_init(|| { build_codex_fingerprint_context(headers, body_json, Uuid::now_v7().to_string()) @@ -34,10 +30,10 @@ impl CodexFingerprintContextSlot { pub(crate) fn resolve_codex_fingerprint_context( parts: &Parts, body_json: &Value, -) -> CodexFingerprintConvergenceContext { +) -> ProviderOutboundRequestContext { if let Some(context) = parts .extensions - .get::() + .get::() .cloned() { return context; @@ -51,7 +47,7 @@ pub(crate) fn resolve_codex_fingerprint_context( pub(crate) fn install_codex_fingerprint_context_slot(parts: &mut Parts) { if parts .extensions - .get::() + .get::() .is_none() && parts .extensions @@ -67,11 +63,11 @@ pub(crate) fn install_codex_fingerprint_context_slot(parts: &mut Parts) { pub(crate) fn ensure_codex_fingerprint_context( parts: &mut Parts, body_json: &Value, -) -> CodexFingerprintConvergenceContext { +) -> ProviderOutboundRequestContext { let context = resolve_codex_fingerprint_context(parts, body_json); if parts .extensions - .get::() + .get::() .is_none() { parts.extensions.remove::(); @@ -84,7 +80,7 @@ pub(crate) fn attach_codex_logical_turn_context( parts: &mut Parts, body_json: &Value, logical_turn_id: &str, -) -> CodexFingerprintConvergenceContext { +) -> ProviderOutboundRequestContext { let context = build_codex_fingerprint_context(&parts.headers, body_json, logical_turn_id.to_string()); parts.extensions.remove::(); @@ -94,7 +90,7 @@ pub(crate) fn attach_codex_logical_turn_context( pub(crate) fn restore_codex_logical_turn_context( parts: &mut Parts, - context: &CodexFingerprintConvergenceContext, + context: &ProviderOutboundRequestContext, ) { parts.extensions.remove::(); parts.extensions.insert(context.clone()); @@ -104,10 +100,9 @@ fn build_codex_fingerprint_context( headers: &HeaderMap, body_json: &Value, logical_turn_id: String, -) -> CodexFingerprintConvergenceContext { +) -> ProviderOutboundRequestContext { let signals = codex_request_signals_from_request(headers, Some(body_json)); - let mut context = - CodexFingerprintConvergenceContext::new(logical_turn_id, current_unix_millis()); + let mut context = ProviderOutboundRequestContext::new(logical_turn_id, current_unix_millis()); if let Some(turn_id) = signals.turn_id { context = context.with_original_turn_id(turn_id); @@ -160,14 +155,14 @@ mod tests { assert_eq!(context.original_client_session_id(), Some("header-thread")); assert_eq!(context.original_prompt_cache_key(), Some("client-cache")); assert_eq!( - parts.extensions.get::(), + parts.extensions.get::(), Some(&context) ); } #[test] fn restored_context_wins_over_retry_request_signals() { - let original = CodexFingerprintConvergenceContext::new("logical-turn", 1234) + let original = ProviderOutboundRequestContext::new("logical-turn", 1234) .with_original_turn_id("original-turn") .with_original_client_session_id("original-thread") .with_original_prompt_cache_key("original-cache"); diff --git a/apps/aether-gateway/src/ai_serving/planner/decision_input.rs b/apps/aether-gateway/src/ai_serving/planner/decision_input.rs index 0de06e2d8..efa915eb0 100644 --- a/apps/aether-gateway/src/ai_serving/planner/decision_input.rs +++ b/apps/aether-gateway/src/ai_serving/planner/decision_input.rs @@ -13,7 +13,7 @@ use http::{HeaderMap, HeaderName, HeaderValue}; use serde_json::{json, Value}; use crate::ai_serving::planner::common::extract_standard_requested_model; -use crate::ai_serving::transport::CodexFingerprintConvergenceContext; +use crate::ai_serving::transport::ProviderOutboundRequestContext; use crate::ai_serving::{ ClientSurface, ExecutionRuntimeAuthContext, GatewayAuthApiKeySnapshot, GatewayCredentialCarrier, GatewayProviderTransportSnapshot, PlannerAppState, @@ -60,7 +60,7 @@ pub(crate) struct LocalRequestedModelDecisionInput { pub(crate) client_surface: Option, pub(crate) gateway_credential_carrier: Option, pub(crate) client_session_affinity: Option, - pub(crate) codex_fingerprint_context: Option, + pub(crate) provider_outbound_context: Option, pub(crate) routing_policy: Option, pub(crate) routing_trace_seed: Option, pub(crate) routing_context: Option, @@ -172,7 +172,7 @@ pub(crate) fn apply_provider_request_routing_policy_to_decision_with_websocket_m provider_api_format.as_str(), ); } - apply_codex_fingerprint_convergence_to_decision( + apply_provider_outbound_request_policies_to_decision( input, decision, transport, @@ -235,7 +235,7 @@ pub(crate) fn apply_provider_request_routing_policy_to_decision_with_websocket_m provider_api_format.as_str(), ); } - apply_codex_fingerprint_convergence_to_decision( + apply_provider_outbound_request_policies_to_decision( input, decision, transport, @@ -358,7 +358,7 @@ pub(crate) fn apply_provider_request_routing_policy_to_decision_with_websocket_m if original_provider_request_body.is_some() { decision.provider_request_body = Some(provider_request_body); } - apply_codex_fingerprint_convergence_to_decision( + apply_provider_outbound_request_policies_to_decision( input, decision, transport, @@ -368,7 +368,7 @@ pub(crate) fn apply_provider_request_routing_policy_to_decision_with_websocket_m Ok(()) } -fn apply_codex_fingerprint_convergence_to_decision( +fn apply_provider_outbound_request_policies_to_decision( input: &LocalRequestedModelDecisionInput, decision: &mut AiExecutionDecision, transport: Option<&GatewayProviderTransportSnapshot>, @@ -379,17 +379,17 @@ fn apply_codex_fingerprint_convergence_to_decision( else { return; }; - let Some(context) = input.codex_fingerprint_context.as_ref() else { + let Some(context) = input.provider_outbound_context.as_ref() else { return; }; - let applied = crate::ai_serving::transport::apply_codex_fingerprint_convergence_with_context( + let results = crate::ai_serving::transport::apply_provider_outbound_request_policies( transport, provider_api_format, context, &mut decision.provider_request_headers, provider_request_body, ); - if applied { + if results.iter().any(|result| result.was_applied()) { decision.prompt_cache_key = provider_request_body .get("prompt_cache_key") .and_then(Value::as_str) @@ -397,6 +397,31 @@ fn apply_codex_fingerprint_convergence_to_decision( .filter(|value| !value.is_empty()) .map(ToOwned::to_owned); } + if results.is_empty() { + return; + } + for result in &results { + tracing::debug!( + event_name = "provider_outbound_policy_evaluated", + log_type = "event", + policy = ?result.policy, + outcome = ?result.outcome, + reason = ?result.reason, + mutation_scope = ?result.mutation_scope, + identity_scope = ?result.identity_scope, + "provider outbound request policy evaluated" + ); + } + let Some(serde_json::Value::Object(report_context)) = decision.report_context.as_mut() else { + return; + }; + report_context.insert( + "provider_outbound_policies".to_string(), + serde_json::json!({ + "schema_version": 1, + "results": results, + }), + ); } struct GatewayAuthenticatedDecisionInputPort<'a> { @@ -485,7 +510,7 @@ pub(crate) fn build_local_requested_model_decision_input( client_surface: None, gateway_credential_carrier: None, client_session_affinity: None, - codex_fingerprint_context: None, + provider_outbound_context: None, routing_policy: None, routing_trace_seed: None, routing_context: None, @@ -500,7 +525,7 @@ pub(crate) async fn attach_routing_policy_to_local_requested_model_input( body_json: &Value, client_api_format: &str, ) -> Result<(), GatewayError> { - input.codex_fingerprint_context = + input.provider_outbound_context = Some(crate::ai_serving::codex_context::resolve_codex_fingerprint_context(parts, body_json)); let explicit_group = routing_header_value_str(&parts.headers, ROUTING_GROUP_HEADER); let selected_group = match state.routing_group_read_repository() { @@ -1390,7 +1415,7 @@ mod tests { client_surface: None, gateway_credential_carrier: None, client_session_affinity: None, - codex_fingerprint_context: None, + provider_outbound_context: None, routing_policy: None, routing_trace_seed: None, model_directive_policy: Default::default(), @@ -1639,7 +1664,7 @@ mod tests { client_surface: None, gateway_credential_carrier: None, client_session_affinity: None, - codex_fingerprint_context: None, + provider_outbound_context: None, routing_policy: None, routing_trace_seed: None, model_directive_policy: Default::default(), @@ -1709,7 +1734,7 @@ mod tests { client_surface: None, gateway_credential_carrier: None, client_session_affinity: None, - codex_fingerprint_context: None, + provider_outbound_context: None, routing_policy: None, routing_trace_seed: None, routing_context: None, @@ -1778,6 +1803,35 @@ mod tests { ); } + #[test] + fn non_codex_provider_outbound_policies_are_terminal_noop() { + let mut input = sample_decision_input(); + input.routing_context = None; + input.provider_outbound_context = Some(ProviderOutboundRequestContext::new( + "logical-turn", + 1_700_000_000_123, + )); + let mut decision = sample_codex_fingerprint_decision(); + decision.provider_type = Some("openai".to_string()); + decision.provider_api_format = Some("openai:responses".to_string()); + decision.client_api_format = Some("openai:responses".to_string()); + let mut transport = sample_codex_fingerprint_transport(); + transport.provider.provider_type = "openai".to_string(); + + let original_headers = decision.provider_request_headers.clone(); + let original_body = decision.provider_request_body.clone(); + apply_provider_request_routing_policy_to_decision(&input, &mut decision, Some(&transport)) + .expect("non-Codex terminal finalization should succeed"); + + assert_eq!(decision.provider_request_headers, original_headers); + assert_eq!(decision.provider_request_body, original_body); + assert!(decision + .report_context + .as_ref() + .and_then(|context| context.get("provider_outbound_policies")) + .is_none()); + } + #[test] fn codex_fingerprint_convergence_runs_at_every_provider_routing_success_exit() { let transport = sample_codex_fingerprint_transport(); @@ -1794,8 +1848,8 @@ mod tests { }); let mut with_mutation = sample_decision_input(); for input in [&mut no_context, &mut empty_mutation, &mut with_mutation] { - input.codex_fingerprint_context = Some( - CodexFingerprintConvergenceContext::new( + input.provider_outbound_context = Some( + ProviderOutboundRequestContext::new( uuid::Uuid::new_v4().to_string(), 1_756_668_000_000, ) @@ -1864,6 +1918,25 @@ mod tests { ); assert_eq!(body["client_metadata"]["x-codex-window-id"], window_id); + let policy_results = decision + .report_context + .as_ref() + .and_then(|context| context.get("provider_outbound_policies")) + .and_then(|policies| policies.get("results")) + .and_then(Value::as_array) + .expect("provider policy results"); + assert_eq!( + policy_results.len(), + 1, + "policy result count at {exit_name}" + ); + assert_eq!( + policy_results[0]["policy"], + json!("codex_fingerprint_convergence") + ); + assert_eq!(policy_results[0]["outcome"], json!("applied")); + assert_eq!(policy_results[0]["reason"], json!("applied")); + let header_metadata: Value = serde_json::from_str(&decision.provider_request_headers["x-codex-turn-metadata"]) .expect("header turn metadata"); diff --git a/apps/aether-gateway/src/ai_serving/planner/standard/family/payload.rs b/apps/aether-gateway/src/ai_serving/planner/standard/family/payload.rs index 85be2f3fe..bd0877b10 100644 --- a/apps/aether-gateway/src/ai_serving/planner/standard/family/payload.rs +++ b/apps/aether-gateway/src/ai_serving/planner/standard/family/payload.rs @@ -378,7 +378,7 @@ mod tests { client_surface: None, gateway_credential_carrier: None, client_session_affinity: None, - codex_fingerprint_context: None, + provider_outbound_context: None, routing_policy: None, routing_trace_seed: None, routing_context: None, diff --git a/apps/aether-gateway/src/ai_serving/planner/standard/openai/chat/decision/request.rs b/apps/aether-gateway/src/ai_serving/planner/standard/openai/chat/decision/request.rs index 36e3d8b4c..4bfa07dea 100644 --- a/apps/aether-gateway/src/ai_serving/planner/standard/openai/chat/decision/request.rs +++ b/apps/aether-gateway/src/ai_serving/planner/standard/openai/chat/decision/request.rs @@ -2183,7 +2183,7 @@ mod tests { client_surface: None, gateway_credential_carrier: None, client_session_affinity: None, - codex_fingerprint_context: None, + provider_outbound_context: None, routing_policy: None, routing_trace_seed: None, routing_context: None, diff --git a/apps/aether-gateway/src/ai_serving/transport.rs b/apps/aether-gateway/src/ai_serving/transport.rs index d75717444..97fe5434a 100644 --- a/apps/aether-gateway/src/ai_serving/transport.rs +++ b/apps/aether-gateway/src/ai_serving/transport.rs @@ -62,7 +62,8 @@ pub(crate) use aether_provider_transport::{ append_transport_diagnostics_to_value, apply_codex_fingerprint_convergence, apply_codex_fingerprint_convergence_with_context, apply_local_auth_config_header_overrides, apply_local_body_rules, apply_local_body_rules_with_request_headers, apply_local_header_rules, - apply_local_header_rules_with_request_headers, apply_standard_provider_request_body_rules, + apply_local_header_rules_with_request_headers, apply_provider_outbound_request_policies, + apply_standard_provider_request_body_rules, apply_standard_provider_request_body_rules_with_request_headers, apply_transport_request_body_semantics, body_rules_are_locally_supported, body_rules_handle_path, body_rules_have_enabled_rules, @@ -112,7 +113,11 @@ pub(crate) use aether_provider_transport::{ GeminiCliRequestAuthSupport, GeminiCliRequestAuthUnsupportedReason, GeminiCliRequestEnvelopeSupport, GeminiFilesHeadersInput, GeminiFilesRequestBodyError, GeminiFilesRequestBodyParts, GrokHeaderInput, LocalResolvedOAuthRequestAuth, - ProviderOpenAiImageHeadersInput, ProviderVideoCreateFamily, ProviderVideoCreateHeadersInput, + ProviderOpenAiImageHeadersInput, ProviderOutboundRequestContext, + ProviderOutboundRequestIdentityScope, ProviderOutboundRequestMutationScope, + ProviderOutboundRequestPolicy, ProviderOutboundRequestPolicyOutcome, + ProviderOutboundRequestPolicyReason, ProviderOutboundRequestPolicyResult, + ProviderVideoCreateFamily, ProviderVideoCreateHeadersInput, SameFormatProviderCompatibilityEdit, SameFormatProviderCompatibilityEditAction, SameFormatProviderFamily, SameFormatProviderHeadersInput, SameFormatProviderRequestBehavior, SameFormatProviderRequestBehaviorParams, SameFormatProviderRequestBodyInput, @@ -121,5 +126,5 @@ pub(crate) use aether_provider_transport::{ StandardProviderRequestHeaders, StandardProviderRequestHeadersInput, TransportRequestBodySemanticsError, TransportRequestUrlParams, GEMINI_CLI_USER_AGENT, GEMINI_CLI_V1INTERNAL_ENVELOPE_NAME, GROK_CHAT_PATH, GROK_INTERNAL_HEADER, - GROK_RATE_LIMITS_PATH, WINDSURF_ENVELOPE_NAME, + GROK_RATE_LIMITS_PATH, PROVIDER_OUTBOUND_CONTEXT_MAX_VALUE_BYTES, WINDSURF_ENVELOPE_NAME, }; diff --git a/apps/aether-gateway/src/handlers/proxy/websocket/live/http.rs b/apps/aether-gateway/src/handlers/proxy/websocket/live/http.rs index c728a0c64..45b71b063 100644 --- a/apps/aether-gateway/src/handlers/proxy/websocket/live/http.rs +++ b/apps/aether-gateway/src/handlers/proxy/websocket/live/http.rs @@ -189,6 +189,7 @@ async fn handle_live_http( client_model, dialect, None, + None, ) .await { @@ -243,6 +244,12 @@ async fn handle_live_http( ); }; let lease = LivePoolLeaseGuard::new(state, &candidate); + let provider_outbound_context = candidate + .execution + .provider_type + .as_deref() + .is_some_and(|provider_type| provider_type.trim().eq_ignore_ascii_case("codex")) + .then(|| candidate.provider_outbound_context.clone()); let binding = LiveCallBinding::from_candidate(&candidate); let mut provider_session = offer.session.clone(); provider_session @@ -528,6 +535,7 @@ async fn handle_live_http( auth_context.api_key_id.as_str(), call_id.as_str(), &binding, + provider_outbound_context.as_ref(), ) .await { diff --git a/apps/aether-gateway/src/handlers/proxy/websocket/live/planner.rs b/apps/aether-gateway/src/handlers/proxy/websocket/live/planner.rs index 83c951879..9019ff8cf 100644 --- a/apps/aether-gateway/src/handlers/proxy/websocket/live/planner.rs +++ b/apps/aether-gateway/src/handlers/proxy/websocket/live/planner.rs @@ -16,6 +16,7 @@ use serde_json::json; use sha2::{Digest, Sha256}; use url::{form_urlencoded, Url}; +use crate::ai_serving::transport::ProviderOutboundRequestContext; use crate::ai_serving::{ build_standard_stream_plan_from_decision, maybe_build_pinned_stream_local_same_format_provider_decision_payload, AiExecutionDecision, @@ -49,8 +50,7 @@ pub(super) enum LiveAuthMode { pub(super) struct PlannedLiveCandidate { pub(super) execution: AiExecutionDecision, pub(super) pinned_candidate: ResponsesWebSocketPinnedCandidate, - pub(super) codex_fingerprint_context: - aether_provider_transport::CodexFingerprintConvergenceContext, + pub(super) provider_outbound_context: ProviderOutboundRequestContext, pub(super) client_model: String, pub(super) provider_model: String, pub(super) auth_mode: LiveAuthMode, @@ -198,6 +198,7 @@ pub(super) async fn plan_live_candidate( client_model: &str, dialect: LiveRouteDialect, pinned_candidate: Option<&ResponsesWebSocketPinnedCandidate>, + provider_outbound_context: Option<&ProviderOutboundRequestContext>, ) -> Result { let diagnostic_guard = LiveRuntimeMissDiagnosticGuard::new(state, trace_id); let candidate = plan_live_candidate_inner( @@ -209,6 +210,7 @@ pub(super) async fn plan_live_candidate( client_model, dialect, pinned_candidate, + provider_outbound_context, ) .await?; Ok(LiveCandidatePlanningOutcome { @@ -226,13 +228,18 @@ async fn plan_live_candidate_inner( client_model: &str, dialect: LiveRouteDialect, pinned_candidate: Option<&ResponsesWebSocketPinnedCandidate>, + provider_outbound_context: Option<&ProviderOutboundRequestContext>, ) -> Result, GatewayError> { if validate_model(client_model).is_err() || client_model.len() > MAX_LIVE_MODEL_BYTES { return Ok(None); } let mut parts = build_live_planning_parts(headers, remote_addr); let body = json!({"model": client_model, "input": []}); - crate::ai_serving::codex_context::install_codex_fingerprint_context_slot(&mut parts); + if let Some(context) = provider_outbound_context { + crate::ai_serving::codex_context::restore_codex_logical_turn_context(&mut parts, context); + } else { + crate::ai_serving::codex_context::install_codex_fingerprint_context_slot(&mut parts); + } let execution = maybe_build_pinned_stream_local_same_format_provider_decision_payload( state, &parts, @@ -341,7 +348,7 @@ async fn plan_live_candidate_inner( Ok(Some(PlannedLiveCandidate { execution, pinned_candidate, - codex_fingerprint_context: + provider_outbound_context: crate::ai_serving::codex_context::resolve_codex_fingerprint_context(&parts, &body), client_model: client_model.to_string(), provider_model, @@ -568,7 +575,7 @@ pub(super) fn build_live_stream_admission_attempt( let mut parts = build_live_planning_parts(headers, remote_addr); crate::ai_serving::codex_context::restore_codex_logical_turn_context( &mut parts, - &candidate.codex_fingerprint_context, + &candidate.provider_outbound_context, ); let body = json!({"model": candidate.client_model.as_str(), "input": []}); let mut execution = candidate.execution.clone(); @@ -578,6 +585,7 @@ pub(super) fn build_live_stream_admission_attempt( } fn live_auth_mode(provider_type: &str, effective_auth_type: &str) -> Option { + let provider_type = provider_type.trim(); match effective_auth_type.trim().to_ascii_lowercase().as_str() { "api_key" | "bearer" => Some(LiveAuthMode::ApiKey), "oauth" if provider_type.eq_ignore_ascii_case("codex") => Some(LiveAuthMode::ChatGptOauth), @@ -931,11 +939,7 @@ mod tests { "key-1", ) .unwrap(), - codex_fingerprint_context: - aether_provider_transport::CodexFingerprintConvergenceContext::new( - "test-live-turn", - 1, - ), + provider_outbound_context: ProviderOutboundRequestContext::new("test-live-turn", 1), client_model: "global-model".to_string(), provider_model: "provider-model".to_string(), auth_mode, diff --git a/apps/aether-gateway/src/handlers/proxy/websocket/live/registry.rs b/apps/aether-gateway/src/handlers/proxy/websocket/live/registry.rs index b4bb0cc53..443936484 100644 --- a/apps/aether-gateway/src/handlers/proxy/websocket/live/registry.rs +++ b/apps/aether-gateway/src/handlers/proxy/websocket/live/registry.rs @@ -13,6 +13,9 @@ use sha2::{Digest, Sha256}; use tokio::sync::{oneshot, watch}; use tokio::task::JoinHandle; +use crate::ai_serving::transport::{ + ProviderOutboundRequestContext, PROVIDER_OUTBOUND_CONTEXT_MAX_VALUE_BYTES, +}; use crate::ai_serving::ResponsesWebSocketPinnedCandidate; use super::planner::{LiveAuthMode, PlannedLiveCandidate}; @@ -21,6 +24,9 @@ use super::protocol::validate_call_id; const SCHEMA_VERSION: u16 = 2; const RECORD_PREFIX: &str = "codex_live:call:v2:"; const RECORD_DOMAIN: &[u8] = b"aether-codex-live-call-v2"; +const CONTEXT_SCHEMA_VERSION: u16 = 1; +const CONTEXT_PREFIX: &str = "codex_live:call_context:v1:"; +const CONTEXT_DOMAIN: &[u8] = b"aether-codex-live-call-context-v1"; const INDEX_PREFIX: &str = "codex_live:call_index:v2:"; const INDEX_DOMAIN: &[u8] = b"aether-codex-live-call-index-v2"; const LOCK_PREFIX: &str = "codex_live:call_lock:v2:"; @@ -38,6 +44,8 @@ const LOCK_OWNER: &str = "codex_live_call_registry"; const SIDEBAND_LOCK_OWNER: &str = "codex_live_sideband_attachment"; const MAX_RECORDS_PER_PRINCIPAL: usize = 64; const MAX_SERIALIZED_RECORD_BYTES: usize = 4 * 1024; +const MAX_SERIALIZED_CONTEXT_BYTES: usize = 2 * 1024; +const MAX_CONTEXT_FIELD_BYTES: usize = PROVIDER_OUTBOUND_CONTEXT_MAX_VALUE_BYTES; const MAX_PRINCIPAL_BYTES: usize = 256; const MAX_RECORD_ID_BYTES: usize = 256; const LIVE_LOG_TARGET: &str = "aether_gateway::handlers::proxy::codex_live"; @@ -51,7 +59,7 @@ enum RegisterCommitState { #[derive(Debug, Clone, PartialEq, Eq)] pub(super) enum LiveCallLookup { - Found(LiveCallBinding), + Found(LiveCallRecord), Missing, Expired, } @@ -277,6 +285,118 @@ pub(super) struct LiveCallBinding { created_at_unix_ms: u64, } +#[derive(Debug, Clone, PartialEq, Eq)] +pub(super) struct LiveCallRecord { + binding: LiveCallBinding, + provider_outbound_context: Option, +} + +impl LiveCallRecord { + pub(super) fn binding(&self) -> &LiveCallBinding { + &self.binding + } + + pub(super) fn provider_outbound_context(&self) -> Option<&ProviderOutboundRequestContext> { + self.provider_outbound_context.as_ref() + } + + pub(super) fn matches_candidate(&self, candidate: &PlannedLiveCandidate) -> bool { + self.binding.matches_candidate(candidate) + && self + .provider_outbound_context + .as_ref() + .is_none_or(|context| context == &candidate.provider_outbound_context) + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct LiveProviderOutboundContextRecord { + schema_version: u16, + binding_created_at_unix_ms: u64, + logical_turn_id: String, + original_turn_id: Option, + original_client_session_id: Option, + original_prompt_cache_key: Option, + turn_started_at_unix_ms: u64, +} + +impl LiveProviderOutboundContextRecord { + fn from_context( + binding: &LiveCallBinding, + context: &ProviderOutboundRequestContext, + ) -> Result { + let record = Self { + schema_version: CONTEXT_SCHEMA_VERSION, + binding_created_at_unix_ms: binding.created_at_unix_ms, + logical_turn_id: context.logical_turn_id().to_string(), + original_turn_id: context.original_turn_id().map(ToOwned::to_owned), + original_client_session_id: context.original_client_session_id().map(ToOwned::to_owned), + original_prompt_cache_key: context.original_prompt_cache_key().map(ToOwned::to_owned), + turn_started_at_unix_ms: context.turn_started_at_unix_ms(), + }; + record.validate(binding)?; + Ok(record) + } + + fn into_context( + self, + binding: &LiveCallBinding, + ) -> Result { + self.validate(binding)?; + let mut context = + ProviderOutboundRequestContext::new(self.logical_turn_id, self.turn_started_at_unix_ms); + if let Some(value) = self.original_turn_id { + context = context.with_original_turn_id(value); + } + if let Some(value) = self.original_client_session_id { + context = context.with_original_client_session_id(value); + } + if let Some(value) = self.original_prompt_cache_key { + context = context.with_original_prompt_cache_key(value); + } + Ok(context) + } + + fn validate(&self, binding: &LiveCallBinding) -> Result<(), LiveCallRegistryError> { + if self.schema_version != CONTEXT_SCHEMA_VERSION { + return Err(LiveCallRegistryError::InvalidRecord( + "unsupported_context_schema_version", + )); + } + if self.binding_created_at_unix_ms != binding.created_at_unix_ms { + return Err(LiveCallRegistryError::InvalidRecord( + "context_binding_mismatch", + )); + } + if self.turn_started_at_unix_ms == 0 { + return Err(LiveCallRegistryError::InvalidRecord( + "invalid_context_started_at", + )); + } + validate_context_field( + self.logical_turn_id.as_str(), + "invalid_context_logical_turn", + )?; + for (value, error) in [ + (self.original_turn_id.as_deref(), "invalid_context_turn"), + ( + self.original_client_session_id.as_deref(), + "invalid_context_session", + ), + ( + self.original_prompt_cache_key.as_deref(), + "invalid_context_prompt_cache", + ), + ] { + if let Some(value) = value { + validate_context_field(value, error)?; + } + } + Ok(()) + } +} + impl LiveCallBinding { pub(super) fn from_candidate(candidate: &PlannedLiveCandidate) -> Self { Self { @@ -370,9 +490,11 @@ impl LiveCallRegistry { api_key_id: &str, call_id: &str, binding: &LiveCallBinding, + provider_outbound_context: Option<&ProviderOutboundRequestContext>, ) -> Result<(), LiveCallRegistryError> { binding.validate()?; let key = record_key(user_id, api_key_id, call_id)?; + let context_key = context_key(key.as_str())?; let index = index_key(user_id, api_key_id)?; let lock = lock_key(user_id, api_key_id)?; let serialized = @@ -380,12 +502,40 @@ impl LiveCallRegistry { if serialized.len() > MAX_SERIALIZED_RECORD_BYTES { return Err(LiveCallRegistryError::RecordTooLarge); } + let context_record = provider_outbound_context + .map(|context| LiveProviderOutboundContextRecord::from_context(binding, context)) + .transpose()?; + let serialized_context = context_record + .as_ref() + .map(serde_json::to_string) + .transpose() + .map_err(LiveCallRegistryError::Serialization)?; + if serialized_context + .as_ref() + .is_some_and(|serialized| serialized.len() > MAX_SERIALIZED_CONTEXT_BYTES) + { + return Err(LiveCallRegistryError::RecordTooLarge); + } let lease = self.acquire_lock(lock.as_str()).await?; let result = self - .register_locked(key.as_str(), index.as_str(), serialized, binding) + .register_locked( + key.as_str(), + context_key.as_str(), + index.as_str(), + serialized, + serialized_context, + binding, + context_record.as_ref(), + ) .await; let exact_binding_committed = if result.is_err() { - self.exact_binding_is_committed(key.as_str(), binding).await + self.exact_binding_is_committed( + key.as_str(), + context_key.as_str(), + binding, + context_record.as_ref(), + ) + .await } else { false }; @@ -425,13 +575,13 @@ impl LiveCallRegistry { user_id: &str, api_key_id: &str, call_id: &str, - ) -> Result, LiveCallRegistryError> { + ) -> Result, LiveCallRegistryError> { Ok( match self .lookup_with_status(user_id, api_key_id, call_id) .await? { - LiveCallLookup::Found(binding) => Some(binding), + LiveCallLookup::Found(record) => Some(record), LiveCallLookup::Missing | LiveCallLookup::Expired => None, }, ) @@ -444,18 +594,38 @@ impl LiveCallRegistry { call_id: &str, ) -> Result { let key = record_key(user_id, api_key_id, call_id)?; - if let Some(serialized) = self + let context_key = context_key(key.as_str())?; + let values = self .runtime_state - .kv_get(key.as_str()) + .kv_get_many(&[key.clone(), context_key]) .await - .map_err(LiveCallRegistryError::Storage)? - { + .map_err(LiveCallRegistryError::Storage)?; + if let Some(serialized) = values.first().and_then(Option::as_ref) { let binding = serde_json::from_str::(serialized.as_str()) .map_err(LiveCallRegistryError::CorruptRecord)?; binding.validate()?; - return Ok(LiveCallLookup::Found(binding)); + // Missing companion data is a rolling-upgrade compatible legacy + // v2 binding. Sideband planning will use the existing fallback + // behavior until that short-lived binding expires. + let provider_outbound_context = values + .get(1) + .and_then(Option::as_ref) + .map(|serialized| { + serde_json::from_str::(serialized.as_str()) + .map_err(LiveCallRegistryError::CorruptRecord)? + .into_context(&binding) + }) + .transpose()?; + return Ok(LiveCallLookup::Found(LiveCallRecord { + binding, + provider_outbound_context, + })); } + // A companion may outlive the base record by the expiry grace period, + // or be visible briefly before registration commits the base. Do not + // delete it from a lock-free lookup; the bounded TTL and the next + // locked registration handle cleanup without racing the writer. let index = index_key(user_id, api_key_id)?; let indexed = self .runtime_state @@ -521,48 +691,121 @@ impl LiveCallRegistry { } } - async fn exact_binding_is_committed(&self, key: &str, expected: &LiveCallBinding) -> bool { - let stored = - match tokio::time::timeout(COMMIT_VERIFY_TIMEOUT, self.runtime_state.kv_get(key)).await - { - Ok(Ok(Some(stored))) => stored, - Ok(Ok(None) | Err(_)) | Err(_) => return false, - }; - let Ok(actual) = serde_json::from_str::(stored.as_str()) else { + async fn exact_binding_is_committed( + &self, + key: &str, + context_storage_key: &str, + expected: &LiveCallBinding, + expected_context: Option<&LiveProviderOutboundContextRecord>, + ) -> bool { + let keys = [key.to_string(), context_storage_key.to_string()]; + let stored = match tokio::time::timeout( + COMMIT_VERIFY_TIMEOUT, + self.runtime_state.kv_get_many(&keys), + ) + .await + { + Ok(Ok(stored)) => stored, + Ok(Err(_)) | Err(_) => return false, + }; + let Some(Some(stored_binding)) = stored.first() else { return false; }; - actual.validate().is_ok() && actual == *expected + let Ok(actual) = serde_json::from_str::(stored_binding.as_str()) else { + return false; + }; + if actual.validate().is_err() || actual != *expected { + return false; + } + match (stored.get(1).and_then(Option::as_ref), expected_context) { + (None, None) => true, + (Some(stored), Some(expected)) => { + serde_json::from_str::(stored.as_str()) + .is_ok_and(|actual| actual == *expected) + } + _ => false, + } } async fn register_locked( &self, key: &str, + context_storage_key: &str, index: &str, serialized: String, + serialized_context: Option, binding: &LiveCallBinding, + context_record: Option<&LiveProviderOutboundContextRecord>, ) -> Result<(), LiveCallRegistryError> { - if let Some(existing) = self + let existing = self .runtime_state - .kv_get(key) + .kv_get_many(&[key.to_string(), context_storage_key.to_string()]) .await - .map_err(LiveCallRegistryError::Storage)? - { + .map_err(LiveCallRegistryError::Storage)?; + let existing_binding = existing.first().and_then(Option::as_ref); + if let Some(existing) = existing_binding { let existing = serde_json::from_str::(existing.as_str()) .map_err(LiveCallRegistryError::CorruptRecord)?; if existing != *binding { return Err(LiveCallRegistryError::OwnershipConflict); } } - self.runtime_state + let existing_context = existing.get(1).and_then(Option::as_ref); + if existing_binding.is_some() { + match (existing_context, context_record) { + (Some(existing), Some(expected)) => { + let existing = serde_json::from_str::( + existing.as_str(), + ) + .map_err(LiveCallRegistryError::CorruptRecord)?; + if existing != *expected { + return Err(LiveCallRegistryError::OwnershipConflict); + } + } + (Some(_), None) => return Err(LiveCallRegistryError::OwnershipConflict), + _ => {} + } + } else if existing_context.is_some() && serialized_context.is_none() { + // Old binaries and interrupted writes can leave a companion after + // the authoritative base record is gone. Clear it under the + // principal lock before committing a legacy/no-context binding. + self.runtime_state + .kv_delete(context_storage_key) + .await + .map_err(LiveCallRegistryError::Storage)?; + } + if let Some(serialized_context) = serialized_context { + self.runtime_state + // Keep the companion alive through the base record's expiry + // grace period. This prevents a normal TTL race from causing + // a sideband retry to mint a different provider identity. + .kv_set( + context_storage_key, + serialized_context, + Some(self.ttl.saturating_add(EXPIRED_LOOKUP_GRACE)), + ) + .await + .map_err(LiveCallRegistryError::Storage)?; + } + if let Err(error) = self + .runtime_state .kv_set(key, serialized, Some(self.ttl)) .await - .map_err(LiveCallRegistryError::Storage)?; + { + // Keep the companion until exact commit verification. A Redis + // timeout can happen after the base write committed; deleting the + // companion here would turn an idempotent retry into a split record. + // If the base did not commit, the bounded companion expires with + // the registry grace period and a later retry can safely replace it. + return Err(LiveCallRegistryError::Storage(error)); + } if let Err(error) = self .runtime_state .score_set(index, key, now_unix_ms() as f64) .await { let _ = self.runtime_state.kv_delete(key).await; + let _ = self.runtime_state.kv_delete(context_storage_key).await; return Err(LiveCallRegistryError::Storage(error)); } if let Err(error) = self @@ -572,6 +815,7 @@ impl LiveCallRegistry { { let _ = self.runtime_state.score_remove(index, key).await; let _ = self.runtime_state.kv_delete(key).await; + let _ = self.runtime_state.kv_delete(context_storage_key).await; return Err(LiveCallRegistryError::Storage(error)); } let members = self @@ -581,10 +825,15 @@ impl LiveCallRegistry { .map_err(LiveCallRegistryError::Storage)?; let overflow = members.len().saturating_sub(self.max_records_per_principal); for oldest in members.into_iter().take(overflow) { + let oldest_context = context_key(oldest.as_str())?; self.runtime_state .kv_delete(oldest.as_str()) .await .map_err(LiveCallRegistryError::Storage)?; + self.runtime_state + .kv_delete(oldest_context.as_str()) + .await + .map_err(LiveCallRegistryError::Storage)?; self.runtime_state .score_remove(index, oldest.as_str()) .await @@ -621,6 +870,21 @@ fn record_key( )) } +fn context_key(record_key: &str) -> Result { + if !record_key.starts_with(RECORD_PREFIX) + || record_key.len() != RECORD_PREFIX.len() + 64 + || !record_key[RECORD_PREFIX.len()..] + .bytes() + .all(|byte| byte.is_ascii_hexdigit()) + { + return Err(LiveCallRegistryError::InvalidIdentity("invalid_record_key")); + } + Ok(format!( + "{CONTEXT_PREFIX}{}", + digest(CONTEXT_DOMAIN, &[record_key]) + )) +} + fn index_key(user_id: &str, api_key_id: &str) -> Result { validate_principal(user_id, "invalid_user_id")?; validate_principal(api_key_id, "invalid_api_key_id")?; @@ -660,6 +924,19 @@ fn validate_principal(value: &str, error: &'static str) -> Result<(), LiveCallRe Ok(()) } +fn validate_context_field(value: &str, error: &'static str) -> Result<(), LiveCallRegistryError> { + let encoded_len = serde_json::to_string(value) + .map(|encoded| encoded.len()) + .unwrap_or(usize::MAX); + if value.trim().is_empty() + || value.len() > MAX_CONTEXT_FIELD_BYTES + || encoded_len > MAX_CONTEXT_FIELD_BYTES.saturating_add(2) + { + return Err(LiveCallRegistryError::InvalidRecord(error)); + } + Ok(()) +} + fn digest(domain: &[u8], components: &[&str]) -> String { let mut digest = Sha256::new(); digest.update(domain); @@ -704,6 +981,13 @@ mod tests { } } + fn provider_outbound_context(logical_turn_id: &str) -> ProviderOutboundRequestContext { + ProviderOutboundRequestContext::new(logical_turn_id, 1_700_000_000_123) + .with_original_turn_id("client-turn") + .with_original_client_session_id("client-session") + .with_original_prompt_cache_key("client-cache") + } + fn candidate_for_binding(binding: &LiveCallBinding) -> PlannedLiveCandidate { let execution: crate::ai_serving::AiExecutionDecision = serde_json::from_value(serde_json::json!({ @@ -718,11 +1002,7 @@ mod tests { PlannedLiveCandidate { execution, pinned_candidate: binding.pinned_candidate.clone(), - codex_fingerprint_context: - aether_provider_transport::CodexFingerprintConvergenceContext::new( - "test-live-turn", - 1, - ), + provider_outbound_context: ProviderOutboundRequestContext::new("test-live-turn", 1), client_model: binding.client_model.clone(), provider_model: binding.provider_model.clone(), auth_mode: binding.auth_mode, @@ -735,7 +1015,13 @@ mod tests { let state = runtime_state(); let registry = LiveCallRegistry::new(Arc::clone(&state)); registry - .register("user-1", "api-key-1", "rtc_secret", &binding("global")) + .register( + "user-1", + "api-key-1", + "rtc_secret", + &binding("global"), + None, + ) .await .unwrap(); assert!(registry @@ -755,14 +1041,117 @@ mod tests { .is_none()); } + #[tokio::test] + async fn provider_outbound_context_round_trips_in_companion_record() { + let state = runtime_state(); + let registry = LiveCallRegistry::new(Arc::clone(&state)); + let binding = binding("global"); + let context = provider_outbound_context("logical-live-turn"); + registry + .register("user", "key", "rtc_context", &binding, Some(&context)) + .await + .unwrap(); + + let record = registry + .lookup("user", "key", "rtc_context") + .await + .unwrap() + .expect("binding should exist"); + assert_eq!(record.binding(), &binding); + assert_eq!(record.provider_outbound_context(), Some(&context)); + + let key = record_key("user", "key", "rtc_context").unwrap(); + let serialized_binding = state.kv_get(key.as_str()).await.unwrap().unwrap(); + let strict_v2 = serde_json::from_str::(&serialized_binding).unwrap(); + assert_eq!(strict_v2, binding); + assert!(!serialized_binding.contains("provider_outbound_context")); + assert!(!serialized_binding.contains("logical_turn_id")); + } + + #[test] + fn bounded_context_fields_fit_the_companion_record_budget() { + let binding = binding("global"); + let value = "x".repeat(PROVIDER_OUTBOUND_CONTEXT_MAX_VALUE_BYTES); + let context = ProviderOutboundRequestContext::new(value.clone(), 1) + .with_original_turn_id(value.clone()) + .with_original_client_session_id(value.clone()) + .with_original_prompt_cache_key(value); + let record = LiveProviderOutboundContextRecord::from_context(&binding, &context) + .expect("bounded context should be valid"); + let serialized = serde_json::to_string(&record).expect("context should serialize"); + assert!(serialized.len() <= MAX_SERIALIZED_CONTEXT_BYTES); + } + + #[tokio::test] + async fn legacy_binding_without_companion_context_still_loads() { + let state = runtime_state(); + let registry = LiveCallRegistry::new(Arc::clone(&state)); + registry + .register("user", "key", "rtc_legacy", &binding("global"), None) + .await + .unwrap(); + + let record = registry + .lookup("user", "key", "rtc_legacy") + .await + .unwrap() + .expect("legacy binding should exist"); + assert!(record.provider_outbound_context().is_none()); + } + + #[tokio::test] + async fn corrupt_or_mismatched_companion_context_is_rejected() { + let state = runtime_state(); + let registry = LiveCallRegistry::new(Arc::clone(&state)); + let binding = binding("global"); + let context = provider_outbound_context("logical-live-turn"); + registry + .register("user", "key", "rtc_corrupt", &binding, Some(&context)) + .await + .unwrap(); + let key = record_key("user", "key", "rtc_corrupt").unwrap(); + let context_key = context_key(key.as_str()).unwrap(); + state + .kv_set( + context_key.as_str(), + "not-json", + Some(Duration::from_secs(60)), + ) + .await + .unwrap(); + assert!(matches!( + registry.lookup("user", "key", "rtc_corrupt").await, + Err(LiveCallRegistryError::CorruptRecord(_)) + )); + + let mut mismatch = LiveProviderOutboundContextRecord::from_context(&binding, &context) + .expect("context should be valid"); + mismatch.binding_created_at_unix_ms = mismatch.binding_created_at_unix_ms.saturating_add(1); + state + .kv_set( + context_key.as_str(), + serde_json::to_string(&mismatch).unwrap(), + Some(Duration::from_secs(60)), + ) + .await + .unwrap(); + assert!(matches!( + registry.lookup("user", "key", "rtc_corrupt").await, + Err(LiveCallRegistryError::InvalidRecord( + "context_binding_mismatch" + )) + )); + } + #[tokio::test] async fn capacity_evicts_the_oldest_binding() { let state = runtime_state(); let registry = LiveCallRegistry::with_limits(Arc::clone(&state), Duration::from_secs(60), 2); + let context = provider_outbound_context("logical-live-turn"); for call_id in ["rtc_1", "rtc_2", "rtc_3"] { registry - .register("user", "key", call_id, &binding(call_id)) + .register("user", "key", call_id, &binding(call_id), Some(&context)) .await .unwrap(); tokio::time::sleep(Duration::from_millis(2)).await; @@ -772,6 +1161,12 @@ mod tests { .await .unwrap() .is_none()); + let evicted_key = record_key("user", "key", "rtc_1").unwrap(); + assert!(state + .kv_get(context_key(evicted_key.as_str()).unwrap().as_str()) + .await + .unwrap() + .is_none()); assert!(registry .lookup("user", "key", "rtc_2") .await @@ -789,8 +1184,15 @@ mod tests { let state = runtime_state(); let registry = LiveCallRegistry::with_limits(Arc::clone(&state), Duration::from_millis(5), 2); + let context = provider_outbound_context("logical-live-turn"); registry - .register("user", "key", "rtc_expiring", &binding("global")) + .register( + "user", + "key", + "rtc_expiring", + &binding("global"), + Some(&context), + ) .await .unwrap(); tokio::time::sleep(Duration::from_millis(20)).await; @@ -806,6 +1208,53 @@ mod tests { .unwrap(), LiveCallLookup::Expired ); + let expired_key = record_key("user", "key", "rtc_expiring").unwrap(); + assert!(state + .kv_get(context_key(expired_key.as_str()).unwrap().as_str()) + .await + .unwrap() + .is_some()); + } + + #[tokio::test] + async fn orphan_companion_does_not_block_a_new_binding() { + let state = runtime_state(); + let registry = LiveCallRegistry::new(Arc::clone(&state)); + let old_binding = binding("old"); + let old_context = provider_outbound_context("old-turn"); + let key = record_key("user", "key", "rtc_reused").unwrap(); + let context_storage_key = context_key(key.as_str()).unwrap(); + let old_record = + LiveProviderOutboundContextRecord::from_context(&old_binding, &old_context).unwrap(); + state + .kv_set( + context_storage_key.as_str(), + serde_json::to_string(&old_record).unwrap(), + Some(Duration::from_secs(60)), + ) + .await + .unwrap(); + + let new_binding = binding("new"); + let new_context = provider_outbound_context("new-turn"); + registry + .register( + "user", + "key", + "rtc_reused", + &new_binding, + Some(&new_context), + ) + .await + .unwrap(); + + let record = registry + .lookup("user", "key", "rtc_reused") + .await + .unwrap() + .expect("new binding should be committed"); + assert_eq!(record.binding(), &new_binding); + assert_eq!(record.provider_outbound_context(), Some(&new_context)); } #[tokio::test] @@ -814,19 +1263,22 @@ mod tests { let registry = LiveCallRegistry::new(Arc::clone(&state)); let original = binding("global-a"); registry - .register("user", "key", "rtc_shared", &original) + .register("user", "key", "rtc_shared", &original, None) .await .unwrap(); assert!(matches!( registry - .register("user", "key", "rtc_shared", &binding("global-b")) + .register("user", "key", "rtc_shared", &binding("global-b"), None,) .await, Err(LiveCallRegistryError::OwnershipConflict) )); assert_eq!( registry.lookup("user", "key", "rtc_shared").await.unwrap(), - Some(original) + Some(LiveCallRecord { + binding: original, + provider_outbound_context: None, + }) ); } @@ -947,7 +1399,7 @@ mod tests { for call_id in [".", "..", "rtc/escape"] { assert!(matches!( registry - .register("user", "key", call_id, &binding("global")) + .register("user", "key", call_id, &binding("global"), None) .await, Err(LiveCallRegistryError::InvalidIdentity("invalid_call_id")) )); @@ -992,11 +1444,17 @@ mod tests { let state = runtime_state(); let registry = LiveCallRegistry::new(Arc::clone(&state)); let key = record_key("user", "key", "rtc_verify").unwrap(); + let context_storage_key = context_key(key.as_str()).unwrap(); let expected = binding("global"); assert!( !registry - .exact_binding_is_committed(key.as_str(), &expected) + .exact_binding_is_committed( + key.as_str(), + context_storage_key.as_str(), + &expected, + None + ) .await ); @@ -1006,7 +1464,12 @@ mod tests { .unwrap(); assert!( !registry - .exact_binding_is_committed(key.as_str(), &expected) + .exact_binding_is_committed( + key.as_str(), + context_storage_key.as_str(), + &expected, + None + ) .await ); @@ -1020,7 +1483,12 @@ mod tests { .unwrap(); assert!( !registry - .exact_binding_is_committed(key.as_str(), &expected) + .exact_binding_is_committed( + key.as_str(), + context_storage_key.as_str(), + &expected, + None + ) .await ); @@ -1034,7 +1502,12 @@ mod tests { .unwrap(); assert!( registry - .exact_binding_is_committed(key.as_str(), &expected) + .exact_binding_is_committed( + key.as_str(), + context_storage_key.as_str(), + &expected, + None + ) .await ); } diff --git a/apps/aether-gateway/src/handlers/proxy/websocket/live/session.rs b/apps/aether-gateway/src/handlers/proxy/websocket/live/session.rs index 747d67557..3f65bc299 100644 --- a/apps/aether-gateway/src/handlers/proxy/websocket/live/session.rs +++ b/apps/aether-gateway/src/handlers/proxy/websocket/live/session.rs @@ -43,7 +43,7 @@ use super::protocol::{ LEGACY_LIVE_CALL_PATH, REALTIME_SIDEBAND_PATH, }; use super::registry::{ - LiveCallBinding, LiveCallLookup, LiveCallRegistry, LiveCallRegistryError, LiveSidebandLease, + LiveCallLookup, LiveCallRecord, LiveCallRegistry, LiveCallRegistryError, LiveSidebandLease, LiveSidebandLeaseLoss, }; @@ -315,8 +315,8 @@ pub(super) async fn prepare_live_websocket( ), ) .await; - let binding = match lookup { - Ok(Ok(LiveCallLookup::Found(binding))) => binding, + let record = match lookup { + Ok(Ok(LiveCallLookup::Found(record))) => record, Ok(Ok(LiveCallLookup::Missing)) => { info!( target: LIVE_LOG_TARGET, @@ -373,7 +373,7 @@ pub(super) async fn prepare_live_websocket( )); } }; - prepare_sideband_live_websocket(state, context, call_id, binding, dialect) + prepare_sideband_live_websocket(state, context, call_id, record, dialect) .await .map(PreparedLiveWebSocket::Sideband) } @@ -394,6 +394,7 @@ async fn prepare_direct_live_websocket( client_model, dialect, None, + None, ) .await { @@ -570,10 +571,11 @@ async fn prepare_sideband_live_websocket( state: &AppState, context: &WebSocketRequestContext, call_id: String, - binding: LiveCallBinding, + record: LiveCallRecord, dialect: LiveRouteDialect, ) -> Result { let started_at = Instant::now(); + let binding = record.binding(); let Some(auth) = context.decision.auth_context.as_ref() else { return Err(preflight_rejection( context, @@ -648,6 +650,7 @@ async fn prepare_sideband_live_websocket( binding.client_model(), dialect, Some(binding.pinned_candidate()), + record.provider_outbound_context(), ), ) .await; @@ -665,7 +668,7 @@ async fn prepare_sideband_live_websocket( Ok(Ok(LiveCandidatePlanningOutcome { candidate: Some(candidate), .. - })) if binding.matches_candidate(&candidate) => candidate, + })) if record.matches_candidate(&candidate) => candidate, Ok(Ok(LiveCandidatePlanningOutcome { candidate: Some(candidate), .. diff --git a/crates/aether-provider/transport/src/codex_fingerprint.rs b/crates/aether-provider/transport/src/codex_fingerprint.rs index b22e642d5..ab0b4355a 100644 --- a/crates/aether-provider/transport/src/codex_fingerprint.rs +++ b/crates/aether-provider/transport/src/codex_fingerprint.rs @@ -5,72 +5,17 @@ use serde_json::{Map, Value}; use sha2::{Digest, Sha256}; use uuid::Uuid; +use crate::outbound_request_policy::{ + ProviderOutboundRequestContext, ProviderOutboundRequestIdentityScope, + ProviderOutboundRequestMutationScope, ProviderOutboundRequestPolicy, + ProviderOutboundRequestPolicyReason, ProviderOutboundRequestPolicyResult, +}; use crate::snapshot::GatewayProviderTransportSnapshot; pub const CODEX_FINGERPRINT_CONFIG_NAMESPACE: &str = "codex"; pub const CODEX_FINGERPRINT_ENABLED_CONFIG_KEY: &str = "fingerprint_convergence_enabled"; -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct CodexFingerprintConvergenceContext { - logical_turn_id: String, - original_turn_id: Option, - original_client_session_id: Option, - original_prompt_cache_key: Option, - turn_started_at_unix_ms: u64, -} - -impl CodexFingerprintConvergenceContext { - pub fn new(logical_turn_id: impl Into, turn_started_at_unix_ms: u64) -> Self { - Self { - logical_turn_id: logical_turn_id.into().trim().to_string(), - original_turn_id: None, - original_client_session_id: None, - original_prompt_cache_key: None, - turn_started_at_unix_ms, - } - } - - pub fn with_original_turn_id(mut self, original_turn_id: impl Into) -> Self { - self.original_turn_id = non_empty_owned(original_turn_id.into()); - self - } - - pub fn with_original_client_session_id( - mut self, - original_client_session_id: impl Into, - ) -> Self { - self.original_client_session_id = non_empty_owned(original_client_session_id.into()); - self - } - - pub fn with_original_prompt_cache_key( - mut self, - original_prompt_cache_key: impl Into, - ) -> Self { - self.original_prompt_cache_key = non_empty_owned(original_prompt_cache_key.into()); - self - } - - pub fn logical_turn_id(&self) -> &str { - self.logical_turn_id.as_str() - } - - pub fn original_turn_id(&self) -> Option<&str> { - self.original_turn_id.as_deref() - } - - pub fn original_client_session_id(&self) -> Option<&str> { - self.original_client_session_id.as_deref() - } - - pub fn original_prompt_cache_key(&self) -> Option<&str> { - self.original_prompt_cache_key.as_deref() - } - - pub fn turn_started_at_unix_ms(&self) -> u64 { - self.turn_started_at_unix_ms - } -} +pub type CodexFingerprintConvergenceContext = ProviderOutboundRequestContext; #[derive(Debug, Clone, PartialEq, Eq)] struct CodexConvergedFingerprint { @@ -125,36 +70,82 @@ pub fn apply_codex_fingerprint_convergence_with_context( provider_request_headers: &mut BTreeMap, provider_request_body: &mut Value, ) -> bool { - let is_responses = aether_ai_formats::is_openai_responses_format(provider_api_format); - let is_live = aether_ai_formats::api_format_alias_matches(provider_api_format, "codex:live"); - // Convergence is a Codex provider policy, independent of whether the key - // uses OAuth, an API key, or another ordinary auth channel. Agent Identity - // uses a separate signed-identity protocol and is excluded here. + apply_codex_fingerprint_convergence_policy( + transport, + provider_api_format, + context, + provider_request_headers, + provider_request_body, + ) + .was_applied() +} + +pub(crate) fn apply_codex_fingerprint_convergence_policy( + transport: &GatewayProviderTransportSnapshot, + provider_api_format: &str, + context: &ProviderOutboundRequestContext, + provider_request_headers: &mut BTreeMap, + provider_request_body: &mut Value, +) -> ProviderOutboundRequestPolicyResult { + let policy = ProviderOutboundRequestPolicy::CodexFingerprintConvergence; if !transport .provider .provider_type .trim() .eq_ignore_ascii_case("codex") - || crate::agent_identity::is_codex_agent_identity_transport(transport) - || (!is_responses && !is_live) - || is_responses - && aether_ai_formats::openai_responses_request_operation( - provider_api_format, - provider_request_body, - ) == Some(aether_ai_formats::OPENAI_RESPONSES_OPERATION_COMPACT) - || !codex_fingerprint_convergence_enabled( - transport.provider.provider_type.as_str(), - transport.provider.config.as_ref(), - ) - || !provider_request_body.is_object() { - return false; + return ProviderOutboundRequestPolicyResult::skipped( + policy, + ProviderOutboundRequestPolicyReason::ProviderTypeMismatch, + ); + } + if crate::agent_identity::is_codex_agent_identity_transport(transport) { + return ProviderOutboundRequestPolicyResult::skipped( + policy, + ProviderOutboundRequestPolicyReason::AgentIdentityExcluded, + ); + } + + let is_responses = aether_ai_formats::is_openai_responses_format(provider_api_format); + let is_live = aether_ai_formats::api_format_alias_matches(provider_api_format, "codex:live"); + if !is_responses && !is_live { + return ProviderOutboundRequestPolicyResult::skipped( + policy, + ProviderOutboundRequestPolicyReason::UnsupportedApiFormat, + ); + } + if is_responses + && aether_ai_formats::openai_responses_request_operation( + provider_api_format, + provider_request_body, + ) == Some(aether_ai_formats::OPENAI_RESPONSES_OPERATION_COMPACT) + { + return ProviderOutboundRequestPolicyResult::skipped( + policy, + ProviderOutboundRequestPolicyReason::CompactOperationExcluded, + ); + } + if !codex_fingerprint_convergence_enabled( + transport.provider.provider_type.as_str(), + transport.provider.config.as_ref(), + ) { + return ProviderOutboundRequestPolicyResult::skipped( + policy, + ProviderOutboundRequestPolicyReason::Disabled, + ); + } + if !provider_request_body.is_object() { + return ProviderOutboundRequestPolicyResult::skipped( + policy, + ProviderOutboundRequestPolicyReason::RequestBodyNotObject, + ); } let auth_identity = aether_ai_formats::parse_codex_auth_identity( transport.key.decrypted_auth_config.as_deref(), ); - let account_seed = resolve_codex_account_seed(&auth_identity, transport.key.id.as_str()); + let (account_seed, identity_scope) = + resolve_codex_account_seed_with_scope(&auth_identity, transport.key.id.as_str()); // Only namespace a cache key that survived all provider-body conversion and // routing rules. The client-side value in `context` is a retry signal, not // permission to resurrect a field that the terminal body deliberately @@ -178,7 +169,15 @@ pub fn apply_codex_fingerprint_convergence_with_context( if is_responses { apply_converged_client_metadata(provider_request_body, &fingerprint); } - true + ProviderOutboundRequestPolicyResult::applied( + policy, + if is_responses { + ProviderOutboundRequestMutationScope::HeadersAndBody + } else { + ProviderOutboundRequestMutationScope::Headers + }, + identity_scope, + ) } #[cfg(test)] @@ -242,11 +241,6 @@ fn resolve_converged_fingerprint_with_prompt_cache( } } -fn non_empty_owned(value: String) -> Option { - let value = value.trim(); - (!value.is_empty()).then(|| value.to_string()) -} - fn current_unix_millis() -> u64 { SystemTime::now() .duration_since(UNIX_EPOCH) @@ -261,32 +255,62 @@ fn normalized_identity_part(value: Option<&str>) -> Option { .map(str::to_ascii_lowercase) } +#[cfg(test)] fn resolve_codex_account_seed( identity: &aether_ai_formats::CodexAuthIdentity, fallback_key_id: &str, ) -> String { - let fingerprint = normalized_identity_part(identity.codex_identity_fingerprint.as_deref()) - .or_else(|| { - aether_oauth::provider::providers::derive_codex_identity_fingerprint( - identity.account_id.as_deref(), - identity.account_user_id.as_deref(), - identity.user_id.as_deref(), - identity.email.as_deref(), - ) - }); - if let Some(fingerprint) = fingerprint { - return format!("persisted:v1:{fingerprint}"); + resolve_codex_account_seed_with_scope(identity, fallback_key_id).0 +} + +fn resolve_codex_account_seed_with_scope( + identity: &aether_ai_formats::CodexAuthIdentity, + fallback_key_id: &str, +) -> (String, ProviderOutboundRequestIdentityScope) { + if let Some(fingerprint) = + normalized_identity_part(identity.codex_identity_fingerprint.as_deref()) + { + return ( + format!("persisted:v1:{fingerprint}"), + ProviderOutboundRequestIdentityScope::PersistedFingerprint, + ); } let account = normalized_identity_part(identity.account_id.as_deref()); let member = normalized_identity_part(identity.account_user_id.as_deref()) .or_else(|| normalized_identity_part(identity.user_id.as_deref())) .or_else(|| normalized_identity_part(identity.email.as_deref())); + if let Some(fingerprint) = aether_oauth::provider::providers::derive_codex_identity_fingerprint( + account.as_deref(), + member.as_deref(), + None, + None, + ) { + let scope = if account.is_some() { + ProviderOutboundRequestIdentityScope::AccountMember + } else { + ProviderOutboundRequestIdentityScope::Member + }; + return (format!("persisted:v1:{fingerprint}"), scope); + } + match (account, member) { - (Some(account), Some(member)) => format!("account-member:v1:{account}\0{member}"), - (None, Some(member)) => format!("member:v1:{member}"), - (Some(account), None) => format!("account:v1:{account}"), - (None, None) => format!("key:v1:{}", fallback_key_id.trim()), + (Some(account), Some(member)) => ( + format!("account-member:v1:{account}\0{member}"), + ProviderOutboundRequestIdentityScope::AccountMember, + ), + (None, Some(member)) => ( + format!("member:v1:{member}"), + ProviderOutboundRequestIdentityScope::Member, + ), + (Some(account), None) => ( + format!("account:v1:{account}"), + ProviderOutboundRequestIdentityScope::Account, + ), + (None, None) => ( + format!("key:v1:{}", fallback_key_id.trim()), + ProviderOutboundRequestIdentityScope::Key, + ), } } @@ -1201,5 +1225,131 @@ mod tests { )); assert_eq!(headers, original_headers); assert_eq!(body, original_body); + + let context = ProviderOutboundRequestContext::new("logical-turn", 1_700_000_000_123); + let result = apply_codex_fingerprint_convergence_policy( + &transport, + "openai:responses", + &context, + &mut headers, + &mut body, + ); + assert_eq!( + result.reason, + ProviderOutboundRequestPolicyReason::AgentIdentityExcluded + ); + assert_eq!(headers, original_headers); + assert_eq!(body, original_body); + } + + #[test] + fn policy_result_reports_applied_scopes_without_identity_values() { + let transport = sample_transport(); + let context = ProviderOutboundRequestContext::new("logical-turn", 1_700_000_000_123); + let mut headers = BTreeMap::new(); + let mut body = json!({"model": "gpt-5.4"}); + + let result = apply_codex_fingerprint_convergence_policy( + &transport, + "openai:responses", + &context, + &mut headers, + &mut body, + ); + + assert_eq!( + result, + ProviderOutboundRequestPolicyResult { + policy: ProviderOutboundRequestPolicy::CodexFingerprintConvergence, + outcome: + crate::outbound_request_policy::ProviderOutboundRequestPolicyOutcome::Applied, + reason: ProviderOutboundRequestPolicyReason::Applied, + mutation_scope: Some(ProviderOutboundRequestMutationScope::HeadersAndBody), + identity_scope: Some(ProviderOutboundRequestIdentityScope::Account), + } + ); + let serialized = serde_json::to_value(result).expect("serialize policy result"); + let serialized = serialized.as_object().expect("policy result object"); + assert_eq!(serialized.len(), 5); + assert!(!serialized.contains_key("installation_id")); + assert!(!serialized.contains_key("session_id")); + assert!(!serialized.contains_key("turn_id")); + } + + #[test] + fn policy_result_distinguishes_codex_skip_reasons_without_mutation() { + let context = ProviderOutboundRequestContext::new("logical-turn", 1_700_000_000_123); + let original_headers = BTreeMap::from([("x-custom".to_string(), "preserve".to_string())]); + let original_body = json!({"model": "gpt-5.4"}); + + let cases = [ + ( + "provider_type_mismatch", + "openai", + Some(json!({"codex": {"fingerprint_convergence_enabled": true}})), + "openai:responses", + original_body.clone(), + ProviderOutboundRequestPolicyReason::ProviderTypeMismatch, + ), + ( + "unsupported_api_format", + "codex", + Some(json!({"codex": {"fingerprint_convergence_enabled": true}})), + "openai:chat", + original_body.clone(), + ProviderOutboundRequestPolicyReason::UnsupportedApiFormat, + ), + ( + "compact_operation", + "codex", + Some(json!({"codex": {"fingerprint_convergence_enabled": true}})), + "openai:responses", + json!({"model": "gpt-5.4", "input": [{"type": "compaction_trigger"}]}), + ProviderOutboundRequestPolicyReason::CompactOperationExcluded, + ), + ( + "disabled", + "codex", + None, + "openai:responses", + original_body.clone(), + ProviderOutboundRequestPolicyReason::Disabled, + ), + ( + "request_body_not_object", + "codex", + Some(json!({"codex": {"fingerprint_convergence_enabled": true}})), + "openai:responses", + json!(["not-an-object"]), + ProviderOutboundRequestPolicyReason::RequestBodyNotObject, + ), + ]; + + for (name, provider_type, config, api_format, body, expected_reason) in cases { + let mut transport = sample_transport(); + transport.provider.provider_type = provider_type.to_string(); + transport.provider.config = config; + let mut headers = original_headers.clone(); + let mut request_body = body.clone(); + + let result = apply_codex_fingerprint_convergence_policy( + &transport, + api_format, + &context, + &mut headers, + &mut request_body, + ); + + assert_eq!( + result.outcome, + crate::outbound_request_policy::ProviderOutboundRequestPolicyOutcome::Skipped, + "case={name}" + ); + assert_eq!(result.reason, expected_reason, "case={name}"); + assert_eq!(result.mutation_scope, None, "case={name}"); + assert_eq!(result.identity_scope, None, "case={name}"); + assert_eq!(headers, original_headers, "case={name}"); + assert_eq!(request_body, body, "case={name}"); + } } } diff --git a/crates/aether-provider/transport/src/lib.rs b/crates/aether-provider/transport/src/lib.rs index cbd5fb3dd..9fa3ab2ee 100644 --- a/crates/aether-provider/transport/src/lib.rs +++ b/crates/aether-provider/transport/src/lib.rs @@ -17,6 +17,7 @@ pub mod kiro; mod network; pub mod oauth_refresh; mod openai_image; +mod outbound_request_policy; pub mod policy; pub mod provider_types; mod request_body; @@ -120,6 +121,13 @@ pub use openai_image::{ openai_image_transport_unsupported_reason, resolve_openai_image_auth, ProviderOpenAiImageHeadersInput, }; +pub use outbound_request_policy::{ + apply_provider_outbound_request_policies, ProviderOutboundRequestContext, + ProviderOutboundRequestIdentityScope, ProviderOutboundRequestMutationScope, + ProviderOutboundRequestPolicy, ProviderOutboundRequestPolicyOutcome, + ProviderOutboundRequestPolicyReason, ProviderOutboundRequestPolicyResult, + PROVIDER_OUTBOUND_CONTEXT_MAX_VALUE_BYTES, +}; pub use policy::{ local_gemini_transport_unsupported_reason, local_gemini_transport_unsupported_reason_with_network, diff --git a/crates/aether-provider/transport/src/outbound_request_policy.rs b/crates/aether-provider/transport/src/outbound_request_policy.rs new file mode 100644 index 000000000..d6948286d --- /dev/null +++ b/crates/aether-provider/transport/src/outbound_request_policy.rs @@ -0,0 +1,396 @@ +use std::collections::BTreeMap; + +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use sha2::{Digest, Sha256}; + +use crate::snapshot::GatewayProviderTransportSnapshot; + +/// Stable request-level signals that provider-specific outbound policies may use. +/// +/// The context deliberately carries client inputs rather than provider-derived +/// identities. Policies remain responsible for deriving and applying their own +/// wire representation at the terminal transport boundary. +const CONTEXT_HASH_DOMAIN: &[u8] = b"aether-provider-outbound-context-v1"; +const CONTEXT_HASH_PREFIX: &str = "aether:provider-context:v1:"; +/// Maximum byte length of any value retained in a cross-stage provider context. +/// +/// The limit is enforced before a context can reach Live persistence. Values +/// above the limit are represented by a deterministic, field-scoped digest so +/// retries keep the same policy identity without allowing unbounded client +/// input into the registry. +pub const PROVIDER_OUTBOUND_CONTEXT_MAX_VALUE_BYTES: usize = 256; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ProviderOutboundRequestContext { + logical_turn_id: String, + original_turn_id: Option, + original_client_session_id: Option, + original_prompt_cache_key: Option, + turn_started_at_unix_ms: u64, +} + +impl ProviderOutboundRequestContext { + pub fn new(logical_turn_id: impl Into, turn_started_at_unix_ms: u64) -> Self { + Self { + logical_turn_id: canonical_required_value(logical_turn_id.into(), "logical_turn_id"), + original_turn_id: None, + original_client_session_id: None, + original_prompt_cache_key: None, + turn_started_at_unix_ms, + } + } + + pub fn with_original_turn_id(mut self, original_turn_id: impl Into) -> Self { + self.original_turn_id = canonical_optional_value(original_turn_id.into(), "turn_id"); + self + } + + pub fn with_original_client_session_id( + mut self, + original_client_session_id: impl Into, + ) -> Self { + self.original_client_session_id = + canonical_optional_value(original_client_session_id.into(), "client_session_id"); + self + } + + pub fn with_original_prompt_cache_key( + mut self, + original_prompt_cache_key: impl Into, + ) -> Self { + self.original_prompt_cache_key = + canonical_optional_value(original_prompt_cache_key.into(), "prompt_cache_key"); + self + } + + pub fn logical_turn_id(&self) -> &str { + self.logical_turn_id.as_str() + } + + pub fn original_turn_id(&self) -> Option<&str> { + self.original_turn_id.as_deref() + } + + pub fn original_client_session_id(&self) -> Option<&str> { + self.original_client_session_id.as_deref() + } + + pub fn original_prompt_cache_key(&self) -> Option<&str> { + self.original_prompt_cache_key.as_deref() + } + + pub fn turn_started_at_unix_ms(&self) -> u64 { + self.turn_started_at_unix_ms + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ProviderOutboundRequestPolicy { + CodexFingerprintConvergence, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ProviderOutboundRequestPolicyOutcome { + Applied, + Skipped, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ProviderOutboundRequestPolicyReason { + Applied, + ProviderTypeMismatch, + AgentIdentityExcluded, + UnsupportedApiFormat, + CompactOperationExcluded, + Disabled, + RequestBodyNotObject, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ProviderOutboundRequestMutationScope { + Headers, + Body, + HeadersAndBody, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ProviderOutboundRequestIdentityScope { + PersistedFingerprint, + AccountMember, + Member, + Account, + Key, +} + +/// Low-sensitivity report for one selected provider-specific policy. +/// +/// This type intentionally contains only categorical values. Derived identity +/// values, client identifiers, and cache keys must never be added to it. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct ProviderOutboundRequestPolicyResult { + pub policy: ProviderOutboundRequestPolicy, + pub outcome: ProviderOutboundRequestPolicyOutcome, + pub reason: ProviderOutboundRequestPolicyReason, + #[serde(skip_serializing_if = "Option::is_none")] + pub mutation_scope: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub identity_scope: Option, +} + +impl ProviderOutboundRequestPolicyResult { + pub(crate) fn applied( + policy: ProviderOutboundRequestPolicy, + mutation_scope: ProviderOutboundRequestMutationScope, + identity_scope: ProviderOutboundRequestIdentityScope, + ) -> Self { + Self { + policy, + outcome: ProviderOutboundRequestPolicyOutcome::Applied, + reason: ProviderOutboundRequestPolicyReason::Applied, + mutation_scope: Some(mutation_scope), + identity_scope: Some(identity_scope), + } + } + + pub(crate) fn skipped( + policy: ProviderOutboundRequestPolicy, + reason: ProviderOutboundRequestPolicyReason, + ) -> Self { + Self { + policy, + outcome: ProviderOutboundRequestPolicyOutcome::Skipped, + reason, + mutation_scope: None, + identity_scope: None, + } + } + + pub fn was_applied(&self) -> bool { + self.outcome == ProviderOutboundRequestPolicyOutcome::Applied + } +} + +/// Applies the statically registered outbound policies for the final provider. +/// +/// Provider selection has already completed at this boundary. A provider with +/// no registered adapter is a strict no-op and produces no policy result. +pub fn apply_provider_outbound_request_policies( + transport: &GatewayProviderTransportSnapshot, + provider_api_format: &str, + context: &ProviderOutboundRequestContext, + provider_request_headers: &mut BTreeMap, + provider_request_body: &mut Value, +) -> Vec { + if !transport + .provider + .provider_type + .trim() + .eq_ignore_ascii_case("codex") + { + return Vec::new(); + } + + vec![ + crate::codex_fingerprint::apply_codex_fingerprint_convergence_policy( + transport, + provider_api_format, + context, + provider_request_headers, + provider_request_body, + ), + ] +} + +fn canonical_required_value(value: String, field: &str) -> String { + let value = value.trim(); + // Bound the JSON-encoded representation, not just the source bytes. This + // preserves ordinary Unicode while preventing quotes, backslashes, or + // control characters from escaping beyond the aggregate Live record + // budget. + if value.len() <= PROVIDER_OUTBOUND_CONTEXT_MAX_VALUE_BYTES + && serde_json::to_string(value).is_ok_and(|encoded| { + encoded.len() <= PROVIDER_OUTBOUND_CONTEXT_MAX_VALUE_BYTES.saturating_add(2) + }) + { + return value.to_string(); + } + digest_context_value(field, value) +} + +fn canonical_optional_value(value: String, field: &str) -> Option { + let value = value.trim(); + (!value.is_empty()).then(|| canonical_required_value(value.to_string(), field)) +} + +fn digest_context_value(field: &str, value: &str) -> String { + let mut digest = Sha256::new(); + digest.update(CONTEXT_HASH_DOMAIN); + digest.update([0]); + digest.update(field.as_bytes()); + digest.update([0]); + digest.update((value.len() as u64).to_be_bytes()); + digest.update(value.as_bytes()); + format!("{CONTEXT_HASH_PREFIX}{field}:{:x}", digest.finalize()) +} + +#[cfg(test)] +mod tests { + use serde_json::json; + + use super::*; + use crate::snapshot::{ + GatewayProviderTransportEndpoint, GatewayProviderTransportKey, + GatewayProviderTransportProvider, + }; + + fn sample_transport(provider_type: &str) -> GatewayProviderTransportSnapshot { + GatewayProviderTransportSnapshot { + provider: GatewayProviderTransportProvider { + id: "provider-1".to_string(), + name: "Provider".to_string(), + provider_type: provider_type.to_string(), + website: None, + is_active: true, + keep_priority_on_conversion: false, + enable_format_conversion: true, + concurrent_limit: None, + max_retries: None, + proxy: None, + request_timeout_secs: None, + stream_first_byte_timeout_secs: None, + config: Some(json!({ + "codex": {"fingerprint_convergence_enabled": true} + })), + }, + endpoint: GatewayProviderTransportEndpoint { + id: "endpoint-1".to_string(), + provider_id: "provider-1".to_string(), + api_format: "openai:responses".to_string(), + api_family: None, + endpoint_kind: None, + is_active: true, + base_url: "https://example.com".to_string(), + header_rules: None, + body_rules: None, + max_retries: None, + custom_path: None, + config: None, + format_acceptance_config: None, + proxy: None, + }, + key: GatewayProviderTransportKey { + id: "key-1".to_string(), + provider_id: "provider-1".to_string(), + name: "Key".to_string(), + auth_type: "api_key".to_string(), + is_active: true, + api_formats: None, + auth_type_by_format: None, + allow_auth_channel_mismatch_formats: None, + allowed_models: None, + capabilities: None, + rate_multipliers: None, + global_priority_by_format: None, + expires_at_unix_secs: None, + proxy: None, + fingerprint: None, + upstream_metadata: None, + decrypted_api_key: "secret".to_string(), + decrypted_auth_config: None, + }, + } + } + + #[test] + fn dispatcher_leaves_unregistered_provider_requests_unchanged() { + let transport = sample_transport("openai"); + let context = ProviderOutboundRequestContext::new("logical-turn", 1_700_000_000_123); + let original_headers = BTreeMap::from([("x-custom".to_string(), "preserve".to_string())]); + let original_body = json!({"model": "gpt-5.4", "custom": true}); + let mut headers = original_headers.clone(); + let mut body = original_body.clone(); + + let results = apply_provider_outbound_request_policies( + &transport, + "openai:responses", + &context, + &mut headers, + &mut body, + ); + + assert!(results.is_empty()); + assert_eq!(headers, original_headers); + assert_eq!(body, original_body); + } + + #[test] + fn result_serialization_is_categorical_and_snake_case() { + let result = ProviderOutboundRequestPolicyResult::applied( + ProviderOutboundRequestPolicy::CodexFingerprintConvergence, + ProviderOutboundRequestMutationScope::HeadersAndBody, + ProviderOutboundRequestIdentityScope::AccountMember, + ); + + assert_eq!( + serde_json::to_value(result).expect("serialize policy result"), + json!({ + "policy": "codex_fingerprint_convergence", + "outcome": "applied", + "reason": "applied", + "mutation_scope": "headers_and_body", + "identity_scope": "account_member" + }) + ); + } + + #[test] + fn context_values_are_bounded_and_deterministic() { + let oversized = "x".repeat(PROVIDER_OUTBOUND_CONTEXT_MAX_VALUE_BYTES + 1); + let context = ProviderOutboundRequestContext::new(oversized.clone(), 1) + .with_original_turn_id(oversized.clone()) + .with_original_client_session_id(oversized.clone()) + .with_original_prompt_cache_key(oversized); + let same_context = ProviderOutboundRequestContext::new( + "x".repeat(PROVIDER_OUTBOUND_CONTEXT_MAX_VALUE_BYTES + 1), + 1, + ) + .with_original_turn_id("x".repeat(PROVIDER_OUTBOUND_CONTEXT_MAX_VALUE_BYTES + 1)) + .with_original_client_session_id("x".repeat(PROVIDER_OUTBOUND_CONTEXT_MAX_VALUE_BYTES + 1)) + .with_original_prompt_cache_key("x".repeat(PROVIDER_OUTBOUND_CONTEXT_MAX_VALUE_BYTES + 1)); + + assert_eq!(context, same_context); + assert!(context.logical_turn_id().len() <= PROVIDER_OUTBOUND_CONTEXT_MAX_VALUE_BYTES); + assert!(context + .original_turn_id() + .is_some_and(|value| value.len() <= PROVIDER_OUTBOUND_CONTEXT_MAX_VALUE_BYTES)); + assert!(context + .original_client_session_id() + .is_some_and(|value| value.len() <= PROVIDER_OUTBOUND_CONTEXT_MAX_VALUE_BYTES)); + assert!(context + .original_prompt_cache_key() + .is_some_and(|value| value.len() <= PROVIDER_OUTBOUND_CONTEXT_MAX_VALUE_BYTES)); + assert!(context.logical_turn_id().starts_with(CONTEXT_HASH_PREFIX)); + assert_ne!( + context.logical_turn_id(), + ProviderOutboundRequestContext::new( + "y".repeat(PROVIDER_OUTBOUND_CONTEXT_MAX_VALUE_BYTES + 1), + 1, + ) + .logical_turn_id() + ); + + let unicode = ProviderOutboundRequestContext::new("turn-你好", 1); + assert_eq!(unicode.logical_turn_id(), "turn-你好"); + let escaped = ProviderOutboundRequestContext::new( + r#"""#.repeat(PROVIDER_OUTBOUND_CONTEXT_MAX_VALUE_BYTES), + 1, + ); + assert!(escaped.logical_turn_id().starts_with(CONTEXT_HASH_PREFIX)); + } +}