diff --git a/README.md b/README.md index 7b2402d..3adaa08 100644 --- a/README.md +++ b/README.md @@ -11,6 +11,7 @@ leash run sim-mcp # MCP stdio for LLM agents leash run sim-http # localhost HTTP + WebSocket leash run sim-stream-hub # localhost TCP JSONL stream hub leash serve mcp-http # localhost MCP JSON control surface +leash worker run -- node -e 'setInterval(() => {}, 1000)' ``` ## Why Leash @@ -82,6 +83,9 @@ leash agent-interactive # Run a TCP JSONL stream hub for external module processes leash run sim-stream-hub +# Start and supervise an explicit external worker process +leash worker run --name demo-worker -- node -e 'setInterval(() => {}, 1000)' + # Run as a daemon and inspect JSONL logs leash run sim-http --daemon leash log sim-http --json --module http --lines 20 @@ -477,6 +481,23 @@ leash serve stream-hub --profile sim --listen 127.0.0.1:9970 leash run sim-stream-hub ``` +## External Workers + +External workers are opt-in child processes. Leash does not start any external +process unless a caller provides the command explicitly. The supervisor records +the worker name, command, restart policy, process-health check, PID, exit code, +restart count, and required/optional metadata. + +Run a one-shot supervised worker check: + +```bash +leash worker run --name demo-worker --hold-ms 150 -- node -e 'setInterval(() => {}, 1000)' +``` + +The command prints JSON status while the worker is running, then stops and reaps +the child process before exiting. Restart policy is deliberately small today: +`never` by default, or `on-failure` with `--max-restarts N`. + ## Run Logs and Resource Samples Daemon runs write structured JSONL logs under the Leash state directory. Each @@ -547,7 +568,7 @@ See [issues](https://github.com/specdog/leash/issues) for the full plan. Highlig - [x] Agent input channels: one-shot CLI, interactive CLI, and localhost web input - [x] TCP JSONL stream framing for cross-process modules - [x] Runnable TCP JSONL stream hub for cross-process module links -- [ ] External worker lifecycle and supervision +- [x] External worker lifecycle and supervision - [ ] MAVLink drone + manipulator adapters - [x] Localhost command center dashboard - [x] Viewer-ready visualization frames diff --git a/scripts/README.md b/scripts/README.md index ff5f8d8..6befbe4 100644 --- a/scripts/README.md +++ b/scripts/README.md @@ -9,6 +9,7 @@ flowchart TB all --> mcp["smoke-mcp.sh\nstdio MCP initialize, tools, health"] all --> mcphttp["smoke-mcp-http.sh\nlocalhost MCP HTTP, CLI, planner, and patrol calls"] all --> streamhub["smoke-stream-hub.sh\nTCP JSONL stream hub"] + all --> worker["smoke-worker.sh\nexternal worker lifecycle"] all --> replayhttp["smoke-replay-http.sh\nHTTP replay observe"] all --> replaymcp["smoke-replay-mcp.sh\nMCP replay observe"] all --> physical["smoke-physical-gate.sh\nphysical profile refuses without gate"] @@ -24,6 +25,7 @@ flowchart TB - `smoke-mcp.sh`: stdio MCP initialization and tool calls. - `smoke-mcp-http.sh`: localhost MCP HTTP routes, `leash mcp` CLI calls, sim planner set/status calls, and sim patrol start/status/stop calls. - `smoke-stream-hub.sh`: starts the localhost TCP JSONL stream hub, sends valid frames, and proves an invalid peer does not kill the listener. +- `smoke-worker.sh`: starts an explicit child process through the worker supervisor, verifies JSON status, and lets the supervisor stop it. - `smoke-replay-http.sh`: replay mode over HTTP. - `smoke-replay-mcp.sh`: replay mode over MCP. - `smoke-physical-gate.sh`: proves physical startup fails without explicit actuation. diff --git a/scripts/smoke-all.sh b/scripts/smoke-all.sh index 30ee4b2..2ede33f 100755 --- a/scripts/smoke-all.sh +++ b/scripts/smoke-all.sh @@ -58,6 +58,11 @@ const checks = [ argv: ["bash", "scripts/smoke-stream-hub.sh"], proof: "TCP JSONL stream hub accepted valid frames and kept serving after an invalid peer", }, + { + name: "worker-supervision", + argv: ["bash", "scripts/smoke-worker.sh"], + proof: "external worker supervisor started, reported, and stopped a child process", + }, { name: "replay-http-observe", argv: ["bash", "scripts/smoke-replay-http.sh"], diff --git a/scripts/smoke-worker.sh b/scripts/smoke-worker.sh new file mode 100755 index 0000000..3892266 --- /dev/null +++ b/scripts/smoke-worker.sh @@ -0,0 +1,34 @@ +#!/usr/bin/env bash +set -euo pipefail + +output="$( + cargo run --quiet -- worker run \ + --name smoke-worker \ + --hold-ms 150 \ + -- node -e 'setInterval(() => {}, 1000)' +)" + +WORKER_OUTPUT="$output" node <<'EOF' +const payload = JSON.parse(process.env.WORKER_OUTPUT || ""); + +if (payload.ok !== true) throw new Error("worker smoke ok was not true"); +if (!Array.isArray(payload.statuses) || payload.statuses.length !== 1) { + throw new Error(`expected one worker status, got ${JSON.stringify(payload.statuses)}`); +} + +const worker = payload.statuses[0]; +if (worker.name !== "smoke-worker") throw new Error(`unexpected worker name: ${worker.name}`); +if (worker.state !== "running") throw new Error(`unexpected worker state: ${worker.state}`); +if (!Number.isInteger(worker.pid) || worker.pid <= 0) { + throw new Error(`worker pid was invalid: ${worker.pid}`); +} +if (worker.restart_policy !== "never") { + throw new Error(`unexpected restart policy: ${worker.restart_policy}`); +} +if (worker.health_check !== "process") { + throw new Error(`unexpected health check: ${worker.health_check}`); +} +if (worker.required !== true) throw new Error("worker should default to required"); +EOF + +echo "worker smoke ok" diff --git a/src/README.md b/src/README.md index 3bec26e..6057a23 100644 --- a/src/README.md +++ b/src/README.md @@ -15,6 +15,7 @@ flowchart TB lib --> replay["replay.rs\nrecord/playback JSONL"] lib --> daemon["daemon.rs\nbackground runs and logs"] lib --> transport["transport.rs\nmemory and local-pubsub streams"] + lib --> worker["worker.rs\nexternal worker supervision"] lib --> types["types.rs\nshared API structs"] lib --> accelerator["accelerator.rs\nCPU/CUDA probe model"] lib --> agent["agent.rs\nagent model provider glue"] @@ -41,4 +42,5 @@ flowchart TB - `stream_processing.rs`: generic latest-value, rate-limit, quality, and timestamp pairing helpers. - `transport.rs`: stream transport interface plus memory and local pubsub implementations. - `types.rs`: serialized HTTP/MCP/replay/API payload types, including viewer visualization, pose, twist, path, occupancy-grid, costmap, detection, vision, planner, patrol, spatial memory, autonomy overlay, and map metadata frames. +- `worker.rs`: explicit external worker specs, process lifecycle supervision, status, and restart policy. - `bin/`: CLI entrypoint crate target. diff --git a/src/bin/leash.rs b/src/bin/leash.rs index 154e1f0..0cb103d 100644 --- a/src/bin/leash.rs +++ b/src/bin/leash.rs @@ -18,6 +18,10 @@ use leash_harness::{ stack::{built_in_stacks, find_stack, Stack, StackTransport}, transport::{spawn_tcp_jsonl_stream_hub, StreamTransportBackend}, types::RunLogEntry, + worker::{ + ExternalWorkerSpec, ExternalWorkerState, ExternalWorkerStatus, WorkerRestartPolicy, + WorkerSupervisor, + }, Harness, HarnessConfig, Profile, TelemetryStreamFrame, }; use serde::Serialize; @@ -65,6 +69,7 @@ enum Command { Status(StatusArgs), Log(LogArgs), Restart(RestartArgs), + Worker(WorkerArgs), Graph(GraphArgs), ShowConfig(ShowConfigArgs), Health(HttpTarget), @@ -354,6 +359,53 @@ struct RestartArgs { graceful_timeout_ms: u64, } +#[derive(Debug, Args)] +struct WorkerArgs { + #[command(subcommand)] + command: WorkerCommand, +} + +#[derive(Debug, Subcommand)] +enum WorkerCommand { + Run(WorkerRunArgs), +} + +#[derive(Debug, Args)] +struct WorkerRunArgs { + #[arg(long, default_value = "external-worker")] + name: String, + + #[arg(long, value_enum, default_value_t = WorkerRestartArg::Never)] + restart: WorkerRestartArg, + + #[arg(long, default_value_t = 0)] + max_restarts: u32, + + #[arg(long, default_value_t = 100)] + hold_ms: u64, + + #[arg(long = "optional", action = ArgAction::SetFalse)] + required: bool, + + #[arg(required = true, trailing_var_arg = true, value_name = "COMMAND")] + argv: Vec, +} + +#[derive(Debug, Clone, Copy, ValueEnum)] +enum WorkerRestartArg { + Never, + OnFailure, +} + +impl From for WorkerRestartPolicy { + fn from(value: WorkerRestartArg) -> Self { + match value { + WorkerRestartArg::Never => Self::Never, + WorkerRestartArg::OnFailure => Self::OnFailure, + } + } +} + #[derive(Debug, Args)] struct GraphArgs { #[arg(default_value = "sim")] @@ -470,6 +522,9 @@ async fn main() -> Result<()> { restart_daemon_run(&args.name, Duration::from_millis(args.graceful_timeout_ms))?; println!("{}", serde_json::to_string_pretty(&record)?); } + Command::Worker(args) => { + run_worker_command(args).await?; + } Command::Graph(args) => { let graph = graph_from_args(&args)?; match args.format { @@ -527,6 +582,47 @@ async fn main() -> Result<()> { Ok(()) } +#[derive(Debug, Serialize)] +struct WorkerRunOutput { + ok: bool, + statuses: Vec, +} + +async fn run_worker_command(args: WorkerArgs) -> Result<()> { + match args.command { + WorkerCommand::Run(args) => run_worker(args).await, + } +} + +async fn run_worker(args: WorkerRunArgs) -> Result<()> { + let Some((command, command_args)) = args.argv.split_first() else { + bail!("worker command is required"); + }; + let mut spec = ExternalWorkerSpec::new(args.name, command.clone()); + spec.args = command_args.to_vec(); + spec.restart_policy = args.restart.into(); + spec.max_restarts = args.max_restarts; + spec.required = args.required; + + let mut supervisor = WorkerSupervisor::new(); + supervisor.add(spec)?; + supervisor.start_all()?; + tokio::time::sleep(Duration::from_millis(args.hold_ms)).await; + supervisor.poll_all()?; + + let statuses = supervisor.statuses(); + let output = WorkerRunOutput { + ok: statuses + .iter() + .all(|status| status.state == ExternalWorkerState::Running), + statuses, + }; + println!("{}", serde_json::to_string_pretty(&output)?); + io::stdout().flush()?; + supervisor.stop_all()?; + Ok(()) +} + fn print_stack_list(format: ListFormat) -> Result<()> { let stacks = built_in_stacks(); match format { diff --git a/src/lib.rs b/src/lib.rs index 92f5915..749bc08 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -22,6 +22,7 @@ pub mod stack; pub mod stream_processing; pub mod transport; pub mod types; +pub mod worker; pub use accelerator::{AcceleratorProbe, AcceleratorProvider, AcceleratorStatus}; pub use agent::complete as complete_agent_prompt; @@ -63,3 +64,7 @@ pub use types::{ VisualizationPath, COST_FREE, COST_LETHAL, COST_UNKNOWN, OCCUPANCY_FREE, OCCUPANCY_OCCUPIED, OCCUPANCY_UNKNOWN, VISUALIZATION_FRAME_VERSION, }; +pub use worker::{ + ExternalWorkerSpec, ExternalWorkerState, ExternalWorkerStatus, WorkerHealthCheck, + WorkerRestartPolicy, WorkerSupervisor, +}; diff --git a/src/worker.rs b/src/worker.rs new file mode 100644 index 0000000..1bf7f92 --- /dev/null +++ b/src/worker.rs @@ -0,0 +1,368 @@ +use std::{ + collections::BTreeMap, + process::{Child, Command, ExitStatus, Stdio}, +}; + +use anyhow::{bail, Context, Result}; +use serde::{Deserialize, Serialize}; + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "mcp", derive(schemars::JsonSchema))] +pub struct ExternalWorkerSpec { + pub name: String, + pub command: String, + pub args: Vec, + pub env: BTreeMap, + pub restart_policy: WorkerRestartPolicy, + pub max_restarts: u32, + pub health_check: WorkerHealthCheck, + pub required: bool, +} + +impl ExternalWorkerSpec { + pub fn new(name: impl Into, command: impl Into) -> Self { + Self { + name: name.into(), + command: command.into(), + args: Vec::new(), + env: BTreeMap::new(), + restart_policy: WorkerRestartPolicy::Never, + max_restarts: 0, + health_check: WorkerHealthCheck::Process, + required: true, + } + } + + pub fn validate(&self) -> Result<()> { + if self.name.trim().is_empty() { + bail!("external worker name cannot be empty"); + } + if self.command.trim().is_empty() { + bail!("external worker '{}' command cannot be empty", self.name); + } + for key in self.env.keys() { + if key.trim().is_empty() { + bail!("external worker '{}' env key cannot be empty", self.name); + } + } + Ok(()) + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "mcp", derive(schemars::JsonSchema))] +#[serde(rename_all = "kebab-case")] +pub enum WorkerRestartPolicy { + Never, + OnFailure, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "mcp", derive(schemars::JsonSchema))] +#[serde(rename_all = "kebab-case")] +pub enum WorkerHealthCheck { + Process, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "mcp", derive(schemars::JsonSchema))] +#[serde(rename_all = "kebab-case")] +pub enum ExternalWorkerState { + Planned, + Starting, + Running, + Exited, + Failed, + Stopped, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "mcp", derive(schemars::JsonSchema))] +pub struct ExternalWorkerStatus { + pub name: String, + pub state: ExternalWorkerState, + pub pid: Option, + pub exit_code: Option, + pub restarts: u32, + pub message: Option, + pub restart_policy: WorkerRestartPolicy, + pub health_check: WorkerHealthCheck, + pub required: bool, +} + +struct WorkerRuntime { + spec: ExternalWorkerSpec, + child: Option, + status: ExternalWorkerStatus, +} + +pub struct WorkerSupervisor { + workers: BTreeMap, +} + +impl WorkerSupervisor { + pub fn new() -> Self { + Self { + workers: BTreeMap::new(), + } + } + + pub fn add(&mut self, spec: ExternalWorkerSpec) -> Result<()> { + spec.validate()?; + if self.workers.contains_key(&spec.name) { + bail!("duplicate external worker '{}'", spec.name); + } + let status = ExternalWorkerStatus { + name: spec.name.clone(), + state: ExternalWorkerState::Planned, + pid: None, + exit_code: None, + restarts: 0, + message: Some("planned".to_string()), + restart_policy: spec.restart_policy, + health_check: spec.health_check, + required: spec.required, + }; + self.workers.insert( + spec.name.clone(), + WorkerRuntime { + spec, + child: None, + status, + }, + ); + Ok(()) + } + + pub fn start_all(&mut self) -> Result<()> { + for name in self.workers.keys().cloned().collect::>() { + self.start(&name)?; + } + Ok(()) + } + + pub fn start(&mut self, name: &str) -> Result<()> { + let runtime = self + .workers + .get_mut(name) + .with_context(|| format!("unknown external worker '{name}'"))?; + if runtime.child.is_some() { + bail!("external worker '{name}' is already running"); + } + runtime.status.state = ExternalWorkerState::Starting; + runtime.status.message = Some("starting".to_string()); + let child = spawn_child(&runtime.spec)?; + runtime.status.pid = Some(child.id()); + runtime.status.exit_code = None; + runtime.status.state = ExternalWorkerState::Running; + runtime.status.message = Some("running".to_string()); + runtime.child = Some(child); + Ok(()) + } + + pub fn poll_all(&mut self) -> Result<()> { + for name in self.workers.keys().cloned().collect::>() { + self.poll(&name)?; + } + Ok(()) + } + + pub fn poll(&mut self, name: &str) -> Result<()> { + let Some(runtime) = self.workers.get_mut(name) else { + bail!("unknown external worker '{name}'"); + }; + let Some(child) = runtime.child.as_mut() else { + return Ok(()); + }; + let Some(exit) = child.try_wait()? else { + runtime.status.state = ExternalWorkerState::Running; + runtime.status.message = Some("running".to_string()); + return Ok(()); + }; + + let pid = runtime.status.pid; + runtime.child = None; + runtime.status.pid = pid; + runtime.status.exit_code = exit.code(); + + if should_restart(&runtime.spec, &runtime.status, exit) { + runtime.status.restarts += 1; + let child = spawn_child(&runtime.spec)?; + runtime.status.pid = Some(child.id()); + runtime.status.exit_code = None; + runtime.status.state = ExternalWorkerState::Running; + runtime.status.message = Some(format!( + "restarted after exit {}; restart {}/{}", + describe_exit(exit), + runtime.status.restarts, + runtime.spec.max_restarts + )); + runtime.child = Some(child); + } else { + runtime.status.state = ExternalWorkerState::Exited; + runtime.status.message = Some(format!("exited {}", describe_exit(exit))); + } + Ok(()) + } + + pub fn stop_all(&mut self) -> Result<()> { + for name in self + .workers + .keys() + .cloned() + .collect::>() + .into_iter() + .rev() + { + self.stop(&name)?; + } + Ok(()) + } + + pub fn stop(&mut self, name: &str) -> Result<()> { + let Some(runtime) = self.workers.get_mut(name) else { + bail!("unknown external worker '{name}'"); + }; + if let Some(mut child) = runtime.child.take() { + match child.try_wait()? { + Some(exit) => { + runtime.status.exit_code = exit.code(); + runtime.status.message = Some(format!("exited {}", describe_exit(exit))); + } + None => { + child.kill()?; + let exit = child.wait()?; + runtime.status.exit_code = exit.code(); + runtime.status.message = Some("stopped".to_string()); + } + } + } else { + runtime.status.message = Some("stopped".to_string()); + } + runtime.status.pid = None; + runtime.status.state = ExternalWorkerState::Stopped; + Ok(()) + } + + pub fn statuses(&self) -> Vec { + self.workers + .values() + .map(|runtime| runtime.status.clone()) + .collect() + } + + pub fn status(&self, name: &str) -> Option<&ExternalWorkerStatus> { + self.workers.get(name).map(|runtime| &runtime.status) + } +} + +impl Default for WorkerSupervisor { + fn default() -> Self { + Self::new() + } +} + +fn spawn_child(spec: &ExternalWorkerSpec) -> Result { + let mut command = Command::new(&spec.command); + command + .args(&spec.args) + .envs(&spec.env) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()); + command + .spawn() + .with_context(|| format!("failed to start external worker '{}'", spec.name)) +} + +fn should_restart( + spec: &ExternalWorkerSpec, + status: &ExternalWorkerStatus, + exit: ExitStatus, +) -> bool { + spec.restart_policy == WorkerRestartPolicy::OnFailure + && !exit.success() + && status.restarts < spec.max_restarts +} + +fn describe_exit(exit: ExitStatus) -> String { + match exit.code() { + Some(code) => format!("code {code}"), + None => "without exit code".to_string(), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::{thread, time::Duration}; + + fn shell_worker(name: &str, script: &str) -> ExternalWorkerSpec { + let mut spec = ExternalWorkerSpec::new(name, "sh"); + spec.args = vec!["-c".to_string(), script.to_string()]; + spec + } + + #[test] + fn rejects_empty_worker_command() { + let spec = ExternalWorkerSpec::new("empty", " "); + let err = spec.validate().unwrap_err().to_string(); + assert!(err.contains("command cannot be empty")); + } + + #[test] + fn starts_reports_and_stops_worker_process() { + let mut supervisor = WorkerSupervisor::new(); + supervisor + .add(shell_worker("sleeper", "sleep 5")) + .expect("valid worker"); + + supervisor.start("sleeper").unwrap(); + supervisor.poll("sleeper").unwrap(); + + let status = supervisor.status("sleeper").unwrap(); + assert_eq!(status.state, ExternalWorkerState::Running); + assert!(status.pid.is_some()); + + supervisor.stop("sleeper").unwrap(); + let status = supervisor.status("sleeper").unwrap(); + assert_eq!(status.state, ExternalWorkerState::Stopped); + assert!(status.pid.is_none()); + } + + #[test] + fn reports_exited_worker_process() { + let mut supervisor = WorkerSupervisor::new(); + supervisor + .add(shell_worker("exit-seven", "exit 7")) + .expect("valid worker"); + + supervisor.start("exit-seven").unwrap(); + thread::sleep(Duration::from_millis(50)); + supervisor.poll("exit-seven").unwrap(); + + let status = supervisor.status("exit-seven").unwrap(); + assert_eq!(status.state, ExternalWorkerState::Exited); + assert_eq!(status.exit_code, Some(7)); + assert!(status.message.as_deref().unwrap_or("").contains("code 7")); + } + + #[test] + fn restarts_failed_worker_within_limit() { + let mut spec = shell_worker("restart-once", "sleep 0.01; exit 9"); + spec.restart_policy = WorkerRestartPolicy::OnFailure; + spec.max_restarts = 1; + let mut supervisor = WorkerSupervisor::new(); + supervisor.add(spec).unwrap(); + + supervisor.start("restart-once").unwrap(); + thread::sleep(Duration::from_millis(50)); + supervisor.poll("restart-once").unwrap(); + + let status = supervisor.status("restart-once").unwrap(); + assert_eq!(status.state, ExternalWorkerState::Running); + assert_eq!(status.restarts, 1); + + supervisor.stop("restart-once").unwrap(); + } +}