use core::fmt;
use core::num::{NonZeroU32, NonZeroU64};
use core::str::FromStr;
use std::collections::BTreeMap;
use crate::ErrorClass;
use serde::Deserialize;
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ParseError {
what: &'static str,
input: String,
note: Option<&'static str>,
}
impl ParseError {
fn new(what: &'static str, input: &str) -> Self {
Self {
what,
input: input.to_owned(),
note: None,
}
}
fn with_note(what: &'static str, input: &str, note: &'static str) -> Self {
Self {
what,
input: input.to_owned(),
note: Some(note),
}
}
}
impl fmt::Display for ParseError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self.note {
None => write!(f, "unparseable {} literal: {:?}", self.what, self.input),
Some(note) => write!(
f,
"invalid {} literal: {:?} — {note}",
self.what, self.input
),
}
}
}
impl core::error::Error for ParseError {}
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Deserialize)]
#[serde(try_from = "String")]
pub struct DurationMs(pub u64);
impl FromStr for DurationMs {
type Err = ParseError;
fn from_str(s: &str) -> Result<Self, Self::Err> {
let err = || ParseError::new("duration", s);
let s = s.trim();
let unit_at = s.find(|c: char| !c.is_ascii_digit()).ok_or_else(err)?;
let (number, unit) = s.split_at(unit_at);
let n: u64 = number.parse().map_err(|_| err())?;
let mult = match unit {
"ms" => 1,
"s" => 1_000,
"m" => 60_000,
"h" => 3_600_000,
_ => return Err(err()),
};
Ok(Self(n * mult))
}
}
impl TryFrom<String> for DurationMs {
type Error = ParseError;
fn try_from(s: String) -> Result<Self, Self::Error> {
s.parse()
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)]
#[serde(try_from = "String")]
pub struct Rate {
pub limit: NonZeroU64,
pub window_ms: u64,
}
impl FromStr for Rate {
type Err = ParseError;
fn from_str(s: &str) -> Result<Self, Self::Err> {
let err = || ParseError::new("rate", s);
let (limit, window) = s.trim().split_once('/').ok_or_else(err)?;
let limit: NonZeroU64 = limit.trim().parse().map_err(|_| err())?;
let window_ms = match window.trim() {
"s" | "sec" => 1_000,
"min" => 60_000,
"h" | "hour" => 3_600_000,
_ => return Err(err()),
};
Ok(Self { limit, window_ms })
}
}
impl TryFrom<String> for Rate {
type Error = ParseError;
fn try_from(s: String) -> Result<Self, Self::Error> {
s.parse()
}
}
#[derive(Debug, Clone, PartialEq, Deserialize)]
#[serde(try_from = "String")]
pub struct Schedule {
pub segments: Vec<ScheduleSegment>,
}
#[derive(Debug, Clone, Copy, PartialEq)]
pub struct ScheduleSegment {
pub primary: SchedulePrimary,
pub up_to_ms: Option<u64>,
}
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum SchedulePrimary {
Exp {
base_ms: u64,
factor: f64,
cap_ms: u64,
jitter: bool,
},
Fixed {
period_ms: u64,
},
}
impl SchedulePrimary {
#[expect(
clippy::cast_precision_loss,
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::cast_possible_wrap,
reason = "backoff arithmetic: values are small and non-negative by construction"
)]
fn wait_ms(self, attempt: u32) -> u64 {
match self {
Self::Exp {
base_ms,
factor,
cap_ms,
..
} => {
let wait = base_ms as f64 * factor.powi(attempt as i32 - 1);
wait.min(cap_ms as f64).round() as u64
}
Self::Fixed { period_ms } => period_ms,
}
}
fn jitter(self) -> bool {
matches!(self, Self::Exp { jitter: true, .. })
}
}
impl Default for Schedule {
fn default() -> Self {
Self {
segments: vec![ScheduleSegment {
primary: SchedulePrimary::Exp {
base_ms: 200,
factor: 2.0,
cap_ms: 30_000,
jitter: true,
},
up_to_ms: None,
}],
}
}
}
impl Schedule {
#[must_use]
pub fn wait_ms(&self, attempt: u32) -> u64 {
self.wait_and_jitter(attempt).0
}
#[must_use]
pub fn wait_and_jitter(&self, attempt: u32) -> (u64, bool) {
let attempt = attempt.max(1);
let last = self.segments.len() - 1;
let (mut i, mut a, mut e) = (0_usize, 1_u32, 0_u64);
let mut emitted = 0_u32;
loop {
let segment = self.segments[i];
let wait = segment.primary.wait_ms(a);
if i < last
&& let Some(bound) = segment.up_to_ms
&& e.saturating_add(wait) > bound
{
(i, a, e) = (i + 1, 1, 0);
continue;
}
emitted += 1;
if emitted == attempt {
return (wait, segment.primary.jitter());
}
a += 1;
e = e.saturating_add(wait);
}
}
}
impl FromStr for SchedulePrimary {
type Err = ParseError;
fn from_str(s: &str) -> Result<Self, Self::Err> {
let err = || ParseError::new("schedule", s);
let s = s.trim();
if let Some(inner) = s.strip_prefix("exp(").and_then(|r| r.strip_suffix(')')) {
let parts: Vec<&str> = inner.split(',').map(str::trim).collect();
let [base, factor, rest @ ..] = parts.as_slice() else {
return Err(err());
};
let base_ms = base.parse::<DurationMs>().map_err(|_| err())?.0;
let factor: f64 = factor
.strip_prefix('x')
.ok_or_else(err)?
.parse()
.map_err(|_| err())?;
let mut cap_ms = u64::MAX;
let mut jitter = false;
for part in rest {
if let Some(d) = part.strip_prefix("max ") {
cap_ms = d.parse::<DurationMs>().map_err(|_| err())?.0;
} else if *part == "jitter" {
jitter = true;
} else {
return Err(err());
}
}
Ok(Self::Exp {
base_ms,
factor,
cap_ms,
jitter,
})
} else if let Some(inner) = s.strip_prefix("fixed(").and_then(|r| r.strip_suffix(')')) {
let period_ms = inner.parse::<DurationMs>().map_err(|_| err())?.0;
Ok(Self::Fixed { period_ms })
} else {
Err(err())
}
}
}
impl FromStr for Schedule {
type Err = ParseError;
fn from_str(s: &str) -> Result<Self, Self::Err> {
let err = || ParseError::new("schedule", s);
let tokens: Vec<&str> = s.split_whitespace().collect();
if tokens.is_empty() {
return Err(err());
}
let mut segments = Vec::new();
for segment_tokens in tokens.split(|t| *t == "andThen") {
let (primary_tokens, up_to_ms) = match segment_tokens.iter().position(|t| *t == "upTo")
{
None => (segment_tokens, None),
Some(pos) => {
let [duration] = &segment_tokens[pos + 1..] else {
return Err(err());
};
let bound = duration.parse::<DurationMs>().map_err(|_| err())?.0;
(&segment_tokens[..pos], Some(bound))
}
};
if primary_tokens.is_empty() {
return Err(err());
}
let primary: SchedulePrimary = primary_tokens.join(" ").parse().map_err(|_| err())?;
segments.push(ScheduleSegment { primary, up_to_ms });
}
let last = segments.len() - 1;
if segments
.iter()
.enumerate()
.any(|(i, segment)| (i < last) != segment.up_to_ms.is_some())
{
return Err(ParseError::with_note(
"schedule",
s,
"`upTo` must bound every segment except the last, and never the last \
(an unbounded segment never hands off; a bounded tail would leave \
attempts without a wait — cap total retrying with `attempts`)",
));
}
Ok(Self { segments })
}
}
impl TryFrom<String> for Schedule {
type Error = ParseError;
fn try_from(s: String) -> Result<Self, Self::Error> {
s.parse()
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)]
#[serde(try_from = "String")]
pub enum Condition {
Conn,
Timeout,
Cancelled,
Other,
Class4xx,
Class5xx,
Status(u16),
}
impl Condition {
pub fn matches(self, class: ErrorClass, http_status: Option<u16>) -> bool {
match self {
Self::Conn => class == ErrorClass::Conn,
Self::Timeout => class == ErrorClass::Timeout,
Self::Cancelled => class == ErrorClass::Cancelled,
Self::Other => class == ErrorClass::Other,
Self::Class4xx => {
class == ErrorClass::Http && http_status.is_some_and(|s| (400..=499).contains(&s))
}
Self::Class5xx => {
class == ErrorClass::Http && http_status.is_some_and(|s| (500..=599).contains(&s))
}
Self::Status(want) => class == ErrorClass::Http && http_status == Some(want),
}
}
}
impl FromStr for Condition {
type Err = ParseError;
fn from_str(s: &str) -> Result<Self, Self::Err> {
match s {
"conn" => Ok(Self::Conn),
"timeout" => Ok(Self::Timeout),
"cancelled" => Ok(Self::Cancelled),
"other" => Ok(Self::Other),
"4xx" => Ok(Self::Class4xx),
"5xx" => Ok(Self::Class5xx),
exact if exact.len() == 3 && exact.bytes().all(|b| b.is_ascii_digit()) => {
let code: u16 = exact
.parse()
.map_err(|_| ParseError::new("retry condition", s))?;
if (100..=599).contains(&code) {
Ok(Self::Status(code))
} else {
Err(ParseError::new("retry condition", s))
}
}
_ => Err(ParseError::new("retry condition", s)),
}
}
}
impl TryFrom<String> for Condition {
type Error = ParseError;
fn try_from(s: String) -> Result<Self, Self::Error> {
s.parse()
}
}
#[derive(Debug, Clone, PartialEq, Deserialize)]
#[serde(default, deny_unknown_fields)]
pub struct RetryPolicy {
pub attempts: NonZeroU32,
pub schedule: Schedule,
pub on: Vec<Condition>,
}
impl RetryPolicy {
pub const DEFAULT_ATTEMPTS: NonZeroU32 = NonZeroU32::new(3).unwrap();
pub fn default_on() -> Vec<Condition> {
vec![
Condition::Conn,
Condition::Timeout,
Condition::Status(429),
Condition::Class5xx,
]
}
pub fn is_retryable(&self, class: ErrorClass, http_status: Option<u16>) -> bool {
self.on.iter().any(|c| c.matches(class, http_status))
}
}
impl Default for RetryPolicy {
fn default() -> Self {
Self {
attempts: Self::DEFAULT_ATTEMPTS,
schedule: Schedule::default(),
on: Self::default_on(),
}
}
}
#[derive(Debug, Clone, PartialEq, Deserialize)]
#[serde(try_from = "BreakerPolicyDe")]
pub struct BreakerPolicy {
pub failures: Option<NonZeroU64>,
pub cooldown: DurationMs,
pub window: Option<DurationMs>,
pub failure_rate: Option<f64>,
pub min_calls: Option<NonZeroU32>,
}
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum BreakerMode {
Count { failures: NonZeroU64 },
Rate {
window: DurationMs,
failure_rate: f64,
min_calls: NonZeroU32,
},
}
impl BreakerPolicy {
pub const DEFAULT_FAILURES: NonZeroU64 = NonZeroU64::new(5).unwrap();
pub const DEFAULT_MIN_CALLS: NonZeroU32 = NonZeroU32::new(10).unwrap();
#[must_use]
pub fn mode(&self) -> BreakerMode {
match (self.failures, self.window, self.failure_rate) {
(None, Some(window), Some(failure_rate)) => BreakerMode::Rate {
window,
failure_rate,
min_calls: self.min_calls.unwrap_or(Self::DEFAULT_MIN_CALLS),
},
(failures, _, _) => BreakerMode::Count {
failures: failures.unwrap_or(Self::DEFAULT_FAILURES),
},
}
}
#[must_use]
pub fn has_inert_rate_knobs(&self) -> bool {
self.failures.is_some()
&& (self.window.is_some() || self.failure_rate.is_some() || self.min_calls.is_some())
}
}
#[derive(Deserialize)]
#[serde(default, deny_unknown_fields)]
struct BreakerPolicyDe {
failures: Option<NonZeroU64>,
cooldown: DurationMs,
window: Option<DurationMs>,
#[serde(deserialize_with = "de_failure_rate")]
failure_rate: Option<f64>,
min_calls: Option<NonZeroU32>,
}
impl Default for BreakerPolicyDe {
fn default() -> Self {
Self {
failures: None,
cooldown: DurationMs(15_000),
window: None,
failure_rate: None,
min_calls: None,
}
}
}
impl TryFrom<BreakerPolicyDe> for BreakerPolicy {
type Error = String;
fn try_from(de: BreakerPolicyDe) -> Result<Self, Self::Error> {
let rate_pair = de.window.is_some() && de.failure_rate.is_some();
let any_rate_knob =
de.window.is_some() || de.failure_rate.is_some() || de.min_calls.is_some();
if de.failures.is_none() && any_rate_knob && !rate_pair {
return Err(String::from(
"breaker rate mode requires both `window` and `failure_rate` \
(count mode sets `failures` instead)",
));
}
Ok(Self {
failures: de.failures,
cooldown: de.cooldown,
window: de.window,
failure_rate: de.failure_rate,
min_calls: de.min_calls,
})
}
}
fn de_failure_rate<'de, D>(deserializer: D) -> Result<Option<f64>, D::Error>
where
D: serde::Deserializer<'de>,
{
let value = Option::<f64>::deserialize(deserializer)?;
if let Some(rate) = value
&& !(rate > 0.0 && rate <= 1.0)
{
return Err(serde::de::Error::custom(format!(
"breaker.failure_rate must be greater than 0 and at most 1 (got {rate})"
)));
}
Ok(value)
}
impl Default for BreakerPolicy {
fn default() -> Self {
Self {
failures: None,
cooldown: DurationMs(15_000),
window: None,
failure_rate: None,
min_calls: None,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum CacheScope {
#[default]
Memory,
Persistent,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum CacheMode {
#[default]
Always,
Dev,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum CacheKeySource {
#[default]
Args,
Url,
}
#[derive(Debug, Clone, PartialEq, Default, Deserialize)]
#[serde(default, deny_unknown_fields)]
pub struct CachePolicy {
pub ttl: Option<DurationMs>,
pub scope: CacheScope,
pub mode: CacheMode,
pub key: CacheKeySource,
}
#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct IdempotencyPolicy {
pub header: String,
}
#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
#[serde(try_from = "PollUntilRaw")]
pub struct PollUntil {
pub field: String,
pub terminal: Vec<String>,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct PollUntilRaw {
field: String,
terminal: Vec<String>,
}
impl TryFrom<PollUntilRaw> for PollUntil {
type Error = ParseError;
fn try_from(raw: PollUntilRaw) -> Result<Self, Self::Error> {
if raw.field.is_empty() {
return Err(ParseError::new("poll until.field", "(empty)"));
}
if raw.terminal.is_empty() {
return Err(ParseError::new("poll until.terminal", "(empty array)"));
}
Ok(Self {
field: raw.field,
terminal: raw.terminal,
})
}
}
#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
#[serde(try_from = "PollPolicyRaw")]
pub struct PollPolicy {
pub interval: DurationMs,
pub deadline: DurationMs,
pub until: PollUntil,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct PollPolicyRaw {
interval: DurationMs,
deadline: DurationMs,
until: PollUntil,
}
impl TryFrom<PollPolicyRaw> for PollPolicy {
type Error = ParseError;
fn try_from(raw: PollPolicyRaw) -> Result<Self, Self::Error> {
if raw.interval.0 == 0 {
return Err(ParseError::with_note(
"poll.interval",
"0",
"must be a nonzero duration",
));
}
Ok(Self {
interval: raw.interval,
deadline: raw.deadline,
until: raw.until,
})
}
}
#[derive(Debug, Clone, PartialEq, Default, Deserialize)]
#[serde(default, deny_unknown_fields)]
pub struct TargetPolicy {
pub timeout: Option<DurationMs>,
pub retry: Option<RetryPolicy>,
pub breaker: Option<BreakerPolicy>,
pub rate: Option<Rate>,
pub cache: Option<CachePolicy>,
pub idempotency: Option<IdempotencyPolicy>,
pub poll: Option<PollPolicy>,
pub fallback: Option<Vec<String>>,
pub budget: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Default, Deserialize)]
#[serde(default, deny_unknown_fields)]
pub struct Defaults {
pub outbound: Option<TargetPolicy>,
pub llm: Option<TargetPolicy>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum NondeterminismResponse {
#[default]
Fail,
Warn,
Branch,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum OnBusy {
#[default]
Skip,
Wait,
Fail,
}
#[derive(Debug, Clone, PartialEq, Eq, Default, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct FlowMatchRule {
pub argv: Vec<String>,
}
#[derive(Debug, Clone, PartialEq, Eq, Default, Deserialize)]
#[serde(default, deny_unknown_fields)]
pub struct FlowsPolicy {
pub entrypoints: Vec<String>,
pub on_nondeterminism: NondeterminismResponse,
pub on_busy: OnBusy,
#[serde(rename = "match")]
pub match_: Option<BTreeMap<String, FlowMatchRule>>,
}
#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
#[serde(try_from = "String")]
pub struct JournalLocation(pub String);
impl FromStr for JournalLocation {
type Err = ParseError;
fn from_str(s: &str) -> Result<Self, Self::Err> {
let valid = s.strip_prefix("file:").is_some_and(|rest| !rest.is_empty())
|| s.strip_prefix("postgres://")
.is_some_and(|rest| !rest.is_empty());
if valid {
Ok(Self(s.to_owned()))
} else {
Err(ParseError::new("journal location", s))
}
}
}
impl TryFrom<String> for JournalLocation {
type Error = ParseError;
fn try_from(s: String) -> Result<Self, Self::Error> {
s.parse()
}
}
#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
#[serde(default, deny_unknown_fields)]
pub struct TelemetryPolicy {
pub otlp_endpoint: Option<String>,
pub console: bool,
}
impl Default for TelemetryPolicy {
fn default() -> Self {
Self {
otlp_endpoint: None,
console: true,
}
}
}
#[derive(Debug, Clone, PartialEq, Default, Deserialize)]
#[serde(default, deny_unknown_fields)]
pub struct Policy {
pub defaults: Defaults,
pub target: BTreeMap<String, TargetPolicy>,
pub flows: Option<FlowsPolicy>,
pub journal: Option<JournalLocation>,
pub telemetry: Option<TelemetryPolicy>,
}
#[derive(Debug, Clone, Default)]
pub struct ResolvedPolicy {
pub timeout: Option<DurationMs>,
pub retry: Option<RetryPolicy>,
pub breaker: Option<BreakerPolicy>,
pub rate: Option<Rate>,
pub cache: Option<CachePolicy>,
pub idempotency: Option<IdempotencyPolicy>,
pub poll: Option<PollPolicy>,
}
impl Policy {
pub fn resolve(&self, target: &str) -> ResolvedPolicy {
ResolvedPolicy {
timeout: self.layer(target, |t| t.timeout.as_ref()).copied(),
retry: self.layer(target, |t| t.retry.as_ref()).cloned(),
breaker: self.layer(target, |t| t.breaker.as_ref()).cloned(),
rate: self.layer(target, |t| t.rate.as_ref()).copied(),
cache: self.layer(target, |t| t.cache.as_ref()).cloned(),
idempotency: self.layer(target, |t| t.idempotency.as_ref()).cloned(),
poll: self.layer(target, |t| t.poll.as_ref()).cloned(),
}
}
fn layer<'a, T>(
&'a self,
target: &str,
pick: impl Fn(&'a TargetPolicy) -> Option<&'a T>,
) -> Option<&'a T> {
if let Some(t) = self.target.get(target)
&& let Some(v) = pick(t)
{
return Some(v);
}
if target.starts_with("llm:")
&& let Some(llm) = self.defaults.llm.as_ref()
&& let Some(v) = pick(llm)
{
return Some(v);
}
self.defaults.outbound.as_ref().and_then(pick)
}
}
const OUTBOUND_METHODS: [&str; 7] = ["GET", "HEAD", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"];
const CLASS_PREFIXES: [&str; 6] = ["py:", "ts:", "rs:", "llm:", "tool:", "mcp:"];
const VERTEX_REGIONAL_SUFFIX: &str = "-aiplatform.googleapis.com";
const LLM_HOST_PROVIDERS: &[(&str, &str)] = &[
("api.openai.com", "openai"),
("api.anthropic.com", "anthropic"),
("generativelanguage.googleapis.com", "google-genai"),
("aiplatform.googleapis.com", "google-genai"),
];
fn scheme_port(scheme: Option<&str>) -> Option<u16> {
match scheme {
Some("http") => Some(80),
Some("https") => Some(443),
_ => None,
}
}
fn glob_match(pattern: &[u8], text: &[u8]) -> bool {
let (mut p, mut t) = (0usize, 0usize);
let (mut star, mut mark) = (None::<usize>, 0usize);
while t < text.len() {
if p < pattern.len() && pattern[p] == b'*' {
star = Some(p);
mark = t;
p += 1;
} else if p < pattern.len() && pattern[p] == text[t] {
p += 1;
t += 1;
} else if let Some(sp) = star {
p = sp + 1;
mark += 1;
t = mark;
} else {
return false;
}
}
while p < pattern.len() && pattern[p] == b'*' {
p += 1;
}
p == pattern.len()
}
#[allow(clippy::type_complexity)]
fn parse_outbound_key(key: &str) -> Option<(Option<String>, String, Option<u16>, Option<String>)> {
let mut method: Option<String> = None;
let mut rest = key;
for m in OUTBOUND_METHODS {
if let Some(stripped) = rest.strip_prefix(m).and_then(|s| s.strip_prefix(' ')) {
method = Some(m.to_owned());
rest = stripped;
break;
}
}
let mut path: Option<String> = None;
if let Some(slash) = rest.find('/') {
path = Some(rest[slash..].to_owned());
rest = &rest[..slash];
}
let (mut host, mut port) = (rest.to_owned(), None);
if let Some((head, tail)) = rest.rsplit_once(':')
&& !tail.is_empty()
&& tail.bytes().all(|b| b.is_ascii_digit())
&& let Ok(n) = tail.parse::<u16>()
{
head.clone_into(&mut host);
port = Some(n);
}
if host.is_empty() {
return None;
}
Some((method, host, port, path))
}
fn is_bare_host_key(key: &str) -> bool {
!key.contains('*')
&& parse_outbound_key(key)
.is_some_and(|(m, _, port, path)| m.is_none() && port.is_none() && path.is_none())
}
struct OutboundPattern {
key: String,
method: Option<String>,
host_glob: String, port: Option<u16>,
path_glob: Option<String>,
wildcards: usize,
literal: usize,
}
impl Policy {
#[must_use]
pub fn resolve_target(
&self,
method: &str,
host: &str,
scheme: Option<&str>,
port: Option<u16>,
path: Option<&str>,
) -> String {
let provider = LLM_HOST_PROVIDERS
.iter()
.find(|(h, _)| *h == host)
.map(|(_, p)| *p)
.or_else(|| {
host.ends_with(VERTEX_REGIONAL_SUFFIX)
.then_some("google-genai")
});
if let Some(p) = provider {
return format!("llm:{p}");
}
if !CLASS_PREFIXES.iter().any(|c| host.starts_with(c))
&& self.target.contains_key(host)
&& is_bare_host_key(host)
{
return host.to_owned();
}
let mut patterns: Vec<OutboundPattern> = Vec::new();
for key in self.target.keys() {
if CLASS_PREFIXES.iter().any(|c| key.starts_with(c)) || is_bare_host_key(key) {
continue;
}
let Some((m, h, pt, pa)) = parse_outbound_key(key) else {
continue;
};
let wildcards = key.matches('*').count();
patterns.push(OutboundPattern {
key: key.clone(),
method: m,
host_glob: h.to_lowercase(),
port: pt,
path_glob: pa,
wildcards,
literal: key.chars().count() - wildcards,
});
}
patterns.sort_by(|a, b| {
a.wildcards
.cmp(&b.wildcards)
.then(b.literal.cmp(&a.literal))
.then(a.method.is_none().cmp(&b.method.is_none()))
.then(a.key.cmp(&b.key))
});
let effective_port = port.or_else(|| scheme_port(scheme));
let host_l = host.to_lowercase();
let method_u = if method.is_empty() {
"GET".to_owned()
} else {
method.to_uppercase()
};
let path_n = match path {
Some(p) if !p.is_empty() => p,
_ => "/",
};
for p in &patterns {
if let Some(m) = &p.method
&& *m != method_u
{
continue;
}
if !glob_match(p.host_glob.as_bytes(), host_l.as_bytes()) {
continue;
}
if let Some(pt) = p.port
&& Some(pt) != effective_port
{
continue;
}
if let Some(pg) = &p.path_glob
&& !glob_match(pg.as_bytes(), path_n.as_bytes())
{
continue;
}
return p.key.clone();
}
host.to_owned()
}
#[must_use]
pub fn known_llm_hosts() -> Vec<(&'static str, &'static str)> {
LLM_HOST_PROVIDERS.to_vec()
}
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
#[test]
fn duration_literals() {
assert_eq!("200ms".parse(), Ok(DurationMs(200)));
assert_eq!("30s".parse(), Ok(DurationMs(30_000)));
assert_eq!("5m".parse(), Ok(DurationMs(300_000)));
assert_eq!("2h".parse(), Ok(DurationMs(7_200_000)));
assert!("30".parse::<DurationMs>().is_err());
assert!("30sec".parse::<DurationMs>().is_err());
assert!("-1s".parse::<DurationMs>().is_err());
}
#[test]
fn rate_literals() {
let rate: Rate = "90/s".parse().unwrap();
assert_eq!((rate.limit.get(), rate.window_ms), (90, 1_000));
let rate: Rate = "60/min".parse().unwrap();
assert_eq!((rate.limit.get(), rate.window_ms), (60, 60_000));
assert!("0/s".parse::<Rate>().is_err(), "zero limit unrepresentable");
assert!("10/day".parse::<Rate>().is_err());
}
#[test]
fn schedule_exp_waits_and_cap() {
let schedule: Schedule = "exp(1s, x2, max 4s)".parse().unwrap();
let waits: Vec<u64> = (1..=4).map(|n| schedule.wait_ms(n)).collect();
assert_eq!(waits, [1_000, 2_000, 4_000, 4_000]);
}
#[test]
fn schedule_fixed_and_rejections() {
assert_eq!(
"fixed(1s)".parse::<Schedule>(),
Ok(Schedule {
segments: vec![ScheduleSegment {
primary: SchedulePrimary::Fixed { period_ms: 1_000 },
up_to_ms: None,
}],
})
);
assert!("linear(1s)".parse::<Schedule>().is_err());
}
#[test]
fn schedule_composition_parses_the_spec_example() {
let schedule: Schedule = "exp(1s, x2, max 5m) upTo 10m andThen fixed(1m)"
.parse()
.unwrap();
assert_eq!(
schedule.segments,
vec![
ScheduleSegment {
primary: SchedulePrimary::Exp {
base_ms: 1_000,
factor: 2.0,
cap_ms: 300_000,
jitter: false,
},
up_to_ms: Some(600_000),
},
ScheduleSegment {
primary: SchedulePrimary::Fixed { period_ms: 60_000 },
up_to_ms: None,
},
]
);
assert_eq!(
"exp(1s, x2, max 5m) upTo 10m andThen fixed(1m)".parse::<Schedule>(),
Ok(schedule)
);
}
#[test]
fn schedule_composition_hands_off_when_the_bound_would_be_overshot() {
let schedule: Schedule = "exp(1s, x2) upTo 4s andThen fixed(500ms)".parse().unwrap();
let waits: Vec<u64> = (1..=5).map(|n| schedule.wait_ms(n)).collect();
assert_eq!(waits, [1_000, 2_000, 500, 500, 500]);
}
#[test]
fn schedule_composition_exact_fit_stays_and_cascade_skips() {
let schedule: Schedule =
"fixed(1s) upTo 3s andThen fixed(10s) upTo 5s andThen fixed(250ms)"
.parse()
.unwrap();
let waits: Vec<u64> = (1..=6).map(|n| schedule.wait_ms(n)).collect();
assert_eq!(waits, [1_000, 1_000, 1_000, 250, 250, 250]);
}
#[test]
fn schedule_composition_restarts_exp_and_tracks_jitter_per_segment() {
let schedule: Schedule = "fixed(1s) upTo 2s andThen exp(100ms, x3, jitter)"
.parse()
.unwrap();
let waits: Vec<u64> = (1..=5).map(|n| schedule.wait_ms(n)).collect();
assert_eq!(waits, [1_000, 1_000, 100, 300, 900]);
assert_eq!(schedule.wait_and_jitter(1), (1_000, false));
assert_eq!(schedule.wait_and_jitter(3), (100, true));
}
#[test]
fn schedule_composition_shape_rule_rejections() {
for degenerate in [
"fixed(1s) andThen fixed(2s)",
"exp(1s, x2, max 5m) upTo 10m",
"fixed(1s) upTo 3s andThen fixed(2s) andThen fixed(4s)",
"fixed(1s) upTo 3s andThen fixed(2s) upTo 5s",
] {
let error = degenerate.parse::<Schedule>().unwrap_err();
assert!(
error.to_string().contains("upTo"),
"{degenerate}: expected the shape-rule note, got {error}"
);
}
for broken in [
"fixed(1s) upTo",
"upTo 3s andThen fixed(1s)",
"fixed(1s) upTo 1s upTo 2s andThen fixed(1s)",
"fixed(1s) andThen",
"andThen fixed(1s)",
"fixed(1s) upTo 3s fixed(2s)",
] {
assert!(
broken.parse::<Schedule>().is_err(),
"{broken} must be rejected"
);
}
}
#[test]
fn condition_matching() {
let on = RetryPolicy::default_on();
let matches = |class, status| on.iter().any(|c| c.matches(class, status));
assert!(matches(ErrorClass::Conn, None));
assert!(matches(ErrorClass::Http, Some(429)));
assert!(matches(ErrorClass::Http, Some(503)));
assert!(!matches(ErrorClass::Http, Some(400)));
assert!(!matches(ErrorClass::Cancelled, None));
assert!("teapot".parse::<Condition>().is_err());
assert_eq!("429".parse::<Condition>(), Ok(Condition::Status(429)));
assert_eq!("100".parse::<Condition>(), Ok(Condition::Status(100)));
assert_eq!("599".parse::<Condition>(), Ok(Condition::Status(599)));
for bad in ["999", "099", "600", "000", "12", "1234", "1x9"] {
assert!(bad.parse::<Condition>().is_err(), "{bad} must be rejected");
}
}
#[test]
fn zero_attempts_is_unrepresentable() {
let doc = json!({ "target": { "x": { "retry": { "attempts": 0 } } } });
let err = serde_path_to_error::deserialize::<_, Policy>(&doc).unwrap_err();
assert_eq!(err.path().to_string(), "target.x.retry.attempts");
}
#[test]
fn breaker_failure_rate_range_is_enforced() {
let bad = |rate: serde_json::Value| {
let doc = json!({ "target": { "x": { "breaker": { "failure_rate": rate } } } });
serde_path_to_error::deserialize::<_, Policy>(&doc)
};
for rate in [json!(0.0), json!(-0.1), json!(1.5), json!(2.0)] {
let err = bad(rate.clone()).unwrap_err();
assert_eq!(
err.path().to_string(),
"target.x.breaker.failure_rate",
"out-of-range failure_rate {rate} must fail at its path"
);
}
for rate in [0.01_f64, 0.5, 1.0] {
let doc = json!({
"target": { "x": { "breaker": { "window": "30s", "failure_rate": rate } } }
});
let policy = serde_path_to_error::deserialize::<_, Policy>(&doc).unwrap();
let breaker = policy.target["x"].breaker.as_ref().unwrap();
assert_eq!(breaker.failure_rate, Some(rate));
}
}
#[test]
fn breaker_mode_selection_follows_the_schema() {
let breaker = |doc: serde_json::Value| -> BreakerPolicy {
let doc = json!({ "target": { "x": { "breaker": doc } } });
let policy: Policy = serde_path_to_error::deserialize(&doc).unwrap();
policy.target["x"].breaker.clone().unwrap()
};
assert_eq!(
breaker(json!({})).mode(),
BreakerMode::Count {
failures: BreakerPolicy::DEFAULT_FAILURES
}
);
assert_eq!(
breaker(json!({ "window": "30s", "failure_rate": 0.5 })).mode(),
BreakerMode::Rate {
window: DurationMs(30_000),
failure_rate: 0.5,
min_calls: BreakerPolicy::DEFAULT_MIN_CALLS,
}
);
assert_eq!(
breaker(json!({ "window": "10s", "failure_rate": 1.0, "min_calls": 4 })).mode(),
BreakerMode::Rate {
window: DurationMs(10_000),
failure_rate: 1.0,
min_calls: NonZeroU32::new(4).unwrap(),
}
);
let mixed = breaker(json!({ "failures": 3, "window": "30s", "failure_rate": 0.5 }));
assert_eq!(
mixed.mode(),
BreakerMode::Count {
failures: NonZeroU64::new(3).unwrap()
}
);
assert!(mixed.has_inert_rate_knobs());
assert!(!breaker(json!({ "failures": 3 })).has_inert_rate_knobs());
}
#[test]
fn half_configured_breaker_rate_mode_is_rejected() {
for doc in [
json!({ "window": "30s" }),
json!({ "failure_rate": 0.5 }),
json!({ "min_calls": 10 }),
json!({ "window": "30s", "min_calls": 10 }),
json!({ "failure_rate": 0.5, "min_calls": 10 }),
] {
let policy = json!({ "target": { "x": { "breaker": doc } } });
let err = serde_path_to_error::deserialize::<_, Policy>(&policy).unwrap_err();
assert_eq!(err.path().to_string(), "target.x.breaker", "doc: {doc}");
assert!(
err.inner().to_string().contains("rate mode requires both"),
"doc {doc}: got {}",
err.inner()
);
}
let policy =
json!({ "target": { "x": { "breaker": { "failures": 3, "window": "30s" } } } });
assert!(serde_path_to_error::deserialize::<_, Policy>(&policy).is_ok());
}
#[test]
fn unknown_key_is_rejected_with_its_path() {
let doc = json!({ "target": { "api.stripe.com": { "retry": { "atempts": 10 } } } });
let err = serde_path_to_error::deserialize::<_, Policy>(&doc).unwrap_err();
assert!(
err.inner().to_string().contains("atempts")
|| err.inner().to_string().contains("unknown field"),
"expected an unknown-field error, got {}",
err.inner()
);
}
#[test]
fn unknown_top_level_and_layer_keys_are_rejected() {
assert!(
serde_path_to_error::deserialize::<_, Policy>(&json!({ "bogus_top": true })).is_err()
);
assert!(
serde_path_to_error::deserialize::<_, Policy>(
&json!({ "target": { "api.x": { "retrys": {} } } })
)
.is_err(),
"a mistyped layer table must be rejected, not dropped"
);
}
#[test]
fn journal_and_telemetry_parse_and_validate() {
let doc = json!({
"journal": "file:/srv/keel/journal.db",
"telemetry": { "otlp_endpoint": "http://collector:4317" }
});
let policy: Policy = serde_path_to_error::deserialize(&doc).unwrap();
assert_eq!(
policy.journal.unwrap(),
JournalLocation("file:/srv/keel/journal.db".to_owned())
);
let telemetry = policy.telemetry.unwrap();
assert_eq!(
telemetry.otlp_endpoint.as_deref(),
Some("http://collector:4317")
);
assert!(telemetry.console, "schema default console = true");
let bad = json!({ "journal": "sqlite:/tmp/x.db" });
assert!(serde_path_to_error::deserialize::<_, Policy>(&bad).is_err());
}
#[test]
fn idempotency_resolves_like_any_other_layer() {
let doc = json!({
"defaults": {
"outbound": { "idempotency": { "header": "X-Idem" } },
"llm": { "idempotency": { "header": "X-Llm-Idem" } }
},
"target": {
"api.stripe.com": { "idempotency": { "header": "Idempotency-Key" } },
"api.plain.example": { "timeout": "1s" }
}
});
let policy: Policy = serde_path_to_error::deserialize(&doc).unwrap();
let stripe = policy.resolve("api.stripe.com");
assert_eq!(
stripe.idempotency.as_ref().map(|i| i.header.as_str()),
Some("Idempotency-Key")
);
let llm = policy.resolve("llm:openai");
assert_eq!(
llm.idempotency.as_ref().map(|i| i.header.as_str()),
Some("X-Llm-Idem")
);
let plain = policy.resolve("api.plain.example");
assert_eq!(
plain.idempotency.as_ref().map(|i| i.header.as_str()),
Some("X-Idem")
);
let empty: Policy = serde_path_to_error::deserialize(&json!({})).unwrap();
assert!(empty.resolve("api.stripe.com").idempotency.is_none());
}
#[test]
fn poll_policy_parses_and_resolves() {
let policy: Policy = serde_json::from_value(serde_json::json!({
"target": { "api.jobs.example": { "poll": {
"interval": "10s", "deadline": "90s",
"until": { "field": "status", "terminal": ["completed", "failed"] }
} } }
}))
.expect("valid poll policy");
let resolved = policy.resolve("api.jobs.example");
let poll = resolved.poll.expect("poll resolved");
assert_eq!(poll.interval.0, 10_000);
assert_eq!(poll.deadline.0, 90_000);
assert_eq!(poll.until.field, "status");
assert_eq!(poll.until.terminal, vec!["completed", "failed"]);
}
#[test]
fn poll_rejects_empty_terminal_and_empty_field() {
for bad in [
serde_json::json!({ "interval": "10s", "deadline": "90s",
"until": { "field": "status", "terminal": [] } }),
serde_json::json!({ "interval": "10s", "deadline": "90s",
"until": { "field": "", "terminal": ["done"] } }),
serde_json::json!({ "interval": "10s",
"until": { "field": "status", "terminal": ["done"] } }),
] {
let doc = serde_json::json!({ "target": { "x": { "poll": bad } } });
assert!(serde_json::from_value::<Policy>(doc).is_err());
}
}
#[test]
fn poll_rejects_zero_interval() {
let doc = serde_json::json!({ "target": { "x": { "poll": {
"interval": "0ms", "deadline": "90s",
"until": { "field": "status", "terminal": ["done"] }
} } } });
assert!(serde_json::from_value::<Policy>(doc).is_err());
let doc = serde_json::json!({ "target": { "x": { "poll": {
"interval": "1ms", "deadline": "90s",
"until": { "field": "status", "terminal": ["done"] }
} } } });
assert!(serde_json::from_value::<Policy>(doc).is_ok());
}
#[test]
fn layer_resolution_precedence() {
let doc = json!({
"defaults": {
"outbound": { "retry": { "attempts": 3 }, "rate": "9/s" },
"llm": { "retry": { "attempts": 6 } }
},
"target": { "llm:openai": { "cache": { "ttl": "10m" } } }
});
let policy: Policy = serde_path_to_error::deserialize(&doc).unwrap();
let llm = policy.resolve("llm:openai");
assert_eq!(llm.cache.unwrap().ttl, Some(DurationMs(600_000)));
assert_eq!(llm.retry.unwrap().attempts.get(), 6);
assert_eq!(llm.rate.unwrap().limit.get(), 9);
let plain = policy.resolve("api.example.com");
assert_eq!(plain.retry.unwrap().attempts.get(), 3);
assert!(plain.cache.is_none());
}
#[test]
fn flows_on_busy_parses_with_skip_default() {
let p: Policy = serde_json::from_value(serde_json::json!({
"flows": { "entrypoints": ["cmd:autonomous-run"], "on_busy": "wait" }
}))
.unwrap();
assert_eq!(p.flows.as_ref().unwrap().on_busy, OnBusy::Wait);
let p: Policy = serde_json::from_value(serde_json::json!({ "flows": {} })).unwrap();
assert_eq!(p.flows.unwrap().on_busy, OnBusy::Skip);
}
#[test]
fn flows_match_cmd_rules_parse() {
let p: Policy = serde_json::from_value(serde_json::json!({
"flows": {
"entrypoints": ["cmd:nightly-etl"],
"match": { "cmd:nightly-etl": { "argv": ["*/run_etl.sh", "--env=prod"] } }
}
}))
.unwrap();
let flows = p.flows.unwrap();
let rules = flows.match_.expect("match table carried");
assert_eq!(
rules["cmd:nightly-etl"].argv,
vec!["*/run_etl.sh".to_owned(), "--env=prod".to_owned()]
);
let p: Policy = serde_json::from_value(serde_json::json!({ "flows": {} })).unwrap();
assert!(p.flows.unwrap().match_.is_none());
assert!(
serde_json::from_value::<Policy>(serde_json::json!({
"flows": { "match": { "cmd:x": { "argv": ["a"], "bogus": 1 } } }
}))
.is_err()
);
}
}
#[cfg(test)]
mod resolve_target_tests {
use super::*;
fn policy(keys: &[&str]) -> Policy {
let mut p = Policy::default();
for k in keys {
p.target.insert((*k).to_owned(), TargetPolicy::default());
}
p
}
#[test]
fn no_table_returns_bare_host() {
assert_eq!(
Policy::default().resolve_target("GET", "api.example.com", None, None, None),
"api.example.com"
);
}
#[test]
fn exact_beats_pattern() {
let p = policy(&["api.example.com", "*.example.com"]);
assert_eq!(
p.resolve_target("GET", "api.example.com", None, None, None),
"api.example.com"
);
}
#[test]
fn host_wildcard_crosses_dots() {
let p = policy(&["*.internal.corp"]);
assert_eq!(
p.resolve_target("GET", "a.b.internal.corp", None, None, None),
"*.internal.corp"
);
assert_eq!(
p.resolve_target("GET", "internal.corp", None, None, None),
"internal.corp"
);
}
#[test]
fn host_is_case_insensitive() {
let p = policy(&["*.Internal.Corp"]);
assert_eq!(
p.resolve_target("GET", "DB.INTERNAL.CORP", None, None, None),
"*.Internal.Corp"
);
}
#[test]
fn path_glob_crosses_slashes_case_sensitive() {
let p = policy(&["api.catalog.internal/*"]);
assert_eq!(
p.resolve_target("GET", "api.catalog.internal", None, None, Some("/a/b/c")),
"api.catalog.internal/*"
);
let p2 = policy(&["api.x/A/*"]);
assert_eq!(
p2.resolve_target("GET", "api.x", None, None, Some("/a/y")),
"api.x"
);
}
#[test]
fn missing_path_normalizes_to_slash() {
let p = policy(&["api.x/*"]);
assert_eq!(
p.resolve_target("GET", "api.x", None, None, None),
"api.x/*"
);
assert_eq!(
p.resolve_target("GET", "api.x", None, None, Some("")),
"api.x/*"
);
}
#[test]
fn method_prefix_must_match() {
let p = policy(&["POST api.example.com"]);
assert_eq!(
p.resolve_target("GET", "api.example.com", None, None, None),
"api.example.com"
);
assert_eq!(
p.resolve_target("POST", "api.example.com", None, None, None),
"POST api.example.com"
);
assert_eq!(
p.resolve_target("post", "api.example.com", None, None, None),
"POST api.example.com"
);
}
#[test]
fn port_uses_scheme_default() {
let p = policy(&["api.example.com:443"]);
assert_eq!(
p.resolve_target("GET", "api.example.com", Some("https"), None, None),
"api.example.com:443"
);
assert_eq!(
p.resolve_target("GET", "api.example.com", Some("http"), None, None),
"api.example.com"
);
}
#[test]
fn explicit_port_overrides_scheme() {
let p = policy(&["api.example.com:8443"]);
assert_eq!(
p.resolve_target("GET", "api.example.com", Some("https"), Some(8443), None),
"api.example.com:8443"
);
assert_eq!(
p.resolve_target("GET", "api.example.com", Some("https"), Some(443), None),
"api.example.com"
);
}
#[test]
fn most_specific_by_literal_length() {
let p = policy(&["*.example.com", "GET api.example.com/*"]);
assert_eq!(
p.resolve_target("GET", "api.example.com", None, None, Some("/v1/x")),
"GET api.example.com/*"
);
}
#[test]
fn lexicographic_tie_break_is_total() {
let p = policy(&["api.example.com/x/*", "api.example.com/*/y"]);
assert_eq!(
p.resolve_target("GET", "api.example.com", None, None, Some("/x/y")),
"api.example.com/*/y"
);
}
#[test]
fn class_prefixed_keys_are_not_hosts() {
let p = policy(&["py:pkg.mod.fn", "llm:openai"]);
assert_eq!(
p.resolve_target("GET", "py:pkg.mod.fn", None, None, None),
"py:pkg.mod.fn"
);
}
#[test]
fn llm_host_map_wins_over_patterns() {
let p = policy(&["*.openai.com"]);
assert_eq!(
p.resolve_target("POST", "api.openai.com", None, None, None),
"llm:openai"
);
}
#[test]
fn vertex_regional_suffix_maps_to_google_genai() {
assert_eq!(
Policy::default().resolve_target(
"POST",
"us-central1-aiplatform.googleapis.com",
None,
None,
None
),
"llm:google-genai"
);
}
#[test]
fn known_llm_hosts_matches_resolve_target_for_every_pair() {
let hosts = Policy::known_llm_hosts();
assert!(hosts.contains(&("api.openai.com", "openai")));
assert!(hosts.contains(&("api.anthropic.com", "anthropic")));
assert!(hosts.contains(&("generativelanguage.googleapis.com", "google-genai")));
for (host, provider) in hosts {
assert_eq!(
Policy::default().resolve_target("GET", host, None, None, None),
format!("llm:{provider}")
);
}
}
}