#[cfg(feature = "http")]
use std::io::{self, Write};
use std::{collections::BTreeMap, fmt, net::SocketAddr, path::PathBuf, time::Duration};
use anyhow::{bail, Result};
use clap::{ArgAction, Args, Parser, Subcommand, ValueEnum};
use leash_harness::{
capability::default_capability_descriptors,
config::{
resolve_config, AcceleratorBackend, AgentProvider, ConfigRequest, PartialHarnessConfig,
PolicyMode,
},
daemon::{
spawn_daemon, stop_process, tail_file, tail_jsonl_file, RunRecord, RunRegistry,
StopOutcome, DEFAULT_RUN_NAME,
},
module::default_module_graph,
replay::{scaled_delay, ReplayEvent, ReplayEventKind, ReplayRecording, REPLAY_FORMAT_VERSION},
stack::{built_in_stacks, find_stack, Stack, StackTransport},
transport::StreamTransportBackend,
types::RunLogEntry,
Harness, HarnessConfig, Profile, TelemetryStreamFrame,
};
use serde::Serialize;
#[cfg(feature = "mcp")]
use serde_json::{json, Map, Value};
use tracing::{
field::{Field, Visit},
Event, Subscriber,
};
use tracing_subscriber::{
fmt::{format::Writer, FmtContext, FormatEvent, FormatFields},
registry::LookupSpan,
EnvFilter,
};
#[derive(Debug, Parser)]
#[command(
name = "leash",
version,
about = "Composable local-LLM and robot harness"
)]
struct Cli {
#[arg(long, env = "LEASH_CONFIG", global = true)]
config: Option<PathBuf>,
#[command(subcommand)]
command: Command,
}
#[derive(Debug, Subcommand)]
enum Command {
List(ListArgs),
Serve(Serve),
Record(RecordArgs),
Replay(ReplayArgs),
#[cfg(any(feature = "http", feature = "mcp"))]
Run(RunArgs),
#[cfg(feature = "http")]
AgentSend(AgentSendArgs),
#[cfg(feature = "http")]
#[command(alias = "agent-chat")]
AgentInteractive(AgentInteractiveArgs),
#[cfg(feature = "mcp")]
Mcp(McpArgs),
Status(StatusArgs),
Log(LogArgs),
Restart(RestartArgs),
Graph(GraphArgs),
ShowConfig(ShowConfigArgs),
Health(HttpTarget),
Stop(StopArgs),
}
#[derive(Debug, Args)]
struct ListArgs {
#[arg(long, value_enum, default_value_t = ListFormat::Table)]
format: ListFormat,
}
#[derive(Debug, Clone, Copy, ValueEnum)]
enum ListFormat {
Table,
Json,
}
#[derive(Debug, Args)]
struct Serve {
#[command(subcommand)]
transport: Transport,
}
#[derive(Debug, Subcommand)]
enum Transport {
#[cfg(feature = "mcp")]
Mcp(RuntimeArgs),
#[cfg(all(feature = "http", feature = "mcp"))]
McpHttp(McpHttpServeArgs),
#[cfg(feature = "http")]
Http(HttpServeArgs),
}
#[derive(Debug, Args)]
struct RuntimeArgs {
#[arg(long)]
role: Option<String>,
#[arg(long, value_enum)]
profile: Option<Profile>,
#[arg(long, value_enum)]
stream_transport: Option<StreamTransportBackend>,
#[arg(long)]
replay_source: Option<PathBuf>,
#[arg(long)]
replay_speed: Option<f64>,
#[arg(long, action = ArgAction::SetTrue)]
allow_untokened_drive: bool,
#[arg(long, action = ArgAction::SetTrue)]
no_untokened_drive: bool,
#[arg(long, action = ArgAction::SetTrue)]
allow_physical_actuation: bool,
#[arg(long)]
deadman_ms: Option<u64>,
#[arg(long)]
soft_odometry_limit_m: Option<f64>,
#[arg(long)]
serial_port: Option<String>,
#[arg(long)]
serial_baud: Option<u32>,
#[arg(long, action = ArgAction::SetTrue)]
drive_invert: bool,
#[arg(long, action = ArgAction::SetTrue)]
drive_swap: bool,
#[arg(long, value_enum)]
accelerator: Option<AcceleratorBackend>,
#[arg(long, action = ArgAction::SetTrue)]
require_accelerator: bool,
#[arg(long, action = ArgAction::SetTrue)]
resource_sampling: bool,
#[arg(long, action = ArgAction::SetTrue)]
no_resource_sampling: bool,
#[arg(long, value_enum)]
agent_provider: Option<AgentProvider>,
#[arg(long)]
agent_model: Option<String>,
#[arg(long)]
agent_base_url: Option<String>,
#[arg(long)]
agent_api_key: Option<String>,
#[arg(long)]
agent_timeout_ms: Option<u64>,
#[arg(long, value_enum)]
policy_mode: Option<PolicyMode>,
}
#[derive(Debug, Args)]
struct HttpServeArgs {
#[command(flatten)]
runtime: RuntimeArgs,
#[arg(long)]
listen: Option<SocketAddr>,
}
#[derive(Debug, Args)]
struct McpHttpServeArgs {
#[command(flatten)]
runtime: RuntimeArgs,
#[arg(long, default_value = "127.0.0.1:9990")]
listen: SocketAddr,
}
#[derive(Debug, Args)]
struct HttpTarget {
#[arg(long, env = "LEASH_URL", default_value = "http://127.0.0.1:8000")]
url: String,
}
#[derive(Debug, Args)]
struct RecordArgs {
#[arg(short, long)]
output: PathBuf,
#[arg(long, default_value_t = 5)]
samples: usize,
#[arg(long, default_value_t = 50)]
interval_ms: u64,
#[command(flatten)]
runtime: RuntimeArgs,
}
#[derive(Debug, Args)]
struct ReplayArgs {
input: PathBuf,
#[arg(long, default_value_t = 1.0)]
speed: f64,
}
#[derive(Debug, Args)]
struct RunArgs {
stack: Option<String>,
#[arg(long)]
name: Option<String>,
#[arg(long, action = ArgAction::SetTrue)]
daemon: bool,
#[command(flatten)]
runtime: RuntimeArgs,
#[arg(long)]
listen: Option<SocketAddr>,
}
#[derive(Debug, Args)]
struct McpArgs {
#[command(subcommand)]
command: McpCommand,
}
#[derive(Debug, Subcommand)]
enum McpCommand {
ListTools(McpTargetArgs),
Call(McpCallArgs),
Status(McpTargetArgs),
Modules(McpTargetArgs),
}
#[derive(Debug, Args)]
struct McpTargetArgs {
#[arg(long, env = "LEASH_MCP_URL")]
url: Option<String>,
#[command(flatten)]
runtime: RuntimeArgs,
}
#[derive(Debug, Args)]
struct McpCallArgs {
tool: String,
#[arg(value_name = "KEY=VALUE")]
args: Vec<String>,
#[arg(long, value_name = "JSON")]
json: Option<String>,
#[arg(long, env = "LEASH_MCP_URL")]
url: Option<String>,
#[command(flatten)]
runtime: RuntimeArgs,
}
#[cfg(feature = "http")]
#[derive(Debug, Args)]
struct AgentSendArgs {
text: String,
#[arg(long, env = "LEASH_URL", default_value = "http://127.0.0.1:8000")]
url: String,
#[arg(long, default_value = "cli")]
source: String,
}
#[cfg(feature = "http")]
#[derive(Debug, Args)]
struct AgentInteractiveArgs {
#[arg(long, env = "LEASH_URL", default_value = "http://127.0.0.1:8000")]
url: String,
#[arg(long, default_value = "interactive")]
source: String,
}
#[derive(Debug, Args)]
struct StatusArgs {
name: Option<String>,
}
#[derive(Debug, Args)]
struct LogArgs {
#[arg(default_value = DEFAULT_RUN_NAME)]
name: String,
#[arg(long, default_value_t = 80)]
lines: usize,
#[arg(long)]
module: Option<String>,
#[arg(long, action = ArgAction::SetTrue)]
json: bool,
}
#[derive(Debug, Args)]
struct StopArgs {
#[arg(default_value = DEFAULT_RUN_NAME)]
name: String,
#[arg(long)]
url: Option<String>,
#[arg(long, default_value_t = 2_000)]
graceful_timeout_ms: u64,
}
#[derive(Debug, Args)]
struct RestartArgs {
#[arg(default_value = DEFAULT_RUN_NAME)]
name: String,
#[arg(long, default_value_t = 2_000)]
graceful_timeout_ms: u64,
}
#[derive(Debug, Args)]
struct GraphArgs {
#[arg(default_value = "sim")]
stack: String,
#[arg(long, value_enum, default_value_t = GraphFormat::Json)]
format: GraphFormat,
#[arg(long, env = "LEASH_ROLE", default_value = "robot")]
role: String,
#[arg(long, value_enum, default_value_t = StreamTransportBackend::LocalPubsub)]
stream_transport: StreamTransportBackend,
}
#[derive(Debug, Args)]
struct ShowConfigArgs {
stack: Option<String>,
#[command(flatten)]
runtime: RuntimeArgs,
#[arg(long)]
listen: Option<SocketAddr>,
}
#[derive(Debug, Clone, Copy, ValueEnum)]
enum GraphFormat {
Json,
Dot,
}
#[tokio::main]
async fn main() -> Result<()> {
init_tracing();
let cli = Cli::parse();
match cli.command {
Command::List(args) => {
print_stack_list(args.format)?;
}
Command::Serve(serve) => match serve.transport {
#[cfg(feature = "mcp")]
Transport::Mcp(args) => {
let harness =
Harness::new(config_from_args(args, None, cli.config.clone(), None)?)?;
leash_harness::mcp::serve_stdio(harness).await?;
}
#[cfg(all(feature = "http", feature = "mcp"))]
Transport::McpHttp(args) => {
let config =
config_from_args(args.runtime, Some(args.listen), cli.config.clone(), None)?;
let listen = config.listen;
let harness = Harness::new(config)?;
leash_harness::http::serve_mcp_http(harness, listen).await?;
}
#[cfg(feature = "http")]
Transport::Http(args) => {
let config = config_from_args(args.runtime, args.listen, cli.config.clone(), None)?;
let listen = config.listen;
let harness = Harness::new(config)?;
leash_harness::http::serve_http(harness, listen).await?;
}
},
Command::Record(args) => {
let output = record_stream(args, cli.config.clone()).await?;
println!("{}", serde_json::to_string_pretty(&output)?);
}
Command::Replay(args) => {
replay_file(args).await?;
}
#[cfg(any(feature = "http", feature = "mcp"))]
Command::Run(args) => {
run_stack(args, cli.config.clone()).await?;
}
#[cfg(feature = "http")]
Command::AgentSend(args) => {
let output = send_agent_message_http(&args.url, &args.source, &args.text).await?;
println!("{}", serde_json::to_string_pretty(&output)?);
}
#[cfg(feature = "http")]
Command::AgentInteractive(args) => {
run_agent_interactive(args).await?;
}
#[cfg(feature = "mcp")]
Command::Mcp(args) => {
run_mcp_command(args, cli.config.clone()).await?;
}
Command::Status(args) => {
let output = daemon_status(args.name.as_deref())?;
println!("{}", serde_json::to_string_pretty(&output)?);
}
Command::Log(args) => {
let registry = RunRegistry::from_env()?;
let Some(record) = registry.read(&args.name)? else {
bail!("run '{}' was not found", args.name);
};
let text = if args.json || args.module.is_some() {
tail_jsonl_file(&record.log_path, args.lines, args.module.as_deref())?
} else {
tail_file(&record.log_path, args.lines)?
};
if !text.is_empty() {
println!("{text}");
}
}
Command::Restart(args) => {
let record =
restart_daemon_run(&args.name, Duration::from_millis(args.graceful_timeout_ms))?;
println!("{}", serde_json::to_string_pretty(&record)?);
}
Command::Graph(args) => {
let graph = graph_from_args(&args)?;
match args.format {
GraphFormat::Json => println!("{}", serde_json::to_string_pretty(&graph)?),
GraphFormat::Dot => print!("{}", graph.to_dot()),
}
}
Command::ShowConfig(args) => {
let mut cli_overrides = args.runtime.into_partial_config()?;
if let Some(listen) = args.listen {
cli_overrides.listen = Some(listen);
}
let config_stack = args
.stack
.as_deref()
.map(resolve_config_stack)
.transpose()?;
let resolved = resolve_config(
ConfigRequest::from_process(
cli.config.clone(),
config_stack.as_ref().map(|stack| stack.profile),
cli_overrides,
)
.with_stack_defaults(
config_stack
.as_ref()
.map(|stack| stack.defaults.clone())
.unwrap_or_default(),
),
)?;
resolved.config.validate()?;
println!("{}", serde_json::to_string_pretty(&resolved)?);
}
Command::Health(target) => {
let value: serde_json::Value =
reqwest::get(format!("{}/health", target.url.trim_end_matches('/')))
.await?
.error_for_status()?
.json()
.await?;
println!("{}", serde_json::to_string_pretty(&value)?);
}
Command::Stop(args) => {
if let Some(url) = args.url {
let value = stop_http_target(&url).await?;
println!("{}", serde_json::to_string_pretty(&value)?);
} else {
let output =
stop_daemon_run(&args.name, Duration::from_millis(args.graceful_timeout_ms))?;
println!("{}", serde_json::to_string_pretty(&output)?);
}
}
}
Ok(())
}
fn print_stack_list(format: ListFormat) -> Result<()> {
let stacks = built_in_stacks();
match format {
ListFormat::Json => println!("{}", serde_json::to_string_pretty(&stacks)?),
ListFormat::Table => {
println!(
"{:<22} {:<13} {:<9} {:<8} {:<28} COMMAND",
"NAME", "PROFILE", "TRANSPORT", "HARDWARE", "FEATURES"
);
for stack in stacks {
let features = stack.required_features.join(",");
println!(
"{:<22} {:<13} {:<9} {:<8} {:<28} {}",
stack.name,
stack.profile.as_str(),
stack.transport.kind.as_str(),
if stack.hardware_required { "yes" } else { "no" },
features,
stack.command
);
}
}
}
Ok(())
}
#[cfg(feature = "mcp")]
async fn run_mcp_command(args: McpArgs, config_path: Option<PathBuf>) -> Result<()> {
match args.command {
McpCommand::ListTools(args) => {
if let Some(url) = args.url {
print_json(&mcp_get(&url, "tools").await?)?;
} else {
let _ = args.runtime;
print_json(&leash_harness::mcp::tool_list())?;
}
}
McpCommand::Call(args) => {
let call_args = parse_mcp_call_args(args.json.as_deref(), &args.args)?;
if let Some(url) = args.url {
print_json(&mcp_post_call(&url, &args.tool, call_args).await?)?;
} else {
let harness =
Harness::new(config_from_args(args.runtime, None, config_path, None)?)?;
print_json(&leash_harness::mcp::call_tool_with_origin(
&harness,
&args.tool,
call_args,
leash_harness::capability::InvocationOrigin::Cli,
)?)?;
}
}
McpCommand::Status(args) => {
if let Some(url) = args.url {
print_json(&mcp_get(&url, "status").await?)?;
} else {
let harness =
Harness::new(config_from_args(args.runtime, None, config_path, None)?)?;
print_json(&leash_harness::mcp::status(&harness, "local"))?;
}
}
McpCommand::Modules(args) => {
if let Some(url) = args.url {
print_json(&mcp_get(&url, "modules").await?)?;
} else {
let harness =
Harness::new(config_from_args(args.runtime, None, config_path, None)?)?;
print_json(&leash_harness::mcp::module_tool_map(&harness))?;
}
}
}
Ok(())
}
#[cfg(feature = "mcp")]
fn print_json(value: &impl Serialize) -> Result<()> {
println!("{}", serde_json::to_string_pretty(value)?);
Ok(())
}
#[cfg(feature = "mcp")]
async fn mcp_get(base_url: &str, path: &str) -> Result<Value> {
Ok(reqwest::get(mcp_endpoint(base_url, path))
.await?
.error_for_status()?
.json()
.await?)
}
#[cfg(feature = "mcp")]
async fn mcp_post_call(base_url: &str, tool: &str, args: Value) -> Result<Value> {
let client = reqwest::Client::new();
Ok(client
.post(mcp_endpoint(base_url, "call"))
.json(&json!({ "tool": tool, "args": args }))
.send()
.await?
.error_for_status()?
.json()
.await?)
}
#[cfg(feature = "mcp")]
fn mcp_endpoint(base_url: &str, path: &str) -> String {
let base = base_url.trim_end_matches('/');
let path = path.trim_start_matches('/');
if base.ends_with("/mcp") {
format!("{base}/{path}")
} else {
format!("{base}/mcp/{path}")
}
}
#[cfg(feature = "mcp")]
fn parse_mcp_call_args(json_arg: Option<&str>, raw_args: &[String]) -> Result<Value> {
let mut map = if let Some(json_arg) = json_arg {
parse_json_object(json_arg)?
} else if raw_args.len() == 1 && raw_args[0].trim_start().starts_with('{') {
parse_json_object(&raw_args[0])?
} else {
Map::new()
};
let skip_positional_json =
json_arg.is_none() && raw_args.len() == 1 && raw_args[0].trim_start().starts_with('{');
for (index, arg) in raw_args.iter().enumerate() {
if skip_positional_json && index == 0 {
continue;
}
let Some((key, value)) = arg.split_once('=') else {
bail!("expected MCP call argument '{arg}' to use key=value or pass --json '{{...}}'");
};
if key.trim().is_empty() {
bail!("MCP call argument key cannot be empty");
}
map.insert(key.to_string(), parse_mcp_arg_value(value));
}
Ok(Value::Object(map))
}
#[cfg(feature = "mcp")]
fn parse_json_object(text: &str) -> Result<Map<String, Value>> {
match serde_json::from_str(text)? {
Value::Object(map) => Ok(map),
_ => bail!("MCP JSON args must be a JSON object"),
}
}
#[cfg(feature = "mcp")]
fn parse_mcp_arg_value(value: &str) -> Value {
serde_json::from_str(value).unwrap_or_else(|_| Value::String(value.to_string()))
}
#[cfg(any(feature = "http", feature = "mcp"))]
async fn run_stack(args: RunArgs, config_path: Option<PathBuf>) -> Result<()> {
let selection = RunSelection::from_args(&args)?;
let config = config_from_args(
args.runtime,
args.listen,
config_path,
selection.stack.as_ref(),
)?;
match selection.transport {
StackTransport::Http => {
#[cfg(feature = "http")]
{
if args.daemon {
let record = start_daemon_run(&selection.name, &config)?;
println!("{}", serde_json::to_string_pretty(&record)?);
} else {
let listen = config.listen;
let harness = Harness::new(config)?;
leash_harness::http::serve_http(harness, listen).await?;
}
}
#[cfg(not(feature = "http"))]
{
let _ = config;
bail!("HTTP stacks require the 'http' feature");
}
}
StackTransport::Mcp => {
if args.daemon {
bail!("daemon mode is only supported for HTTP stacks");
}
#[cfg(feature = "mcp")]
{
let harness = Harness::new(config)?;
leash_harness::mcp::serve_stdio(harness).await?;
}
#[cfg(not(feature = "mcp"))]
{
let _ = config;
bail!("MCP stacks require the 'mcp' feature");
}
}
}
Ok(())
}
#[derive(Debug)]
struct RunSelection {
#[cfg_attr(not(feature = "http"), allow(dead_code))]
name: String,
stack: Option<Stack>,
transport: StackTransport,
}
impl RunSelection {
fn from_args(args: &RunArgs) -> Result<Self> {
if let Some(name) = args.stack.as_deref() {
if let Some(stack) = find_stack(name) {
stack.validate()?;
return Ok(Self {
name: args.name.clone().unwrap_or_else(|| stack.name.clone()),
transport: stack.transport.kind,
stack: Some(stack),
});
}
return Ok(Self {
name: args.name.clone().unwrap_or_else(|| name.to_string()),
stack: None,
transport: StackTransport::Http,
});
}
Ok(Self {
name: args
.name
.clone()
.unwrap_or_else(|| DEFAULT_RUN_NAME.to_string()),
stack: None,
transport: StackTransport::Http,
})
}
}
#[derive(Debug)]
struct ConfigStack {
profile: Profile,
defaults: PartialHarnessConfig,
}
fn resolve_config_stack(name: &str) -> Result<ConfigStack> {
if let Some(stack) = find_stack(name) {
stack.validate()?;
return Ok(ConfigStack {
profile: stack.profile,
defaults: stack.config_overrides,
});
}
let profile = match name {
"sim" => Profile::Sim,
"replay" => Profile::Replay,
"waveshare-ugv" => Profile::WaveshareUgv,
other => {
let stacks = built_in_stacks()
.into_iter()
.map(|stack| stack.name)
.collect::<Vec<_>>()
.join(", ");
bail!(
"unknown stack or profile '{other}'; expected sim, replay, waveshare-ugv, or one of: {stacks}"
);
}
};
Ok(ConfigStack {
profile,
defaults: PartialHarnessConfig::default(),
})
}
impl RuntimeArgs {
fn into_partial_config(self) -> Result<PartialHarnessConfig> {
if self.allow_untokened_drive && self.no_untokened_drive {
bail!("use either --allow-untokened-drive or --no-untokened-drive, not both");
}
if self.resource_sampling && self.no_resource_sampling {
bail!("use either --resource-sampling or --no-resource-sampling, not both");
}
let profile = if self.replay_source.is_some() && self.profile.is_none() {
Some(Profile::Replay)
} else {
self.profile
};
Ok(PartialHarnessConfig {
role: self.role,
profile,
stream_transport: self.stream_transport,
replay_source: self.replay_source,
replay_speed: self.replay_speed,
allow_untokened_drive: if self.allow_untokened_drive {
Some(true)
} else if self.no_untokened_drive {
Some(false)
} else {
None
},
allow_physical_actuation: self.allow_physical_actuation.then_some(true),
deadman_ms: self.deadman_ms,
soft_odometry_limit_m: self.soft_odometry_limit_m,
serial_port: self.serial_port,
serial_baud: self.serial_baud,
drive_invert: self.drive_invert.then_some(true),
drive_swap: self.drive_swap.then_some(true),
accelerator: self.accelerator,
require_accelerator: self.require_accelerator.then_some(true),
resource_sampling: if self.resource_sampling {
Some(true)
} else if self.no_resource_sampling {
Some(false)
} else {
None
},
agent_provider: self.agent_provider,
agent_model: self.agent_model,
agent_base_url: self.agent_base_url,
agent_api_key: self.agent_api_key,
agent_timeout_ms: self.agent_timeout_ms,
policy_mode: self.policy_mode,
..PartialHarnessConfig::default()
})
}
}
#[derive(Debug, Serialize)]
struct StatusOutput {
ok: bool,
state_dir: String,
stale_removed: Vec<String>,
runs: Vec<RunStatus>,
}
#[derive(Debug, Serialize)]
struct RunStatus {
name: String,
pid: u32,
running: bool,
transport: String,
profile: String,
listen: String,
log_path: String,
args: Vec<String>,
}
#[derive(Debug, Serialize)]
struct StopOutput {
ok: bool,
name: String,
pid: u32,
outcome: StopOutcome,
}
#[derive(Debug, Serialize)]
struct RecordOutput {
ok: bool,
format: &'static str,
path: String,
samples: usize,
events: usize,
profile: String,
}
async fn record_stream(args: RecordArgs, config_path: Option<PathBuf>) -> Result<RecordOutput> {
if args.samples == 0 {
bail!("record --samples must be at least 1");
}
let interval = Duration::from_millis(args.interval_ms);
let config = config_from_args(args.runtime, None, config_path, None)?;
let profile = config.profile.as_str().to_string();
let harness = Harness::new(config)?;
let mut events = Vec::with_capacity(args.samples * 4);
for sample in 0..args.samples {
let mut frame = harness.telemetry_stream_frame();
let ts_ms = sample as u128 * args.interval_ms as u128;
normalize_replay_frame_timestamps(&mut frame, ts_ms);
let seq = sample as u64 * 4;
events.push(ReplayEvent::new(
ts_ms,
seq,
ReplayEventKind::Telemetry,
serde_json::to_value(&frame)?,
));
events.push(ReplayEvent::new(
ts_ms,
seq + 1,
ReplayEventKind::Sensors,
serde_json::to_value(&frame.telemetry.sensors)?,
));
events.push(ReplayEvent::new(
ts_ms,
seq + 2,
ReplayEventKind::Camera,
serde_json::to_value(&frame.telemetry.sensors.camera)?,
));
events.push(ReplayEvent::new(
ts_ms,
seq + 3,
ReplayEventKind::Command,
serde_json::to_value(&frame.command)?,
));
if sample + 1 < args.samples {
tokio::time::sleep(interval).await;
}
}
let recording = ReplayRecording::new(events);
recording.write_path(&args.output)?;
Ok(RecordOutput {
ok: true,
format: REPLAY_FORMAT_VERSION,
path: args.output.display().to_string(),
samples: args.samples,
events: recording.events().len(),
profile,
})
}
async fn replay_file(args: ReplayArgs) -> Result<()> {
let recording = ReplayRecording::read_path(&args.input)?;
let mut previous_ts_ms = None;
for event in recording.events() {
if let Some(previous_ts_ms) = previous_ts_ms {
tokio::time::sleep(scaled_delay(previous_ts_ms, event.ts_ms, args.speed)?).await;
}
println!("{}", serde_json::to_string(event)?);
previous_ts_ms = Some(event.ts_ms);
}
Ok(())
}
fn normalize_replay_frame_timestamps(frame: &mut TelemetryStreamFrame, ts_ms: u128) {
frame.ts_ms = ts_ms;
frame.telemetry.ts_ms = ts_ms;
frame.telemetry.profile = "replay".to_string();
frame.telemetry.source = "replay".to_string();
frame.telemetry.sensors.raw_frame.source = "replay".to_string();
frame.telemetry.sensors.raw_frame.last_ms = Some(ts_ms);
frame.health.mode = "replay".to_string();
frame.health.replay = true;
frame.health.profile = "replay".to_string();
frame.health.uptime_ms = ts_ms;
frame.health.physical_actuation_enabled = false;
frame.safety.physical_actuation_enabled = false;
}
#[cfg(feature = "http")]
fn start_daemon_run(name: &str, config: &HarnessConfig) -> Result<RunRecord> {
let registry = RunRegistry::from_env()?;
registry.cleanup_stale()?;
if let Some(existing) = registry.read(name)? {
bail!(
"run '{}' is already registered with pid {}",
name,
existing.pid
);
}
let args = serve_http_args(config);
spawn_run_record(name, config, args)
}
fn restart_daemon_run(name: &str, graceful_timeout: Duration) -> Result<RunRecord> {
let registry = RunRegistry::from_env()?;
let Some(record) = registry.read(name)? else {
bail!("run '{name}' was not found");
};
if leash_harness::daemon::is_process_alive(record.pid) {
stop_process(record.pid, graceful_timeout)?;
}
registry.remove(name)?;
let pid = spawn_daemon(
&std::env::current_exe()?,
&record.args,
&record.log_path,
&record.name,
)?;
let now = leash_harness::daemon::now_ms();
let restarted = RunRecord {
pid,
started_at_ms: now,
updated_at_ms: now,
..record
};
registry.write(&restarted)?;
Ok(restarted)
}
fn stop_daemon_run(name: &str, graceful_timeout: Duration) -> Result<StopOutput> {
let registry = RunRegistry::from_env()?;
let Some(record) = registry.read(name)? else {
bail!("run '{name}' was not found");
};
let outcome = stop_process(record.pid, graceful_timeout)?;
registry.remove(name)?;
Ok(StopOutput {
ok: true,
name: record.name,
pid: record.pid,
outcome,
})
}
fn daemon_status(name: Option<&str>) -> Result<StatusOutput> {
let registry = RunRegistry::from_env()?;
let stale_removed = registry
.cleanup_stale()?
.into_iter()
.map(|record| record.name)
.collect::<Vec<_>>();
let records = if let Some(name) = name {
registry.read(name)?.into_iter().collect()
} else {
registry.list()?
};
let runs = records
.into_iter()
.map(|record| RunStatus {
running: leash_harness::daemon::is_process_alive(record.pid),
name: record.name,
pid: record.pid,
transport: record.transport,
profile: record.profile,
listen: record.listen,
log_path: record.log_path.display().to_string(),
args: record.args,
})
.collect::<Vec<_>>();
Ok(StatusOutput {
ok: true,
state_dir: registry.root().display().to_string(),
stale_removed,
runs,
})
}
#[cfg(feature = "http")]
fn spawn_run_record(name: &str, config: &HarnessConfig, args: Vec<String>) -> Result<RunRecord> {
let registry = RunRegistry::from_env()?;
let log_path = registry.log_path(name)?;
let pid = spawn_daemon(&std::env::current_exe()?, &args, &log_path, name)?;
let now = leash_harness::daemon::now_ms();
let record = RunRecord {
name: name.to_string(),
pid,
transport: "http".to_string(),
profile: config.profile.as_str().to_string(),
listen: config.listen.to_string(),
log_path,
args,
started_at_ms: now,
updated_at_ms: now,
};
registry.write(&record)?;
Ok(record)
}
#[cfg(feature = "http")]
fn serve_http_args(config: &HarnessConfig) -> Vec<String> {
let mut args = vec![
"serve".to_string(),
"http".to_string(),
"--role".to_string(),
config.role.clone(),
"--profile".to_string(),
config.profile.as_str().to_string(),
"--stream-transport".to_string(),
config.stream_transport.as_str().to_string(),
"--listen".to_string(),
config.listen.to_string(),
"--deadman-ms".to_string(),
config.deadman_ms.to_string(),
"--soft-odometry-limit-m".to_string(),
config.soft_odometry_limit_m.to_string(),
"--serial-port".to_string(),
config.serial_port.clone(),
"--serial-baud".to_string(),
config.serial_baud.to_string(),
"--accelerator".to_string(),
config.accelerator.as_str().to_string(),
];
if config.allow_untokened_drive {
args.push("--allow-untokened-drive".to_string());
} else {
args.push("--no-untokened-drive".to_string());
}
if config.allow_physical_actuation {
args.push("--allow-physical-actuation".to_string());
}
if let Some(path) = &config.replay_source {
args.push("--replay-source".to_string());
args.push(path.display().to_string());
args.push("--replay-speed".to_string());
args.push(config.replay_speed.to_string());
}
if config.drive_invert {
args.push("--drive-invert".to_string());
}
if config.drive_swap {
args.push("--drive-swap".to_string());
}
if config.require_accelerator {
args.push("--require-accelerator".to_string());
}
if config.resource_sampling {
args.push("--resource-sampling".to_string());
} else {
args.push("--no-resource-sampling".to_string());
}
args.push("--agent-provider".to_string());
args.push(config.agent_provider.as_str().to_string());
args.push("--agent-model".to_string());
args.push(config.agent_model.clone());
if let Some(base_url) = &config.agent_base_url {
args.push("--agent-base-url".to_string());
args.push(base_url.clone());
}
args.push("--agent-timeout-ms".to_string());
args.push(config.agent_timeout_ms.to_string());
args
}
async fn stop_http_target(url: &str) -> Result<serde_json::Value> {
let client = reqwest::Client::new();
Ok(client
.post(format!("{}/motors/stop", url.trim_end_matches('/')))
.send()
.await?
.error_for_status()?
.json()
.await?)
}
#[cfg(feature = "http")]
async fn run_agent_interactive(args: AgentInteractiveArgs) -> Result<()> {
let mut line = String::new();
loop {
print!("agent> ");
io::stdout().flush()?;
line.clear();
if io::stdin().read_line(&mut line)? == 0 {
break;
}
let text = line.trim();
if text.is_empty() {
continue;
}
if matches!(text, "/quit" | "/exit") {
break;
}
let output = send_agent_message_http(&args.url, &args.source, text).await?;
println!("{}", serde_json::to_string_pretty(&output)?);
}
Ok(())
}
#[cfg(feature = "http")]
async fn send_agent_message_http(url: &str, source: &str, text: &str) -> Result<serde_json::Value> {
let client = reqwest::Client::new();
Ok(client
.post(agent_messages_endpoint(url))
.json(&serde_json::json!({
"source": source,
"text": text,
}))
.send()
.await?
.error_for_status()?
.json()
.await?)
}
#[cfg(feature = "http")]
fn agent_messages_endpoint(base_url: &str) -> String {
format!("{}/agent/messages", base_url.trim_end_matches('/'))
}
fn graph_from_args(args: &GraphArgs) -> Result<leash_harness::ModuleGraph> {
let config_stack = resolve_config_stack(&args.stack)?;
let capabilities = default_capability_descriptors()
.into_iter()
.map(|descriptor| descriptor.name)
.collect();
let config = HarnessConfig {
role: args.role.clone(),
profile: config_stack.profile,
stream_transport: args.stream_transport,
..HarnessConfig::default()
};
Ok(default_module_graph(&config, capabilities))
}
fn config_from_args(
args: RuntimeArgs,
listen: Option<SocketAddr>,
config_path: Option<PathBuf>,
stack: Option<&Stack>,
) -> Result<HarnessConfig> {
let mut cli = args.into_partial_config()?;
if let Some(listen) = listen {
cli.listen = Some(listen);
}
Ok(resolve_config(
ConfigRequest::from_process(config_path, stack.map(|stack| stack.profile), cli)
.with_stack_defaults(
stack
.map(|stack| stack.config_overrides.clone())
.unwrap_or_default(),
),
)?
.config)
}
fn init_tracing() {
let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
if let Some(run_id) = std::env::var("LEASH_RUN_ID")
.ok()
.filter(|run_id| !run_id.trim().is_empty())
{
tracing_subscriber::fmt()
.event_format(JsonlEventFormat { run_id })
.with_env_filter(filter)
.with_writer(std::io::stderr)
.init();
} else {
tracing_subscriber::fmt()
.with_env_filter(filter)
.with_writer(std::io::stderr)
.init();
}
}
struct JsonlEventFormat {
run_id: String,
}
impl<S, N> FormatEvent<S, N> for JsonlEventFormat
where
S: Subscriber + for<'a> LookupSpan<'a>,
N: for<'a> FormatFields<'a> + 'static,
{
fn format_event(
&self,
_ctx: &FmtContext<'_, S, N>,
mut writer: Writer<'_>,
event: &Event<'_>,
) -> fmt::Result {
let mut visitor = JsonFieldVisitor::default();
event.record(&mut visitor);
let metadata = event.metadata();
let entry = RunLogEntry {
timestamp: leash_harness::daemon::now_ms(),
run_id: self.run_id.clone(),
module: metadata.target().to_string(),
event: visitor
.message
.unwrap_or_else(|| metadata.name().to_string()),
level: metadata.level().as_str().to_ascii_lowercase(),
fields: visitor.fields,
};
let line = serde_json::to_string(&entry).map_err(|_| fmt::Error)?;
writeln!(writer, "{line}")
}
}
#[derive(Default)]
struct JsonFieldVisitor {
message: Option<String>,
fields: BTreeMap<String, serde_json::Value>,
}
impl JsonFieldVisitor {
fn insert(&mut self, field: &Field, value: serde_json::Value) {
if field.name() == "message" {
if let Some(message) = value.as_str() {
self.message = Some(message.to_string());
}
return;
}
self.fields.insert(field.name().to_string(), value);
}
}
impl Visit for JsonFieldVisitor {
fn record_bool(&mut self, field: &Field, value: bool) {
self.insert(field, serde_json::json!(value));
}
fn record_i64(&mut self, field: &Field, value: i64) {
self.insert(field, serde_json::json!(value));
}
fn record_u64(&mut self, field: &Field, value: u64) {
self.insert(field, serde_json::json!(value));
}
fn record_f64(&mut self, field: &Field, value: f64) {
self.insert(field, serde_json::json!(value));
}
fn record_str(&mut self, field: &Field, value: &str) {
self.insert(field, serde_json::json!(value));
}
fn record_debug(&mut self, field: &Field, value: &dyn fmt::Debug) {
self.insert(field, serde_json::json!(format!("{value:?}")));
}
}
#[cfg(all(test, feature = "mcp"))]
mod tests {
use super::*;
#[test]
fn parses_key_value_mcp_args_as_json_scalars() {
let args = parse_mcp_call_args(
None,
&[
"capability=drive".to_string(),
"left=0.2".to_string(),
"right=0".to_string(),
"speed_mode=low".to_string(),
],
)
.unwrap();
assert_eq!(
args,
json!({
"capability": "drive",
"left": 0.2,
"right": 0,
"speed_mode": "low",
})
);
}
#[test]
fn parses_json_mcp_args_and_allows_key_value_overrides() {
let args = parse_mcp_call_args(
Some(r#"{"capability":"speed_mode","speed_mode":"medium"}"#),
&["speed_mode=low".to_string()],
)
.unwrap();
assert_eq!(
args,
json!({
"capability": "speed_mode",
"speed_mode": "low",
})
);
}
#[test]
fn builds_mcp_urls_without_duplicate_prefixes() {
assert_eq!(
mcp_endpoint("http://127.0.0.1:9990", "tools"),
"http://127.0.0.1:9990/mcp/tools"
);
assert_eq!(
mcp_endpoint("http://127.0.0.1:9990/mcp", "tools"),
"http://127.0.0.1:9990/mcp/tools"
);
}
}
#[cfg(all(test, feature = "http"))]
mod http_tests {
use super::*;
#[test]
fn builds_agent_message_urls_without_trailing_slash_duplication() {
assert_eq!(
agent_messages_endpoint("http://127.0.0.1:8000"),
"http://127.0.0.1:8000/agent/messages"
);
assert_eq!(
agent_messages_endpoint("http://127.0.0.1:8000/"),
"http://127.0.0.1:8000/agent/messages"
);
}
}