use std::path::{Path, PathBuf};
use std::sync::Arc;
use anyhow::{Context, Result};
use clap::{Args, CommandFactory, Parser, Subcommand, ValueEnum};
use bitrouter::actions::administration::{PolicyInput, PolicyView};
use bitrouter::actions::requests::RequestFilters;
use bitrouter::actions::route::RouteInput;
use bitrouter::administration_target::{InspectionTarget, ReloadSubmission};
use bitrouter::commands;
use bitrouter::daemon::{self, DaemonCommand, DaemonResponse};
use bitrouter::output::reports::admin::{
KeySignReport, PolicyCreateReport, ProviderLoginReport, ProviderLogoutReport,
};
use bitrouter::output::reports::agents::{
AgentCheckRow, AgentConformanceReport, AgentConformanceTier, AgentInstallReport,
AgentRegistryRow, AgentRow, AgentSessionMutationReport, AgentSessionsReport, AgentsCheckReport,
AgentsListReport, BackgroundRunReport,
};
use bitrouter::output::reports::config::{UnsetVar, ValidateReport};
use bitrouter::output::reports::daemon::DaemonActionReport;
use bitrouter::output::reports::eval::EvalReport;
use bitrouter::output::reports::mcp::{
McpAddReport, McpCheckReport, McpCheckRow, McpRegistryReport, McpRegistryRow,
};
use bitrouter::output::reports::observe::ObserveStatusReport;
use bitrouter::output::reports::optimization::{
ArmReport, OptimizationControllerReport, TreatmentReport,
};
use bitrouter::output::reports::policy::PolicyReport;
use bitrouter::output::reports::routing::{ProviderRow, ProvidersReport};
use bitrouter::output::reports::tools::{
ServerStatusView, ServerToolsView, ToolInfo, ToolsDiscoverReport, ToolsListReport,
ToolsStatusReport,
};
use bitrouter::output::reports::trajectory::{
TrajectoryPruneReport, inspect_report as trajectory_inspect_report,
replay_report as trajectory_replay_report,
};
use bitrouter::output::{CliReport, Output};
use bitrouter::remote_control::operations::{OperationReport, OperationStatus};
use bitrouter_sdk::config;
async fn supervise_http_shutdown<Http, Control, Hup, Term>(
http: Http,
control: Control,
hup: Hup,
term: Term,
shutdown: tokio::sync::oneshot::Sender<()>,
control_shutdown: tokio::sync::oneshot::Sender<()>,
) -> Result<()>
where
Http: std::future::Future<Output = Result<()>> + Send,
Control: std::future::Future<Output = Result<()>> + Send,
Hup: std::future::Future<Output = Result<()>> + Send,
Term: std::future::Future<Output = Result<()>> + Send,
{
let mut http = Box::pin(http);
let mut control = Box::pin(control);
let mut hup = Box::pin(hup);
let mut term = Box::pin(term);
let mut hup_open = true;
enum Trigger {
Http(Result<()>),
Control(Result<()>),
Term,
}
let trigger = loop {
tokio::select! {
result = &mut http => break Trigger::Http(result),
result = &mut control => break Trigger::Control(result),
result = &mut term => {
if result.is_err() {
tracing::warn!(
reason = "termination_signal_listener_unavailable",
"termination-signal listener unavailable"
);
}
break Trigger::Term;
}
result = &mut hup, if hup_open => {
if result.is_err() {
tracing::warn!(
reason = "sighup_listener_unavailable",
"SIGHUP listener unavailable"
);
}
hup_open = false;
}
}
};
drop(hup);
drop(term);
match trigger {
Trigger::Http(http_result) => {
let _ = control_shutdown.send(());
control.await?;
http_result
}
Trigger::Control(control_result) => {
let _ = shutdown.send(());
http.await?;
control_result
}
Trigger::Term => {
let _ = control_shutdown.send(());
control.await?;
let _ = shutdown.send(());
http.await
}
}
}
async fn request_daemon_shutdown(socket: &Path) -> Result<DaemonResponse> {
let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(2);
loop {
match daemon::send_command(socket, &DaemonCommand::Stop).await {
Ok(response) => return Ok(response),
Err(error) if tokio::time::Instant::now() < deadline => {
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
tracing::debug!(error = %error, "waiting for local control endpoint before shutdown");
}
Err(error) => return Err(error),
}
}
}
#[derive(Parser)]
#[command(
name = bitrouter_sdk::invocation::DEFAULT,
version,
about,
after_help = "Global compatibility options (place before COMMAND): --json, --human, --context <NAME>"
)]
struct Cli {
#[arg(
short = 'j',
long,
global = true,
hide = true,
conflicts_with = "human"
)]
json: bool,
#[arg(long, global = true, hide = true)]
human: bool,
#[arg(short = 'H', hide = true, conflicts_with = "json")]
human_short: bool,
#[arg(long, global = true, hide = true, value_name = "NAME")]
context: Option<String>,
#[command(subcommand)]
command: Option<Command>,
}
#[derive(Args)]
struct NativeLaunchArgs {
#[arg(long, value_name = "ID")]
model: Option<String>,
#[arg(short, long)]
config: Option<PathBuf>,
#[arg(long)]
base_url: Option<String>,
#[arg(long)]
no_install: bool,
#[arg(long)]
no_start: bool,
#[arg(long)]
check: bool,
#[arg(last = true, allow_hyphen_values = true)]
agent_args: Vec<String>,
}
#[derive(Args)]
struct CodeArgs {
agent: Option<String>,
#[arg(
long,
value_name = "NATIVE_SESSION_ID",
conflicts_with = "resume",
requires = "agent"
)]
load: Option<String>,
#[arg(
long,
value_name = "NATIVE_SESSION_ID",
conflicts_with = "load",
requires = "agent"
)]
resume: Option<String>,
#[arg(long, value_name = "SECS")]
turn_timeout: Option<u64>,
#[command(flatten)]
routing: bitrouter::acp_cli::RoutingOptions,
#[arg(short, long)]
config: Option<PathBuf>,
#[arg(long)]
socket: Option<PathBuf>,
}
#[derive(Subcommand)]
enum Command {
Serve {
#[arg(short, long)]
config: Option<PathBuf>,
},
Start {
#[arg(short, long)]
config: Option<PathBuf>,
#[arg(long)]
log: Option<PathBuf>,
},
Stop {
#[arg(short, long)]
config: Option<PathBuf>,
#[arg(long)]
socket: Option<PathBuf>,
},
Restart {
#[arg(short, long)]
config: Option<PathBuf>,
#[arg(long)]
socket: Option<PathBuf>,
#[arg(long)]
log: Option<PathBuf>,
},
Reload {
#[arg(short, long)]
config: Option<PathBuf>,
#[arg(long)]
socket: Option<PathBuf>,
},
Operations {
#[command(subcommand)]
action: OperationsAction,
},
Status {
#[arg(short, long)]
config: Option<PathBuf>,
#[arg(long)]
socket: Option<PathBuf>,
#[arg(short, long, hide = true)]
requests: bool,
},
Requests {
#[arg(long, default_value_t = bitrouter::actions::requests::MAX_REQUEST_ROWS)]
limit: u64,
#[arg(long)]
since: Option<String>,
#[arg(long)]
until: Option<String>,
#[arg(long)]
model: Option<String>,
#[arg(long)]
provider: Option<String>,
#[arg(short, long)]
config: Option<PathBuf>,
#[arg(long)]
socket: Option<PathBuf>,
},
Checks {
#[arg(short, long, global = true)]
config: Option<PathBuf>,
#[arg(long, global = true)]
socket: Option<PathBuf>,
#[command(subcommand)]
action: Option<ChecksAction>,
},
Route {
model: String,
#[arg(long)]
prompt: Option<String>,
#[arg(short, long)]
config: Option<PathBuf>,
#[arg(long)]
socket: Option<PathBuf>,
},
Init {
#[arg(short, long)]
config: Option<PathBuf>,
#[arg(short = 'y', long)]
yes: bool,
#[arg(long)]
force: bool,
#[arg(long)]
reset: bool,
#[arg(long)]
cloud_login: bool,
#[arg(long, value_name = "BRK_KEY")]
api_key: Option<String>,
#[arg(long = "provider", value_name = "ID")]
providers: Vec<String>,
#[arg(long = "provider-api-key", value_name = "KEY")]
provider_api_keys: Vec<String>,
#[arg(long)]
use_detected: bool,
#[arg(long = "harness", value_enum)]
harnesses: Vec<bitrouter::spawn::SpawnAgent>,
#[arg(long, value_enum)]
after: Option<bitrouter::onboarding::AfterAction>,
#[arg(long, value_name = "ID")]
model: Option<String>,
},
Config {
#[command(subcommand)]
action: ConfigAction,
},
Key {
#[command(subcommand)]
action: KeyAction,
},
Models {
#[arg(short, long)]
config: Option<PathBuf>,
#[arg(short, long)]
provider: Option<String>,
},
Context {
#[command(subcommand)]
action: ContextAction,
},
#[command(hide = true)]
Tools {
#[command(subcommand)]
action: ToolsAction,
},
Observe {
#[command(subcommand)]
action: ObserveAction,
},
Policy {
#[command(subcommand)]
action: PolicyAction,
},
Eval {
#[command(subcommand)]
action: EvalAction,
},
Optimize {
#[command(subcommand)]
action: OptimizeAction,
},
Trajectory {
#[arg(short, long, global = true)]
config: Option<PathBuf>,
#[command(subcommand)]
action: TrajectoryAction,
},
Providers {
#[command(subcommand)]
action: ProviderAction,
},
Agents {
#[command(subcommand)]
action: Option<AgentsAction>,
},
#[command(override_usage = "bro launch <AGENT> [OPTIONS] [-- <AGENT_ARGS>...]")]
Launch {
#[arg(required_unless_present = "agent_compat")]
agent: Option<String>,
#[arg(
short = 'a',
long = "agent",
value_name = "ID",
hide = true,
conflicts_with = "agent"
)]
agent_compat: Option<String>,
#[command(flatten)]
options: NativeLaunchArgs,
},
Claude {
#[command(flatten)]
options: NativeLaunchArgs,
},
ClaudeCode {
#[command(flatten)]
options: NativeLaunchArgs,
},
Codex {
#[command(flatten)]
options: NativeLaunchArgs,
},
Run {
agent: String,
#[arg(conflicts_with = "prompt_file")]
prompt: Option<String>,
#[arg(long, value_name = "PATH", conflicts_with = "prompt")]
prompt_file: Option<PathBuf>,
#[arg(long, value_name = "NATIVE_SESSION_ID", conflicts_with = "resume")]
load: Option<String>,
#[arg(long, value_name = "NATIVE_SESSION_ID", conflicts_with = "load")]
resume: Option<String>,
#[arg(long, value_name = "PATH")]
cwd: Option<PathBuf>,
#[command(flatten)]
routing: bitrouter::acp_cli::RoutingOptions,
#[arg(long, value_name = "SECS")]
turn_timeout: Option<u64>,
#[arg(long, hide = true, conflicts_with = "background")]
no_wait: bool,
#[arg(long)]
background: bool,
#[arg(long, requires = "background")]
allow_shared_directory: bool,
#[arg(long, value_name = "JSON|@PATH", conflicts_with = "no_wait")]
result_schema: Option<String>,
#[command(flatten)]
headless: bitrouter::acp_cli::HeadlessOptions,
#[arg(short, long)]
config: Option<PathBuf>,
},
#[command(hide = true)]
Spawn {
#[arg(required_unless_present = "legacy_agent")]
agent: Option<String>,
#[arg(short = 'p', long, value_name = "TEXT")]
prompt: Option<String>,
#[arg(long, conflicts_with = "prompt")]
serve: bool,
#[arg(long, conflicts_with_all = ["prompt", "serve"])]
check: bool,
#[arg(long)]
direct: bool,
#[arg(long)]
model: Option<String>,
#[arg(long)]
base_url: Option<String>,
#[arg(long)]
no_start: bool,
#[arg(long, value_name = "SECS")]
turn_timeout: Option<u64>,
#[arg(long, requires = "prompt")]
no_wait: bool,
#[arg(
long,
value_name = "JSON|@PATH",
requires = "prompt",
conflicts_with = "no_wait"
)]
result_schema: Option<String>,
#[command(flatten)]
headless: bitrouter::acp_cli::HeadlessOptions,
#[arg(short, long)]
config: Option<PathBuf>,
#[arg(long = "agent", short = 'a', hide = true, value_enum)]
legacy_agent: Option<bitrouter::spawn::SpawnAgent>,
#[arg(long, hide = true)]
no_install: bool,
#[arg(last = true, allow_hyphen_values = true)]
agent_args: Vec<String>,
},
Cloud {
#[command(subcommand)]
action: bitrouter::cloud::cli::CloudAction,
},
Skills {
#[command(subcommand)]
action: bitrouter::skills::cli::SkillsAction,
},
Mcp {
#[command(subcommand)]
action: McpAction,
},
WorkflowState {
#[command(subcommand)]
action: WorkflowStateAction,
},
Update {
#[arg(long)]
check: bool,
#[arg(long)]
tag: Option<String>,
#[arg(long)]
stable: bool,
#[arg(long)]
restart: bool,
#[arg(short = 'y', long)]
yes: bool,
},
Acp {
#[command(subcommand)]
cmd: AcpCmd,
},
Code {
#[command(flatten)]
options: CodeArgs,
},
#[command(hide = true)]
Tui {
#[command(flatten)]
options: CodeArgs,
},
#[command(hide = true)]
Chat {
agent: String,
#[arg(long, value_name = "SECS")]
turn_timeout: Option<u64>,
#[command(flatten)]
routing: bitrouter::acp_cli::RoutingOptions,
#[arg(short, long)]
config: Option<PathBuf>,
},
}
#[derive(Subcommand)]
enum ContextAction {
Add {
name: String,
#[arg(long)]
endpoint: String,
#[arg(long, value_name = "NAME")]
token_env: String,
},
List,
Show { name: String },
Remove { name: String },
}
#[derive(Subcommand)]
enum ConfigAction {
MigrateRouters {
#[arg(short, long)]
config: Option<PathBuf>,
#[arg(long)]
candidate: Option<PathBuf>,
#[arg(long, requires_all = ["candidate", "source_digest"])]
apply: bool,
#[arg(long, requires = "apply")]
source_digest: Option<String>,
},
Validate {
#[arg(short, long)]
config: Option<PathBuf>,
},
}
fn parse_unit_interval_ppm(value: &str) -> std::result::Result<u32, String> {
let value = value.trim();
let (whole, fractional) = value.split_once('.').unwrap_or((value, ""));
if !matches!(whole, "0" | "1")
|| fractional.len() > 6
|| !fractional.bytes().all(|byte| byte.is_ascii_digit())
|| (whole == "1" && fractional.bytes().any(|byte| byte != b'0'))
{
return Err(format!(
"expected a decimal between 0 and 1 with at most six fractional digits, got {value}"
));
}
let fractional_value = if fractional.is_empty() {
0
} else {
fractional
.parse::<u32>()
.map_err(|error| format!("expected a decimal between 0 and 1, got {value}: {error}"))?
};
let scale = 10_u32.pow(
u32::try_from(6_usize.saturating_sub(fractional.len()))
.map_err(|error| format!("could not scale unit interval {value}: {error}"))?,
);
Ok(if whole == "1" {
1_000_000
} else {
fractional_value.saturating_mul(scale)
})
}
#[derive(Subcommand)]
enum WorkflowStateAction {
ClassifierBakeoff {
#[arg(long)]
fixtures: PathBuf,
#[arg(long)]
submission: Option<PathBuf>,
#[arg(long)]
output: PathBuf,
},
Bundle {
#[arg(long)]
run_label: String,
#[arg(long)]
traces: PathBuf,
#[arg(long)]
cloud_usage: PathBuf,
#[arg(long)]
outcomes: Option<PathBuf>,
#[arg(long)]
policy_decisions: Option<PathBuf>,
#[arg(long)]
output_dir: PathBuf,
},
MeteringUsage {
#[arg(long)]
database_url: String,
#[arg(long)]
output: PathBuf,
#[arg(long = "impute-price")]
impute_prices: Vec<String>,
#[arg(long)]
since: Option<String>,
#[arg(long)]
until: Option<String>,
},
ReliabilityReport {
#[arg(long)]
database_url: String,
#[arg(long)]
config: PathBuf,
#[arg(long)]
output: PathBuf,
},
PolicyOracle {
#[arg(long)]
traces: PathBuf,
#[arg(long)]
cloud_usage: PathBuf,
#[arg(long)]
policy_lock: PathBuf,
#[arg(long)]
policy: String,
#[arg(long = "effective-cost-factor", value_parser = parse_unit_interval_ppm)]
effective_cost_factor_ppm: u32,
#[arg(long = "target-savings", required = true, value_parser = parse_unit_interval_ppm)]
target_savings_ppm: Vec<u32>,
#[arg(long)]
output: PathBuf,
},
ReconcileMetering {
#[arg(long)]
database_url: String,
#[arg(long, default_value = "https://api.bitrouter.ai/v1")]
api_base: String,
#[arg(long, default_value = "BITROUTER_API_KEY")]
api_key_env: String,
#[arg(long)]
credentials_file: Option<PathBuf>,
#[arg(long = "request-id", required = true)]
request_ids: Vec<String>,
#[arg(long = "price")]
prices: Vec<String>,
#[arg(long, default_value_t = 12)]
max_attempts: u32,
#[arg(long, default_value_t = 1_000)]
poll_interval_ms: u64,
},
ApplyRewardFeedback {
#[arg(long)]
database_url: String,
#[arg(long)]
traces: PathBuf,
#[arg(long)]
cloud_usage: PathBuf,
#[arg(long)]
outcomes: PathBuf,
#[arg(long)]
policy_decisions: PathBuf,
},
}
#[derive(Subcommand)]
enum McpAction {
Check {
server: Option<String>,
#[arg(short, long)]
config: Option<PathBuf>,
},
#[command(hide = true)]
Search {
query: String,
#[arg(long, default_value_t = 50)]
limit: usize,
},
#[command(hide = true)]
List {
#[arg(long, default_value_t = 50)]
limit: usize,
},
#[command(hide = true)]
Add {
name: String,
},
}
#[derive(Subcommand)]
enum AgentsAction {
Sessions {
#[arg(short, long)]
config: Option<PathBuf>,
#[arg(long)]
socket: Option<PathBuf>,
},
Attach {
agent_run_id: String,
#[arg(short, long)]
config: Option<PathBuf>,
#[arg(long)]
socket: Option<PathBuf>,
},
Stop {
agent_run_id: String,
#[arg(short, long)]
config: Option<PathBuf>,
#[arg(long)]
socket: Option<PathBuf>,
},
Remove {
agent_run_id: String,
#[arg(short, long)]
config: Option<PathBuf>,
#[arg(long)]
socket: Option<PathBuf>,
},
List {
#[arg(long)]
remote: bool,
#[arg(short, long)]
config: Option<PathBuf>,
#[arg(long)]
socket: Option<PathBuf>,
},
Inspect {
agent: String,
#[command(flatten)]
routing: bitrouter::acp_cli::RoutingOptions,
#[arg(long, value_name = "MS", default_value_t = 2000)]
wait_ms: u64,
#[arg(long, value_enum)]
source: Option<CommandSourceArg>,
#[arg(short, long)]
config: Option<PathBuf>,
},
Check {
agent: Option<String>,
#[command(flatten)]
routing: bitrouter::acp_cli::RoutingOptions,
#[arg(short, long)]
config: Option<PathBuf>,
},
Conformance {
id: String,
},
Scaffold {
id: String,
},
#[command(hide = true)]
Install { id: String },
}
#[derive(Subcommand)]
enum ObserveAction {
Status {
#[arg(short, long)]
config: Option<PathBuf>,
#[arg(long)]
socket: Option<PathBuf>,
},
}
#[derive(Subcommand)]
enum ToolsAction {
List {
#[arg(short, long)]
config: Option<PathBuf>,
},
Status {
#[arg(short, long)]
config: Option<PathBuf>,
},
Discover {
server: String,
#[arg(short, long)]
config: Option<PathBuf>,
},
}
#[derive(Subcommand)]
enum KeyAction {
Sign {
#[arg(short, long)]
user: String,
#[arg(short, long, default_value = "sqlite://./bitrouter.db")]
db: String,
#[arg(long)]
policy: Option<String>,
},
}
#[derive(Subcommand)]
enum OperationsAction {
Show {
request_id: String,
#[arg(long)]
instance: String,
},
}
#[derive(Subcommand)]
enum ChecksAction {
Probe {
checker: String,
},
Receipts {
#[arg(long, default_value_t = bitrouter::actions::checks::DEFAULT_RECEIPT_LIMIT)]
limit: usize,
},
Receipt {
#[arg(value_name = "REQUEST_ID_OR_RECEIPT_ID")]
request_id_or_receipt_id: String,
#[arg(long)]
incarnation: Option<String>,
},
}
#[derive(Subcommand)]
enum PolicyAction {
Create {
id: String,
#[arg(long, default_value = "./policies")]
dir: PathBuf,
},
Init {
name: String,
#[arg(long, conflicts_with = "preset")]
router: Option<String>,
#[arg(long, conflicts_with = "router")]
preset: Option<String>,
#[arg(long)]
strong: Option<String>,
#[arg(long, requires = "strong")]
strong_effort: Option<bitrouter_sdk::language_model::types::ReasoningEffort>,
#[arg(long)]
economy: String,
#[arg(long)]
economy_effort: Option<bitrouter_sdk::language_model::types::ReasoningEffort>,
#[arg(short, long)]
config: Option<PathBuf>,
},
Check {
#[arg(short, long)]
config: Option<PathBuf>,
},
Verify {
#[arg(long)]
evidence: bool,
#[arg(short, long)]
config: Option<PathBuf>,
},
Status {
#[arg(long, value_enum)]
view: Option<PolicyViewArg>,
#[arg(short, long)]
config: Option<PathBuf>,
#[arg(long)]
socket: Option<PathBuf>,
},
Show {
name: String,
#[arg(long, value_enum)]
view: Option<PolicyViewArg>,
#[arg(short, long)]
config: Option<PathBuf>,
#[arg(long)]
socket: Option<PathBuf>,
},
Reload {
#[arg(short, long)]
config: Option<PathBuf>,
#[arg(long)]
socket: Option<PathBuf>,
},
Compile {
#[arg(long, value_name = "FILE")]
output: PathBuf,
#[arg(long, value_name = "UNIX_MS")]
snapshot_time: Option<i64>,
#[arg(long, value_name = "SHA256")]
eval_snapshot: Option<String>,
#[arg(short, long)]
config: Option<PathBuf>,
},
Diff { active: PathBuf, candidate: PathBuf },
Publish {
candidate: PathBuf,
#[arg(short, long)]
config: Option<PathBuf>,
#[arg(long)]
socket: Option<PathBuf>,
},
Evolve {
#[arg(long, conflicts_with = "output")]
apply: bool,
#[arg(long, value_name = "FILE", conflicts_with = "apply")]
output: Option<PathBuf>,
#[arg(short, long)]
config: Option<PathBuf>,
},
Rollback {
digest: String,
#[arg(short, long)]
config: Option<PathBuf>,
#[arg(long)]
socket: Option<PathBuf>,
},
}
#[derive(Debug, Clone, Copy, ValueEnum)]
enum PolicyViewArg {
Active,
Disk,
}
impl From<PolicyViewArg> for PolicyView {
fn from(view: PolicyViewArg) -> Self {
match view {
PolicyViewArg::Active => Self::Active,
PolicyViewArg::Disk => Self::Disk,
}
}
}
#[derive(Subcommand)]
enum EvalAction {
Subject {
#[command(subcommand)]
action: EvalSubjectAction,
},
Result {
#[command(subcommand)]
action: EvalResultAction,
},
Snapshot {
#[command(subcommand)]
action: EvalSnapshotAction,
},
Status {
#[arg(short, long)]
config: Option<PathBuf>,
},
}
#[derive(Subcommand)]
enum OptimizeAction {
Run {
#[arg(long, default_value = "auto")]
policy: String,
#[arg(long)]
candidate_tier: Option<String>,
#[arg(long, default_value_t = 100_000)]
exploration_ppm: u32,
#[arg(long, default_value_t = 3)]
minimum_tasks: u32,
#[arg(long, default_value_t = 20)]
maximum_tasks: u32,
#[arg(long, default_value_t = 900_000)]
minimum_pass_rate_ppm: u32,
#[arg(long)]
evaluator_config_digest: Option<String>,
#[arg(short, long, default_value = "bitrouter.yaml")]
config: PathBuf,
#[arg(long)]
socket: Option<PathBuf>,
},
Status {
#[arg(long, default_value = "auto")]
policy: String,
#[arg(short, long, default_value = "bitrouter.yaml")]
config: PathBuf,
},
}
#[derive(Subcommand)]
enum TrajectoryAction {
Inspect {
episode_id: String,
},
Replay {
episode_id: String,
},
Prune {
#[arg(long, value_parser = parse_rfc3339_argument)]
before: String,
#[arg(long)]
dry_run: bool,
},
}
fn parse_rfc3339_argument(value: &str) -> std::result::Result<String, String> {
chrono::DateTime::parse_from_rfc3339(value)
.map(|_| value.to_owned())
.map_err(|error| format!("expected an RFC3339 timestamp: {error}"))
}
#[derive(Subcommand)]
enum EvalSubjectAction {
Seal {
draft: PathBuf,
#[arg(long, value_name = "FILE")]
output: PathBuf,
},
Put {
file: PathBuf,
#[arg(short, long)]
config: Option<PathBuf>,
},
Get {
eval_id: String,
#[arg(short, long)]
config: Option<PathBuf>,
},
List {
#[arg(short, long)]
config: Option<PathBuf>,
},
}
#[derive(Subcommand)]
enum EvalResultAction {
Submit {
file: PathBuf,
#[arg(short, long)]
config: Option<PathBuf>,
},
}
#[derive(Subcommand)]
enum EvalSnapshotAction {
Freeze {
#[arg(long)]
at: Option<String>,
#[arg(short, long)]
config: Option<PathBuf>,
},
Get {
evidence_root: String,
#[arg(short, long)]
config: Option<PathBuf>,
},
}
#[derive(Subcommand)]
enum ProviderAction {
List {
#[arg(short, long)]
config: Option<PathBuf>,
#[arg(long)]
socket: Option<PathBuf>,
},
Login {
provider: String,
#[arg(short, long, default_value = "default")]
label: String,
#[arg(long)]
import_existing: bool,
#[arg(long)]
no_browser: bool,
#[arg(long, value_name = "KEY", conflicts_with_all = ["import_existing", "no_browser", "key_stdin"])]
api_key: Option<String>,
#[arg(long, conflicts_with_all = ["import_existing", "no_browser"])]
key_stdin: bool,
},
Logout {
provider: String,
},
}
#[derive(Subcommand)]
enum AcpRecordingAction {
List {
#[arg(long)]
agent: String,
},
Show {
#[arg(long)]
agent: String,
session: String,
},
Delete {
#[arg(long)]
agent: String,
session: String,
},
}
#[derive(Subcommand)]
enum AcpCheckpointAction {
Judge {
checkpoint: String,
#[arg(long)]
model: String,
},
JudgeJob {
job_id: String,
#[arg(long)]
resume: bool,
},
Rubric {
#[command(subcommand)]
action: bitrouter::evolution::scoring::ScoringCommand,
},
Create {
#[arg(long)]
watermark: i64,
},
List,
Show { checkpoint: String },
Resources {
checkpoint: String,
#[arg(long)]
refresh: bool,
},
Submit { file: PathBuf },
History,
Effective,
Family,
}
#[derive(Subcommand)]
enum AcpCmd {
Evolution {
#[arg(short, long)]
config: Option<PathBuf>,
#[command(subcommand)]
action: bitrouter::evolution::operator::EvolutionAction,
},
Checkpoints {
#[arg(long)]
agent: String,
session: String,
#[arg(short, long)]
config: Option<PathBuf>,
#[command(subcommand)]
action: AcpCheckpointAction,
},
Recordings {
#[command(subcommand)]
action: AcpRecordingAction,
#[arg(short, long)]
config: Option<PathBuf>,
},
#[command(override_usage = "bro acp serve <AGENT> [OPTIONS]")]
Serve {
agent: Option<String>,
#[arg(
long = "agent",
value_name = "ID",
hide = true,
conflicts_with = "agent"
)]
agent_compat: Option<String>,
#[arg(long, value_name = "SECS")]
turn_timeout: Option<u64>,
#[command(flatten)]
routing: bitrouter::acp_cli::RoutingOptions,
#[arg(short, long)]
config: Option<PathBuf>,
},
#[command(hide = true)]
Prompt {
#[arg(long)]
agent: String,
#[arg(long, value_name = "SECS")]
turn_timeout: Option<u64>,
#[arg(long)]
no_wait: bool,
#[command(flatten)]
routing: bitrouter::acp_cli::RoutingOptions,
#[command(flatten)]
headless: bitrouter::acp_cli::HeadlessOptions,
#[arg(short, long)]
config: Option<PathBuf>,
text: String,
},
#[command(hide = true)]
Commands {
#[arg(long)]
agent: String,
#[command(flatten)]
routing: bitrouter::acp_cli::RoutingOptions,
#[arg(long, value_name = "MS", default_value_t = 2000)]
wait_ms: u64,
#[arg(long, value_enum)]
source: Option<CommandSourceArg>,
#[arg(short, long)]
config: Option<PathBuf>,
},
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, clap::ValueEnum)]
enum CommandSourceArg {
Bitrouter,
Config,
Agent,
}
impl From<CommandSourceArg> for bitrouter::actions::commands::CommandSource {
fn from(arg: CommandSourceArg) -> Self {
match arg {
CommandSourceArg::Bitrouter => Self::Bitrouter,
CommandSourceArg::Config => Self::Config,
CommandSourceArg::Agent => Self::Agent,
}
}
}
const CLI_MAIN_STACK_SIZE: usize = 8 * 1024 * 1024;
const DEPRECATED_INVOCATION: &str = "bitrouter";
fn command_for_invocation(invoked: &'static str) -> clap::Command {
Cli::command()
.name(invoked)
.bin_name(invoked)
.mut_subcommand("launch", |command| {
command.override_usage(format!(
"{invoked} launch <AGENT> [OPTIONS] [-- <AGENT_ARGS>...]"
))
})
.mut_subcommand("acp", |command| {
command.mut_subcommand("serve", |command| {
command.override_usage(format!("{invoked} acp serve <AGENT> [OPTIONS]"))
})
})
}
fn parse_cli(invoked: &'static str) -> Cli {
use clap::FromArgMatches;
let matches = command_for_invocation(invoked).get_matches();
match Cli::from_arg_matches(&matches) {
Ok(cli) => cli,
Err(error) => error.exit(),
}
}
fn warn_deprecated_invocation() {
let palette = bitrouter::style::Palette::for_stderr();
eprintln!(
"{cyan}note:{reset} `{old}` is now `{new}`. This alias is removed in 1.0.0.",
cyan = palette.cyan,
reset = palette.reset,
old = DEPRECATED_INVOCATION,
new = bitrouter_sdk::invocation::DEFAULT,
);
}
fn main() {
let worker = std::thread::Builder::new()
.name("bitrouter-main".to_owned())
.stack_size(CLI_MAIN_STACK_SIZE)
.spawn(async_main);
match worker {
Ok(handle) => {
if handle.join().is_err() {
std::process::exit(101);
}
}
Err(error) => {
eprintln!("error: failed to start BitRouter CLI runtime: {error}");
std::process::exit(1);
}
}
}
#[tokio::main]
async fn async_main() {
let invoked = bitrouter_sdk::invocation::record_argv0();
if invoked == DEPRECATED_INVOCATION {
warn_deprecated_invocation();
}
let cli = parse_cli(invoked);
let raw_cloud_api = matches!(
&cli.command,
Some(Command::Cloud {
action: bitrouter::cloud::cli::CloudAction::Api(_)
})
);
let raw_agent_stream = matches!(
&cli.command,
Some(Command::Run {
background: false,
..
}) | Some(Command::Acp {
cmd: AcpCmd::Prompt { .. } | AcpCmd::Serve { .. },
}) | Some(Command::Spawn {
prompt: Some(_),
legacy_agent: None,
..
}) | Some(Command::Spawn {
serve: true,
legacy_agent: None,
..
})
);
let output = bitrouter::output::Output::from_flags(cli.json, cli.human || cli.human_short);
match Box::pin(run(cli, &output)).await {
Ok(()) => {}
Err(e) => {
if raw_cloud_api || raw_agent_stream {
eprintln!("error: {e:#}");
} else if let Some(routing) = e.downcast_ref::<bitrouter::acp_cli::RoutingError>() {
eprintln!("error: {routing}");
} else {
let _ = output.emit(&bitrouter::output::error::envelope_from_anyhow(&e));
}
std::process::exit(1);
}
}
}
async fn run(cli: Cli, output: &bitrouter::output::Output) -> Result<()> {
let is_acp = matches!(
&cli.command,
None | Some(
Command::Acp { .. }
| Command::Run { .. }
| Command::Spawn { .. }
| Command::Chat { .. }
| Command::Code { .. }
| Command::Tui { .. }
| Command::Agents {
action: None | Some(AgentsAction::Attach { .. })
}
| Command::Init { .. }
)
);
let owns_the_terminal = matches!(
&cli.command,
None | Some(
Command::Chat { .. }
| Command::Code { .. }
| Command::Tui { .. }
| Command::Agents {
action: None | Some(AgentsAction::Attach { .. })
}
| Command::Init { .. }
)
);
if matches!(cli.command, Some(Command::Serve { .. })) {
} else if is_acp {
if let Some(path) = init_session_log_tracing_subscriber(!owns_the_terminal) {
tracing::debug!(log = %path.display(), "session log");
bitrouter::chat::session::remember_session_log(path);
}
} else {
init_basic_tracing_subscriber();
}
let Some(command) = cli.command else {
if cli.context.as_deref().is_some_and(|name| name != "local") {
return Err(bitrouter_sdk::BitrouterError::bad_request(
"a remote context requires an explicit administration action",
)
.into());
}
return bitrouter::onboarding::entry(output).await;
};
let remote_context = if matches!(&command, Command::Context { .. }) {
if cli.context.as_deref().is_some_and(|name| name != "local") {
return Err(bitrouter_sdk::BitrouterError::bad_request(format!(
"`{} context` manages targets and does not run against a remote context",
bitrouter_sdk::invocation::name()
))
.into());
}
None
} else {
cli.context
.as_deref()
.map(bitrouter::contexts::resolve)
.transpose()?
.flatten()
};
validate_remote_invocation(&command, remote_context.is_some())?;
match command {
Command::Serve { config } => {
let source = bitrouter::paths::resolve_config(config.as_deref())?;
serve(&source).await
}
Command::Start { config, log } => {
let source = bitrouter::paths::resolve_config(config.as_deref())?;
let log_path = resolve_log_path(source.home(), log.as_deref());
output.emit(&start(&source, &log_path, "start").await?)?;
Ok(())
}
Command::Stop { config, socket } => {
let socket = resolve_client_socket(config.as_deref(), socket.as_deref()).await?;
output.emit(&stop(&socket).await?)?;
Ok(())
}
Command::Restart {
config,
socket,
log,
} => {
let source = bitrouter::paths::resolve_config(config.as_deref())?;
let socket = resolve_client_socket_from(&source, socket.as_deref()).await?;
let log_path = resolve_log_path(source.home(), log.as_deref());
output.emit(&restart(&source, &socket, &log_path).await?)?;
Ok(())
}
Command::Reload { config, socket } => {
if remote_context.is_some() {
let target = inspection_target(
remote_context.as_ref(),
cli.context.as_deref(),
config.as_deref(),
socket.as_deref(),
)
.await?;
let ReloadSubmission::Remote(report) = target.reload().await? else {
anyhow::bail!("remote reload resolved to a local control target")
};
return emit_operation_result(output, &report);
}
let socket = resolve_client_socket(config.as_deref(), socket.as_deref()).await?;
output.emit(&reload(&socket).await?)?;
Ok(())
}
Command::Operations { action } => {
let remote = remote_context.as_ref().ok_or_else(|| {
bitrouter_sdk::BitrouterError::bad_request(
"`operations show` requires a named remote context",
)
})?;
let target =
inspection_target(Some(remote), cli.context.as_deref(), None, None).await?;
match action {
OperationsAction::Show {
request_id,
instance,
} => {
let report = target.operation(&request_id, &instance).await?;
emit_operation_result(output, &report)
}
}
}
Command::Status {
config,
socket,
requests,
} => {
if requests {
eprintln!(
"note: `{cli} status --requests` is a compatibility form; use \
`{cli} requests`.",
cli = bitrouter_sdk::invocation::name()
);
}
let target = inspection_target(
remote_context.as_ref(),
cli.context.as_deref(),
config.as_deref(),
socket.as_deref(),
)
.await?;
if requests {
output.emit(&target.requests(RequestFilters::default()).await?)?;
} else {
output.emit(&target.status().await?)?;
}
Ok(())
}
Command::Requests {
limit,
since,
until,
model,
provider,
config,
socket,
} => {
let target = inspection_target(
remote_context.as_ref(),
cli.context.as_deref(),
config.as_deref(),
socket.as_deref(),
)
.await?;
output.emit(
&target
.requests(RequestFilters {
limit,
since,
until,
model,
provider,
})
.await?,
)?;
Ok(())
}
Command::Checks {
config,
socket,
action,
} => {
let target = inspection_target(
remote_context.as_ref(),
cli.context.as_deref(),
config.as_deref(),
socket.as_deref(),
)
.await?;
match action {
None => output.emit(&target.checks().await?)?,
Some(ChecksAction::Probe { checker }) => {
output.emit(&target.checks_probe(&checker).await?)?
}
Some(ChecksAction::Receipts { limit }) => {
output.emit(&target.check_receipts(limit).await?)?
}
Some(ChecksAction::Receipt {
request_id_or_receipt_id,
incarnation,
}) => output.emit(
&target
.check_receipt(&request_id_or_receipt_id, incarnation.as_deref())
.await?,
)?,
}
Ok(())
}
Command::Route {
model,
prompt,
config,
socket,
} => {
let target = inspection_target(
remote_context.as_ref(),
cli.context.as_deref(),
config.as_deref(),
socket.as_deref(),
)
.await?;
output.emit(&target.route(RouteInput { model, prompt }).await?)?;
Ok(())
}
Command::Init {
config,
yes,
force,
reset,
cloud_login,
api_key,
providers,
provider_api_keys,
use_detected,
harnesses,
after,
model,
} => {
let flags = bitrouter::onboarding::OnboardingFlags {
config,
yes,
force,
reset,
cloud_login,
api_key,
providers,
provider_api_keys,
use_detected,
harnesses,
after,
model,
};
bitrouter::onboarding::run(flags, output).await
}
Command::Config { action } => config_cmd(action, output).await,
Command::Key { action } => {
output.emit(&key(action).await?)?;
Ok(())
}
Command::Models { config, provider } => {
let target = inspection_target(
remote_context.as_ref(),
cli.context.as_deref(),
config.as_deref(),
None,
)
.await?;
output.emit(&target.models(provider.as_deref()).await?)?;
Ok(())
}
Command::Context { action } => {
match action {
ContextAction::Add {
name,
endpoint,
token_env,
} => output.emit(&bitrouter::contexts::add(&name, &endpoint, &token_env)?)?,
ContextAction::List => output.emit(&bitrouter::contexts::list()?)?,
ContextAction::Show { name } => output.emit(&bitrouter::contexts::show(&name)?)?,
ContextAction::Remove { name } => {
output.emit(&bitrouter::contexts::remove(&name)?)?
}
}
Ok(())
}
Command::Tools { action } => tools(action, output).await,
Command::Observe { action } => {
administration_observe(
action,
output,
remote_context.as_ref(),
cli.context.as_deref(),
)
.await
}
Command::Policy { action } => {
administration_policy(
action,
output,
remote_context.as_ref(),
cli.context.as_deref(),
)
.await
}
Command::Eval { action } => eval(action, output).await,
Command::Optimize { action } => optimize(action, output).await,
Command::Trajectory { config, action } => {
trajectory(config.as_deref(), action, output).await
}
Command::Providers { action } => {
administration_providers(
action,
output,
remote_context.as_ref(),
cli.context.as_deref(),
)
.await
}
Command::Agents { action } => {
administration_agents(
action,
output,
remote_context.as_ref(),
cli.context.as_deref(),
)
.await
}
Command::Launch {
agent,
agent_compat,
options,
} => {
if agent_compat.is_some() {
eprintln!(
"note: `{cli} launch --agent <agent>` is a compatibility form; use \
`{cli} launch <agent>`.",
cli = bitrouter_sdk::invocation::name()
);
}
let agent = agent.or(agent_compat).ok_or_else(|| {
bitrouter_sdk::BitrouterError::bad_request(format!(
"launch requires an agent, for example `{} launch claude`",
bitrouter_sdk::invocation::name()
))
})?;
run_native_launch(&agent, options, output).await
}
Command::Claude { options } | Command::ClaudeCode { options } => {
run_native_launch("claude", options, output).await
}
Command::Codex { options } => run_native_launch("codex", options, output).await,
Command::Run {
agent,
prompt,
prompt_file,
load,
resume,
cwd,
routing,
turn_timeout,
no_wait,
background,
allow_shared_directory,
result_schema,
headless,
config,
} => {
if no_wait {
eprintln!(
"note: `--no-wait` is deprecated and does not detach; use `acp serve \
<agent>` when an ACP client should own the connection."
);
}
let prompt = resolve_prompt_input(prompt, prompt_file.as_deref()).await?;
let source = bitrouter::paths::resolve_config(config.as_deref())?;
let cfg = bitrouter::paths::load_config(&source).await?;
let agent = bitrouter::acp_cli::resolve_agent_id(&cfg, &agent)?;
let session = match (load, resume) {
(Some(id), None) => bitrouter::acp_cli::SessionSelection::Load(id),
(None, Some(id)) => bitrouter::acp_cli::SessionSelection::Resume(id),
(None, None) => bitrouter::acp_cli::SessionSelection::New,
(Some(_), Some(_)) => {
return Err(bitrouter_sdk::BitrouterError::bad_request(
"--load and --resume cannot be combined",
)
.into());
}
};
if background {
if headless.format != bitrouter::acp_cli::PromptFormat::Ndjson {
return Err(bitrouter_sdk::BitrouterError::bad_request(
"`--format text|quiet` is unavailable with `--background`; the command returns a structured run report",
)
.into());
}
let schema = result_schema
.as_deref()
.map(bitrouter::result_contract::ResultContract::from_flag)
.transpose()?
.map(|contract| contract.schema().clone());
let permission_policy = headless.supervised_permission_policy()?;
let cwd = match cwd {
Some(cwd) => cwd,
None => {
std::env::current_dir().context("resolving the run working directory")?
}
};
if allow_shared_directory {
eprintln!(
"warning: --allow-shared-directory disables BitRouter's single-writer protection for this supervised run; concurrent agents may overwrite each other's work"
);
}
let no_start = routing.no_start;
let snapshot = bitrouter::agent_sessions::start_background(
&source,
&cfg,
no_start,
bitrouter::supervisor::StartRunRequest {
action_request_id: uuid::Uuid::new_v4().to_string(),
client_id: None,
label: None,
agent_id: agent,
prompt: Some(prompt),
cwd,
routing,
launch: bitrouter::acp_cli::launch_options(turn_timeout),
session,
presentation: bitrouter::supervisor::Presentation::Background,
parent_run_id: None,
allow_shared_directory,
permission_policy,
result_schema: schema,
},
)
.await?;
output.emit(&BackgroundRunReport::from_snapshot(&snapshot))?;
return Ok(());
}
run_agent_prompt(
&source,
cfg,
&agent,
AgentPromptRequest {
prompt: &prompt,
routing,
turn_timeout,
no_wait,
result_schema: result_schema.as_deref(),
headless,
session,
cwd,
},
)
.await
}
Command::Spawn {
agent,
prompt,
serve,
check,
direct,
model,
base_url,
no_start,
turn_timeout,
no_wait,
result_schema,
headless,
config,
legacy_agent,
no_install,
agent_args,
} => {
if let Some(legacy) = legacy_agent {
if agent.is_some() || prompt.is_some() || serve {
anyhow::bail!(
"`--agent` selects the deprecated interactive launcher; it cannot be \
combined with a positional agent id, `-p`, or `--serve`. Use \
`{cli} launch {}` for the TUI, or drop `--agent` to spawn \
an ACP sub-agent.",
legacy.spec().id,
cli = bitrouter_sdk::invocation::name()
);
}
eprintln!(
"note: `{cli} spawn --agent` is deprecated — use \
`{cli} launch {}` (this alias will be removed).",
legacy.spec().id,
cli = bitrouter_sdk::invocation::name()
);
let opts = bitrouter::spawn::SpawnOptions {
agent: bitrouter::spawn::resolve_launch_agent(legacy.spec().id)?,
model: model.clone(),
agent_args,
base_url,
no_install,
no_start,
check,
};
return run_launch(config.as_deref(), opts, output).await;
}
let Some(agent) = agent else {
anyhow::bail!(
"spawn: provide an agent id and a mode, e.g. \
`{cli} spawn claude-acp -p \"summarize README\"`, \
`{cli} spawn codex-acp --serve`, or `--check`.",
cli = bitrouter_sdk::invocation::name()
);
};
let source = bitrouter::paths::resolve_config(config.as_deref())?;
let cfg = bitrouter::paths::load_config(&source).await?;
let agent = bitrouter::acp_cli::resolve_agent_id(&cfg, &agent)?;
let routing = bitrouter::acp_cli::RoutingOptions {
direct,
base_url,
model,
no_start,
};
if prompt.is_none() && headless != bitrouter::acp_cli::HeadlessOptions::default() {
anyhow::bail!(
"spawn: `--approve-all`, `--approve-reads`, `--deny-all`, \
`--permission-policy`, and `--format` apply to `-p` only; \
`--serve` hands permissions to the ACP client and `--check` asks nothing."
);
}
if check {
eprintln!(
"note: `{cli} spawn <agent> --check` is a compatibility form; use \
`{cli} agents check <agent>`.",
cli = bitrouter_sdk::invocation::name()
);
let report = bitrouter::acp_cli::spawn_check(cfg, &agent, &routing).await?;
output.emit(&report)?;
if report.exit_code() == 0 {
Ok(())
} else {
std::process::exit(report.exit_code());
}
} else if serve {
let options = bitrouter::acp_cli::launch_options(turn_timeout);
let ctx = bitrouter::acp_cli::SpawnContext {
source: &source,
config: cfg,
agent_id: &agent,
options,
routing,
};
bitrouter::acp_cli::serve(ctx).await
} else if let Some(text) = prompt {
eprintln!(
"note: `{cli} spawn <agent> -p` is a compatibility form; use \
`{cli} run <agent> <prompt>`.",
cli = bitrouter_sdk::invocation::name()
);
run_agent_prompt(
&source,
cfg,
&agent,
AgentPromptRequest {
prompt: &text,
routing,
turn_timeout,
no_wait,
result_schema: result_schema.as_deref(),
headless,
session: bitrouter::acp_cli::SessionSelection::New,
cwd: None,
},
)
.await
} else {
anyhow::bail!(
"spawn: choose a mode — `-p \"<prompt>\"` (NDJSON), \
`--serve` (ACP over stdio), or `--check` (preflight)."
);
}
}
Command::Cloud { action } => bitrouter::cloud::cli::run(action, output.format()).await,
Command::Skills { action } => bitrouter::skills::cli::run(action, output),
Command::Mcp { action } => mcp_cmd(action, output).await,
Command::WorkflowState { action } => workflow_state_cmd(action).await,
Command::Acp { cmd } => acp_cmd(cmd, output).await,
Command::Code { options } => {
run_code(options, remote_context, cli.context.as_deref()).await
}
Command::Tui { options } => {
eprintln!(
"note: `{cli} tui` is a compatibility alias; use `{cli} code`.",
cli = bitrouter_sdk::invocation::name()
);
run_code(options, remote_context, cli.context.as_deref()).await
}
Command::Chat {
agent,
turn_timeout,
routing,
config,
} => {
use std::io::IsTerminal as _;
eprintln!(
"note: `{cli} chat` is a compatibility alias; use `{cli} code {agent}`.",
cli = bitrouter_sdk::invocation::name()
);
if !std::io::stdin().is_terminal() || !std::io::stdout().is_terminal() {
let source = bitrouter::paths::resolve_config(config.as_deref())?;
let cfg = bitrouter::paths::load_config(&source).await?;
let agent = bitrouter::acp_cli::resolve_agent_id(&cfg, &agent)?;
return bitrouter::acp_cli::chat(bitrouter::acp_cli::SpawnContext {
source: &source,
config: cfg,
agent_id: &agent,
options: bitrouter::acp_cli::launch_options(turn_timeout),
routing,
})
.await;
}
run_code(
CodeArgs {
agent: Some(agent),
load: None,
resume: None,
turn_timeout,
routing,
config,
socket: None,
},
None,
None,
)
.await
}
Command::Update {
check,
tag,
stable,
restart: restart_after,
yes,
} => {
let source = bitrouter::paths::resolve_config(None)?;
let socket = resolve_client_socket_from(&source, None).await?;
let opts = bitrouter::update::UpdateOptions {
check,
tag,
stable,
restart: restart_after,
yes,
};
let outcome = bitrouter::update::run(opts, &socket).await?;
if outcome.restart_needed {
let log_path = resolve_log_path(source.home(), None);
restart(&source, &socket, &log_path).await?;
}
output.emit(&outcome.report)?;
Ok(())
}
}
}
async fn config_cmd(action: ConfigAction, output: &Output) -> Result<()> {
match action {
ConfigAction::MigrateRouters {
config,
candidate,
apply,
source_digest,
} => {
let source = bitrouter::paths::resolve_config(config.as_deref())?;
let path = require_policy_config_path(&source)?;
let report = bitrouter::router_migration::migrate(
path,
candidate.as_deref(),
apply,
source_digest.as_deref(),
)
.await?;
output.emit(&report)?;
Ok(())
}
ConfigAction::Validate { config } => {
let source = bitrouter::paths::resolve_config(config.as_deref())?;
let loaded = bitrouter::paths::load_config(&source).await?;
bitrouter::actions::session::prompt_commands(&loaded.chat)?;
let report = validate_config(&source).await?;
output.emit(&report)?;
if report.valid {
Ok(())
} else {
std::process::exit(1)
}
}
}
}
async fn settlement_api_key(
explicit_api_key: Option<&str>,
credentials_file: Option<&Path>,
api_base: &str,
) -> Result<String> {
if let Some(api_key) = explicit_api_key.filter(|api_key| !api_key.is_empty()) {
return Ok(api_key.to_owned());
}
let credentials_file = credentials_file.ok_or_else(|| {
anyhow::anyhow!("settlement requires a static BitRouter API key or credentials file")
})?;
let manager =
bitrouter_providers::hosted::account::manager::CredentialManager::new(credentials_file)
.context("build settlement credential manager")?;
manager
.resolve_api_key(None, Some(api_base))
.await
.map(|credential| credential.secret().to_owned())
.map_err(anyhow::Error::from)
}
async fn workflow_state_cmd(action: WorkflowStateAction) -> Result<()> {
match action {
WorkflowStateAction::ClassifierBakeoff {
fixtures,
submission,
output,
} => {
use bitrouter::workflow_state::classifier_bakeoff::ClassifierBakeoffArtifact;
use bitrouter::workflow_state::fixture::WorkflowTraceFixture;
use bitrouter::workflow_state::shadow_classifier::ShadowClassifierSubmission;
let fixtures = WorkflowTraceFixture::load_tree(&fixtures)
.with_context(|| format!("read classifier fixtures {}", fixtures.display()))?;
let submission = submission
.as_ref()
.map(ShadowClassifierSubmission::load_json)
.transpose()?;
let artifact = ClassifierBakeoffArtifact::build(&fixtures, submission)?;
artifact.write_json(&output)?;
println!(
"✓ wrote shadow classifier bake-off to {} (accepted: {}/{})",
output.display(),
artifact.report.accepted_count,
artifact.report.total_count
);
Ok(())
}
WorkflowStateAction::Bundle {
run_label,
traces,
cloud_usage,
outcomes,
policy_decisions,
output_dir,
} => {
use bitrouter::workflow_state::archive::{
CloudUsageRecord, TraceArchive, WorkflowRunArtifact,
};
use bitrouter::workflow_state::decision::PolicyDecisionRecord;
use bitrouter::workflow_state::real_trace::TraceSanitizer;
use bitrouter::workflow_state::reward::BenchmarkOutcomeRecord;
let traces = TraceArchive::read_jsonl(&traces)
.with_context(|| format!("read workflow traces {}", traces.display()))?;
let usage = CloudUsageRecord::load_snapshot_jsonl(&cloud_usage)
.with_context(|| format!("read cloud usage {}", cloud_usage.display()))?;
let outcomes = match outcomes {
Some(path) => BenchmarkOutcomeRecord::load_jsonl(&path)
.with_context(|| format!("read benchmark outcomes {}", path.display()))?,
None => Vec::new(),
};
let decisions = match policy_decisions {
Some(path) => PolicyDecisionRecord::load_jsonl(&path)
.with_context(|| format!("read policy decisions {}", path.display()))?,
None => Vec::new(),
};
let artifact = WorkflowRunArtifact::write_bundle_with_decisions(
run_label,
&output_dir,
&traces,
&usage,
&outcomes,
&decisions,
&TraceSanitizer::default(),
)
.with_context(|| format!("write workflow bundle {}", output_dir.display()))?;
println!(
"✓ wrote workflow bundle to {} (traces: {}, reward matches: {})",
output_dir.display(),
artifact.trace_count,
artifact.reward_join.matched_trace_count
);
Ok(())
}
WorkflowStateAction::MeteringUsage {
database_url,
output,
impute_prices,
since,
until,
} => {
use bitrouter::metering::{
MeteringStore, MeteringUsageRecord, TimeWindow, UsagePriceOverride,
};
let impute_prices = impute_prices
.iter()
.map(|value| UsagePriceOverride::parse(value))
.collect::<std::result::Result<Vec<_>, _>>()
.map_err(anyhow::Error::from)?;
let window = match since {
Some(since) => {
let start = parse_rfc3339_utc(&since, "--since")?;
let end = match until {
Some(until) => parse_rfc3339_utc(&until, "--until")?,
None => chrono::Utc::now(),
};
TimeWindow::Custom { start, end }
}
None => TimeWindow::ThisMonth,
};
let db = bitrouter::db::connect(&database_url)
.await
.with_context(|| format!("connect metering database {database_url}"))?;
let mut records = MeteringStore::new(db)
.export_usage(window)
.await
.with_context(|| format!("export metering usage from {database_url}"))?;
MeteringUsageRecord::apply_price_overrides(&mut records, &impute_prices);
MeteringUsageRecord::write_jsonl(&output, &records)
.with_context(|| format!("write metering usage {}", output.display()))?;
println!(
"✓ wrote {} metering usage records to {}",
records.len(),
output.display()
);
Ok(())
}
WorkflowStateAction::ReliabilityReport {
database_url,
config,
output,
} => {
use bitrouter::adequacy::report::ReliabilityReport;
use bitrouter::adequacy::store::AdequacyStore;
let config_document = bitrouter_sdk::config::load(&config)
.await
.with_context(|| format!("load reliability config {}", config.display()))?;
let db = bitrouter::db::connect(&database_url)
.await
.with_context(|| format!("connect reliability database {database_url}"))?;
let rows = AdequacyStore::new(db)
.load_reliability_events()
.await
.with_context(|| format!("load reliability events from {database_url}"))?;
let report = ReliabilityReport::build(&config_document.policy_table.adequacy, &rows)
.context("replay provider reliability events")?;
report
.write(&output)
.with_context(|| format!("write reliability report {}", output.display()))?;
println!(
"✓ wrote {} reliability events to {}",
report.event_count,
output.display()
);
Ok(())
}
WorkflowStateAction::PolicyOracle {
traces,
cloud_usage,
policy_lock,
policy,
effective_cost_factor_ppm,
target_savings_ppm,
output,
} => {
use bitrouter::workflow_state::archive::{CloudUsageRecord, TraceArchive};
use bitrouter::workflow_state::counterfactual::analyze_policy_counterfactual;
let traces = TraceArchive::read_jsonl(&traces)
.with_context(|| format!("read workflow traces {}", traces.display()))?;
let usage = CloudUsageRecord::load_snapshot_jsonl(&cloud_usage)
.with_context(|| format!("read cloud usage {}", cloud_usage.display()))?;
let lock = bitrouter::policy_lock::load(&policy_lock)
.await
.with_context(|| format!("load policy lock {}", policy_lock.display()))?;
let definition = lock.document.policies.get(&policy).ok_or_else(|| {
anyhow::anyhow!(
"policy {policy} does not exist in {}",
policy_lock.display()
)
})?;
let report = analyze_policy_counterfactual(
&traces,
&usage,
definition,
effective_cost_factor_ppm,
&target_savings_ppm,
)
.context("analyze policy counterfactual")?;
let encoded = serde_json::to_vec_pretty(&report)
.context("serialize policy counterfactual report")?;
std::fs::write(&output, encoded)
.with_context(|| format!("write policy oracle report {}", output.display()))?;
println!(
"✓ analyzed {} requests: {:.1}% cost coverage, {:.1}% projected savings -> {}",
report.trace_count,
f64::from(report.cost_weighted_coverage_ppm) / 10_000.0,
f64::from(report.projected_savings_ppm) / 10_000.0,
output.display()
);
Ok(())
}
WorkflowStateAction::ReconcileMetering {
database_url,
api_base,
api_key_env,
credentials_file,
request_ids,
prices,
max_attempts,
poll_interval_ms,
} => {
use std::time::Duration;
use bitrouter::cloud::settlement::SettlementClient;
use bitrouter::metering::{MeteringStore, UsagePriceOverride, reconcile_requests};
let prices = prices
.iter()
.map(|value| UsagePriceOverride::parse(value))
.collect::<std::result::Result<Vec<_>, _>>()
.map_err(anyhow::Error::from)?;
let explicit_api_key = std::env::var(&api_key_env)
.ok()
.filter(|api_key| !api_key.is_empty());
let api_key = match settlement_api_key(
explicit_api_key.as_deref(),
credentials_file.as_deref(),
&api_base,
)
.await
{
Ok(api_key) => api_key,
Err(error) if credentials_file.is_none() => {
return Err(error.context(format!(
"provide a non-empty {api_key_env} environment variable or a credentials file containing a static BitRouter API key"
)));
}
Err(error) => return Err(error),
};
let client = SettlementClient::new(&api_base, api_key)
.context("build request settlement client")?;
let db = bitrouter::db::connect(&database_url)
.await
.with_context(|| format!("connect metering database {database_url}"))?;
bitrouter::db::run_migrations(&db)
.await
.context("run metering reconciliation migrations")?;
let summary = reconcile_requests(
&MeteringStore::new(db),
&client,
&request_ids,
&prices,
max_attempts,
Duration::from_millis(poll_interval_ms),
)
.await
.context("reconcile selected metering requests")?;
if !summary.accepted() {
anyhow::bail!(
"metering reconciliation failed closed: requested={}, computed={}, \
not_charged={}, unknown={}, attempts={}",
summary.requested,
summary.computed,
summary.not_charged,
summary.unknown,
summary.attempts
);
}
println!(
"✓ reconciled {} metering requests (computed: {}, not charged: {}, attempts: {})",
summary.requested, summary.computed, summary.not_charged, summary.attempts
);
Ok(())
}
WorkflowStateAction::ApplyRewardFeedback {
database_url,
traces,
cloud_usage,
outcomes,
policy_decisions,
} => {
use bitrouter::eval::EvalService;
use bitrouter::eval::store::EvalStore;
use bitrouter::workflow_state::archive::{
CloudUsageRecord, TraceArchive, WorkflowRunArtifact,
};
use bitrouter::workflow_state::decision::PolicyDecisionRecord;
use bitrouter::workflow_state::reward::BenchmarkOutcomeRecord;
use bitrouter::workflow_state::reward_feedback::import_semantic_reward_feedback;
let traces = TraceArchive::read_jsonl(&traces)
.with_context(|| format!("read workflow traces {}", traces.display()))?;
let outcomes = BenchmarkOutcomeRecord::load_jsonl(&outcomes)
.with_context(|| format!("read benchmark outcomes {}", outcomes.display()))?;
let usage = CloudUsageRecord::load_snapshot_jsonl(&cloud_usage)
.with_context(|| format!("read cloud usage {}", cloud_usage.display()))?;
let decisions = PolicyDecisionRecord::load_jsonl(&policy_decisions)
.with_context(|| format!("read policy decisions {}", policy_decisions.display()))?;
WorkflowRunArtifact::validate_reward_feedback_integrity(
&traces, &usage, &outcomes, &decisions,
)
.context("validate reward feedback admission")?;
let artifact = WorkflowRunArtifact::build_with_decisions(
"reward-feedback",
&traces,
&usage,
&outcomes,
&decisions,
)
.context("build workflow artifact for reward feedback")?;
let db = bitrouter::db::connect(&database_url)
.await
.with_context(|| format!("connect eval exchange database {database_url}"))?;
bitrouter::db::run_migrations(&db).await?;
let summary = import_semantic_reward_feedback(
&EvalService::new(EvalStore::new(db), Default::default()),
&artifact.semantic_policy_transition_candidates,
)
.await
.context("import reward feedback into generic eval exchange")?;
println!(
"✓ imported reward feedback: {} candidates, {} admitted eval results, {} skipped",
summary.candidate_count, summary.admitted_count, summary.skipped_count
);
for eval_id in summary.eval_ids {
println!(" admitted {eval_id}");
}
Ok(())
}
}
}
fn parse_rfc3339_utc(value: &str, flag: &str) -> Result<chrono::DateTime<chrono::Utc>> {
chrono::DateTime::parse_from_rfc3339(value)
.map(|dt| dt.with_timezone(&chrono::Utc))
.with_context(|| format!("{flag} must be RFC3339, got {value:?}"))
}
async fn validate_config(source: &bitrouter::paths::ConfigSource) -> Result<ValidateReport> {
use bitrouter::paths::ConfigSource;
let path = match source {
ConfigSource::File(p) => p,
ConfigSource::Default { .. } => anyhow::bail!(
"no config file found to validate — looked in ./bitrouter.yaml, \
$BITROUTER_HOME, and ~/.bitrouter. Pass --config <path>."
),
};
let raw = tokio::fs::read_to_string(path)
.await
.with_context(|| format!("reading {}", path.display()))?;
let missing: std::cell::RefCell<std::collections::BTreeSet<String>> =
std::cell::RefCell::new(std::collections::BTreeSet::new());
let parsed = config::parse_with(&raw, |name| {
Some(config::env_lookup(name).unwrap_or_else(|| {
missing.borrow_mut().insert(name.to_string());
"https://env-placeholder.invalid".to_string()
}))
});
let missing = missing.into_inner();
match parsed {
Ok(cfg) => match bitrouter::policy_lock::load_for_config(&cfg, Some(path)).await {
Ok(_) => Ok(ValidateReport::valid(
path.display().to_string(),
cfg.providers.len(),
cfg.models.len(),
cfg.presets.len(),
cfg.variants.len(),
missing
.into_iter()
.map(|name| UnsetVar { unset_env: name })
.collect(),
)
.with_routers(cfg.routers.len())
.with_ignored_config(bitrouter::assemble::ignored_config_file_warnings(&cfg))),
Err(error) => Ok(ValidateReport::invalid(
path.display().to_string(),
error.to_string(),
)),
},
Err(e) => Ok(ValidateReport::invalid(
path.display().to_string(),
e.to_string(),
)),
}
}
async fn mcp_cmd(action: McpAction, output: &Output) -> Result<()> {
match action {
McpAction::Check { server, config } => {
let source = bitrouter::paths::resolve_config(config.as_deref())?;
let config = bitrouter::paths::load_config(&source).await?;
let servers = bitrouter::tools::check(&config, server.as_deref())
.await
.map_err(bitrouter_sdk::BitrouterError::bad_request)?
.into_iter()
.map(|row| match row.outcome {
Ok(tools) => McpCheckRow {
server: row.server,
transport: row.transport,
ok: true,
latency_ms: row.latency.as_millis(),
capabilities: vec!["tools".to_string()],
tools: tools.into_iter().map(|tool| tool.name).collect(),
error: None,
},
Err(error) => McpCheckRow {
server: row.server,
transport: row.transport,
ok: false,
latency_ms: row.latency.as_millis(),
capabilities: Vec::new(),
tools: Vec::new(),
error: Some(error),
},
})
.collect();
let report = McpCheckReport { servers };
let ok = report.servers.iter().all(|server| server.ok);
output.emit(&report)?;
if ok { Ok(()) } else { exit_with(1) }
}
McpAction::Search { query, limit } => {
eprintln!(
"note: `{cli} mcp search` is deprecated and hidden from help.",
cli = bitrouter_sdk::invocation::name()
);
let outcome = bitrouter::mcp_registry::RegistryClient::new()?
.servers(Some(&query), limit)
.await?;
output.emit(&mcp_registry_report(outcome)?)?;
Ok(())
}
McpAction::List { limit } => {
eprintln!(
"note: `{cli} mcp list` is deprecated and hidden from help.",
cli = bitrouter_sdk::invocation::name()
);
let outcome = bitrouter::mcp_registry::RegistryClient::new()?
.servers(None, limit)
.await?;
output.emit(&mcp_registry_report(outcome)?)?;
Ok(())
}
McpAction::Add { name } => {
eprintln!(
"note: `{cli} mcp add` is deprecated and hidden from help.",
cli = bitrouter_sdk::invocation::name()
);
let outcome = bitrouter::mcp_registry::RegistryClient::new()?
.latest(&name)
.await?;
match bitrouter::mcp_registry::add_stub(&outcome.data) {
Ok(stub) => {
output.emit(&McpAddReport {
name,
id: stub.id,
yaml: stub.yaml,
})?;
Ok(())
}
Err(e) => anyhow::bail!(e),
}
}
}
}
fn mcp_registry_report(
outcome: bitrouter::mcp_registry::FetchOutcome<Vec<bitrouter::mcp_registry::ServerEntry>>,
) -> Result<McpRegistryReport> {
let servers = bitrouter::mcp_registry::registry_rows(&outcome.data)
.into_iter()
.map(|row| McpRegistryRow {
name: row.name,
version: row.version,
install: row.install.to_string(),
description: row.description,
})
.collect();
Ok(McpRegistryReport {
servers,
from_cache: outcome.from_cache,
})
}
async fn resolve_client_socket(config: Option<&Path>, socket: Option<&Path>) -> Result<PathBuf> {
if let Some(s) = socket {
return Ok(s.to_path_buf());
}
if let Some(located) = bitrouter::daemon_locator::locate(config).await? {
return Ok(located.socket().to_path_buf());
}
let source = bitrouter::daemon_locator::selected_source(config)?;
match &source {
bitrouter::paths::ConfigSource::File(path) => {
let socket_str = match config::load(path).await {
Ok(cfg) => cfg.server.control_socket,
Err(_) => daemon::DEFAULT_CONTROL_SOCKET.to_string(),
};
Ok(daemon::resolve_socket_path(path, &socket_str))
}
bitrouter::paths::ConfigSource::Default { home } => Ok(home.join("bitrouter.sock")),
}
}
const DEFAULT_LOG_FILTER: &str = "info";
fn resolve_env_filter(
config_log_level: Option<&str>,
) -> (tracing_subscriber::EnvFilter, Option<String>) {
let rust_log = std::env::var(tracing_subscriber::EnvFilter::DEFAULT_ENV).ok();
resolve_env_filter_from(rust_log.as_deref(), config_log_level)
}
fn resolve_env_filter_from(
rust_log: Option<&str>,
config_log_level: Option<&str>,
) -> (tracing_subscriber::EnvFilter, Option<String>) {
let non_blank = |s: &&str| !s.trim().is_empty();
let (source, raw) = match (
rust_log.filter(non_blank),
config_log_level.filter(non_blank),
) {
(Some(raw), _) => (tracing_subscriber::EnvFilter::DEFAULT_ENV, raw),
(None, Some(level)) => ("server.log_level", level),
(None, None) => return (tracing_subscriber::EnvFilter::new(DEFAULT_LOG_FILTER), None),
};
match parse_log_filter(raw) {
Ok(filter) => (filter, None),
Err(reason) => (
tracing_subscriber::EnvFilter::new(DEFAULT_LOG_FILTER),
Some(format!(
"invalid {source} value {raw:?}: {reason} — falling back to `{DEFAULT_LOG_FILTER}`"
)),
),
}
}
fn parse_log_filter(raw: &str) -> std::result::Result<tracing_subscriber::EnvFilter, String> {
let raw = raw.trim();
if !raw.contains('=')
&& !raw.contains(',')
&& raw
.parse::<tracing_subscriber::filter::LevelFilter>()
.is_err()
{
return Err(
"expected a level (`trace`, `debug`, `info`, `warn`, `error`, `off`) or an \
`EnvFilter` directive list such as `info,bitrouter=debug`"
.to_string(),
);
}
tracing_subscriber::EnvFilter::try_new(raw).map_err(|e| e.to_string())
}
fn init_basic_tracing_subscriber() {
let (env_filter, warning) = resolve_env_filter(None);
tracing_subscriber::fmt()
.with_writer(std::io::stderr)
.with_env_filter(env_filter)
.init();
if let Some(warning) = warning {
tracing::warn!("{warning}");
}
}
fn init_stderr_tracing_subscriber() {
let (env_filter, warning) = resolve_env_filter(None);
tracing_subscriber::fmt()
.with_writer(std::io::stderr)
.with_env_filter(env_filter)
.init();
if let Some(warning) = warning {
tracing::warn!("{warning}");
}
}
fn session_log_path() -> anyhow::Result<PathBuf> {
let dir = bitrouter::paths::runtime_home()?.join("logs");
std::fs::create_dir_all(&dir)?;
let stamp = chrono::Local::now().format("%Y%m%dT%H%M%S");
Ok(dir.join(format!("session-{stamp}-{}.log", std::process::id())))
}
fn init_session_log_tracing_subscriber(also_stderr: bool) -> Option<PathBuf> {
use tracing_subscriber::fmt::writer::MakeWriterExt;
let env_filter = || {
tracing_subscriber::EnvFilter::try_from_default_env()
.unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info"))
};
let opened = session_log_path().and_then(|path| {
let file = std::fs::OpenOptions::new()
.create(true)
.append(true)
.open(&path)?;
Ok((path, file))
});
match opened {
Ok((path, file)) => {
let file = std::sync::Arc::new(file);
if also_stderr {
tracing_subscriber::fmt()
.with_ansi(false)
.with_writer(std::io::stderr.and(file))
.with_env_filter(env_filter())
.init();
} else {
tracing_subscriber::fmt()
.with_ansi(false)
.with_writer(file)
.with_env_filter(env_filter())
.init();
}
Some(path)
}
Err(e) => {
init_stderr_tracing_subscriber();
tracing::warn!("no session log ({e}); logging to stderr only");
None
}
}
}
fn init_serve_tracing_subscriber(
exporter: Option<&bitrouter_telemetry::otel::OtelExporter>,
config_log_level: &str,
) {
use tracing_subscriber::layer::SubscriberExt;
use tracing_subscriber::util::SubscriberInitExt;
let (env_filter, warning) = resolve_env_filter(Some(config_log_level));
let registry = tracing_subscriber::registry()
.with(env_filter)
.with(tracing_subscriber::fmt::layer());
match exporter {
Some(exp) => registry
.with(bitrouter_telemetry::otel::subscriber::tracing_subscriber_layer(exp))
.init(),
None => registry.init(),
}
if let Some(warning) = warning {
tracing::warn!("{warning}");
}
}
fn resolve_log_path(home: &Path, log: Option<&Path>) -> PathBuf {
if let Some(l) = log {
return l.to_path_buf();
}
home.join("bitrouter.log")
}
async fn resolve_client_socket_from(
source: &bitrouter::paths::ConfigSource,
socket: Option<&Path>,
) -> Result<PathBuf> {
if let Some(s) = socket {
return Ok(s.to_path_buf());
}
if let Some(located) = bitrouter::daemon_locator::locate_source(source).await? {
return Ok(located.socket().to_path_buf());
}
match source {
bitrouter::paths::ConfigSource::File(path) => {
let cfg = config::load(path)
.await
.with_context(|| format!("loading {}", path.display()))?;
Ok(daemon::resolve_socket_path(
path,
&cfg.server.control_socket,
))
}
bitrouter::paths::ConfigSource::Default { home } => Ok(home.join("bitrouter.sock")),
}
}
fn reject_remote_local_target_flags(config: Option<&Path>, socket: Option<&Path>) -> Result<()> {
bitrouter::administration_target::reject_remote_local_flags(config, socket)
}
fn emit_operation_result(output: &Output, report: &OperationReport) -> Result<()> {
output.emit(report)?;
if report.succeeded() {
return Ok(());
}
let status = match report.status {
OperationStatus::Running => "is still running",
OperationStatus::Failed => "failed",
OperationStatus::PartiallyApplied => "partially applied",
OperationStatus::Unknown => "has an unknown outcome",
OperationStatus::Succeeded => "succeeded",
};
anyhow::bail!(
"reload {status}; inspect `{cli} --context <name> operations show {} --instance {}` before retrying",
report.request_id,
report.server_instance_id,
cli = bitrouter_sdk::invocation::name()
)
}
fn validate_remote_invocation(command: &Command, remote: bool) -> Result<()> {
if !remote {
return Ok(());
}
match command {
Command::Agents {
action:
Some(AgentsAction::List {
remote: true, ..
}),
} => Err(bitrouter_sdk::BitrouterError::bad_request(
"`agents list --remote` fetches the external ACP registry and cannot run against a remote BitRouter context",
)
.into()),
Command::Agents {
action:
Some(AgentsAction::List {
remote: false,
config,
socket,
}),
} => reject_remote_local_target_flags(config.as_deref(), socket.as_deref()),
Command::Agents {
action: Some(AgentsAction::Check { .. }),
} => Err(bitrouter_sdk::BitrouterError::bad_request(
"`agents check` launches an agent and is unavailable for a remote context",
)
.into()),
Command::Code { options } | Command::Tui { options } if options.agent.is_none() => {
reject_remote_local_target_flags(options.config.as_deref(), options.socket.as_deref())
}
_ => {
if let Some(leaf) = remote_cli_leaf(command) {
bitrouter::remote_control::inventory::by_cli(leaf).ok_or_else(|| {
anyhow::anyhow!("control inventory has no action for CLI leaf `{leaf}`")
})?;
let (config, socket) = remote_target_flags(command).ok_or_else(|| {
anyhow::anyhow!("remote CLI leaf `{leaf}` has no target-flag contract")
})?;
return reject_remote_local_target_flags(config, socket);
}
if let Some(leaf) = remote_resource_leaf(command) {
bitrouter::remote_control::inventory::RESOURCES
.iter()
.find(|resource| resource.cli_leaf == Some(leaf))
.ok_or_else(|| {
anyhow::anyhow!("control inventory has no resource for CLI leaf `{leaf}`")
})?;
return Ok(());
}
Err(bitrouter_sdk::BitrouterError::bad_request(
"this command is local-only for a remote context",
)
.into())
}
}
}
fn remote_cli_leaf(command: &Command) -> Option<&'static str> {
match command {
Command::Reload { .. } => Some("reload"),
Command::Status { requests: true, .. } | Command::Requests { .. } => Some("requests"),
Command::Status {
requests: false, ..
} => Some("status"),
Command::Models { .. } => Some("models"),
Command::Route { .. } => Some("route"),
Command::Providers {
action: ProviderAction::List { .. },
} => Some("providers list"),
Command::Observe {
action: ObserveAction::Status { .. },
} => Some("observe status"),
Command::Policy {
action: PolicyAction::Status { .. },
} => Some("policy status"),
Command::Policy {
action: PolicyAction::Show { .. },
} => Some("policy show"),
Command::Agents {
action: Some(AgentsAction::List { remote: false, .. }),
} => Some("agents list"),
Command::Checks { action: None, .. } => Some("checks"),
Command::Checks {
action: Some(ChecksAction::Probe { .. }),
..
} => Some("checks probe"),
Command::Checks {
action: Some(ChecksAction::Receipts { .. }),
..
} => Some("checks receipts"),
Command::Checks {
action: Some(ChecksAction::Receipt { .. }),
..
} => Some("checks receipt"),
_ => None,
}
}
fn remote_resource_leaf(command: &Command) -> Option<&'static str> {
match command {
Command::Operations {
action: OperationsAction::Show { .. },
} => Some("operations show"),
_ => None,
}
}
fn remote_target_flags(command: &Command) -> Option<(Option<&Path>, Option<&Path>)> {
match command {
Command::Reload { config, socket }
| Command::Status { config, socket, .. }
| Command::Requests { config, socket, .. }
| Command::Route { config, socket, .. }
| Command::Providers {
action: ProviderAction::List { config, socket },
}
| Command::Observe {
action: ObserveAction::Status { config, socket },
}
| Command::Policy {
action:
PolicyAction::Status { config, socket, .. } | PolicyAction::Show { config, socket, .. },
}
| Command::Agents {
action: Some(AgentsAction::List { config, socket, .. }),
}
| Command::Checks { config, socket, .. } => Some((config.as_deref(), socket.as_deref())),
Command::Models { config, .. } => Some((config.as_deref(), None)),
_ => None,
}
}
async fn serve(source: &bitrouter::paths::ConfigSource) -> Result<()> {
if let Some(located) = bitrouter::daemon_locator::locate_source(source).await? {
anyhow::bail!(
"bitrouter is already running (pid {}); use `restart` or `stop` first",
located.pid()
);
}
let home = source.home();
bitrouter::paths::ensure_home_directory(home)?;
std::env::set_current_dir(home)
.with_context(|| format!("chdir to bitrouter home {}", home.display()))?;
let startup_configuration = bitrouter::reload::load_configuration_baseline(source).await?;
let mut cfg = startup_configuration.config().clone();
bitrouter::claude_code::enable_if_logged_in(&mut cfg);
bitrouter::merge_registry_into(&mut cfg).await;
announce_zero_config(source, &cfg);
maybe_announce_telemetry(home);
let listen = cfg.server.listen.clone();
let socket_path = daemon::socket_path_for(source, &cfg);
let pid_path = pid_path_for(&socket_path);
let remote_control = bitrouter::remote_control::ControlServer::from_config(
&cfg.control,
source.clone(),
socket_path.clone(),
)?;
let config_path_for_reload = match source {
bitrouter::paths::ConfigSource::File(path) => Some(path.as_path()),
bitrouter::paths::ConfigSource::Default { .. } => None,
};
let assembled = bitrouter::build_app_with_path(&cfg, config_path_for_reload).await?;
init_serve_tracing_subscriber(assembled.otel_exporter.as_deref(), &cfg.server.log_level);
if let Some(msg) = &assembled.otel_init_error {
tracing::error!("{msg}");
}
for msg in &assembled.ignored_config {
tracing::warn!("{msg}");
}
let workflow_trace_capture =
bitrouter::workflow_state::real_trace::capture_from_env().map_err(anyhow::Error::from)?;
if workflow_trace_capture.is_some() {
tracing::info!(
env = bitrouter::workflow_state::real_trace::WORKFLOW_TRACE_JSONL_ENV,
"workflow trace capture enabled"
);
}
let app = Arc::new(assembled.app);
let eval_router = bitrouter::eval::api::router(
assembled.eval_service.clone(),
assembled.db.clone(),
cfg.server.skip_auth,
);
let policy_store = assembled.policy_store;
let trajectory_outbox_for_shutdown = assembled.trajectory_outbox_publisher.clone();
let observe_provider = assembled.observe;
let observe_for_shutdown = observe_provider.clone();
let reload_source = match source {
bitrouter::paths::ConfigSource::File(path) => {
bitrouter::reload::ReloadSource::File(path.clone())
}
bitrouter::paths::ConfigSource::Default { .. } => bitrouter::reload::ReloadSource::Default,
};
let administration = bitrouter::actions::administration::Administration {
source: source.clone(),
routing: assembled.routing_table.clone(),
policy: assembled.policy_runtime.clone(),
observe: observe_provider.clone(),
request_checks: Some(assembled.request_checks.clone()),
};
let acp_runtime_for_control = assembled.acp_runtime.clone();
let reloader = bitrouter::reload::AppReloader::new(
policy_store.clone(),
assembled.routing_table,
assembled.upstream_executor,
reload_source,
)
.with_startup_configuration(startup_configuration)
.with_policy_runtime(assembled.policy_runtime)
.with_policy_table_router(assembled.policy_table_router);
let server_instance_id = daemon::DaemonReloader::reload_state(&reloader)
.ok_or_else(|| anyhow::anyhow!("daemon reload state is unavailable at startup"))?
.server_instance_id;
let reloader: Arc<dyn daemon::DaemonReloader> = Arc::new(reloader);
let remote_control = match remote_control {
Some(server) => Some(
server
.with_administration(administration.clone())
.with_reloader(reloader.clone())
.bind()
.await?,
),
None => None,
};
daemon::write_pid_file(&pid_path).await?;
println!(
"bitrouter {} — serving on {listen} (control: {})",
bitrouter::VERSION,
socket_path.display()
);
if let Some(server) = &remote_control {
println!(
"bitrouter {} — remote control on {}",
bitrouter::VERSION,
server.listen()
);
}
let http_app = app.clone();
let http_listen = listen.clone();
let otel_router_wrapper = assembled
.otel_exporter
.as_deref()
.map(bitrouter_telemetry::otel::http_layer::router_wrapper);
let (http_shutdown_tx, http_shutdown_rx) = tokio::sync::oneshot::channel();
let http = async move {
let (inference_shutdown_tx, inference_shutdown_rx) = tokio::sync::oneshot::channel();
let (remote_shutdown_tx, remote_shutdown_rx) = tokio::sync::oneshot::channel();
let otel_wrapper = move |router: axum::Router| match &otel_router_wrapper {
Some(wrapper) => wrapper(router),
None => router,
};
let inference_shutdown = async move {
let _ = inference_shutdown_rx.await;
};
let inference = async move {
match workflow_trace_capture {
Some(capture) => {
let workflow_wrapper = capture.router_wrapper();
let eval_router = eval_router.clone();
http_app
.serve_with_router_wrapper_and_shutdown(
&http_listen,
move |router| {
workflow_wrapper(otel_wrapper(router.merge(eval_router.clone())))
},
inference_shutdown,
)
.await
}
None => {
http_app
.serve_with_router_wrapper_and_shutdown(
&http_listen,
move |router| otel_wrapper(router.merge(eval_router.clone())),
inference_shutdown,
)
.await
}
}
.map_err(anyhow::Error::from)
};
let remote = async move {
match remote_control {
Some(server) => {
server
.serve_with_shutdown(async move {
let _ = remote_shutdown_rx.await;
})
.await
}
None => {
let _ = remote_shutdown_rx.await;
Ok(())
}
}
};
let mut inference = Box::pin(inference);
let mut remote = Box::pin(remote);
let mut shutdown = Box::pin(async move {
let _ = http_shutdown_rx.await;
});
tokio::select! {
result = &mut inference => {
let _ = remote_shutdown_tx.send(());
remote.await?;
result
}
result = &mut remote => {
let _ = inference_shutdown_tx.send(());
inference.await?;
result
}
_ = &mut shutdown => {
let _ = inference_shutdown_tx.send(());
let _ = remote_shutdown_tx.send(());
let (inference_result, remote_result) = tokio::join!(inference, remote);
inference_result?;
remote_result
}
}
};
let locator_socket = socket_path.clone();
let stop_socket = socket_path.clone();
let locator_source = source.clone();
let locator_instance = server_instance_id;
let (control_shutdown_tx, control_shutdown_rx) = tokio::sync::oneshot::channel();
let control = daemon::run_control_socket_with_acp_runtime_and_administration(
socket_path,
app.clone(),
listen,
reloader.clone(),
observe_provider,
daemon::AcpControlPlane {
runtime: acp_runtime_for_control,
metering: bitrouter::metering::MeteringStore::new(assembled.db.clone()),
inventory: Some(assembled.evolution.inventory()),
evolution: Some(assembled.evolution.clone()),
},
Some(administration),
);
let control = async move {
let mut control = Box::pin(control);
loop {
let probe = bitrouter::daemon_locator::endpoint_matches(
&locator_socket,
std::process::id(),
&locator_instance,
);
tokio::pin!(probe);
let matches = tokio::select! {
result = &mut control => return result,
matches = &mut probe => matches,
};
if matches {
let _locator = bitrouter::daemon_locator::publish(
&locator_source,
&locator_socket,
&locator_instance,
)?;
return control.await;
}
tokio::select! {
result = &mut control => return result,
_ = tokio::time::sleep(std::time::Duration::from_millis(10)) => {}
}
}
};
let control = async move {
let mut control = Box::pin(control);
tokio::select! {
result = &mut control => result,
_ = control_shutdown_rx => {
let stop = request_daemon_shutdown(&stop_socket);
tokio::pin!(stop);
let response = tokio::select! {
result = &mut control => return result,
response = &mut stop => response,
}?;
match response {
DaemonResponse::Ok => control.await,
DaemonResponse::Error { message } => anyhow::bail!(message),
response => anyhow::bail!(
"daemon returned an unexpected shutdown response: {response:?}"
),
}
}
}
};
let hup_reloader = reloader.clone();
let hup = async move {
#[cfg(unix)]
{
use tokio::signal::unix::{SignalKind, signal};
let mut hup = match signal(SignalKind::hangup()) {
Ok(s) => s,
Err(e) => return Err::<(), anyhow::Error>(anyhow::Error::from(e)),
};
loop {
if hup.recv().await.is_none() {
return Ok(());
}
match hup_reloader.reload().await {
Ok(()) => tracing::info!("SIGHUP — reload succeeded"),
Err(e) => tracing::warn!(error = %e, "SIGHUP reload failed"),
}
}
}
#[cfg(not(unix))]
{
let _keep = &hup_reloader;
std::future::pending::<()>().await;
Ok::<(), anyhow::Error>(())
}
};
let term = async {
#[cfg(unix)]
{
use tokio::signal::unix::{SignalKind, signal};
let mut sigint = signal(SignalKind::interrupt()).map_err(anyhow::Error::from)?;
let mut sigterm = signal(SignalKind::terminate()).map_err(anyhow::Error::from)?;
tokio::select! {
_ = sigint.recv() => tracing::info!("SIGINT — shutting down"),
_ = sigterm.recv() => tracing::info!("SIGTERM — shutting down"),
}
Ok::<(), anyhow::Error>(())
}
#[cfg(windows)]
{
use tokio::signal::windows;
let mut ctrl_c = windows::ctrl_c().map_err(anyhow::Error::from)?;
let mut ctrl_break = windows::ctrl_break().map_err(anyhow::Error::from)?;
let mut ctrl_close = windows::ctrl_close().map_err(anyhow::Error::from)?;
let mut ctrl_shutdown = windows::ctrl_shutdown().map_err(anyhow::Error::from)?;
tokio::select! {
_ = ctrl_c.recv() => tracing::info!("Ctrl-C — shutting down"),
_ = ctrl_break.recv() => tracing::info!("Ctrl-Break — shutting down"),
_ = ctrl_close.recv() => tracing::info!("console close — shutting down"),
_ = ctrl_shutdown.recv() => tracing::info!("system shutdown — shutting down"),
}
Ok::<(), anyhow::Error>(())
}
};
let evolution_stop = tokio_util::sync::CancellationToken::new();
let evolution_worker = app.language_model().cloned().map(|pipeline| {
let worker =
bitrouter::evolution::scheduler::EvolutionScheduler::new(assembled.evolution.clone());
let stop = evolution_stop.clone();
tokio::spawn(async move { worker.run(pipeline, stop).await })
});
let control = async {
let result = control.await;
evolution_stop.cancel();
result
};
let term = async {
let result = term.await;
evolution_stop.cancel();
result
};
let result = supervise_http_shutdown(
http,
control,
hup,
term,
http_shutdown_tx,
control_shutdown_tx,
)
.await;
evolution_stop.cancel();
if let Some(worker) = evolution_worker
&& worker.await.is_err()
{
tracing::warn!("checkpoint evolution worker ended unexpectedly");
}
if let Some(publisher) = trajectory_outbox_for_shutdown {
match publisher.drain_after_active_worker().await {
Ok(summary) if summary.failed > 0 => tracing::warn!(
attempted = summary.attempted,
delivered = summary.delivered,
failed = summary.failed,
"trajectory outbox drain completed with poison items still pending"
),
Ok(_) => {}
Err(_) => tracing::warn!(
reason = "shutdown_drain_failed",
"trajectory outbox shutdown drain failed"
),
}
}
observe_for_shutdown.shutdown().await;
daemon::remove_pid_file(&pid_path).await;
result
}
async fn start(
source: &bitrouter::paths::ConfigSource,
log_path: &Path,
action: &'static str,
) -> Result<DaemonActionReport> {
bitrouter::paths::ensure_home_directory(source.home())?;
if let Some(located) = bitrouter::daemon_locator::locate_source(source).await? {
anyhow::bail!(
"bitrouter is already running (pid {}); use `restart` or `stop` first",
located.pid()
);
}
let cfg_socket_path: Option<PathBuf> = match source {
bitrouter::paths::ConfigSource::File(path) => match config::load(path).await {
Ok(cfg) => Some(daemon::socket_path_for(source, &cfg)),
Err(_) => None,
},
bitrouter::paths::ConfigSource::Default { home } => Some(home.join("bitrouter.sock")),
};
if let Some(socket) = &cfg_socket_path {
let pid_path = pid_path_for(socket);
if let Some(pid) = daemon::read_pid_file(&pid_path).await {
if process_is_alive(pid) {
anyhow::bail!(
"bitrouter is already running (pid {pid}); use `restart` or `stop` first"
);
}
daemon::remove_pid_file(&pid_path).await;
}
}
let outcome = daemon::start_and_wait(
source,
log_path,
cfg_socket_path.as_deref(),
daemon::DAEMON_READY_TIMEOUT,
)
.await?;
match outcome {
daemon::DaemonStartOutcome::Ready(info) => Ok(DaemonActionReport::started(
action,
"started",
info.pid,
info.listen,
info.models,
log_path.display().to_string(),
)),
daemon::DaemonStartOutcome::NotReadyInTime { pid } => {
let p = bitrouter::style::Palette::for_stderr();
eprintln!(
"{cyan}note:{reset} bitrouter daemon started (pid {pid}) but has not become \
ready after {secs}s — check logs at {log}",
cyan = p.cyan,
reset = p.reset,
secs = daemon::DAEMON_READY_TIMEOUT.as_secs(),
log = log_path.display(),
);
Ok(DaemonActionReport::not_ready(
action,
pid,
log_path.display().to_string(),
))
}
daemon::DaemonStartOutcome::Exited { status, log_tail } => {
daemon::eprint_failure_log(log_path, &log_tail);
anyhow::bail!("daemon exited during startup ({status})")
}
}
}
fn announce_zero_config(
source: &bitrouter::paths::ConfigSource,
cfg: &bitrouter_sdk::config::Config,
) {
if !source.is_default() {
return;
}
let enabled: Vec<&str> = cfg.providers.keys().map(String::as_str).collect();
if enabled.is_empty() {
print_onboarding_hint();
} else {
bitrouter::error_report::info(format_args!(
"zero-config mode — auto-enabled providers: {}",
enabled.join(", ")
));
}
}
fn maybe_announce_telemetry(home: &std::path::Path) {
match bitrouter::paths::mark_telemetry_notice_shown(home) {
Ok(true) => {}
Ok(false) => return,
Err(e) => {
tracing::debug!("telemetry notice sentinel: {e:#}");
return;
}
}
let p = bitrouter::style::Palette::for_stderr();
eprintln!(
"{cyan}{bold}info:{reset} optional usage telemetry is available — and OFF by default.",
cyan = p.cyan,
bold = p.bold,
reset = p.reset,
);
eprintln!();
eprintln!(" Nothing is sent unless you opt in. Two levels are offered:");
eprintln!(
" • metadata — model, tokens, latency, finish reason, routing (no message content)"
);
eprintln!(" • full — the above plus request + response message content");
eprintln!();
eprintln!(" Enable it under plugins.bitrouter-telemetry.telemetry in your config:");
eprintln!();
eprintln!(" plugins:");
eprintln!(" bitrouter-telemetry:");
eprintln!(" telemetry:");
eprintln!(" enabled: true");
eprintln!(" level: metadata # or: full");
eprintln!();
eprintln!(" Remove the block (or set enabled: false) to turn it off again.");
eprintln!();
}
fn print_onboarding_hint() {
let p = bitrouter::style::Palette::for_stderr();
let cli = bitrouter_sdk::invocation::name();
eprintln!(
"{cyan}{bold}info:{reset} no providers are configured yet. Choose one:",
cyan = p.cyan,
bold = p.bold,
reset = p.reset,
);
eprintln!();
eprintln!(" 1. Sign in to BitRouter Cloud — one account covers every model:");
eprintln!();
eprintln!(" {cli} cloud login");
eprintln!(" {cli} cloud --help # manage keys, usage, policies, billing");
eprintln!();
eprintln!(" 2. Or paste a BitRouter API key:");
eprintln!();
eprintln!(" export BITROUTER_API_KEY=brk_…");
eprintln!();
eprintln!(" 3. Or use a provider you already pay for, locally:");
eprintln!();
eprintln!(" {cli} providers login claude-code # Claude Pro/Max subscription");
eprintln!(" {cli} providers login github-copilot # GitHub Copilot subscription");
eprintln!(" {cli} providers login openai-codex # ChatGPT subscription");
eprintln!();
eprintln!(" …or set an API-key env var:");
eprintln!();
let env_vars = other_provider_env_var_hints();
for var in &env_vars {
eprintln!(" export {var}=…");
}
eprintln!();
}
fn other_provider_env_var_hints() -> Vec<String> {
let mut vars: Vec<String> = bitrouter_providers::zero_config_env_var_providers()
.into_iter()
.map(|(_, env)| env)
.filter(|v| v != "BITROUTER_API_KEY")
.collect();
vars.sort();
vars.dedup();
vars
}
async fn stop(socket: &Path) -> Result<DaemonActionReport> {
match daemon::send_command(socket, &DaemonCommand::Stop).await? {
DaemonResponse::Ok => Ok(DaemonActionReport::simple("stop", "stopped")),
DaemonResponse::Error { message } => Err(anyhow::anyhow!(message)),
other => Err(anyhow::anyhow!("unexpected response: {other:?}")),
}
}
const RESTART_GRACE_PERIOD: std::time::Duration = std::time::Duration::from_secs(30);
const RESTART_CLEANUP_PERIOD: std::time::Duration = std::time::Duration::from_secs(2);
const RESTART_POLL_INTERVAL: std::time::Duration = std::time::Duration::from_millis(50);
struct RestartRelease {
ready: bool,
}
#[derive(Clone, Copy)]
struct RestartTarget {
pid: u32,
}
async fn wait_for_restart_condition(
timeout: std::time::Duration,
mut ready: impl FnMut() -> bool,
) -> bool {
let deadline = tokio::time::Instant::now() + timeout;
loop {
if ready() {
return true;
}
let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
if remaining.is_zero() {
return ready();
}
tokio::time::sleep(RESTART_POLL_INTERVAL.min(remaining)).await;
}
}
async fn await_restart_release<IsAlive, EndpointInUse>(
socket: &Path,
target: Option<RestartTarget>,
mut is_alive: IsAlive,
mut endpoint_in_use: EndpointInUse,
) -> RestartRelease
where
IsAlive: FnMut(u32) -> bool,
EndpointInUse: FnMut(&Path) -> bool,
{
let Some(target) = target else {
let ready =
wait_for_restart_condition(RESTART_GRACE_PERIOD, || !endpoint_in_use(socket)).await;
return RestartRelease { ready };
};
let pid = target.pid;
let process_exited = wait_for_restart_condition(RESTART_GRACE_PERIOD, || !is_alive(pid)).await;
if process_exited {
let ready =
wait_for_restart_condition(RESTART_CLEANUP_PERIOD, || !endpoint_in_use(socket)).await;
return RestartRelease { ready };
}
RestartRelease { ready: false }
}
async fn restart_control_phase<SendCommand, SendCommandFuture, IsAlive, EndpointInUse>(
socket: &Path,
pidfile_pid: Option<u32>,
endpoint_was_in_use: bool,
mut send_command: SendCommand,
is_alive: IsAlive,
endpoint_in_use: EndpointInUse,
) -> Result<RestartRelease>
where
SendCommand: FnMut(DaemonCommand) -> SendCommandFuture,
SendCommandFuture: std::future::Future<Output = Result<DaemonResponse>>,
IsAlive: FnMut(u32) -> bool,
EndpointInUse: FnMut(&Path) -> bool,
{
let target = if endpoint_was_in_use {
let status = send_command(DaemonCommand::Status).await.map_err(|_| {
anyhow::anyhow!("running daemon process identity could not be verified for restart")
})?;
let pid = match status {
DaemonResponse::Status { pid, .. } if pid != 0 => pid,
_ => anyhow::bail!("running daemon process identity could not be verified for restart"),
};
match send_command(DaemonCommand::Stop).await {
Ok(DaemonResponse::Ok) => {}
Ok(DaemonResponse::Error { message }) => return Err(anyhow::anyhow!(message)),
Ok(other) => return Err(anyhow::anyhow!("unexpected response: {other:?}")),
Err(e) => tracing::warn!(error = %e, "stop failed — proceeding to start"),
}
Some(RestartTarget { pid })
} else {
pidfile_pid.map(|pid| RestartTarget { pid })
};
let release = await_restart_release(socket, target, is_alive, endpoint_in_use).await;
Ok(release)
}
async fn restart(
source: &bitrouter::paths::ConfigSource,
socket: &Path,
log_path: &Path,
) -> Result<DaemonActionReport> {
let pid_path = pid_path_for(socket);
let pidfile_pid = daemon::read_pid_file(&pid_path)
.await
.filter(|pid| process_is_alive(*pid));
let endpoint_was_in_use = daemon::endpoint_in_use(socket);
let release = restart_control_phase(
socket,
pidfile_pid,
endpoint_was_in_use,
|command| async move { daemon::send_command(socket, &command).await },
process_is_alive,
daemon::endpoint_in_use,
)
.await?;
if !release.ready {
anyhow::bail!("old daemon did not release its process and control endpoint in time");
}
start(source, log_path, "restart").await
}
async fn reload(socket: &Path) -> Result<DaemonActionReport> {
let env: Vec<(String, String)> = bitrouter_providers::zero_config_env_var_providers()
.into_iter()
.filter_map(|(_, var)| {
std::env::var(&var)
.ok()
.filter(|v| !v.is_empty())
.map(|v| (var, v))
})
.collect();
match daemon::send_command(socket, &DaemonCommand::Reload { env }).await? {
DaemonResponse::Ok => Ok(DaemonActionReport::simple("reload", "reloaded")),
DaemonResponse::Error { message } => Err(anyhow::anyhow!(message)),
other => Err(anyhow::anyhow!("unexpected response: {other:?}")),
}
}
fn read_api_key_from_stdin() -> Result<String> {
use std::io::Read;
let mut buf = String::new();
std::io::stdin()
.read_to_string(&mut buf)
.context("reading API key from stdin")?;
let key = buf.trim().to_string();
if key.is_empty() {
anyhow::bail!("no API key on stdin (--key-stdin)");
}
Ok(key)
}
async fn key(action: KeyAction) -> Result<KeySignReport> {
match action {
KeyAction::Sign { user, db, policy } => {
let key = commands::key_sign(&db, &user, policy.as_deref()).await?;
Ok(KeySignReport {
id: key.id,
user,
secret: key.secret,
policy,
hash_stored: true,
})
}
}
}
async fn inspection_target(
remote_context: Option<&bitrouter::contexts::RemoteContext>,
context_name: Option<&str>,
config: Option<&Path>,
socket: Option<&Path>,
) -> Result<InspectionTarget> {
let remote = match remote_context {
Some(context) => {
let name = context_name
.ok_or_else(|| anyhow::anyhow!("remote context name is unavailable"))?;
Some((name.to_string(), context.clone()))
}
None => None,
};
InspectionTarget::resolve(remote, config, socket).await
}
fn selected_policy_view(view: Option<PolicyViewArg>, remote: bool) -> PolicyView {
match view {
Some(view) => view.into(),
None if remote => PolicyView::Active,
None => PolicyView::Disk,
}
}
async fn administration_policy(
action: PolicyAction,
output: &Output,
remote_context: Option<&bitrouter::contexts::RemoteContext>,
context_name: Option<&str>,
) -> Result<()> {
if remote_context.is_none() && is_legacy_local_policy_read(&action) {
return policy(action, output).await;
}
match action {
PolicyAction::Status {
view,
config,
socket,
} => {
let target = inspection_target(
remote_context,
context_name,
config.as_deref(),
socket.as_deref(),
)
.await?;
output.emit(
&target
.policy(PolicyInput {
view: selected_policy_view(view, remote_context.is_some()),
name: None,
})
.await?,
)?;
Ok(())
}
PolicyAction::Show {
name,
view,
config,
socket,
} => {
let target = inspection_target(
remote_context,
context_name,
config.as_deref(),
socket.as_deref(),
)
.await?;
output.emit(
&target
.policy(PolicyInput {
view: selected_policy_view(view, remote_context.is_some()),
name: Some(name),
})
.await?,
)?;
Ok(())
}
action => {
if remote_context.is_some() {
return Err(bitrouter_sdk::BitrouterError::bad_request(
"this policy action is local-only for a remote context",
)
.into());
}
policy(action, output).await
}
}
}
fn is_legacy_local_policy_read(action: &PolicyAction) -> bool {
matches!(
action,
PolicyAction::Status { view: None, .. } | PolicyAction::Show { view: None, .. }
)
}
async fn administration_providers(
action: ProviderAction,
output: &Output,
remote_context: Option<&bitrouter::contexts::RemoteContext>,
context_name: Option<&str>,
) -> Result<()> {
match action {
ProviderAction::List { config, socket } => {
let target = inspection_target(
remote_context,
context_name,
config.as_deref(),
socket.as_deref(),
)
.await?;
if let Some(source) = target.local_source() {
output.emit(&legacy_providers_report(source).await?)?;
} else {
output.emit(&target.providers().await?)?;
}
Ok(())
}
action => {
if remote_context.is_some() {
return Err(bitrouter_sdk::BitrouterError::bad_request(
"provider credential administration is local-only for a remote context",
)
.into());
}
providers(action, output).await
}
}
}
async fn administration_observe(
action: ObserveAction,
output: &Output,
remote_context: Option<&bitrouter::contexts::RemoteContext>,
context_name: Option<&str>,
) -> Result<()> {
match action {
ObserveAction::Status { config, socket } => {
let target = inspection_target(
remote_context,
context_name,
config.as_deref(),
socket.as_deref(),
)
.await?;
if let Some(socket) = target.local_socket() {
output.emit(&observe_status(socket).await?)?;
} else {
output.emit(&target.observe().await?)?;
}
Ok(())
}
}
}
async fn observe_status(socket: &Path) -> Result<ObserveStatusReport> {
use bitrouter_telemetry::OTEL_ENABLED;
let (snapshot, daemon_reachable) =
match daemon::send_command(socket, &DaemonCommand::ObserveStatus).await {
Ok(DaemonResponse::ObserveStatus { payload }) => (payload, true),
Ok(DaemonResponse::Error { message }) => return Err(anyhow::anyhow!(message)),
Ok(other) => return Err(anyhow::anyhow!("unexpected response: {other:?}")),
Err(error) if daemon::is_not_reachable(&error) => {
(daemon::ObserveStatusPayload::unwired(OTEL_ENABLED), false)
}
Err(error) => return Err(error),
};
Ok(ObserveStatusReport {
daemon_reachable,
snapshot,
socket: socket.display().to_string(),
})
}
async fn administration_agents(
action: Option<AgentsAction>,
output: &Output,
remote_context: Option<&bitrouter::contexts::RemoteContext>,
context_name: Option<&str>,
) -> Result<()> {
match action {
None => {
if remote_context.is_some() {
return Err(bitrouter_sdk::BitrouterError::bad_request(
"the standalone agent manager is local-only for a remote context",
)
.into());
}
bitrouter::agent_sessions::run_standalone().await
}
Some(AgentsAction::List {
remote: false,
config,
socket,
}) => {
let target = inspection_target(
remote_context,
context_name,
config.as_deref(),
socket.as_deref(),
)
.await?;
output.emit(&target.agents().await?)?;
Ok(())
}
Some(action) => {
if remote_context.is_some() {
return Err(bitrouter_sdk::BitrouterError::bad_request(
"agent launch and external-registry actions are local-only for a remote context",
)
.into());
}
agents_cmd(action, output).await
}
}
}
async fn policy(action: PolicyAction, output: &Output) -> Result<()> {
match action {
PolicyAction::Create { id, dir } => {
let path = commands::create_policy(&dir, &id).await?;
output.emit(&PolicyCreateReport {
id,
path: path.display().to_string(),
created: true,
})?;
}
PolicyAction::Init {
name,
router,
preset,
strong,
strong_effort,
economy,
economy_effort,
config,
} => {
let source = bitrouter::paths::resolve_config(config.as_deref())?;
let config_path = require_policy_config_path(&source)?;
let source_raw = tokio::fs::read_to_string(config_path).await?;
let source_config = bitrouter_sdk::config::parse(&source_raw)?;
if let Some(strong_model) = strong.as_deref() {
bitrouter::policy_lock::validate_routable_model(&source_config, strong_model)
.await?;
bitrouter::policy_lock::validate_routable_effort(
&source_config,
strong_model,
strong_effort,
)
.await?;
}
bitrouter::policy_lock::validate_routable_model(&source_config, &economy).await?;
bitrouter::policy_lock::validate_routable_effort(
&source_config,
&economy,
economy_effort,
)
.await?;
let update = match (router.as_deref(), preset.as_deref()) {
(Some(router), None) => {
bitrouter::policy_lock::initialize_router_files_with_efforts(
config_path,
&name,
router,
strong.as_deref(),
strong_effort,
&economy,
economy_effort,
)
.await?
}
(None, Some(preset)) => {
bitrouter::policy_lock::initialize_files_with_efforts(
config_path,
&name,
preset,
strong.as_deref(),
strong_effort,
&economy,
economy_effort,
)
.await?
}
(None, None) => {
bitrouter::policy_lock::initialize_router_files_with_efforts(
config_path,
&name,
"coding",
strong.as_deref(),
strong_effort,
&economy,
economy_effort,
)
.await?
}
(Some(_), Some(_)) => {
anyhow::bail!("--router and --preset are mutually exclusive")
}
};
let mut report =
routing_policy_report(config_path, "init", true, update.changes, None).await?;
report.config_activation =
Some(bitrouter::output::reports::config::ConfigActivation::SavedOnly);
output.emit(&report)?;
}
PolicyAction::Check { config } => {
let source = bitrouter::paths::resolve_config(config.as_deref())?;
let config_path = require_policy_config_path(&source)?;
output.emit(
&routing_policy_report(config_path, "check", false, Vec::new(), None).await?,
)?;
}
PolicyAction::Verify { evidence, config } => {
if !evidence {
anyhow::bail!("policy verify currently requires `--evidence`");
}
let source = bitrouter::paths::resolve_config(config.as_deref())?;
let config_path = require_policy_config_path(&source)?;
let verification = bitrouter::policy_lock::verify_evidence_files(config_path).await?;
output.emit(&PolicyReport {
action: "verify-evidence".into(),
path: Some(config_path.display().to_string()),
candidate_path: None,
digest: Some(verification.policy_digest.clone()),
source: None,
mode: "n/a".into(),
policies: Vec::new(),
bindings: Default::default(),
changes: vec![format!(
"verified {} eval results under {}",
verification.eval_results, verification.evidence_root
)],
policy: Some(serde_json::to_value(verification)?),
applied: false,
config_activation: None,
})?;
}
PolicyAction::Status { config, .. } => {
let source = bitrouter::paths::resolve_config(config.as_deref())?;
let config_path = require_policy_config_path(&source)?;
output.emit(
&routing_policy_report(config_path, "status", false, Vec::new(), None).await?,
)?;
}
PolicyAction::Show { name, config, .. } => {
let source = bitrouter::paths::resolve_config(config.as_deref())?;
let config_path = require_policy_config_path(&source)?;
output.emit(
&routing_policy_report(config_path, "show", false, Vec::new(), Some(&name)).await?,
)?;
}
PolicyAction::Reload { config, socket } => {
let source = bitrouter::paths::resolve_config(config.as_deref())?;
let socket = resolve_client_socket_from(&source, socket.as_deref()).await?;
output.emit(&reload(&socket).await?)?;
}
PolicyAction::Compile {
output: candidate_path,
snapshot_time,
eval_snapshot,
config,
} => {
let source = bitrouter::paths::resolve_config(config.as_deref())?;
let config_path = require_policy_config_path(&source)?;
let snapshot_time =
snapshot_time.unwrap_or_else(|| chrono::Utc::now().timestamp_millis());
let update = bitrouter::policy_lock::compile_files_with_eval(
config_path,
snapshot_time,
eval_snapshot.as_deref(),
)
.await?;
let digest = bitrouter::policy_lock::export_candidate_file(
&update.path,
&candidate_path,
&update.document,
)?;
let mut changes = update.changes;
changes.extend(
update
.conflicts
.into_iter()
.map(|conflict| format!("conflict: {conflict}")),
);
let mut report =
routing_policy_report(config_path, "compile", false, changes, None).await?;
report.candidate_path = Some(candidate_path.display().to_string());
report.digest = Some(digest);
output.emit(&report)?;
}
PolicyAction::Diff { active, candidate } => {
let active_lock = bitrouter::policy_lock::load(&active).await?;
let candidate_lock = bitrouter::policy_lock::load(&candidate).await?;
let changes = bitrouter::policy_lock::diff_explanations(
&active_lock.document,
&candidate_lock.document,
);
output.emit(&PolicyReport {
action: "diff".into(),
path: Some(active.display().to_string()),
candidate_path: Some(candidate.display().to_string()),
digest: Some(candidate_lock.digest),
source: None,
mode: "n/a".into(),
policies: candidate_lock.document.policies.keys().cloned().collect(),
bindings: Default::default(),
changes,
policy: None,
applied: false,
config_activation: None,
})?;
}
PolicyAction::Publish {
candidate,
config,
socket,
} => {
let source = bitrouter::paths::resolve_config(config.as_deref())?;
let config_path = require_policy_config_path(&source)?;
let update =
bitrouter::policy_lock::publish_candidate_file(config_path, &candidate).await?;
reload_published_policy_or_restore(&source, &update, socket.as_deref()).await?;
let changes = update.changes.clone();
let mut report =
routing_policy_report(config_path, "publish", true, changes, None).await?;
report.candidate_path = Some(candidate.display().to_string());
report.digest = Some(update.digest);
output.emit(&report)?;
}
PolicyAction::Evolve {
apply,
output: candidate_output,
config,
} => {
let source = bitrouter::paths::resolve_config(config.as_deref())?;
let config_path = require_policy_config_path(&source)?;
let update = bitrouter::policy_lock::evolve_files(config_path, apply).await?;
let action = if candidate_output.is_some() {
"evolve-export"
} else if apply {
"evolve"
} else {
"evolve-dry-run"
};
let published = apply;
if published {
reload_published_policy_or_restore(&source, &update, None).await?;
}
let mut changes = update.changes.clone();
changes.extend(
update
.conflicts
.iter()
.map(|conflict| format!("conflict: {conflict}")),
);
let mut report =
routing_policy_report(config_path, action, published, changes, None).await?;
if let Some(candidate_path) = candidate_output {
report.digest = Some(bitrouter::policy_lock::export_candidate_file(
&update.path,
&candidate_path,
&update.document,
)?);
report.candidate_path = Some(candidate_path.display().to_string());
} else {
report.digest = Some(update.digest);
}
output.emit(&report)?;
}
PolicyAction::Rollback {
digest,
config,
socket,
} => {
let source = bitrouter::paths::resolve_config(config.as_deref())?;
let config_path = require_policy_config_path(&source)?;
let cfg = config::load(config_path).await?;
if cfg.policy.mode == config::PolicyRuntimeMode::Frozen {
anyhow::bail!("policy runtime mode is frozen; rollback requires adaptive mode");
}
if cfg.policy.mode == config::PolicyRuntimeMode::Frozen {
anyhow::bail!(
"policy runtime mode is frozen; use adaptive mode for policy publication"
);
}
let loaded = bitrouter::policy_lock::load_for_config(&cfg, Some(config_path))
.await?
.ok_or_else(|| anyhow::anyhow!("no policy lock is configured"))?;
let history_dir = bitrouter::policy_lock::default_history_dir(&loaded.path);
let record = bitrouter::policy_lock::rollback_to_digest(
&loaded.path,
&loaded.digest,
&digest,
&history_dir,
)?;
if let Err(error) = reload_policy_if_reachable(&source, socket.as_deref()).await {
bitrouter::policy_lock::rollback_to_digest(
&loaded.path,
&record.child_digest,
&record.parent_digest,
&history_dir,
)?;
let _ = reload_policy_if_reachable(&source, socket.as_deref()).await;
return Err(error.context("daemon rejected rollback; restored previous lock"));
}
output.emit(
&routing_policy_report(
config_path,
"rollback",
true,
vec![format!("restored {}", record.child_digest)],
None,
)
.await?,
)?;
}
}
Ok(())
}
async fn optimize(action: OptimizeAction, output: &Output) -> Result<()> {
match action {
OptimizeAction::Run {
policy,
candidate_tier,
exploration_ppm,
minimum_tasks,
maximum_tasks,
minimum_pass_rate_ppm,
evaluator_config_digest,
config,
socket,
} => {
let source = bitrouter::paths::resolve_config(Some(&config))?;
let config_path = require_policy_config_path(&source)?.to_path_buf();
let options = bitrouter::optimization::controller::OptimizationOptions {
policy: policy.clone(),
candidate_tier,
challenger_exposure_ppm: exploration_ppm,
minimum_tasks_per_arm: minimum_tasks,
maximum_challenger_tasks: maximum_tasks,
minimum_pass_rate_ppm,
evaluator_config_digest,
};
let prepared =
bitrouter::optimization::controller::prepare_files(&config_path, options).await?;
let mut report = optimization_run_report(&policy, &prepared);
let publication =
bitrouter::optimization::controller::publish_prepared(prepared).await?;
report.active_policy_digest = publication.active_policy_digest.clone();
report.published = publication.published;
if publication.update.is_some() {
let endpoint = resolve_client_socket_from(&source, socket.as_deref()).await?;
report.reload_attempted = daemon::endpoint_in_use(&endpoint);
reload_optimization_publication_or_restore(
&source,
&publication,
socket.as_deref(),
)
.await?;
}
output.emit(&report)?;
Ok(())
}
OptimizeAction::Status { policy, config } => {
let source = bitrouter::paths::resolve_config(Some(&config))?;
let config_path = require_policy_config_path(&source)?;
let status =
bitrouter::optimization::controller::read_status(config_path, &policy).await?;
output.emit(&optimization_status_report(status))?;
Ok(())
}
}
}
fn optimization_run_report(
policy: &str,
prepared: &bitrouter::optimization::controller::PreparedOptimizationStep,
) -> OptimizationControllerReport {
let cohort = prepared.step.evidence.cohort.as_ref();
OptimizationControllerReport {
action: "optimize.run",
policy: policy.into(),
decision: controller_action_name(prepared.step.action),
parent_policy_digest: Some(prepared.parent_policy_digest.clone()),
active_policy_digest: prepared.parent_policy_digest.clone(),
eval_snapshot_root: Some(prepared.step.evidence.eval_snapshot_root.clone()),
observed_subject_digest: Some(prepared.step.evidence.observed_subject_digest.clone()),
treatment: prepared.treatment.as_ref().map(treatment_report),
control: cohort.map(|assessment| arm_report(&assessment.control)),
challenger: cohort.map(|assessment| arm_report(&assessment.challenger)),
cost_delta_micro_usd: cohort.and_then(|assessment| assessment.cost_delta_micro_usd),
evaluator_config_digest: cohort
.and_then(|assessment| assessment.evaluator_config_digest.clone())
.or_else(|| {
prepared
.treatment
.as_ref()
.and_then(|treatment| treatment.gate.evaluator_config_digest.clone())
}),
published: false,
reload_attempted: false,
}
}
fn optimization_status_report(
status: bitrouter::optimization::controller::OptimizationStatus,
) -> OptimizationControllerReport {
let decision = if status.active_experiment.is_some() {
"exploring"
} else {
"idle"
};
OptimizationControllerReport {
action: "optimize.status",
policy: status.policy,
decision,
parent_policy_digest: status.parent_policy_digest,
active_policy_digest: status.active_policy_digest,
eval_snapshot_root: status.eval_snapshot_root,
observed_subject_digest: status.observed_subject_digest,
treatment: status.active_experiment.as_ref().map(treatment_report),
control: None,
challenger: None,
cost_delta_micro_usd: None,
evaluator_config_digest: status
.active_experiment
.and_then(|experiment| experiment.gate.evaluator_config_digest),
published: false,
reload_attempted: false,
}
}
fn treatment_report(
exploration: &bitrouter::optimization::exploration::RouteExploration,
) -> TreatmentReport {
TreatmentReport {
target_request_key: exploration.target_request_key.clone(),
champion_tier: exploration.champion_tier.clone(),
challenger_tier: exploration.challenger_tier.clone(),
challenger_exposure_ppm: exploration.challenger_exposure_ppm,
minimum_tasks_per_arm: exploration.gate.minimum_tasks_per_arm,
maximum_challenger_tasks: exploration.gate.maximum_challenger_tasks,
minimum_pass_rate_ppm: exploration.gate.minimum_pass_rate_ppm,
}
}
fn arm_report(assessment: &bitrouter::optimization::cohort::ArmAssessment) -> ArmReport {
ArmReport {
observed: assessment.observed,
eligible: assessment.eligible,
excluded: assessment.excluded,
conclusive: assessment.conclusive,
pass: assessment.pass,
fail: assessment.fail,
hard_violations: assessment.hard_violations,
pass_rate_ppm: assessment.pass_rate_ppm,
mean_cost_micro_usd: assessment.mean_cost_micro_usd,
}
}
fn controller_action_name(
action: bitrouter::optimization::controller::ControllerAction,
) -> &'static str {
match action {
bitrouter::optimization::controller::ControllerAction::Explore => "explore",
bitrouter::optimization::controller::ControllerAction::Promote => "promote",
bitrouter::optimization::controller::ControllerAction::Retreat => "retreat",
bitrouter::optimization::controller::ControllerAction::Hold => "hold",
bitrouter::optimization::controller::ControllerAction::Converged => "converged",
}
}
async fn eval(action: EvalAction, output: &Output) -> Result<()> {
use bitrouter::eval::admission::SubmissionPrincipal;
use bitrouter::eval::types::{EvalSubject, EvaluationResult};
let (action_name, data) = match action {
EvalAction::Subject { action } => match action {
EvalSubjectAction::Seal {
draft,
output: path,
} => {
let mut subject: EvalSubject = read_eval_file(&draft)?;
subject
.evidence
.sort_by(|left, right| left.evidence_id.cmp(&right.evidence_id));
subject.evidence_digest =
bitrouter::eval::types::evidence_digest(&subject.evidence)?;
bitrouter::eval::types::validate_subject(&subject)?;
let sealed = serde_json::to_string_pretty(&subject)? + "\n";
std::fs::write(&path, sealed)
.with_context(|| format!("writing sealed eval subject {}", path.display()))?;
(
"subject-seal",
serde_json::json!({
"eval_id": subject.eval_id,
"evidence_digest": subject.evidence_digest,
"output": path,
}),
)
}
EvalSubjectAction::Put { file, config } => {
let subject: EvalSubject = read_eval_file(&file)?;
let service = local_eval_service(config.as_deref()).await?;
let outcome = service.store().insert_subject(&subject).await?;
(
"subject-put",
serde_json::json!({
"eval_id": subject.eval_id,
"outcome": format!("{outcome:?}").to_ascii_lowercase(),
}),
)
}
EvalSubjectAction::Get { eval_id, config } => {
let service = local_eval_service(config.as_deref()).await?;
let subject = service
.store()
.subject_for_owner(&eval_id, "local")
.await?
.ok_or_else(|| anyhow::anyhow!("eval subject '{eval_id}' not found"))?;
("subject-get", serde_json::to_value(subject)?)
}
EvalSubjectAction::List { config } => {
let service = local_eval_service(config.as_deref()).await?;
(
"subject-list",
serde_json::to_value(service.store().list_subjects_for_owner("local").await?)?,
)
}
},
EvalAction::Result { action } => match action {
EvalResultAction::Submit { file, config } => {
let result: EvaluationResult = read_eval_file(&file)?;
let service = local_eval_service(config.as_deref()).await?;
let outcome = service
.submit(result, SubmissionPrincipal::LocalOperator)
.await?;
("result-submit", serde_json::to_value(outcome)?)
}
},
EvalAction::Snapshot { action } => match action {
EvalSnapshotAction::Freeze { at, config } => {
let service = local_eval_service(config.as_deref()).await?;
let frozen_at = at.unwrap_or_else(|| chrono::Utc::now().to_rfc3339());
let snapshot = service
.store()
.freeze_snapshot_for_owner(&frozen_at, "local")
.await?;
("snapshot-freeze", serde_json::to_value(snapshot)?)
}
EvalSnapshotAction::Get {
evidence_root,
config,
} => {
let service = local_eval_service(config.as_deref()).await?;
let snapshot = service
.store()
.snapshot_by_root_for_owner(&evidence_root, "local")
.await?
.ok_or_else(|| anyhow::anyhow!("eval snapshot '{evidence_root}' not found"))?;
("snapshot-get", serde_json::to_value(snapshot)?)
}
},
EvalAction::Status { config } => {
let service = local_eval_service(config.as_deref()).await?;
let subjects = service.store().list_subjects_for_owner("local").await?;
let admissions = service.store().latest_admissions_for_owner("local").await?;
let mut counts = std::collections::BTreeMap::<String, usize>::new();
for event in admissions.values() {
*counts
.entry(format!("{:?}", event.status).to_ascii_lowercase())
.or_default() += 1;
}
(
"status",
serde_json::json!({
"subjects": subjects.len(),
"results": admissions.len(),
"admission": counts,
}),
)
}
};
output.emit(&EvalReport {
action: action_name.into(),
data,
})?;
Ok(())
}
async fn trajectory(
config_path: Option<&Path>,
action: TrajectoryAction,
output: &Output,
) -> Result<()> {
let source = bitrouter::paths::resolve_config(config_path)?;
let (config, store) = local_trajectory_store(&source).await?;
match action {
TrajectoryAction::Inspect { episode_id } => {
output.emit(&trajectory_inspect_report(&store, &episode_id).await?)?;
}
TrajectoryAction::Replay { episode_id } => {
output.emit(&trajectory_replay_report(&store, &episode_id).await?)?;
}
TrajectoryAction::Prune { before, dry_run } => {
let summary = store
.prune_before(&before, dry_run, config.trajectory.outbox_batch_size)
.await?;
output.emit(&TrajectoryPruneReport::new(before, dry_run, summary))?;
}
}
Ok(())
}
async fn local_trajectory_store(
source: &bitrouter::paths::ConfigSource,
) -> Result<(
bitrouter_sdk::config::Config,
bitrouter::trajectory::store::TrajectoryStore,
)> {
let config = bitrouter::paths::load_config(source).await?;
let database_url = bitrouter::db::anchor_url(&config.database.url, source.home());
let db = bitrouter::db::connect(&database_url).await?;
bitrouter::db::run_migrations(&db).await?;
Ok((
config,
bitrouter::trajectory::store::TrajectoryStore::new(db),
))
}
async fn local_eval_service(config_path: Option<&Path>) -> Result<bitrouter::eval::EvalService> {
let source = bitrouter::paths::resolve_config(config_path)?;
let config = bitrouter::paths::load_config(&source).await?;
let db = bitrouter::db::connect(&config.database.url).await?;
bitrouter::db::run_migrations(&db).await?;
Ok(bitrouter::eval::EvalService::new(
bitrouter::eval::store::EvalStore::new(db),
config.eval,
))
}
fn read_eval_file<T>(path: &Path) -> Result<T>
where
T: serde::de::DeserializeOwned,
{
let content = std::fs::read_to_string(path)
.with_context(|| format!("reading eval exchange file {}", path.display()))?;
match serde_json::from_str(&content) {
Ok(value) => Ok(value),
Err(json_error) => serde_saphyr::from_str(&content)
.with_context(|| format!("parsing {} as JSON ({json_error}) or YAML", path.display())),
}
}
async fn reload_policy_if_reachable(
source: &bitrouter::paths::ConfigSource,
socket_override: Option<&Path>,
) -> Result<()> {
let socket = resolve_client_socket_from(source, socket_override).await?;
if !daemon::endpoint_in_use(&socket) {
return Ok(());
}
reload(&socket).await.map(|_| ())
}
#[derive(Debug, Eq, PartialEq)]
enum OptimizationRecoveryRevision {
Parent,
Child,
Both,
Other(String),
}
#[derive(Debug, Eq, PartialEq)]
struct OptimizationRecoveryPair {
config: OptimizationRecoveryRevision,
policy: OptimizationRecoveryRevision,
}
impl OptimizationRecoveryPair {
fn description(&self) -> String {
format!(
"config={}, policy={}",
self.config.description(),
self.policy.description()
)
}
}
impl OptimizationRecoveryRevision {
fn description(&self) -> &str {
match self {
Self::Parent => "parent",
Self::Child => "child",
Self::Both => "parent and child (identical bytes)",
Self::Other(detail) => detail,
}
}
fn is_child(&self) -> bool {
matches!(self, Self::Child | Self::Both)
}
}
async fn observe_optimization_recovery_pair(
publication: &bitrouter::optimization::controller::OptimizationPublication,
update: &bitrouter::policy_lock::PolicyFileUpdate,
) -> Result<OptimizationRecoveryPair> {
let config = std::fs::read_to_string(&publication.config_path)
.with_context(|| format!("reading {}", publication.config_path.display()))?;
let config = if publication.config_before == publication.config_after
&& config == publication.config_before
{
OptimizationRecoveryRevision::Both
} else if config == publication.config_before {
OptimizationRecoveryRevision::Parent
} else if config == publication.config_after {
OptimizationRecoveryRevision::Child
} else {
OptimizationRecoveryRevision::Other("different bytes".into())
};
let active = bitrouter::policy_lock::load(&update.path).await?;
let policy = if active.digest == publication.parent_policy_digest {
OptimizationRecoveryRevision::Parent
} else if active.digest == publication.active_policy_digest {
OptimizationRecoveryRevision::Child
} else {
OptimizationRecoveryRevision::Other(active.digest)
};
Ok(OptimizationRecoveryPair { config, policy })
}
fn optimizer_recovery_failure(
initial_error: &anyhow::Error,
message: impl std::fmt::Display,
recovery_errors: &[String],
) -> anyhow::Error {
let recovery_detail = if recovery_errors.is_empty() {
String::new()
} else {
format!(
"; recovery operation errors: {}",
recovery_errors.join("; ")
)
};
anyhow::anyhow!("{initial_error:#}").context(format!("{message}{recovery_detail}"))
}
async fn reload_reconciled_optimization_pair<Reload, ReloadFuture>(
reload: &Reload,
initial_error: &anyhow::Error,
pair: &str,
recovery_errors: &[String],
) -> Result<()>
where
Reload: Fn() -> ReloadFuture,
ReloadFuture: std::future::Future<Output = Result<()>>,
{
match reload().await {
Ok(()) => Err(optimizer_recovery_failure(
initial_error,
format!(
"daemon rejected optimization publication; reconciled and reloaded the {pair} config and policy"
),
recovery_errors,
)),
Err(reload_error) => {
let mut errors = recovery_errors.to_vec();
errors.push(format!(
"reloading reconciled {pair} pair: {reload_error:#}"
));
Err(optimizer_recovery_failure(
initial_error,
format!(
"daemon rejected optimization publication; reconciled the {pair} config and policy but recovery reload failed"
),
&errors,
))
}
}
}
async fn reload_published_policy_or_restore(
source: &bitrouter::paths::ConfigSource,
update: &bitrouter::policy_lock::PolicyFileUpdate,
socket_override: Option<&Path>,
) -> Result<()> {
if let Err(error) = reload_policy_if_reachable(source, socket_override).await {
let parent_digest = update
.document
.artifact
.as_ref()
.and_then(|artifact| artifact.parent_digest.as_deref())
.ok_or_else(|| anyhow::anyhow!("published policy has no parent digest"))?;
let history_dir = bitrouter::policy_lock::default_history_dir(&update.path);
let restore = bitrouter::policy_lock::rollback_to_digest(
&update.path,
&update.digest,
parent_digest,
&history_dir,
);
let restore_reload = if restore.is_ok() {
reload_policy_if_reachable(source, socket_override).await
} else {
Ok(())
};
return match (restore, restore_reload) {
(Ok(_), Ok(())) => {
Err(error.context("daemon rejected candidate; restored previous lock"))
}
(restore, restore_reload) => Err(error.context(format!(
"daemon rejected candidate and recovery failed (policy: {}; reload: {})",
restore
.err()
.map(|value| format!("{value:#}"))
.unwrap_or_else(|| "ok".into()),
restore_reload
.err()
.map(|value| format!("{value:#}"))
.unwrap_or_else(|| "ok".into()),
))),
};
}
Ok(())
}
async fn reload_optimization_publication_or_restore(
source: &bitrouter::paths::ConfigSource,
publication: &bitrouter::optimization::controller::OptimizationPublication,
socket_override: Option<&Path>,
) -> Result<()> {
reload_optimization_publication_or_restore_with(
publication,
move || reload_policy_if_reachable(source, socket_override),
|| Ok(()),
)
.await
}
async fn reload_optimization_publication_or_restore_with<Reload, ReloadFuture, BeforeRecovery>(
publication: &bitrouter::optimization::controller::OptimizationPublication,
reload: Reload,
before_recovery: BeforeRecovery,
) -> Result<()>
where
Reload: Fn() -> ReloadFuture,
ReloadFuture: std::future::Future<Output = Result<()>>,
BeforeRecovery: FnOnce() -> Result<()>,
{
reload_optimization_publication_or_restore_with_mutations(
publication,
reload,
before_recovery,
|path, expected_digest, target_digest, history_dir| {
bitrouter::policy_lock::rollback_to_digest_unlocked(
path,
expected_digest,
target_digest,
history_dir,
)
.map(|_| ())
},
bitrouter::policy_lock::write_text_atomic_unlocked,
)
.await
}
async fn reload_optimization_publication_or_restore_with_mutations<
Reload,
ReloadFuture,
BeforeRecovery,
MutatePolicy,
MutateConfig,
>(
publication: &bitrouter::optimization::controller::OptimizationPublication,
reload: Reload,
before_recovery: BeforeRecovery,
mutate_policy: MutatePolicy,
mutate_config: MutateConfig,
) -> Result<()>
where
Reload: Fn() -> ReloadFuture,
ReloadFuture: std::future::Future<Output = Result<()>>,
BeforeRecovery: FnOnce() -> Result<()>,
MutatePolicy: Fn(&Path, &str, &str, &Path) -> Result<()>,
MutateConfig: Fn(&Path, &str, &str) -> Result<()>,
{
let initial_error = match reload().await {
Ok(()) => return Ok(()),
Err(error) => error,
};
let initial_detail = format!("{initial_error:#}");
if let Err(seam_error) = before_recovery() {
return Err(initial_error.context(format!(
"daemon rejected optimization publication and recovery setup failed: {seam_error:#}"
)));
}
let Some(update) = publication.update.as_ref() else {
return Err(anyhow::anyhow!(
"daemon rejected optimization publication ({initial_detail}); recovery has no policy update"
));
};
let _config_guard =
bitrouter::policy_lock::acquire_publication_lock(&publication.config_path).map_err(
|recovery| {
anyhow::anyhow!(
"daemon rejected optimization publication ({initial_detail}); acquiring config recovery lock failed: {recovery:#}"
)
},
)?;
let _policy_guard = bitrouter::policy_lock::acquire_publication_lock(&update.path).map_err(
|recovery| {
anyhow::anyhow!(
"daemon rejected optimization publication ({initial_detail}); acquiring policy recovery lock failed: {recovery:#}"
)
},
)?;
if update.digest != publication.active_policy_digest {
return Err(initial_error.context(format!(
"daemon rejected optimization publication; recovery is stale because update {} does not match published policy {}",
update.digest, publication.active_policy_digest
)));
}
let initial_pair = observe_optimization_recovery_pair(publication, update)
.await
.map_err(|recovery| {
anyhow::anyhow!(
"daemon rejected optimization publication ({initial_detail}); reading recovery state failed: {recovery:#}"
)
})?;
if !initial_pair.config.is_child() || initial_pair.policy != OptimizationRecoveryRevision::Child
{
return Err(initial_error.context(format!(
"daemon rejected optimization publication; recovery is stale because the active pair is {}",
initial_pair.description()
)));
}
let history_dir = bitrouter::policy_lock::default_history_dir(&update.path);
let mut recovery_errors = Vec::new();
if let Err(policy_error) = mutate_policy(
&update.path,
&publication.active_policy_digest,
&publication.parent_policy_digest,
&history_dir,
) {
recovery_errors.push(format!("restoring parent policy: {policy_error:#}"));
}
let after_policy = observe_optimization_recovery_pair(publication, update)
.await
.map_err(|recovery| {
optimizer_recovery_failure(
&initial_error,
format!(
"daemon rejected optimization publication; reconciling disk state after policy recovery failed: {recovery:#}"
),
&recovery_errors,
)
})?;
match (&after_policy.config, &after_policy.policy) {
(
OptimizationRecoveryRevision::Child | OptimizationRecoveryRevision::Both,
OptimizationRecoveryRevision::Child,
) => {
return reload_reconciled_optimization_pair(
&reload,
&initial_error,
"child",
&recovery_errors,
)
.await;
}
(OptimizationRecoveryRevision::Both, OptimizationRecoveryRevision::Parent) => {
return reload_reconciled_optimization_pair(
&reload,
&initial_error,
"parent",
&recovery_errors,
)
.await;
}
(OptimizationRecoveryRevision::Child, OptimizationRecoveryRevision::Parent) => {}
_ => {
return Err(optimizer_recovery_failure(
&initial_error,
format!(
"daemon rejected optimization publication; recovery is stale after policy recovery because the pair is {}",
after_policy.description()
),
&recovery_errors,
));
}
}
if let Err(config_error) = mutate_config(
&publication.config_path,
&publication.config_after,
&publication.config_before,
) {
recovery_errors.push(format!("restoring parent config: {config_error:#}"));
}
let after_config = observe_optimization_recovery_pair(publication, update)
.await
.map_err(|recovery| {
optimizer_recovery_failure(
&initial_error,
format!(
"daemon rejected optimization publication; reconciling disk state after config recovery failed: {recovery:#}"
),
&recovery_errors,
)
})?;
match (&after_config.config, &after_config.policy) {
(OptimizationRecoveryRevision::Parent, OptimizationRecoveryRevision::Parent) => {
return reload_reconciled_optimization_pair(
&reload,
&initial_error,
"parent",
&recovery_errors,
)
.await;
}
(OptimizationRecoveryRevision::Child, OptimizationRecoveryRevision::Parent) => {}
_ => {
return Err(optimizer_recovery_failure(
&initial_error,
format!(
"daemon rejected optimization publication; recovery is stale after config recovery because the pair is {}",
after_config.description()
),
&recovery_errors,
));
}
}
if let Err(compensation_error) = mutate_policy(
&update.path,
&publication.parent_policy_digest,
&publication.active_policy_digest,
&history_dir,
) {
recovery_errors.push(format!(
"restoring child policy after config recovery failure: {compensation_error:#}"
));
}
let after_compensation = observe_optimization_recovery_pair(publication, update)
.await
.map_err(|recovery| {
optimizer_recovery_failure(
&initial_error,
format!(
"daemon rejected optimization publication; reconciling disk state after policy compensation failed: {recovery:#}"
),
&recovery_errors,
)
})?;
if after_compensation.config.is_child()
&& after_compensation.policy == OptimizationRecoveryRevision::Child
{
return reload_reconciled_optimization_pair(
&reload,
&initial_error,
"child",
&recovery_errors,
)
.await;
}
Err(optimizer_recovery_failure(
&initial_error,
format!(
"daemon rejected optimization publication; recovery is stale after policy compensation because the pair is {}",
after_compensation.description()
),
&recovery_errors,
))
}
fn require_policy_config_path(source: &bitrouter::paths::ConfigSource) -> Result<&Path> {
match source {
bitrouter::paths::ConfigSource::File(path) => Ok(path),
bitrouter::paths::ConfigSource::Default { .. } => anyhow::bail!(
"routing policies require a file-backed bitrouter.yaml; run `{} init` first",
bitrouter_sdk::invocation::name()
),
}
}
async fn routing_policy_report(
config_path: &Path,
action: &str,
applied: bool,
changes: Vec<String>,
show: Option<&str>,
) -> Result<PolicyReport> {
let raw = tokio::fs::read_to_string(config_path)
.await
.with_context(|| format!("reading {}", config_path.display()))?;
let cfg = config::parse(&raw).context("parsing bitrouter.yaml")?;
let loaded = bitrouter::policy_lock::load_for_config(&cfg, Some(config_path)).await?;
let policy = match show {
Some(name) => {
let lock = loaded
.as_ref()
.ok_or_else(|| anyhow::anyhow!("no policy lock is configured"))?;
let definition = lock
.document
.policies
.get(name)
.ok_or_else(|| anyhow::anyhow!("policy '{name}' does not exist"))?;
Some(serde_json::to_value(definition).context("serializing policy")?)
}
None => None,
};
let bindings = cfg
.router_policy_bindings()
.map(|(name, policy, _)| (name.to_owned(), policy.to_owned()))
.collect();
let path = loaded
.as_ref()
.map(|lock| lock.path.clone())
.or_else(|| bitrouter::policy_lock::resolve_path(&cfg, Some(config_path)));
let policies = loaded
.as_ref()
.map(|lock| lock.document.policies.keys().cloned().collect())
.unwrap_or_default();
Ok(PolicyReport {
action: action.to_string(),
path: path.map(|path| path.display().to_string()),
candidate_path: None,
digest: loaded.as_ref().map(|lock| lock.digest.clone()),
source: matches!(action, "status" | "show").then(|| "disk".to_string()),
mode: match cfg.policy.mode {
config::PolicyRuntimeMode::Frozen => "frozen",
config::PolicyRuntimeMode::Adaptive => "adaptive",
}
.to_string(),
policies,
bindings,
changes,
policy,
applied,
config_activation: None,
})
}
async fn providers(action: ProviderAction, output: &Output) -> Result<()> {
match action {
ProviderAction::List { config, .. } => {
let source = bitrouter::paths::resolve_config(config.as_deref())?;
output.emit(&legacy_providers_report(&source).await?)?;
Ok(())
}
ProviderAction::Login {
provider,
label,
import_existing,
no_browser,
api_key,
key_stdin,
} => {
let api_key = if key_stdin {
Some(read_api_key_from_stdin()?)
} else {
api_key
};
if provider == "bitrouter" {
if import_existing || no_browser {
anyhow::bail!(
"`{cli} providers login bitrouter` uses BitRouter Cloud OAuth; \
--import-existing/--no-browser apply to upstream provider logins",
cli = bitrouter_sdk::invocation::name()
);
}
bitrouter::cloud::cli::run(
bitrouter::cloud::cli::CloudAction::Login {
authorization_server: None,
client_id: None,
scope: None,
api_key,
},
output.format(),
)
.await
} else {
let outcome = bitrouter::commands::login_provider_with_options(
&provider,
&label,
bitrouter::commands::ProviderLoginOptions {
import_existing,
no_browser,
api_key,
},
)
.await?;
output.emit(&ProviderLoginReport {
provider: outcome.provider,
label: outcome.label,
method: outcome.method,
credential: "saved",
path: outcome.path,
})?;
Ok(())
}
}
ProviderAction::Logout { provider } => {
if provider == "bitrouter" {
bitrouter::cloud::cli::run(
bitrouter::cloud::cli::CloudAction::Logout {
authorization_server: None,
client_id: None,
},
output.format(),
)
.await
} else {
let removed = bitrouter::commands::logout_provider(&provider).await?;
output.emit(&ProviderLogoutReport { provider, removed })?;
Ok(())
}
}
}
}
async fn legacy_providers_report(
source: &bitrouter::paths::ConfigSource,
) -> Result<ProvidersReport> {
let cfg = bitrouter::paths::load_config(source).await?;
let providers = commands::list_providers(&cfg)
.into_iter()
.map(|provider| ProviderRow {
id: provider.id,
models: provider.model_count,
active: provider.active,
api_base: provider.api_base,
})
.collect();
Ok(ProvidersReport { providers })
}
async fn tools(action: ToolsAction, output: &Output) -> Result<()> {
use bitrouter::tools as tools_cmd;
match action {
ToolsAction::List { config } => {
eprintln!(
"note: `{cli} tools list` is a compatibility form; use `{cli} mcp check`.",
cli = bitrouter_sdk::invocation::name()
);
let source = bitrouter::paths::resolve_config(config.as_deref())?;
let cfg = bitrouter::paths::load_config(&source).await?;
let servers = tools_cmd::list(&cfg)
.await
.into_iter()
.map(|row| match row.outcome {
Ok(tools) => ServerToolsView {
server: row.server,
tools: Some(
tools
.into_iter()
.map(|t| ToolInfo {
name: t.name,
description: t.description,
})
.collect(),
),
error: None,
},
Err(e) => ServerToolsView {
server: row.server,
tools: None,
error: Some(e),
},
})
.collect();
output.emit(&ToolsListReport { servers })?;
Ok(())
}
ToolsAction::Status { config } => {
eprintln!(
"note: `{cli} tools status` is a compatibility form; use `{cli} mcp check`.",
cli = bitrouter_sdk::invocation::name()
);
let source = bitrouter::paths::resolve_config(config.as_deref())?;
let cfg = bitrouter::paths::load_config(&source).await?;
let servers = tools_cmd::status(&cfg)
.await
.into_iter()
.map(|row| {
let (ok, latency_ms, error) = match row.outcome {
Ok(d) => (true, Some(d.as_millis()), None),
Err(e) => (false, None, Some(e)),
};
ServerStatusView {
server: row.server,
ok,
latency_ms,
transport: row.transport,
error,
}
})
.collect();
output.emit(&ToolsStatusReport { servers })?;
Ok(())
}
ToolsAction::Discover { server, config } => {
eprintln!(
"note: `{cli} tools discover` is deprecated; use `{cli} mcp check <server>`.",
cli = bitrouter_sdk::invocation::name()
);
let source = bitrouter::paths::resolve_config(config.as_deref())?;
let cfg = bitrouter::paths::load_config(&source).await?;
match tools_cmd::discover(&cfg, &server).await {
Ok(yaml) => {
output.emit(&ToolsDiscoverReport { server, yaml })?;
Ok(())
}
Err(e) => anyhow::bail!("discover '{server}': {e}"),
}
}
}
}
async fn agents_cmd(action: AgentsAction, output: &Output) -> Result<()> {
use bitrouter::agents as agents_cmd;
match action {
AgentsAction::Sessions { config, socket } => {
let socket = resolve_client_socket(config.as_deref(), socket.as_deref()).await?;
let runs = bitrouter::agent_sessions::BackgroundClient::list_only(socket)
.summaries()
.await?;
output.emit(&AgentSessionsReport { runs })?;
Ok(())
}
AgentsAction::Attach {
agent_run_id,
config,
socket,
} => {
let socket = resolve_client_socket(config.as_deref(), socket.as_deref()).await?;
bitrouter::agent_sessions::run_attached(socket, &agent_run_id).await
}
AgentsAction::Stop {
agent_run_id,
config,
socket,
} => {
let socket = resolve_client_socket(config.as_deref(), socket.as_deref()).await?;
let acknowledgement = bitrouter::agent_sessions::BackgroundClient::stop_only(socket)
.stop(&agent_run_id)
.await?;
output.emit(&AgentSessionMutationReport {
action: "stopped",
agent_run_id: acknowledgement.snapshot.run_id,
state: format!("{:?}", acknowledgement.snapshot.process).to_ascii_lowercase(),
})?;
Ok(())
}
AgentsAction::Remove {
agent_run_id,
config,
socket,
} => {
let socket = resolve_client_socket(config.as_deref(), socket.as_deref()).await?;
let removed = bitrouter::agent_sessions::BackgroundClient::remove_only(socket)
.remove(&agent_run_id)
.await?;
output.emit(&AgentSessionMutationReport {
action: "removed",
agent_run_id: removed,
state: "removed".to_string(),
})?;
Ok(())
}
AgentsAction::List { remote, config, .. } => {
let source = bitrouter::paths::resolve_config(config.as_deref())?;
let cfg = bitrouter::paths::load_config(&source).await?;
let agents = agents_cmd::list(&cfg)
.into_iter()
.map(|row| AgentRow {
id: row.id,
configured: row.configured,
in_catalog: row.in_catalog,
description: row.description,
})
.collect();
let registry = if remote {
let fetched =
bitrouter::agent_registry::fetch(bitrouter::agent_registry::REGISTRY_URL)
.await?;
Some(
agents_cmd::registry_rows(&fetched)
.into_iter()
.map(|row| AgentRegistryRow {
id: row.id,
version: row.version,
install: row.install.to_string(),
description: row.description,
})
.collect(),
)
} else {
None
};
output.emit(&AgentsListReport { agents, registry })?;
Ok(())
}
AgentsAction::Inspect {
agent,
routing,
wait_ms,
source: only,
config,
} => {
let source = bitrouter::paths::resolve_config(config.as_deref())?;
let cfg = bitrouter::paths::load_config(&source).await?;
let agent = bitrouter::acp_cli::resolve_agent_id(&cfg, &agent)?;
let ctx = bitrouter::acp_cli::SpawnContext {
source: &source,
config: cfg,
agent_id: &agent,
options: bitrouter::acp_cli::launch_options(None),
routing,
};
let mut report = bitrouter::acp_cli::commands(ctx, wait_ms).await?;
if let Some(only) = only {
let only = only.into();
report.commands.retain(|row| row.source == only);
}
output.emit(&report)?;
Ok(())
}
AgentsAction::Check {
agent,
routing,
config,
} => {
let source = bitrouter::paths::resolve_config(config.as_deref())?;
let cfg = bitrouter::paths::load_config(&source).await?;
if let Some(agent) = agent {
let agent = bitrouter::acp_cli::resolve_agent_id(&cfg, &agent)?;
let report = bitrouter::acp_cli::spawn_check(cfg, &agent, &routing).await?;
let code = report.exit_code();
output.emit(&report)?;
return exit_with(code);
}
if routing != bitrouter::acp_cli::RoutingOptions::default() {
return Err(bitrouter_sdk::BitrouterError::bad_request(
"routing flags on `agents check` require a specific agent",
)
.into());
}
let agents = agents_cmd::check(&cfg)
.await
.into_iter()
.map(|row| {
let (ok, latency_ms, error) = match row.outcome {
Ok(d) => (true, Some(d.as_millis()), None),
Err(e) => (false, None, Some(e)),
};
AgentCheckRow {
id: row.id,
ok,
latency_ms,
error,
}
})
.collect();
let report = AgentsCheckReport { agents };
let ok = report.agents.iter().all(|agent| agent.ok);
output.emit(&report)?;
if ok { Ok(()) } else { exit_with(1) }
}
AgentsAction::Conformance { id } => {
let harness_id = id
.split_once('/')
.map_or(id.as_str(), |(runtime, harness)| {
if runtime == "local" {
harness
} else {
id.as_str()
}
});
let harness = bitrouter::harness::by_id(harness_id).ok_or_else(|| {
anyhow::anyhow!(
"'{id}' is not a catalog agent. Run `{cli} agents list` to see the ids.",
cli = bitrouter_sdk::invocation::name()
)
})?;
let scratch = tempfile::tempdir()?;
let report = bitrouter::conformance::run(harness, &id, scratch.path()).await;
let verified = report
.tiers
.iter()
.any(|result| matches!(result.outcome, bitrouter::conformance::Outcome::Pass));
let passed = report.passed() && verified;
let registry_block = report.registry_block();
output.emit(&AgentConformanceReport {
agent: report.agent.clone(),
suite: report.suite.to_string(),
suite_version: report.suite_version.to_string(),
agent_version: report.agent_version.clone(),
passed,
tiers: report
.tiers
.iter()
.map(|result| {
use bitrouter::conformance::Outcome;
let (outcome, reason) = match &result.outcome {
Outcome::Pass => ("pass".to_string(), None),
Outcome::Fail { reason } => ("fail".to_string(), Some(reason.clone())),
Outcome::Skipped { reason } => {
("skipped".to_string(), Some(reason.clone()))
}
};
AgentConformanceTier {
tier: result.tier.key().to_string(),
outcome,
reason,
duration_ms: result.duration_ms,
}
})
.collect(),
registry_block,
})?;
if passed {
Ok(())
} else if verified {
anyhow::bail!("{id} did not pass {}", report.suite)
} else {
anyhow::bail!(
"{} checked nothing for {id} — no record is emitted",
report.suite
)
}
}
AgentsAction::Scaffold { id } => scaffold_agent(&id, output).await,
AgentsAction::Install { id } => {
eprintln!(
"note: `{cli} agents install` is a compatibility form; use \
`{cli} agents scaffold`.",
cli = bitrouter_sdk::invocation::name()
);
scaffold_agent(&id, output).await
}
}
}
async fn scaffold_agent(id: &str, output: &Output) -> Result<()> {
use bitrouter::agents as agents_cmd;
match agents_cmd::install(id) {
Ok(yaml) => {
output.emit(&AgentInstallReport {
id: id.to_string(),
yaml,
})?;
Ok(())
}
Err(catalog_miss) => {
let registry =
match bitrouter::agent_registry::fetch(bitrouter::agent_registry::REGISTRY_URL)
.await
{
Ok(registry) => registry,
Err(fetch_err) => {
anyhow::bail!(
"{catalog_miss}\n(also failed to consult the ACP registry: {fetch_err})"
)
}
};
match agents_cmd::install_from_registry(®istry, id) {
Ok(yaml) => {
output.emit(&AgentInstallReport {
id: id.to_string(),
yaml,
})?;
Ok(())
}
Err(e) => anyhow::bail!(e),
}
}
}
}
async fn run_launch(
config: Option<&std::path::Path>,
opts: bitrouter::spawn::SpawnOptions,
output: &bitrouter::output::Output,
) -> Result<()> {
let source = bitrouter::paths::resolve_config(config)?;
let cfg = bitrouter::paths::load_config(&source).await?;
if opts.check {
let report = bitrouter::spawn::check(&cfg, &opts).await?;
output.emit(&report)?;
if report.exit_code() == 0 {
Ok(())
} else {
std::process::exit(report.exit_code());
}
} else {
bitrouter::spawn::run(&source, &cfg, opts).await
}
}
async fn run_native_launch(
agent: &str,
options: NativeLaunchArgs,
output: &bitrouter::output::Output,
) -> Result<()> {
let NativeLaunchArgs {
model,
config,
base_url,
no_install,
no_start,
check,
agent_args,
} = options;
let opts = bitrouter::spawn::SpawnOptions {
agent: bitrouter::spawn::resolve_launch_agent(agent)?,
model,
agent_args,
base_url,
no_install,
no_start,
check,
};
run_launch(config.as_deref(), opts, output).await
}
async fn run_code(
options: CodeArgs,
remote_context: Option<bitrouter::contexts::RemoteContext>,
context_name: Option<&str>,
) -> Result<()> {
let CodeArgs {
agent,
load,
resume,
turn_timeout,
routing,
config,
socket,
} = options;
let initial_session = if let Some(agent) = agent {
if remote_context.is_some() {
return Err(bitrouter_sdk::BitrouterError::bad_request(
"remote ACP conversations are not available; use bare `code` for operations",
)
.into());
}
if socket.is_some() {
return Err(bitrouter_sdk::BitrouterError::bad_request(
"`code <agent>` does not accept --socket",
)
.into());
}
let selection = match (load, resume) {
(Some(session_id), None) => bitrouter::acp_cli::SessionSelection::Load(session_id),
(None, Some(session_id)) => bitrouter::acp_cli::SessionSelection::Resume(session_id),
(None, None) => bitrouter::acp_cli::SessionSelection::New,
(Some(_), Some(_)) => {
return Err(bitrouter_sdk::BitrouterError::bad_request(
"--load and --resume are mutually exclusive",
)
.into());
}
};
Some(bitrouter::dashboard::SessionRequest {
agent,
selection,
turn_timeout,
routing,
})
} else {
if turn_timeout.is_some() || routing != bitrouter::acp_cli::RoutingOptions::default() {
return Err(bitrouter_sdk::BitrouterError::bad_request(
"--turn-timeout, --direct, --base-url, --model, and --no-start apply to \
`code <agent>`, not the agent-selection screen",
)
.into());
}
None
};
let remote = remote_context.map(|context| {
let name = context_name.unwrap_or("remote").to_string();
(name, context)
});
bitrouter::dashboard::run(
remote,
config.as_deref(),
socket.as_deref(),
initial_session,
)
.await
}
async fn resolve_prompt_input(
prompt: Option<String>,
prompt_file: Option<&Path>,
) -> Result<String> {
use std::io::IsTerminal;
use tokio::io::AsyncReadExt;
if let Some(path) = prompt_file {
return tokio::fs::read_to_string(path)
.await
.with_context(|| format!("reading prompt file {}", path.display()));
}
match prompt {
Some(prompt) if prompt == "-" => {}
Some(prompt) => return Ok(prompt),
None if std::io::stdin().is_terminal() => {
return Err(bitrouter_sdk::BitrouterError::bad_request(
"run requires a prompt, `--prompt-file <path>`, or piped stdin",
)
.into());
}
None => {}
}
let mut prompt = String::new();
tokio::io::stdin()
.read_to_string(&mut prompt)
.await
.context("reading prompt from stdin")?;
Ok(prompt)
}
async fn acp_cmd(cmd: AcpCmd, output: &Output) -> Result<()> {
match cmd {
AcpCmd::Evolution { config, action } => {
let source = bitrouter::paths::resolve_config(config.as_deref())?;
let cfg = bitrouter::paths::load_config(&source).await?;
let socket = daemon::socket_path_for(&source, &cfg);
let command = daemon::DaemonCommand::Evolution {
operation: action.operation().await?,
};
match daemon::send_command(&socket, &command)
.await
.context("checkpoint evolution requires the local serving daemon")?
{
daemon::DaemonResponse::Evolution { report } => output.emit(report.as_ref())?,
daemon::DaemonResponse::Error { message } => anyhow::bail!(message),
_ => anyhow::bail!("daemon returned an unexpected evolution response"),
}
Ok(())
}
AcpCmd::Checkpoints {
agent,
session,
config,
action,
} => {
use bitrouter::acp_trajectory::checkpoint::types::{CheckpointReport, RevisionInput};
let source = bitrouter::paths::resolve_config(config.as_deref())?;
let cfg = bitrouter::paths::load_config(&source).await?;
let db = bitrouter::db::connect(&bitrouter::db::anchor_url(
&cfg.database.url,
source.home(),
))
.await?;
bitrouter::db::run_migrations(&db).await?;
let store = bitrouter::acp_trajectory::CanonicalStore::new(db.clone());
let identity = bitrouter::acp_trajectory::SessionIdentity {
owner: "local".into(),
source: agent,
native_session_id: session,
};
let report = match action {
AcpCheckpointAction::Judge { checkpoint, model } => {
let jobs = bitrouter::evolution::jobs::JudgeJobs::new(
bitrouter::evolution::store::EvolutionStore::new(db, &identity.owner)?,
);
let job = jobs.enqueue(&identity, &checkpoint, &model, None).await?;
let mut judge_config = cfg.clone();
judge_config.database.url =
bitrouter::db::anchor_url(&cfg.database.url, source.home());
judge_config.server_tools = Default::default();
judge_config.mcp_servers.clear();
let judge_path = match &source {
bitrouter::paths::ConfigSource::File(path) => path.clone(),
bitrouter::paths::ConfigSource::Default { home } => {
home.join("bitrouter.yaml")
}
};
let assembled =
bitrouter::build_app_with_path(&judge_config, Some(&judge_path)).await?;
let pipeline = assembled
.app
.language_model()
.context("judge requires a language model pipeline")?;
let result = jobs.run(&job.job_id, pipeline).await.with_context(|| {
format!("judge job {} failed; inspect it with judge-job", job.job_id)
})?;
output.emit(&bitrouter::evolution::scoring::ScoringReport::JudgeJob(
result,
))?;
return Ok(());
}
AcpCheckpointAction::JudgeJob { job_id, resume } => {
let jobs = bitrouter::evolution::jobs::JudgeJobs::new(
bitrouter::evolution::store::EvolutionStore::new(db, &identity.owner)?,
);
let job = jobs.get(&job_id).await?;
anyhow::ensure!(
job.identity == identity,
"judge job belongs to another native session"
);
let result = if resume {
let mut judge_config = cfg.clone();
judge_config.database.url =
bitrouter::db::anchor_url(&cfg.database.url, source.home());
judge_config.server_tools = Default::default();
judge_config.mcp_servers.clear();
let judge_path = match &source {
bitrouter::paths::ConfigSource::File(path) => path.clone(),
bitrouter::paths::ConfigSource::Default { home } => {
home.join("bitrouter.yaml")
}
};
let assembled =
bitrouter::build_app_with_path(&judge_config, Some(&judge_path))
.await?;
jobs.run(
&job_id,
assembled
.app
.language_model()
.context("judge requires a language model pipeline")?,
)
.await?
} else {
job
};
output.emit(&bitrouter::evolution::scoring::ScoringReport::JudgeJob(
result,
))?;
return Ok(());
}
AcpCheckpointAction::Rubric { action } => {
output.emit(
&bitrouter::evolution::scoring::command(&store, &identity, action).await?,
)?;
return Ok(());
}
AcpCheckpointAction::Create { watermark } => CheckpointReport::Checkpoint(
store.freeze_checkpoint(&identity, watermark).await?,
),
AcpCheckpointAction::List => {
CheckpointReport::Checkpoints(store.checkpoints(&identity).await?)
}
AcpCheckpointAction::Show { checkpoint } => CheckpointReport::Content(Box::new(
store.checkpoint_content(&identity, &checkpoint).await?,
)),
AcpCheckpointAction::Resources {
checkpoint,
refresh,
} => {
if refresh {
CheckpointReport::Resources(
store
.observe_checkpoint_resources(&identity, &checkpoint)
.await?,
)
} else {
CheckpointReport::ResourceHistory(
store
.checkpoint_resource_history(&identity, &checkpoint)
.await?,
)
}
}
AcpCheckpointAction::Submit { file } => {
let input: RevisionInput = serde_json::from_slice(
&tokio::fs::read(&file)
.await
.context("reading assessment JSON")?,
)?;
CheckpointReport::Revision(store.submit_assessment(&identity, input).await?)
}
AcpCheckpointAction::History => {
CheckpointReport::History(store.assessment_history(&identity).await?)
}
AcpCheckpointAction::Effective => CheckpointReport::Effective(Box::new(
store.effective_assessment(&identity).await?,
)),
AcpCheckpointAction::Family => {
CheckpointReport::Family(Box::new(store.checkpoint_family(&identity).await?))
}
};
output.emit(&report)?;
Ok(())
}
AcpCmd::Recordings { action, config } => {
let source = bitrouter::paths::resolve_config(config.as_deref())?;
let cfg = bitrouter::paths::load_config(&source).await?;
let url = bitrouter::db::anchor_url(&cfg.database.url, source.home());
let db = bitrouter::db::connect(&url).await?;
bitrouter::db::run_migrations(&db).await?;
let store = bitrouter::acp_trajectory::CanonicalStore::new(db);
match action {
AcpRecordingAction::List { agent } => {
output.emit(&bitrouter::acp_trajectory::RecordingReport::Sessions(
store.list("local", &agent).await?,
))?
}
AcpRecordingAction::Show { agent, session } => {
let identity = bitrouter::acp_trajectory::SessionIdentity {
owner: "local".into(),
source: agent,
native_session_id: session,
};
output.emit(&bitrouter::acp_trajectory::RecordingReport::Transcript(
Box::new(store.transcript(&identity).await?),
))?;
}
AcpRecordingAction::Delete { agent, session } => {
let identity = bitrouter::acp_trajectory::SessionIdentity {
owner: "local".into(),
source: agent,
native_session_id: session,
};
store.delete(&identity).await?;
output.emit(&bitrouter::acp_trajectory::RecordingReport::Deleted(
identity,
))?;
}
}
Ok(())
}
AcpCmd::Serve {
agent,
agent_compat,
turn_timeout,
routing,
config,
} => {
let agent = agent.or(agent_compat).ok_or_else(|| {
bitrouter_sdk::BitrouterError::bad_request(format!(
"acp serve requires an agent, for example `{} acp serve claude`",
bitrouter_sdk::invocation::name()
))
})?;
let source = bitrouter::paths::resolve_config(config.as_deref())?;
let cfg = bitrouter::paths::load_config(&source).await?;
let options = bitrouter::acp_cli::launch_options(turn_timeout);
let ctx = bitrouter::acp_cli::SpawnContext {
source: &source,
config: cfg,
agent_id: &agent,
options,
routing,
};
bitrouter::acp_cli::serve(ctx).await
}
AcpCmd::Prompt {
agent,
turn_timeout,
no_wait,
routing,
headless,
config,
text,
} => {
eprintln!(
"note: `{cli} acp prompt` is a compatibility form; use \
`{cli} run <agent> <prompt>`.",
cli = bitrouter_sdk::invocation::name()
);
let source = bitrouter::paths::resolve_config(config.as_deref())?;
let cfg = bitrouter::paths::load_config(&source).await?;
let agent = bitrouter::acp_cli::resolve_agent_id(&cfg, &agent)?;
run_agent_prompt(
&source,
cfg,
&agent,
AgentPromptRequest {
prompt: &text,
routing,
turn_timeout,
no_wait,
result_schema: None,
headless,
session: bitrouter::acp_cli::SessionSelection::New,
cwd: None,
},
)
.await
}
AcpCmd::Commands {
agent,
routing,
wait_ms,
source: only,
config,
} => {
eprintln!(
"note: `{cli} acp commands` is a compatibility form; use \
`{cli} agents inspect <agent>`.",
cli = bitrouter_sdk::invocation::name()
);
let source = bitrouter::paths::resolve_config(config.as_deref())?;
let cfg = bitrouter::paths::load_config(&source).await?;
let agent = bitrouter::acp_cli::resolve_agent_id(&cfg, &agent)?;
let ctx = bitrouter::acp_cli::SpawnContext {
source: &source,
config: cfg,
agent_id: &agent,
options: bitrouter::acp_cli::launch_options(None),
routing,
};
let mut report = bitrouter::acp_cli::commands(ctx, wait_ms).await?;
if let Some(only) = only {
let only = only.into();
report.commands.retain(|row| row.source == only);
}
output.emit(&report)?;
Ok(())
}
}
}
struct AgentPromptRequest<'a> {
prompt: &'a str,
routing: bitrouter::acp_cli::RoutingOptions,
turn_timeout: Option<u64>,
no_wait: bool,
result_schema: Option<&'a str>,
headless: bitrouter::acp_cli::HeadlessOptions,
session: bitrouter::acp_cli::SessionSelection,
cwd: Option<PathBuf>,
}
async fn run_agent_prompt(
source: &bitrouter::paths::ConfigSource,
config: bitrouter_sdk::config::Config,
agent: &str,
request: AgentPromptRequest<'_>,
) -> Result<()> {
let contract = request
.result_schema
.map(bitrouter::result_contract::ResultContract::from_flag)
.transpose()?;
let prompt_options = bitrouter::acp_cli::PromptOptions {
no_wait: request.no_wait,
contract,
policy: request.headless.policy()?,
format: request.headless.format,
session: request.session,
cwd: request.cwd,
};
let mut stdout = tokio::io::stdout();
let ctx = bitrouter::acp_cli::SpawnContext {
source,
config,
agent_id: agent,
options: bitrouter::acp_cli::launch_options(request.turn_timeout),
routing: request.routing,
};
let tally =
bitrouter::acp_cli::prompt(ctx, request.prompt, prompt_options, &mut stdout).await?;
exit_with(tally.exit_code())
}
fn exit_with(code: i32) -> Result<()> {
if code == 0 {
Ok(())
} else {
std::process::exit(code);
}
}
fn pid_path_for(socket: &Path) -> PathBuf {
let mut p = socket.to_path_buf();
p.set_extension("pid");
p
}
#[cfg(unix)]
fn process_is_alive(pid: u32) -> bool {
std::process::Command::new("kill")
.args(["-0", &pid.to_string()])
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.status()
.map(|s| s.success())
.unwrap_or(false)
}
#[cfg(windows)]
fn process_is_alive(pid: u32) -> bool {
let output = std::process::Command::new("tasklist")
.args(["/FI", &format!("PID eq {pid}"), "/NH", "/FO", "CSV"])
.stdout(std::process::Stdio::piped())
.stderr(std::process::Stdio::null())
.output();
match output {
Ok(out) => {
let text = String::from_utf8_lossy(&out.stdout);
text.contains(&format!("\"{pid}\""))
}
Err(_) => false,
}
}
#[cfg(test)]
mod tests {
use super::*;
use clap::CommandFactory;
#[test]
fn run_is_the_unambiguous_headless_agent_surface() -> anyhow::Result<()> {
let cli = Cli::try_parse_from([
"bitrouter",
"run",
"claude-acp",
"summarize this repository",
"--format",
"quiet",
])?;
match cli.command {
Some(Command::Run {
agent,
prompt,
headless,
..
}) => {
assert_eq!(agent, "claude-acp");
assert_eq!(prompt.as_deref(), Some("summarize this repository"));
assert_eq!(headless.format, bitrouter::acp_cli::PromptFormat::Quiet);
}
_ => anyhow::bail!("run did not parse as the headless agent command"),
}
Ok(())
}
#[test]
fn background_run_is_explicit_and_not_no_wait() -> anyhow::Result<()> {
let cli = Cli::try_parse_from([
"bitrouter",
"run",
"codex",
"audit this repository",
"--background",
"--allow-shared-directory",
])?;
assert!(matches!(
cli.command,
Some(Command::Run {
background: true,
allow_shared_directory: true,
no_wait: false,
..
})
));
assert!(
Cli::try_parse_from([
"bitrouter",
"run",
"codex",
"audit",
"--background",
"--no-wait",
])
.is_err()
);
assert!(
Cli::try_parse_from([
"bitrouter",
"run",
"codex",
"audit",
"--allow-shared-directory",
])
.is_err()
);
Ok(())
}
#[test]
fn agents_keeps_catalog_verbs_and_adds_session_management() -> anyhow::Result<()> {
let manager = Cli::try_parse_from(["bitrouter", "agents"])?;
assert!(matches!(
manager.command,
Some(Command::Agents { action: None })
));
let sessions = Cli::try_parse_from(["bitrouter", "agents", "sessions", "--json"])?;
assert!(sessions.json);
assert!(matches!(
sessions.command,
Some(Command::Agents {
action: Some(AgentsAction::Sessions { .. })
})
));
for args in [
["bitrouter", "agents", "attach", "run-1"],
["bitrouter", "agents", "stop", "run-1"],
["bitrouter", "agents", "remove", "run-1"],
["bitrouter", "agents", "inspect", "codex"],
["bitrouter", "agents", "check", "codex"],
["bitrouter", "agents", "conformance", "codex"],
["bitrouter", "agents", "scaffold", "codex"],
] {
assert!(
Cli::try_parse_from(args).is_ok(),
"failed to parse {args:?}"
);
}
Ok(())
}
#[test]
fn code_distinguishes_dashboard_from_agent_session() -> anyhow::Result<()> {
let dashboard = Cli::try_parse_from(["bitrouter", "--context", "work", "code"])?;
assert_eq!(dashboard.context.as_deref(), Some("work"));
assert!(matches!(
dashboard.command,
Some(Command::Code {
options: CodeArgs { agent: None, .. }
})
));
let session = Cli::try_parse_from(["bitrouter", "code", "codex"])?;
assert!(matches!(
session.command,
Some(Command::Code {
options: CodeArgs {
agent: Some(agent),
..
}
}) if agent == "codex"
));
let loaded =
Cli::try_parse_from(["bitrouter", "code", "claude", "--load", "native-session"])?;
assert!(matches!(
loaded.command,
Some(Command::Code {
options: CodeArgs {
agent: Some(agent),
load: Some(session),
..
}
}) if agent == "claude" && session == "native-session"
));
Ok(())
}
#[test]
fn canonical_agent_commands_are_positional() -> anyhow::Result<()> {
let launch = Cli::try_parse_from(["bitrouter", "launch", "codex", "--", "--search"])?;
assert!(matches!(
launch.command,
Some(Command::Launch {
agent: Some(agent),
options: NativeLaunchArgs { agent_args, .. },
..
}) if agent == "codex" && agent_args == ["--search"]
));
let serve = Cli::try_parse_from(["bitrouter", "acp", "serve", "claude"])?;
assert!(matches!(
serve.command,
Some(Command::Acp {
cmd: AcpCmd::Serve {
agent: Some(agent),
..
}
}) if agent == "claude"
));
Ok(())
}
#[test]
fn compatibility_commands_are_hidden_from_help() {
let command = Cli::command();
let visible: Vec<_> = command
.get_subcommands()
.filter(|command| !command.is_hide_set())
.map(clap::Command::get_name)
.collect();
for hidden in ["spawn", "chat", "tui", "tools"] {
assert!(!visible.contains(&hidden), "{hidden} leaked into root help");
}
for canonical in [
"code", "run", "requests", "checks", "launch", "claude", "codex",
] {
assert!(
visible.contains(&canonical),
"{canonical} missing from root help"
);
}
}
#[test]
fn agent_and_protocol_help_scopes_global_compatibility_flags() {
let mut command = command_for_invocation("bro");
for name in ["code", "run", "launch", "claude", "codex"] {
let help = command
.find_subcommand_mut(name)
.map(clap::Command::render_long_help)
.map(|help| help.to_string())
.unwrap_or_default();
for hidden in ["--json", "--human", "--context"] {
assert!(!help.contains(hidden), "{hidden} leaked into {name} help");
}
}
let acp = command
.find_subcommand_mut("acp")
.and_then(|acp| acp.find_subcommand_mut("serve"))
.map(clap::Command::render_long_help)
.map(|help| help.to_string())
.unwrap_or_default();
assert!(acp.contains("bro acp serve <AGENT>"));
assert!(!acp.contains("--json"));
assert!(!acp.contains("--human"));
assert!(!acp.contains("--context"));
let launch = command
.find_subcommand_mut("launch")
.map(clap::Command::render_long_help)
.map(|help| help.to_string())
.unwrap_or_default();
assert!(launch.contains("bro launch <AGENT>"));
let mut alias = command_for_invocation("bitrouter");
let alias_acp = alias
.find_subcommand_mut("acp")
.and_then(|acp| acp.find_subcommand_mut("serve"))
.map(clap::Command::render_long_help)
.map(|help| help.to_string())
.unwrap_or_default();
assert!(alias_acp.contains("bitrouter acp serve <AGENT>"));
let alias_launch = alias
.find_subcommand_mut("launch")
.map(clap::Command::render_long_help)
.map(|help| help.to_string())
.unwrap_or_default();
assert!(alias_launch.contains("bitrouter launch <AGENT>"));
}
#[test]
fn config_log_level_is_used_when_rust_log_is_unset() {
let (filter, warning) = resolve_env_filter_from(None, Some("debug"));
assert_eq!(filter.to_string(), "debug");
assert!(warning.is_none());
}
#[test]
fn config_log_level_accepts_per_target_filter_syntax() {
let (filter, warning) = resolve_env_filter_from(None, Some("warn,bitrouter=trace"));
let rendered = filter.to_string();
assert!(rendered.contains("warn"), "{rendered}");
assert!(rendered.contains("bitrouter=trace"), "{rendered}");
assert!(warning.is_none());
}
#[test]
fn rust_log_wins_over_config_log_level() {
let (filter, warning) = resolve_env_filter_from(Some("trace"), Some("error"));
assert_eq!(filter.to_string(), "trace");
assert!(warning.is_none());
}
#[test]
fn defaults_to_info_when_neither_source_is_set() {
let (filter, warning) = resolve_env_filter_from(None, None);
assert_eq!(filter.to_string(), DEFAULT_LOG_FILTER);
assert!(warning.is_none());
}
#[test]
fn blank_sources_fall_through_rather_than_erroring() {
let (filter, warning) = resolve_env_filter_from(Some(" "), Some("debug"));
assert_eq!(filter.to_string(), "debug");
assert!(warning.is_none());
let (filter, warning) = resolve_env_filter_from(None, Some(""));
assert_eq!(filter.to_string(), DEFAULT_LOG_FILTER);
assert!(warning.is_none());
}
#[test]
fn invalid_config_log_level_warns_and_falls_back() {
let (filter, warning) = resolve_env_filter_from(None, Some("not-a-level"));
assert_eq!(filter.to_string(), DEFAULT_LOG_FILTER);
let warning = warning.expect("an unparseable log_level must be reported");
assert!(warning.contains("server.log_level"), "{warning}");
assert!(warning.contains("not-a-level"), "{warning}");
}
#[test]
fn typo_level_does_not_silently_mute_the_daemon() {
assert_eq!(
tracing_subscriber::EnvFilter::try_new("dbug")
.expect("EnvFilter accepts a bare word as a target")
.to_string(),
"dbug=trace",
);
let (filter, warning) = resolve_env_filter_from(None, Some("dbug"));
assert_eq!(filter.to_string(), DEFAULT_LOG_FILTER);
assert!(warning.is_some(), "a typo'd level must warn, not mute");
}
#[test]
fn every_level_name_is_accepted() {
for level in ["trace", "debug", "info", "warn", "error", "off"] {
let (filter, warning) = resolve_env_filter_from(None, Some(level));
assert!(warning.is_none(), "{level} should be valid: {warning:?}");
assert_eq!(filter.to_string(), level);
}
}
#[test]
fn invalid_rust_log_warns_and_does_not_fall_back_to_config() {
let (filter, warning) = resolve_env_filter_from(Some("not-a-level"), Some("debug"));
assert_eq!(filter.to_string(), DEFAULT_LOG_FILTER);
let warning = warning.expect("an unparseable RUST_LOG must be reported");
assert!(warning.contains("RUST_LOG"), "{warning}");
}
#[test]
fn server_config_default_log_level_is_a_valid_filter() {
let default_level = config::ServerConfig::default().log_level;
let (filter, warning) = resolve_env_filter_from(None, Some(&default_level));
assert_eq!(filter.to_string(), DEFAULT_LOG_FILTER);
assert!(warning.is_none());
}
enum RestartCommandKind {
Status,
Stop,
}
impl RestartCommandKind {
fn matches(&self, command: &DaemonCommand) -> bool {
matches!(
(self, command),
(Self::Status, DaemonCommand::Status) | (Self::Stop, DaemonCommand::Stop)
)
}
}
#[derive(Clone)]
struct RestartGateHarness {
old_pid: u32,
alive: Arc<std::sync::atomic::AtomicBool>,
endpoint_in_use: Arc<std::sync::atomic::AtomicBool>,
sent_commands: Arc<std::sync::atomic::AtomicUsize>,
}
impl RestartGateHarness {
fn new(old_pid: u32, alive: bool, endpoint_in_use: bool) -> Self {
Self {
old_pid,
alive: Arc::new(std::sync::atomic::AtomicBool::new(alive)),
endpoint_in_use: Arc::new(std::sync::atomic::AtomicBool::new(endpoint_in_use)),
sent_commands: Arc::new(std::sync::atomic::AtomicUsize::new(0)),
}
}
fn spawn(&self, target: Option<RestartTarget>) -> tokio::task::JoinHandle<RestartRelease> {
let expected_pid = self.old_pid;
let alive = self.alive.clone();
let endpoint_in_use = self.endpoint_in_use.clone();
tokio::spawn(async move {
await_restart_release(
Path::new("restart.sock"),
target,
move |pid| {
pid == expected_pid && alive.load(std::sync::atomic::Ordering::SeqCst)
},
move |_| endpoint_in_use.load(std::sync::atomic::Ordering::SeqCst),
)
.await
})
}
fn spawn_control(
&self,
pidfile_pid: Option<u32>,
endpoint_was_in_use: bool,
responses: Vec<(RestartCommandKind, anyhow::Result<DaemonResponse>)>,
) -> tokio::task::JoinHandle<anyhow::Result<RestartRelease>> {
let expected_pid = self.old_pid;
let alive = self.alive.clone();
let endpoint_in_use = self.endpoint_in_use.clone();
let sent_commands = self.sent_commands.clone();
tokio::spawn(async move {
let mut responses = std::collections::VecDeque::from(responses);
restart_control_phase(
Path::new("restart.sock"),
pidfile_pid,
endpoint_was_in_use,
move |command| {
sent_commands.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
let response = match responses.pop_front() {
Some((expected, response)) if expected.matches(&command) => response,
Some(_) => Err(anyhow::anyhow!("unexpected daemon command order")),
None => Err(anyhow::anyhow!("unexpected extra daemon command")),
};
std::future::ready(response)
},
move |pid| {
pid == expected_pid && alive.load(std::sync::atomic::Ordering::SeqCst)
},
move |_| endpoint_in_use.load(std::sync::atomic::Ordering::SeqCst),
)
.await
})
}
fn set_alive(&self, alive: bool) {
self.alive.store(alive, std::sync::atomic::Ordering::SeqCst);
}
fn set_endpoint_in_use(&self, endpoint_in_use: bool) {
self.endpoint_in_use
.store(endpoint_in_use, std::sync::atomic::Ordering::SeqCst);
}
fn sent_commands(&self) -> usize {
self.sent_commands.load(std::sync::atomic::Ordering::SeqCst)
}
}
#[tokio::test(start_paused = true)]
async fn restart_waits_for_exact_old_pid_after_endpoint_release() -> anyhow::Result<()> {
let harness = RestartGateHarness::new(4_242, true, false);
let gate = harness.spawn(Some(RestartTarget { pid: 4_242 }));
tokio::task::yield_now().await;
assert!(
!gate.is_finished(),
"endpoint release incorrectly proved exact process exit"
);
harness.set_alive(false);
tokio::time::advance(RESTART_POLL_INTERVAL).await;
tokio::task::yield_now().await;
let release = gate.await?;
assert!(release.ready);
Ok(())
}
#[tokio::test(start_paused = true)]
async fn restart_pid_reuse_timeout_never_signals_the_reused_process() -> anyhow::Result<()> {
let harness = RestartGateHarness::new(4_252, true, false);
let gate = harness.spawn(Some(RestartTarget { pid: 4_252 }));
tokio::task::yield_now().await;
harness.set_alive(false);
harness.set_alive(true);
tokio::time::advance(RESTART_GRACE_PERIOD).await;
tokio::task::yield_now().await;
let release = gate.await?;
assert!(
!release.ready,
"reused live pid was treated as released instead of returning a bounded error"
);
Ok(())
}
#[tokio::test(start_paused = true)]
async fn restart_pidfile_only_target_times_out_without_force_kill() -> anyhow::Result<()> {
let harness = RestartGateHarness::new(4_248, true, false);
let gate = harness.spawn_control(Some(4_248), false, vec![]);
tokio::task::yield_now().await;
tokio::time::advance(RESTART_GRACE_PERIOD).await;
tokio::task::yield_now().await;
let release = gate.await??;
assert!(!release.ready, "stale live pid evidence became ready");
assert_eq!(harness.sent_commands(), 0);
Ok(())
}
#[tokio::test(start_paused = true)]
async fn restart_control_status_pid_survives_missing_pidfile_and_endpoint_release()
-> anyhow::Result<()> {
let harness = RestartGateHarness::new(4_249, true, false);
let control = harness.spawn_control(
None,
true,
vec![
(
RestartCommandKind::Status,
Ok(DaemonResponse::Status {
pid: 4_249,
listen: "127.0.0.1:4356".to_string(),
models: 0,
providers: Vec::new(),
saved_routers: None,
running_routers: None,
router_restart_required: None,
config_state: None,
}),
),
(RestartCommandKind::Stop, Ok(DaemonResponse::Ok)),
],
);
tokio::task::yield_now().await;
assert_eq!(harness.sent_commands(), 2);
assert!(
!control.is_finished(),
"endpoint release bypassed Status pid"
);
harness.set_alive(false);
tokio::time::advance(RESTART_POLL_INTERVAL).await;
tokio::task::yield_now().await;
let release = control.await??;
assert!(release.ready);
Ok(())
}
#[tokio::test(start_paused = true)]
async fn restart_control_status_pid_times_out_even_with_different_pidfile() -> anyhow::Result<()>
{
let harness = RestartGateHarness::new(4_250, true, false);
let control = harness.spawn_control(
Some(9_999),
true,
vec![
(
RestartCommandKind::Status,
Ok(DaemonResponse::Status {
pid: 4_250,
listen: "127.0.0.1:4356".to_string(),
models: 0,
providers: Vec::new(),
saved_routers: None,
running_routers: None,
router_restart_required: None,
config_state: None,
}),
),
(RestartCommandKind::Stop, Ok(DaemonResponse::Ok)),
],
);
tokio::task::yield_now().await;
tokio::time::advance(RESTART_GRACE_PERIOD).await;
tokio::task::yield_now().await;
assert_eq!(harness.sent_commands(), 2);
let release = control.await??;
assert!(!release.ready);
Ok(())
}
#[tokio::test(start_paused = true)]
async fn restart_status_failures_never_promote_pidfile_to_kill_authority() -> anyhow::Result<()>
{
let cases = [
(
"zero pid",
Ok(DaemonResponse::Status {
pid: 0,
listen: "127.0.0.1:4356".to_string(),
models: 0,
providers: Vec::new(),
saved_routers: None,
running_routers: None,
router_restart_required: None,
config_state: None,
}),
),
(
"daemon error",
Ok(DaemonResponse::Error {
message: "status unavailable".to_string(),
}),
),
("unexpected response", Ok(DaemonResponse::Ok)),
(
"transport error",
Err(anyhow::anyhow!("control transport unavailable")),
),
];
for (name, response) in cases {
let harness = RestartGateHarness::new(4_251, true, false);
let control = harness.spawn_control(
Some(4_251),
true,
vec![(RestartCommandKind::Status, response)],
);
let outcome = control.await?;
assert!(outcome.is_err(), "{name} did not fail closed");
assert_eq!(
harness.sent_commands(),
1,
"{name} sent Stop without authenticated process identity"
);
}
Ok(())
}
#[tokio::test(start_paused = true)]
async fn restart_waits_for_endpoint_after_exact_old_pid_exits() -> anyhow::Result<()> {
let harness = RestartGateHarness::new(4_243, false, true);
let gate = harness.spawn(Some(RestartTarget { pid: 4_243 }));
tokio::task::yield_now().await;
assert!(
!gate.is_finished(),
"exact process exit bypassed endpoint cleanup"
);
harness.set_endpoint_in_use(false);
tokio::time::advance(RESTART_POLL_INTERVAL).await;
tokio::task::yield_now().await;
let release = gate.await?;
assert!(release.ready);
Ok(())
}
#[tokio::test(start_paused = true)]
async fn restart_without_valid_pid_uses_endpoint_fallback_without_force_kill()
-> anyhow::Result<()> {
let harness = RestartGateHarness::new(4_246, false, true);
let gate = harness.spawn(None);
tokio::task::yield_now().await;
tokio::time::advance(RESTART_GRACE_PERIOD - RESTART_POLL_INTERVAL).await;
tokio::task::yield_now().await;
assert!(!gate.is_finished());
harness.set_endpoint_in_use(false);
tokio::time::advance(RESTART_POLL_INTERVAL).await;
tokio::task::yield_now().await;
let release = gate.await?;
assert!(release.ready);
let timed_out_harness = RestartGateHarness::new(4_247, false, true);
let timed_out = timed_out_harness.spawn(None);
tokio::task::yield_now().await;
tokio::time::advance(RESTART_GRACE_PERIOD).await;
tokio::task::yield_now().await;
let release = timed_out.await?;
assert!(!release.ready);
Ok(())
}
#[tokio::test]
async fn outer_shutdown_keeps_http_future_until_required_drain_recovers() -> anyhow::Result<()>
{
for control_trigger in [true, false] {
let accepting = Arc::new(std::sync::atomic::AtomicBool::new(true));
let drain_attempts = Arc::new(std::sync::atomic::AtomicUsize::new(0));
let recovery = Arc::new(tokio::sync::Notify::new());
let inflight_release = Arc::new(tokio::sync::Notify::new());
let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel();
let (control_tx, control_rx) = tokio::sync::oneshot::channel();
let (control_shutdown_tx, control_shutdown_rx) = tokio::sync::oneshot::channel();
let (term_tx, term_rx) = tokio::sync::oneshot::channel();
let http_accepting = accepting.clone();
let http_attempts = drain_attempts.clone();
let http_recovery = recovery.clone();
let http_inflight_release = inflight_release.clone();
let http = async move {
shutdown_rx
.await
.map_err(|_| anyhow::anyhow!("outer shutdown sender disappeared"))?;
http_accepting.store(false, std::sync::atomic::Ordering::SeqCst);
http_inflight_release.notified().await;
http_attempts.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
http_recovery.notified().await;
http_attempts.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
Ok(())
};
let control = async move {
tokio::select! {
result = control_rx => {
result.map_err(|_| anyhow::anyhow!("test control sender disappeared"))?;
}
result = control_shutdown_rx => {
result.map_err(|_| anyhow::anyhow!("test control shutdown sender disappeared"))?;
}
}
Ok(())
};
let term = async move {
term_rx
.await
.map_err(|_| anyhow::anyhow!("test term sender disappeared"))?;
Ok(())
};
let hup = async move {
if control_trigger {
return Err(anyhow::anyhow!("test HUP setup unavailable"));
}
std::future::pending::<anyhow::Result<()>>().await
};
let supervision = tokio::spawn(supervise_http_shutdown(
http,
control,
hup,
term,
shutdown_tx,
control_shutdown_tx,
));
tokio::task::yield_now().await;
assert!(
!supervision.is_finished(),
"a HUP setup error must not drop the HTTP server"
);
assert!(accepting.load(std::sync::atomic::Ordering::SeqCst));
if control_trigger {
control_tx
.send(())
.map_err(|_| anyhow::anyhow!("test control receiver disappeared"))?;
} else {
term_tx
.send(())
.map_err(|_| anyhow::anyhow!("test term receiver disappeared"))?;
}
tokio::task::yield_now().await;
assert!(!accepting.load(std::sync::atomic::Ordering::SeqCst));
assert_eq!(drain_attempts.load(std::sync::atomic::Ordering::SeqCst), 0);
assert!(
!supervision.is_finished(),
"in-flight HTTP work must finish before required drain"
);
inflight_release.notify_one();
tokio::task::yield_now().await;
assert_eq!(drain_attempts.load(std::sync::atomic::Ordering::SeqCst), 1);
assert!(
!supervision.is_finished(),
"the outer supervisor dropped a server waiting on required drain"
);
recovery.notify_one();
tokio::task::yield_now().await;
assert_eq!(drain_attempts.load(std::sync::atomic::Ordering::SeqCst), 2);
supervision.await??;
}
Ok(())
}
#[tokio::test]
async fn outer_shutdown_returns_http_errors_without_waiting_for_a_trigger() -> anyhow::Result<()>
{
let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel();
let (control_shutdown_tx, control_shutdown_rx) = tokio::sync::oneshot::channel();
let error = supervise_http_shutdown(
async { Err(anyhow::anyhow!("test HTTP failure")) },
async move {
control_shutdown_rx
.await
.map_err(|_| anyhow::anyhow!("test control shutdown sender disappeared"))?;
Ok(())
},
std::future::pending::<anyhow::Result<()>>(),
std::future::pending::<anyhow::Result<()>>(),
shutdown_tx,
control_shutdown_tx,
)
.await
.err()
.ok_or_else(|| anyhow::anyhow!("HTTP failure unexpectedly succeeded"))?;
assert_eq!(error.to_string(), "test HTTP failure");
assert!(shutdown_rx.await.is_err());
Ok(())
}
#[test]
fn trajectory_inspect_replay_and_prune_flags_parse() {
use clap::Parser;
let inspect = Cli::try_parse_from([
"bitrouter",
"trajectory",
"--config",
"/tmp/custom-bitrouter.yaml",
"inspect",
"episode-1",
"--json",
])
.expect("parse trajectory inspect");
assert!(inspect.json);
assert!(matches!(
inspect.command,
Some(Command::Trajectory {
config: Some(config),
action: TrajectoryAction::Inspect { episode_id },
}) if episode_id == "episode-1" && config == Path::new("/tmp/custom-bitrouter.yaml")
));
let replay = Cli::try_parse_from([
"bitrouter",
"trajectory",
"replay",
"episode-2",
"--config",
"/tmp/custom-bitrouter.yaml",
"--human",
])
.expect("parse trajectory replay");
assert!(replay.human);
assert!(matches!(
replay.command,
Some(Command::Trajectory {
config: Some(config),
action: TrajectoryAction::Replay { episode_id },
}) if episode_id == "episode-2" && config == Path::new("/tmp/custom-bitrouter.yaml")
));
let prune = Cli::try_parse_from([
"bitrouter",
"trajectory",
"prune",
"--before",
"2026-08-01T00:00:00Z",
"--dry-run",
])
.expect("parse trajectory prune");
assert!(matches!(
prune.command,
Some(Command::Trajectory {
config: None,
action: TrajectoryAction::Prune { before, dry_run: true },
}) if before == "2026-08-01T00:00:00Z"
));
assert!(
Cli::try_parse_from([
"bitrouter",
"trajectory",
"prune",
"--before",
"not-a-timestamp",
])
.is_err()
);
}
#[tokio::test]
async fn trajectory_store_anchors_file_and_zero_config_databases_to_daemon_home()
-> anyhow::Result<()> {
use sea_orm::{ConnectionTrait, DatabaseBackend, Statement};
async fn seed_episode(database_url: &str, episode_id: &str) -> anyhow::Result<()> {
let db = bitrouter::db::connect(database_url).await?;
bitrouter::db::run_migrations(&db).await?;
db.execute(Statement::from_sql_and_values(
DatabaseBackend::Sqlite,
"INSERT INTO trajectory_episodes \
(episode_id, owner_user_id, correlation_source, correlation_key_id, \
correlation_digest, history_completeness, next_sequence, first_captured_at, \
last_captured_at, closed_at, latest_request_id) \
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
[
episode_id.into(),
"owner-home".into(),
"explicit_root".into(),
"key-home".into(),
format!("hmac-sha256:key-home:{}", "0".repeat(64)).into(),
"complete".into(),
1_i64.into(),
"2026-08-01T00:00:00Z".into(),
"2026-08-01T00:00:00Z".into(),
Option::<String>::None.into(),
Option::<String>::None.into(),
],
))
.await?;
Ok(())
}
let file_home = tempfile::tempdir()?;
let config_path = file_home.path().join("bitrouter.yaml");
std::fs::write(
&config_path,
"inherit_defaults: false\ndatabase:\n url: sqlite://./daemon.db\n",
)?;
let file_db = file_home.path().join("daemon.db");
seed_episode(
&format!("sqlite://{}?mode=rwc", file_db.display()),
"episode-file-home",
)
.await?;
assert_ne!(std::env::current_dir()?, file_home.path());
let file_source = bitrouter::paths::ConfigSource::File(config_path);
let (_, file_store) = local_trajectory_store(&file_source).await?;
assert_eq!(
file_store
.resolve_episode_owner("episode-file-home")
.await?
.as_deref(),
Some("owner-home")
);
let zero_home = tempfile::tempdir()?;
let zero_db = zero_home.path().join("bitrouter.db");
seed_episode(
&format!("sqlite://{}?mode=rwc", zero_db.display()),
"episode-zero-home",
)
.await?;
assert_ne!(std::env::current_dir()?, zero_home.path());
let zero_source = bitrouter::paths::ConfigSource::Default {
home: zero_home.path().to_path_buf(),
};
let (_, zero_store) = local_trajectory_store(&zero_source).await?;
assert_eq!(
zero_store
.resolve_episode_owner("episode-zero-home")
.await?
.as_deref(),
Some("owner-home")
);
Ok(())
}
#[test]
fn reconcile_metering_accepts_explicit_cloud_credentials_file() {
use clap::Parser;
let cli = Cli::try_parse_from([
"bitrouter",
"workflow-state",
"reconcile-metering",
"--database-url",
"sqlite:///tmp/usage.db",
"--credentials-file",
"/tmp/account-credentials.json",
"--request-id",
"req-1",
])
.expect("parse credentials file");
match cli.command {
Some(Command::WorkflowState {
action:
WorkflowStateAction::ReconcileMetering {
credentials_file, ..
},
}) => assert_eq!(
credentials_file.as_deref(),
Some(Path::new("/tmp/account-credentials.json"))
),
_ => panic!("expected reconcile-metering"),
}
}
#[tokio::test]
async fn settlement_explicit_key_bypasses_malformed_credentials_file() -> anyhow::Result<()> {
let directory = tempfile::tempdir()?;
let path = directory.path().join("account-credentials.json");
std::fs::write(&path, b"malformed")?;
let key = settlement_api_key(
Some("brk_explicit.secret"),
Some(&path),
"https://api.bitrouter.ai/v1",
)
.await?;
assert_eq!(key, "brk_explicit.secret");
Ok(())
}
#[tokio::test]
async fn settlement_credentials_file_rejects_oauth() -> anyhow::Result<()> {
use bitrouter_providers::hosted::account::credentials::{Credentials, StoredCredential};
use bitrouter_providers::hosted::account::manager::CredentialManager;
let directory = tempfile::tempdir()?;
let path = directory.path().join("account-credentials.json");
let manager = CredentialManager::with_client(&path, reqwest::Client::new());
manager
.save(StoredCredential::from(Credentials {
access_token: "oauth-access".to_owned(),
refresh_token: Some("oauth-refresh".to_owned()),
expires_at: chrono::Utc::now() + chrono::Duration::minutes(10),
refresh_token_expires_at: None,
token_type: "Bearer".to_owned(),
scope: "inference:invoke".to_owned(),
client_id: "bitrouter-cli".to_owned(),
authorization_server: "https://api.bitrouter.ai".to_owned(),
namespace_id: Some("ns-test".to_owned()),
subject: None,
}))
.await?;
let error = match settlement_api_key(None, Some(&path), "https://api.bitrouter.ai/v1").await
{
Ok(_) => anyhow::bail!("OAuth unexpectedly resolved for settlement"),
Err(error) => error,
};
assert!(
error
.to_string()
.contains("requires a static BitRouter API key")
);
Ok(())
}
#[tokio::test]
async fn settlement_credentials_file_rejects_wrong_origin() -> anyhow::Result<()> {
use bitrouter_providers::hosted::account::credentials::StoredCredential;
use bitrouter_providers::hosted::account::manager::CredentialManager;
let directory = tempfile::tempdir()?;
let path = directory.path().join("account-credentials.json");
let manager = CredentialManager::with_client(&path, reqwest::Client::new());
manager
.save(StoredCredential::api_key(
"brk_stored.secret".to_owned(),
"https://other.example".to_owned(),
))
.await?;
let error = match settlement_api_key(None, Some(&path), "https://api.bitrouter.ai/v1").await
{
Ok(_) => anyhow::bail!("credential unexpectedly crossed origins"),
Err(error) => error,
};
assert!(error.to_string().contains("does not match"));
Ok(())
}
#[test]
fn every_tui_command_has_an_actions_row() {
use bitrouter::actions::ACTIONS;
use bitrouter_tui::machine::{ALIASES, REDUCER_OWNED};
let named: Vec<&str> = ACTIONS.iter().filter_map(|row| row.tui_command).collect();
for owned in REDUCER_OWNED {
let row = ACTIONS.iter().find(|row| &row.id == owned);
match row {
Some(row) => assert!(
row.tui_command.is_some(),
"`{owned}` is dispatched by the reducer but its row carries no \
`tui_command`, so nothing can reach it"
),
None => panic!(
"`{owned}` is dispatched by the reducer but is not an `ACTIONS` row. \
Known rows: {:?}",
ACTIONS.iter().map(|row| row.id).collect::<Vec<_>>()
),
}
}
for (index, name) in named.iter().enumerate() {
assert!(
!named[index + 1..].contains(name),
"two rows both claim `/{name}`; a name reaches one action"
);
}
for (alias, target) in ALIASES {
assert!(
named.contains(target),
"alias `/{alias}` points at `{target}`, which no row offers"
);
assert!(
!named.contains(alias),
"`/{alias}` is both an alias and a command's own name"
);
}
for row in ACTIONS.iter().filter(|row| row.tui_command.is_some()) {
assert!(
!bitrouter::actions::session::summary_for(row.id).is_empty(),
"`{}` is offered in a session but `summary_for` has no arm for it",
row.id
);
}
}
#[test]
fn every_actions_row_resolves_to_a_cli_leaf() {
use clap::CommandFactory;
let root = Cli::command();
for action in bitrouter::actions::ACTIONS {
let Some(leaf) = action.cli_leaf else {
continue;
};
let mut command = &root;
for segment in leaf.split_whitespace() {
let found = command
.get_subcommands()
.find(|sub| sub.get_name() == segment);
match found {
Some(sub) => command = sub,
None => panic!(
"action `{}` names CLI leaf `{leaf}`, but `{segment}` is not a \
subcommand of `{}`",
action.id,
command.get_name()
),
}
}
}
}
fn control_action_arguments(leaf: &str) -> Option<Vec<&'static str>> {
let args = match leaf {
"reload" => vec!["bitrouter", "--context", "workstation", "reload"],
"status" => vec!["bitrouter", "--context", "workstation", "status"],
"models" => vec![
"bitrouter",
"--context",
"workstation",
"models",
"--provider",
"openai",
],
"route" => vec![
"bitrouter",
"--context",
"workstation",
"route",
"openai/gpt-5",
],
"requests" => vec![
"bitrouter",
"--context",
"workstation",
"requests",
"--limit",
"100",
"--since",
"2026-01-01T00:00:00Z",
"--until",
"2026-01-01T01:00:00Z",
"--model",
"openai/gpt-5",
"--provider",
"openai",
],
"providers list" => vec!["bitrouter", "--context", "workstation", "providers", "list"],
"observe status" => vec!["bitrouter", "--context", "workstation", "observe", "status"],
"policy status" => vec![
"bitrouter",
"--context",
"workstation",
"policy",
"status",
"--view",
"active",
],
"policy show" => vec![
"bitrouter",
"--context",
"workstation",
"policy",
"show",
"default",
"--view",
"disk",
],
"agents list" => vec!["bitrouter", "--context", "workstation", "agents", "list"],
"checks" => vec!["bitrouter", "--context", "workstation", "checks"],
"checks probe" => vec![
"bitrouter",
"--context",
"workstation",
"checks",
"probe",
"company",
],
"checks receipts" => vec![
"bitrouter",
"--context",
"workstation",
"checks",
"receipts",
"--limit",
"10",
],
"checks receipt" => vec![
"bitrouter",
"--context",
"workstation",
"checks",
"receipt",
"request-1",
"--incarnation",
"incarnation-1",
],
_ => return None,
};
Some(args)
}
#[test]
fn every_control_action_parses_and_is_remote_eligible() -> anyhow::Result<()> {
for action in bitrouter::remote_control::inventory::ACTIONS {
let args = control_action_arguments(action.cli_leaf).ok_or_else(|| {
anyhow::anyhow!(
"control action `{}` has no Clap argument fixture for `{}`",
action.id,
action.cli_leaf
)
})?;
let cli = Cli::try_parse_from(args)
.with_context(|| format!("parse control action `{}`", action.id))?;
let command = cli.command.ok_or_else(|| {
anyhow::anyhow!("control action `{}` parsed without a command", action.id)
})?;
assert_eq!(remote_cli_leaf(&command), Some(action.cli_leaf));
validate_remote_invocation(&command, true)
.with_context(|| format!("validate control action `{}`", action.id))?;
}
for resource in bitrouter::remote_control::inventory::RESOURCES {
let Some(leaf) = resource.cli_leaf else {
continue;
};
let cli = Cli::try_parse_from([
"bitrouter",
"--context",
"workstation",
"operations",
"show",
"7ab707f8-50d1-4ba0-9aeb-8f4cb196712b",
"--instance",
"a68fed5e-18e0-4ed8-9023-9ad286a8351d",
])?;
let command = cli.command.ok_or_else(|| {
anyhow::anyhow!(
"control resource `{}` parsed without a command",
resource.id
)
})?;
assert_eq!(remote_resource_leaf(&command), Some(leaf));
validate_remote_invocation(&command, true)
.with_context(|| format!("validate control resource `{}`", resource.id))?;
}
Ok(())
}
#[test]
fn remote_control_actions_reject_local_target_flags_before_execution() -> anyhow::Result<()> {
for action in bitrouter::remote_control::inventory::ACTIONS {
let mut args = control_action_arguments(action.cli_leaf).ok_or_else(|| {
anyhow::anyhow!("missing control fixture for `{}`", action.cli_leaf)
})?;
args.extend(["--config", "/must-not-be-read/bitrouter.yaml"]);
if action.cli_leaf != "models" {
args.extend(["--socket", "/must-not-be-read/bitrouter.sock"]);
}
let cli = Cli::try_parse_from(args)
.with_context(|| format!("parse target flags for `{}`", action.id))?;
let command = cli.command.ok_or_else(|| {
anyhow::anyhow!("control action `{}` parsed without a command", action.id)
})?;
let error = validate_remote_invocation(&command, true)
.err()
.ok_or_else(|| anyhow::anyhow!("{} accepted a local target flag", action.id))?;
assert!(
error
.to_string()
.contains("does not accept --config or --socket"),
"{}: {error:#}",
action.id
);
}
Ok(())
}
#[test]
fn local_policy_reads_without_view_keep_the_legacy_contract() {
assert!(is_legacy_local_policy_read(&PolicyAction::Status {
view: None,
config: None,
socket: None,
}));
assert!(is_legacy_local_policy_read(&PolicyAction::Show {
name: "default".to_string(),
view: None,
config: None,
socket: None,
}));
assert!(!is_legacy_local_policy_read(&PolicyAction::Status {
view: Some(PolicyViewArg::Disk),
config: None,
socket: None,
}));
}
#[test]
fn cli_definition_is_valid() {
use clap::CommandFactory;
Cli::command().debug_assert();
}
#[test]
fn mcp_cli_exposes_check_but_not_the_removed_origin_commands() {
use clap::Parser;
assert!(matches!(
Cli::try_parse_from(["bro", "mcp", "check"]),
Ok(Cli {
command: Some(Command::Mcp {
action: McpAction::Check { .. }
}),
..
})
));
assert!(Cli::try_parse_from(["bro", "mcp", "serve"]).is_err());
assert!(Cli::try_parse_from(["bro", "mcp", "install"]).is_err());
}
#[test]
fn first_party_plugins_are_skill_only() {
for (name, manifest) in [
(
"claude",
include_str!("../../../.claude-plugin/plugin.json"),
),
("codex", include_str!("../../../.codex-plugin/plugin.json")),
] {
let parsed = serde_json::from_str::<serde_json::Value>(manifest);
assert!(parsed.is_ok(), "{name} plugin manifest is not valid JSON");
let Ok(parsed) = parsed else {
continue;
};
assert!(
parsed.get("mcpServers").is_none(),
"{name} plugin must not register an OSS origin MCP server"
);
assert!(
!manifest.contains("mcp serve"),
"{name} plugin references the removed origin command"
);
}
}
#[test]
fn generic_eval_commands_parse_as_nested_exchange_operations() {
use clap::Parser;
let submit = Cli::try_parse_from(["bitrouter", "eval", "result", "submit", "result.json"])
.expect("parse eval result submit");
assert!(matches!(
submit.command,
Some(Command::Eval {
action: EvalAction::Result {
action: EvalResultAction::Submit { .. }
}
})
));
let freeze = Cli::try_parse_from([
"bitrouter",
"eval",
"snapshot",
"freeze",
"--at",
"2026-07-30T00:00:00Z",
])
.expect("parse eval snapshot freeze");
assert!(matches!(
freeze.command,
Some(Command::Eval {
action: EvalAction::Snapshot {
action: EvalSnapshotAction::Freeze { .. }
}
})
));
}
#[test]
fn workflow_bundle_allows_eval_exchange_to_own_task_outcomes() {
use clap::Parser;
let cli = Cli::try_parse_from([
"bitrouter",
"workflow-state",
"bundle",
"--run-label",
"generic-eval-run",
"--traces",
"traces.jsonl",
"--cloud-usage",
"usage.jsonl",
"--output-dir",
"artifact",
])
.expect("parse workflow bundle without request-scoped outcomes");
assert!(matches!(
cli.command,
Some(Command::WorkflowState {
action: WorkflowStateAction::Bundle { outcomes: None, .. }
})
));
}
#[test]
fn shadow_classifier_bakeoff_flags_parse_without_a_submission() {
use clap::Parser;
let cli = Cli::try_parse_from([
"bitrouter",
"workflow-state",
"classifier-bakeoff",
"--fixtures",
"fixtures",
"--output",
"bakeoff.json",
])
.expect("parse shadow classifier bake-off");
assert!(matches!(
cli.command,
Some(Command::WorkflowState {
action: WorkflowStateAction::ClassifierBakeoff {
submission: None,
..
}
})
));
}
#[test]
fn policy_oracle_accepts_human_readable_cost_assumptions() {
use clap::Parser;
let cli = Cli::try_parse_from([
"bitrouter",
"workflow-state",
"policy-oracle",
"--traces",
"traces.jsonl",
"--cloud-usage",
"usage.jsonl",
"--policy-lock",
"policy-lock.yaml",
"--policy",
"auto",
"--effective-cost-factor",
"0.24",
"--target-savings",
"0.30",
"--target-savings",
"0.40",
"--output",
"oracle.json",
])
.expect("parse policy counterfactual oracle");
assert!(matches!(
cli.command,
Some(Command::WorkflowState {
action: WorkflowStateAction::PolicyOracle {
effective_cost_factor_ppm: 240_000,
target_savings_ppm,
..
}
}) if target_savings_ppm == vec![300_000, 400_000]
));
}
#[tokio::test]
async fn eval_subject_seal_derives_a_digest_without_the_ledger() -> Result<()> {
use clap::Parser;
let directory = tempfile::tempdir()?;
let draft = directory.path().join("subject-draft.json");
let reversed_draft = directory.path().join("subject-draft-reversed.json");
let sealed = directory.path().join("subject-sealed.json");
let sealed_again = directory.path().join("subject-sealed-again.json");
std::fs::write(
&draft,
r#"{
"schema_version": 1,
"eval_id": "eval-generic-1",
"scope": "task",
"subject_id": "task-generic-1",
"policy_digest": "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"preset": null,
"cohort": null,
"holdout": false,
"decisions": [
{
"decision_id": "decision-1",
"policy": "auto",
"route_projection": "agent_route/v1|code:generation|implement|normal",
"request_key": "agent_route/v1|unknown|implement|normal",
"selected_tier": "economy",
"baseline_tier": "strong",
"policy_digest": "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
}
],
"requested_dimensions": ["quality.pass"],
"evidence": [
{
"evidence_id": "verifier-result",
"kind": "task.verifier",
"digest": "sha256:0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
"redacted": true,
"attributes": {}
},
{
"evidence_id": "audit-result",
"kind": "task.audit",
"digest": "sha256:fedcba9876543210fedcba9876543210fedcba9876543210fedcba9876543210",
"redacted": true,
"attributes": {"status": "clean"}
}
],
"evidence_digest": "",
"observed_at": "2026-07-31T00:00:00Z"
}
"#,
)?;
let mut reversed =
serde_json::from_str::<serde_json::Value>(&std::fs::read_to_string(&draft)?)?;
let evidence = reversed
.get_mut("evidence")
.and_then(serde_json::Value::as_array_mut)
.ok_or_else(|| anyhow::anyhow!("draft evidence must be an array"))?;
evidence.reverse();
std::fs::write(&reversed_draft, serde_json::to_string_pretty(&reversed)?)?;
let cli = Cli::try_parse_from([
"bitrouter",
"eval",
"subject",
"seal",
draft
.to_str()
.ok_or_else(|| anyhow::anyhow!("draft path is not UTF-8"))?,
"--output",
sealed
.to_str()
.ok_or_else(|| anyhow::anyhow!("sealed path is not UTF-8"))?,
])?;
let Some(Command::Eval { action }) = cli.command else {
anyhow::bail!("expected eval subject seal command")
};
eval(action, &Output::new(bitrouter::output::Format::Json)).await?;
let sealed_text = std::fs::read_to_string(&sealed)?;
let subject: bitrouter::eval::types::EvalSubject = serde_json::from_str(&sealed_text)?;
bitrouter::eval::types::validate_subject(&subject)?;
let repeat = Cli::try_parse_from([
"bitrouter",
"eval",
"subject",
"seal",
reversed_draft
.to_str()
.ok_or_else(|| anyhow::anyhow!("reversed draft path is not UTF-8"))?,
"--output",
sealed_again
.to_str()
.ok_or_else(|| anyhow::anyhow!("repeat output path is not UTF-8"))?,
])?;
let Some(Command::Eval { action }) = repeat.command else {
anyhow::bail!("expected repeated eval subject seal command")
};
eval(action, &Output::new(bitrouter::output::Format::Json)).await?;
let repeated_text = std::fs::read_to_string(&sealed_again)?;
let repeated: bitrouter::eval::types::EvalSubject = serde_json::from_str(&repeated_text)?;
bitrouter::eval::types::validate_subject(&repeated)?;
assert_eq!(subject.evidence_digest, repeated.evidence_digest);
assert_eq!(
sealed_text, repeated_text,
"equivalent evidence order must produce deterministic pretty JSON"
);
assert!(
!directory.path().join("bitrouter.db").exists(),
"sealing must not initialize a ledger"
);
Ok(())
}
#[test]
fn headless_permission_flags_parse_on_both_front_doors() {
use bitrouter::acp_cli::{HeadlessOptions, PromptFormat};
use clap::Parser;
let cli = Cli::try_parse_from([
"bitrouter",
"spawn",
"claude-acp",
"-p",
"hi",
"--approve-reads",
"--format",
"quiet",
])
.expect("parse");
match cli.command {
Some(Command::Spawn { headless, .. }) => {
assert!(headless.approve_reads);
assert_eq!(headless.format, PromptFormat::Quiet);
}
_ => panic!("expected `spawn -p`"),
}
let cli = Cli::try_parse_from(["bitrouter", "acp", "prompt", "--agent", "a", "hi"])
.expect("parse");
match cli.command {
Some(Command::Acp {
cmd: AcpCmd::Prompt { headless, .. },
}) => assert_eq!(headless, HeadlessOptions::default()),
_ => panic!("expected `acp prompt`"),
}
assert!(
Cli::try_parse_from([
"bitrouter",
"acp",
"prompt",
"--agent",
"a",
"--approve-all",
"--deny-all",
"hi"
])
.is_err(),
"one mode at a time"
);
}
#[test]
fn update_flags_parse() {
use clap::Parser;
let cli =
Cli::try_parse_from(["bitrouter", "update", "--check", "--tag", "1.0.0-alpha.18"])
.expect("parse");
match cli.command {
Some(Command::Update {
check,
tag,
stable,
restart,
yes,
}) => {
assert!(check);
assert_eq!(tag.as_deref(), Some("1.0.0-alpha.18"));
assert!(!stable && !restart && !yes);
}
_ => panic!("expected Update"),
}
}
#[test]
fn bare_invocation_parses_to_no_subcommand() {
use clap::Parser;
let cli = Cli::try_parse_from(["bitrouter"]).expect("parse");
assert!(cli.command.is_none());
let human = Cli::try_parse_from(["bitrouter", "--human"]).expect("parse");
assert!(human.command.is_none() && human.human);
}
#[test]
fn init_wizard_flags_parse() {
use clap::Parser;
let cli = Cli::try_parse_from([
"bitrouter",
"init",
"--yes",
"--force",
"--reset",
"--api-key",
"brk_abc.secret",
"--provider",
"openai",
"--provider-api-key",
"sk-openai",
"--harness",
"claude",
"--harness",
"codex",
"--after",
"exit",
"--model",
"openai/gpt-5",
])
.expect("parse");
match cli.command {
Some(Command::Init {
yes,
force,
reset,
api_key,
providers,
provider_api_keys,
harnesses,
after,
model,
..
}) => {
assert!(yes && force && reset);
assert_eq!(api_key.as_deref(), Some("brk_abc.secret"));
assert_eq!(providers, vec!["openai"]);
assert_eq!(provider_api_keys, vec!["sk-openai"]);
assert_eq!(
harnesses,
vec![
bitrouter::spawn::SpawnAgent::Claude,
bitrouter::spawn::SpawnAgent::Codex
]
);
assert_eq!(after, Some(bitrouter::onboarding::AfterAction::Exit));
assert_eq!(model.as_deref(), Some("openai/gpt-5"));
}
_ => panic!("expected Init"),
}
}
#[test]
fn init_onboarding_rejects_removed_optimization_flags() {
use clap::Parser;
for removed_args in [
&["--optimize"][..],
&["--optimize-workflow-command", "./run-eval"][..],
&["--optimize-workflow-arg", "--smoke"][..],
&["--optimize-workflow-input", ".venv"][..],
&["--optimize-success", "tests pass"][..],
&["--optimize-strong", "openai:gpt-5"][..],
&["--optimize-strong-effort", "high"][..],
&["--optimize-economy", "openai:gpt-5-mini"][..],
&["--optimize-economy-effort", "low"][..],
&["--optimize-normalized-price", "openai:gpt-5=100"][..],
&["--optimize-preference", "balanced"][..],
] {
let mut command = vec!["bitrouter", "init"];
command.extend_from_slice(removed_args);
assert!(
Cli::try_parse_from(command).is_err(),
"removed init arguments {removed_args:?} still parse"
);
}
}
#[test]
fn providers_login_api_key_flag_parses() {
use clap::Parser;
let cli = Cli::try_parse_from([
"bitrouter",
"providers",
"login",
"openai",
"--api-key",
"sk-abc",
])
.expect("parse");
match cli.command {
Some(Command::Providers {
action:
ProviderAction::Login {
provider,
api_key,
key_stdin,
..
},
}) => {
assert_eq!(provider, "openai");
assert_eq!(api_key.as_deref(), Some("sk-abc"));
assert!(!key_stdin);
}
_ => panic!("expected provider login"),
}
}
#[test]
fn providers_login_api_key_conflicts_with_oauth_flags() {
use clap::Parser;
let parsed = Cli::try_parse_from([
"bitrouter",
"providers",
"login",
"openai",
"--api-key",
"sk-abc",
"--no-browser",
]);
let err = match parsed {
Ok(_) => panic!("api-key + oauth flags must conflict"),
Err(err) => err,
};
assert_eq!(err.kind(), clap::error::ErrorKind::ArgumentConflict);
}
#[test]
fn provider_login_import_existing_flags_parse() {
use clap::Parser;
let cli = Cli::try_parse_from([
"bitrouter",
"providers",
"login",
"openai-codex",
"--import-existing",
"--no-browser",
"--label",
"work",
])
.expect("parse");
match cli.command {
Some(Command::Providers {
action:
ProviderAction::Login {
provider,
label,
import_existing,
no_browser,
..
},
}) => {
assert_eq!(provider, "openai-codex");
assert_eq!(label, "work");
assert!(import_existing);
assert!(no_browser);
}
_ => panic!("expected provider login"),
}
}
#[test]
fn routing_policy_init_and_evolve_flags_parse() {
use clap::Parser;
let init = Cli::try_parse_from([
"bitrouter",
"policy",
"init",
"terminal-bench",
"--router",
"coding",
"--strong",
"openai-codex:gpt-5.6-sol",
"--strong-effort",
"high",
"--economy",
"openai-codex:gpt-5.6-sol",
"--economy-effort",
"low",
"--config",
"team/bitrouter.yaml",
])
.expect("parse init");
match init.command {
Some(Command::Policy {
action:
PolicyAction::Init {
name,
router,
preset,
strong,
strong_effort,
economy,
economy_effort,
config,
},
}) => {
assert_eq!(name, "terminal-bench");
assert_eq!(router.as_deref(), Some("coding"));
assert_eq!(preset, None);
assert_eq!(strong.as_deref(), Some("openai-codex:gpt-5.6-sol"));
assert_eq!(
strong_effort,
Some(bitrouter_sdk::language_model::types::ReasoningEffort::High)
);
assert_eq!(economy, "openai-codex:gpt-5.6-sol");
assert_eq!(
economy_effort,
Some(bitrouter_sdk::language_model::types::ReasoningEffort::Low)
);
assert_eq!(config, Some(PathBuf::from("team/bitrouter.yaml")));
}
_ => panic!("expected policy init"),
}
let default_coding = Cli::try_parse_from([
"bitrouter",
"policy",
"init",
"coding",
"--strong",
"openai-codex:gpt-5.6-sol",
"--economy",
"openai-codex:gpt-5.6-luna",
]);
assert!(matches!(
default_coding,
Ok(Cli {
command: Some(Command::Policy {
action: PolicyAction::Init {
router: None,
preset: None,
..
}
}),
..
})
));
assert!(
Cli::try_parse_from([
"bitrouter",
"policy",
"init",
"coding",
"--router",
"coding",
"--preset",
"coding",
"--economy",
"openai-codex:gpt-5.6-luna",
])
.is_err()
);
assert!(
Cli::try_parse_from([
"bitrouter",
"policy",
"init",
"coding",
"--strong",
"openai-codex:gpt-5.6-sol",
])
.is_err()
);
assert!(
Cli::try_parse_from([
"bitrouter",
"policy",
"init",
"auto",
"--preset",
"auto",
"--strong",
"openai-codex:gpt-5.6-sol",
"--economy",
"openai-codex:gpt-5.6-luna",
])
.is_ok()
);
let evolve = Cli::try_parse_from(["bitrouter", "policy", "evolve", "--apply"])
.expect("parse evolve");
assert!(matches!(
evolve.command,
Some(Command::Policy {
action: PolicyAction::Evolve { apply: true, .. }
})
));
let export = Cli::try_parse_from([
"bitrouter",
"policy",
"evolve",
"--output",
"candidate.yaml",
])
.expect("parse candidate export");
assert!(matches!(
export.command,
Some(Command::Policy {
action: PolicyAction::Evolve {
apply: false,
output: Some(path),
..
}
}) if path == Path::new("candidate.yaml")
));
assert!(Cli::try_parse_from(["bitrouter", "policy", "evolve", "--freeze"]).is_err());
assert!(Cli::try_parse_from(["bitrouter", "policy", "lock"]).is_err());
assert!(Cli::try_parse_from(["bitrouter", "policy", "unlock"]).is_err());
assert!(
Cli::try_parse_from([
"bitrouter",
"policy",
"evolve",
"--apply",
"--output",
"candidate.yaml",
])
.is_err()
);
}
#[test]
fn policy_compile_diff_and_rollback_commands_parse() {
use clap::Parser;
assert!(
Cli::try_parse_from([
"bitrouter",
"policy",
"compile",
"--output",
"candidate.yaml"
])
.is_ok()
);
assert!(
Cli::try_parse_from(["bitrouter", "policy", "diff", "active.yaml", "next.yaml"])
.is_ok()
);
assert!(
Cli::try_parse_from([
"bitrouter",
"policy",
"publish",
"candidate.yaml",
"--config",
"bitrouter.yaml",
])
.is_ok()
);
assert!(Cli::try_parse_from(["bitrouter", "policy", "rollback", "sha256:abc"]).is_ok());
}
#[test]
fn optimize_accepts_only_history_controller_commands() -> anyhow::Result<()> {
use clap::Parser;
assert!(
Cli::try_parse_from([
"bitrouter",
"optimize",
"run",
"--policy",
"auto",
"--candidate-tier",
"economy",
])
.is_ok()
);
assert!(
Cli::try_parse_from(["bitrouter", "optimize", "status", "--policy", "auto",]).is_ok()
);
for removed in ["setup", "resolve", "review", "publish", "rollback"] {
assert!(
Cli::try_parse_from(["bitrouter", "optimize", removed]).is_err(),
"removed optimize command {removed} still parses"
);
}
for removed_args in [
&["run", "--workflow-command", "./run-eval"][..],
&["run", "--workflow-arg", "--smoke"][..],
&["run", "--workflow-input", ".venv"][..],
&["run", "--evaluator-agent", "codex-acp"][..],
&["run", "--evaluator-model", "gpt-5.6"][..],
&["run", "--evaluator-via-cloud"][..],
&["run", "--enable-adaptive"][..],
] {
let mut command = vec!["bitrouter", "optimize"];
command.extend_from_slice(removed_args);
assert!(
Cli::try_parse_from(command).is_err(),
"removed optimize arguments {removed_args:?} still parse"
);
}
let run = Cli::try_parse_from(["bitrouter", "optimize", "run"])?;
assert!(matches!(
run.command,
Some(Command::Optimize {
action: OptimizeAction::Run {
policy,
candidate_tier: None,
exploration_ppm: 100_000,
minimum_tasks: 3,
maximum_tasks: 20,
minimum_pass_rate_ppm: 900_000,
evaluator_config_digest: None,
config,
socket: None,
}
}) if policy == "auto"
&& config == Path::new("bitrouter.yaml")
));
Ok(())
}
#[test]
fn optimize_status_without_active_exploration_is_idle() {
let report = optimization_status_report(
bitrouter::optimization::controller::OptimizationStatus {
policy: "auto".into(),
policy_mode: bitrouter_sdk::config::PolicyRuntimeMode::Adaptive,
active_policy_digest:
"sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa".into(),
parent_policy_digest: None,
eval_snapshot_root: None,
observed_subject_digest: None,
active_experiment: None,
},
);
assert_eq!(report.action, "optimize.status");
assert_eq!(report.decision, "idle");
}
#[test]
fn optimize_status_with_active_exploration_is_exploring() {
let report = optimization_status_report(
bitrouter::optimization::controller::OptimizationStatus {
policy: "auto".into(),
policy_mode: bitrouter_sdk::config::PolicyRuntimeMode::Adaptive,
active_policy_digest:
"sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa".into(),
parent_policy_digest: None,
eval_snapshot_root: None,
observed_subject_digest: None,
active_experiment: Some(bitrouter::optimization::exploration::RouteExploration {
experiment_id: "experiment-1".into(),
target_request_key: "agent_route/v1|unknown|implement|normal".into(),
champion_tier: "strong".into(),
challenger_tier: "economy".into(),
challenger_exposure_ppm: 100_000,
gate: bitrouter::optimization::exploration::OptimizationGate {
minimum_tasks_per_arm: 3,
maximum_challenger_tasks: 20,
minimum_pass_rate_ppm: 900_000,
evaluator_config_digest: None,
},
}),
},
);
assert_eq!(report.action, "optimize.status");
assert_eq!(report.decision, "exploring");
}
#[tokio::test]
async fn optimize_explore_json_reports_the_explicit_evaluator_pin() -> anyhow::Result<()> {
use std::collections::{BTreeMap, BTreeSet};
use bitrouter::eval::EvalService;
use bitrouter::eval::admission::SubmissionPrincipal;
use bitrouter::eval::store::EvalStore;
use bitrouter::eval::types::{
EVAL_SCHEMA_VERSION, EvalDecisionRef, EvalScope, EvalSubject, EvalVerdict,
EvaluationResult, EvaluatorIdentity, EvaluatorKind, EvidenceItem, evidence_digest,
};
const PIN: &str = "sha256:ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff";
let directory = tempfile::tempdir()?;
let config_path = directory.path().join("bitrouter.yaml");
let database_path = directory.path().join("eval.db");
let database_url = format!("sqlite://{}?mode=rwc", database_path.display());
std::fs::write(
&config_path,
"database:\n url: sqlite://./eval.db\npolicy:\n mode: frozen\n path: policy-lock.yaml\npresets:\n auto:\n model: strong-model\n",
)?;
let database = bitrouter::db::connect(&database_url).await?;
bitrouter::db::run_migrations(&database).await?;
drop(database);
let initialized = bitrouter::policy_lock::initialize_files(
&config_path,
"auto",
"auto",
Some("strong-model"),
"economy-model",
)
.await?;
let store = EvalStore::new(bitrouter::db::connect(&database_url).await?);
let service = EvalService::new(store.clone(), bitrouter_sdk::config::EvalConfig::default());
for (scope, subject_id, cost) in [
(EvalScope::Request, "request-report-pin", Some("900")),
(EvalScope::Task, "task-report-pin", None),
] {
let evidence = cost.map_or_else(Vec::new, |cost| {
vec![EvidenceItem {
evidence_id: format!("evidence-{subject_id}"),
kind: "request.outcome".into(),
digest: PIN.into(),
redacted: true,
attributes: BTreeMap::from([("cost_micro_usd".into(), cost.into())]),
}]
});
let digest = evidence_digest(&evidence)?;
let eval_id = format!("eval-{subject_id}");
let subject = EvalSubject {
schema_version: EVAL_SCHEMA_VERSION,
eval_id: eval_id.clone(),
scope,
subject_id: subject_id.into(),
policy_digest: initialized.digest.clone(),
preset: Some("auto".into()),
cohort: None,
holdout: false,
decisions: vec![EvalDecisionRef {
decision_id: format!("decision-{subject_id}"),
policy: "auto".into(),
route_projection: "agent_route/v1|unknown|implement|normal".into(),
request_key: "agent_route/v1|unknown|implement|normal".into(),
selected_tier: "strong".into(),
selected_effort: None,
baseline_tier: None,
baseline_effort: None,
policy_digest: initialized.digest.clone(),
experiment: None,
route_measurement: None,
}],
requested_dimensions: BTreeSet::new(),
evidence,
evidence_digest: digest.clone(),
observed_at: "2026-08-17T00:00:00Z".into(),
};
store.insert_subject(&subject).await?;
service
.submit(
EvaluationResult {
schema_version: EVAL_SCHEMA_VERSION,
eval_id,
evidence_digest: digest,
evaluator: EvaluatorIdentity {
authority_id: "local".into(),
evaluator_id: "history".into(),
kind: EvaluatorKind::TaskNative,
version: "1".into(),
config_digest: PIN.into(),
},
verdict: if scope == EvalScope::Request {
EvalVerdict::Inconclusive
} else {
EvalVerdict::Pass
},
metrics: BTreeMap::new(),
hard_violations: Vec::new(),
confidence_ppm: None,
evidence_refs: Vec::new(),
decision_credit: BTreeMap::new(),
idempotency_key: format!("result-{subject_id}"),
submitted_at: "2026-08-17T00:00:01Z".into(),
},
SubmissionPrincipal::LocalOperator,
)
.await?;
}
let prepared = bitrouter::optimization::controller::prepare_files(
&config_path,
bitrouter::optimization::controller::OptimizationOptions {
evaluator_config_digest: Some(PIN.into()),
..Default::default()
},
)
.await?;
let report = optimization_run_report("auto", &prepared);
let json = serde_json::to_value(report)?;
assert_eq!(json["decision"], "explore");
assert_eq!(json["evaluator_config_digest"], PIN);
Ok(())
}
struct RecoveryFixture {
_directory: tempfile::TempDir,
config_path: PathBuf,
policy_path: PathBuf,
config_before: String,
config_after: String,
parent_digest: String,
child_document: bitrouter::policy_lock::PolicyLock,
publication: bitrouter::optimization::controller::OptimizationPublication,
}
fn recovery_fixture() -> anyhow::Result<RecoveryFixture> {
let directory = tempfile::tempdir()?;
let config_path = directory.path().join("bitrouter.yaml");
let policy_path = directory.path().join("policy-lock.yaml");
let config_before =
"# keep operator bytes\npolicy:\n mode: frozen\n path: policy-lock.yaml\n"
.to_string();
let config_after = bitrouter::policy_lock::edit_config_mode(
&config_before,
config::PolicyRuntimeMode::Adaptive,
)?;
std::fs::write(&config_path, &config_after)?;
let parent = bitrouter::policy_lock::PolicyLock::default();
let parent_digest =
bitrouter::policy_lock::write_atomic_unlocked(&policy_path, None, &parent)?;
let mut child_document = parent;
let artifact = child_document
.artifact
.as_mut()
.ok_or_else(|| anyhow::anyhow!("default compiled policy has no artifact"))?;
artifact.parent_digest = Some(parent_digest.clone());
artifact.source_snapshot_time_unix_ms = 1;
let history = bitrouter::policy_lock::default_history_dir(&policy_path);
let record = bitrouter::policy_lock::publish_candidate_unlocked(
&policy_path,
&parent_digest,
&child_document,
&history,
)?;
let update = bitrouter::policy_lock::PolicyFileUpdate {
path: policy_path.clone(),
digest: record.child_digest.clone(),
document: child_document.clone(),
changes: Vec::new(),
conflicts: Vec::new(),
};
let publication = bitrouter::optimization::controller::OptimizationPublication {
action: bitrouter::optimization::controller::ControllerAction::Explore,
parent_policy_digest: parent_digest.clone(),
active_policy_digest: record.child_digest,
eval_snapshot_root:
"sha256:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855".into(),
published: true,
config_activated: true,
config_path: config_path.clone(),
config_before: config_before.clone(),
config_after: config_after.clone(),
update: Some(update),
};
Ok(RecoveryFixture {
_directory: directory,
config_path,
policy_path,
config_before,
config_after,
parent_digest,
child_document,
publication,
})
}
#[tokio::test]
async fn optimizer_reload_recovery_restores_an_adaptive_publication_with_identical_config()
-> anyhow::Result<()> {
use std::collections::{BTreeMap, BTreeSet};
use bitrouter::eval::EvalService;
use bitrouter::eval::admission::SubmissionPrincipal;
use bitrouter::eval::store::EvalStore;
use bitrouter::eval::types::{
EVAL_SCHEMA_VERSION, EvalDecisionRef, EvalScope, EvalSubject, EvalVerdict,
EvaluationResult, EvaluatorIdentity, EvaluatorKind, EvidenceItem, evidence_digest,
};
const EVIDENCE_DIGEST: &str =
"sha256:ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff";
let directory = tempfile::tempdir()?;
let config_path = directory.path().join("bitrouter.yaml");
let database_path = directory.path().join("eval.db");
let database_url = format!("sqlite://{}?mode=rwc", database_path.display());
std::fs::write(
&config_path,
"database:\n url: sqlite://./eval.db\npolicy:\n mode: frozen\n path: policy-lock.yaml\npresets:\n auto:\n model: strong-model\n",
)?;
let database = bitrouter::db::connect(&database_url).await?;
bitrouter::db::run_migrations(&database).await?;
drop(database);
let initialized = bitrouter::policy_lock::initialize_files(
&config_path,
"auto",
"auto",
Some("strong-model"),
"economy-model",
)
.await?;
let frozen_config = std::fs::read_to_string(&config_path)?;
let adaptive_config = bitrouter::policy_lock::edit_config_mode(
&frozen_config,
config::PolicyRuntimeMode::Adaptive,
)?;
std::fs::write(&config_path, &adaptive_config)?;
let store = EvalStore::new(bitrouter::db::connect(&database_url).await?);
let service = EvalService::new(store.clone(), bitrouter_sdk::config::EvalConfig::default());
for (scope, subject_id, cost) in [
(EvalScope::Request, "request-adaptive-recovery", Some("900")),
(EvalScope::Task, "task-adaptive-recovery", None),
] {
let evidence = cost.map_or_else(Vec::new, |cost| {
vec![EvidenceItem {
evidence_id: format!("evidence-{subject_id}"),
kind: "request.outcome".into(),
digest: EVIDENCE_DIGEST.into(),
redacted: true,
attributes: BTreeMap::from([("cost_micro_usd".into(), cost.into())]),
}]
});
let digest = evidence_digest(&evidence)?;
let eval_id = format!("eval-{subject_id}");
let subject = EvalSubject {
schema_version: EVAL_SCHEMA_VERSION,
eval_id: eval_id.clone(),
scope,
subject_id: subject_id.into(),
policy_digest: initialized.digest.clone(),
preset: Some("auto".into()),
cohort: None,
holdout: false,
decisions: vec![EvalDecisionRef {
decision_id: format!("decision-{subject_id}"),
policy: "auto".into(),
route_projection: "agent_route/v1|unknown|implement|normal".into(),
request_key: "agent_route/v1|unknown|implement|normal".into(),
selected_tier: "strong".into(),
selected_effort: None,
baseline_tier: None,
baseline_effort: None,
policy_digest: initialized.digest.clone(),
experiment: None,
route_measurement: None,
}],
requested_dimensions: BTreeSet::new(),
evidence,
evidence_digest: digest.clone(),
observed_at: "2026-08-17T00:00:00Z".into(),
};
store.insert_subject(&subject).await?;
service
.submit(
EvaluationResult {
schema_version: EVAL_SCHEMA_VERSION,
eval_id,
evidence_digest: digest,
evaluator: EvaluatorIdentity {
authority_id: "local".into(),
evaluator_id: "history".into(),
kind: EvaluatorKind::TaskNative,
version: "1".into(),
config_digest: EVIDENCE_DIGEST.into(),
},
verdict: if scope == EvalScope::Request {
EvalVerdict::Inconclusive
} else {
EvalVerdict::Pass
},
metrics: BTreeMap::new(),
hard_violations: Vec::new(),
confidence_ppm: None,
evidence_refs: Vec::new(),
decision_credit: BTreeMap::new(),
idempotency_key: format!("result-{subject_id}"),
submitted_at: "2026-08-17T00:00:01Z".into(),
},
SubmissionPrincipal::LocalOperator,
)
.await?;
}
let prepared = bitrouter::optimization::controller::prepare_files(
&config_path,
bitrouter::optimization::controller::OptimizationOptions::default(),
)
.await?;
let publication = bitrouter::optimization::controller::publish_prepared(prepared).await?;
let policy_path = publication
.update
.as_ref()
.map(|update| update.path.clone())
.context("adaptive optimizer publication has no policy update")?;
assert_eq!(publication.config_before, adaptive_config);
assert_eq!(publication.config_after, adaptive_config);
assert!(!publication.config_activated);
let attempts = Arc::new(std::sync::atomic::AtomicUsize::new(0));
let live = Arc::new(std::sync::Mutex::new((
config::PolicyRuntimeMode::Adaptive,
initialized.digest.clone(),
)));
let reload = {
let attempts = attempts.clone();
let live = live.clone();
let config_path = config_path.clone();
let policy_path = policy_path.clone();
move || {
let attempts = attempts.clone();
let live = live.clone();
let config_path = config_path.clone();
let policy_path = policy_path.clone();
async move {
if attempts.fetch_add(1, std::sync::atomic::Ordering::SeqCst) == 0 {
anyhow::bail!("injected adaptive daemon rejection of child policy")
}
let raw = std::fs::read_to_string(config_path)?;
let mode = bitrouter_sdk::config::parse(&raw)?.policy.mode;
let digest = bitrouter::policy_lock::load(&policy_path).await?.digest;
*live
.lock()
.map_err(|_| anyhow::anyhow!("modeled daemon state lock poisoned"))? =
(mode, digest);
Ok(())
}
}
};
let error =
reload_optimization_publication_or_restore_with(&publication, reload, || Ok(()))
.await
.err()
.context("injected adaptive daemon rejection unexpectedly succeeded")?;
let detail = format!("{error:#}");
assert!(detail.contains("injected adaptive daemon rejection of child policy"));
assert!(detail.contains("reconciled and reloaded the parent"));
assert_eq!(attempts.load(std::sync::atomic::Ordering::SeqCst), 2);
assert_eq!(std::fs::read_to_string(&config_path)?, adaptive_config);
assert_eq!(
bitrouter_sdk::config::parse(&std::fs::read_to_string(&config_path)?)?
.policy
.mode,
config::PolicyRuntimeMode::Adaptive
);
assert_eq!(
bitrouter::policy_lock::load(&policy_path).await?.digest,
initialized.digest
);
assert_eq!(
*live
.lock()
.map_err(|_| anyhow::anyhow!("modeled daemon state lock poisoned"))?,
(config::PolicyRuntimeMode::Adaptive, initialized.digest)
);
Ok(())
}
#[tokio::test]
async fn optimizer_reload_recovery_restores_one_disk_and_live_parent_pair() -> anyhow::Result<()>
{
let fixture = recovery_fixture()?;
let attempts = Arc::new(std::sync::atomic::AtomicUsize::new(0));
let live = Arc::new(std::sync::Mutex::new((
config::PolicyRuntimeMode::Frozen,
fixture.parent_digest.clone(),
)));
let reload = {
let attempts = attempts.clone();
let live = live.clone();
let config_path = fixture.config_path.clone();
let policy_path = fixture.policy_path.clone();
move || {
let attempts = attempts.clone();
let live = live.clone();
let config_path = config_path.clone();
let policy_path = policy_path.clone();
async move {
if attempts.fetch_add(1, std::sync::atomic::Ordering::SeqCst) == 0 {
anyhow::bail!("injected daemon rejection of child policy")
}
let raw = std::fs::read_to_string(config_path)?;
let mode = bitrouter_sdk::config::parse(&raw)?.policy.mode;
let digest = bitrouter::policy_lock::load(&policy_path).await?.digest;
*live
.lock()
.map_err(|_| anyhow::anyhow!("modeled daemon state lock poisoned"))? =
(mode, digest);
Ok(())
}
}
};
let error = reload_optimization_publication_or_restore_with(
&fixture.publication,
reload,
|| Ok(()),
)
.await
.err()
.context("injected daemon rejection unexpectedly succeeded")?;
assert!(format!("{error:#}").contains("injected daemon rejection"));
assert_eq!(attempts.load(std::sync::atomic::Ordering::SeqCst), 2);
assert_eq!(
std::fs::read_to_string(&fixture.config_path)?,
fixture.config_before
);
assert_eq!(
bitrouter::policy_lock::load(&fixture.policy_path)
.await?
.digest,
fixture.parent_digest
);
assert_eq!(
*live
.lock()
.map_err(|_| anyhow::anyhow!("modeled daemon state lock poisoned"))?,
(config::PolicyRuntimeMode::Frozen, fixture.parent_digest)
);
Ok(())
}
#[tokio::test]
async fn optimizer_reload_recovery_reconciles_a_post_commit_policy_error() -> anyhow::Result<()>
{
let fixture = recovery_fixture()?;
let attempts = Arc::new(std::sync::atomic::AtomicUsize::new(0));
let events = Arc::new(std::sync::Mutex::new(Vec::<String>::new()));
let live = Arc::new(std::sync::Mutex::new((
config::PolicyRuntimeMode::Frozen,
fixture.parent_digest.clone(),
)));
let reload = {
let attempts = attempts.clone();
let events = events.clone();
let live = live.clone();
let config_path = fixture.config_path.clone();
let policy_path = fixture.policy_path.clone();
move || {
let attempts = attempts.clone();
let events = events.clone();
let live = live.clone();
let config_path = config_path.clone();
let policy_path = policy_path.clone();
async move {
if attempts.fetch_add(1, std::sync::atomic::Ordering::SeqCst) == 0 {
events
.lock()
.map_err(|_| anyhow::anyhow!("recovery event lock poisoned"))?
.push("reload-child-rejected".into());
anyhow::bail!("injected daemon rejection of child policy")
}
let raw = std::fs::read_to_string(config_path)?;
let mode = bitrouter_sdk::config::parse(&raw)?.policy.mode;
let digest = bitrouter::policy_lock::load(&policy_path).await?.digest;
events
.lock()
.map_err(|_| anyhow::anyhow!("recovery event lock poisoned"))?
.push("reload-parent".into());
*live
.lock()
.map_err(|_| anyhow::anyhow!("modeled daemon state lock poisoned"))? =
(mode, digest);
Ok(())
}
}
};
let policy_events = events.clone();
let config_events = events.clone();
let error = reload_optimization_publication_or_restore_with_mutations(
&fixture.publication,
reload,
|| Ok(()),
move |path, expected, target, history| {
bitrouter::policy_lock::rollback_to_digest_unlocked(
path, expected, target, history,
)?;
policy_events
.lock()
.map_err(|_| anyhow::anyhow!("recovery event lock poisoned"))?
.push("policy-parent-post-commit-error".into());
anyhow::bail!("injected post-commit policy failure")
},
move |path, expected, updated| {
bitrouter::policy_lock::write_text_atomic_unlocked(path, expected, updated)?;
config_events
.lock()
.map_err(|_| anyhow::anyhow!("recovery event lock poisoned"))?
.push("config-parent".into());
Ok(())
},
)
.await
.err()
.context("post-commit policy failure unexpectedly succeeded")?;
let detail = format!("{error:#}");
assert!(detail.contains("injected daemon rejection of child policy"));
assert!(detail.contains("injected post-commit policy failure"));
assert_eq!(attempts.load(std::sync::atomic::Ordering::SeqCst), 2);
assert_eq!(
*events
.lock()
.map_err(|_| anyhow::anyhow!("recovery event lock poisoned"))?,
[
"reload-child-rejected",
"policy-parent-post-commit-error",
"config-parent",
"reload-parent",
]
);
assert_eq!(
std::fs::read_to_string(&fixture.config_path)?,
fixture.config_before
);
assert_eq!(
bitrouter::policy_lock::load(&fixture.policy_path)
.await?
.digest,
fixture.parent_digest
);
assert_eq!(
*live
.lock()
.map_err(|_| anyhow::anyhow!("modeled daemon state lock poisoned"))?,
(config::PolicyRuntimeMode::Frozen, fixture.parent_digest)
);
Ok(())
}
#[tokio::test]
async fn optimizer_reload_recovery_reconciles_a_post_rename_config_error() -> anyhow::Result<()>
{
let fixture = recovery_fixture()?;
let attempts = Arc::new(std::sync::atomic::AtomicUsize::new(0));
let events = Arc::new(std::sync::Mutex::new(Vec::<String>::new()));
let live = Arc::new(std::sync::Mutex::new((
config::PolicyRuntimeMode::Frozen,
fixture.parent_digest.clone(),
)));
let reload = {
let attempts = attempts.clone();
let events = events.clone();
let live = live.clone();
let config_path = fixture.config_path.clone();
let policy_path = fixture.policy_path.clone();
move || {
let attempts = attempts.clone();
let events = events.clone();
let live = live.clone();
let config_path = config_path.clone();
let policy_path = policy_path.clone();
async move {
if attempts.fetch_add(1, std::sync::atomic::Ordering::SeqCst) == 0 {
events
.lock()
.map_err(|_| anyhow::anyhow!("recovery event lock poisoned"))?
.push("reload-child-rejected".into());
anyhow::bail!("injected daemon rejection of child policy")
}
let raw = std::fs::read_to_string(config_path)?;
let mode = bitrouter_sdk::config::parse(&raw)?.policy.mode;
let digest = bitrouter::policy_lock::load(&policy_path).await?.digest;
events
.lock()
.map_err(|_| anyhow::anyhow!("recovery event lock poisoned"))?
.push("reload-parent".into());
*live
.lock()
.map_err(|_| anyhow::anyhow!("modeled daemon state lock poisoned"))? =
(mode, digest);
Ok(())
}
}
};
let policy_events = events.clone();
let config_events = events.clone();
let error = reload_optimization_publication_or_restore_with_mutations(
&fixture.publication,
reload,
|| Ok(()),
move |path, expected, target, history| {
bitrouter::policy_lock::rollback_to_digest_unlocked(
path, expected, target, history,
)?;
policy_events
.lock()
.map_err(|_| anyhow::anyhow!("recovery event lock poisoned"))?
.push("policy-parent".into());
Ok(())
},
move |path, expected, updated| {
bitrouter::policy_lock::write_text_atomic_unlocked(path, expected, updated)?;
config_events
.lock()
.map_err(|_| anyhow::anyhow!("recovery event lock poisoned"))?
.push("config-parent-post-rename-error".into());
anyhow::bail!("injected post-rename config failure")
},
)
.await
.err()
.context("post-rename config failure unexpectedly succeeded")?;
let detail = format!("{error:#}");
assert!(detail.contains("injected daemon rejection of child policy"));
assert!(detail.contains("injected post-rename config failure"));
assert_eq!(attempts.load(std::sync::atomic::Ordering::SeqCst), 2);
assert_eq!(
*events
.lock()
.map_err(|_| anyhow::anyhow!("recovery event lock poisoned"))?,
[
"reload-child-rejected",
"policy-parent",
"config-parent-post-rename-error",
"reload-parent",
]
);
assert_eq!(
std::fs::read_to_string(&fixture.config_path)?,
fixture.config_before
);
assert_eq!(
bitrouter::policy_lock::load(&fixture.policy_path)
.await?
.digest,
fixture.parent_digest
);
assert_eq!(
*live
.lock()
.map_err(|_| anyhow::anyhow!("modeled daemon state lock poisoned"))?,
(config::PolicyRuntimeMode::Frozen, fixture.parent_digest)
);
Ok(())
}
#[tokio::test]
async fn optimizer_reload_recovery_never_overwrites_a_concurrent_policy_winner()
-> anyhow::Result<()> {
let fixture = recovery_fixture()?;
let attempts = Arc::new(std::sync::atomic::AtomicUsize::new(0));
let winner_digest = Arc::new(std::sync::Mutex::new(None::<String>));
let live = Arc::new(std::sync::Mutex::new((
config::PolicyRuntimeMode::Frozen,
fixture.parent_digest.clone(),
)));
let reload = {
let attempts = attempts.clone();
move || {
let attempts = attempts.clone();
async move {
attempts.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
anyhow::bail!("injected daemon rejection before concurrent publication")
}
}
};
let before_recovery = {
let policy_path = fixture.policy_path.clone();
let child_digest = fixture.publication.active_policy_digest.clone();
let mut winner = fixture.child_document.clone();
let winner_digest = winner_digest.clone();
let live = live.clone();
move || {
let artifact = winner
.artifact
.as_mut()
.ok_or_else(|| anyhow::anyhow!("winner artifact is missing"))?;
artifact.parent_digest = Some(child_digest.clone());
artifact.source_snapshot_time_unix_ms = 2;
let record = bitrouter::policy_lock::publish_candidate(
&policy_path,
&child_digest,
&winner,
&bitrouter::policy_lock::default_history_dir(&policy_path),
)?;
*winner_digest
.lock()
.map_err(|_| anyhow::anyhow!("winner digest lock poisoned"))? =
Some(record.child_digest.clone());
*live
.lock()
.map_err(|_| anyhow::anyhow!("modeled daemon state lock poisoned"))? =
(config::PolicyRuntimeMode::Adaptive, record.child_digest);
Ok(())
}
};
let error = reload_optimization_publication_or_restore_with(
&fixture.publication,
reload,
before_recovery,
)
.await
.err()
.context("stale recovery unexpectedly succeeded")?;
let winner_digest = winner_digest
.lock()
.map_err(|_| anyhow::anyhow!("winner digest lock poisoned"))?
.clone()
.context("concurrent winner was not published")?;
assert!(error.to_string().contains("recovery is stale"));
assert_eq!(attempts.load(std::sync::atomic::Ordering::SeqCst), 1);
assert_eq!(
std::fs::read_to_string(&fixture.config_path)?,
fixture.config_after
);
assert_eq!(
bitrouter::policy_lock::load(&fixture.policy_path)
.await?
.digest,
winner_digest
);
assert_eq!(
*live
.lock()
.map_err(|_| anyhow::anyhow!("modeled daemon state lock poisoned"))?,
(config::PolicyRuntimeMode::Adaptive, winner_digest)
);
Ok(())
}
#[tokio::test]
async fn optimizer_reload_recovery_reports_rejection_and_recovery_failures()
-> anyhow::Result<()> {
let fixture = recovery_fixture()?;
let config_path = fixture.config_path.clone();
let error = reload_optimization_publication_or_restore_with(
&fixture.publication,
|| async { anyhow::bail!("injected daemon rejection for combined error") },
move || {
std::fs::remove_file(&config_path)?;
Ok(())
},
)
.await
.err()
.context("recovery with a missing config unexpectedly succeeded")?;
let detail = format!("{error:#}");
assert!(detail.contains("injected daemon rejection for combined error"));
assert!(detail.contains("reading"));
assert!(detail.contains("bitrouter.yaml"));
Ok(())
}
#[test]
fn workflow_state_reliability_report_flags_parse() {
use clap::Parser;
let cli = Cli::try_parse_from([
"bitrouter",
"workflow-state",
"reliability-report",
"--database-url",
"sqlite:///tmp/bitrouter.db",
"--config",
"/tmp/bitrouter.yaml",
"--output",
"/tmp/reliability.json",
])
.expect("parse reliability report");
match cli.command {
Some(Command::WorkflowState {
action:
WorkflowStateAction::ReliabilityReport {
database_url,
config,
output,
},
}) => {
assert_eq!(database_url, "sqlite:///tmp/bitrouter.db");
assert_eq!(config, PathBuf::from("/tmp/bitrouter.yaml"));
assert_eq!(output, PathBuf::from("/tmp/reliability.json"));
}
_ => panic!("expected reliability report"),
}
}
#[test]
fn cloud_api_owns_header_short_flag_in_full_command_tree() {
let cli = Cli::try_parse_from([
"bitrouter",
"cloud",
"api",
"/v1/models",
"-H",
"X-Test: value",
])
.unwrap();
match cli.command {
Some(Command::Cloud {
action: bitrouter::cloud::cli::CloudAction::Api(args),
}) => assert_eq!(args.headers, ["X-Test: value"]),
_ => panic!("expected cloud API command"),
}
let human = Cli::try_parse_from(["bitrouter", "-H", "cloud", "whoami"]).unwrap();
assert!(human.human_short);
}
#[tokio::test]
async fn config_validate_rejects_a_missing_bound_policy() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("bitrouter.yaml");
tokio::fs::write(
&path,
r#"presets:
coding:
model: anthropic/claude-opus-4.8
policy: missing
"#,
)
.await
.unwrap();
let source = bitrouter::paths::ConfigSource::File(path);
let report = validate_config(&source).await.unwrap();
assert!(!report.valid);
assert!(report.errors[0].contains("policy lock"));
}
}