use crate::catalog::{
CatalogError, CatalogSnapshot, NormalizedModelSpec, ProviderCatalogDiscovery,
};
use crate::http_transport::{ResolvedSecret, SecretResolutionError, SecretResolver};
use futures_util::StreamExt;
use reqwest::header::{ACCEPT, AUTHORIZATION, CONTENT_TYPE};
use router_protocol::{
ANTHROPIC_ORIGIN, CatalogSourceKind, DASHSCOPE_TOKEN_PLAN_ORIGIN, DEEPSEEK_ORIGIN,
GEMINI_ORIGIN, MOONSHOT_ORIGIN, XAI_ORIGIN, ZAI_ORIGIN,
};
use serde::Deserialize;
use serde_json::{Map, Value, json};
use sha2::{Digest, Sha256};
use std::collections::{BTreeSet, HashSet};
use std::fmt;
use std::future::Future;
use std::pin::Pin;
use std::time::Duration;
use thiserror::Error;
pub const MAX_CATALOG_RESPONSE_BYTES: usize = 2 * 1024 * 1024;
pub const MAX_CATALOG_MODELS: usize = 512;
pub const MAX_SHOW_ENRICHMENTS: usize = 16;
pub const MAX_SHOW_RESPONSE_BYTES: usize = 256 * 1024;
pub const CATALOG_REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
const MAX_ERROR_RESPONSE_BYTES: usize = 32 * 1024;
const MAX_REQUEST_BODY_BYTES: usize = 512;
const CATALOG_USER_AGENT: &str = "steadymux-catalog/0.1";
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum CatalogEndpoint {
OllamaCloudTags,
OllamaCloudShow,
OpenRouterUserModels,
OpenRouterPublicModels,
LocalOllamaTags,
LocalOllamaShow,
DeepseekModels,
QwenModels,
XaiModels,
MoonshotModels,
ZaiModels,
AnthropicModels,
GeminiModels,
}
impl CatalogEndpoint {
const fn method(self) -> &'static str {
match self {
Self::OllamaCloudShow | Self::LocalOllamaShow => "POST",
Self::OllamaCloudTags
| Self::OpenRouterUserModels
| Self::OpenRouterPublicModels
| Self::LocalOllamaTags
| Self::DeepseekModels
| Self::QwenModels
| Self::XaiModels
| Self::MoonshotModels
| Self::ZaiModels
| Self::AnthropicModels
| Self::GeminiModels => "GET",
}
}
const fn origin(self) -> &'static str {
match self {
Self::OllamaCloudTags | Self::OllamaCloudShow => "https://ollama.com",
Self::OpenRouterUserModels | Self::OpenRouterPublicModels => "https://openrouter.ai",
Self::LocalOllamaTags | Self::LocalOllamaShow => "http://127.0.0.1:11434",
Self::DeepseekModels => DEEPSEEK_ORIGIN,
Self::QwenModels => DASHSCOPE_TOKEN_PLAN_ORIGIN,
Self::XaiModels => XAI_ORIGIN,
Self::MoonshotModels => MOONSHOT_ORIGIN,
Self::ZaiModels => ZAI_ORIGIN,
Self::AnthropicModels => ANTHROPIC_ORIGIN,
Self::GeminiModels => GEMINI_ORIGIN,
}
}
const fn path(self) -> &'static str {
match self {
Self::OllamaCloudTags | Self::LocalOllamaTags => "/api/tags",
Self::OllamaCloudShow | Self::LocalOllamaShow => "/api/show",
Self::OpenRouterUserModels => "/api/v1/models/user",
Self::OpenRouterPublicModels => "/api/v1/models",
Self::DeepseekModels => "/models",
Self::QwenModels => "/compatible-mode/v1/models",
Self::XaiModels | Self::MoonshotModels => "/v1/models",
Self::ZaiModels => "/api/paas/v4/models",
Self::AnthropicModels | Self::GeminiModels => "/v1/models",
}
}
const fn requires_authorization(self) -> bool {
matches!(
self,
Self::OllamaCloudTags
| Self::OllamaCloudShow
| Self::OpenRouterUserModels
| Self::DeepseekModels
| Self::QwenModels
| Self::XaiModels
| Self::MoonshotModels
| Self::ZaiModels
| Self::AnthropicModels
| Self::GeminiModels
)
}
const fn allows_authorization(self) -> bool {
self.requires_authorization()
}
const fn response_limit(self) -> usize {
match self {
Self::OllamaCloudShow | Self::LocalOllamaShow => MAX_SHOW_RESPONSE_BYTES,
_ => MAX_CATALOG_RESPONSE_BYTES,
}
}
}
pub struct CatalogHttpRequest {
endpoint: CatalogEndpoint,
body: Vec<u8>,
authorization: Option<ResolvedSecret>,
pub auth_header_name: Option<&'static str>,
pub version_header: Option<(&'static str, &'static str)>,
}
impl CatalogHttpRequest {
pub const fn endpoint(&self) -> CatalogEndpoint {
self.endpoint
}
pub const fn method(&self) -> &'static str {
self.endpoint.method()
}
pub const fn origin(&self) -> &'static str {
self.endpoint.origin()
}
pub const fn path(&self) -> &'static str {
self.endpoint.path()
}
pub fn body_for_dispatch(&self) -> &[u8] {
&self.body
}
pub fn authorization_for_dispatch(&self) -> Option<&[u8]> {
self.authorization
.as_ref()
.map(ResolvedSecret::expose_for_dispatch)
}
}
impl fmt::Debug for CatalogHttpRequest {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("CatalogHttpRequest")
.field("endpoint", &self.endpoint)
.field("body", &"[REDACTED]")
.field(
"authorization",
&self.authorization.as_ref().map(|_| "[REDACTED]"),
)
.field("auth_header_name", &self.auth_header_name)
.field("version_header", &self.version_header)
.finish()
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct CatalogHttpResponse {
pub status: u16,
pub content_type: Option<String>,
pub redirected: bool,
pub body: Vec<u8>,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum CatalogHttpErrorKind {
Cancelled,
Timeout,
Transport,
InvalidRequest,
ResponseTooLarge,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Error)]
#[error("catalog HTTP execution failed")]
pub struct CatalogHttpError {
pub kind: CatalogHttpErrorKind,
}
pub type CatalogHttpFuture<'a> =
Pin<Box<dyn Future<Output = Result<CatalogHttpResponse, CatalogHttpError>> + Send + 'a>>;
pub trait CatalogHttpExecutor: Send + Sync {
fn execute<'a>(&'a self, request: CatalogHttpRequest) -> CatalogHttpFuture<'a>;
}
#[derive(Clone)]
pub struct ReqwestCatalogExecutor {
client: reqwest::Client,
}
impl fmt::Debug for ReqwestCatalogExecutor {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("ReqwestCatalogExecutor")
.field("redirects", &"disabled")
.field("proxy", &"disabled")
.finish()
}
}
impl ReqwestCatalogExecutor {
pub fn new(
connect_timeout: Duration,
read_timeout: Duration,
) -> Result<Self, CatalogHttpError> {
if connect_timeout.is_zero() || read_timeout.is_zero() {
return Err(catalog_http_error(CatalogHttpErrorKind::InvalidRequest));
}
let client = reqwest::Client::builder()
.use_rustls_tls()
.redirect(reqwest::redirect::Policy::none())
.no_proxy()
.user_agent(CATALOG_USER_AGENT)
.connect_timeout(connect_timeout)
.read_timeout(read_timeout)
.build()
.map_err(|_| catalog_http_error(CatalogHttpErrorKind::InvalidRequest))?;
Ok(Self { client })
}
async fn execute_bounded(
&self,
request: CatalogHttpRequest,
) -> Result<CatalogHttpResponse, CatalogHttpError> {
validate_catalog_request(&request)?;
let exact_url = format!("{}{}", request.origin(), request.path());
let mut builder = if request.method() == "GET" {
self.client.get(&exact_url)
} else {
self.client
.post(&exact_url)
.header(CONTENT_TYPE, "application/json")
.body(request.body)
}
.header(ACCEPT, "application/json");
if let Some(authorization) = request.authorization {
let mut header =
reqwest::header::HeaderValue::from_bytes(authorization.expose_for_dispatch())
.map_err(|_| catalog_http_error(CatalogHttpErrorKind::InvalidRequest))?;
header.set_sensitive(true);
builder = match request.auth_header_name {
Some(name) => builder.header(name, header),
None => builder.header(AUTHORIZATION, header),
};
}
if let Some((name, value)) = request.version_header {
let header = reqwest::header::HeaderValue::from_str(value)
.map_err(|_| catalog_http_error(CatalogHttpErrorKind::InvalidRequest))?;
builder = builder.header(name, header);
}
let response = builder.send().await.map_err(|error| {
catalog_http_error(if error.is_timeout() {
CatalogHttpErrorKind::Timeout
} else {
CatalogHttpErrorKind::Transport
})
})?;
let status = response.status().as_u16();
let content_type = response
.headers()
.get(CONTENT_TYPE)
.and_then(|value| value.to_str().ok())
.filter(|value| value.len() <= 128)
.map(str::to_owned);
let redirected = response.url().as_str() != exact_url;
let limit = if status == 200 {
request.endpoint.response_limit()
} else {
MAX_ERROR_RESPONSE_BYTES
};
let mut stream = response.bytes_stream();
let mut body = Vec::new();
while let Some(chunk) = stream.next().await {
let chunk = chunk.map_err(|error| {
catalog_http_error(if error.is_timeout() {
CatalogHttpErrorKind::Timeout
} else {
CatalogHttpErrorKind::Transport
})
})?;
if body.len().saturating_add(chunk.len()) > limit {
return Err(catalog_http_error(CatalogHttpErrorKind::ResponseTooLarge));
}
body.extend_from_slice(&chunk);
}
Ok(CatalogHttpResponse {
status,
content_type,
redirected,
body,
})
}
}
impl CatalogHttpExecutor for ReqwestCatalogExecutor {
fn execute<'a>(&'a self, request: CatalogHttpRequest) -> CatalogHttpFuture<'a> {
Box::pin(async move { self.execute_bounded(request).await })
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct CatalogSnapshotRequest {
pub snapshot_id: String,
pub catalog_version: String,
pub observed_at_epoch_seconds: u64,
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct OllamaDiscoveryOptions {
pub max_show_enrichments: usize,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum OpenRouterMalformedStage {
Json,
DataEnvelope,
ModelObject,
ModelId,
DuplicateModelId,
ContextLength,
Modalities,
MissingConfiguredModel,
Normalization,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Error)]
pub enum CatalogDiscoveryError {
#[error("catalog discovery request is invalid")]
InvalidRequest,
#[error("catalog credential resolution failed")]
SecretResolution,
#[error("catalog discovery timed out")]
Timeout,
#[error("catalog discovery transport failed")]
Transport,
#[error("catalog discovery response exceeded its bound")]
ResponseBounds,
#[error("catalog discovery returned a disallowed HTTP response")]
HttpResponse,
#[error("catalog discovery response was malformed")]
Malformed,
#[error("OpenRouter catalog discovery was malformed")]
OpenRouterMalformed(OpenRouterMalformedStage),
#[error("catalog snapshot was rejected")]
SnapshotRejected,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct OllamaNativeCatalog {
models: Vec<OllamaNativeModel>,
}
#[derive(Clone, Debug, PartialEq, Eq)]
struct OllamaNativeModel {
id: String,
location: OllamaModelLocation,
context_tokens: u64,
modalities: Vec<String>,
tool_support: bool,
reasoning_modes: Vec<String>,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum OllamaModelLocation {
Cloud,
Local,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct OllamaCatalogCodec {
source: CatalogSourceKind,
}
impl OllamaCatalogCodec {
pub const fn cloud() -> Self {
Self {
source: CatalogSourceKind::OllamaCloudCatalog,
}
}
pub const fn local() -> Self {
Self {
source: CatalogSourceKind::OllamaLocalInventory,
}
}
fn includes(self, model: &OllamaNativeModel) -> bool {
matches!(
(self.source, model.location),
(
CatalogSourceKind::OllamaCloudCatalog,
OllamaModelLocation::Cloud
) | (
CatalogSourceKind::OllamaLocalInventory,
OllamaModelLocation::Local
)
)
}
}
impl ProviderCatalogDiscovery for OllamaCatalogCodec {
type NativeResponse = OllamaNativeCatalog;
fn source(&self) -> CatalogSourceKind {
self.source
}
fn normalize(
&self,
response: &Self::NativeResponse,
observed_at_epoch_seconds: u64,
) -> Result<Vec<NormalizedModelSpec>, CatalogError> {
let pricing_provenance = match self.source {
CatalogSourceKind::OllamaCloudCatalog => "ollama_cloud_unpriced",
CatalogSourceKind::OllamaLocalInventory => "local_compute",
_ => return Err(CatalogError::Malformed),
};
let mut models = response
.models
.iter()
.map(|model| NormalizedModelSpec {
provider_model_id: model.id.clone(),
context_tokens: model.context_tokens,
modalities: model.modalities.clone(),
tool_support: model.tool_support,
reasoning_modes: model.reasoning_modes.clone(),
pricing_provenance: pricing_provenance.to_owned(),
availability: "available".to_owned(),
observed_at_epoch_seconds,
})
.collect::<Vec<_>>();
models.sort_by(|left, right| left.provider_model_id.cmp(&right.provider_model_id));
Ok(models)
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct OpenRouterNativeCatalog {
models: Vec<OpenRouterNativeModel>,
authenticated: bool,
}
#[derive(Clone, Debug, PartialEq, Eq)]
struct OpenRouterNativeModel {
id: String,
context_tokens: u64,
modalities: Vec<String>,
tool_support: bool,
reasoning_modes: Vec<String>,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct OpenRouterCatalogCodec;
impl ProviderCatalogDiscovery for OpenRouterCatalogCodec {
type NativeResponse = OpenRouterNativeCatalog;
fn source(&self) -> CatalogSourceKind {
CatalogSourceKind::OpenrouterCatalog
}
fn normalize(
&self,
response: &Self::NativeResponse,
observed_at_epoch_seconds: u64,
) -> Result<Vec<NormalizedModelSpec>, CatalogError> {
let provenance = if response.authenticated {
"openrouter_user_catalog"
} else {
"openrouter_public_catalog"
};
let mut models = response
.models
.iter()
.map(|model| NormalizedModelSpec {
provider_model_id: model.id.clone(),
context_tokens: model.context_tokens,
modalities: model.modalities.clone(),
tool_support: model.tool_support,
reasoning_modes: model.reasoning_modes.clone(),
pricing_provenance: provenance.to_owned(),
availability: "available".to_owned(),
observed_at_epoch_seconds,
})
.collect::<Vec<_>>();
models.sort_by(|left, right| left.provider_model_id.cmp(&right.provider_model_id));
Ok(models)
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct OpenAiCompatNativeCatalog {
models: Vec<OpenAiCompatNativeModel>,
}
#[derive(Clone, Debug, PartialEq, Eq)]
struct OpenAiCompatNativeModel {
id: String,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct OpenAiCompatCatalogCodec {
source: CatalogSourceKind,
}
impl OpenAiCompatCatalogCodec {
pub const fn deepseek() -> Self {
Self {
source: CatalogSourceKind::DeepseekCatalog,
}
}
pub const fn qwen() -> Self {
Self {
source: CatalogSourceKind::QwenCatalog,
}
}
pub const fn xai() -> Self {
Self {
source: CatalogSourceKind::GrokCatalog,
}
}
pub const fn moonshot() -> Self {
Self {
source: CatalogSourceKind::KimiCatalog,
}
}
pub const fn zai() -> Self {
Self {
source: CatalogSourceKind::ZaiCatalog,
}
}
pub const fn anthropic() -> Self {
Self {
source: CatalogSourceKind::ClaudeCatalog,
}
}
}
impl ProviderCatalogDiscovery for OpenAiCompatCatalogCodec {
type NativeResponse = OpenAiCompatNativeCatalog;
fn source(&self) -> CatalogSourceKind {
self.source
}
fn normalize(
&self,
response: &Self::NativeResponse,
observed_at_epoch_seconds: u64,
) -> Result<Vec<NormalizedModelSpec>, CatalogError> {
let pricing_provenance = match self.source {
CatalogSourceKind::DeepseekCatalog => "deepseek_api_catalog",
CatalogSourceKind::QwenCatalog => "qwen_token_plan_catalog",
CatalogSourceKind::GrokCatalog => "grok_api_catalog",
CatalogSourceKind::KimiCatalog => "kimi_api_catalog",
CatalogSourceKind::ZaiCatalog => "zai_api_catalog",
CatalogSourceKind::ClaudeCatalog => "anthropic_api_catalog",
CatalogSourceKind::GeminiCatalog => "gemini_api_catalog",
_ => return Err(CatalogError::Malformed),
};
let mut models = response
.models
.iter()
.map(|model| NormalizedModelSpec {
provider_model_id: model.id.clone(),
context_tokens: 1,
modalities: vec!["text".to_owned()],
tool_support: false,
reasoning_modes: Vec::new(),
pricing_provenance: pricing_provenance.to_owned(),
availability: "available".to_owned(),
observed_at_epoch_seconds,
})
.collect::<Vec<_>>();
models.sort_by(|left, right| left.provider_model_id.cmp(&right.provider_model_id));
Ok(models)
}
}
fn decode_openai_compat_models(
bytes: &[u8],
) -> Result<OpenAiCompatNativeCatalog, CatalogDiscoveryError> {
let root: Value =
serde_json::from_slice(bytes).map_err(|_| CatalogDiscoveryError::Malformed)?;
let values = root
.get("data")
.and_then(Value::as_array)
.ok_or(CatalogDiscoveryError::Malformed)?;
if values.is_empty() || values.len() > MAX_CATALOG_MODELS {
return Err(CatalogDiscoveryError::ResponseBounds);
}
let mut seen = HashSet::with_capacity(values.len());
let mut models = Vec::with_capacity(values.len());
for value in values {
let Some(id) = value.get("id").and_then(Value::as_str) else {
return Err(CatalogDiscoveryError::Malformed);
};
if !valid_model_id(id) || !seen.insert(id.to_owned()) {
return Err(CatalogDiscoveryError::Malformed);
}
models.push(OpenAiCompatNativeModel { id: id.to_owned() });
}
models.sort_by(|left, right| left.id.cmp(&right.id));
Ok(OpenAiCompatNativeCatalog { models })
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct GeminiNativeModel {
id: String,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct GeminiNativeCatalog {
pub models: Vec<GeminiNativeModel>,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct GeminiCatalogCodec;
impl ProviderCatalogDiscovery for GeminiCatalogCodec {
type NativeResponse = GeminiNativeCatalog;
fn source(&self) -> CatalogSourceKind {
CatalogSourceKind::GeminiCatalog
}
fn normalize(
&self,
response: &Self::NativeResponse,
observed_at_epoch_seconds: u64,
) -> Result<Vec<NormalizedModelSpec>, CatalogError> {
let mut models = response
.models
.iter()
.map(|model| NormalizedModelSpec {
provider_model_id: model.id.clone(),
context_tokens: 1,
modalities: vec!["text".to_owned()],
tool_support: false,
reasoning_modes: Vec::new(),
pricing_provenance: "gemini_api_catalog".to_owned(),
availability: "available".to_owned(),
observed_at_epoch_seconds,
})
.collect::<Vec<_>>();
models.sort_by(|left, right| left.provider_model_id.cmp(&right.provider_model_id));
Ok(models)
}
}
fn decode_gemini_models(bytes: &[u8]) -> Result<GeminiNativeCatalog, CatalogDiscoveryError> {
let root: Value =
serde_json::from_slice(bytes).map_err(|_| CatalogDiscoveryError::Malformed)?;
let values = root
.get("models")
.and_then(Value::as_array)
.ok_or(CatalogDiscoveryError::Malformed)?;
if values.is_empty() || values.len() > MAX_CATALOG_MODELS {
return Err(CatalogDiscoveryError::ResponseBounds);
}
let mut seen = HashSet::with_capacity(values.len());
let mut models = Vec::with_capacity(values.len());
for value in values {
let name = value
.get("name")
.and_then(Value::as_str)
.ok_or(CatalogDiscoveryError::Malformed)?;
let id = name
.strip_prefix("models/")
.ok_or(CatalogDiscoveryError::Malformed)?;
if !valid_model_id(id) || !seen.insert(id.to_owned()) {
return Err(CatalogDiscoveryError::Malformed);
}
models.push(GeminiNativeModel { id: id.to_owned() });
}
models.sort_by(|left, right| left.id.cmp(&right.id));
Ok(GeminiNativeCatalog { models })
}
pub async fn discover_deepseek(
executor: &dyn CatalogHttpExecutor,
resolver: &dyn SecretResolver,
secret_ref: &str,
snapshot: CatalogSnapshotRequest,
) -> Result<CatalogSnapshot, CatalogDiscoveryError> {
let request = catalog_request(
CatalogEndpoint::DeepseekModels,
None,
Some(CatalogCredential::Resolver {
resolver,
secret_ref,
}),
)?;
let response = execute_catalog(executor, request).await?;
validate_success_response(CatalogEndpoint::DeepseekModels, &response)?;
let native = decode_openai_compat_models(&response.body)?;
let models = OpenAiCompatCatalogCodec::deepseek()
.normalize(&native, snapshot.observed_at_epoch_seconds)
.map_err(|_| CatalogDiscoveryError::Malformed)?;
accept_snapshot(
snapshot,
"deepseek",
CatalogSourceKind::DeepseekCatalog,
receipt_for(&[response.body]),
models,
)
}
pub async fn discover_qwen(
executor: &dyn CatalogHttpExecutor,
resolver: &dyn SecretResolver,
secret_ref: &str,
snapshot: CatalogSnapshotRequest,
) -> Result<CatalogSnapshot, CatalogDiscoveryError> {
let request = catalog_request(
CatalogEndpoint::QwenModels,
None,
Some(CatalogCredential::Resolver {
resolver,
secret_ref,
}),
)?;
let response = execute_catalog(executor, request).await?;
validate_success_response(CatalogEndpoint::QwenModels, &response)?;
let native = decode_openai_compat_models(&response.body)?;
let models = OpenAiCompatCatalogCodec::qwen()
.normalize(&native, snapshot.observed_at_epoch_seconds)
.map_err(|_| CatalogDiscoveryError::Malformed)?;
accept_snapshot(
snapshot,
"qwen",
CatalogSourceKind::QwenCatalog,
receipt_for(&[response.body]),
models,
)
}
pub async fn discover_xai(
executor: &dyn CatalogHttpExecutor,
resolver: &dyn SecretResolver,
secret_ref: &str,
snapshot: CatalogSnapshotRequest,
) -> Result<CatalogSnapshot, CatalogDiscoveryError> {
let request = catalog_request(
CatalogEndpoint::XaiModels,
None,
Some(CatalogCredential::Resolver {
resolver,
secret_ref,
}),
)?;
let response = execute_catalog(executor, request).await?;
validate_success_response(CatalogEndpoint::XaiModels, &response)?;
let native = decode_openai_compat_models(&response.body)?;
let models = OpenAiCompatCatalogCodec::xai()
.normalize(&native, snapshot.observed_at_epoch_seconds)
.map_err(|_| CatalogDiscoveryError::Malformed)?;
accept_snapshot(
snapshot,
"grok",
CatalogSourceKind::GrokCatalog,
receipt_for(&[response.body]),
models,
)
}
pub async fn discover_moonshot(
executor: &dyn CatalogHttpExecutor,
resolver: &dyn SecretResolver,
secret_ref: &str,
snapshot: CatalogSnapshotRequest,
) -> Result<CatalogSnapshot, CatalogDiscoveryError> {
let request = catalog_request(
CatalogEndpoint::MoonshotModels,
None,
Some(CatalogCredential::Resolver {
resolver,
secret_ref,
}),
)?;
let response = execute_catalog(executor, request).await?;
validate_success_response(CatalogEndpoint::MoonshotModels, &response)?;
let native = decode_openai_compat_models(&response.body)?;
let models = OpenAiCompatCatalogCodec::moonshot()
.normalize(&native, snapshot.observed_at_epoch_seconds)
.map_err(|_| CatalogDiscoveryError::Malformed)?;
accept_snapshot(
snapshot,
"kimi",
CatalogSourceKind::KimiCatalog,
receipt_for(&[response.body]),
models,
)
}
pub async fn discover_zai(
executor: &dyn CatalogHttpExecutor,
resolver: &dyn SecretResolver,
secret_ref: &str,
snapshot: CatalogSnapshotRequest,
) -> Result<CatalogSnapshot, CatalogDiscoveryError> {
let request = catalog_request(
CatalogEndpoint::ZaiModels,
None,
Some(CatalogCredential::Resolver {
resolver,
secret_ref,
}),
)?;
let response = execute_catalog(executor, request).await?;
validate_success_response(CatalogEndpoint::ZaiModels, &response)?;
let native = decode_openai_compat_models(&response.body)?;
let models = OpenAiCompatCatalogCodec::zai()
.normalize(&native, snapshot.observed_at_epoch_seconds)
.map_err(|_| CatalogDiscoveryError::Malformed)?;
accept_snapshot(
snapshot,
"zai",
CatalogSourceKind::ZaiCatalog,
receipt_for(&[response.body]),
models,
)
}
pub async fn discover_anthropic(
executor: &dyn CatalogHttpExecutor,
resolver: &dyn SecretResolver,
secret_ref: &str,
snapshot: CatalogSnapshotRequest,
) -> Result<CatalogSnapshot, CatalogDiscoveryError> {
let request = catalog_request(
CatalogEndpoint::AnthropicModels,
None,
Some(CatalogCredential::Resolver {
resolver,
secret_ref,
}),
)?;
let response = execute_catalog(executor, request).await?;
validate_success_response(CatalogEndpoint::AnthropicModels, &response)?;
let native = decode_openai_compat_models(&response.body)?;
let models = OpenAiCompatCatalogCodec::anthropic()
.normalize(&native, snapshot.observed_at_epoch_seconds)
.map_err(|_| CatalogDiscoveryError::Malformed)?;
accept_snapshot(
snapshot,
"claude",
CatalogSourceKind::ClaudeCatalog,
receipt_for(&[response.body]),
models,
)
}
pub async fn discover_gemini(
executor: &dyn CatalogHttpExecutor,
resolver: &dyn SecretResolver,
secret_ref: &str,
snapshot: CatalogSnapshotRequest,
) -> Result<CatalogSnapshot, CatalogDiscoveryError> {
let request = catalog_request(
CatalogEndpoint::GeminiModels,
None,
Some(CatalogCredential::Resolver {
resolver,
secret_ref,
}),
)?;
let response = execute_catalog(executor, request).await?;
validate_success_response(CatalogEndpoint::GeminiModels, &response)?;
let native = decode_gemini_models(&response.body)?;
let models = GeminiCatalogCodec
.normalize(&native, snapshot.observed_at_epoch_seconds)
.map_err(|_| CatalogDiscoveryError::Malformed)?;
accept_snapshot(
snapshot,
"gemini",
CatalogSourceKind::GeminiCatalog,
receipt_for(&[response.body]),
models,
)
}
pub async fn discover_ollama_cloud(
executor: &dyn CatalogHttpExecutor,
resolver: &dyn SecretResolver,
secret_ref: &str,
snapshot: CatalogSnapshotRequest,
options: OllamaDiscoveryOptions,
) -> Result<CatalogSnapshot, CatalogDiscoveryError> {
discover_ollama(
executor,
Some(CatalogCredential::Resolver {
resolver,
secret_ref,
}),
CatalogEndpoint::OllamaCloudTags,
CatalogEndpoint::OllamaCloudShow,
snapshot,
options,
OllamaCatalogCodec::cloud(),
"ollama_cloud",
&[],
false,
)
.await
}
pub async fn discover_ollama_cloud_with_configured_models(
executor: &dyn CatalogHttpExecutor,
resolver: &dyn SecretResolver,
secret_ref: &str,
snapshot: CatalogSnapshotRequest,
options: OllamaDiscoveryOptions,
configured_model_ids: &[String],
) -> Result<CatalogSnapshot, CatalogDiscoveryError> {
discover_ollama(
executor,
Some(CatalogCredential::Resolver {
resolver,
secret_ref,
}),
CatalogEndpoint::OllamaCloudTags,
CatalogEndpoint::OllamaCloudShow,
snapshot,
options,
OllamaCatalogCodec::cloud(),
"ollama_cloud",
configured_model_ids,
false,
)
.await
}
pub async fn discover_ollama_cloud_with_resolved_secret_and_configured_models(
executor: &dyn CatalogHttpExecutor,
credential: ResolvedSecret,
snapshot: CatalogSnapshotRequest,
options: OllamaDiscoveryOptions,
configured_model_ids: &[String],
) -> Result<CatalogSnapshot, CatalogDiscoveryError> {
discover_ollama(
executor,
Some(CatalogCredential::Resolved(&credential)),
CatalogEndpoint::OllamaCloudTags,
CatalogEndpoint::OllamaCloudShow,
snapshot,
options,
OllamaCatalogCodec::cloud(),
"ollama_cloud",
configured_model_ids,
false,
)
.await
}
pub async fn discover_ollama_cloud_relay(
executor: &dyn CatalogHttpExecutor,
snapshot: CatalogSnapshotRequest,
options: OllamaDiscoveryOptions,
) -> Result<CatalogSnapshot, CatalogDiscoveryError> {
discover_ollama(
executor,
None,
CatalogEndpoint::LocalOllamaTags,
CatalogEndpoint::LocalOllamaShow,
snapshot,
options,
OllamaCatalogCodec::cloud(),
"ollama_cloud",
&[],
true,
)
.await
}
pub async fn discover_ollama_cloud_relay_with_configured_models(
executor: &dyn CatalogHttpExecutor,
snapshot: CatalogSnapshotRequest,
options: OllamaDiscoveryOptions,
configured_model_ids: &[String],
) -> Result<CatalogSnapshot, CatalogDiscoveryError> {
discover_ollama(
executor,
None,
CatalogEndpoint::LocalOllamaTags,
CatalogEndpoint::LocalOllamaShow,
snapshot,
options,
OllamaCatalogCodec::cloud(),
"ollama_cloud",
configured_model_ids,
true,
)
.await
}
pub async fn discover_local_ollama(
executor: &dyn CatalogHttpExecutor,
snapshot: CatalogSnapshotRequest,
options: OllamaDiscoveryOptions,
) -> Result<CatalogSnapshot, CatalogDiscoveryError> {
discover_ollama(
executor,
None,
CatalogEndpoint::LocalOllamaTags,
CatalogEndpoint::LocalOllamaShow,
snapshot,
options,
OllamaCatalogCodec::local(),
"ollama_local",
&[],
false,
)
.await
}
#[allow(clippy::too_many_arguments)]
async fn discover_ollama(
executor: &dyn CatalogHttpExecutor,
credential: Option<CatalogCredential<'_>>,
tags_endpoint: CatalogEndpoint,
show_endpoint: CatalogEndpoint,
snapshot: CatalogSnapshotRequest,
options: OllamaDiscoveryOptions,
codec: OllamaCatalogCodec,
provider_id: &str,
configured_model_ids: &[String],
configured_ids_require_cloud_suffix: bool,
) -> Result<CatalogSnapshot, CatalogDiscoveryError> {
if options.max_show_enrichments > MAX_SHOW_ENRICHMENTS
|| configured_model_ids.len() > MAX_SHOW_ENRICHMENTS
|| configured_model_ids
.len()
.saturating_add(options.max_show_enrichments)
> MAX_SHOW_ENRICHMENTS
{
return Err(CatalogDiscoveryError::InvalidRequest);
}
let mut configured = HashSet::with_capacity(configured_model_ids.len());
if configured_model_ids.iter().any(|model_id| {
!valid_model_id(model_id)
|| !configured.insert(model_id.as_str())
|| (cloud_base_model_id(model_id).is_some() != configured_ids_require_cloud_suffix)
}) {
return Err(CatalogDiscoveryError::InvalidRequest);
}
let tags_request = catalog_request(tags_endpoint, None, credential)?;
let tags_response = execute_catalog(executor, tags_request).await?;
validate_success_response(tags_endpoint, &tags_response)?;
let mut receipt_bodies = vec![tags_response.body.clone()];
let mut native = decode_ollama_tags(&tags_response.body)?;
native.models.retain(|model| codec.includes(model));
let mut probed = HashSet::new();
for model_id in configured_model_ids {
if native.models.iter().any(|model| model.id == *model_id) {
continue;
}
let show_request = catalog_request(show_endpoint, Some(model_id), credential)?;
let show_response = execute_catalog(executor, show_request).await?;
validate_success_response(show_endpoint, &show_response)?;
let mut model = OllamaNativeModel {
id: model_id.clone(),
location: OllamaModelLocation::Cloud,
context_tokens: 1,
modalities: vec!["text".to_owned()],
tool_support: false,
reasoning_modes: Vec::new(),
};
apply_ollama_show(&mut model, &show_response.body)?;
native.models.push(model);
probed.insert(model_id.as_str());
receipt_bodies.push(show_response.body);
}
native.models.sort_by(|left, right| left.id.cmp(&right.id));
if native.models.is_empty() {
return Err(CatalogDiscoveryError::SnapshotRejected);
}
let mut remaining_enrichments = options.max_show_enrichments;
for model in &mut native.models {
if remaining_enrichments == 0 {
break;
}
if probed.contains(model.id.as_str()) {
continue;
}
let show_request = catalog_request(show_endpoint, Some(&model.id), credential)?;
let show_response = execute_catalog(executor, show_request).await?;
validate_success_response(show_endpoint, &show_response)?;
apply_ollama_show(model, &show_response.body)?;
receipt_bodies.push(show_response.body);
remaining_enrichments -= 1;
}
let models = codec
.normalize(&native, snapshot.observed_at_epoch_seconds)
.map_err(|_| CatalogDiscoveryError::Malformed)?;
accept_snapshot(
snapshot,
provider_id,
codec.source(),
receipt_for(&receipt_bodies),
models,
)
}
#[derive(Clone, Copy)]
enum CatalogCredential<'a> {
Resolver {
resolver: &'a dyn SecretResolver,
secret_ref: &'a str,
},
Resolved(&'a ResolvedSecret),
}
pub async fn discover_openrouter(
executor: &dyn CatalogHttpExecutor,
credential: Option<(&dyn SecretResolver, &str)>,
snapshot: CatalogSnapshotRequest,
) -> Result<CatalogSnapshot, CatalogDiscoveryError> {
let (response, authenticated) = if let Some(credential) = credential {
let request = catalog_request(
CatalogEndpoint::OpenRouterUserModels,
None,
Some(CatalogCredential::Resolver {
resolver: credential.0,
secret_ref: credential.1,
}),
)?;
let response = execute_catalog(executor, request).await?;
if matches!(response.status, 401 | 403 | 404) {
let public = catalog_request(CatalogEndpoint::OpenRouterPublicModels, None, None)?;
(execute_catalog(executor, public).await?, false)
} else {
(response, true)
}
} else {
let public = catalog_request(CatalogEndpoint::OpenRouterPublicModels, None, None)?;
(execute_catalog(executor, public).await?, false)
};
let endpoint = if authenticated {
CatalogEndpoint::OpenRouterUserModels
} else {
CatalogEndpoint::OpenRouterPublicModels
};
validate_success_response(endpoint, &response)?;
let native = decode_openrouter(&response.body, authenticated)?;
let models = OpenRouterCatalogCodec
.normalize(&native, snapshot.observed_at_epoch_seconds)
.map_err(|_| {
CatalogDiscoveryError::OpenRouterMalformed(OpenRouterMalformedStage::Normalization)
})?;
accept_snapshot(
snapshot,
"openrouter",
CatalogSourceKind::OpenrouterCatalog,
receipt_for(&[response.body]),
models,
)
}
pub async fn discover_openrouter_authenticated(
executor: &dyn CatalogHttpExecutor,
resolver: &dyn SecretResolver,
secret_ref: &str,
snapshot: CatalogSnapshotRequest,
) -> Result<CatalogSnapshot, CatalogDiscoveryError> {
let request = catalog_request(
CatalogEndpoint::OpenRouterUserModels,
None,
Some(CatalogCredential::Resolver {
resolver,
secret_ref,
}),
)?;
let response = execute_catalog(executor, request).await?;
validate_success_response(CatalogEndpoint::OpenRouterUserModels, &response)?;
let native = decode_openrouter(&response.body, true)?;
let models = OpenRouterCatalogCodec
.normalize(&native, snapshot.observed_at_epoch_seconds)
.map_err(|_| {
CatalogDiscoveryError::OpenRouterMalformed(OpenRouterMalformedStage::Normalization)
})?;
accept_snapshot(
snapshot,
"openrouter",
CatalogSourceKind::OpenrouterCatalog,
receipt_for(&[response.body]),
models,
)
}
pub async fn discover_openrouter_authenticated_with_configured_models(
executor: &dyn CatalogHttpExecutor,
resolver: &dyn SecretResolver,
secret_ref: &str,
snapshot: CatalogSnapshotRequest,
configured_model_ids: &[String],
) -> Result<CatalogSnapshot, CatalogDiscoveryError> {
let credential = resolver
.resolve(secret_ref)
.map_err(|_: SecretResolutionError| CatalogDiscoveryError::SecretResolution)?;
discover_openrouter_authenticated_with_resolved_secret_and_configured_models(
executor,
credential,
snapshot,
configured_model_ids,
)
.await
}
pub async fn discover_openrouter_authenticated_with_resolved_secret_and_configured_models(
executor: &dyn CatalogHttpExecutor,
credential: ResolvedSecret,
snapshot: CatalogSnapshotRequest,
configured_model_ids: &[String],
) -> Result<CatalogSnapshot, CatalogDiscoveryError> {
if configured_model_ids.is_empty()
|| configured_model_ids.len() > MAX_CATALOG_MODELS
|| configured_model_ids.iter().any(|id| !valid_model_id(id))
{
return Err(CatalogDiscoveryError::InvalidRequest);
}
let configured = configured_model_ids
.iter()
.map(String::as_str)
.collect::<HashSet<_>>();
if configured.len() != configured_model_ids.len() {
return Err(CatalogDiscoveryError::InvalidRequest);
}
let request = catalog_request_with_resolved_secret(
CatalogEndpoint::OpenRouterUserModels,
None,
Some(credential),
)?;
let response = execute_catalog(executor, request).await?;
validate_success_response(CatalogEndpoint::OpenRouterUserModels, &response)?;
let native = decode_openrouter_configured(&response.body, true, &configured)?;
let models = OpenRouterCatalogCodec
.normalize(&native, snapshot.observed_at_epoch_seconds)
.map_err(|_| {
CatalogDiscoveryError::OpenRouterMalformed(OpenRouterMalformedStage::Normalization)
})?;
accept_snapshot(
snapshot,
"openrouter",
CatalogSourceKind::OpenrouterCatalog,
receipt_for(&[response.body]),
models,
)
}
fn accept_snapshot(
snapshot: CatalogSnapshotRequest,
provider_id: &str,
source: CatalogSourceKind,
receipt: String,
models: Vec<NormalizedModelSpec>,
) -> Result<CatalogSnapshot, CatalogDiscoveryError> {
CatalogSnapshot::accept(
snapshot.snapshot_id,
provider_id,
snapshot.catalog_version,
source,
receipt,
snapshot.observed_at_epoch_seconds,
models,
)
.map_err(|_| CatalogDiscoveryError::SnapshotRejected)
}
fn catalog_request(
endpoint: CatalogEndpoint,
model: Option<&str>,
credential: Option<CatalogCredential<'_>>,
) -> Result<CatalogHttpRequest, CatalogDiscoveryError> {
let credential = match (endpoint.requires_authorization(), credential) {
(
true,
Some(CatalogCredential::Resolver {
resolver,
secret_ref,
}),
) => Some(
resolver
.resolve(secret_ref)
.map_err(|_: SecretResolutionError| CatalogDiscoveryError::SecretResolution)?,
),
(true, Some(CatalogCredential::Resolved(secret))) => Some(
ResolvedSecret::new(secret.expose_for_dispatch().to_vec())
.map_err(|_| CatalogDiscoveryError::SecretResolution)?,
),
(false, None) => None,
_ => return Err(CatalogDiscoveryError::InvalidRequest),
};
catalog_request_with_resolved_secret(endpoint, model, credential)
}
fn catalog_request_with_resolved_secret(
endpoint: CatalogEndpoint,
model: Option<&str>,
credential: Option<ResolvedSecret>,
) -> Result<CatalogHttpRequest, CatalogDiscoveryError> {
let body = if matches!(
endpoint,
CatalogEndpoint::OllamaCloudShow | CatalogEndpoint::LocalOllamaShow
) {
let model = model
.filter(|value| valid_model_id(value))
.ok_or(CatalogDiscoveryError::InvalidRequest)?;
serde_json::to_vec(&json!({"model": model}))
.map_err(|_| CatalogDiscoveryError::InvalidRequest)?
} else {
if model.is_some() {
return Err(CatalogDiscoveryError::InvalidRequest);
}
Vec::new()
};
let custom = matches!(
endpoint,
CatalogEndpoint::AnthropicModels | CatalogEndpoint::GeminiModels
);
let authorization = match (endpoint.requires_authorization(), credential) {
(true, Some(secret)) => Some(if custom {
secret
} else {
bearer_secret(secret)?
}),
(false, None) => None,
_ => return Err(CatalogDiscoveryError::InvalidRequest),
};
let auth_header_name = match endpoint {
CatalogEndpoint::AnthropicModels => Some(router_protocol::ANTHROPIC_API_KEY_HEADER),
CatalogEndpoint::GeminiModels => Some(router_protocol::GEMINI_API_KEY_HEADER),
_ => None,
};
let version_header = match endpoint {
CatalogEndpoint::AnthropicModels => Some((
router_protocol::ANTHROPIC_VERSION_HEADER,
router_protocol::ANTHROPIC_VERSION,
)),
_ => None,
};
Ok(CatalogHttpRequest {
endpoint,
body,
authorization,
auth_header_name,
version_header,
})
}
fn bearer_secret(secret: ResolvedSecret) -> Result<ResolvedSecret, CatalogDiscoveryError> {
let mut value = Vec::with_capacity(7usize.saturating_add(secret.expose_for_dispatch().len()));
value.extend_from_slice(b"Bearer ");
value.extend_from_slice(secret.expose_for_dispatch());
ResolvedSecret::new(value).map_err(|_| CatalogDiscoveryError::SecretResolution)
}
async fn execute_catalog(
executor: &dyn CatalogHttpExecutor,
request: CatalogHttpRequest,
) -> Result<CatalogHttpResponse, CatalogDiscoveryError> {
execute_with_absolute_deadline(executor, request, CATALOG_REQUEST_TIMEOUT).await
}
async fn execute_with_absolute_deadline(
executor: &dyn CatalogHttpExecutor,
request: CatalogHttpRequest,
deadline: Duration,
) -> Result<CatalogHttpResponse, CatalogDiscoveryError> {
let endpoint = request.endpoint;
let response = tokio::time::timeout(deadline, executor.execute(request))
.await
.map_err(|_| CatalogDiscoveryError::Timeout)?
.map_err(map_http_error)?;
if response.body.len() > endpoint.response_limit() {
return Err(CatalogDiscoveryError::ResponseBounds);
}
Ok(response)
}
fn map_http_error(error: CatalogHttpError) -> CatalogDiscoveryError {
match error.kind {
CatalogHttpErrorKind::Timeout => CatalogDiscoveryError::Timeout,
CatalogHttpErrorKind::ResponseTooLarge => CatalogDiscoveryError::ResponseBounds,
CatalogHttpErrorKind::Cancelled
| CatalogHttpErrorKind::Transport
| CatalogHttpErrorKind::InvalidRequest => CatalogDiscoveryError::Transport,
}
}
fn validate_catalog_request(request: &CatalogHttpRequest) -> Result<(), CatalogHttpError> {
let auth_present = request.authorization.is_some();
let body_valid = if request.method() == "POST" {
!request.body.is_empty() && request.body.len() <= MAX_REQUEST_BODY_BYTES
} else {
request.body.is_empty()
};
if !body_valid
|| request.endpoint.requires_authorization() != auth_present
|| (!request.endpoint.allows_authorization() && auth_present)
{
return Err(catalog_http_error(CatalogHttpErrorKind::InvalidRequest));
}
Ok(())
}
fn validate_success_response(
endpoint: CatalogEndpoint,
response: &CatalogHttpResponse,
) -> Result<(), CatalogDiscoveryError> {
if response.body.len() > endpoint.response_limit() {
return Err(CatalogDiscoveryError::ResponseBounds);
}
if response.redirected || response.status != 200 {
return Err(CatalogDiscoveryError::HttpResponse);
}
if !response.content_type.as_deref().is_some_and(|value| {
value
.split(';')
.next()
.is_some_and(|mime| mime.trim().eq_ignore_ascii_case("application/json"))
}) {
return Err(CatalogDiscoveryError::HttpResponse);
}
Ok(())
}
#[derive(Deserialize)]
struct OllamaTagModel {
#[serde(default)]
name: String,
#[serde(default)]
model: String,
#[serde(default)]
remote_model: Option<String>,
#[serde(default)]
remote_host: Option<String>,
}
fn decode_ollama_tags(bytes: &[u8]) -> Result<OllamaNativeCatalog, CatalogDiscoveryError> {
let root: Value =
serde_json::from_slice(bytes).map_err(|_| CatalogDiscoveryError::Malformed)?;
let values = root
.get("models")
.and_then(Value::as_array)
.ok_or(CatalogDiscoveryError::Malformed)?;
if values.is_empty() || values.len() > MAX_CATALOG_MODELS {
return Err(CatalogDiscoveryError::ResponseBounds);
}
let mut seen = HashSet::with_capacity(values.len());
let mut models = Vec::with_capacity(values.len());
for value in values {
let item: OllamaTagModel =
serde_json::from_value(value.clone()).map_err(|_| CatalogDiscoveryError::Malformed)?;
let id = if item.model.is_empty() {
item.name
} else {
item.model
};
if !valid_model_id(&id) || !seen.insert(id.clone()) {
return Err(CatalogDiscoveryError::Malformed);
}
let location = match (item.remote_model.as_deref(), item.remote_host.as_deref()) {
(Some(remote_model), Some(remote_host))
if cloud_base_model_id(&id) == Some(remote_model)
&& valid_ollama_remote_host(remote_host) =>
{
OllamaModelLocation::Cloud
}
(None, None) if cloud_base_model_id(&id).is_none() => OllamaModelLocation::Local,
_ => return Err(CatalogDiscoveryError::Malformed),
};
models.push(OllamaNativeModel {
id,
location,
context_tokens: 1,
modalities: vec!["text".to_owned()],
tool_support: false,
reasoning_modes: Vec::new(),
});
}
Ok(OllamaNativeCatalog { models })
}
fn apply_ollama_show(
model: &mut OllamaNativeModel,
bytes: &[u8],
) -> Result<(), CatalogDiscoveryError> {
let root: Value =
serde_json::from_slice(bytes).map_err(|_| CatalogDiscoveryError::Malformed)?;
let capabilities = root
.get("capabilities")
.and_then(Value::as_array)
.map(|values| {
values
.iter()
.filter_map(Value::as_str)
.collect::<BTreeSet<_>>()
})
.unwrap_or_default();
let model_info = root
.get("model_info")
.and_then(Value::as_object)
.ok_or(CatalogDiscoveryError::Malformed)?;
let context = context_length(model_info).ok_or(CatalogDiscoveryError::Malformed)?;
model.context_tokens = context;
model.tool_support = capabilities.contains("tools");
if capabilities.contains("vision") {
model.modalities.push("image".to_owned());
}
if capabilities.contains("thinking") {
model.reasoning_modes.push("reasoning".to_owned());
}
model.modalities.sort();
model.modalities.dedup();
model.reasoning_modes.sort();
model.reasoning_modes.dedup();
Ok(())
}
fn context_length(model_info: &Map<String, Value>) -> Option<u64> {
model_info
.iter()
.filter(|(key, _)| key.ends_with(".context_length"))
.filter_map(|(_, value)| value.as_u64())
.filter(|value| *value > 0)
.max()
}
fn decode_openrouter(
bytes: &[u8],
authenticated: bool,
) -> Result<OpenRouterNativeCatalog, CatalogDiscoveryError> {
decode_openrouter_selected(bytes, authenticated, None)
}
fn decode_openrouter_configured(
bytes: &[u8],
authenticated: bool,
configured: &HashSet<&str>,
) -> Result<OpenRouterNativeCatalog, CatalogDiscoveryError> {
decode_openrouter_selected(bytes, authenticated, Some(configured))
}
fn decode_openrouter_selected(
bytes: &[u8],
authenticated: bool,
configured: Option<&HashSet<&str>>,
) -> Result<OpenRouterNativeCatalog, CatalogDiscoveryError> {
let root: Value = serde_json::from_slice(bytes)
.map_err(|_| CatalogDiscoveryError::OpenRouterMalformed(OpenRouterMalformedStage::Json))?;
let values = root.get("data").and_then(Value::as_array).ok_or(
CatalogDiscoveryError::OpenRouterMalformed(OpenRouterMalformedStage::DataEnvelope),
)?;
if values.is_empty() || values.len() > MAX_CATALOG_MODELS {
return Err(CatalogDiscoveryError::ResponseBounds);
}
let mut models = Vec::with_capacity(values.len());
let mut seen = HashSet::with_capacity(values.len());
for value in values {
let Some(object) = value.as_object() else {
if configured.is_some() {
continue;
}
return Err(CatalogDiscoveryError::OpenRouterMalformed(
OpenRouterMalformedStage::ModelObject,
));
};
let Some(id) = object.get("id").and_then(Value::as_str) else {
if configured.is_some() {
continue;
}
return Err(CatalogDiscoveryError::OpenRouterMalformed(
OpenRouterMalformedStage::ModelId,
));
};
if configured.is_some_and(|wanted| !wanted.contains(id)) {
continue;
}
if !valid_model_id(id) {
return Err(CatalogDiscoveryError::OpenRouterMalformed(
OpenRouterMalformedStage::ModelId,
));
}
let context_tokens = object
.get("context_length")
.and_then(Value::as_u64)
.filter(|value| *value > 0)
.ok_or(CatalogDiscoveryError::OpenRouterMalformed(
OpenRouterMalformedStage::ContextLength,
))?;
if !seen.insert(id.to_owned()) {
return Err(CatalogDiscoveryError::OpenRouterMalformed(
OpenRouterMalformedStage::DuplicateModelId,
));
}
let supported = object
.get("supported_parameters")
.and_then(Value::as_array)
.map(|values| {
values
.iter()
.filter_map(Value::as_str)
.collect::<BTreeSet<_>>()
})
.unwrap_or_default();
let mut modalities = object
.get("architecture")
.and_then(Value::as_object)
.and_then(|architecture| architecture.get("input_modalities"))
.and_then(Value::as_array)
.map(|values| {
values
.iter()
.filter_map(Value::as_str)
.filter(|value| valid_token(value))
.map(str::to_owned)
.collect::<Vec<_>>()
})
.unwrap_or_else(|| vec!["text".to_owned()]);
modalities.sort();
modalities.dedup();
if modalities.is_empty() {
return Err(CatalogDiscoveryError::OpenRouterMalformed(
OpenRouterMalformedStage::Modalities,
));
}
let mut reasoning_modes = Vec::new();
if supported.contains("reasoning") || supported.contains("include_reasoning") {
reasoning_modes.push("reasoning".to_owned());
}
models.push(OpenRouterNativeModel {
id: id.to_owned(),
context_tokens,
modalities,
tool_support: supported.contains("tools"),
reasoning_modes,
});
}
if configured.is_some_and(|wanted| seen.len() != wanted.len()) {
return Err(CatalogDiscoveryError::OpenRouterMalformed(
OpenRouterMalformedStage::MissingConfiguredModel,
));
}
models.sort_by(|left, right| left.id.cmp(&right.id));
Ok(OpenRouterNativeCatalog {
models,
authenticated,
})
}
fn receipt_for(bodies: &[Vec<u8>]) -> String {
let mut digest = Sha256::new();
for body in bodies {
digest.update((body.len() as u64).to_be_bytes());
digest.update(body);
}
let digest = digest.finalize();
let encoded = digest
.iter()
.map(|byte| format!("{byte:02x}"))
.collect::<String>();
format!("receipt:sha256:{encoded}")
}
fn valid_model_id(value: &str) -> bool {
!value.is_empty()
&& value.len() <= 256
&& !value.starts_with('/')
&& !value.ends_with('/')
&& !value.contains("..")
&& !value.contains("//")
&& value.bytes().all(|byte| {
byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.' | b':' | b'/')
})
}
fn cloud_base_model_id(value: &str) -> Option<&str> {
let base = value
.strip_suffix(":cloud")
.or_else(|| value.strip_suffix("-cloud"))?;
(!base.is_empty()
&& valid_model_id(base)
&& !base.ends_with(":cloud")
&& !base.ends_with("-cloud"))
.then_some(base)
}
fn valid_ollama_remote_host(value: &str) -> bool {
matches!(value, "https://ollama.com" | "https://ollama.com:443")
}
fn valid_token(value: &str) -> bool {
!value.is_empty()
&& value.len() <= 64
&& value
.bytes()
.all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_'))
}
const fn catalog_http_error(kind: CatalogHttpErrorKind) -> CatalogHttpError {
CatalogHttpError { kind }
}
#[cfg(test)]
mod tests {
use super::*;
use std::collections::VecDeque;
use std::sync::{Arc, Mutex};
#[derive(Clone, Debug, PartialEq, Eq)]
struct Observation {
endpoint: CatalogEndpoint,
authorized: bool,
body: Vec<u8>,
auth_header_name: Option<&'static str>,
version_header: Option<(&'static str, &'static str)>,
debug: String,
}
struct FakeExecutor {
responses: Mutex<VecDeque<Result<CatalogHttpResponse, CatalogHttpError>>>,
observations: Arc<Mutex<Vec<Observation>>>,
}
impl CatalogHttpExecutor for FakeExecutor {
fn execute<'a>(&'a self, request: CatalogHttpRequest) -> CatalogHttpFuture<'a> {
let response = self
.responses
.lock()
.expect("response queue")
.pop_front()
.expect("one response per request");
self.observations
.lock()
.expect("observations")
.push(Observation {
endpoint: request.endpoint(),
authorized: request.authorization_for_dispatch().is_some(),
body: request.body_for_dispatch().to_vec(),
auth_header_name: request.auth_header_name,
version_header: request.version_header,
debug: format!("{request:?}"),
});
Box::pin(async move { response })
}
}
struct NeverCompletesExecutor;
impl CatalogHttpExecutor for NeverCompletesExecutor {
fn execute<'a>(&'a self, _request: CatalogHttpRequest) -> CatalogHttpFuture<'a> {
Box::pin(async move {
tokio::time::sleep(CATALOG_REQUEST_TIMEOUT.saturating_mul(100)).await;
unreachable!("the absolute catalog deadline must cancel this executor")
})
}
}
#[derive(Default)]
struct FixtureResolver;
impl SecretResolver for FixtureResolver {
fn resolve(&self, _secret_ref: &str) -> Result<ResolvedSecret, SecretResolutionError> {
ResolvedSecret::new(b"catalog-secret".to_vec())
}
}
fn response(body: Value) -> Result<CatalogHttpResponse, CatalogHttpError> {
Ok(CatalogHttpResponse {
status: 200,
content_type: Some("application/json; charset=utf-8".to_owned()),
redirected: false,
body: serde_json::to_vec(&body).expect("fixture JSON"),
})
}
fn snapshot(id: &str) -> CatalogSnapshotRequest {
CatalogSnapshotRequest {
snapshot_id: id.to_owned(),
catalog_version: "catalog.1".to_owned(),
observed_at_epoch_seconds: 100,
}
}
#[tokio::test]
async fn cloud_tags_and_bounded_show_normalize_deterministically_without_secret_output() {
let observations = Arc::new(Mutex::new(Vec::new()));
let executor = FakeExecutor {
responses: Mutex::new(VecDeque::from([
response(json!({"models": [
{"model": "zeta:cloud", "remote_model": "zeta", "remote_host": "https://ollama.com"},
{"model": "alpha:cloud", "remote_model": "alpha", "remote_host": "https://ollama.com:443"}
]})),
response(
json!({"capabilities": ["tools", "thinking", "vision"], "model_info": {"fixture.context_length": 65536}}),
),
])),
observations: Arc::clone(&observations),
};
let first = discover_ollama_cloud(
&executor,
&FixtureResolver,
"secret:fixture",
snapshot("snapshot.cloud.1"),
OllamaDiscoveryOptions {
max_show_enrichments: 1,
},
)
.await
.expect("cloud catalog");
assert_eq!(
first.model("alpha:cloud").expect("alpha").context_tokens,
65_536
);
assert_eq!(first.model("zeta:cloud").expect("zeta").context_tokens, 1);
assert!(first.model("alpha:cloud").expect("alpha").tool_support);
let observed = observations.lock().expect("observations");
assert_eq!(observed.len(), 2);
assert!(observed.iter().all(|item| item.authorized));
assert_eq!(observed[0].endpoint, CatalogEndpoint::OllamaCloudTags);
assert_eq!(observed[1].endpoint, CatalogEndpoint::OllamaCloudShow);
assert_eq!(observed[1].body, br#"{"model":"alpha:cloud"}"#);
assert!(
observed
.iter()
.all(|item| !item.debug.contains("catalog-secret"))
);
}
#[tokio::test]
async fn local_catalog_never_authenticates_and_never_requests_more_than_show_bound() {
let observations = Arc::new(Mutex::new(Vec::new()));
let executor = FakeExecutor {
responses: Mutex::new(VecDeque::from([response(json!({
"models": [
{"name": "qwen3-coder:480b-cloud", "remote_model": "qwen3-coder:480b", "remote_host": "https://ollama.com:443"},
{"name": "mistral:latest"}
]
}))])),
observations: Arc::clone(&observations),
};
let catalog = discover_local_ollama(
&executor,
snapshot("snapshot.local.1"),
OllamaDiscoveryOptions::default(),
)
.await
.expect("local catalog");
assert!(catalog.model("mistral:latest").is_some());
assert!(catalog.model("qwen3-coder:480b-cloud").is_none());
assert!(
observations
.lock()
.expect("observations")
.iter()
.all(|item| !item.authorized)
);
assert_eq!(
discover_local_ollama(
&executor,
snapshot("snapshot.local.2"),
OllamaDiscoveryOptions {
max_show_enrichments: MAX_SHOW_ENRICHMENTS + 1,
},
)
.await,
Err(CatalogDiscoveryError::InvalidRequest)
);
}
#[tokio::test]
async fn deepseek_and_qwen_model_lists_authenticate_and_normalize_conservatively() {
let observations = Arc::new(Mutex::new(Vec::new()));
let executor = FakeExecutor {
responses: Mutex::new(VecDeque::from([
response(json!({"data": [
{"id": "deepseek-v4-flash", "object": "model"},
{"id": "deepseek-v4-pro", "object": "model"}
]})),
response(json!({"data": [
{"id": "qwen3.6-flash", "object": "model"},
{"id": "qwen3.7-plus", "object": "model"},
{"id": "qwen3.8-max", "object": "model"}
]})),
])),
observations: Arc::clone(&observations),
};
let deepseek = discover_deepseek(
&executor,
&FixtureResolver,
"secret:fixture",
snapshot("snapshot.deepseek.1"),
)
.await
.expect("deepseek catalog");
assert_eq!(deepseek.models().len(), 2);
let flash = deepseek.model("deepseek-v4-flash").expect("flash");
assert_eq!(flash.context_tokens, 1);
assert!(!flash.tool_support);
assert_eq!(
serde_json::to_value(&deepseek)
.expect("snapshot JSON")
.get("source")
.and_then(Value::as_str),
Some("deepseek_catalog")
);
let qwen = discover_qwen(
&executor,
&FixtureResolver,
"secret:fixture",
snapshot("snapshot.qwen.1"),
)
.await
.expect("qwen catalog");
assert_eq!(qwen.models().len(), 3);
assert_eq!(
serde_json::to_value(&qwen)
.expect("snapshot JSON")
.get("source")
.and_then(Value::as_str),
Some("qwen_catalog")
);
let observed = observations.lock().expect("observations");
assert_eq!(observed.len(), 2);
assert!(observed.iter().all(|item| item.authorized));
assert_eq!(observed[0].endpoint, CatalogEndpoint::DeepseekModels);
assert_eq!(observed[1].endpoint, CatalogEndpoint::QwenModels);
assert!(
observed
.iter()
.all(|item| !item.debug.contains("catalog-secret"))
);
}
#[tokio::test]
async fn openai_compat_model_lists_fail_closed_on_missing_data_and_duplicates() {
let executor = FakeExecutor {
responses: Mutex::new(VecDeque::from([response(json!({"models": []}))])),
observations: Arc::new(Mutex::new(Vec::new())),
};
assert!(matches!(
discover_deepseek(
&executor,
&FixtureResolver,
"secret:fixture",
snapshot("snapshot.deepseek.2"),
)
.await,
Err(CatalogDiscoveryError::Malformed)
));
let executor = FakeExecutor {
responses: Mutex::new(VecDeque::from([response(json!({"data": [
{"id": "qwen3.6-flash", "object": "model"},
{"id": "qwen3.6-flash", "object": "model"}
]}))])),
observations: Arc::new(Mutex::new(Vec::new())),
};
assert!(matches!(
discover_qwen(
&executor,
&FixtureResolver,
"secret:fixture",
snapshot("snapshot.qwen.2"),
)
.await,
Err(CatalogDiscoveryError::Malformed)
));
}
#[tokio::test]
async fn grok_kimi_and_zai_model_lists_authenticate_and_normalize_conservatively() {
let observations = Arc::new(Mutex::new(Vec::new()));
let executor = FakeExecutor {
responses: Mutex::new(VecDeque::from([
response(json!({"data": [
{"id": "grok-4.5", "object": "model"},
{"id": "grok-4.6", "object": "model"}
]})),
response(json!({"data": [
{"id": "kimi-k2.7-code", "object": "model"},
{"id": "kimi-k3", "object": "model"}
]})),
response(json!({"data": [
{"id": "glm-4.7-flash", "object": "model"},
{"id": "glm-5.3", "object": "model"}
]})),
])),
observations: Arc::clone(&observations),
};
let grok = discover_xai(
&executor,
&FixtureResolver,
"secret:fixture",
snapshot("snapshot.grok.1"),
)
.await
.expect("grok catalog");
assert_eq!(grok.models().len(), 2);
assert_eq!(grok.provider_id(), "grok");
let flash = grok.model("grok-4.5").expect("grok-4.5");
assert_eq!(flash.context_tokens, 1);
assert!(!flash.tool_support);
assert_eq!(
serde_json::to_value(&grok)
.expect("snapshot JSON")
.get("source")
.and_then(Value::as_str),
Some("grok_catalog")
);
let kimi = discover_moonshot(
&executor,
&FixtureResolver,
"secret:fixture",
snapshot("snapshot.kimi.1"),
)
.await
.expect("kimi catalog");
assert_eq!(kimi.models().len(), 2);
assert_eq!(kimi.provider_id(), "kimi");
assert_eq!(
serde_json::to_value(&kimi)
.expect("snapshot JSON")
.get("source")
.and_then(Value::as_str),
Some("kimi_catalog")
);
let zai = discover_zai(
&executor,
&FixtureResolver,
"secret:fixture",
snapshot("snapshot.zai.1"),
)
.await
.expect("zai catalog");
assert_eq!(zai.models().len(), 2);
assert_eq!(zai.provider_id(), "zai");
assert_eq!(
serde_json::to_value(&zai)
.expect("snapshot JSON")
.get("source")
.and_then(Value::as_str),
Some("zai_catalog")
);
let observed = observations.lock().expect("observations");
assert_eq!(observed.len(), 3);
assert!(observed.iter().all(|item| item.authorized));
assert_eq!(observed[0].endpoint, CatalogEndpoint::XaiModels);
assert_eq!(observed[1].endpoint, CatalogEndpoint::MoonshotModels);
assert_eq!(observed[2].endpoint, CatalogEndpoint::ZaiModels);
assert!(
observed
.iter()
.all(|item| !item.debug.contains("catalog-secret"))
);
}
#[tokio::test]
async fn grok_kimi_and_zai_model_lists_fail_closed_on_missing_data_and_duplicates() {
let executor = FakeExecutor {
responses: Mutex::new(VecDeque::from([response(json!({"models": []}))])),
observations: Arc::new(Mutex::new(Vec::new())),
};
assert!(matches!(
discover_xai(
&executor,
&FixtureResolver,
"secret:fixture",
snapshot("snapshot.grok.2"),
)
.await,
Err(CatalogDiscoveryError::Malformed)
));
let executor = FakeExecutor {
responses: Mutex::new(VecDeque::from([response(json!({"data": [
{"id": "kimi-k3", "object": "model"},
{"id": "kimi-k3", "object": "model"}
]}))])),
observations: Arc::new(Mutex::new(Vec::new())),
};
assert!(matches!(
discover_moonshot(
&executor,
&FixtureResolver,
"secret:fixture",
snapshot("snapshot.kimi.2"),
)
.await,
Err(CatalogDiscoveryError::Malformed)
));
let executor = FakeExecutor {
responses: Mutex::new(VecDeque::from([response(json!({"data": [
{"id": "glm-5.3", "object": "model"},
{"id": "glm-5.3", "object": "model"}
]}))])),
observations: Arc::new(Mutex::new(Vec::new())),
};
assert!(matches!(
discover_zai(
&executor,
&FixtureResolver,
"secret:fixture",
snapshot("snapshot.zai.2"),
)
.await,
Err(CatalogDiscoveryError::Malformed)
));
}
#[tokio::test]
async fn claude_and_gemini_model_lists_use_native_api_key_headers_and_strip_gemini_prefix() {
let observations = Arc::new(Mutex::new(Vec::new()));
let executor = FakeExecutor {
responses: Mutex::new(VecDeque::from([
response(json!({"data": [
{"id": "claude-opus-5", "type": "model"},
{"id": "claude-sonnet-5", "type": "model"}
]})),
response(json!({"models": [
{"name": "models/gemini-3.6-flash", "display_name": "Gemini 3.6 Flash"},
{"name": "models/gemini-3.5-flash-lite", "display_name": "Gemini 3.5 Flash Lite"}
]})),
])),
observations: Arc::clone(&observations),
};
let claude = discover_anthropic(
&executor,
&FixtureResolver,
"secret:fixture",
snapshot("snapshot.claude.1"),
)
.await
.expect("claude catalog");
assert_eq!(claude.models().len(), 2);
assert_eq!(claude.provider_id(), "claude");
assert_eq!(
claude
.model("claude-sonnet-5")
.expect("claude-sonnet-5")
.context_tokens,
1
);
assert_eq!(
serde_json::to_value(&claude)
.expect("snapshot JSON")
.get("source")
.and_then(Value::as_str),
Some("claude_catalog")
);
let gemini = discover_gemini(
&executor,
&FixtureResolver,
"secret:fixture",
snapshot("snapshot.gemini.1"),
)
.await
.expect("gemini catalog");
assert_eq!(gemini.models().len(), 2);
assert_eq!(gemini.provider_id(), "gemini");
assert_eq!(
gemini
.model("gemini-3.6-flash")
.expect("gemini-3.6-flash")
.context_tokens,
1
);
assert!(
gemini.model("models/gemini-3.6-flash").is_none(),
"the models/ prefix must be stripped from the catalog id"
);
assert_eq!(
serde_json::to_value(&gemini)
.expect("snapshot JSON")
.get("source")
.and_then(Value::as_str),
Some("gemini_catalog")
);
let observed = observations.lock().expect("observations");
assert_eq!(observed.len(), 2);
assert!(observed.iter().all(|item| item.authorized));
assert_eq!(observed[0].endpoint, CatalogEndpoint::AnthropicModels);
assert_eq!(
observed[0].auth_header_name,
Some(router_protocol::ANTHROPIC_API_KEY_HEADER)
);
assert_eq!(
observed[0].version_header,
Some((
router_protocol::ANTHROPIC_VERSION_HEADER,
router_protocol::ANTHROPIC_VERSION
))
);
assert_eq!(observed[1].endpoint, CatalogEndpoint::GeminiModels);
assert_eq!(
observed[1].auth_header_name,
Some(router_protocol::GEMINI_API_KEY_HEADER)
);
assert_eq!(observed[1].version_header, None);
assert!(
observed
.iter()
.all(|item| !item.debug.contains("catalog-secret"))
);
}
#[tokio::test]
async fn claude_and_gemini_model_lists_fail_closed_on_duplicates_and_bad_prefix() {
let executor = FakeExecutor {
responses: Mutex::new(VecDeque::from([response(json!({"data": [
{"id": "claude-opus-5", "type": "model"},
{"id": "claude-opus-5", "type": "model"}
]}))])),
observations: Arc::new(Mutex::new(Vec::new())),
};
assert!(matches!(
discover_anthropic(
&executor,
&FixtureResolver,
"secret:fixture",
snapshot("snapshot.claude.2"),
)
.await,
Err(CatalogDiscoveryError::Malformed)
));
let executor = FakeExecutor {
responses: Mutex::new(VecDeque::from([response(json!({"models": [
{"name": "gemini-3.6-flash", "display_name": "Gemini"}
]}))])),
observations: Arc::new(Mutex::new(Vec::new())),
};
assert!(matches!(
discover_gemini(
&executor,
&FixtureResolver,
"secret:fixture",
snapshot("snapshot.gemini.2"),
)
.await,
Err(CatalogDiscoveryError::Malformed)
));
}
#[tokio::test]
async fn signed_in_local_relay_partitions_actual_cloud_alias_shape_without_api_key() {
let observations = Arc::new(Mutex::new(Vec::new()));
let executor = FakeExecutor {
responses: Mutex::new(VecDeque::from([response(json!({
"models": [
{"name": "qwen3-coder:480b-cloud", "remote_model": "qwen3-coder:480b", "remote_host": "https://ollama.com:443", "size": 382},
{"name": "deepseek-v4-flash:cloud", "remote_model": "deepseek-v4-flash", "remote_host": "https://ollama.com:443", "size": 382},
{"name": "deepseek-v4-pro:cloud", "remote_model": "deepseek-v4-pro", "remote_host": "https://ollama.com", "size": 382},
{"name": "qwen3-coder-next:cloud", "remote_model": "qwen3-coder-next", "remote_host": "https://ollama.com:443", "size": 382},
{"name": "mistral:latest", "size": 4_372_824_384u64}
]
}))])),
observations: Arc::clone(&observations),
};
let catalog = discover_ollama_cloud_relay(
&executor,
snapshot("snapshot.cloud.relay.1"),
OllamaDiscoveryOptions::default(),
)
.await
.expect("signed-in relay cloud catalog");
assert!(catalog.model("qwen3-coder:480b-cloud").is_some());
assert!(catalog.model("deepseek-v4-flash:cloud").is_some());
assert!(catalog.model("mistral:latest").is_none());
assert_eq!(
serde_json::to_value(&catalog)
.expect("snapshot JSON")
.get("source")
.and_then(Value::as_str),
Some("ollama_cloud_catalog")
);
let observed = observations.lock().expect("observations");
assert_eq!(observed.len(), 1);
assert_eq!(observed[0].endpoint, CatalogEndpoint::LocalOllamaTags);
assert!(!observed[0].authorized);
}
#[tokio::test]
async fn configured_unlisted_cloud_model_is_bounded_and_probed_only_through_relay() {
let observations = Arc::new(Mutex::new(Vec::new()));
let executor = FakeExecutor {
responses: Mutex::new(VecDeque::from([
response(json!({"models": [{"name": "mistral:latest"}]})),
response(json!({
"capabilities": ["thinking"],
"model_info": {"fixture.context_length": 32768}
})),
])),
observations: Arc::clone(&observations),
};
let catalog = discover_ollama_cloud_relay_with_configured_models(
&executor,
snapshot("snapshot.cloud.relay.configured.1"),
OllamaDiscoveryOptions::default(),
&["glm-5.2:cloud".to_owned()],
)
.await
.expect("configured cloud probe");
let model = catalog.model("glm-5.2:cloud").expect("probed model");
assert_eq!(model.context_tokens, 32_768);
assert_eq!(model.reasoning_modes, ["reasoning"]);
let observed = observations.lock().expect("observations");
assert_eq!(observed.len(), 2);
assert_eq!(observed[0].endpoint, CatalogEndpoint::LocalOllamaTags);
assert_eq!(observed[1].endpoint, CatalogEndpoint::LocalOllamaShow);
assert_eq!(observed[1].body, br#"{"model":"glm-5.2:cloud"}"#);
assert!(observed.iter().all(|item| !item.authorized));
}
#[tokio::test]
async fn configured_cloud_probe_rejects_transport_identity_and_request_count_confusion() {
let executor = FakeExecutor {
responses: Mutex::new(VecDeque::new()),
observations: Arc::new(Mutex::new(Vec::new())),
};
assert_eq!(
discover_ollama_cloud_relay_with_configured_models(
&executor,
snapshot("snapshot.cloud.relay.invalid.1"),
OllamaDiscoveryOptions::default(),
&["glm-5.2".to_owned()],
)
.await,
Err(CatalogDiscoveryError::InvalidRequest)
);
assert_eq!(
discover_ollama_cloud_relay_with_configured_models(
&executor,
snapshot("snapshot.cloud.relay.invalid.2"),
OllamaDiscoveryOptions {
max_show_enrichments: MAX_SHOW_ENRICHMENTS,
},
&["glm-5.2:cloud".to_owned()],
)
.await,
Err(CatalogDiscoveryError::InvalidRequest)
);
}
#[test]
fn cloud_markers_reject_host_and_suffix_confusion() {
let hostile = [
("model:cloud", "model", "http://ollama.com"),
("model:cloud", "model", "https://ollama.com:444"),
("model:cloud", "model", "https://ollama.com/path"),
("model:cloud", "model", "https://ollama.com?query=1"),
("model:cloud", "model", "https://ollama.com#fragment"),
("model:cloud", "model", "https://user@ollama.com"),
("model:cloud", "model", "https://ollama.com.evil"),
("model:cloud", "other", "https://ollama.com"),
("model-cloud", "model:cloud", "https://ollama.com"),
("model", "model", "https://ollama.com"),
("model:cloud:cloud", "model:cloud", "https://ollama.com"),
];
for (id, remote_model, remote_host) in hostile {
let bytes = serde_json::to_vec(&json!({
"models": [{
"name": id,
"remote_model": remote_model,
"remote_host": remote_host
}]
}))
.expect("hostile fixture JSON");
assert_eq!(
decode_ollama_tags(&bytes),
Err(CatalogDiscoveryError::Malformed),
"accepted confused cloud identity for {id} via {remote_host}"
);
}
}
#[tokio::test]
async fn openrouter_user_unauthorized_falls_back_without_forwarding_auth() {
let observations = Arc::new(Mutex::new(Vec::new()));
let executor = FakeExecutor {
responses: Mutex::new(VecDeque::from([
Ok(CatalogHttpResponse {
status: 401,
content_type: Some("application/json".to_owned()),
redirected: false,
body: b"{}".to_vec(),
}),
response(json!({"data": [
{"id": "vendor/z-model", "context_length": 8192, "supported_parameters": []},
{"id": "vendor/a-model", "context_length": 32768, "supported_parameters": ["tools", "reasoning"], "architecture": {"input_modalities": ["text", "image"]}}
]})),
])),
observations: Arc::clone(&observations),
};
let catalog = discover_openrouter(
&executor,
Some((&FixtureResolver, "secret:fixture")),
snapshot("snapshot.openrouter.1"),
)
.await
.expect("public fallback catalog");
let selected = catalog.model("vendor/a-model").expect("slash model ID");
assert!(selected.tool_support);
assert_eq!(selected.reasoning_modes, ["reasoning"]);
let observed = observations.lock().expect("observations");
assert_eq!(observed[0].endpoint, CatalogEndpoint::OpenRouterUserModels);
assert!(observed[0].authorized);
assert_eq!(
observed[1].endpoint,
CatalogEndpoint::OpenRouterPublicModels
);
assert!(!observed[1].authorized);
}
#[tokio::test]
async fn strict_authenticated_openrouter_discovery_never_falls_back_public() {
let observations = Arc::new(Mutex::new(Vec::new()));
let executor = FakeExecutor {
responses: Mutex::new(VecDeque::from([Ok(CatalogHttpResponse {
status: 401,
content_type: Some("application/json".to_owned()),
redirected: false,
body: b"{}".to_vec(),
})])),
observations: Arc::clone(&observations),
};
assert_eq!(
discover_openrouter_authenticated(
&executor,
&FixtureResolver,
"secret:fixture",
snapshot("snapshot.openrouter.strict.1"),
)
.await,
Err(CatalogDiscoveryError::HttpResponse)
);
let observed = observations.lock().expect("observations");
assert_eq!(observed.len(), 1);
assert_eq!(observed[0].endpoint, CatalogEndpoint::OpenRouterUserModels);
assert!(observed[0].authorized);
}
#[tokio::test]
async fn openrouter_public_catalog_accepts_observed_367_model_inventory() {
let models = (0..367)
.map(|index| {
json!({
"id": format!("vendor/model-{index}"),
"context_length": 8_192,
"supported_parameters": []
})
})
.collect::<Vec<_>>();
let executor = FakeExecutor {
responses: Mutex::new(VecDeque::from([response(json!({"data": models}))])),
observations: Arc::new(Mutex::new(Vec::new())),
};
let catalog =
discover_openrouter(&executor, None, snapshot("snapshot.openrouter.public.367"))
.await
.expect("observed public inventory fits the bounded snapshot");
assert!(catalog.model("vendor/model-0").is_some());
assert!(catalog.model("vendor/model-366").is_some());
}
#[test]
fn openrouter_malformed_predicates_are_content_free_and_stable() {
let cases = [
(
b"not-json".as_slice(),
OpenRouterMalformedStage::Json,
),
(br#"{}"#, OpenRouterMalformedStage::DataEnvelope),
(
br#"{"data":[null]}"#,
OpenRouterMalformedStage::ModelObject,
),
(
br#"{"data":[{"id":"model-id-canary?","context_length":8192}]}"#,
OpenRouterMalformedStage::ModelId,
),
(
br#"{"data":[{"id":"vendor/model","context_length":0}]}"#,
OpenRouterMalformedStage::ContextLength,
),
(
br#"{"data":[{"id":"vendor/model","context_length":8192},{"id":"vendor/model","context_length":8192}]}"#,
OpenRouterMalformedStage::DuplicateModelId,
),
(
br#"{"data":[{"id":"vendor/model","context_length":8192,"architecture":{"input_modalities":["modalities-canary!"]}}]}"#,
OpenRouterMalformedStage::Modalities,
),
];
for (body, expected_stage) in cases {
let error = decode_openrouter(body, true).expect_err("malformed fixture");
assert_eq!(
error,
CatalogDiscoveryError::OpenRouterMalformed(expected_stage)
);
let diagnostic = format!("{error:?} {error}");
assert!(!diagnostic.contains("model-id-canary"));
assert!(!diagnostic.contains("modalities-canary"));
assert!(!diagnostic.contains("not-json"));
}
}
#[test]
fn openrouter_normalization_diagnostic_is_content_free() {
let error =
CatalogDiscoveryError::OpenRouterMalformed(OpenRouterMalformedStage::Normalization);
assert_eq!(format!("{error:?}"), "OpenRouterMalformed(Normalization)");
assert_eq!(
error.to_string(),
"OpenRouter catalog discovery was malformed"
);
}
#[tokio::test]
async fn openrouter_authenticated_catalog_accepts_observed_340_model_inventory() {
let models = (0..340)
.map(|index| {
json!({
"id": format!("vendor/model-{index}"),
"context_length": 8_192,
"supported_parameters": []
})
})
.collect::<Vec<_>>();
let executor = FakeExecutor {
responses: Mutex::new(VecDeque::from([response(json!({"data": models}))])),
observations: Arc::new(Mutex::new(Vec::new())),
};
let catalog = discover_openrouter_authenticated(
&executor,
&FixtureResolver,
"secret:fixture",
snapshot("snapshot.openrouter.authenticated.340"),
)
.await
.expect("observed authenticated inventory fits the bounded snapshot");
assert_eq!(catalog.models().len(), 340);
}
#[tokio::test]
async fn configured_openrouter_ignores_malformed_unconfigured_rows_only() {
let executor = FakeExecutor {
responses: Mutex::new(VecDeque::from([response(json!({"data": [
null,
{"context_length": 8192},
{"id": "unconfigured?invalid", "context_length": 8192},
{"id": "vendor/unconfigured-missing-context"},
{"id": "vendor/unconfigured-modalities", "context_length": 8192, "architecture": {"input_modalities": ["invalid!"]}},
{"id": "approved/standard", "context_length": 32768, "supported_parameters": ["tools"]},
{"id": "approved/complex", "context_length": 65536, "supported_parameters": ["reasoning"]}
]}))])),
observations: Arc::new(Mutex::new(Vec::new())),
};
let configured = vec![
"approved/standard".to_owned(),
"approved/complex".to_owned(),
];
let catalog = discover_openrouter_authenticated_with_configured_models(
&executor,
&FixtureResolver,
"secret:fixture",
snapshot("snapshot.openrouter.configured.1"),
&configured,
)
.await
.expect("unconfigured provider rows cannot poison configured discovery");
assert_eq!(catalog.models().len(), 2);
assert!(catalog.model("approved/standard").is_some());
assert!(catalog.model("approved/complex").is_some());
assert!(
catalog
.model("vendor/unconfigured-missing-context")
.is_none()
);
}
#[test]
fn configured_openrouter_requires_every_configured_row_once_and_strictly_valid() {
let configured = HashSet::from(["approved/standard", "approved/complex"]);
let cases = [
(
json!({"data": [
{"id": "approved/standard", "context_length": 8192}
]}),
OpenRouterMalformedStage::MissingConfiguredModel,
),
(
json!({"data": [
{"id": "approved/standard", "context_length": 8192},
{"id": "approved/standard", "context_length": 8192},
{"id": "approved/complex", "context_length": 8192}
]}),
OpenRouterMalformedStage::DuplicateModelId,
),
(
json!({"data": [
{"id": "approved/standard", "context_length": 0},
{"id": "approved/complex", "context_length": 8192}
]}),
OpenRouterMalformedStage::ContextLength,
),
(
json!({"data": [
{"id": "approved/standard", "context_length": 8192, "architecture": {"input_modalities": ["invalid!"]}},
{"id": "approved/complex", "context_length": 8192}
]}),
OpenRouterMalformedStage::Modalities,
),
];
for (body, expected) in cases {
let bytes = serde_json::to_vec(&body).expect("fixture JSON");
assert_eq!(
decode_openrouter_configured(&bytes, true, &configured),
Err(CatalogDiscoveryError::OpenRouterMalformed(expected))
);
}
}
#[tokio::test]
async fn configured_openrouter_rejects_invalid_requested_model_set_before_http() {
let observations = Arc::new(Mutex::new(Vec::new()));
let executor = FakeExecutor {
responses: Mutex::new(VecDeque::new()),
observations: Arc::clone(&observations),
};
for configured in [
Vec::<String>::new(),
vec!["invalid?model".to_owned()],
vec!["approved/model".to_owned(), "approved/model".to_owned()],
] {
assert_eq!(
discover_openrouter_authenticated_with_configured_models(
&executor,
&FixtureResolver,
"secret:fixture",
snapshot("snapshot.openrouter.configured.invalid"),
&configured,
)
.await,
Err(CatalogDiscoveryError::InvalidRequest)
);
}
assert!(observations.lock().expect("observations").is_empty());
}
#[tokio::test]
async fn configured_openrouter_keeps_full_response_model_bound() {
let models = (0..=MAX_CATALOG_MODELS)
.map(|index| {
json!({
"id": format!("unconfigured/model-{index}"),
"context_length": 8192
})
})
.collect::<Vec<_>>();
let executor = FakeExecutor {
responses: Mutex::new(VecDeque::from([response(json!({"data": models}))])),
observations: Arc::new(Mutex::new(Vec::new())),
};
assert_eq!(
discover_openrouter_authenticated_with_configured_models(
&executor,
&FixtureResolver,
"secret:fixture",
snapshot("snapshot.openrouter.configured.bounds"),
&["approved/model".to_owned()],
)
.await,
Err(CatalogDiscoveryError::ResponseBounds)
);
}
#[tokio::test]
async fn response_and_model_count_bounds_fail_before_typed_normalization() {
let oversized = FakeExecutor {
responses: Mutex::new(VecDeque::from([Ok(CatalogHttpResponse {
status: 200,
content_type: Some("application/json".to_owned()),
redirected: false,
body: vec![b' '; MAX_CATALOG_RESPONSE_BYTES + 1],
})])),
observations: Arc::new(Mutex::new(Vec::new())),
};
assert_eq!(
discover_local_ollama(
&oversized,
snapshot("snapshot.bounds.1"),
OllamaDiscoveryOptions::default(),
)
.await,
Err(CatalogDiscoveryError::ResponseBounds)
);
let many = (0..=MAX_CATALOG_MODELS)
.map(|index| json!({"model": format!("model-{index}")}))
.collect::<Vec<_>>();
let executor = FakeExecutor {
responses: Mutex::new(VecDeque::from([response(json!({"models": many}))])),
observations: Arc::new(Mutex::new(Vec::new())),
};
assert_eq!(
discover_local_ollama(
&executor,
snapshot("snapshot.bounds.2"),
OllamaDiscoveryOptions::default(),
)
.await,
Err(CatalogDiscoveryError::ResponseBounds)
);
let oversized_openrouter = FakeExecutor {
responses: Mutex::new(VecDeque::from([Ok(CatalogHttpResponse {
status: 200,
content_type: Some("application/json".to_owned()),
redirected: false,
body: vec![b' '; MAX_CATALOG_RESPONSE_BYTES + 1],
})])),
observations: Arc::new(Mutex::new(Vec::new())),
};
assert_eq!(
discover_openrouter_authenticated(
&oversized_openrouter,
&FixtureResolver,
"secret:fixture",
snapshot("snapshot.openrouter.bounds.bytes"),
)
.await,
Err(CatalogDiscoveryError::ResponseBounds)
);
let many_openrouter = (0..=MAX_CATALOG_MODELS)
.map(|index| {
json!({
"id": format!("vendor/model-{index}"),
"context_length": 8_192
})
})
.collect::<Vec<_>>();
let openrouter_executor = FakeExecutor {
responses: Mutex::new(VecDeque::from([response(json!({
"data": many_openrouter
}))])),
observations: Arc::new(Mutex::new(Vec::new())),
};
assert_eq!(
discover_openrouter_authenticated(
&openrouter_executor,
&FixtureResolver,
"secret:fixture",
snapshot("snapshot.openrouter.bounds.models"),
)
.await,
Err(CatalogDiscoveryError::ResponseBounds)
);
}
#[tokio::test]
async fn catalog_request_absolute_deadline_times_out_non_completing_executor() {
let request =
catalog_request_with_resolved_secret(CatalogEndpoint::LocalOllamaTags, None, None)
.expect("valid local catalog request");
let started = std::time::Instant::now();
let error = execute_with_absolute_deadline(
&NeverCompletesExecutor,
request,
CATALOG_REQUEST_TIMEOUT,
)
.await
.expect_err("absolute catalog request deadline");
let elapsed = started.elapsed();
assert_eq!(error, CatalogDiscoveryError::Timeout);
assert_eq!(error.to_string(), "catalog discovery timed out");
assert!(elapsed >= CATALOG_REQUEST_TIMEOUT, "elapsed: {elapsed:?}");
assert!(
elapsed <= CATALOG_REQUEST_TIMEOUT + Duration::from_secs(1),
"elapsed: {elapsed:?}"
);
}
#[tokio::test]
async fn normalized_snapshot_hash_is_independent_of_provider_list_order() {
async fn discover(models: Value, snapshot_id: &str) -> CatalogSnapshot {
let executor = FakeExecutor {
responses: Mutex::new(VecDeque::from([response(json!({"models": models}))])),
observations: Arc::new(Mutex::new(Vec::new())),
};
discover_local_ollama(
&executor,
snapshot(snapshot_id),
OllamaDiscoveryOptions::default(),
)
.await
.expect("deterministic local snapshot")
}
let forward = discover(
json!([{"name": "alpha:latest"}, {"name": "zeta:latest"}]),
"snapshot.order.1",
)
.await;
let reverse = discover(
json!([{"name": "zeta:latest"}, {"name": "alpha:latest"}]),
"snapshot.order.2",
)
.await;
assert_eq!(forward.content_sha256(), reverse.content_sha256());
}
}