diff --git a/CHANGELOG.md b/CHANGELOG.md index 58aecec..512b6d3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,30 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +## [0.7.0] - 2026-08-30 + +### Added +- **`ctl` control plane (C1) — opt-in.** amux can now host a *controllable* + hierarchy of agents. Launch with `amux --allow-ctl [--max-depth ] …` and + amux binds a per-process control channel (a Windows **named pipe** / unix + **socket**, zero third-party deps), injecting its address (`AMUX_CTL`) and each + pane's id (`AMUX_PANE`) into every pane it spawns. From inside a pane: + - `amux ctl spawn [--role R] -- ` opens a **visible** new worker pane + (returns its agent id + session id as JSON); + - `amux ctl list` returns the spawn tree (id, parent, role, depth, live + `agsess` status) as JSON. + The channel is drained non-blocking from the run loop (no thread). **Off by + default:** without `--allow-ctl` there is no pipe, `amux ctl` refuses, and + behavior is identical to before. +- **Spawn-tree safety guards.** `ctl spawn` accepts only commands on the agent + allowlist (`{claude}`, extensible per-session via `AMUX_CTL_ALLOW`), and a + `--max-depth` ceiling (default 6, `0` = unlimited) bounds *recursion* — never + fleet width — as a fork-bomb circuit-breaker. Every worker is a normal pane: + visible in the bar, killable, in the org chart. + +_Not yet: `ctl send` / `status` / `kill`, identity delegation, `--here` splits +(they arrive in C2–C3)._ + ## [0.6.1] - 2026-08-29 ### Fixed diff --git a/Cargo.toml b/Cargo.toml index 1b87d58..56aad17 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "amux" -version = "0.6.1" +version = "0.7.0" edition = "2021" rust-version = "1.70" description = "tmux for agents: a multi-agent terminal that hosts your own CLIs in switchable panes with agent-aware chrome. 100% nativelite - zero third-party dependencies." diff --git a/src/ctl.rs b/src/ctl.rs new file mode 100644 index 0000000..cf5985b --- /dev/null +++ b/src/ctl.rs @@ -0,0 +1,562 @@ +//! `amux ctl` — the verb layer over the control channel ([`crate::ipc`]). +//! +//! Two halves live here: +//! * The **client** ([`ctl_cmd`]): parse `amux ctl …` argv into one JSON +//! request, read `AMUX_CTL` (the endpoint) and `AMUX_PANE` (the caller's agent +//! id) from the environment amux injected, send it, and print the JSON reply. +//! * The **protocol + policy** the server (the run loop) applies: [`parse_request`] +//! turns a request line into a typed [`Request`], and [`evaluate_spawn`] is the +//! *pure* guard — allowlist + depth cap — so the safety rules are unit-tested +//! without a pty or a running amux. +//! +//! C1 surface: `spawn` (open a visible worker pane) and `list` (the org chart). +//! `send` / `status` / `kill` arrive in C2–C3. + +use std::process::ExitCode; + +use json::{Number, Value}; + +use crate::bind; + +/// Environment variable naming the control endpoint, injected into every pane a +/// `--allow-ctl` amux spawns. Absent → `amux ctl` refuses (not in a ctl session). +pub const ENV_ADDRESS: &str = "AMUX_CTL"; +/// Environment variable carrying the *caller* pane's agent id, so the server can +/// attribute a spawn to its parent (spawn tree + depth). Injected per pane. +pub const ENV_PANE: &str = "AMUX_PANE"; + +/// The default spawn-depth ceiling: the recursion circuit-breaker (design §5). +/// Generous — a real hierarchy is CEO→lead→IC (depth 2–3); this only stops a +/// runaway self-spawning agent. Overridable/removable via `--max-depth`. +pub const DEFAULT_MAX_DEPTH: usize = 6; + +/// A parsed control request. +#[derive(Debug, Clone, PartialEq)] +pub struct Request { + /// The agent id of the pane that issued this (from `AMUX_PANE`), if any. + pub caller: Option, + pub cmd: Cmd, +} + +/// The C1 command set. +#[derive(Debug, Clone, PartialEq)] +pub enum Cmd { + /// Open a new worker pane running `argv`, tagged `role`. + Spawn(SpawnReq), + /// Report the spawn tree. + List, +} + +/// A `spawn` request's payload. +#[derive(Debug, Clone, PartialEq)] +pub struct SpawnReq { + pub role: Option, + /// The command to host, e.g. `["claude"]`. Must be non-empty and on the + /// agent allowlist (checked by [`evaluate_spawn`]). + pub argv: Vec, + /// Open in a new window (true, the C1 default) vs. split the caller (later). + pub new_window: bool, +} + +/// Why a spawn was refused. Each maps to a clear reply the caller can act on. +#[derive(Debug, Clone, PartialEq)] +pub enum SpawnDenied { + EmptyCommand, + NotAllowed(String), + DepthExceeded { attempted: usize, max: usize }, +} + +impl SpawnDenied { + pub fn message(&self) -> String { + match self { + SpawnDenied::EmptyCommand => "spawn needs a command (e.g. `-- claude`)".to_string(), + SpawnDenied::NotAllowed(stem) => format!( + "{stem:?} is not on the agent allowlist ({}); ctl spawns agents only", + bind::AGENT_STEMS.join(", ") + ), + SpawnDenied::DepthExceeded { attempted, max } => { + format!("spawn depth {attempted} exceeds --max-depth {max} (recursion guard)") + } + } + } +} + +/// The **pure** spawn guard: is this command allowed to spawn at this depth? +/// `caller_depth` is the depth of the requesting pane (a human/root pane is 0); +/// the new worker would be at `caller_depth + 1`. `extra_allow` extends the +/// built-in agent allowlist ([`bind::AGENT_STEMS`]) with operator-approved stems +/// (the `AMUX_CTL_ALLOW` knob). Returns the new worker's depth on success. +/// Unit-tested in isolation — this is the heart of the safety model. +pub fn evaluate_spawn( + argv: &[String], + caller_depth: usize, + max_depth: usize, + extra_allow: &[String], +) -> Result { + let Some(first) = argv.first() else { + return Err(SpawnDenied::EmptyCommand); + }; + if first.is_empty() { + return Err(SpawnDenied::EmptyCommand); + } + let stem = std::path::Path::new(first) + .file_stem() + .map(|s| s.to_string_lossy().into_owned()) + .unwrap_or_else(|| first.clone()); + let allowed = bind::is_agent_stem(&stem) || extra_allow.iter().any(|s| s == &stem); + if !allowed { + return Err(SpawnDenied::NotAllowed(stem)); + } + let attempted = caller_depth + 1; + if attempted > max_depth { + return Err(SpawnDenied::DepthExceeded { + attempted, + max: max_depth, + }); + } + Ok(attempted) +} + +/// Environment knob (comma-separated) that extends the ctl agent allowlist +/// beyond [`bind::AGENT_STEMS`]. Opt-in on top of `--allow-ctl` and set by the +/// human who launches amux, so it never weakens the confused-agent guard for a +/// session that did not ask for it. Empty/unset ⇒ agents-only (`{claude}`). +pub const ENV_ALLOW: &str = "AMUX_CTL_ALLOW"; + +/// Read [`ENV_ALLOW`] into the extra-allow list (trimmed, empties dropped). +pub fn extra_allow_from_env() -> Vec { + std::env::var(ENV_ALLOW) + .ok() + .map(|v| { + v.split(',') + .map(str::trim) + .filter(|s| !s.is_empty()) + .map(str::to_string) + .collect() + }) + .unwrap_or_default() +} + +/// Pull amux's own ctl meta-flags off the front of the (already identity- +/// stripped) argument vector, before the hosted command begins — exactly the way +/// [`crate::spawn::parse`] pulls `-n`/`--grid`. Returns `(allow_ctl, max_depth, +/// rest)`, where `rest` is the untouched remainder (grid flags + hosted command). +/// +/// * `--allow-ctl` — opt in to the control channel (off by default). +/// * `--max-depth ` — the recursion guard ceiling (default +/// [`DEFAULT_MAX_DEPTH`]); `0` means unlimited (the guard is removed). +/// +/// Parsing stops at the first non-flag token, so a `--max-depth` the hosted +/// program takes is never eaten. A bad `--max-depth` value is a clear error. +pub fn parse_flags(args: &[String]) -> Result<(bool, usize, Vec), String> { + let mut allow = false; + let mut max_depth = DEFAULT_MAX_DEPTH; + let mut i = 0; + while i < args.len() { + match args[i].as_str() { + "--allow-ctl" => { + allow = true; + i += 1; + } + "--max-depth" => { + let val = args + .get(i + 1) + .ok_or_else(|| "--max-depth needs a value (0 = unlimited)".to_string())?; + max_depth = parse_depth(val)?; + i += 2; + } + s if s.starts_with("--max-depth=") => { + max_depth = parse_depth(&s["--max-depth=".len()..])?; + i += 1; + } + _ => return Ok((allow, effective_depth(max_depth), args[i..].to_vec())), + } + } + Ok((allow, effective_depth(max_depth), Vec::new())) +} + +fn parse_depth(val: &str) -> Result { + val.parse::() + .map_err(|_| "--max-depth must be a non-negative integer (0 = unlimited)".to_string()) +} + +/// `0` from the user means "no limit"; represent it as the max so the guard in +/// [`evaluate_spawn`] can never trip. +fn effective_depth(d: usize) -> usize { + if d == 0 { + usize::MAX + } else { + d + } +} + +/// Parse a request line (one JSON object) into a [`Request`], or a clear error. +pub fn parse_request(line: &str) -> Result { + let v = json::parse(line).map_err(|e| format!("bad request json: {e}"))?; + let caller = v.get("caller").and_then(Value::as_i64).map(|n| n as usize); + let cmd = match v.get("cmd").and_then(Value::as_str) { + Some("spawn") => { + let argv = match v.get("argv").and_then(Value::as_array) { + Some(items) => items + .iter() + .map(|it| it.as_str().map(str::to_string)) + .collect::>>() + .ok_or_else(|| "argv must be an array of strings".to_string())?, + None => return Err("spawn needs an argv array".to_string()), + }; + let role = v.get("role").and_then(Value::as_str).map(str::to_string); + let new_window = v.get("window").and_then(Value::as_bool).unwrap_or(true); + Cmd::Spawn(SpawnReq { + role, + argv, + new_window, + }) + } + Some("list") => Cmd::List, + Some(other) => return Err(format!("unknown command {other:?}")), + None => return Err("request has no \"cmd\"".to_string()), + }; + Ok(Request { caller, cmd }) +} + +// ---- reply builders (compact JSON via `Value`'s Display) ------------------ + +fn obj(pairs: Vec<(&str, Value)>) -> Value { + Value::Object(pairs.into_iter().map(|(k, v)| (k.to_string(), v)).collect()) +} + +fn s(v: &str) -> Value { + Value::String(v.to_string()) +} + +fn i(n: usize) -> Value { + Value::Number(Number::Int(n as i64)) +} + +/// `{"ok":false,"err":""}` +pub fn reply_err(msg: &str) -> String { + obj(vec![("ok", Value::Bool(false)), ("err", s(msg))]).to_string() +} + +/// `{"ok":true,"pane":,"role":,"session":}` +pub fn reply_spawned(pane: usize, role: Option<&str>, session: Option<&str>) -> String { + obj(vec![ + ("ok", Value::Bool(true)), + ("pane", i(pane)), + ("role", role.map(s).unwrap_or(Value::Null)), + ("session", session.map(s).unwrap_or(Value::Null)), + ]) + .to_string() +} + +/// One node in the org chart, as the run loop knows it. +pub struct TreeNode<'a> { + pub id: usize, + pub parent: Option, + pub role: Option<&'a str>, + pub title: &'a str, + pub depth: usize, + pub status: Option<&'a str>, +} + +/// `{"ok":true,"tree":[{id,parent,role,title,depth,status}, …]}` +pub fn reply_list(nodes: &[TreeNode]) -> String { + let arr = nodes + .iter() + .map(|n| { + obj(vec![ + ("id", i(n.id)), + ("parent", n.parent.map(i).unwrap_or(Value::Null)), + ("role", n.role.map(s).unwrap_or(Value::Null)), + ("title", s(n.title)), + ("depth", i(n.depth)), + ("status", n.status.map(s).unwrap_or(Value::Null)), + ]) + }) + .collect(); + obj(vec![("ok", Value::Bool(true)), ("tree", Value::Array(arr))]).to_string() +} + +// ---- client -------------------------------------------------------------- + +/// `amux ctl …`: build a request from argv, send it to `AMUX_CTL`, print +/// the reply. Exit code reflects the reply's `ok`. +pub fn ctl_cmd(args: &[String]) -> ExitCode { + let Some(address) = std::env::var(ENV_ADDRESS).ok().filter(|a| !a.is_empty()) else { + eprintln!( + "amux ctl: not inside a ctl-enabled amux session ({ENV_ADDRESS} unset).\n\ + Start amux with `--allow-ctl` and run `amux ctl` from one of its panes." + ); + return ExitCode::FAILURE; + }; + let caller = std::env::var(ENV_PANE) + .ok() + .and_then(|p| p.parse::().ok()); + + let request = match build_request(args, caller) { + Ok(r) => r, + Err(msg) => { + eprintln!("amux ctl: {msg}"); + eprintln!("usage: amux ctl spawn [--role R] [-- ] | amux ctl list"); + return ExitCode::FAILURE; + } + }; + + match crate::ipc::request(&address, &request) { + Ok(reply) => { + println!("{reply}"); + let ok = json::parse(&reply) + .ok() + .and_then(|v| v.get("ok").and_then(json::Value::as_bool)) + .unwrap_or(false); + if ok { + ExitCode::SUCCESS + } else { + ExitCode::FAILURE + } + } + Err(e) => { + eprintln!("amux ctl: {e}"); + ExitCode::FAILURE + } + } +} + +/// Turn `amux ctl` argv (after the `ctl` word) + caller id into a JSON request +/// line. Pure and testable. Grammar: +/// `spawn [--role R] [--here | --window] [-- ]` +/// `list` +pub fn build_request(args: &[String], caller: Option) -> Result { + let mut pairs: Vec<(&str, Value)> = Vec::new(); + if let Some(c) = caller { + pairs.push(("caller", Value::Number(Number::Int(c as i64)))); + } + match args.first().map(String::as_str) { + Some("spawn") => { + pairs.push(("cmd", s("spawn"))); + let mut role: Option = None; + let mut new_window = true; + let mut argv: Vec = Vec::new(); + let mut i = 1; + while i < args.len() { + match args[i].as_str() { + "--role" => { + role = Some( + args.get(i + 1) + .cloned() + .ok_or_else(|| "--role needs a value".to_string())?, + ); + i += 2; + } + "--here" => { + new_window = false; + i += 1; + } + "--window" => { + new_window = true; + i += 1; + } + "--" => { + argv = args[i + 1..].to_vec(); + break; + } + other => return Err(format!("unexpected argument {other:?} (use `-- `)")), + } + } + if argv.is_empty() { + return Err("spawn needs a command after `--` (e.g. `-- claude`)".to_string()); + } + if let Some(r) = role { + pairs.push(("role", Value::String(r))); + } + pairs.push(("window", Value::Bool(new_window))); + pairs.push(( + "argv", + Value::Array(argv.into_iter().map(Value::String).collect()), + )); + } + Some("list") => { + pairs.push(("cmd", s("list"))); + } + Some(other) => return Err(format!("unknown subcommand {other:?}")), + None => return Err("needs a subcommand: spawn | list".to_string()), + } + Ok(obj(pairs).to_string()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn v(parts: &[&str]) -> Vec { + parts.iter().map(|s| s.to_string()).collect() + } + + #[test] + fn allowlisted_agent_at_shallow_depth_is_allowed() { + assert_eq!(evaluate_spawn(&v(&["claude"]), 0, 6, &[]), Ok(1)); + assert_eq!( + evaluate_spawn(&v(&["claude", "--continue"]), 2, 6, &[]), + Ok(3) + ); + } + + #[test] + fn non_allowlisted_command_is_refused() { + assert_eq!( + evaluate_spawn(&v(&["rm", "-rf", "/"]), 0, 6, &[]), + Err(SpawnDenied::NotAllowed("rm".to_string())) + ); + } + + #[test] + fn extra_allow_admits_an_operator_approved_stem() { + // `sh` is not a built-in agent, but AMUX_CTL_ALLOW=sh lets it spawn. + assert_eq!( + evaluate_spawn(&v(&["sh"]), 0, 6, &["sh".to_string()]), + Ok(1) + ); + // …and only the approved one; a different command is still refused. + assert_eq!( + evaluate_spawn(&v(&["bash"]), 0, 6, &["sh".to_string()]), + Err(SpawnDenied::NotAllowed("bash".to_string())) + ); + } + + #[test] + fn path_qualified_agent_matches_by_stem() { + // A full path to claude still resolves to the "claude" stem. + assert_eq!(evaluate_spawn(&v(&["/usr/bin/claude"]), 0, 6, &[]), Ok(1)); + } + + #[test] + fn depth_over_the_cap_is_refused() { + assert_eq!( + evaluate_spawn(&v(&["claude"]), 6, 6, &[]), + Err(SpawnDenied::DepthExceeded { + attempted: 7, + max: 6 + }) + ); + } + + #[test] + fn empty_command_is_refused() { + assert_eq!( + evaluate_spawn(&[], 0, 6, &[]), + Err(SpawnDenied::EmptyCommand) + ); + } + + #[test] + fn build_spawn_request_roundtrips_through_parse() { + let line = + build_request(&v(&["spawn", "--role", "dev_1", "--", "claude"]), Some(0)).unwrap(); + let req = parse_request(&line).unwrap(); + assert_eq!(req.caller, Some(0)); + match req.cmd { + Cmd::Spawn(sp) => { + assert_eq!(sp.role.as_deref(), Some("dev_1")); + assert_eq!(sp.argv, v(&["claude"])); + assert!(sp.new_window); + } + _ => panic!("expected spawn"), + } + } + + #[test] + fn build_here_sets_window_false() { + let line = build_request(&v(&["spawn", "--here", "--", "claude"]), None).unwrap(); + let req = parse_request(&line).unwrap(); + match req.cmd { + Cmd::Spawn(sp) => assert!(!sp.new_window), + _ => panic!("expected spawn"), + } + } + + #[test] + fn build_list_request() { + let line = build_request(&v(&["list"]), Some(3)).unwrap(); + let req = parse_request(&line).unwrap(); + assert_eq!(req.caller, Some(3)); + assert_eq!(req.cmd, Cmd::List); + } + + #[test] + fn spawn_without_command_is_an_error() { + let err = build_request(&v(&["spawn", "--role", "x"]), None).unwrap_err(); + assert!(err.contains("needs a command"), "{err}"); + } + + #[test] + fn parse_rejects_unknown_command() { + let err = parse_request(r#"{"cmd":"frobnicate"}"#).unwrap_err(); + assert!(err.contains("unknown command"), "{err}"); + } + + #[test] + fn flags_default_off_and_pass_command_through() { + let (allow, depth, rest) = parse_flags(&v(&["claude", "--continue"])).unwrap(); + assert!(!allow); + assert_eq!(depth, DEFAULT_MAX_DEPTH); + assert_eq!(rest, v(&["claude", "--continue"])); + } + + #[test] + fn flags_allow_ctl_and_max_depth() { + let (allow, depth, rest) = + parse_flags(&v(&["--allow-ctl", "--max-depth", "3", "claude"])).unwrap(); + assert!(allow); + assert_eq!(depth, 3); + assert_eq!(rest, v(&["claude"])); + } + + #[test] + fn flags_max_depth_zero_is_unlimited() { + let (_, depth, _) = + parse_flags(&v(&["--allow-ctl", "--max-depth", "0", "claude"])).unwrap(); + assert_eq!(depth, usize::MAX); + } + + #[test] + fn flags_stop_at_command_so_child_keeps_its_flags() { + // A `--max-depth` after the command belongs to the child, untouched. + let (allow, _, rest) = + parse_flags(&v(&["--allow-ctl", "claude", "--max-depth", "9"])).unwrap(); + assert!(allow); + assert_eq!(rest, v(&["claude", "--max-depth", "9"])); + } + + #[test] + fn flags_bad_max_depth_errors() { + let err = parse_flags(&v(&["--max-depth", "lots"])).unwrap_err(); + assert!(err.contains("--max-depth"), "{err}"); + } + + #[test] + fn reply_builders_are_valid_json() { + let spawned = reply_spawned(3, Some("dev_1"), Some("abc-123")); + let v = json::parse(&spawned).unwrap(); + assert_eq!(v.get("ok").and_then(Value::as_bool), Some(true)); + assert_eq!(v.get("pane").and_then(Value::as_i64), Some(3)); + + let nodes = [TreeNode { + id: 0, + parent: None, + role: Some("ceo"), + title: "claude", + depth: 0, + status: Some("working"), + }]; + let listed = reply_list(&nodes); + let v = json::parse(&listed).unwrap(); + assert_eq!( + v.get("tree").and_then(Value::as_array).map(<[_]>::len), + Some(1) + ); + + let err = reply_err("nope"); + let v = json::parse(&err).unwrap(); + assert_eq!(v.get("ok").and_then(Value::as_bool), Some(false)); + } +} diff --git a/src/ipc.rs b/src/ipc.rs new file mode 100644 index 0000000..f68f974 --- /dev/null +++ b/src/ipc.rs @@ -0,0 +1,550 @@ +//! The `ctl` control channel: a tiny, zero-dependency, **local** request/reply +//! transport — a Windows named pipe or a unix-domain socket, chosen at compile +//! time. It carries one JSON request line and one JSON reply line per client +//! connection; amux ("the server") owns the endpoint and drains it from its run +//! loop without blocking, and each `amux ctl ` invocation is a short-lived +//! client ([`request`]). +//! +//! The API is deliberately small: +//! * [`default_address`] — the per-process endpoint amux binds and injects as +//! `AMUX_CTL` into every pane it spawns. +//! * [`Listener::bind`] / [`Listener::poll`] / [`Listener::respond`] — the +//! server side. `poll` is **non-blocking**: it returns `Ok(None)` when no +//! request is ready, so it fits the 15 ms run-loop tick with no thread (the +//! C0 spike proved this on Windows via `PIPE_NOWAIT`; unix uses a +//! non-blocking `accept`). One request is outstanding at a time — the caller +//! must [`respond`](Listener::respond) before the next `poll`. +//! * [`request`] — the client: connect, send one line, read one reply line. +//! +//! Framing is one `\n`-terminated line each way. This is a control plane for a +//! handful of agents, not a high-throughput bus; serving one request per tick is +//! plenty and keeps the state machine trivial. + +use std::io; + +/// The endpoint amux binds for this process, unique per pid so two amux +/// instances never collide. Injected as `AMUX_CTL` into spawned panes. +pub fn default_address() -> String { + sys::default_address(std::process::id()) +} + +/// The server endpoint. Owns the OS handle/socket; one request is outstanding at +/// a time (poll → respond → poll …). +pub struct Listener { + sys: sys::Listener, +} + +impl Listener { + /// Bind the control endpoint at `addr`. On unix a stale socket file at the + /// path is removed first; the socket file is unlinked on drop. + pub fn bind(addr: &str) -> io::Result { + Ok(Listener { + sys: sys::Listener::bind(addr)?, + }) + } + + /// Non-blocking: the next complete request line if one has arrived, else + /// `Ok(None)`. After a `Some(_)` the caller MUST call [`respond`](Listener::respond) + /// before polling again. + pub fn poll(&mut self) -> io::Result> { + self.sys.poll() + } + + /// Send the reply line for the request the last [`poll`](Listener::poll) + /// returned, and release the connection so the endpoint can accept the next + /// client. A trailing newline is added if absent. + pub fn respond(&mut self, reply: &str) -> io::Result<()> { + self.sys.respond(reply) + } +} + +/// Client: connect to `addr`, send `req` (one line), return the reply line +/// (newline trimmed). Used by `amux ctl `. +pub fn request(addr: &str, req: &str) -> io::Result { + sys::request(addr, req) +} + +// --------------------------------------------------------------------------- +// Windows: a single named-pipe instance, reused per client. Connection detection +// and reads are non-blocking (`PIPE_NOWAIT`); the reply write flushes before the +// disconnect so the client always sees it. Mirrors the verified C0 spike. +// --------------------------------------------------------------------------- +#[cfg(windows)] +mod sys { + use std::ffi::c_void; + use std::io; + use std::os::windows::ffi::OsStrExt; + use std::thread; + use std::time::Duration; + + type Handle = *mut c_void; + + #[link(name = "kernel32")] + extern "system" { + fn CreateNamedPipeW( + name: *const u16, + open_mode: u32, + pipe_mode: u32, + max_instances: u32, + out_buf: u32, + in_buf: u32, + default_timeout: u32, + sec: *mut c_void, + ) -> Handle; + fn ConnectNamedPipe(handle: Handle, overlapped: *mut c_void) -> i32; + fn DisconnectNamedPipe(handle: Handle) -> i32; + fn CreateFileW( + name: *const u16, + access: u32, + share: u32, + sec: *mut c_void, + disposition: u32, + flags: u32, + template: Handle, + ) -> Handle; + fn ReadFile( + handle: Handle, + buf: *mut u8, + len: u32, + read: *mut u32, + overlapped: *mut c_void, + ) -> i32; + fn WriteFile( + handle: Handle, + buf: *const u8, + len: u32, + written: *mut u32, + overlapped: *mut c_void, + ) -> i32; + fn SetNamedPipeHandleState( + handle: Handle, + mode: *mut u32, + max_collect: *mut u32, + collect_timeout: *mut u32, + ) -> i32; + fn FlushFileBuffers(handle: Handle) -> i32; + fn CloseHandle(handle: Handle) -> i32; + } + + const PIPE_ACCESS_DUPLEX: u32 = 0x0000_0003; + const PIPE_TYPE_MESSAGE: u32 = 0x0000_0004; + const PIPE_READMODE_MESSAGE: u32 = 0x0000_0002; + const PIPE_NOWAIT: u32 = 0x0000_0001; + const PIPE_UNLIMITED_INSTANCES: u32 = 255; + const GENERIC_READ: u32 = 0x8000_0000; + const GENERIC_WRITE: u32 = 0x4000_0000; + const OPEN_EXISTING: u32 = 3; + const ERROR_FILE_NOT_FOUND: i32 = 2; + const ERROR_PIPE_BUSY: i32 = 231; + const ERROR_NO_DATA: i32 = 232; + const ERROR_PIPE_NOT_CONNECTED: i32 = 233; + const ERROR_BROKEN_PIPE: i32 = 109; + const ERROR_PIPE_CONNECTED: i32 = 535; + const INVALID_HANDLE_VALUE: Handle = usize::MAX as Handle; + + fn wide(s: &str) -> Vec { + std::ffi::OsStr::new(s).encode_wide().chain([0]).collect() + } + + fn last() -> i32 { + io::Error::last_os_error().raw_os_error().unwrap_or(0) + } + + pub fn default_address(pid: u32) -> String { + format!(r"\\.\pipe\amux-ctl-{pid}") + } + + pub struct Listener { + handle: Handle, + connected: bool, + } + + // The handle is owned solely by this Listener, used only from the run loop. + unsafe impl Send for Listener {} + + impl Listener { + fn create_instance(addr: &[u16]) -> io::Result { + let h = unsafe { + CreateNamedPipeW( + addr.as_ptr(), + PIPE_ACCESS_DUPLEX, + PIPE_TYPE_MESSAGE | PIPE_READMODE_MESSAGE | PIPE_NOWAIT, + PIPE_UNLIMITED_INSTANCES, + 8192, + 8192, + 0, + std::ptr::null_mut(), + ) + }; + if h == INVALID_HANDLE_VALUE { + return Err(io::Error::last_os_error()); + } + Ok(h) + } + + pub fn bind(addr: &str) -> io::Result { + let addr = wide(addr); + let handle = Self::create_instance(&addr)?; + Ok(Listener { + handle, + connected: false, + }) + } + + pub fn poll(&mut self) -> io::Result> { + if !self.connected { + // NOWAIT: returns immediately. A waiting client shows up as + // ERROR_PIPE_CONNECTED; ERROR_PIPE_LISTENING / no-data means + // "still nobody", which is the common idle path. + let r = unsafe { ConnectNamedPipe(self.handle, std::ptr::null_mut()) }; + if r != 0 { + self.connected = true; + } else { + let e = last(); + if e == ERROR_PIPE_CONNECTED { + self.connected = true; + } else { + return Ok(None); + } + } + } + // Connected: try one non-blocking read. + let mut buf = [0u8; 8192]; + let mut n = 0u32; + let ok = unsafe { + ReadFile( + self.handle, + buf.as_mut_ptr(), + buf.len() as u32, + &mut n, + std::ptr::null_mut(), + ) + }; + if ok != 0 && n > 0 { + let line = String::from_utf8_lossy(&buf[..n as usize]) + .trim_end_matches(['\r', '\n']) + .to_string(); + Ok(Some(line)) + } else { + let e = last(); + if e == ERROR_BROKEN_PIPE || e == ERROR_PIPE_NOT_CONNECTED { + // Client vanished before sending; recycle the instance. + self.recycle(); + } + Ok(None) // ERROR_NO_DATA: connected but nothing yet — try next tick. + } + } + + pub fn respond(&mut self, reply: &str) -> io::Result<()> { + let mut line = reply.to_string(); + if !line.ends_with('\n') { + line.push('\n'); + } + let bytes = line.as_bytes(); + let mut written = 0u32; + let ok = unsafe { + WriteFile( + self.handle, + bytes.as_ptr(), + bytes.len() as u32, + &mut written, + std::ptr::null_mut(), + ) + }; + unsafe { + FlushFileBuffers(self.handle); + } + self.recycle(); + if ok == 0 { + return Err(io::Error::last_os_error()); + } + Ok(()) + } + + /// Drop the current client and re-arm the same instance to listen again. + fn recycle(&mut self) { + unsafe { + DisconnectNamedPipe(self.handle); + } + self.connected = false; + } + } + + impl Drop for Listener { + fn drop(&mut self) { + unsafe { + DisconnectNamedPipe(self.handle); + CloseHandle(self.handle); + } + } + } + + pub fn request(addr: &str, req: &str) -> io::Result { + let name = wide(addr); + // The server serves one client at a time; a brief busy/absent window + // between clients is normal — retry a bounded number of times. + let mut h = INVALID_HANDLE_VALUE; + for _ in 0..100 { + h = unsafe { + CreateFileW( + name.as_ptr(), + GENERIC_READ | GENERIC_WRITE, + 0, + std::ptr::null_mut(), + OPEN_EXISTING, + 0, + std::ptr::null_mut(), + ) + }; + if h != INVALID_HANDLE_VALUE { + break; + } + let e = last(); + if e == ERROR_PIPE_BUSY || e == ERROR_FILE_NOT_FOUND { + thread::sleep(Duration::from_millis(5)); + continue; + } + return Err(io::Error::from_raw_os_error(e)); + } + if h == INVALID_HANDLE_VALUE { + return Err(io::Error::new( + io::ErrorKind::NotConnected, + "amux control channel did not answer", + )); + } + + let mut mode = PIPE_READMODE_MESSAGE; + unsafe { + SetNamedPipeHandleState(h, &mut mode, std::ptr::null_mut(), std::ptr::null_mut()); + } + + let mut line = req.to_string(); + if !line.ends_with('\n') { + line.push('\n'); + } + let bytes = line.as_bytes(); + let mut written = 0u32; + let ok = unsafe { + WriteFile( + h, + bytes.as_ptr(), + bytes.len() as u32, + &mut written, + std::ptr::null_mut(), + ) + }; + if ok == 0 { + let e = last(); + unsafe { CloseHandle(h) }; + return Err(io::Error::from_raw_os_error(e)); + } + + // Read the reply, tolerating the small window before the server writes. + let mut buf = [0u8; 8192]; + let mut out = String::new(); + for _ in 0..600 { + let mut n = 0u32; + let ok = unsafe { + ReadFile( + h, + buf.as_mut_ptr(), + buf.len() as u32, + &mut n, + std::ptr::null_mut(), + ) + }; + if ok != 0 && n > 0 { + out = String::from_utf8_lossy(&buf[..n as usize]) + .trim_end_matches(['\r', '\n']) + .to_string(); + break; + } + let e = last(); + if e == ERROR_NO_DATA { + thread::sleep(Duration::from_millis(5)); + continue; + } + unsafe { CloseHandle(h) }; + return Err(io::Error::from_raw_os_error(e)); + } + unsafe { CloseHandle(h) }; + if out.is_empty() { + return Err(io::Error::new( + io::ErrorKind::TimedOut, + "no reply from amux control channel", + )); + } + Ok(out) + } +} + +// --------------------------------------------------------------------------- +// Unix: a non-blocking `UnixListener`; each client is one short-lived stream. +// --------------------------------------------------------------------------- +#[cfg(unix)] +mod sys { + use std::io::{self, Read, Write}; + use std::os::unix::net::{UnixListener, UnixStream}; + use std::path::PathBuf; + use std::time::Duration; + + pub fn default_address(pid: u32) -> String { + std::env::temp_dir() + .join(format!("amux-ctl-{pid}.sock")) + .to_string_lossy() + .into_owned() + } + + pub struct Listener { + listener: UnixListener, + path: PathBuf, + cur: Option, + } + + impl Listener { + pub fn bind(addr: &str) -> io::Result { + let path = PathBuf::from(addr); + // A stale socket file from a crashed prior run would block the bind. + let _ = std::fs::remove_file(&path); + let listener = UnixListener::bind(&path)?; + listener.set_nonblocking(true)?; + Ok(Listener { + listener, + path, + cur: None, + }) + } + + pub fn poll(&mut self) -> io::Result> { + match self.listener.accept() { + Ok((mut stream, _)) => { + // The client sends immediately then waits; a short blocking + // read of one line is safe and keeps framing simple. + stream.set_nonblocking(false)?; + stream.set_read_timeout(Some(Duration::from_secs(2)))?; + let line = read_line(&mut stream)?; + self.cur = Some(stream); + Ok(Some(line)) + } + Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => Ok(None), + Err(e) => Err(e), + } + } + + pub fn respond(&mut self, reply: &str) -> io::Result<()> { + if let Some(mut stream) = self.cur.take() { + let mut line = reply.to_string(); + if !line.ends_with('\n') { + line.push('\n'); + } + stream.write_all(line.as_bytes())?; + stream.flush()?; + } + Ok(()) + } + } + + impl Drop for Listener { + fn drop(&mut self) { + let _ = std::fs::remove_file(&self.path); + } + } + + fn read_line(stream: &mut UnixStream) -> io::Result { + let mut buf = Vec::with_capacity(256); + let mut byte = [0u8; 1]; + loop { + match stream.read(&mut byte) { + Ok(0) => break, + Ok(_) => { + if byte[0] == b'\n' { + break; + } + buf.push(byte[0]); + } + Err(e) => return Err(e), + } + if buf.len() > 65536 { + break; // never grow unbounded on a malformed client + } + } + Ok(String::from_utf8_lossy(&buf) + .trim_end_matches('\r') + .to_string()) + } + + pub fn request(addr: &str, req: &str) -> io::Result { + let mut stream = UnixStream::connect(addr)?; + stream.set_read_timeout(Some(Duration::from_secs(5)))?; + let mut line = req.to_string(); + if !line.ends_with('\n') { + line.push('\n'); + } + stream.write_all(line.as_bytes())?; + stream.flush()?; + read_line(&mut stream) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::thread; + use std::time::Duration; + + /// A unique endpoint per test (tests run in parallel; a shared per-pid name + /// would collide). Platform-appropriate: a named pipe on Windows, a temp + /// socket path on unix. + fn test_addr(nonce: u32) -> String { + let pid = std::process::id(); + #[cfg(windows)] + { + format!(r"\\.\pipe\amux-ctl-test-{pid}-{nonce}") + } + #[cfg(unix)] + { + std::env::temp_dir() + .join(format!("amux-ctl-test-{pid}-{nonce}.sock")) + .to_string_lossy() + .into_owned() + } + } + + /// Full roundtrip over the real transport: bind a listener, dial it from a + /// client thread, and confirm the server sees the request and the client + /// sees the reply. This is the C0 spike promoted to a permanent test. + #[test] + fn roundtrip_request_reply() { + let addr = test_addr(1); + let mut server = Listener::bind(&addr).expect("bind"); + let addr2 = addr.clone(); + + let client = thread::spawn(move || request(&addr2, r#"{"cmd":"list"}"#).expect("request")); + + // Serve one request within a bounded number of non-blocking ticks. + let mut reply_sent = false; + for _ in 0..400 { + if let Some(req) = server.poll().expect("poll") { + assert!(req.contains("\"list\""), "server saw: {req}"); + server.respond(r#"{"ok":true,"tree":[]}"#).expect("respond"); + reply_sent = true; + break; + } + thread::sleep(Duration::from_millis(5)); + } + assert!(reply_sent, "server never saw the request"); + + let reply = client.join().expect("client thread"); + assert!(reply.contains("\"ok\":true"), "client saw: {reply}"); + } + + #[test] + fn poll_is_non_blocking_when_idle() { + let addr = test_addr(2); + let mut server = Listener::bind(&addr).expect("bind"); + // With no client, poll must return immediately with None, not hang. + for _ in 0..10 { + assert_eq!(server.poll().expect("poll"), None); + } + } +} diff --git a/src/lib.rs b/src/lib.rs index 297bc38..77d28cd 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -17,10 +17,12 @@ pub mod bar; pub mod bind; +pub mod ctl; pub mod filter; pub mod fleet; pub mod identity; pub mod input; +pub mod ipc; pub mod layout; pub mod resolve; pub mod spawn; diff --git a/src/main.rs b/src/main.rs index 2f82f75..be6067c 100644 --- a/src/main.rs +++ b/src/main.rs @@ -27,8 +27,25 @@ use amux::layout::{self, Rect, Tree}; use amux::tile::{compose, AgentMark, PaneState, PaneView}; use std::io::Write; use std::process::ExitCode; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::OnceLock; use std::time::{Duration, Instant}; +/// A process-global, monotonic **agent id** stamped on every pane amux hosts — +/// the stable key of the ctl spawn tree (`Pane.id` is only unique within a +/// window; this is unique across the whole run). Incremented once per spawn. +static NEXT_AGENT: AtomicUsize = AtomicUsize::new(0); + +/// The ctl channel address, set once at startup iff `--allow-ctl` was given. +/// The spawn path reads it to inject `AMUX_CTL`/`AMUX_PANE` into each pane so an +/// agent inside can drive `amux ctl`. `None` (unset) ⇒ ctl is off and every +/// spawn is byte-identical to pre-ctl amux. +static CTL_ADDRESS: OnceLock = OnceLock::new(); + +fn next_agent_id() -> usize { + NEXT_AGENT.fetch_add(1, Ordering::Relaxed) +} + /// One hosted terminal: a pty, its emulator (for tiled compositing), its /// passthrough filter (for the passthrough / zoom path), and bar metadata. Each /// pane has a stable `id` the window's split tree refers to. @@ -50,6 +67,20 @@ struct Pane { /// alone; the resolved values live only for the spawn call and are dropped /// immediately. Surfaced in the chrome as a `·` tag. identity: Option, + /// Process-global agent id (see [`NEXT_AGENT`]): the ctl spawn-tree key, + /// stable across windows. Injected into the pane as `AMUX_PANE` so an agent + /// inside can attribute its own `ctl spawn` calls. + agent_id: usize, + /// The ctl role label this pane was spawned under (`dev_1`), if any. `None` + /// for the human's own panes and shells. + role: Option, + /// The agent id of the pane whose `ctl spawn` created this one. `None` for a + /// root pane the human opened. + parent: Option, + /// Depth in the spawn tree: 0 for a root/human pane, parent.depth + 1 for a + /// ctl-spawned worker. The `--max-depth` recursion guard is checked against + /// this. + depth: usize, } /// One window: a split tree over a set of panes, plus a zoom flag. Windows are @@ -91,6 +122,12 @@ fn main() -> ExitCode { if args.first().map(String::as_str) == Some("fleet") { return fleet_cmd(&args[1..]); } + // `amux ctl …` is the control-channel client: it connects to the running + // amux's `AMUX_CTL` endpoint, so it is a command family, not a hosted + // program — dispatch it before flag parsing too. + if args.first().map(String::as_str) == Some("ctl") { + return amux::ctl::ctl_cmd(&args[1..]); + } // amux's own `--identity ` / `-I ` is stripped off the front, // before the hosted command begins; it tags the initial agent pane and is // inherited by every split/new pane (stored, re-resolved per spawn). Only @@ -100,13 +137,25 @@ fn main() -> ExitCode { let (identity, rest) = amux::identity::parse(&args); if rest.first().map(String::as_str) == Some("--help") { eprintln!( - "usage: amux [--identity ] [-n | --grid x] [command [args...]] (Ctrl+A ? in the bar shows keys)" + "usage: amux [--identity ] [--allow-ctl [--max-depth ]] [-n | --grid x] [command [args...]]\n\ + \x20 amux ctl spawn [--role R] [-- ] | amux ctl list (inside a --allow-ctl session)\n\ + \x20 (Ctrl+A ? in the bar shows keys)" ); return ExitCode::SUCCESS; } if rest.first().map(String::as_str) == Some("--stdin-probe") { return stdin_probe(); } + // amux's own ctl meta-flags (`--allow-ctl` / `--max-depth `) are stripped + // next — after `--identity`, before mass-spawn flags and the hosted command. + // A bad `--max-depth` is a startup error, never a silent fallback. + let (allow_ctl, max_depth, rest) = match amux::ctl::parse_flags(&rest) { + Ok(triple) => triple, + Err(msg) => { + eprintln!("amux: {msg}"); + return ExitCode::FAILURE; + } + }; // amux's own mass-spawn flags (`-n ` / `--grid x`) are stripped off // the front, after `--identity`, before the hosted command. A bad value is a // startup error, surfaced on stderr — never a silent fallback to one pane. @@ -129,7 +178,15 @@ fn main() -> ExitCode { return ExitCode::FAILURE; } }; - run(&mut term, &command, identity.as_deref(), grid, None) + run( + &mut term, + &command, + identity.as_deref(), + grid, + None, + allow_ctl, + max_depth, + ) } fn run( @@ -143,6 +200,12 @@ fn run( // still drive later `Ctrl+A c` new panes and splits (which host `command` // under `identity`), so a fleet's new panes open a shell as a scratch pane. initial_window: Option, + // ctl control plane (§ amux-ctl-control-plane): when `allow_ctl`, amux binds + // a per-process control endpoint and injects its address into every pane, so + // an agent inside a pane can `amux ctl spawn/list`. `max_depth` is the + // recursion guard (usize::MAX == unlimited). Off ⇒ pre-ctl behavior verbatim. + allow_ctl: bool, + max_depth: usize, ) -> ExitCode { let mut out = std::io::stdout(); let (mut rows, mut cols) = term.size().unwrap_or((24, 80)); @@ -158,6 +221,31 @@ fn run( // bar and the pane spawns *without* the credential — never silently, and // never unauthenticated-without-saying-so (§7). let mut flash: Option<(String, Instant)> = None; + // ctl control channel (opt-in). Bind the endpoint and publish its address to + // the spawn path *before* the first pane is spawned, so every pane — the + // initial one included — is born with `AMUX_CTL`/`AMUX_PANE` in its + // environment. A bind failure is non-fatal: amux still runs, just without + // ctl, and says so in the bar (never silently unavailable). + let mut ctl_listener: Option = None; + // Operator-approved extra allowlist stems (AMUX_CTL_ALLOW); empty ⇒ the + // built-in agents-only guard. Read once at startup. + let ctl_extra_allow = if allow_ctl { + amux::ctl::extra_allow_from_env() + } else { + Vec::new() + }; + if allow_ctl { + let addr = amux::ipc::default_address(); + match amux::ipc::Listener::bind(&addr) { + Ok(l) => { + let _ = CTL_ADDRESS.set(addr); + ctl_listener = Some(l); + } + Err(e) => { + flash = Some((format!("ctl channel disabled: {e}"), Instant::now())); + } + } + } // A pre-built window (fleet) is used as-is; otherwise mass-spawn opens one // window of N tiles in a balanced grid, or the 0.1 single-pane path. All // share the same spawn machinery (each pane its own session). @@ -340,6 +428,33 @@ fn run( } } + // 1b. ctl control channel: drain up to a few requests this tick + // (non-blocking; usually zero). Each request is applied as a pane + // operation and answered on the same connection. A spawn appends a + // visible new window, so we force a repaint after any request. + if let Some(listener) = ctl_listener.as_mut() { + for _ in 0..8 { + match listener.poll() { + Ok(Some(line)) => { + let reply = apply_ctl( + &line, + &mut windows, + rows, + cols, + max_depth, + &ctl_extra_allow, + &world, + ); + let _ = listener.respond(&reply); + prev_master = None; + force_repaint = true; + } + Ok(None) => break, + Err(_) => break, + } + } + } + // 2. drain every pane in the active window (all panes are live in // tiled mode); feed the emulator, and in passthrough also write the // focused pane's cleaned bytes straight through. @@ -903,12 +1018,16 @@ fn fleet_up(name: &str) -> ExitCode { // New panes / splits opened later host a shell under the fleet's default // identity — a scratch pane in-role, not another copy of an agent. let scratch = vec![default_shell()]; + // ctl is opt-in via the `amux --allow-ctl` path; the fleet path runs without + // it in C1 (a fleet + live ctl-spawn combination lands later). run( &mut term, &scratch, fleet.identity.as_deref(), None, Some(window), + false, + amux::ctl::DEFAULT_MAX_DEPTH, ) } @@ -980,6 +1099,98 @@ fn spawn_fleet_window( }) } +/// The agsess status label the ctl protocol reports (stable strings the calling +/// agent can match on). +fn status_label(s: agsess::Status) -> &'static str { + match s { + agsess::Status::Working => "working", + agsess::Status::WaitingApproval => "waiting-approval", + agsess::Status::WaitingPrompt => "waiting-prompt", + agsess::Status::Idle => "idle", + } +} + +/// Apply one ctl request against the live window set and return the JSON reply +/// line. This is the server side of the control channel: `list` serializes the +/// spawn tree (agsess statuses folded in), `spawn` creates a visible worker pane +/// after the pure [`amux::ctl::evaluate_spawn`] guard (allowlist + depth) passes. +fn apply_ctl( + line: &str, + windows: &mut Vec, + rows: u16, + cols: u16, + max_depth: usize, + extra_allow: &[String], + world: &agsess::World, +) -> String { + use amux::ctl::{self, Cmd}; + + let req = match ctl::parse_request(line) { + Ok(r) => r, + Err(e) => return ctl::reply_err(&e), + }; + + match req.cmd { + Cmd::List => { + let mut panes: Vec<&Pane> = windows.iter().flat_map(|w| w.panes.iter()).collect(); + panes.sort_by_key(|p| p.agent_id); + let nodes: Vec = panes + .iter() + .map(|p| ctl::TreeNode { + id: p.agent_id, + parent: p.parent, + role: p.role.as_deref(), + title: &p.title, + depth: p.depth, + status: amux::bind::status_for(p.session_id.as_deref(), &world.sessions) + .map(status_label), + }) + .collect(); + ctl::reply_list(&nodes) + } + Cmd::Spawn(sp) => { + if !sp.new_window { + return ctl::reply_err( + "`--here` (split the caller) lands in C2; C1 opens a new window — rerun without --here", + ); + } + // The caller's depth (0 if the caller is unknown — e.g. a human pane + // amux did not spawn): the worker will sit one below it. + let caller_depth = req + .caller + .and_then(|cid| { + windows + .iter() + .flat_map(|w| w.panes.iter()) + .find(|p| p.agent_id == cid) + }) + .map(|p| p.depth) + .unwrap_or(0); + match ctl::evaluate_spawn(&sp.argv, caller_depth, max_depth, extra_allow) { + Err(denied) => ctl::reply_err(&denied.message()), + Ok(new_depth) => { + let mut flash = None; + match spawn_window(&sp.argv, rows, cols, windows.len(), None, &mut flash) { + Ok(mut w) => { + // A new window from ctl spawn is a single pane; stamp + // its spawn-tree fields before it joins the set. + let pane = &mut w.panes[0]; + pane.role = sp.role.clone(); + pane.parent = req.caller; + pane.depth = new_depth; + let agent_id = pane.agent_id; + let session = pane.session_id.clone(); + windows.push(w); + ctl::reply_spawned(agent_id, sp.role.as_deref(), session.as_deref()) + } + Err(e) => ctl::reply_err(&format!("spawn failed: {e}")), + } + } + } + } + } +} + fn spawn_window( command: &[String], rows: u16, @@ -1107,6 +1318,20 @@ fn spawn_pane_full( let r = rows.max(1); let c = cols.max(1); + // Stamp the process-global agent id now — it is both the pane's spawn-tree + // key and the `AMUX_PANE` value injected below, so an agent inside can + // attribute its own `ctl spawn` calls back to this pane. + let agent_id = next_agent_id(); + + // ctl env (non-secret): when the control channel is on, every pane learns + // the endpoint (`AMUX_CTL`) and its own id (`AMUX_PANE`). This is the base + // env; identity secrets (if any) are merged on top for this one spawn. + let mut base_env: Vec<(String, String)> = Vec::new(); + if let Some(addr) = CTL_ADDRESS.get() { + base_env.push((amux::ctl::ENV_ADDRESS.to_string(), addr.clone())); + base_env.push((amux::ctl::ENV_PANE.to_string(), agent_id.to_string())); + } + // Identity injection (path B): only for an agent pane with an identity set. // Decide ONCE so the spawn path and the pane's stored tag can never diverge // — a pane tagged with an identity is exactly a pane spawned with its env. @@ -1124,9 +1349,12 @@ fn spawn_pane_full( let name = identity.expect("wants_env implies Some"); match akey::resolve(name) { Ok(env) => { - // `env` holds secret values; used for this one spawn only, then - // dropped. Deliberately never formatted, logged, or stored. - pty::Pty::spawn_full(&effective[0], &argrefs, r, c, &env, cwd)? + // `env` holds secret values; merged with the (non-secret) ctl + // base env for this one spawn, then dropped. Deliberately never + // formatted, logged, or stored. + let mut merged = base_env.clone(); + merged.extend(env); + pty::Pty::spawn_full(&effective[0], &argrefs, r, c, &merged, cwd)? } Err(e) => { // Name only in the message — `e` is akey's own error text @@ -1136,11 +1364,11 @@ fn spawn_pane_full( format!("identity {name:?} unresolved: {e} — running without it"), Instant::now(), )); - pty::Pty::spawn_full(&effective[0], &argrefs, r, c, &[], cwd)? + pty::Pty::spawn_full(&effective[0], &argrefs, r, c, &base_env, cwd)? } } } else { - pty::Pty::spawn_full(&effective[0], &argrefs, r, c, &[], cwd)? + pty::Pty::spawn_full(&effective[0], &argrefs, r, c, &base_env, cwd)? }; Ok(Pane { id, @@ -1156,6 +1384,12 @@ fn spawn_pane_full( // tag on a pane that was never credentialed. `inject` is the same // decision the spawn used, so tag and env can never disagree. identity: identity.filter(|_| inject).map(str::to_string), + agent_id, + // Spawn-tree fields default to "root pane the human opened"; the ctl + // spawn handler overrides role/parent/depth for a ctl-created worker. + role: None, + parent: None, + depth: 0, }) } diff --git a/tests/amux.rs b/tests/amux.rs index da5c9df..4226aab 100644 --- a/tests/amux.rs +++ b/tests/amux.rs @@ -868,3 +868,87 @@ fn double_prefix_reaches_the_child() { p.write(b"\x01q").unwrap(); assert_eq!(wait_exit(&mut p, 15), 0); } + +// --- end to end: the ctl control channel (C1) ------------------------------- + +/// Spawn `amux --allow-ctl ` in a pty, with `AMUX_CTL_ALLOW` set so the +/// harmless shell counts as a spawnable worker, and return the pty once the bar +/// (pane 1) is up. The hosted shell inherits `AMUX_CTL`/`AMUX_PANE`, so a client +/// typed into it drives the real channel — exactly as an agent-in-a-pane would. +fn spawn_amux_ctl_shell() -> (pty::Pty, &'static str, &'static str) { + let (shell, flag): (&str, &str) = if cfg!(windows) { + ("cmd", "/Q") + } else { + ("sh", "-i") + }; + let mut p = pty::Pty::spawn_full( + env!("CARGO_BIN_EXE_amux"), + &["--allow-ctl", shell, flag], + 24, + 100, + &[("AMUX_CTL_ALLOW".to_string(), shell.to_string())], + None, + ) + .unwrap(); + let bar: &[u8] = if cfg!(windows) { b"1:cmd" } else { b"1:sh" }; + read_until(&mut p, bar, Duration::from_secs(15)); + (p, shell, flag) +} + +/// `amux ctl list`, run inside a live `--allow-ctl` amux pane, connects over the +/// real channel and returns the org chart as JSON — proving the whole loop: +/// bind → inject env → client connect → non-blocking server drain → reply. +#[test] +fn ctl_list_roundtrips_through_a_live_amux() { + let (mut p, _shell, _flag) = spawn_amux_ctl_shell(); + let amux = env!("CARGO_BIN_EXE_amux"); + p.write(format!("\"{amux}\" ctl list\r\n").as_bytes()) + .unwrap(); + let out = read_until(&mut p, b"\"tree\"", Duration::from_secs(20)); + assert!( + contains(&out, b"\"ok\":true") && contains(&out, b"\"tree\""), + "no ctl list reply in: {:?}", + String::from_utf8_lossy(&out) + ); + p.write(b"\x01q").unwrap(); + let _ = wait_exit(&mut p, 15); +} + +/// `amux ctl spawn -- ` opens a *visible* new worker window: after the +/// call the bar gains a second pane entry. This is the C1 "an in-pane +/// `ctl spawn` opens a visible worker" done-criterion, end to end. +#[test] +fn ctl_spawn_opens_a_visible_worker_window() { + let (mut p, shell, _flag) = spawn_amux_ctl_shell(); + let amux = env!("CARGO_BIN_EXE_amux"); + p.write(format!("\"{amux}\" ctl spawn --role dev_1 -- {shell}\r\n").as_bytes()) + .unwrap(); + let two: &[u8] = if cfg!(windows) { b"2:cmd" } else { b"2:sh" }; + let out = read_until(&mut p, two, Duration::from_secs(20)); + assert!( + contains(&out, two), + "no second (worker) pane in the bar after ctl spawn: {:?}", + String::from_utf8_lossy(&out) + ); + p.write(b"\x01q").unwrap(); + let _ = wait_exit(&mut p, 15); +} + +/// A spawn of a command that is *not* on the allowlist is refused cleanly over +/// the channel — the guard reaches the client as a JSON error, no worker opens. +#[test] +fn ctl_spawn_off_the_allowlist_is_refused() { + let (mut p, _shell, _flag) = spawn_amux_ctl_shell(); + let amux = env!("CARGO_BIN_EXE_amux"); + // `whoami` is a real binary on both platforms but not an agent/allowlisted. + p.write(format!("\"{amux}\" ctl spawn -- whoami\r\n").as_bytes()) + .unwrap(); + let out = read_until(&mut p, b"allowlist", Duration::from_secs(20)); + assert!( + contains(&out, b"\"ok\":false") && contains(&out, b"allowlist"), + "expected an allowlist refusal, got: {:?}", + String::from_utf8_lossy(&out) + ); + p.write(b"\x01q").unwrap(); + let _ = wait_exit(&mut p, 15); +}