use crate::net::http::{Stream, Url};
use crate::wire::intel::{Request, Response};
use std::cell::RefCell;
use std::fmt;
use std::time::Duration;
use super::endpoints::EndpointList;
use super::{anthropic, bedrock, failover, openai};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Provider {
OpenAiCompatible,
Anthropic,
Bedrock,
}
impl Provider {
pub(super) fn default_path(self) -> &'static str {
match self {
Provider::OpenAiCompatible => openai::DEFAULT_PATH,
Provider::Anthropic => anthropic::DEFAULT_PATH,
Provider::Bedrock => bedrock::DEFAULT_PATH,
}
}
pub fn from_dialect(dialect: Option<&str>) -> Option<Provider> {
match dialect.map(str::trim).filter(|s| !s.is_empty()) {
None | Some("openai") | Some("openai-compatible") => Some(Provider::OpenAiCompatible),
Some("anthropic") => Some(Provider::Anthropic),
Some("bedrock") => Some(Provider::Bedrock),
Some(_) => None,
}
}
pub(super) fn request_path(self, configured: &str, req: &Request) -> String {
match self {
Provider::Bedrock => bedrock::converse_path(&req.model),
_ => configured.to_string(),
}
}
}
#[derive(Debug)]
pub enum IntelError {
Transport(std::io::Error),
Http(u16, String),
Parse(String),
Unsupported(String),
AllEndpointsDown(Option<Box<IntelError>>),
}
impl fmt::Display for IntelError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
IntelError::Transport(e) => write!(f, "intelligence transport error: {e}"),
IntelError::Http(code, body) => write!(f, "intelligence HTTP {code}: {body}"),
IntelError::Parse(m) => write!(f, "{m}"),
IntelError::Unsupported(m) => write!(f, "{m}"),
IntelError::AllEndpointsDown(cause) => match cause {
Some(e) => write!(f, "all intelligence endpoints down (last error: {e})"),
None => write!(f, "all intelligence endpoints down"),
},
}
}
}
impl std::error::Error for IntelError {}
impl From<std::io::Error> for IntelError {
fn from(e: std::io::Error) -> Self {
IntelError::Transport(e)
}
}
pub struct IntelClient {
list: RefCell<EndpointList>,
timeout: Duration,
trace_id: Option<String>,
alldown: Option<AllDownPolicy>,
health_reporter: Option<Box<dyn Fn(IntelHealthReport)>>,
last_all_down: std::cell::Cell<bool>,
}
pub struct IntelHealthReport {
pub all_down: bool,
pub active: Option<(usize, &'static str)>,
}
#[derive(Debug, Clone, Copy)]
pub struct AllDownPolicy {
pub max_retries: u32,
pub base: Duration,
pub max: Duration,
}
impl Default for AllDownPolicy {
fn default() -> AllDownPolicy {
AllDownPolicy {
max_retries: 8,
base: Duration::from_secs(1),
max: Duration::from_secs(30),
}
}
}
#[derive(Debug)]
pub enum Transport {
Tcp { host: String, port: u16, tls: bool },
}
impl IntelClient {
pub fn from_parts(uri: &str, default_token: Option<String>) -> Result<IntelClient, IntelError> {
let list = EndpointList::parse(uri, default_token)?;
Ok(IntelClient {
list: RefCell::new(list),
timeout: Duration::from_secs(120),
trace_id: None,
alldown: None,
health_reporter: None,
last_all_down: std::cell::Cell::new(false),
})
}
pub fn with_headers(self, headers: Vec<(String, String)>) -> IntelClient {
if !headers.is_empty() {
self.list.borrow_mut().set_extra_headers(headers);
}
self
}
pub fn with_signer(
self,
signer: Option<std::sync::Arc<dyn ::mcp::http::RequestSigner>>,
) -> IntelClient {
if let Some(s) = signer {
self.list.borrow_mut().set_signer(s);
}
self
}
pub fn with_dialect(self, dialect: Option<&str>) -> IntelClient {
if let Some(p) = Provider::from_dialect(dialect)
&& p != Provider::OpenAiCompatible
{
self.list.borrow_mut().set_provider(p);
}
self
}
pub fn set_health_reporter(&mut self, reporter: Box<dyn Fn(IntelHealthReport)>) {
self.health_reporter = Some(reporter);
}
pub fn set_trace_id(&mut self, trace_id: Option<String>) {
self.trace_id = trace_id;
}
pub fn enable_alldown_backoff(&mut self, policy: AllDownPolicy) {
self.alldown = Some(policy);
}
pub fn endpoint_count(&self) -> usize {
self.list.borrow().len()
}
pub fn trace_id(&self) -> Option<&str> {
self.trace_id.as_deref()
}
pub fn alldown_enabled(&self) -> bool {
self.alldown.is_some()
}
pub fn complete(&self, req: &Request) -> Result<Response, IntelError> {
crate::obs::metrics::record_intel_call();
let mut attempt: u32 = 0;
loop {
let sweep = {
let mut list = self.list.borrow_mut();
failover::complete_resilient(&mut list, req, self.timeout, self.trace_id.as_deref())
};
{
let list = self.list.borrow();
let all_down = list.all_down();
let active_up = list.ep(list.active()).health.is_up();
crate::obs::metrics::set_intel_up(active_up && !all_down);
if let Some(report) = &self.health_reporter
&& self.last_all_down.replace(all_down) != all_down
{
let active = (!all_down).then(|| list.active_identity());
report(IntelHealthReport { all_down, active });
}
}
match sweep.outcome {
Ok(resp) => return Ok(resp),
Err(e) => {
crate::obs::metrics::record_intel_error(error_reason(&e));
let backoff = match (&self.alldown, &e) {
(Some(p), IntelError::AllEndpointsDown(cause))
if !cause.as_deref().is_some_and(failover::is_auth)
&& attempt < p.max_retries =>
{
*p
}
_ => return Err(e),
};
let delay = backoff_delay(&backoff, attempt);
attempt += 1;
std::thread::sleep(delay);
}
}
}
}
pub fn with_list<R>(&self, f: impl FnOnce(&EndpointList) -> R) -> R {
f(&self.list.borrow())
}
}
fn backoff_delay(policy: &AllDownPolicy, attempt: u32) -> Duration {
let shift = attempt.min(20);
let scaled = policy.base.saturating_mul(1u32 << shift).min(policy.max);
let ms = scaled.as_millis() as u64;
let seed = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_nanos() as u64)
.unwrap_or(0)
^ (attempt as u64).wrapping_mul(0x9E37_79B9_7F4A_7C15);
let mut z = seed.wrapping_add(0x9E37_79B9_7F4A_7C15);
z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9);
z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB);
z ^= z >> 31;
let lo = ms.saturating_sub(ms / 4);
let window = (ms / 2) + 1;
Duration::from_millis(lo + z % window)
}
fn error_reason(e: &IntelError) -> &'static str {
match e {
IntelError::Transport(io) => match io.kind() {
std::io::ErrorKind::TimedOut | std::io::ErrorKind::WouldBlock => "timeout",
_ => "unreachable",
},
IntelError::Http(401 | 403, _) => "auth",
IntelError::Http(c, _) if (500..600).contains(c) => "5xx",
IntelError::Http(_, _) => "other",
IntelError::Parse(_) | IntelError::Unsupported(_) => "other",
IntelError::AllEndpointsDown(Some(cause)) => error_reason(cause),
IntelError::AllEndpointsDown(None) => "unreachable",
}
}
pub(super) fn resolve(
uri: &str,
provider: Provider,
) -> Result<(Transport, String, String), IntelError> {
#[cfg(any(feature = "internal-mocks", debug_assertions))]
if let Some(script) = uri.strip_prefix("mock:") {
let addr = super::mock::inprocess(script).map_err(IntelError::Unsupported)?;
return resolve(&format!("http://{addr}"), provider);
}
#[cfg(not(any(feature = "internal-mocks", debug_assertions)))]
if uri.starts_with("mock:") {
return Err(IntelError::Unsupported(
"mock: intelligence needs a build with --features internal-mocks".into(),
));
}
let url = Url::parse(uri).map_err(|_| {
IntelError::Unsupported(format!(
"intelligence endpoint must be https://host[:port][/path] (got: {uri})"
))
})?;
let tls = url.is_tls();
if !tls && !crate::net::http::is_loopback_host(&url.host) {
return Err(IntelError::Unsupported(format!(
"plaintext http:// intelligence is allowed for loopback only (dev); use https:// (got: {uri})"
)));
}
let http_path = if url.path == "/" {
provider.default_path().to_string()
} else {
url.path.clone()
};
let host_header = url.host_header();
Ok((
Transport::Tcp {
host: url.host,
port: url.port,
tls,
},
http_path,
host_header,
))
}
impl Transport {
pub(super) fn connect(&self, timeout: Duration) -> Result<Box<dyn Stream>, IntelError> {
use crate::net::http;
match self {
Transport::Tcp {
host,
port,
tls: false,
} => Ok(Box::new(http::connect_tcp(host, *port, timeout)?)),
Transport::Tcp {
host,
port,
tls: true,
} => connect_tls(host, *port, timeout),
}
}
}
#[cfg(feature = "tls")]
fn connect_tls(host: &str, port: u16, timeout: Duration) -> Result<Box<dyn Stream>, IntelError> {
let tcp = crate::net::http::connect_tcp(host, port, timeout)?;
Ok(Box::new(
crate::net::tls::connect(tcp, host, None).map_err(IntelError::Transport)?,
))
}
#[cfg(not(feature = "tls"))]
fn connect_tls(_host: &str, _port: u16, _timeout: Duration) -> Result<Box<dyn Stream>, IntelError> {
Err(IntelError::Unsupported(
"https:// intelligence requires building with --features tls".into(),
))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn resolve_rejects_non_https_transports() {
for uri in ["unix:/run/intel.sock", "vsock:2:8080", "not-a-url"] {
let err = resolve(uri, Provider::OpenAiCompatible).unwrap_err();
assert!(
matches!(err, IntelError::Unsupported(_)),
"{uri} must be rejected, got: {err:?}"
);
}
}
#[test]
fn resolve_allows_plaintext_http_for_loopback_only() {
for uri in [
"http://127.0.0.1:8080",
"http://localhost:8080",
"http://[::1]:8080",
] {
let (t, _p, _h) = resolve(uri, Provider::OpenAiCompatible).unwrap();
assert!(matches!(t, Transport::Tcp { tls: false, .. }), "{uri}");
}
let err = resolve("http://intel.example:8080", Provider::OpenAiCompatible).unwrap_err();
assert!(
matches!(err, IntelError::Unsupported(m) if m.contains("loopback")),
"non-loopback plaintext must be rejected"
);
}
#[test]
fn resolve_https_full_url() {
let (t, path, host) = resolve(
"https://api.openai.com/v1/chat/completions",
Provider::OpenAiCompatible,
)
.unwrap();
assert!(matches!(
t,
Transport::Tcp {
tls: true,
port: 443,
..
}
));
assert_eq!(path, "/v1/chat/completions");
assert_eq!(host, "api.openai.com");
}
#[test]
fn resolve_https_host_only_uses_default_path() {
let (_t, path, _host) =
resolve("https://gateway.local", Provider::OpenAiCompatible).unwrap();
assert_eq!(path, "/v1/chat/completions");
}
#[test]
fn single_endpoint_client_builds() {
let c = IntelClient::from_parts("https://intel.example", None).unwrap();
assert_eq!(c.endpoint_count(), 1);
}
#[test]
fn comma_list_client_builds_with_all_endpoints() {
let c = IntelClient::from_parts(
"https://a.example,https://b.example,https://c.example",
None,
)
.unwrap();
assert_eq!(c.endpoint_count(), 3);
}
#[test]
fn all_endpoints_down_maps_to_unreachable_reason() {
let cause = Box::new(IntelError::Http(503, "x".into()));
assert_eq!(
error_reason(&IntelError::AllEndpointsDown(Some(cause))),
"5xx"
);
assert_eq!(
error_reason(&IntelError::AllEndpointsDown(None)),
"unreachable"
);
assert_eq!(error_reason(&IntelError::Http(401, "x".into())), "auth");
}
#[test]
fn trace_header_propagates_to_endpoint_dialect() {
let mut c = IntelClient::from_parts("https://intel.example", None).unwrap();
assert!(c.trace_id.is_none());
c.set_trace_id(Some("1234567890abcdef1234567890abcdef".into()));
assert!(c.trace_id.is_some());
}
}