use std::sync::Arc;
use anyhow::{Context, Result};
use sea_orm::DatabaseConnection;
use bitrouter_sdk::App;
use bitrouter_sdk::acp::{AcpStdioExecutor, ConfigAcpRoutingTable};
use bitrouter_sdk::config::{Config, ConfigRoutingTable};
use bitrouter_sdk::language_model::protocol::OutboundDispatch;
use bitrouter_sdk::language_model::{AuthAppliers, HttpExecutor, HttpTimeouts};
use bitrouter_sdk::mcp::aggregating_executor::AggregatingExecutor;
use bitrouter_sdk::mcp::caching_executor::{CacheTtls, CachingExecutor};
use bitrouter_sdk::mcp::config_routing::{ConfigMcpRoutingTable, McpServerAggregateConfig};
use bitrouter_sdk::mcp::rmcp_executor::RmcpExecutor;
use bitrouter_guardrails::{GuardrailConfig, GuardrailsPlugin};
use bitrouter_observe::OTEL_ENABLED;
use bitrouter_observe::otel::{OtelConfig, OtelExporter, OtelObserveHook};
use bitrouter_sdk::MetricsRenderer;
use crate::auth::AuthHook;
use crate::daemon::{NoopObserveStatus, ObserveStatusPayload, ObserveStatusProvider};
use crate::metering::{ContextTier, MeteringRecorder, MeteringStore, ModelPricing, PricingTable};
use crate::policy::{PolicyHook, PolicyStore};
pub struct Assembled {
pub app: App,
pub db: DatabaseConnection,
pub policy_store: Arc<PolicyStore>,
pub routing_table: Arc<ConfigRoutingTable>,
pub observe: Arc<dyn ObserveStatusProvider>,
pub otel_exporter: Option<Arc<OtelExporter>>,
pub otel_init_error: Option<String>,
}
struct OtelExporterStatus {
exporter: Arc<OtelExporter>,
}
#[async_trait::async_trait]
impl ObserveStatusProvider for OtelExporterStatus {
fn status(&self) -> ObserveStatusPayload {
let s = self.exporter.status();
ObserveStatusPayload {
compiled_in: s.compiled_in,
exporter_wired: s.exporter_wired,
endpoint: s.endpoint,
header_count: s.header_count,
service_name: s.service_name,
resource_attribute_count: s.resource_attribute_count,
sampler: s.sampler,
sampler_arg: s.sampler_arg,
metrics_enabled: s.metrics_enabled,
api_key_count: s.api_key_count,
api_key_cap: s.api_key_cap,
user_id_count: s.user_id_count,
user_id_cap: s.user_id_cap,
active_spans: s.active_spans,
}
}
async fn shutdown(&self) {
let exporter = self.exporter.clone();
if let Err(e) = tokio::task::spawn_blocking(move || exporter.shutdown()).await {
tracing::warn!(error = %e, "OTel exporter shutdown task panicked");
}
}
}
pub async fn build_app(config: &Config) -> Result<Assembled> {
build_app_with_path(config, None).await
}
pub async fn build_app_with_path(
config: &Config,
config_path: Option<&std::path::Path>,
) -> Result<Assembled> {
let db = crate::db::connect(&config.database.url)
.await
.with_context(|| format!("connecting to database {}", config.database.url))?;
crate::db::run_migrations(&db)
.await
.context("running database migrations")?;
#[cfg(unix)]
if let Some(path) = sqlite_file_path(&config.database.url) {
use std::os::unix::fs::PermissionsExt;
if let Ok(meta) = std::fs::metadata(&path) {
let mut perms = meta.permissions();
perms.set_mode(0o600);
if let Err(e) = std::fs::set_permissions(&path, perms) {
tracing::warn!(?path, %e, "failed to chmod 0600 on sqlite db file");
}
}
}
let mut resolved = config.clone();
bitrouter_providers::apply_builtin_defaults(&mut resolved);
bitrouter_sdk::config::discover_models(&mut resolved).await;
let routing_table = Arc::new(match config_path {
Some(path) => ConfigRoutingTable::from_config_with_path(resolved, path),
None => ConfigRoutingTable::from_config(resolved),
});
let routing_table_for_reload = routing_table.clone();
let auth_appliers = build_auth_appliers(config)?;
let executor = Arc::new(
HttpExecutor::with_dispatch_and_auth(
HttpTimeouts::default(),
OutboundDispatch::builtin(),
auth_appliers,
)
.context("building the upstream HTTP executor")?,
);
let pricing = Arc::new(build_pricing_table(config));
let metering_store = MeteringStore::new(db.clone());
let metering_store_for_policy = metering_store.clone();
let metering_store_for_recorder = metering_store.clone();
let pricing_for_recorder = pricing.clone();
let policy_store: Arc<PolicyStore> = Arc::new(load_policy_store(config).await?);
let policy_store_for_reload = policy_store.clone();
let guardrail_rules = build_guardrail_config(config)?
.compile()
.context("compiling guardrail patterns")?;
let metrics_renderer: Arc<dyn MetricsRenderer> = Arc::new(EmptyMetricsRenderer);
let (otel_exporter, otel_init_error): (Option<Arc<OtelExporter>>, Option<String>) =
match build_otel_config(config)? {
Some(c) => match OtelExporter::new(c) {
Ok(exporter) => (Some(Arc::new(exporter)), None),
Err(e) => (
None,
Some(format!("failed to initialise OpenTelemetry: {e}")),
),
},
None => (None, None),
};
let observe_provider: Arc<dyn ObserveStatusProvider> = match otel_exporter.clone() {
Some(exporter) => Arc::new(OtelExporterStatus { exporter }),
None => Arc::new(NoopObserveStatus {
compiled_in: OTEL_ENABLED,
}),
};
let otel_for_hook = otel_exporter.clone();
let otel_for_assembled = otel_exporter;
let mcp_aggregate_route = if config.mcp.aggregate.enabled {
Some(config.mcp.aggregate.route.clone())
} else {
None
};
if let Some(route) = mcp_aggregate_route.as_deref() {
let agg_last = route.trim_end_matches('/').rsplit('/').next().unwrap_or("");
if !agg_last.is_empty() && config.mcp_servers.keys().any(|k| k.as_str() == agg_last) {
anyhow::bail!(
"mcp_servers entry '{agg_last}' would be shadowed by the per-server mount at \
'{route}/{agg_last}' (derived from the aggregate route '{route}'). Rename the \
server or move the aggregate route. Note: this check only catches the \
last-segment overlap; axum mounts may still reject other shapes at startup."
);
}
}
let mcp_routing = if config.mcp_servers.is_empty() {
None
} else {
Some(Arc::new(
ConfigMcpRoutingTable::from_configs(config.mcp_servers.iter().map(|(k, v)| {
let agg = McpServerAggregateConfig {
aggregate: v.aggregate,
tool_prefix: v.tool_prefix.clone().unwrap_or_else(|| format!("{k}__")),
};
(k.clone(), v.clone(), agg)
}))
.context("building the MCP routing table from config.mcp_servers")?,
))
};
let mcp_executor = mcp_routing.as_ref().map(|_| {
let rmcp: Arc<RmcpExecutor> = Arc::new(RmcpExecutor::new());
let inner_for_cache: Arc<RmcpExecutor> = rmcp.clone();
if config.mcp.cache.enabled {
let ttls: CacheTtls = (&config.mcp.cache).into();
let cached: Arc<CachingExecutor<RmcpExecutor>> = Arc::new(
CachingExecutor::new(inner_for_cache, ttls)
.with_invalidation(rmcp.invalidation_receiver()),
);
Arc::new(AggregatingExecutor::new(cached)) as Arc<dyn bitrouter_sdk::mcp::Executor>
} else {
Arc::new(AggregatingExecutor::new(inner_for_cache))
as Arc<dyn bitrouter_sdk::mcp::Executor>
}
});
let acp_routing = if config.agents.is_empty() {
None
} else {
Some(Arc::new(
ConfigAcpRoutingTable::from_configs(
config.agents.iter().map(|(k, v)| (k.clone(), v.clone())),
)
.context("building the ACP routing table from config.agents")?,
))
};
let acp_executor: Option<Arc<AcpStdioExecutor>> = acp_routing
.as_ref()
.map(|_| Arc::new(AcpStdioExecutor::new()));
let db_for_hooks = db.clone();
let app = App::builder()
.skip_auth(config.server.skip_auth)
.metrics_renderer(metrics_renderer)
.language_model(move |lm| {
lm.routing_table(routing_table).executor(executor);
lm.pre_request_hook(AuthHook::new(db_for_hooks.clone()));
lm.pre_request_hook(PolicyHook::new(
policy_store.clone(),
Some(metering_store_for_policy),
));
if let Some(exporter) = otel_for_hook {
lm.observe_hook(OtelObserveHook::new(exporter));
}
lm.settlement_recorder(MeteringRecorder::new(
metering_store_for_recorder,
pricing_for_recorder,
));
});
let app = if guardrail_rules.is_empty() {
app
} else {
app.plugin(GuardrailsPlugin::with_static(guardrail_rules))
};
let app = match (mcp_routing, mcp_executor) {
(Some(table), Some(exec)) => {
let app = app.mcp(move |m| {
m.routing_table(table).executor(exec);
});
if let Some(route) = mcp_aggregate_route {
app.mcp_aggregate_route(route)
} else {
app
}
}
_ => app,
};
let app = match (acp_routing, acp_executor) {
(Some(table), Some(exec)) => app.acp(move |a| {
a.routing_table(table).executor(exec);
}),
_ => app,
};
let app = app.build().context("building the App")?;
Ok(Assembled {
app,
db,
policy_store: policy_store_for_reload,
routing_table: routing_table_for_reload,
observe: observe_provider,
otel_exporter: otel_for_assembled,
otel_init_error,
})
}
fn build_auth_appliers(config: &Config) -> Result<AuthAppliers> {
let mut appliers = AuthAppliers::new();
let store_path = bitrouter_providers::oauth::credential_store::CredentialStore::default_path()
.map(|s| s.path().to_path_buf())
.context("resolving credential store path")?;
crate::cloud::register_if_configured(config, &mut appliers)?;
if config.providers.contains_key("github-copilot") {
let applier = bitrouter_providers::copilot::CopilotAuthApplier::new(&store_path)
.context("building the github-copilot AuthApplier")?;
appliers.register("github-copilot", Arc::new(applier));
}
if config.providers.contains_key("anthropic") {
let applier = bitrouter_providers::anthropic::AnthropicOAuthApplier::new(&store_path)
.context("building the anthropic AuthApplier")?;
appliers.register("anthropic", Arc::new(applier));
}
if config.providers.contains_key("openai-codex") {
let applier = bitrouter_providers::codex::OpenAiCodexAuthApplier::new(&store_path)
.context("building the openai-codex AuthApplier")?;
appliers.register("openai-codex", Arc::new(applier));
}
Ok(appliers)
}
fn build_pricing_table(config: &Config) -> PricingTable {
let mut table = PricingTable::new();
for (provider_id, provider) in &config.providers {
for model in &provider.models {
if let Some(pricing) = &model.pricing {
let mut model_pricing = ModelPricing::new(
pricing.input_micro_usd_per_token,
pricing.output_micro_usd_per_token,
);
model_pricing.context_tiers = pricing
.context_tiers
.iter()
.map(|t| ContextTier {
above_input_tokens: t.above_input_tokens,
input_micro_usd_per_token: Some(t.input_micro_usd_per_token),
output_micro_usd_per_token: Some(t.output_micro_usd_per_token),
})
.collect();
table.insert(provider_id.clone(), model.id.clone(), model_pricing);
}
}
}
table
}
async fn load_policy_store(config: &Config) -> Result<PolicyStore> {
let dir = config
.plugins
.get("bitrouter-policy")
.and_then(|c| c.get("policy_dir"))
.and_then(|v| v.as_str());
match dir {
Some(dir) => PolicyStore::load_dir(dir)
.await
.with_context(|| format!("loading policies from {dir}")),
None => Ok(PolicyStore::new()),
}
}
fn build_guardrail_config(config: &Config) -> Result<GuardrailConfig> {
let Some(value) = config.plugins.get("bitrouter-guardrails") else {
return Ok(GuardrailConfig::default());
};
serde_json::from_value(value.clone()).context("plugins.bitrouter-guardrails failed to parse")
}
struct EmptyMetricsRenderer;
impl MetricsRenderer for EmptyMetricsRenderer {
fn render(&self) -> String {
"# Prometheus metrics have been removed in favor of OpenTelemetry.\n".to_string()
+ "# Configure OTLP export via plugins.bitrouter-observe.otel\n"
}
}
fn build_otel_config(config: &Config) -> Result<Option<OtelConfig>> {
let observe = config.plugins.get("bitrouter-observe");
if let Some(otel_value) = observe.and_then(|c| c.get("otel")) {
let cfg = serde_json::from_value::<OtelConfig>(otel_value.clone())
.context("plugins.bitrouter-observe.otel failed to parse")?;
return Ok(Some(cfg));
}
if let Some(endpoint) = observe
.and_then(|c| c.get("otlp_endpoint"))
.and_then(|v| v.as_str())
{
let cfg = OtelConfig {
endpoint: endpoint.to_string(),
..OtelConfig::default()
};
tracing::warn!(
"plugins.bitrouter-observe.otlp_endpoint is deprecated; switch to plugins.bitrouter-observe.otel",
);
return Ok(Some(cfg));
}
if std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").is_ok() {
return Ok(Some(OtelConfig::default()));
}
Ok(None)
}
#[cfg(unix)]
fn sqlite_file_path(url: &str) -> Option<std::path::PathBuf> {
let after_scheme = url
.strip_prefix("sqlite://")
.or_else(|| url.strip_prefix("sqlite:"))?;
let path = after_scheme.split('?').next().unwrap_or(after_scheme);
if path.is_empty() || path == ":memory:" {
return None;
}
Some(std::path::PathBuf::from(path))
}
#[cfg(all(test, unix))]
mod sqlite_path_tests {
use super::sqlite_file_path;
use std::path::PathBuf;
#[test]
fn parses_common_sqlite_urls() {
assert_eq!(
sqlite_file_path("sqlite:///var/lib/bitrouter.db"),
Some(PathBuf::from("/var/lib/bitrouter.db"))
);
assert_eq!(
sqlite_file_path("sqlite:bitrouter.db"),
Some(PathBuf::from("bitrouter.db"))
);
assert_eq!(
sqlite_file_path("sqlite:bitrouter.db?cache=shared"),
Some(PathBuf::from("bitrouter.db"))
);
assert_eq!(sqlite_file_path(":memory:"), None);
assert_eq!(sqlite_file_path("sqlite::memory:"), None);
assert_eq!(sqlite_file_path("postgres://u:p@host/bitrouter"), None);
assert_eq!(sqlite_file_path("mysql://u:p@host/bitrouter"), None);
}
}
#[cfg(test)]
mod otel_config_tests {
use super::{Config, build_otel_config};
fn config_with_observe(observe: serde_json::Value) -> Config {
let mut config = Config::default();
config
.plugins
.insert("bitrouter-observe".to_string(), observe);
config
}
#[test]
fn malformed_otel_block_is_a_hard_error() {
let config = config_with_observe(serde_json::json!({
"otel": { "sampler": "not_a_real_sampler" }
}));
assert!(
build_otel_config(&config).is_err(),
"a malformed otel block must be a hard error",
);
}
#[test]
fn valid_otel_block_parses() {
let config = config_with_observe(serde_json::json!({
"otel": { "endpoint": "http://collector:4318" }
}));
let cfg = build_otel_config(&config)
.expect("valid otel block is Ok")
.expect("valid otel block yields Some");
assert_eq!(cfg.endpoint, "http://collector:4318");
}
#[test]
fn legacy_otlp_endpoint_shim_still_works() {
let config = config_with_observe(serde_json::json!({
"otlp_endpoint": "http://legacy:4318"
}));
let cfg = build_otel_config(&config)
.expect("legacy shim is Ok")
.expect("legacy shim yields Some");
assert_eq!(cfg.endpoint, "http://legacy:4318");
}
}