Add external worker supervision

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

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.

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"],

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"

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.

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<String>,
+}
+
+#[derive(Debug, Clone, Copy, ValueEnum)]
+enum WorkerRestartArg {
+    Never,
+    OnFailure,
+}
+
+impl From<WorkerRestartArg> 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<ExternalWorkerStatus>,
+}
+
+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 {

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,
+};