use std::collections::{BTreeMap, BTreeSet, VecDeque};
use std::fmt;
use std::path::PathBuf;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use crate::daemon::DaemonReloader;
use crate::policy::PolicyStore;
use chrono::{SecondsFormat, Utc};
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
const MAX_MIXED_STATE_HISTORY: usize = 32;
pub const PREPARATION_TIMEOUT_SECONDS: u64 = 60;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum ConfigSourceKind {
File,
Default,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum SavedConfigState {
Available,
Generated,
Missing,
Invalid,
Unavailable,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum RunningConfigState {
InSync,
ReloadRequired,
RestartRequired,
Mixed,
Unknown,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum AuxiliaryConfigState {
InSync,
ReloadRequired,
NotConfigured,
Missing,
Invalid,
Unavailable,
Unknown,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
pub struct ConfigurationState {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub server_instance_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub generation: Option<u64>,
pub source: ConfigSourceKind,
pub saved: SavedConfigState,
pub running: RunningConfigState,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub reload_required_fields: Vec<String>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub restart_required_fields: Vec<String>,
pub named_policy: AuxiliaryConfigState,
pub access_policies: AuxiliaryConfigState,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub last_reload: Option<ReloadReport>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub mixed_state_history: Vec<ReloadReport>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum ReloadConsistency {
Consistent,
Mixed,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum ReloadOutcome {
Succeeded,
Failed,
PartiallyApplied,
Unknown,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum ReloadParticipant {
RoutingTable,
UpstreamTimeoutClients,
NamedPolicyRuntime,
AccessPolicyStore,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum ReloadParticipantOutcome {
Applied,
Unchanged,
Failed,
NotAttempted,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
pub struct ReloadFailure {
pub code: String,
pub message: String,
}
impl ReloadFailure {
fn new(code: &str, message: &str) -> Self {
Self {
code: code.to_string(),
message: message.to_string(),
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
pub struct ReloadParticipantReport {
pub participant: ReloadParticipant,
pub outcome: ReloadParticipantOutcome,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub error: Option<ReloadFailure>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
pub struct ReloadReport {
pub server_instance_id: String,
pub generation: u64,
pub outcome: ReloadOutcome,
pub participants: Vec<ReloadParticipantReport>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub restart_required_fields: Vec<String>,
pub started_at: String,
pub completed_at: String,
}
impl ReloadReport {
fn pending(instance: String, generation: u64) -> Self {
Self {
server_instance_id: instance,
generation,
outcome: ReloadOutcome::Failed,
participants: ReloadParticipant::all()
.into_iter()
.map(|participant| ReloadParticipantReport {
participant,
outcome: ReloadParticipantOutcome::NotAttempted,
error: None,
})
.collect(),
restart_required_fields: Vec::new(),
started_at: now_rfc3339(),
completed_at: String::new(),
}
}
#[cfg(test)]
pub(crate) fn succeeded(instance: String, generation: u64) -> Self {
let mut report = Self::pending(instance, generation);
for participant in &mut report.participants {
participant.outcome = ReloadParticipantOutcome::Unchanged;
}
report.outcome = ReloadOutcome::Succeeded;
report.completed_at = now_rfc3339();
report
}
pub(crate) fn unsupported() -> Self {
let mut report = Self::pending("unsupported".to_string(), 0);
report.participant_failed(
ReloadParticipant::RoutingTable,
ReloadFailure::new(
"unsupported_reload",
"this daemon does not support coordinated reload",
),
);
report.outcome = ReloadOutcome::Failed;
report.completed_at = now_rfc3339();
report
}
fn participant_mut(
&mut self,
participant: ReloadParticipant,
) -> Option<&mut ReloadParticipantReport> {
self.participants
.iter_mut()
.find(|entry| entry.participant == participant)
}
fn participant_applied(&mut self, participant: ReloadParticipant) {
if let Some(entry) = self.participant_mut(participant) {
entry.outcome = ReloadParticipantOutcome::Applied;
entry.error = None;
}
}
fn participant_unchanged(&mut self, participant: ReloadParticipant) {
if let Some(entry) = self.participant_mut(participant) {
entry.outcome = ReloadParticipantOutcome::Unchanged;
entry.error = None;
}
}
fn participant_failed(&mut self, participant: ReloadParticipant, error: ReloadFailure) {
if let Some(entry) = self.participant_mut(participant) {
entry.outcome = ReloadParticipantOutcome::Failed;
entry.error = Some(error);
}
}
fn finish(&mut self, outcome: ReloadOutcome) {
self.outcome = outcome;
self.completed_at = now_rfc3339();
}
fn safe_summary(&self) -> &'static str {
match self.outcome {
ReloadOutcome::Succeeded => "reload succeeded",
ReloadOutcome::Failed => "reload failed before applying every requested change",
ReloadOutcome::PartiallyApplied => {
"reload partially applied; inspect participant results"
}
ReloadOutcome::Unknown => {
"reload outcome is unknown; inspect live state before retrying"
}
}
}
fn any_applied(&self) -> bool {
self.participants
.iter()
.any(|entry| entry.outcome == ReloadParticipantOutcome::Applied)
}
}
impl ReloadParticipant {
fn all() -> [Self; 4] {
[
Self::RoutingTable,
Self::UpstreamTimeoutClients,
Self::NamedPolicyRuntime,
Self::AccessPolicyStore,
]
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
pub struct ReloadState {
pub server_instance_id: String,
pub generation: u64,
pub running: bool,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub running_generation: Option<u64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub last_outcome: Option<ReloadReport>,
pub consistency: ReloadConsistency,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub mixed_state_history: Vec<ReloadReport>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum ReloadAdmissionError {
Unsupported,
ServerInstanceChanged,
StaleGeneration,
ReloadInProgress,
GenerationExhausted,
}
impl ReloadAdmissionError {
pub fn code(&self) -> &'static str {
match self {
Self::Unsupported => "unsupported_reload",
Self::ServerInstanceChanged => "server_instance_changed",
Self::StaleGeneration => "stale_generation",
Self::ReloadInProgress => "reload_in_progress",
Self::GenerationExhausted => "generation_exhausted",
}
}
}
impl fmt::Display for ReloadAdmissionError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str(self.code())
}
}
impl std::error::Error for ReloadAdmissionError {}
pub struct ReloadReservation {
coordinator: Arc<ReloadCoordinator>,
ticket: uuid::Uuid,
server_instance_id: String,
generation: u64,
mutation_started: AtomicBool,
completed: AtomicBool,
progress: Mutex<Option<ReloadReport>>,
}
impl ReloadReservation {
pub fn server_instance_id(&self) -> &str {
&self.server_instance_id
}
pub fn generation(&self) -> u64 {
self.generation
}
fn mark_mutation_started(&self) {
self.mutation_started.store(true, Ordering::Release);
}
fn update_progress(&self, report: &ReloadReport) {
let mut progress = match self.progress.lock() {
Ok(progress) => progress,
Err(poisoned) => poisoned.into_inner(),
};
*progress = Some(report.clone());
}
fn progress(&self) -> Option<ReloadReport> {
match self.progress.lock() {
Ok(progress) => progress.clone(),
Err(poisoned) => poisoned.into_inner().clone(),
}
}
fn belongs_to(&self, coordinator: &Arc<ReloadCoordinator>) -> bool {
Arc::ptr_eq(&self.coordinator, coordinator)
}
}
impl Drop for ReloadReservation {
fn drop(&mut self) {
if self.completed.load(Ordering::Acquire) {
return;
}
if self.mutation_started.load(Ordering::Acquire) {
self.coordinator.interrupted(self, self.progress());
} else {
self.coordinator.abandon(self);
}
}
}
#[derive(Debug)]
struct CoordinatorState {
server_instance_id: String,
generation: u64,
running: Option<(uuid::Uuid, u64)>,
last_outcome: Option<ReloadReport>,
consistency: ReloadConsistency,
mixed_state_history: VecDeque<ReloadReport>,
}
pub(crate) struct ReloadCoordinator {
state: Mutex<CoordinatorState>,
}
impl ReloadCoordinator {
pub(crate) fn new() -> Arc<Self> {
Arc::new(Self {
state: Mutex::new(CoordinatorState {
server_instance_id: uuid::Uuid::new_v4().to_string(),
generation: 0,
running: None,
last_outcome: None,
consistency: ReloadConsistency::Consistent,
mixed_state_history: VecDeque::new(),
}),
})
}
pub(crate) fn state(&self) -> ReloadState {
let state = match self.state.lock() {
Ok(state) => state,
Err(poisoned) => poisoned.into_inner(),
};
ReloadState {
server_instance_id: state.server_instance_id.clone(),
generation: state.generation,
running: state.running.is_some(),
running_generation: state.running.map(|(_, generation)| generation),
last_outcome: state.last_outcome.clone(),
consistency: state.consistency,
mixed_state_history: state.mixed_state_history.iter().cloned().collect(),
}
}
pub(crate) fn reserve_remote(
self: &Arc<Self>,
expected_instance: &str,
expected_generation: u64,
) -> Result<ReloadReservation, ReloadAdmissionError> {
let mut state = match self.state.lock() {
Ok(state) => state,
Err(poisoned) => poisoned.into_inner(),
};
if state.server_instance_id != expected_instance {
return Err(ReloadAdmissionError::ServerInstanceChanged);
}
if state.generation != expected_generation {
return Err(ReloadAdmissionError::StaleGeneration);
}
Self::reserve_locked(self, &mut state)
}
pub(crate) fn reserve_local(
self: &Arc<Self>,
) -> Result<ReloadReservation, ReloadAdmissionError> {
let mut state = match self.state.lock() {
Ok(state) => state,
Err(poisoned) => poisoned.into_inner(),
};
Self::reserve_locked(self, &mut state)
}
fn reserve_locked(
coordinator: &Arc<Self>,
state: &mut CoordinatorState,
) -> Result<ReloadReservation, ReloadAdmissionError> {
if state.running.is_some() {
return Err(ReloadAdmissionError::ReloadInProgress);
}
let generation = state
.generation
.checked_add(1)
.ok_or(ReloadAdmissionError::GenerationExhausted)?;
let ticket = uuid::Uuid::new_v4();
state.running = Some((ticket, generation));
Ok(ReloadReservation {
coordinator: Arc::clone(coordinator),
ticket,
server_instance_id: state.server_instance_id.clone(),
generation,
mutation_started: AtomicBool::new(false),
completed: AtomicBool::new(false),
progress: Mutex::new(None),
})
}
pub(crate) fn complete(&self, reservation: &ReloadReservation, mut report: ReloadReport) {
report.completed_at = now_rfc3339();
let mut state = match self.state.lock() {
Ok(state) => state,
Err(poisoned) => poisoned.into_inner(),
};
if state.running != Some((reservation.ticket, reservation.generation)) {
return;
}
state.running = None;
state.generation = reservation.generation;
state.consistency = match report.outcome {
ReloadOutcome::Succeeded => ReloadConsistency::Consistent,
ReloadOutcome::PartiallyApplied | ReloadOutcome::Unknown => ReloadConsistency::Mixed,
ReloadOutcome::Failed => state.consistency,
};
if matches!(
report.outcome,
ReloadOutcome::PartiallyApplied | ReloadOutcome::Unknown
) {
state.mixed_state_history.push_back(report.clone());
while state.mixed_state_history.len() > MAX_MIXED_STATE_HISTORY {
let _ = state.mixed_state_history.pop_front();
}
}
state.last_outcome = Some(report);
reservation.completed.store(true, Ordering::Release);
}
fn abandon(&self, reservation: &ReloadReservation) {
let mut state = match self.state.lock() {
Ok(state) => state,
Err(poisoned) => poisoned.into_inner(),
};
if state.running == Some((reservation.ticket, reservation.generation)) {
state.running = None;
}
}
fn interrupted(&self, reservation: &ReloadReservation, progress: Option<ReloadReport>) {
let mut report = match progress {
Some(report) => report,
None => ReloadReport::pending(
reservation.server_instance_id.clone(),
reservation.generation,
),
};
report.finish(ReloadOutcome::Unknown);
self.complete(reservation, report);
}
}
fn now_rfc3339() -> String {
Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true)
}
fn restart_required_fields(
current: &bitrouter_sdk::config::Config,
candidate: &bitrouter_sdk::config::Config,
unclassified_fields: Option<&BTreeSet<String>>,
) -> Vec<String> {
let mut fields = BTreeSet::new();
if let Some(unclassified_fields) = unclassified_fields {
fields.extend(unclassified_fields.iter().cloned());
}
if current.server.listen != candidate.server.listen {
fields.insert("server.listen".to_string());
}
if current.server.control_socket != candidate.server.control_socket {
fields.insert("server.control_socket".to_string());
}
if current.server.log_level != candidate.server.log_level {
fields.insert("server.log_level".to_string());
}
if current.server.require_known_pricing != candidate.server.require_known_pricing {
fields.insert("server.require_known_pricing".to_string());
}
if current.server.skip_auth != candidate.server.skip_auth {
fields.insert("server.skip_auth".to_string());
}
if current.inherit_defaults != candidate.inherit_defaults {
fields.insert("inherit_defaults".to_string());
}
if current.decision_model != candidate.decision_model {
fields.insert("decision_model".to_string());
}
if current.control != candidate.control {
fields.insert("control".to_string());
}
if current.chat != candidate.chat {
fields.insert("chat".to_string());
}
if current.database.url != candidate.database.url {
fields.insert("database.url".to_string());
}
if serialized_values_differ(¤t.eval, &candidate.eval) {
fields.insert("eval".to_string());
}
if current.trajectory != candidate.trajectory {
fields.insert("trajectory".to_string());
}
if current.acp_recording != candidate.acp_recording {
fields.insert("acp_recording".to_string());
}
if current.continuation != candidate.continuation {
fields.insert("continuation".to_string());
}
if current.routers != candidate.routers {
fields.insert("routers".to_string());
}
if current.checkers != candidate.checkers {
fields.insert("checkers".to_string());
}
if current.plugins != candidate.plugins {
fields.insert("plugins".to_string());
}
if mcp_config_changed(current, candidate) {
fields.insert("mcp".to_string());
}
if mcp_servers_changed(current, candidate) {
fields.insert("mcp_servers".to_string());
}
if server_tools_changed(current, candidate) {
fields.insert("server_tools".to_string());
}
if agents_changed(current, candidate) {
fields.insert("agents".to_string());
}
if current.upstream.fallback_backoff_ms != candidate.upstream.fallback_backoff_ms {
fields.insert("upstream.fallback_backoff_ms".to_string());
}
if pricing_signature(current) != pricing_signature(candidate) {
fields.insert("providers.*.models.*.pricing".to_string());
}
for provider_id in [
"bitrouter",
"github-copilot",
"anthropic",
"claude-code",
"openai-codex",
"supergrok",
] {
if current.providers.contains_key(provider_id)
!= candidate.providers.contains_key(provider_id)
{
fields.insert(format!("providers.{provider_id}"));
}
}
fields.into_iter().collect()
}
#[cfg(test)]
fn unclassified_config_paths(value: &serde_json::Value) -> BTreeSet<String> {
unclassified_config(value).0
}
#[derive(Clone)]
struct UnclassifiedConfigValue {
path: String,
value: serde_json::Value,
}
type UnclassifiedConfigValues = BTreeMap<String, UnclassifiedConfigValue>;
fn unclassified_config(value: &serde_json::Value) -> (BTreeSet<String>, UnclassifiedConfigValues) {
let schema = schemars::schema_for!(bitrouter_sdk::config::Config);
let root = schema.as_value();
let mut fields = BTreeSet::new();
let mut values = BTreeMap::new();
collect_unclassified_schema_paths(value, root, root, "", "", &mut fields, &mut values);
(fields, values)
}
fn collect_unclassified_schema_paths(
value: &serde_json::Value,
schema: &serde_json::Value,
root: &serde_json::Value,
path: &str,
location: &str,
fields: &mut BTreeSet<String>,
unclassified_values: &mut UnclassifiedConfigValues,
) {
let Some(schema) = resolve_schema(schema, root) else {
record_unclassified_value(value, path, location, fields, unclassified_values);
return;
};
if schema.as_bool() == Some(false) {
record_unclassified_value(value, path, location, fields, unclassified_values);
return;
}
if schema.as_bool() == Some(true) {
return;
}
if !schema_accepts_value_kind(schema, value) {
record_unclassified_value(value, path, location, fields, unclassified_values);
return;
}
for union in ["anyOf", "oneOf"] {
if let Some(variants) = schema.get(union).and_then(serde_json::Value::as_array) {
let mut best: Option<(BTreeSet<String>, UnclassifiedConfigValues)> = None;
for variant in variants {
let mut candidate = BTreeSet::new();
let mut candidate_values = BTreeMap::new();
collect_unclassified_schema_paths(
value,
variant,
root,
path,
location,
&mut candidate,
&mut candidate_values,
);
if candidate.is_empty() {
return;
}
if best
.as_ref()
.is_none_or(|(current, _)| candidate.len() < current.len())
{
best = Some((candidate, candidate_values));
}
}
if let Some((best_fields, best_values)) = best {
fields.extend(best_fields);
unclassified_values.extend(best_values);
}
return;
}
}
let all_of = schema.get("allOf").and_then(serde_json::Value::as_array);
let has_direct_object_shape = schema.get("properties").is_some()
|| schema.get("additionalProperties").is_some()
|| schema.get("unevaluatedProperties").is_some()
|| schema.get("patternProperties").is_some();
if all_of.is_none() || has_direct_object_shape {
match value {
serde_json::Value::Object(values) => {
let properties = schema
.get("properties")
.and_then(serde_json::Value::as_object);
let additional = schema
.get("additionalProperties")
.or_else(|| schema.get("unevaluatedProperties"));
for (name, child) in values {
let child_path = schema_child_path(path, name);
let child_location = schema_child_location(location, name);
if let Some(property) = schema_property(properties, path, name) {
collect_unclassified_schema_paths(
child,
property,
root,
&child_path,
&child_location,
fields,
unclassified_values,
);
} else if let Some(additional) = additional {
if additional.as_bool() == Some(false) {
fields.insert(child_path);
unclassified_values.insert(
child_location,
UnclassifiedConfigValue {
path: schema_child_path(path, name),
value: child.clone(),
},
);
} else if additional.as_bool() != Some(true) {
collect_unclassified_schema_paths(
child,
additional,
root,
&child_path,
&child_location,
fields,
unclassified_values,
);
}
} else {
fields.insert(child_path.clone());
unclassified_values.insert(
child_location,
UnclassifiedConfigValue {
path: child_path,
value: child.clone(),
},
);
}
}
}
serde_json::Value::Array(values) => {
if let Some(items) = schema.get("items") {
for (index, child) in values.iter().enumerate() {
let child_path = format!("{path}[{index}]");
let child_location = format!("{location}/{index}");
collect_unclassified_schema_paths(
child,
items,
root,
&child_path,
&child_location,
fields,
unclassified_values,
);
}
}
}
_ => {}
}
}
if let Some(all_of) = all_of {
for part in all_of {
collect_unclassified_schema_paths(
value,
part,
root,
path,
location,
fields,
unclassified_values,
);
}
}
}
fn record_unclassified_value(
value: &serde_json::Value,
path: &str,
location: &str,
fields: &mut BTreeSet<String>,
values: &mut UnclassifiedConfigValues,
) {
if path.is_empty() {
return;
}
fields.insert(path.to_string());
values.insert(
location.to_string(),
UnclassifiedConfigValue {
path: path.to_string(),
value: value.clone(),
},
);
}
fn schema_child_location(parent: &str, child: &str) -> String {
let escaped = child.replace('~', "~0").replace('/', "~1");
format!("{parent}/{escaped}")
}
fn changed_unclassified_config_paths(
startup: Option<&UnclassifiedConfigValues>,
candidate: Option<&UnclassifiedConfigValues>,
candidate_fields: Option<&BTreeSet<String>>,
) -> BTreeSet<String> {
match (startup, candidate) {
(Some(startup), Some(candidate)) => startup
.keys()
.chain(candidate.keys())
.filter(|location| {
startup.get(*location).map(|entry| &entry.value)
!= candidate.get(*location).map(|entry| &entry.value)
})
.filter_map(|location| {
candidate
.get(location)
.or_else(|| startup.get(location))
.map(|entry| entry.path.clone())
})
.collect(),
(None, Some(_)) => candidate_fields.cloned().unwrap_or_default(),
(Some(startup), None) => startup.values().map(|entry| entry.path.clone()).collect(),
(None, None) => BTreeSet::new(),
}
}
fn schema_property<'a>(
properties: Option<&'a serde_json::Map<String, serde_json::Value>>,
parent_path: &str,
name: &str,
) -> Option<&'a serde_json::Value> {
let properties = properties?;
if let Some(property) = properties.get(name) {
return Some(property);
}
if parent_path == "policy" && name == "writeback" {
return properties.get("mode");
}
None
}
fn resolve_schema<'a>(
schema: &'a serde_json::Value,
root: &'a serde_json::Value,
) -> Option<&'a serde_json::Value> {
match schema.get("$ref").and_then(serde_json::Value::as_str) {
Some(reference) => reference
.strip_prefix('#')
.and_then(|pointer| root.pointer(pointer)),
None => Some(schema),
}
}
fn schema_child_path(parent: &str, child: &str) -> String {
if parent.is_empty() {
child.to_string()
} else {
format!("{parent}.{child}")
}
}
fn schema_accepts_value_kind(schema: &serde_json::Value, value: &serde_json::Value) -> bool {
let Some(types) = schema.get("type") else {
return true;
};
match types {
serde_json::Value::String(kind) => json_value_matches_kind(value, kind),
serde_json::Value::Array(kinds) => kinds.iter().any(|kind| {
kind.as_str()
.is_some_and(|kind| json_value_matches_kind(value, kind))
}),
_ => true,
}
}
fn json_value_matches_kind(value: &serde_json::Value, kind: &str) -> bool {
match kind {
"object" => value.is_object(),
"array" => value.is_array(),
"string" => value.is_string(),
"boolean" => value.is_boolean(),
"null" => value.is_null(),
"number" => value.is_number(),
"integer" => value.as_i64().is_some() || value.as_u64().is_some(),
_ => true,
}
}
fn serialized_values_differ<T: Serialize>(current: &T, candidate: &T) -> bool {
match (
serde_json::to_value(current),
serde_json::to_value(candidate),
) {
(Ok(current), Ok(candidate)) => current != candidate,
_ => true,
}
}
fn mcp_config_changed(
current: &bitrouter_sdk::config::Config,
candidate: &bitrouter_sdk::config::Config,
) -> bool {
let current = ¤t.mcp;
let candidate = &candidate.mcp;
current.aggregate.enabled != candidate.aggregate.enabled
|| current.aggregate.route != candidate.aggregate.route
|| current.cache.enabled != candidate.cache.enabled
|| current.cache.tools_list_ttl_secs != candidate.cache.tools_list_ttl_secs
|| current.cache.resources_list_ttl_secs != candidate.cache.resources_list_ttl_secs
|| current.cache.resources_templates_list_ttl_secs
!= candidate.cache.resources_templates_list_ttl_secs
|| current.cache.prompts_list_ttl_secs != candidate.cache.prompts_list_ttl_secs
|| current.cache.max_entries_per_server != candidate.cache.max_entries_per_server
|| current.upstream_protocol != candidate.upstream_protocol
}
fn mcp_servers_changed(
current: &bitrouter_sdk::config::Config,
candidate: &bitrouter_sdk::config::Config,
) -> bool {
current.mcp_servers.len() != candidate.mcp_servers.len()
|| current.mcp_servers.iter().any(|(id, current_server)| {
candidate
.mcp_servers
.get(id)
.is_none_or(|candidate_server| {
current_server.name != candidate_server.name
|| current_server.transport != candidate_server.transport
|| current_server.aggregate != candidate_server.aggregate
|| current_server.tool_prefix != candidate_server.tool_prefix
})
})
}
fn agents_changed(
current: &bitrouter_sdk::config::Config,
candidate: &bitrouter_sdk::config::Config,
) -> bool {
current.agents.len() != candidate.agents.len()
|| current.agents.iter().any(|(id, current_agent)| {
candidate.agents.get(id).is_none_or(|candidate_agent| {
current_agent.name != candidate_agent.name
|| current_agent.transport != candidate_agent.transport
})
})
}
fn server_tools_changed(
current: &bitrouter_sdk::config::Config,
candidate: &bitrouter_sdk::config::Config,
) -> bool {
format!("{:?}", current.server_tools) != format!("{:?}", candidate.server_tools)
}
fn pricing_signature(config: &bitrouter_sdk::config::Config) -> Vec<String> {
let mut entries = Vec::new();
for (provider_id, provider) in &config.providers {
if provider
.models
.iter()
.any(|model| model.pricing.is_some() || !model.pricing_by_protocol.is_empty())
{
entries.push(format!(
"{provider_id}|endpoint|{}",
crate::metering::tariff::endpoint_profile(&provider.api_base)
));
for (protocol, endpoint) in &provider.protocol_endpoints {
entries.push(format!(
"{provider_id}|endpoint|{protocol}|{}",
crate::metering::tariff::endpoint_profile(endpoint)
));
}
}
for model in &provider.models {
if let Some(pricing) = &model.pricing {
let provider_model_id = model.provider_model_id.as_deref().unwrap_or_default();
entries.push(format!(
"{provider_id}|{}|{}|{pricing:?}",
model.id, provider_model_id
));
}
for (protocol, pricing) in &model.pricing_by_protocol {
let native_id = model.provider_model_id.as_deref().unwrap_or_default();
entries.push(format!(
"{provider_id}|{}|{native_id}|{protocol}|{pricing:?}",
model.id
));
}
}
}
entries.sort();
entries
}
fn activate_stored_credential_providers(config: &mut bitrouter_sdk::config::Config) {
if let Ok(store) = crate::provider_credentials::load_default() {
crate::providers::apply::activate_stored_credential_providers(config, &store);
}
}
async fn resolve_reloadable_config(config: &mut bitrouter_sdk::config::Config) {
crate::providers::apply::apply_builtin_defaults(config);
crate::claude_code::enable_if_logged_in(config);
crate::assemble::merge_registry_into(config).await;
activate_stored_credential_providers(config);
bitrouter_sdk::config::discover_models(config).await;
}
pub enum ReloadSource {
File(PathBuf),
Default,
}
#[derive(Clone, PartialEq, Eq)]
enum ConfigurationRevision {
Document(serde_json::Value),
Generated(u64),
}
#[derive(Clone)]
pub struct ConfigurationBaseline {
source: crate::paths::ConfigSource,
loaded_source: crate::paths::ConfigSource,
config: bitrouter_sdk::config::Config,
revision: ConfigurationRevision,
unclassified_fields: Option<BTreeSet<String>>,
unclassified_values: Option<UnclassifiedConfigValues>,
}
impl ConfigurationBaseline {
pub fn config(&self) -> &bitrouter_sdk::config::Config {
&self.config
}
}
fn config_source_kind(source: &crate::paths::ConfigSource) -> ConfigSourceKind {
match source {
crate::paths::ConfigSource::File(_) => ConfigSourceKind::File,
crate::paths::ConfigSource::Default { .. } => ConfigSourceKind::Default,
}
}
pub async fn configuration_state_without_runtime(
source: &crate::paths::ConfigSource,
) -> ConfigurationState {
let saved = match load_configuration_baseline_at(source, 0, true).await {
Ok(baseline)
if matches!(
baseline.loaded_source,
crate::paths::ConfigSource::Default { .. }
) =>
{
SavedConfigState::Generated
}
Ok(_) => SavedConfigState::Available,
Err(failure) => failure.saved,
};
ConfigurationState {
server_instance_id: None,
generation: None,
source: config_source_kind(source),
saved,
running: RunningConfigState::Unknown,
reload_required_fields: Vec::new(),
restart_required_fields: Vec::new(),
named_policy: AuxiliaryConfigState::Unknown,
access_policies: AuxiliaryConfigState::Unknown,
last_reload: None,
mixed_state_history: Vec::new(),
}
}
fn fixed_top_level_field(name: &str) -> &'static str {
match name {
"server" => "server",
"control" => "control",
"chat" => "chat",
"upstream" => "upstream",
"database" => "database",
"eval" => "eval",
"trajectory" => "trajectory",
"acp_recording" => "acp_recording",
"continuation" => "continuation",
"providers" => "providers",
"models" => "models",
"routers" => "routers",
"checkers" => "checkers",
"presets" => "presets",
"variants" => "variants",
"plugins" => "plugins",
"mcp" => "mcp",
"mcp_servers" => "mcp_servers",
"server_tools" => "server_tools",
"agents" => "agents",
"inherit_defaults" => "inherit_defaults",
"registry" => "registry",
"policy" => "policy",
"configuration_source" => "configuration_source",
_ => "unclassified",
}
}
fn changed_configuration_fields(
running: &ConfigurationRevision,
saved: &ConfigurationRevision,
) -> BTreeSet<String> {
match (running, saved) {
(ConfigurationRevision::Generated(left), ConfigurationRevision::Generated(right)) => {
if left == right {
BTreeSet::new()
} else {
BTreeSet::from(["environment".to_string()])
}
}
(ConfigurationRevision::Document(left), ConfigurationRevision::Document(right)) => {
let (Some(left), Some(right)) = (left.as_object(), right.as_object()) else {
return BTreeSet::from(["unclassified".to_string()]);
};
left.keys()
.chain(right.keys())
.filter(|key| left.get(*key) != right.get(*key))
.map(|key| fixed_top_level_field(key).to_string())
.collect()
}
_ => BTreeSet::from(["configuration_source".to_string()]),
}
}
fn fixed_restart_field(path: &str) -> String {
match path {
"server.listen"
| "server.control_socket"
| "server.log_level"
| "server.skip_auth"
| "server.require_known_pricing" => path.to_string(),
"inherit_defaults"
| "control"
| "decision_model"
| "chat"
| "database.url"
| "eval"
| "trajectory"
| "acp_recording"
| "continuation"
| "routers"
| "checkers"
| "plugins"
| "mcp"
| "mcp_servers"
| "server_tools"
| "agents"
| "upstream.fallback_backoff_ms"
| "providers.*.models.*.pricing"
| "configuration_source" => path.to_string(),
_ if path.starts_with("providers.") => "providers".to_string(),
_ => "unclassified".to_string(),
}
}
fn fixed_restart_fields(fields: Vec<String>) -> Vec<String> {
fields
.into_iter()
.map(|field| fixed_restart_field(&field))
.collect::<BTreeSet<_>>()
.into_iter()
.collect()
}
fn externally_safe_reload_report(mut report: ReloadReport) -> ReloadReport {
report.restart_required_fields = fixed_restart_fields(report.restart_required_fields);
report
}
fn restart_field_category(path: &str) -> &'static str {
if path.starts_with("server.") {
"server"
} else if path.starts_with("database.") {
"database"
} else if path.starts_with("upstream.") {
"upstream"
} else if path.starts_with("providers") {
"providers"
} else {
fixed_top_level_field(path)
}
}
struct ConfigurationLoadError {
saved: SavedConfigState,
error: anyhow::Error,
guardrails_migration_required: bool,
}
pub async fn load_configuration_baseline(
source: &crate::paths::ConfigSource,
) -> anyhow::Result<ConfigurationBaseline> {
load_configuration_baseline_at(source, 0, false)
.await
.map_err(|failure| failure.error)
}
pub(crate) async fn inspect_configuration_baseline(
source: &crate::paths::ConfigSource,
) -> anyhow::Result<ConfigurationBaseline> {
load_configuration_baseline_at(source, 0, true)
.await
.map_err(|failure| failure.error)
}
async fn load_configuration_baseline_at(
source: &crate::paths::ConfigSource,
generated_revision: u64,
inspect_default_file: bool,
) -> Result<ConfigurationBaseline, ConfigurationLoadError> {
let effective_source = match source {
crate::paths::ConfigSource::Default { home }
if inspect_default_file && !home.as_os_str().is_empty() =>
{
let candidate = home.join("bitrouter.yaml");
match tokio::fs::metadata(&candidate).await {
Ok(_) => crate::paths::ConfigSource::File(candidate),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => source.clone(),
Err(error) => {
return Err(ConfigurationLoadError {
saved: SavedConfigState::Unavailable,
error: anyhow::anyhow!(
"inspecting configuration {}: {error}",
candidate.display()
),
guardrails_migration_required: false,
});
}
}
}
crate::paths::ConfigSource::Default { .. } | crate::paths::ConfigSource::File(_) => {
source.clone()
}
};
match &effective_source {
crate::paths::ConfigSource::File(path) => {
let raw = tokio::fs::read_to_string(path).await.map_err(|error| {
let saved = if error.kind() == std::io::ErrorKind::NotFound {
SavedConfigState::Missing
} else {
SavedConfigState::Unavailable
};
ConfigurationLoadError {
saved,
error: anyhow::anyhow!("reading configuration {}: {error}", path.display()),
guardrails_migration_required: false,
}
})?;
let mut config =
bitrouter_sdk::config::parse_with(&raw, bitrouter_sdk::config::env_lookup)
.map_err(|error| ConfigurationLoadError {
saved: SavedConfigState::Invalid,
error: anyhow::Error::new(error),
guardrails_migration_required: false,
})?;
crate::policy_lock::apply_defaults(&mut config, Some(path));
let guardrails_migration_required = config.plugins.contains_key("bitrouter-guardrails");
crate::assemble::validate_host_configuration(&config).map_err(|error| {
ConfigurationLoadError {
saved: SavedConfigState::Invalid,
error,
guardrails_migration_required,
}
})?;
let substituted = bitrouter_sdk::config::substitute_env(&raw).map_err(|error| {
ConfigurationLoadError {
saved: SavedConfigState::Invalid,
error: anyhow::Error::new(error),
guardrails_migration_required: false,
}
})?;
let document =
serde_saphyr::from_str::<serde_json::Value>(&substituted).map_err(|error| {
ConfigurationLoadError {
saved: SavedConfigState::Invalid,
error: anyhow::Error::new(error),
guardrails_migration_required: false,
}
})?;
if !document.is_object() {
return Err(ConfigurationLoadError {
saved: SavedConfigState::Invalid,
error: anyhow::anyhow!("configuration must be a mapping"),
guardrails_migration_required: false,
});
}
let (unclassified_fields, unclassified_values) = unclassified_config(&document);
Ok(ConfigurationBaseline {
source: source.clone(),
loaded_source: effective_source.clone(),
config,
revision: ConfigurationRevision::Document(document),
unclassified_fields: Some(unclassified_fields),
unclassified_values: Some(unclassified_values),
})
}
crate::paths::ConfigSource::Default { .. } => {
let mut config = crate::providers::apply::zero_config();
crate::cloud::enable_in_zero_config(&mut config);
crate::policy_lock::apply_defaults(&mut config, None);
Ok(ConfigurationBaseline {
source: source.clone(),
loaded_source: effective_source.clone(),
config,
revision: ConfigurationRevision::Generated(generated_revision),
unclassified_fields: None,
unclassified_values: None,
})
}
}
}
pub struct AppReloader {
policy_store: Arc<PolicyStore>,
routing_table: Arc<bitrouter_sdk::config::ConfigRoutingTable>,
startup_config: bitrouter_sdk::config::Config,
startup_source_config: Option<bitrouter_sdk::config::Config>,
startup_configuration_source: Option<crate::paths::ConfigSource>,
startup_unclassified_values: Option<UnclassifiedConfigValues>,
running_baseline: Mutex<Option<ConfigurationBaseline>>,
environment_revision: AtomicU64,
upstream_executor: Arc<bitrouter_sdk::language_model::HttpExecutor>,
policy_runtime: Option<Arc<crate::policy_lock::PolicyRuntime>>,
coordinator: Arc<ReloadCoordinator>,
source: ReloadSource,
#[cfg(test)]
test_apply_failure: Option<ReloadParticipant>,
#[cfg(test)]
test_pause_after_prepare: Option<Arc<TestPreparationPause>>,
#[cfg(test)]
test_pause_during_prepare: Option<Arc<TestPreparationPause>>,
#[cfg(test)]
test_preparation_timeout: Option<Duration>,
}
#[derive(Clone, Copy)]
enum ReloadInvocation {
Local,
Remote,
}
#[cfg(test)]
struct TestPreparationPause {
prepared: tokio::sync::Notify,
resume: tokio::sync::Notify,
}
#[cfg(test)]
impl TestPreparationPause {
fn new() -> Self {
Self {
prepared: tokio::sync::Notify::new(),
resume: tokio::sync::Notify::new(),
}
}
}
struct PreparedReload {
baseline: ConfigurationBaseline,
config: bitrouter_sdk::config::Config,
timeout_clients: bitrouter_sdk::language_model::executor::PreparedProviderTimeouts,
named_policy_runtime: Option<crate::policy_lock::PreparedPolicySnapshot>,
access_policy_store: Option<crate::policy::store::PreparedPolicyStore>,
}
struct PreparationError {
participant: ReloadParticipant,
failure: ReloadFailure,
restart_required_fields: Vec<String>,
}
impl PreparationError {
fn failed(participant: ReloadParticipant, code: &str, message: &str) -> Self {
Self {
participant,
failure: ReloadFailure::new(code, message),
restart_required_fields: Vec::new(),
}
}
fn restart_required(fields: Vec<String>) -> Self {
Self {
participant: ReloadParticipant::RoutingTable,
failure: ReloadFailure::new(
"restart_required",
"candidate changes a field that is only applied at daemon startup",
),
restart_required_fields: fields,
}
}
}
impl AppReloader {
pub fn new(
policy_store: Arc<PolicyStore>,
routing_table: Arc<bitrouter_sdk::config::ConfigRoutingTable>,
upstream_executor: Arc<bitrouter_sdk::language_model::HttpExecutor>,
source: ReloadSource,
) -> Self {
let startup_config = routing_table.snapshot_config();
Self {
policy_store,
routing_table,
startup_config,
startup_source_config: None,
startup_configuration_source: None,
startup_unclassified_values: None,
running_baseline: Mutex::new(None),
environment_revision: AtomicU64::new(0),
upstream_executor,
policy_runtime: None,
coordinator: ReloadCoordinator::new(),
source,
#[cfg(test)]
test_apply_failure: None,
#[cfg(test)]
test_pause_after_prepare: None,
#[cfg(test)]
test_pause_during_prepare: None,
#[cfg(test)]
test_preparation_timeout: None,
}
}
pub fn with_startup_configuration(mut self, baseline: ConfigurationBaseline) -> Self {
self.startup_source_config = Some(baseline.config.clone());
self.startup_configuration_source = Some(baseline.source.clone());
self.startup_unclassified_values = baseline.unclassified_values.clone();
self.running_baseline = Mutex::new(Some(baseline));
self
}
#[cfg(test)]
fn with_test_apply_failure(mut self, participant: ReloadParticipant) -> Self {
self.test_apply_failure = Some(participant);
self
}
#[cfg(test)]
fn with_test_pause_after_prepare(mut self, pause: Arc<TestPreparationPause>) -> Self {
self.test_pause_after_prepare = Some(pause);
self
}
#[cfg(test)]
fn with_test_pause_during_prepare(mut self, pause: Arc<TestPreparationPause>) -> Self {
self.test_pause_during_prepare = Some(pause);
self
}
#[cfg(test)]
fn with_test_preparation_timeout(mut self, timeout: Duration) -> Self {
self.test_preparation_timeout = Some(timeout);
self
}
fn should_fail_apply(&self, participant: ReloadParticipant) -> bool {
#[cfg(test)]
{
self.test_apply_failure == Some(participant)
}
#[cfg(not(test))]
{
let _ = participant;
false
}
}
async fn wait_after_prepare(&self) {
#[cfg(test)]
if let Some(pause) = &self.test_pause_after_prepare {
pause.prepared.notify_one();
pause.resume.notified().await;
}
}
async fn wait_during_prepare(&self) {
#[cfg(test)]
if let Some(pause) = &self.test_pause_during_prepare {
pause.prepared.notify_one();
pause.resume.notified().await;
}
}
fn preparation_timeout(&self) -> Duration {
#[cfg(test)]
if let Some(timeout) = self.test_preparation_timeout {
return timeout;
}
Duration::from_secs(PREPARATION_TIMEOUT_SECONDS)
}
async fn prepare_candidate(&self) -> Result<ConfigurationBaseline, PreparationError> {
let source = self.source_for_inspection();
let revision = self.environment_revision.load(Ordering::Acquire);
load_configuration_baseline_at(&source, revision, true)
.await
.map_err(|failure| {
tracing::warn!(error = %failure.error, "reload could not prepare configuration candidate");
if failure.guardrails_migration_required {
return PreparationError::failed(
ReloadParticipant::RoutingTable,
"guardrails_migration_required",
"saved configuration contains plugins.bitrouter-guardrails; migration is required before reload because router-bound request checkers cover input only and do not replace global or output protection",
);
}
let message = match failure.saved {
SavedConfigState::Missing | SavedConfigState::Unavailable => {
"reload could not read the server configuration"
}
_ => "reload configuration is invalid",
};
PreparationError::failed(
ReloadParticipant::RoutingTable,
"config_prepare_failed",
message,
)
})
}
async fn prepare_resolved_candidate(&self) -> Result<ConfigurationBaseline, PreparationError> {
let mut baseline = self.prepare_candidate().await?;
if config_source_kind(&baseline.source) != config_source_kind(&baseline.loaded_source) {
return Err(PreparationError::restart_required(vec![
"configuration_source".to_string(),
]));
}
resolve_reloadable_config(&mut baseline.config).await;
Ok(baseline)
}
async fn prepare(
&self,
_invocation: ReloadInvocation,
) -> Result<PreparedReload, PreparationError> {
self.wait_during_prepare().await;
let baseline = self.prepare_resolved_candidate().await?;
let config = baseline.config.clone();
let changed_unclassified = changed_unclassified_config_paths(
self.startup_unclassified_values.as_ref(),
baseline.unclassified_values.as_ref(),
baseline.unclassified_fields.as_ref(),
);
let restart_required =
restart_required_fields(&self.startup_config, &config, Some(&changed_unclassified));
if !restart_required.is_empty() {
return Err(PreparationError::restart_required(restart_required));
}
let named_policy_runtime = match &self.policy_runtime {
Some(runtime) => {
let path = match &self.source {
ReloadSource::File(path) => Some(path.as_path()),
ReloadSource::Default => None,
};
Some(runtime.prepare_for_config(&config, path).await.map_err(|error| {
tracing::warn!(error = %error, "reload named policy preparation failed");
PreparationError::failed(
ReloadParticipant::NamedPolicyRuntime,
"policy_prepare_failed",
"named policy preparation failed",
)
})?)
}
None => None,
};
let (global_timeouts, provider_timeouts) =
crate::assemble::resolved_upstream_timeouts(&config);
let timeout_clients = self
.upstream_executor
.prepare_provider_timeouts(global_timeouts, provider_timeouts)
.map_err(|error| {
tracing::warn!(error = %error, "reload timeout client preparation failed");
PreparationError::failed(
ReloadParticipant::UpstreamTimeoutClients,
"timeout_prepare_failed",
"upstream timeout client preparation failed",
)
})?;
let access_policy_store = self.policy_store.prepare_reload().await.map_err(|error| {
tracing::warn!(error = %error, "reload access-policy preparation failed");
PreparationError::failed(
ReloadParticipant::AccessPolicyStore,
"access_policy_prepare_failed",
"access-policy preparation failed",
)
})?;
Ok(PreparedReload {
baseline,
config,
timeout_clients,
named_policy_runtime,
access_policy_store,
})
}
async fn execute_reservation(
&self,
reservation: ReloadReservation,
invocation: ReloadInvocation,
env: Vec<(String, String)>,
) -> ReloadReport {
let mut report = ReloadReport::pending(
reservation.server_instance_id.clone(),
reservation.generation,
);
reservation.update_progress(&report);
if !reservation.belongs_to(&self.coordinator) {
report.participant_failed(
ReloadParticipant::RoutingTable,
ReloadFailure::new(
"invalid_reservation",
"reload reservation does not belong to this daemon",
),
);
report.finish(ReloadOutcome::Failed);
reservation.update_progress(&report);
return report;
}
if matches!(invocation, ReloadInvocation::Local) && !env.is_empty() {
let overrides = env.into_iter().collect();
bitrouter_sdk::config::set_env_overrides(overrides);
let mut revision = self.environment_revision.load(Ordering::Acquire);
loop {
match self.environment_revision.compare_exchange_weak(
revision,
revision.saturating_add(1),
Ordering::AcqRel,
Ordering::Acquire,
) {
Ok(_) => break,
Err(current) => revision = current,
}
}
tracing::info!("env override map updated by local reload");
}
let prepared = match tokio::time::timeout(
self.preparation_timeout(),
self.prepare(invocation),
)
.await
{
Ok(Ok(prepared)) => prepared,
Ok(Err(error)) => {
report.participant_failed(error.participant, error.failure);
report.restart_required_fields =
fixed_restart_fields(error.restart_required_fields);
report.finish(ReloadOutcome::Failed);
reservation.update_progress(&report);
self.coordinator.complete(&reservation, report.clone());
return report;
}
Err(_) => {
report.participant_failed(
ReloadParticipant::RoutingTable,
ReloadFailure::new(
"preparation_timed_out",
"reload preparation exceeded its time limit",
),
);
report.finish(ReloadOutcome::Failed);
reservation.update_progress(&report);
self.coordinator.complete(&reservation, report.clone());
return report;
}
};
self.wait_after_prepare().await;
let applied_baseline = prepared.baseline.clone();
reservation.mark_mutation_started();
if self.should_fail_apply(ReloadParticipant::RoutingTable) {
report.participant_failed(
ReloadParticipant::RoutingTable,
ReloadFailure::new("fault_injected", "routing table update failed"),
);
} else {
match self
.routing_table
.replace_prepared_config(prepared.config)
.await
{
Ok(()) => report.participant_applied(ReloadParticipant::RoutingTable),
Err(error) => {
tracing::warn!(error = %error, "reload routing table swap failed");
report.participant_failed(
ReloadParticipant::RoutingTable,
ReloadFailure::new("routing_apply_failed", "routing table update failed"),
);
}
}
}
reservation.update_progress(&report);
if self.should_fail_apply(ReloadParticipant::UpstreamTimeoutClients) {
report.participant_failed(
ReloadParticipant::UpstreamTimeoutClients,
ReloadFailure::new("fault_injected", "upstream timeout client update failed"),
);
} else {
self.upstream_executor
.commit_provider_timeouts(prepared.timeout_clients);
report.participant_applied(ReloadParticipant::UpstreamTimeoutClients);
}
reservation.update_progress(&report);
if self.should_fail_apply(ReloadParticipant::NamedPolicyRuntime) {
report.participant_failed(
ReloadParticipant::NamedPolicyRuntime,
ReloadFailure::new("fault_injected", "named policy runtime update failed"),
);
} else {
match (self.policy_runtime.as_ref(), prepared.named_policy_runtime) {
(Some(runtime), Some(prepared_runtime)) => {
runtime.commit(prepared_runtime);
report.participant_applied(ReloadParticipant::NamedPolicyRuntime);
}
_ => report.participant_unchanged(ReloadParticipant::NamedPolicyRuntime),
}
}
reservation.update_progress(&report);
if self.should_fail_apply(ReloadParticipant::AccessPolicyStore) {
report.participant_failed(
ReloadParticipant::AccessPolicyStore,
ReloadFailure::new("fault_injected", "access-policy store update failed"),
);
} else {
match prepared.access_policy_store {
Some(prepared_store) => match self.policy_store.commit_prepared(prepared_store) {
Ok(()) => report.participant_applied(ReloadParticipant::AccessPolicyStore),
Err(error) => {
tracing::warn!(error = %error, "reload access-policy store swap failed");
report.participant_failed(
ReloadParticipant::AccessPolicyStore,
ReloadFailure::new(
"access_policy_apply_failed",
"access-policy store update failed",
),
);
}
},
None => report.participant_unchanged(ReloadParticipant::AccessPolicyStore),
}
}
let failed = report
.participants
.iter()
.any(|entry| entry.outcome == ReloadParticipantOutcome::Failed);
let outcome = if !failed {
ReloadOutcome::Succeeded
} else if report.any_applied() {
ReloadOutcome::PartiallyApplied
} else {
ReloadOutcome::Failed
};
if outcome == ReloadOutcome::Succeeded {
let mut running = match self.running_baseline.lock() {
Ok(running) => running,
Err(poisoned) => poisoned.into_inner(),
};
*running = Some(applied_baseline);
}
report.finish(outcome);
reservation.update_progress(&report);
self.coordinator.complete(&reservation, report.clone());
report
}
fn source_for_inspection(&self) -> crate::paths::ConfigSource {
self.startup_configuration_source
.clone()
.unwrap_or_else(|| match &self.source {
ReloadSource::File(path) => crate::paths::ConfigSource::File(path.clone()),
ReloadSource::Default => crate::paths::ConfigSource::Default {
home: PathBuf::new(),
},
})
}
async fn named_policy_configuration_state(
&self,
baseline: &ConfigurationBaseline,
) -> AuxiliaryConfigState {
let Some(runtime) = &self.policy_runtime else {
return AuxiliaryConfigState::NotConfigured;
};
let path = match &baseline.loaded_source {
crate::paths::ConfigSource::File(path) => Some(path.as_path()),
crate::paths::ConfigSource::Default { .. } => None,
};
let loaded = match crate::policy_lock::load_for_config(&baseline.config, path).await {
Ok(loaded) => loaded,
Err(_) => {
let Some(policy_path) = crate::policy_lock::resolve_path(&baseline.config, path)
else {
return AuxiliaryConfigState::Invalid;
};
return match tokio::fs::read_to_string(policy_path).await {
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
AuxiliaryConfigState::Missing
}
Err(_) => AuxiliaryConfigState::Unavailable,
Ok(_) => AuxiliaryConfigState::Invalid,
};
}
};
let active = runtime.administration_snapshot();
let disk = crate::actions::administration::PolicyReport::from_loaded(
&baseline.config,
loaded.as_ref(),
crate::actions::administration::PolicyView::Disk,
);
if active.availability == "not_configured" && disk.availability == "not_configured" {
AuxiliaryConfigState::NotConfigured
} else if active.availability == disk.availability && active.digest == disk.digest {
AuxiliaryConfigState::InSync
} else {
AuxiliaryConfigState::ReloadRequired
}
}
async fn access_policy_configuration_state(&self) -> AuxiliaryConfigState {
use crate::policy::store::PolicySourceState;
match self.policy_store.source_state().await {
PolicySourceState::NotConfigured => AuxiliaryConfigState::NotConfigured,
PolicySourceState::InSync => AuxiliaryConfigState::InSync,
PolicySourceState::Changed => AuxiliaryConfigState::ReloadRequired,
PolicySourceState::Missing => AuxiliaryConfigState::Missing,
PolicySourceState::Invalid => AuxiliaryConfigState::Invalid,
PolicySourceState::Unavailable => AuxiliaryConfigState::Unavailable,
}
}
async fn inspect_configuration_state(&self) -> ConfigurationState {
let before = self.coordinator.state();
let source = self.source_for_inspection();
let generated_revision = self.environment_revision.load(Ordering::Acquire);
let saved_baseline =
load_configuration_baseline_at(&source, generated_revision, true).await;
let (
saved,
mut running,
mut reload_fields,
mut restart_fields,
named_policy,
access_policies,
) = match saved_baseline {
Err(failure) => (
failure.saved,
RunningConfigState::Unknown,
Vec::new(),
Vec::new(),
AuxiliaryConfigState::Unknown,
AuxiliaryConfigState::Unknown,
),
Ok(saved_baseline) => {
let source_changed = config_source_kind(&source)
!= config_source_kind(&saved_baseline.loaded_source);
let saved = if matches!(
&saved_baseline.loaded_source,
crate::paths::ConfigSource::Default { .. }
) {
SavedConfigState::Generated
} else {
SavedConfigState::Available
};
let named_policy = self.named_policy_configuration_state(&saved_baseline).await;
let access_policies = self.access_policy_configuration_state().await;
let running_baseline = match self.running_baseline.lock() {
Ok(running) => running.clone(),
Err(poisoned) => poisoned.into_inner().clone(),
};
let Some(running_baseline) = running_baseline else {
return ConfigurationState {
server_instance_id: Some(before.server_instance_id),
generation: Some(before.generation),
source: config_source_kind(&source),
saved,
running: RunningConfigState::Unknown,
reload_required_fields: Vec::new(),
restart_required_fields: Vec::new(),
named_policy,
access_policies,
last_reload: before.last_outcome.map(externally_safe_reload_report),
mixed_state_history: before
.mixed_state_history
.into_iter()
.map(externally_safe_reload_report)
.collect(),
};
};
let mut changed = changed_configuration_fields(
&running_baseline.revision,
&saved_baseline.revision,
);
let raw_restart_fields = if source_changed {
vec!["configuration_source".to_string()]
} else {
let changed_unclassified = changed_unclassified_config_paths(
self.startup_unclassified_values.as_ref(),
saved_baseline.unclassified_values.as_ref(),
saved_baseline.unclassified_fields.as_ref(),
);
self.startup_source_config
.as_ref()
.map(|startup| {
restart_required_fields(
startup,
&saved_baseline.config,
Some(&changed_unclassified),
)
})
.unwrap_or_default()
};
for field in &raw_restart_fields {
let _ = changed.remove(restart_field_category(field));
}
let restart_fields = fixed_restart_fields(raw_restart_fields);
let mut reload_fields = changed.into_iter().collect::<Vec<_>>();
if named_policy == AuxiliaryConfigState::ReloadRequired {
reload_fields.push("policy_lock".to_string());
}
if access_policies == AuxiliaryConfigState::ReloadRequired {
reload_fields.push("access_policies".to_string());
}
reload_fields.sort();
reload_fields.dedup();
let auxiliary_unknown = [named_policy, access_policies].into_iter().any(|state| {
matches!(
state,
AuxiliaryConfigState::Missing
| AuxiliaryConfigState::Invalid
| AuxiliaryConfigState::Unavailable
| AuxiliaryConfigState::Unknown
)
});
let running = if !restart_fields.is_empty() {
RunningConfigState::RestartRequired
} else if !reload_fields.is_empty() {
RunningConfigState::ReloadRequired
} else if auxiliary_unknown {
RunningConfigState::Unknown
} else {
RunningConfigState::InSync
};
(
saved,
running,
reload_fields,
restart_fields,
named_policy,
access_policies,
)
}
};
let after = self.coordinator.state();
if before.consistency == ReloadConsistency::Mixed
|| after.consistency == ReloadConsistency::Mixed
{
running = RunningConfigState::Mixed;
} else if before.running
|| after.running
|| before.server_instance_id != after.server_instance_id
|| before.generation != after.generation
{
running = RunningConfigState::Unknown;
reload_fields.clear();
restart_fields.clear();
}
ConfigurationState {
server_instance_id: Some(after.server_instance_id),
generation: Some(after.generation),
source: config_source_kind(&source),
saved,
running,
reload_required_fields: reload_fields,
restart_required_fields: restart_fields,
named_policy,
access_policies,
last_reload: after.last_outcome.map(externally_safe_reload_report),
mixed_state_history: after
.mixed_state_history
.into_iter()
.map(externally_safe_reload_report)
.collect(),
}
}
pub fn with_policy_runtime(mut self, runtime: Arc<crate::policy_lock::PolicyRuntime>) -> Self {
self.policy_runtime = Some(runtime);
self
}
}
#[async_trait::async_trait]
impl DaemonReloader for AppReloader {
async fn reload(&self) -> anyhow::Result<()> {
self.reload_with_env(Vec::new()).await
}
async fn reload_with_env(&self, env: Vec<(String, String)>) -> anyhow::Result<()> {
let reservation = self
.coordinator
.reserve_local()
.map_err(anyhow::Error::from)?;
let report = self
.execute_reservation(reservation, ReloadInvocation::Local, env)
.await;
if report.outcome == ReloadOutcome::Succeeded {
Ok(())
} else {
Err(anyhow::anyhow!(report.safe_summary()))
}
}
fn reload_state(&self) -> Option<ReloadState> {
Some(self.coordinator.state())
}
async fn configuration_state(&self) -> Option<ConfigurationState> {
Some(self.inspect_configuration_state().await)
}
fn reserve_remote(
&self,
expected_instance: &str,
expected_generation: u64,
) -> Result<ReloadReservation, ReloadAdmissionError> {
self.coordinator
.reserve_remote(expected_instance, expected_generation)
}
async fn reload_reserved(&self, reservation: ReloadReservation) -> ReloadReport {
self.execute_reservation(reservation, ReloadInvocation::Remote, Vec::new())
.await
}
}
#[cfg(test)]
mod tests {
use super::*;
use bitrouter_ai::client::HttpTimeouts;
use bitrouter_ai::types::{ApiProtocol, GenerationParams, Message, Prompt, Role};
use bitrouter_sdk::config::{self, ConfigRoutingTable};
use bitrouter_sdk::language_model::{
Executor, HttpExecutor, PipelineContext, PipelineRequest, RoutingTarget,
};
use std::time::Duration;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
fn temp_config_path() -> (std::path::PathBuf, std::path::PathBuf) {
let dir = tempfile::Builder::new()
.prefix("bitrouter-reload-")
.tempdir()
.expect("create temp config dir")
.keep();
(dir.join("bitrouter.yaml"), dir)
}
fn config_yaml(read_secs: u64) -> String {
format!(
r#"
inherit_defaults: false
upstream:
timeouts:
read_secs: {read_secs}
providers:
slow:
api_base: https://api.example.com/v1
api_key: k
api_protocol:
- "*": chat_completions
models:
- {{ id: m }}
"#
)
}
async fn stalled_json_server() -> String {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
.await
.expect("bind test server");
let addr = listener.local_addr().expect("test server addr");
tokio::spawn(async move {
let (mut socket, _) = listener.accept().await.expect("accept request");
let mut request_buf = [0_u8; 1024];
let _ = socket.read(&mut request_buf).await;
socket
.write_all(
b"HTTP/1.1 200 OK\r\n\
content-type: application/json\r\n\
content-length: 1024\r\n\
\r\n\
{\"id\":\"partial\"",
)
.await
.expect("write partial response");
tokio::time::sleep(Duration::from_secs(5)).await;
});
format!("http://{addr}/v1")
}
fn prompt() -> Prompt {
Prompt {
model: "m".into(),
system: None,
system_provider_metadata: Default::default(),
messages: vec![Message::text(Role::User, "hi")],
tools: vec![],
params: GenerationParams::default(),
response_format: None,
tool_choice: None,
stream: false,
}
}
struct FaultFixture {
_home: tempfile::TempDir,
reloader: Arc<AppReloader>,
routing_table: Arc<ConfigRoutingTable>,
policy_runtime: Arc<crate::policy_lock::PolicyRuntime>,
policy_store: Arc<PolicyStore>,
initial_policy_digest: String,
candidate_policy_digest: String,
config_path: std::path::PathBuf,
lock_path: std::path::PathBuf,
access_policy_dir: std::path::PathBuf,
}
fn coordinated_config_yaml(model: &str, read_secs: u64) -> String {
format!(
r#"inherit_defaults: false
upstream:
timeouts:
read_secs: {read_secs}
providers:
alpha:
api_base: https://alpha.example.com/v1
api_key: alpha-key
api_protocol:
- "*": chat_completions
models:
- {{ id: m }}
beta:
api_base: https://beta.example.com/v1
api_key: beta-key
api_protocol:
- "*": chat_completions
models:
- {{ id: m }}
presets:
coding:
model: {model}
policy: coding
"#
)
}
fn coordinated_policy_lock_yaml(model: &str) -> String {
format!(
r#"lockfileVersion: 4
artifact:
evidence_root: "sha256:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
source_snapshot_time_unix_ms: 0
compiler:
id: fixture
version: 1
config_digest: "sha256:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
policies:
coding:
tiers: {{ strong: {{ model: "{model}", context: evidence }} }}
default_tier: strong
tool_use_tier: strong
tool_safe_tiers: [strong]
"#
)
}
async fn fault_fixture(failure: ReloadParticipant) -> anyhow::Result<FaultFixture> {
coordinated_fixture(Some(failure), None).await
}
async fn coordinated_fixture(
failure: Option<ReloadParticipant>,
pause: Option<Arc<TestPreparationPause>>,
) -> anyhow::Result<FaultFixture> {
let home = tempfile::tempdir()?;
let config_path = home.path().join("bitrouter.yaml");
let lock_path = home.path().join("policy-lock.yaml");
let access_policy_dir = home.path().join("access-policies");
std::fs::create_dir(&access_policy_dir)?;
std::fs::write(&config_path, coordinated_config_yaml("alpha:m", 30))?;
std::fs::write(&lock_path, coordinated_policy_lock_yaml("alpha:m"))?;
std::fs::write(
access_policy_dir.join("operator.yaml"),
"id: operator\nallowed_models: [alpha:m]\n",
)?;
let initial_baseline =
load_configuration_baseline(&crate::paths::ConfigSource::File(config_path.clone()))
.await?;
let mut initial = initial_baseline.config.clone();
resolve_reloadable_config(&mut initial).await;
let routing_table = Arc::new(ConfigRoutingTable::from_config(initial.clone()));
let executor = Arc::new(HttpExecutor::new(HttpTimeouts::default())?);
let db = crate::db::connect("sqlite::memory:").await?;
crate::db::run_migrations(&db).await?;
let policy_runtime = crate::policy_lock::PolicyRuntime::new(
&initial,
Some(&config_path),
None,
crate::eval::settlement::PendingEvalDecisionStore::default(),
None,
)
.await?;
let initial_policy_digest = policy_runtime
.administration_snapshot()
.digest
.ok_or_else(|| anyhow::anyhow!("initial active policy has no digest"))?;
let policy_store = Arc::new(PolicyStore::load_dir(&access_policy_dir).await?);
std::fs::write(&config_path, coordinated_config_yaml("beta:m", 1))?;
std::fs::write(&lock_path, coordinated_policy_lock_yaml("beta:m"))?;
std::fs::write(
access_policy_dir.join("operator.yaml"),
"id: operator\nallowed_models: [beta:m]\n",
)?;
let mut candidate = config::load(&config_path).await?;
resolve_reloadable_config(&mut candidate).await;
let candidate_policy_digest =
crate::policy_lock::load_for_config(&candidate, Some(&config_path))
.await?
.map(|loaded| loaded.digest)
.ok_or_else(|| anyhow::anyhow!("candidate policy has no digest"))?;
let mut reloader = AppReloader::new(
policy_store.clone(),
routing_table.clone(),
executor,
ReloadSource::File(config_path.clone()),
)
.with_startup_configuration(initial_baseline)
.with_policy_runtime(policy_runtime.clone());
if let Some(failure) = failure {
reloader = reloader.with_test_apply_failure(failure);
}
if let Some(pause) = pause {
reloader = reloader.with_test_pause_after_prepare(pause);
}
Ok(FaultFixture {
_home: home,
reloader: Arc::new(reloader),
routing_table,
policy_runtime,
policy_store,
initial_policy_digest,
candidate_policy_digest,
config_path,
lock_path,
access_policy_dir,
})
}
#[tokio::test]
async fn reload_updates_live_upstream_timeout_clients() -> anyhow::Result<()> {
let (path, dir) = temp_config_path();
std::fs::write(&path, config_yaml(30))?;
let mut initial = config::parse(&config_yaml(30))?;
resolve_reloadable_config(&mut initial).await;
let routing_table = Arc::new(ConfigRoutingTable::from_config(initial));
let executor = Arc::new(HttpExecutor::new(HttpTimeouts {
read: Duration::from_secs(30),
..HttpTimeouts::default()
})?);
let reloader = AppReloader::new(
Arc::new(PolicyStore::new()),
routing_table,
executor.clone(),
ReloadSource::File(path.clone()),
);
std::fs::write(&path, config_yaml(1))?;
reloader.reload().await?;
let api_base = stalled_json_server().await;
let target = RoutingTarget {
provider_name: "slow".into(),
service_id: "m".into(),
api_base,
api_key: "k".into(),
api_protocol: ApiProtocol::ChatCompletions,
chat_token_limit_field: None,
chat_supports_store: None,
chat_supports_stream_options: None,
chat_google_extensions: false,
reasoning_effort: None,
model_constraints: Default::default(),
account_label: None,
api_key_override: None,
api_base_override: None,
auth_scheme: Default::default(),
headers: Vec::new(),
};
let prompt = prompt();
let ctx = PipelineContext::new(PipelineRequest::new(
"m",
bitrouter_sdk::caller::CallerContext::local(),
prompt.clone(),
));
let result = tokio::time::timeout(
Duration::from_secs(3),
executor.execute(&target, &prompt, &ctx),
)
.await?;
std::fs::remove_dir_all(dir).ok();
match result {
Err(bitrouter_sdk::BitrouterError::UpstreamTimeout) => Ok(()),
Err(other) => Err(anyhow::anyhow!("expected UpstreamTimeout, got {other:?}")),
Ok(_) => Err(anyhow::anyhow!("stalled body unexpectedly completed")),
}
}
#[tokio::test]
async fn invalid_policy_candidate_does_not_swap_routing_or_policy() {
use crate::policy_lock::PolicyRuntime;
use bitrouter_sdk::config::PolicyRuntimeMode;
let (path, dir) = temp_config_path();
let config_yaml = |model: &str| {
format!(
r#"inherit_defaults: false
presets:
coding:
model: {model}
policy: coding
"#
)
};
std::fs::write(&path, config_yaml("vendor:old")).expect("write initial config");
std::fs::write(
dir.join("policy-lock.yaml"),
r#"lockfileVersion: 4
artifact:
evidence_root: "sha256:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
source_snapshot_time_unix_ms: 0
compiler:
id: bitrouter-policy-compiler
version: 1
config_digest: "sha256:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
policies:
coding:
key_strategy: agent_trace
tiers: { strong: { model: vendor:old, context: evidence } }
routes: {}
default_tier: strong
tool_use_tier: strong
tool_safe_tiers: [strong]
"#,
)
.expect("write initial policy");
let initial = config::load(&path).await.expect("load initial config");
let routing_table = Arc::new(ConfigRoutingTable::from_config(initial.clone()));
let db = crate::db::connect("sqlite::memory:")
.await
.expect("connect db");
crate::db::run_migrations(&db).await.expect("migrate db");
let runtime = PolicyRuntime::new(
&initial,
Some(&path),
None,
crate::eval::settlement::PendingEvalDecisionStore::default(),
None,
)
.await
.expect("build policy runtime");
let initial_digest = runtime
.status(PolicyRuntimeMode::Frozen)
.digest
.expect("initial digest");
let executor = Arc::new(HttpExecutor::new(HttpTimeouts::default()).expect("executor"));
let reloader = AppReloader::new(
Arc::new(PolicyStore::new()),
routing_table.clone(),
executor,
ReloadSource::File(path.clone()),
)
.with_policy_runtime(runtime.clone());
std::fs::write(&path, config_yaml("vendor:new")).expect("write candidate config");
std::fs::write(dir.join("policy-lock.yaml"), "lockfileVersion: broken\n")
.expect("break candidate policy");
assert!(reloader.reload().await.is_err());
assert_eq!(
routing_table.snapshot_config().presets["coding"]
.model
.as_deref(),
Some("vendor:old")
);
assert_eq!(
runtime.status(PolicyRuntimeMode::Frozen).digest.as_deref(),
Some(initial_digest.as_str())
);
std::fs::remove_dir_all(dir).ok();
}
#[tokio::test]
async fn fault_boundaries_report_each_participant_and_preserve_active_policy_truth()
-> anyhow::Result<()> {
for failed_participant in ReloadParticipant::all() {
let fixture = fault_fixture(failed_participant).await?;
let initial_state = fixture
.reloader
.reload_state()
.ok_or_else(|| anyhow::anyhow!("coordinator state is unavailable"))?;
let reservation = fixture
.reloader
.reserve_remote(&initial_state.server_instance_id, initial_state.generation)?;
let report = fixture.reloader.reload_reserved(reservation).await;
assert_eq!(report.outcome, ReloadOutcome::PartiallyApplied);
assert_eq!(
report
.participants
.iter()
.filter(|entry| entry.outcome == ReloadParticipantOutcome::Applied)
.count(),
3,
"every non-failed participant should have made its prepared change"
);
for entry in &report.participants {
if entry.participant == failed_participant {
assert_eq!(entry.outcome, ReloadParticipantOutcome::Failed);
assert_eq!(
entry.error.as_ref().map(|error| error.code.as_str()),
Some("fault_injected")
);
} else {
assert_eq!(
entry.outcome,
ReloadParticipantOutcome::Applied,
"{failed_participant:?} must not obscure {participant:?}",
participant = entry.participant
);
}
}
let state = fixture
.reloader
.reload_state()
.ok_or_else(|| anyhow::anyhow!("coordinator state disappeared"))?;
assert!(!state.running);
assert_eq!(state.generation, 1);
assert_eq!(state.consistency, ReloadConsistency::Mixed);
assert_eq!(state.mixed_state_history.len(), 1);
assert_eq!(
state.mixed_state_history[0].outcome,
ReloadOutcome::PartiallyApplied
);
assert_eq!(
state.mixed_state_history[0].participants, report.participants,
"the retained mixed-state record must preserve the actual participant outcomes"
);
let configuration = fixture
.reloader
.configuration_state()
.await
.ok_or_else(|| anyhow::anyhow!("configuration state is unavailable"))?;
assert_eq!(configuration.running, RunningConfigState::Mixed);
let expected_routing_model = if failed_participant == ReloadParticipant::RoutingTable {
"alpha:m"
} else {
"beta:m"
};
let active_config = fixture.routing_table.snapshot_config();
let routed_model = active_config
.presets
.get("coding")
.and_then(|preset| preset.model.as_deref())
.ok_or_else(|| anyhow::anyhow!("coding preset is absent after reload"))?;
assert_eq!(routed_model, expected_routing_model);
let expected_access_model =
if failed_participant == ReloadParticipant::AccessPolicyStore {
"alpha:m"
} else {
"beta:m"
};
let access_model = fixture.policy_store.with_policy("operator", |policy| {
policy
.and_then(|policy| policy.allowed_models.as_ref())
.and_then(|models| models.first())
.cloned()
});
assert_eq!(access_model.as_deref(), Some(expected_access_model));
let expected_policy_digest =
if failed_participant == ReloadParticipant::NamedPolicyRuntime {
fixture.initial_policy_digest.as_str()
} else {
fixture.candidate_policy_digest.as_str()
};
assert_eq!(
fixture
.policy_runtime
.administration_snapshot()
.digest
.as_deref(),
Some(expected_policy_digest),
"the active policy report must describe the policy snapshot that actually committed"
);
}
Ok(())
}
#[tokio::test]
async fn later_preparation_failure_keeps_the_mixed_cause_in_configuration_state()
-> anyhow::Result<()> {
let fixture = fault_fixture(ReloadParticipant::NamedPolicyRuntime).await?;
assert!(fixture.reloader.reload().await.is_err());
std::fs::write(&fixture.config_path, "providers: [invalid\n")?;
assert!(fixture.reloader.reload().await.is_err());
let state = fixture
.reloader
.configuration_state()
.await
.ok_or_else(|| anyhow::anyhow!("configuration state is unavailable"))?;
assert_eq!(state.running, RunningConfigState::Mixed);
assert_eq!(
state.last_reload.as_ref().map(|report| report.outcome),
Some(ReloadOutcome::Failed)
);
assert_eq!(state.mixed_state_history.len(), 1);
assert_eq!(
state.mixed_state_history[0].outcome,
ReloadOutcome::PartiallyApplied
);
assert!(
state.mixed_state_history[0]
.participants
.iter()
.any(|entry| {
entry.participant == ReloadParticipant::NamedPolicyRuntime
&& entry.outcome == ReloadParticipantOutcome::Failed
})
);
Ok(())
}
#[tokio::test]
async fn prepared_inputs_are_not_re_read_after_source_changes() -> anyhow::Result<()> {
let pause = Arc::new(TestPreparationPause::new());
let fixture = coordinated_fixture(None, Some(pause.clone())).await?;
let state = fixture
.reloader
.reload_state()
.ok_or_else(|| anyhow::anyhow!("coordinator state is unavailable"))?;
let reservation = fixture
.reloader
.reserve_remote(&state.server_instance_id, state.generation)?;
let reloader = fixture.reloader.clone();
let task = tokio::spawn(async move { reloader.reload_reserved(reservation).await });
tokio::time::timeout(Duration::from_secs(3), pause.prepared.notified()).await?;
std::fs::write(&fixture.config_path, coordinated_config_yaml("alpha:m", 30))?;
std::fs::write(&fixture.lock_path, coordinated_policy_lock_yaml("alpha:m"))?;
std::fs::write(
fixture.access_policy_dir.join("operator.yaml"),
"id: operator\nallowed_models: [alpha:m]\n",
)?;
pause.resume.notify_one();
let report = tokio::time::timeout(Duration::from_secs(3), task).await??;
assert_eq!(report.outcome, ReloadOutcome::Succeeded);
assert!(
report
.participants
.iter()
.all(|entry| entry.outcome == ReloadParticipantOutcome::Applied)
);
let active_config = fixture.routing_table.snapshot_config();
let routed_model = active_config
.presets
.get("coding")
.and_then(|preset| preset.model.as_deref())
.ok_or_else(|| anyhow::anyhow!("coding preset is absent after reload"))?;
assert_eq!(routed_model, "beta:m");
let access_model = fixture.policy_store.with_policy("operator", |policy| {
policy
.and_then(|policy| policy.allowed_models.as_ref())
.and_then(|models| models.first())
.cloned()
});
assert_eq!(access_model.as_deref(), Some("beta:m"));
assert_eq!(
fixture
.policy_runtime
.administration_snapshot()
.digest
.as_deref(),
Some(fixture.candidate_policy_digest.as_str())
);
Ok(())
}
#[tokio::test]
async fn local_and_sighup_entrypoints_cannot_overtake_remote_reload() -> anyhow::Result<()> {
let pause = Arc::new(TestPreparationPause::new());
let fixture = coordinated_fixture(None, Some(pause.clone())).await?;
let state = fixture
.reloader
.reload_state()
.ok_or_else(|| anyhow::anyhow!("coordinator state is unavailable"))?;
let reservation = fixture
.reloader
.reserve_remote(&state.server_instance_id, state.generation)?;
let remote_reloader = fixture.reloader.clone();
let remote =
tokio::spawn(async move { remote_reloader.reload_reserved(reservation).await });
tokio::time::timeout(Duration::from_secs(3), pause.prepared.notified()).await?;
assert!(
fixture
.reloader
.reload_state()
.is_some_and(|state| state.running)
);
let configuration = fixture
.reloader
.configuration_state()
.await
.ok_or_else(|| anyhow::anyhow!("configuration state is unavailable"))?;
assert_eq!(configuration.running, RunningConfigState::Unknown);
let override_name = format!(
"BITROUTER_RELOAD_CONCURRENCY_{}",
uuid::Uuid::new_v4().simple()
);
assert!(bitrouter_sdk::config::env_lookup(&override_name).is_none());
let local = tokio::time::timeout(
Duration::from_secs(1),
fixture.reloader.reload_with_env(vec![(
override_name.clone(),
"must-not-be-installed".to_string(),
)]),
)
.await?;
let local_error = match local {
Ok(()) => return Err(anyhow::anyhow!("local reload overtook remote admission")),
Err(error) => error,
};
assert!(local_error.to_string().contains("reload_in_progress"));
assert!(bitrouter_sdk::config::env_lookup(&override_name).is_none());
let hup = tokio::time::timeout(Duration::from_secs(1), fixture.reloader.reload()).await?;
let hup_error = match hup {
Ok(()) => return Err(anyhow::anyhow!("SIGHUP reload overtook remote admission")),
Err(error) => error,
};
assert!(hup_error.to_string().contains("reload_in_progress"));
pause.resume.notify_one();
let report = tokio::time::timeout(Duration::from_secs(3), remote).await??;
assert_eq!(report.outcome, ReloadOutcome::Succeeded);
assert_eq!(
fixture
.reloader
.reload_state()
.map(|state| state.generation),
Some(1)
);
Ok(())
}
#[tokio::test]
async fn every_entrypoint_rejects_startup_change_before_mutation() -> anyhow::Result<()> {
let (path, dir) = temp_config_path();
let initial_yaml = r#"inherit_defaults: false
server:
listen: 127.0.0.1:4356
presets:
coding:
model: alpha:m
"#;
let candidate_yaml = r#"inherit_defaults: false
server:
listen: 127.0.0.1:9999
presets:
coding:
model: beta:m
"#;
std::fs::write(&path, initial_yaml)?;
let mut initial = config::load(&path).await?;
resolve_reloadable_config(&mut initial).await;
let routing_table = Arc::new(ConfigRoutingTable::from_config(initial));
let reloader = AppReloader::new(
Arc::new(PolicyStore::new()),
routing_table.clone(),
Arc::new(HttpExecutor::new(HttpTimeouts::default())?),
ReloadSource::File(path.clone()),
);
std::fs::write(&path, candidate_yaml)?;
assert!(reloader.reload().await.is_err());
assert_eq!(
routing_table.snapshot_config().server.listen,
"127.0.0.1:4356"
);
let state = reloader
.reload_state()
.ok_or_else(|| anyhow::anyhow!("coordinator state is unavailable"))?;
assert_eq!(state.generation, 1);
let reservation = reloader.reserve_remote(&state.server_instance_id, state.generation)?;
let report = reloader.reload_reserved(reservation).await;
assert_eq!(report.outcome, ReloadOutcome::Failed);
assert!(
report
.restart_required_fields
.iter()
.any(|field| field == "server.listen"),
"remote reload must compare startup-only fields with the daemon baseline"
);
assert_eq!(
report
.participants
.iter()
.find(|entry| entry.participant == ReloadParticipant::RoutingTable)
.and_then(|entry| entry.error.as_ref())
.map(|error| error.code.as_str()),
Some("restart_required")
);
assert_eq!(
routing_table.snapshot_config().server.listen,
"127.0.0.1:4356"
);
let state = reloader
.reload_state()
.ok_or_else(|| anyhow::anyhow!("coordinator state disappeared"))?;
assert_eq!(state.generation, 2);
assert_eq!(state.consistency, ReloadConsistency::Consistent);
let _ = std::fs::remove_dir_all(dir);
Ok(())
}
#[tokio::test]
async fn preparation_timeout_finishes_without_live_mutation() -> anyhow::Result<()> {
let initial =
config::parse("inherit_defaults: false\npresets:\n coding:\n model: alpha:m\n")?;
let routing_table = Arc::new(ConfigRoutingTable::from_config(initial));
let pause = Arc::new(TestPreparationPause::new());
let reloader = Arc::new(
AppReloader::new(
Arc::new(PolicyStore::new()),
routing_table.clone(),
Arc::new(HttpExecutor::new(HttpTimeouts::default())?),
ReloadSource::Default,
)
.with_test_pause_during_prepare(pause.clone())
.with_test_preparation_timeout(Duration::from_millis(10)),
);
let state = reloader
.reload_state()
.ok_or_else(|| anyhow::anyhow!("coordinator state is unavailable"))?;
let reservation = reloader.reserve_remote(&state.server_instance_id, state.generation)?;
let running_reloader = reloader.clone();
let task = tokio::spawn(async move { running_reloader.reload_reserved(reservation).await });
tokio::time::timeout(Duration::from_secs(1), pause.prepared.notified()).await?;
let report = tokio::time::timeout(Duration::from_secs(1), task).await??;
assert_eq!(report.outcome, ReloadOutcome::Failed);
assert!(report.participants.iter().all(|entry| {
entry.participant == ReloadParticipant::RoutingTable
|| entry.outcome == ReloadParticipantOutcome::NotAttempted
}));
let routing_failure = report
.participants
.iter()
.find(|entry| entry.participant == ReloadParticipant::RoutingTable)
.and_then(|entry| entry.error.as_ref())
.ok_or_else(|| anyhow::anyhow!("timeout did not identify preparation failure"))?;
assert_eq!(routing_failure.code, "preparation_timed_out");
let state = reloader
.reload_state()
.ok_or_else(|| anyhow::anyhow!("coordinator state disappeared"))?;
assert!(!state.running);
assert_eq!(state.generation, 1);
assert_eq!(state.consistency, ReloadConsistency::Consistent);
assert_eq!(
state.last_outcome.as_ref().map(|outcome| outcome.outcome),
Some(ReloadOutcome::Failed)
);
let active_config = routing_table.snapshot_config();
assert_eq!(
active_config
.presets
.get("coding")
.and_then(|preset| preset.model.as_deref()),
Some("alpha:m")
);
Ok(())
}
#[test]
fn coordinator_fences_stale_and_concurrent_admission() -> Result<(), ReloadAdmissionError> {
let coordinator = ReloadCoordinator::new();
let initial = coordinator.state();
let first = coordinator.reserve_remote(&initial.server_instance_id, initial.generation)?;
assert!(matches!(
coordinator.reserve_local(),
Err(ReloadAdmissionError::ReloadInProgress)
));
assert!(coordinator.state().running);
drop(first);
let after_abandon = coordinator.state();
assert!(!after_abandon.running);
assert_eq!(after_abandon.generation, 0);
let accepted = coordinator
.reserve_remote(&after_abandon.server_instance_id, after_abandon.generation)?;
let report = ReloadReport::succeeded(
accepted.server_instance_id().to_string(),
accepted.generation(),
);
coordinator.complete(&accepted, report);
drop(accepted);
let completed = coordinator.state();
assert_eq!(completed.generation, 1);
assert!(matches!(
coordinator.reserve_remote(&completed.server_instance_id, 0),
Err(ReloadAdmissionError::StaleGeneration)
));
Ok(())
}
#[test]
fn interrupted_mutation_records_unknown_mixed_state() -> anyhow::Result<()> {
let coordinator = ReloadCoordinator::new();
let reservation = coordinator.reserve_local()?;
let mut progress = ReloadReport::pending(
reservation.server_instance_id().to_string(),
reservation.generation(),
);
progress.participant_applied(ReloadParticipant::RoutingTable);
reservation.update_progress(&progress);
reservation.mark_mutation_started();
drop(reservation);
let state = coordinator.state();
let outcome = state
.last_outcome
.as_ref()
.ok_or_else(|| anyhow::anyhow!("interrupted reservation did not record an outcome"))?;
assert_eq!(state.generation, 1);
assert!(!state.running);
assert_eq!(state.consistency, ReloadConsistency::Mixed);
assert_eq!(outcome.outcome, ReloadOutcome::Unknown);
assert_eq!(
outcome
.participants
.iter()
.find(|entry| entry.participant == ReloadParticipant::RoutingTable)
.map(|entry| entry.outcome),
Some(ReloadParticipantOutcome::Applied),
"interruption must retain already-recorded live changes"
);
assert_eq!(state.mixed_state_history.len(), 1);
Ok(())
}
#[test]
fn failed_preparation_does_not_clear_prior_mixed_state() -> anyhow::Result<()> {
let coordinator = ReloadCoordinator::new();
let partial_reservation = coordinator.reserve_local()?;
let mut partial = ReloadReport::pending(
partial_reservation.server_instance_id().to_string(),
partial_reservation.generation(),
);
partial.participant_applied(ReloadParticipant::RoutingTable);
partial.participant_failed(
ReloadParticipant::NamedPolicyRuntime,
ReloadFailure::new("fault_injected", "policy table update failed"),
);
partial.finish(ReloadOutcome::PartiallyApplied);
coordinator.complete(&partial_reservation, partial);
drop(partial_reservation);
assert_eq!(coordinator.state().consistency, ReloadConsistency::Mixed);
let failed_reservation = coordinator.reserve_local()?;
let mut failed = ReloadReport::pending(
failed_reservation.server_instance_id().to_string(),
failed_reservation.generation(),
);
failed.participant_failed(
ReloadParticipant::RoutingTable,
ReloadFailure::new("config_prepare_failed", "reload configuration is invalid"),
);
failed.finish(ReloadOutcome::Failed);
coordinator.complete(&failed_reservation, failed);
drop(failed_reservation);
let state = coordinator.state();
assert_eq!(state.generation, 2);
assert_eq!(state.consistency, ReloadConsistency::Mixed);
assert_eq!(state.mixed_state_history.len(), 1);
assert_eq!(
state.last_outcome.as_ref().map(|outcome| outcome.outcome),
Some(ReloadOutcome::Failed)
);
Ok(())
}
#[tokio::test]
async fn auxiliary_source_drift_and_invalidity_prevent_in_sync() -> anyhow::Result<()> {
let fixture = coordinated_fixture(None, None).await?;
fixture.reloader.reload().await?;
let synchronized = fixture
.reloader
.configuration_state()
.await
.ok_or_else(|| anyhow::anyhow!("configuration state is unavailable"))?;
assert_eq!(synchronized.running, RunningConfigState::InSync);
std::fs::write(&fixture.lock_path, coordinated_policy_lock_yaml("alpha:m"))?;
std::fs::write(
fixture.access_policy_dir.join("operator.yaml"),
"id: operator\nallowed_models: [alpha:m]\n",
)?;
let changed = fixture
.reloader
.configuration_state()
.await
.ok_or_else(|| anyhow::anyhow!("configuration state is unavailable"))?;
assert_eq!(changed.running, RunningConfigState::ReloadRequired);
assert_eq!(changed.named_policy, AuxiliaryConfigState::ReloadRequired);
assert_eq!(
changed.access_policies,
AuxiliaryConfigState::ReloadRequired
);
assert!(
changed
.reload_required_fields
.contains(&"policy_lock".into())
);
assert!(
changed
.reload_required_fields
.contains(&"access_policies".into())
);
std::fs::write(&fixture.lock_path, "lockfileVersion: broken\n")?;
let invalid = fixture
.reloader
.configuration_state()
.await
.ok_or_else(|| anyhow::anyhow!("configuration state is unavailable"))?;
assert_eq!(invalid.named_policy, AuxiliaryConfigState::Invalid);
assert_ne!(invalid.running, RunningConfigState::InSync);
std::fs::remove_file(&fixture.lock_path)?;
let missing = fixture
.reloader
.configuration_state()
.await
.ok_or_else(|| anyhow::anyhow!("configuration state is unavailable"))?;
assert_eq!(missing.named_policy, AuxiliaryConfigState::Missing);
assert_ne!(missing.running, RunningConfigState::InSync);
Ok(())
}
#[tokio::test]
async fn default_source_requires_restart_when_a_config_file_appears() -> anyhow::Result<()> {
let home = tempfile::tempdir()?;
let source = crate::paths::ConfigSource::Default {
home: home.path().to_path_buf(),
};
let baseline = load_configuration_baseline(&source).await?;
let running_config = baseline.config.clone();
let reloader = AppReloader::new(
Arc::new(PolicyStore::new()),
Arc::new(ConfigRoutingTable::from_config(running_config)),
Arc::new(HttpExecutor::new(HttpTimeouts::default())?),
ReloadSource::Default,
)
.with_startup_configuration(baseline);
let initial = reloader
.configuration_state()
.await
.ok_or_else(|| anyhow::anyhow!("configuration state is unavailable"))?;
assert_eq!(initial.source, ConfigSourceKind::Default);
assert_eq!(initial.saved, SavedConfigState::Generated);
assert_eq!(initial.running, RunningConfigState::InSync);
let path = home.path().join("bitrouter.yaml");
std::fs::write(&path, "inherit_defaults: false\n")?;
let pending = reloader
.configuration_state()
.await
.ok_or_else(|| anyhow::anyhow!("configuration state is unavailable"))?;
assert_eq!(pending.source, ConfigSourceKind::Default);
assert_eq!(pending.saved, SavedConfigState::Available);
assert_eq!(pending.running, RunningConfigState::RestartRequired);
assert_eq!(pending.restart_required_fields, ["configuration_source"]);
assert!(reloader.reload().await.is_err());
let report = reloader
.reload_state()
.and_then(|state| state.last_outcome)
.ok_or_else(|| anyhow::anyhow!("reload report is unavailable"))?;
assert_eq!(report.restart_required_fields, ["configuration_source"]);
std::fs::write(&path, "providers: [invalid\n")?;
let invalid = reloader
.configuration_state()
.await
.ok_or_else(|| anyhow::anyhow!("configuration state is unavailable"))?;
assert_eq!(invalid.source, ConfigSourceKind::Default);
assert_eq!(invalid.saved, SavedConfigState::Invalid);
assert_eq!(invalid.running, RunningConfigState::Unknown);
Ok(())
}
#[tokio::test]
async fn removed_global_policy_table_is_rejected_before_reload() -> anyhow::Result<()> {
assert!(config::parse("policy_table: {}\n").is_err());
Ok(())
}
#[tokio::test]
async fn interrupted_mutation_makes_configuration_state_mixed() -> anyhow::Result<()> {
let baseline = load_configuration_baseline(&crate::paths::ConfigSource::Default {
home: PathBuf::new(),
})
.await?;
let config = baseline.config.clone();
let reloader = AppReloader::new(
Arc::new(PolicyStore::new()),
Arc::new(ConfigRoutingTable::from_config(config.clone())),
Arc::new(HttpExecutor::new(HttpTimeouts::default())?),
ReloadSource::Default,
)
.with_startup_configuration(baseline);
let reservation = reloader.coordinator.reserve_local()?;
reservation.mark_mutation_started();
drop(reservation);
let state = reloader
.configuration_state()
.await
.ok_or_else(|| anyhow::anyhow!("configuration state is unavailable"))?;
assert_eq!(state.running, RunningConfigState::Mixed);
assert_eq!(
state.last_reload.as_ref().map(|report| report.outcome),
Some(ReloadOutcome::Unknown)
);
Ok(())
}
#[test]
fn remote_restart_classifier_rejects_startup_and_unknown_fields() -> anyhow::Result<()> {
let current = config::parse("inherit_defaults: false\n")?;
let candidate = config::parse(
"inherit_defaults: false\nserver:\n listen: 127.0.0.1:9999\nupstream:\n fallback_backoff_ms: [10]\n",
)?;
let unclassified_fields = ["future_runtime"]
.into_iter()
.map(str::to_string)
.collect::<BTreeSet<_>>();
let fields = restart_required_fields(¤t, &candidate, Some(&unclassified_fields));
assert!(fields.contains(&"server.listen".to_string()));
assert!(fields.contains(&"upstream.fallback_backoff_ms".to_string()));
assert!(fields.contains(&"future_runtime".to_string()));
Ok(())
}
#[test]
fn changing_semantic_backend_requires_restart() -> anyhow::Result<()> {
let current = config::parse("decision_model:\n model: semantic-v1\n")?;
let candidate = config::parse("decision_model:\n model: semantic-v2\n")?;
let fields = restart_required_fields(¤t, &candidate, Some(&BTreeSet::new()));
assert!(fields.contains(&"decision_model".into()));
assert_eq!(fixed_restart_fields(fields), vec!["decision_model"]);
Ok(())
}
#[test]
fn protocol_tariffs_endpoint_profiles_and_known_price_policy_require_restart()
-> anyhow::Result<()> {
use bitrouter_ai::types::ApiProtocol;
let current: bitrouter_sdk::config::Config = serde_json::from_value(serde_json::json!({
"providers":{"fixture":{"api_base":"https://api.openai.com/v1", "models":[{
"id":"test", "pricing_by_protocol":{"decisions":{"input_micro_usd_per_token":0.1,"output_micro_usd_per_token":0}}
}]}}
}))?;
let expected = "providers.*.models.*.pricing".to_string();
let mut rate = current.clone();
let provider = rate
.providers
.get_mut("fixture")
.ok_or_else(|| anyhow::anyhow!("missing fixture"))?;
let model = provider
.models
.first_mut()
.ok_or_else(|| anyhow::anyhow!("missing model"))?;
model
.pricing_by_protocol
.get_mut(&ApiProtocol::Decisions)
.ok_or_else(|| anyhow::anyhow!("missing tariff"))?
.input_micro_usd_per_token = Some(0.2);
assert!(
restart_required_fields(¤t, &rate, Some(&BTreeSet::new())).contains(&expected)
);
let mut endpoint = current.clone();
endpoint
.providers
.get_mut("fixture")
.ok_or_else(|| anyhow::anyhow!("missing fixture"))?
.api_base = "https://eu.api.openai.com/v1".into();
assert!(
restart_required_fields(¤t, &endpoint, Some(&BTreeSet::new()))
.contains(&expected)
);
let mut protocol_endpoint = current.clone();
protocol_endpoint
.providers
.get_mut("fixture")
.ok_or_else(|| anyhow::anyhow!("missing fixture"))?
.protocol_endpoints
.insert("decisions".into(), "https://us.api.openai.com/v1".into());
assert!(
restart_required_fields(¤t, &protocol_endpoint, Some(&BTreeSet::new()))
.contains(&expected)
);
let mut provenance = current.clone();
provenance
.providers
.get_mut("fixture")
.and_then(|provider| provider.models.first_mut())
.and_then(|model| model.pricing_by_protocol.get_mut(&ApiProtocol::Decisions))
.ok_or_else(|| anyhow::anyhow!("missing tariff"))?
.endpoint_profile =
Some(bitrouter_ai::catalog::types::PricingEndpointProfile::OpenaiGlobal);
assert!(
restart_required_fields(¤t, &provenance, Some(&BTreeSet::new()))
.contains(&expected)
);
let mut strict = current.clone();
strict.server.require_known_pricing = true;
assert!(
restart_required_fields(¤t, &strict, Some(&BTreeSet::new()))
.contains(&"server.require_known_pricing".to_string())
);
Ok(())
}
#[test]
fn remote_restart_classifier_allows_timeout_client_changes() -> anyhow::Result<()> {
let current = config::parse("inherit_defaults: false\n")?;
let candidate =
config::parse("inherit_defaults: false\nupstream:\n timeouts:\n read_secs: 10\n")?;
assert!(restart_required_fields(¤t, &candidate, Some(&BTreeSet::new())).is_empty());
Ok(())
}
#[tokio::test]
async fn checker_revision_edits_require_restart_before_any_reload_mutation()
-> anyhow::Result<()> {
let directory = tempfile::tempdir()?;
let path = directory.path().join("bitrouter.yaml");
let document = |revision: u16| {
format!(
"inherit_defaults: false\ncheckers:\n company:\n native:\n revision: rules-{revision}\n"
)
};
std::fs::write(&path, document(18081))?;
let baseline =
load_configuration_baseline(&crate::paths::ConfigSource::File(path.clone())).await?;
let mut initial = baseline.config().clone();
resolve_reloadable_config(&mut initial).await;
let routing_table = Arc::new(ConfigRoutingTable::from_config(initial));
let reloader = AppReloader::new(
Arc::new(PolicyStore::new()),
routing_table.clone(),
Arc::new(HttpExecutor::new(HttpTimeouts::default())?),
ReloadSource::File(path.clone()),
)
.with_startup_configuration(baseline);
std::fs::write(&path, document(18082))?;
assert!(reloader.reload().await.is_err());
let running = routing_table.snapshot_config();
assert_eq!(
running
.checkers
.get("company")
.map(|checker| match checker {
bitrouter_sdk::config::checker::CheckerConfig::Native { native } =>
native.revision.as_str(),
}),
Some("rules-18081")
);
let state = reloader
.reload_state()
.ok_or_else(|| anyhow::anyhow!("missing reload state"))?;
assert_eq!(state.consistency, ReloadConsistency::Consistent);
let report = state
.last_outcome
.ok_or_else(|| anyhow::anyhow!("missing reload report"))?;
assert_eq!(report.restart_required_fields, ["checkers"]);
assert_eq!(fixed_top_level_field("checkers"), "checkers");
assert_eq!(fixed_restart_field("checkers"), "checkers");
Ok(())
}
#[tokio::test]
async fn local_reload_rejects_router_changes_before_mutating_the_live_table()
-> anyhow::Result<()> {
let (path, dir) = temp_config_path();
let config = |prompt: &str| {
format!(
"inherit_defaults: false\nrouters:\n coding:\n selection:\n kind: model\n model: demo/model\n defaults:\n system_prompt: {prompt}\n"
)
};
std::fs::write(&path, config("old"))?;
let mut initial = config::load(&path).await?;
resolve_reloadable_config(&mut initial).await;
let routing_table = Arc::new(ConfigRoutingTable::from_config(initial));
let reloader = AppReloader::new(
Arc::new(PolicyStore::new()),
routing_table.clone(),
Arc::new(HttpExecutor::new(HttpTimeouts::default())?),
ReloadSource::File(path.clone()),
);
std::fs::write(&path, config("new"))?;
if reloader.reload().await.is_ok() {
anyhow::bail!("router edits unexpectedly reloaded without a restart");
}
let running = routing_table.snapshot_config();
assert_eq!(
running
.routers
.get("coding")
.and_then(|router| router.defaults.system_prompt.as_deref()),
Some("old")
);
let state = reloader
.reload_state()
.ok_or_else(|| anyhow::anyhow!("coordinator state unavailable"))?;
assert_eq!(state.generation, 1);
assert_eq!(state.consistency, ReloadConsistency::Consistent);
assert!(
state
.last_outcome
.as_ref()
.is_some_and(|report| report.restart_required_fields == ["routers"])
);
assert!(state.last_outcome.as_ref().is_some_and(|report| {
report.participants.iter().any(|participant| {
participant.participant == ReloadParticipant::RoutingTable
&& participant
.error
.as_ref()
.is_some_and(|error| error.code == "restart_required")
})
}));
let _ = std::fs::remove_dir_all(dir);
Ok(())
}
#[test]
fn remote_restart_classifier_rejects_unknown_nested_fields() -> anyhow::Result<()> {
let raw = r#"inherit_defaults: false
server:
future_field: true
upstream:
future_field: true
providers:
alpha:
api_base: https://alpha.example.com/v1
api_key: key
future_field: true
"#;
let current = config::parse("inherit_defaults: false\n")?;
let candidate = config::parse(raw)?;
let raw_candidate = serde_saphyr::from_str::<serde_json::Value>(raw)?;
let unclassified = unclassified_config_paths(&raw_candidate);
assert!(unclassified.contains("server.future_field"));
assert!(unclassified.contains("upstream.future_field"));
assert!(unclassified.contains("providers.alpha.future_field"));
let fields = restart_required_fields(¤t, &candidate, Some(&unclassified));
assert!(fields.contains(&"server.future_field".to_string()));
assert!(fields.contains(&"upstream.future_field".to_string()));
assert!(fields.contains(&"providers.alpha.future_field".to_string()));
Ok(())
}
#[test]
fn unknown_restart_fields_only_report_drift_from_startup() -> anyhow::Result<()> {
let startup = serde_saphyr::from_str::<serde_json::Value>(
"inherit_defaults: false\nserver:\n future_field: stable\nupstream:\n timeouts:\n read_secs: 30\n",
)?;
let known_edit = serde_saphyr::from_str::<serde_json::Value>(
"inherit_defaults: false\nserver:\n future_field: stable\nupstream:\n timeouts:\n read_secs: 1\n",
)?;
let value_edit = serde_saphyr::from_str::<serde_json::Value>(
"inherit_defaults: false\nserver:\n future_field: changed\nupstream:\n timeouts:\n read_secs: 1\n",
)?;
let removed = serde_saphyr::from_str::<serde_json::Value>(
"inherit_defaults: false\nupstream:\n timeouts:\n read_secs: 1\n",
)?;
let (_, startup_values) = unclassified_config(&startup);
let (known_fields, known_values) = unclassified_config(&known_edit);
let (value_fields, value_values) = unclassified_config(&value_edit);
let (removed_fields, removed_values) = unclassified_config(&removed);
assert!(
changed_unclassified_config_paths(
Some(&startup_values),
Some(&known_values),
Some(&known_fields),
)
.is_empty(),
"a known reloadable edit must not make an unchanged unknown field restart-only"
);
assert_eq!(
changed_unclassified_config_paths(
Some(&startup_values),
Some(&value_values),
Some(&value_fields),
),
BTreeSet::from(["server.future_field".to_string()])
);
assert_eq!(
changed_unclassified_config_paths(
Some(&startup_values),
Some(&removed_values),
Some(&removed_fields),
),
BTreeSet::from(["server.future_field".to_string()])
);
Ok(())
}
#[test]
fn remote_restart_classifier_accepts_legacy_policy_writeback_alias() -> anyhow::Result<()> {
let current = config::parse("inherit_defaults: false\npolicy:\n mode: frozen\n")?;
let raw = "inherit_defaults: false\npolicy:\n writeback: locked\n";
let candidate = config::parse(raw)?;
let raw_candidate = serde_saphyr::from_str::<serde_json::Value>(raw)?;
let unclassified = unclassified_config_paths(&raw_candidate);
assert!(!unclassified.contains("policy.writeback"));
assert!(restart_required_fields(¤t, &candidate, Some(&unclassified)).is_empty());
Ok(())
}
#[test]
fn remote_restart_classifier_rejects_inherit_defaults_change() -> anyhow::Result<()> {
let current = config::parse("inherit_defaults: false\n")?;
let candidate = config::parse("inherit_defaults: true\n")?;
let fields = restart_required_fields(¤t, &candidate, Some(&BTreeSet::new()));
assert!(fields.contains(&"inherit_defaults".to_string()));
Ok(())
}
}