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

Merge pull request #775 from fawney19/codex/provider-policy-hardening

Browse Source 
feat(codex): add provider outbound policy boundary
This commit is contained in:
fawney19
2026-09-05 16:56:23 +08:00
committed by GitHub
parent 882bb43125 9d7a0665c0
commit d1cb0ebecf
12 changed files with 1313 additions and 198 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/aether-gateway/src/ai_serving/codex_context.rs
+14 -19
View File
@@ -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<OnceLock<CodexFingerprintConvergenceContext>>);
pub(crate) struct CodexFingerprintContextSlot(Arc<OnceLock<ProviderOutboundRequestContext>>);
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::<CodexFingerprintConvergenceContext>()
.get::<ProviderOutboundRequestContext>()
.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::<CodexFingerprintConvergenceContext>()
.get::<ProviderOutboundRequestContext>()
.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::<CodexFingerprintConvergenceContext>()
.get::<ProviderOutboundRequestContext>()
.is_none()
{
parts.extensions.remove::<CodexFingerprintContextSlot>();
@@ -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::<CodexFingerprintContextSlot>();
@@ -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::<CodexFingerprintContextSlot>();
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::<CodexFingerprintConvergenceContext>(),
parts.extensions.get::<ProviderOutboundRequestContext>(),
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");
apps/aether-gateway/src/ai_serving/planner/decision_input.rs
+89 -16
View File
@@ -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<ClientSurface>,
pub(crate) gateway_credential_carrier: Option<GatewayCredentialCarrier>,
pub(crate) client_session_affinity: Option<ClientSessionAffinity>,
pub(crate) codex_fingerprint_context: Option<CodexFingerprintConvergenceContext>,
pub(crate) provider_outbound_context: Option<ProviderOutboundRequestContext>,
pub(crate) routing_policy: Option<ResolvedRoutingPolicy>,
pub(crate) routing_trace_seed: Option<RoutingDecisionTrace>,
pub(crate) routing_context: Option<LocalRoutingRequestContext>,
@@ -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");
apps/aether-gateway/src/ai_serving/planner/standard/family/payload.rs
+1 -1
View File
@@ -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,
apps/aether-gateway/src/ai_serving/planner/standard/openai/chat/decision/request.rs
+1 -1
View File
@@ -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,
apps/aether-gateway/src/ai_serving/transport.rs
+8 -3
View File
@@ -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,
};
apps/aether-gateway/src/handlers/proxy/websocket/live/http.rs
+8
View File
@@ -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
{
apps/aether-gateway/src/handlers/proxy/websocket/live/planner.rs
+14 -10
View File
@@ -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<LiveCandidatePlanningOutcome, GatewayError> {
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<Option<PlannedLiveCandidate>, 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<LiveAuthMode> {
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,
apps/aether-gateway/src/handlers/proxy/websocket/live/registry.rs
+514 -41
View File
@@ -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<ProviderOutboundRequestContext>,
}
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<String>,
original_client_session_id: Option<String>,
original_prompt_cache_key: Option<String>,
turn_started_at_unix_ms: u64,
}
impl LiveProviderOutboundContextRecord {
fn from_context(
binding: &LiveCallBinding,
context: &ProviderOutboundRequestContext,
) -> Result<Self, LiveCallRegistryError> {
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<ProviderOutboundRequestContext, LiveCallRegistryError> {
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<Option<LiveCallBinding>, LiveCallRegistryError> {
) -> Result<Option<LiveCallRecord>, 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<LiveCallLookup, LiveCallRegistryError> {
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::<LiveCallBinding>(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::<LiveProviderOutboundContextRecord>(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::<LiveCallBinding>(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::<LiveCallBinding>(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::<LiveProviderOutboundContextRecord>(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<String>,
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::<LiveCallBinding>(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::<LiveProviderOutboundContextRecord>(
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<String, LiveCallRegistryError> {
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<String, LiveCallRegistryError> {
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::<LiveCallBinding>(&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
);
}
apps/aether-gateway/src/handlers/proxy/websocket/live/session.rs
+9 -6
View File
@@ -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<PreparedLiveSideband, LiveWebSocketPreflightRejection> {
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),
..
crates/aether-provider/transport/src/codex_fingerprint.rs
+251 -101
View File
@@ -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<String>,
original_client_session_id: Option<String>,
original_prompt_cache_key: Option<String>,
turn_started_at_unix_ms: u64,
}
impl CodexFingerprintConvergenceContext {
pub fn new(logical_turn_id: impl Into<String>, 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<String>) -> 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<String>,
) -> 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<String>,
) -> 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<String, String>,
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<String, String>,
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<String> {
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<String> {
.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}");
}
}
}
crates/aether-provider/transport/src/lib.rs
+8
View File
@@ -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,
crates/aether-provider/transport/src/outbound_request_policy.rs
+396
View File
@@ -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<String>,
original_client_session_id: Option<String>,
original_prompt_cache_key: Option<String>,
turn_started_at_unix_ms: u64,
}
impl ProviderOutboundRequestContext {
pub fn new(logical_turn_id: impl Into<String>, 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<String>) -> 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<String>,
) -> 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<String>,
) -> 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<ProviderOutboundRequestMutationScope>,
#[serde(skip_serializing_if = "Option::is_none")]
pub identity_scope: Option<ProviderOutboundRequestIdentityScope>,
}
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<String, String>,
provider_request_body: &mut Value,
) -> Vec<ProviderOutboundRequestPolicyResult> {
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<String> {
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));
}
}