use std::path::{Path, PathBuf};
#[cfg(test)]
use std::sync::Arc;
mod tracing_filter;
use tracing_filter::resolve_env_filter;
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, process_is_alive};
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;
#[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>,
#[arg(long)]
thread_id: Option<String>,
#[arg(long)]
check: Option<String>,
#[arg(long, conflicts_with_all = ["check", "thread_id"])]
read_only: bool,
#[arg(long)]
workspace: Option<PathBuf>,
#[arg(long, conflicts_with_all = ["agent", "thread_id"])]
model_policy: bool,
#[arg(long, conflicts_with_all = ["agent", "thread_id"], value_parser = clap::value_parser!(u32).range(1..))]
max_output_tokens: Option<u32>,
}
#[derive(Args)]
struct TaskRunArgs {
prompt: String,
#[arg(long)]
model: Option<String>,
#[arg(long)]
effort: Option<bitrouter_ai::types::ReasoningEffort>,
#[arg(long, value_parser = clap::value_parser!(u32).range(1..))]
max_output_tokens: Option<u32>,
#[arg(long)]
check: Option<String>,
#[arg(long, conflicts_with = "check")]
read_only: bool,
#[arg(long)]
workspace: Option<PathBuf>,
#[arg(short, long)]
config: Option<PathBuf>,
}
#[derive(clap::Args)]
struct ManagedTaskArgs {
#[command(flatten)]
task: TaskRunArgs,
#[arg(long)]
session: Option<String>,
}
#[derive(Subcommand)]
enum TaskAction {
Run(TaskRunArgs),
Managed(ManagedTaskArgs),
}
#[derive(Subcommand)]
enum Command {
Task {
#[command(subcommand)]
action: TaskAction,
},
Serve {
#[arg(short, long)]
config: Option<PathBuf>,
#[arg(long, hide = true)]
managed_child: bool,
},
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>,
},
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, hide = true)]
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, default_value = "auto")]
policy: String,
#[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 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_ai::types::ReasoningEffort>,
#[arg(long)]
economy: String,
#[arg(long)]
economy_effort: Option<bitrouter_ai::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,
..
}) | Some(Command::Task { .. })
);
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::Task { .. }
| 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::Task { action } => match action {
TaskAction::Run(args) => run_native_task(args, remote_context.is_some()).await,
TaskAction::Managed(args) => {
anyhow::ensure!(
remote_context.is_none(),
"managed native tasks run on the local host"
);
let task = args.task;
bitrouter::managed_task::run(bitrouter::managed_task::Options {
prompt: task.prompt,
model: task.model,
effort: task.effort,
check: task.check,
read_only: task.read_only,
workspace: task.workspace,
config: task.config,
session: args.session,
max_output_tokens: task.max_output_tokens,
})
.await
}
},
Command::Serve {
config,
managed_child,
} => {
let source = bitrouter::paths::resolve_config(config.as_deref())?;
bitrouter::host::serve_cli(&source, managed_child).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::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,
thread_id: None,
check: None,
read_only: false,
workspace: None,
model_policy: false,
max_output_tokens: 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 mut outcome = bitrouter::update::run(opts, &socket).await?;
if outcome.restart_needed {
let before_pid = match daemon::send_command(&socket, &DaemonCommand::Status).await {
Ok(DaemonResponse::Status { pid, .. }) => Some(pid),
_ => None,
};
let candidate = std::env::current_exe()?;
let attempt = tokio::process::Command::new(candidate)
.args(["--json", "agents", "sessions"])
.output()
.await;
let (completed, reason) = match attempt {
Ok(result) if result.status.success() => {
let hello =
daemon::send_command(&socket, &DaemonCommand::HandoffHello).await;
let matches_target = matches!(hello,
Ok(DaemonResponse::HandoffHello { ref version, protocol: 1, .. })
if Some(version.as_str()) == outcome.report.target_version.as_deref());
(
matches_target,
"replacement daemon version was not verified".to_string(),
)
}
Ok(result) => {
let message = serde_json::from_slice::<serde_json::Value>(&result.stdout)
.ok()
.and_then(|value| {
value["error"]["message"].as_str().map(str::to_string)
})
.unwrap_or_else(|| {
"the newly installed binary declined safe handoff".to_string()
});
(false, message)
}
Err(error) => (
false,
format!("could not launch the newly installed binary: {error}"),
),
};
if let Ok(DaemonResponse::Status {
pid,
listen,
daemon_version,
..
}) = daemon::send_command(&socket, &DaemonCommand::Status).await
{
outcome.report.daemon_pid = Some(pid);
outcome.report.daemon_listen = Some(listen);
outcome.report.daemon_version = daemon_version;
}
if let Ok(DaemonResponse::HandoffHello { instance_id, .. }) =
daemon::send_command(&socket, &DaemonCommand::HandoffHello).await
{
outcome.report.daemon_instance_id = Some(instance_id);
}
let completed = completed
&& outcome.report.daemon_pid.is_some()
&& outcome.report.daemon_instance_id.is_some();
outcome.report.daemon = Some(if !completed {
"deferred"
} else if before_pid.is_some()
&& outcome.report.daemon_pid.is_some()
&& before_pid != outcome.report.daemon_pid
{
"restarted"
} else {
"compatible"
});
if !completed {
outcome.report.daemon_reason = Some(reason);
}
}
output.emit(&outcome.report)?;
if outcome.report.exit_code() != 0 {
std::process::exit(outcome.report.exit_code());
}
Ok(())
}
}
}
async fn run_native_task(args: TaskRunArgs, remote: bool) -> Result<()> {
use bitrouter::agent_local::{Operation, ReplyResult};
use bitrouter_orchestrator::turn::TurnStatus;
anyhow::ensure!(
!remote,
"remote native tasks require an explicitly configured task API; no local fallback was attempted"
);
let source = bitrouter::paths::resolve_config(args.config.as_deref())?;
let cfg = bitrouter::paths::load_config(&source).await?;
let control_socket = daemon::socket_path_for(&source, &cfg);
let task_socket = bitrouter::agent_local::connect_or_start(&source, &control_socket).await?;
let workspace = match args.workspace {
Some(workspace) => workspace,
None => std::env::current_dir()?,
};
let client = bitrouter::agent_local::ThreadClient::connect(&task_socket).await?;
let model = args
.model
.or_else(|| bitrouter::policy_lock::native_default_model(&cfg).map(str::to_owned))
.context("no native model configured; pass --model or enable defaults")?;
let model_mode = bitrouter::policy_lock::native_model_mode(&cfg, &model)?;
let config = bitrouter_orchestrator::agent::AgentConfig::fixed(model, args.effort)
.with_model_mode(model_mode)
.with_output_reservation(args.max_output_tokens);
let config = if args.read_only {
config.read_only()
} else {
config
};
let (thread, accepted) = client
.create_and_start(workspace, config, args.check, args.prompt)
.await?;
emit_task_json(
serde_json::json!({ "type": "accepted", "thread_id": thread.thread_id, "turn_id": accepted.turn_id, "host": "local", "server_instance_id": thread.server_instance_id, "workspace": thread.workspace, "permission_profile": thread.permission_profile, "cursor": thread.cursor }),
)?;
let mut stream = client
.observe(&thread.thread_id, Some(thread.cursor))
.await?;
let mut view = match client
.request(Operation::ReadThread {
thread_id: thread.thread_id.clone(),
})
.await?
{
ReplyResult::View { view } => *view,
_ => anyhow::bail!("unexpected Thread view"),
};
let mut answered = None;
let mut cancelling = false;
loop {
let observation = tokio::select! {
signal = tokio::signal::ctrl_c(), if !cancelling => {
signal?;
client.request(Operation::CancelTurn { thread_id: thread.thread_id.clone(), turn_id: accepted.turn_id.clone(), idempotency_key: "headless-cancel".into() }).await?;
cancelling = true;
continue;
}
next = stream.next() => next?.ok_or_else(|| anyhow::anyhow!("task observation disconnected; task {} may still be running; no task was resubmitted", accepted.turn_id))?,
};
match observation {
bitrouter_orchestrator::thread::ThreadObservation::Snapshot {
view: fresh,
resynchronized,
catchup,
} => {
for event in catchup {
emit_task_json(serde_json::json!({"type":"event", "event":event}))?;
}
view = *fresh;
emit_task_json(
serde_json::json!({"type":"snapshot", "view":view, "resynchronized":resynchronized}),
)?;
}
bitrouter_orchestrator::thread::ThreadObservation::Event { event } => {
view.apply(&event);
emit_task_json(serde_json::json!({"type":"event", "event":event}))?;
}
bitrouter_orchestrator::thread::ThreadObservation::Live {
event,
after_cursor,
} => {
emit_task_json(
serde_json::json!({"type":"live", "event":event, "after_cursor":after_cursor}),
)?;
}
}
let historical = if view
.latest_turn
.as_ref()
.is_some_and(|turn| turn.turn_id != accepted.turn_id)
{
match client
.request(Operation::ReadTurn {
thread_id: thread.thread_id.clone(),
turn_id: accepted.turn_id.clone(),
})
.await?
{
ReplyResult::Turn { snapshot } => Some(snapshot),
_ => anyhow::bail!("unexpected targeted Turn reply"),
}
} else {
None
};
let Some(snapshot) = historical.as_deref().or(view.latest_turn.as_ref()) else {
continue;
};
if snapshot.status == TurnStatus::WaitingForInput
&& let Some(request_id) = snapshot.pending_input_id.as_ref()
&& answered.as_ref() != Some(request_id)
&& !cancelling
{
let result = client
.request(Operation::Input {
thread_id: thread.thread_id.clone(),
turn_id: snapshot.turn_id.clone(),
request_id: request_id.clone(),
approved: true,
idempotency_key: request_id.clone(),
})
.await;
if let Err(error) = result
&& !error
.downcast_ref::<bitrouter_orchestrator::service::ServiceError>()
.is_some_and(|error| {
error.code == bitrouter_orchestrator::service::ErrorCode::Conflict
})
{
return Err(error);
}
answered = Some(request_id.clone());
}
if snapshot.status == TurnStatus::RecoveryRequired {
emit_task_json(serde_json::json!({"type": "blocked", "snapshot": snapshot}))?;
anyhow::bail!("BRO execution is blocked; inspect recovery state before continuing");
}
if snapshot.status.terminal() {
emit_task_json(serde_json::json!({
"type": "terminal", "thread_id": thread.thread_id, "turn_id": snapshot.turn_id, "status": snapshot.status,
"server_instance_id": snapshot.server_instance_id,
"verification": snapshot.verification, "verification_evidence": snapshot.verification_evidence,
"final_answer": snapshot.final_answer, "detail": snapshot.detail, "unknown_effect": snapshot.unknown_effect,
"host": "local", "workspace": snapshot.workspace, "cursor": snapshot.cursor,
}))?;
drop(stream);
let _ = client
.request(Operation::UnloadThread {
thread_id: thread.thread_id.clone(),
})
.await;
anyhow::ensure!(
snapshot.status == TurnStatus::Completed,
"BRO task ended with {:?}",
snapshot.status
);
return Ok(());
}
}
}
fn emit_task_json(value: serde_json::Value) -> Result<()> {
use std::io::Write;
let mut stdout = std::io::stdout().lock();
serde_json::to_writer(&mut stdout, &value)?;
writeln!(stdout)?;
stdout.flush()?;
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::cloud::account::manager::CredentialManager::new(credentials_file)
.context("build settlement credential manager")?;
manager
.session()
.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 {
policy,
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 lock = bitrouter::policy_lock::load_for_config(&config_document, Some(&config))
.await?
.ok_or_else(|| {
anyhow::anyhow!("reliability report requires a named policy lock")
})?;
let policy = lock.document.policies.get(&policy).ok_or_else(|| {
anyhow::anyhow!("reliability report policy is absent from the lock")
})?;
let report = ReliabilityReport::build(&policy.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 parsed = parsed.map_err(anyhow::Error::new).and_then(|config| {
bitrouter::assemble::validate_host_configuration(&config)?;
Ok(config)
});
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")),
}
}
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 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")),
}
}
async fn resolve_supervisor_socket(
config: Option<&Path>,
socket: Option<&Path>,
) -> Result<PathBuf> {
let selected = resolve_client_socket(config, socket).await?;
if daemon::probe_status(&selected).await?.is_some() {
let source = bitrouter::paths::resolve_config(config)?;
bitrouter::upgrade::ensure_compatible(&source, &selected, false).await?;
}
Ok(selected)
}
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"),
_ => 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, .. }),
} => Some((config.as_deref(), socket.as_deref())),
Command::Models { config, .. } => Some((config.as_deref(), None)),
_ => None,
}
}
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})")
}
}
}
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> {
bitrouter::paths::ensure_home_directory(source.home())?;
let _lock = bitrouter::upgrade::lock(source).await?;
if daemon::endpoint_in_use(socket) {
let cfg = bitrouter::paths::load_config(source).await?;
if cfg.database.url.starts_with("sqlite:") && !cfg.database.url.contains(":memory:") {
let preflight = bitrouter::upgrade_preflight::Preflight::check_explicit(source).await?;
let backup = preflight.backup(source).await?;
eprintln!(
"note: explicit restart may interrupt active agent runs; SQLite recovery backup: {}",
backup.display()
);
} else {
eprintln!(
"note: explicit restart may interrupt active agent runs; database migration preflight is unavailable for this backend"
);
}
}
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::catalog::credential_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.file_path()?);
let record = bitrouter::policy_lock::rollback_to_digest(
loaded.file_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.file_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 mut cfg = config::parse(&raw).context("parsing bitrouter.yaml")?;
bitrouter::policy_lock::apply_defaults(&mut cfg, Some(config_path));
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().and_then(|lock| lock.path.clone());
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(|| {
if loaded.as_ref().is_some_and(|lock| lock.path.is_none()) {
"bundled"
} else {
"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,
} => {
if let Some(message) = bitrouter_ai::providers::retired::provider_message(&provider) {
anyhow::bail!("{message}");
}
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_supervisor_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_supervisor_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_supervisor_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_supervisor_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,
thread_id,
check,
read_only,
workspace,
model_policy,
max_output_tokens,
} = options;
if agent.is_none() && remote_context.is_none() {
if socket.is_some()
&& routing.model.is_none()
&& thread_id.is_none()
&& check.is_none()
&& !read_only
&& workspace.is_none()
&& !model_policy
&& max_output_tokens.is_none()
{
return bitrouter::dashboard::run(None, config.as_deref(), socket.as_deref(), None)
.await;
}
anyhow::ensure!(
!routing.direct
&& routing.base_url.is_none()
&& !routing.no_start
&& turn_timeout.is_none(),
"--direct, --base-url, --no-start, and --turn-timeout apply to `code <agent>`"
);
return bitrouter::native_code::run(
config.as_deref(),
socket.as_deref(),
routing.model,
thread_id,
check,
bitrouter::native_code::Options {
read_only,
model_policy,
max_output_tokens,
},
workspace,
)
.await;
}
anyhow::ensure!(
thread_id.is_none()
&& check.is_none()
&& !read_only
&& workspace.is_none()
&& !model_policy
&& max_output_tokens.is_none(),
"--thread-id, --check, --read-only, --workspace, --model-policy, and --max-output-tokens apply only to local bare `bro code`"
);
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(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", "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>"));
}
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,
daemon_version: None,
handoff_protocol: None,
handoff_build_id: None,
handoff_activity: None,
cli_owned: None,
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,
daemon_version: None,
handoff_protocol: None,
handoff_build_id: None,
handoff_activity: None,
cli_owned: None,
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,
daemon_version: None,
handoff_protocol: None,
handoff_build_id: None,
handoff_activity: None,
cli_owned: None,
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(())
}
#[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_ai::providers::hosted::credentials::{Credentials, StoredCredential};
use bitrouter::cloud::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::cloud::account::manager::CredentialManager;
use bitrouter_ai::providers::hosted::credentials::StoredCredential;
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"],
_ => 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": "semantic_route/v1|code:generation|implement|normal",
"request_key": "semantic_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_ai::types::ReasoningEffort::High)
);
assert_eq!(economy, "openai-codex:gpt-5.6-sol");
assert_eq!(
economy_effort,
Some(bitrouter_ai::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: "semantic_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: "semantic_route/v1|unknown|implement|normal".into(),
request_key: "semantic_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: "semantic_route/v1|unknown|implement|normal".into(),
request_key: "semantic_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 {
policy,
database_url,
config,
output,
},
}) => {
assert_eq!(policy, "auto");
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("missing policy"));
}
}