diff --git a/README.md b/README.md index 8b16ea8..7327755 100644 --- a/README.md +++ b/README.md @@ -192,6 +192,15 @@ st2 up --catalog "$CATALOG" --host --once There is intentionally no resident macOS service path. +For a shortest-path change to one exact task, render only its owning agent and reconcile only that +task in a bounded pass: + +```sh +st2 up --catalog "$CATALOG" --host --once --task +``` + +Unknown, ambiguous, and wrong-host task selectors refuse before workspace writes or PTY inspection. + `st2 doctor` accepts the absence of a live host lock as the normal manual/`--once` mode. For a resident `st2 up` deployment, use `st2 doctor --require-supervisor` to make a missing loop fail the health check. A stale lock left by a dead supervisor is always a failure. The underlying diff --git a/docs/vrs/requirements.md b/docs/vrs/requirements.md index 157f12c..0f71e31 100644 --- a/docs/vrs/requirements.md +++ b/docs/vrs/requirements.md @@ -97,3 +97,9 @@ accepted. - **R17 Durable error propagation:** Lifecycle, harness/eval, provider-turn, task/exec/PTY, hook, and delivery errors are durably reported to the responsible supervisor with agent/task identity and actionable context. +- **R19 Targeted reconciliation:** An exact agent/task selector resolves its + identity and pinned host before mutation; unknown, ambiguous, and wrong-host + targets refuse before writes, listing, or actions. Materialization, hook + gates, PTY inspection, and plan execution are limited to the selected + owner/task; unrelated diagnostics remain visible while unrelated workspaces, + tasks, and live PTY PID/generation stay unchanged. diff --git a/docs/vrs/spec.md b/docs/vrs/spec.md index 1fb72cc..739a82a 100644 --- a/docs/vrs/spec.md +++ b/docs/vrs/spec.md @@ -141,14 +141,21 @@ positive declaration/template wakes, negative runtime/bus events, bounded discovery/materialization/PTY queries and writes, continuous-event starvation, and no-op desired-equals-actual behavior. -## Open design questions +## Targeted reconciliation (R19) + +`st2 up --materialize-only --task ` resolves one exact local +task before writing and renders only its owning agent. `st2 up --once --task +` performs the same owner-only materialization, then inspects +PTY/exec state and executes a plan containing only that task. Unknown, +ambiguous, and wrong-host selectors refuse before writes or runner inspection; +unrelated discovery diagnostics remain visible without preventing the selected +owner/task path. -### Targeted materialization (R13) +`st2 up --materialize-only --agent ` remains the agent-wide rendering +selector. Targeted task reconciliation is intentionally bounded to `--once`; +the resident supervisor continues to reconcile the complete local catalog. -`st2 up --materialize-only --agent ` filters discovery before rendering, -so a declared agent/task change cannot be blocked by unrelated slow or -unreadable workspaces. This selector is materialization-only; live -reconciliation remains separately gated and host-local. +## Open design questions - **DQ1 Scheduled work:** The vision includes per-machine schedulers that form a distributed workflow engine, but the KDL shape, event inbox, deduplication diff --git a/src/lib.rs b/src/lib.rs index 94dc5bc..86105f8 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -45,5 +45,5 @@ pub use host_lock::HostLock; pub use reconcile::{Launch, ReconcilePlan, Session, TaskLaunch, TaskTarget, Teardown, reconcile}; pub use run::{ PtyCli, Runner, SystemRunner, UpReport, detect_host, down, down_specs, exec_state_dir, execute, - up_loop, up_loop_specs, up_once, up_once_specs, + up_loop, up_loop_specs, up_once, up_once_selected, up_once_selected_specs, up_once_specs, }; diff --git a/src/main.rs b/src/main.rs index 7004314..1040fb3 100644 --- a/src/main.rs +++ b/src/main.rs @@ -53,9 +53,13 @@ enum Command { /// Materialize every local agent's render block and exit without reconciling or spawning. #[arg(long, conflicts_with = "once")] materialize_only: bool, - /// Limit materialization/reconciliation to one declared agent identity. + /// Limit materialization to one declared agent identity. #[arg(long)] agent: Option, + /// Select one exact local task. Use with --materialize-only to render only its owner, or + /// with --once to render its owner and reconcile only that task. + #[arg(long, conflicts_with = "agent")] + task: Option, /// Seconds between timer-driven reconcile passes when looping (folder changes reconcile /// immediately regardless). #[arg(long, default_value_t = 30)] @@ -533,9 +537,16 @@ fn main() -> Result<()> { materialize_only, interval, agent, + task, } => { let root = catalog_arg(root)?; - up(&root, host, once, materialize_only, interval, agent) + if task.is_some() && !materialize_only && !once { + anyhow::bail!("--task requires --once or --materialize-only"); + } + if agent.is_some() && !materialize_only { + anyhow::bail!("--agent requires --materialize-only"); + } + up(&root, host, once, materialize_only, interval, agent, task) } Command::Message(cmd) => message_cmd(cmd), Command::Context(cmd) => context_cmd(cmd), @@ -1753,10 +1764,14 @@ fn up( materialize_only: bool, interval: u64, agent: Option, + task: Option, ) -> Result<()> { // An st2-SPEC path (a `*.kdl` file, or a folder with one top-level spec `*.kdl`) supervises its // top-level team directly — no catalog discovery. Otherwise, the classic catalog reconcile loop. if let Some(spec_file) = st2::eval_run::resolve_spec_path(root) { + if task.is_some() { + anyhow::bail!("--task is for folder catalogs, not single-file specs"); + } if materialize_only { anyhow::bail!( "--materialize-only is for folder catalogs with agent render{{}} blocks, not single-file specs" @@ -1770,6 +1785,14 @@ fn up( if materialize_only { let mut found = discover(&catalog_root); + if let Some(selector) = task.as_deref() { + let (owner, _, _) = st2::reconcile::resolve_task(&found.specs, selector, &this_host)?; + let owner_identity = owner.identity.clone(); + let owner_path = owner.path.clone(); + found + .specs + .retain(|spec| spec.identity == owner_identity && spec.path == owner_path); + } if let Some(identity) = agent.as_deref() { found .specs @@ -1809,7 +1832,7 @@ fn up( return Ok(()); } - let runner = SystemRunner::new(catalog_root, exec_state_dir(&this_host)); + let runner = SystemRunner::new(catalog_root.clone(), exec_state_dir(&this_host)); // One supervisor per (folder, host). A single `--once` pass must also refuse while a loop owns // the lock (it would double-spawn) — but it does NOT take the lock itself (that would clobber the @@ -1821,12 +1844,21 @@ fn up( } if once { - let report = up_once(root, &this_host, &runner)?; + let targeted = task.is_some(); + let report = match task.as_deref() { + Some(selector) => { + st2::run::up_once_selected(&catalog_root, selector, &this_host, &runner)? + } + None => up_once(root, &this_host, &runner)?, + }; println!("reconcile pass on host '{this_host}':"); print_report(&report); if report.skipped { anyhow::bail!("one-shot reconcile pass was skipped"); } + if targeted && !report.errors.is_empty() { + anyhow::bail!("targeted one-shot reconcile pass reported errors"); + } return Ok(()); } diff --git a/src/reconcile.rs b/src/reconcile.rs index 0f4f054..baf822f 100644 --- a/src/reconcile.rs +++ b/src/reconcile.rs @@ -89,6 +89,105 @@ pub struct ReconcilePlan<'a> { pub gc: Vec, } +/// Resolve one exact local task selector (`host.agent.task` or explicit task id) without mutation. +pub fn resolve_task<'a>( + specs: &'a [AgentSpec], + selector: &str, + this_host: &str, +) -> anyhow::Result<(&'a AgentSpec, &'a crate::spec::Task, String)> { + let mut matches = Vec::new(); + for spec in specs { + if spec.resolved_host(this_host) != this_host { + continue; + } + for task in &spec.tasks { + let runtime = task + .id + .clone() + .unwrap_or_else(|| format!("{}.{}", spec.bus_id(this_host), task.name)); + let qualified = format!("{}.{}", spec.bus_id(this_host), task.name); + if selector == runtime || selector == qualified { + matches.push((spec, task, runtime)); + } + } + } + match matches.as_slice() { + [(spec, task, runtime)] => Ok((*spec, *task, runtime.clone())), + [] => anyhow::bail!("task selector {selector:?} did not resolve to one local task"), + _ => anyhow::bail!("task selector {selector:?} is ambiguous"), + } +} + +/// Pure task-scoped plan: resolve first, then retain only the selected runtime target. +pub fn reconcile_selected<'a>( + specs: &'a [AgentSpec], + sessions: &[Session], + this_host: &str, + selector: &str, +) -> anyhow::Result> { + let (owner, task, runtime) = resolve_task(specs, selector, this_host)?; + let mut plan = ReconcilePlan::default(); + let actual = sessions.iter().find(|s| s.pty_id == runtime); + if owner.retired { + if let Some(s) = actual { + if s.alive { + plan.teardown.push(Teardown { + spec: owner, + pty_ids: vec![runtime], + }); + } else if !(task.keep || owner.keep) { + plan.gc.push(runtime); + } + } + return Ok(plan); + } + let launch = match (&task.command, &task.argv) { + (Some(command), None) => TaskLaunch::Shell(command.clone()), + (None, Some(argv)) => TaskLaunch::Argv(argv.clone()), + (None, None) => { + plan.unrunnable.push(owner); + return Ok(plan); + } + (Some(_), Some(_)) => { + unreachable!("discovery rejects tasks carrying both command and argv") + } + }; + let bus_id = owner.bus_id(this_host); + let mut env = task.env.clone(); + if let Some(supervisor) = &owner.supervisor { + env.insert("ST_SUPERVISOR".into(), supervisor.clone()); + } else { + env.remove("ST_SUPERVISOR"); + } + let target = TaskTarget { + kind: task.kind, + pty_id: runtime.clone(), + bus_id, + name: task.name.clone(), + launch, + cwd: task.cwd.clone(), + workspace: owner.workspace.clone(), + tags: task.tags.clone(), + env, + keep: task.keep || owner.keep, + }; + match actual { + Some(s) if s.alive || target.keep => plan.adopt.push(owner), + Some(_) => { + plan.gc.push(runtime); + plan.launch.push(Launch { + spec: owner, + tasks: vec![target], + }); + } + _ => plan.launch.push(Launch { + spec: owner, + tasks: vec![target], + }), + } + Ok(plan) +} + /// The state of a declared task's session in the ACTUAL world. enum SessionState { Alive, diff --git a/src/run.rs b/src/run.rs index 49c624e..0fa3d7e 100644 --- a/src/run.rs +++ b/src/run.rs @@ -564,6 +564,20 @@ pub struct UpReport { } impl UpReport { + fn absorb(&mut self, mut other: UpReport) { + self.skipped |= other.skipped; + self.launched.append(&mut other.launched); + self.torn_down.append(&mut other.torn_down); + self.gc.append(&mut other.gc); + self.deferred.append(&mut other.deferred); + self.flapping.append(&mut other.flapping); + self.crash_loops.append(&mut other.crash_loops); + self.adopted.append(&mut other.adopted); + self.other_host.append(&mut other.other_host); + self.unrunnable.append(&mut other.unrunnable); + self.warnings.append(&mut other.warnings); + self.errors.append(&mut other.errors); + } /// True when the pass actually changed something (or hit an error) — used to keep the loop's log /// quiet on no-op ticks. pub fn is_noteworthy(&self) -> bool { @@ -943,6 +957,91 @@ pub fn up_once_specs( ) } +/// One bounded task-scoped pass over already-discovered specs. Selector resolution precedes any runner call. +pub fn up_once_selected_specs( + catalog_root: &Path, + specs: &[crate::spec::AgentSpec], + selector: &str, + this_host: &str, + runner: &dyn Runner, +) -> anyhow::Result { + up_once_selected_specs_with_gates(catalog_root, specs, selector, this_host, runner, || { + crate::hooks::verify_installed().map(|_| ()) + }) +} + +/// Discover a folder catalog once, resolve one task before any owner hook/render mutation, then +/// materialize only that owner and execute the selected plan. +pub fn up_once_selected( + catalog_root: &Path, + selector: &str, + this_host: &str, + runner: &dyn Runner, +) -> anyhow::Result { + let found = crate::discovery::discover(catalog_root); + let (owner, _, _) = crate::reconcile::resolve_task(&found.specs, selector, this_host)?; + let mut report = UpReport::default(); + report.warnings.extend(found.warnings); + report.errors.extend( + found + .errors + .into_iter() + .map(|e| format!("{}: {}", e.path.display(), e.message)), + ); + let owner = owner.clone(); + if crate::hooks::required_by_codex_agent(&owner, this_host, catalog_root) + && let Err(error) = crate::hooks::verify_installed() + { + report + .errors + .push(format!("verify lifecycle hooks: {error}")); + return Ok(report); + } + let materialized = crate::materialize::materialize_catalog( + catalog_root, + std::slice::from_ref(&owner), + this_host, + ); + report.warnings.extend(materialized.warnings); + let owner_materialization_failed = !materialized.failed_agents.is_empty(); + report.errors.extend(materialized.errors); + if owner_materialization_failed { + return Ok(report); + } + let execution = up_once_selected_specs_with_gates( + catalog_root, + &found.specs, + selector, + this_host, + runner, + || Ok(()), + )?; + report.absorb(execution); + Ok(report) +} + +fn up_once_selected_specs_with_gates( + catalog_root: &Path, + specs: &[crate::spec::AgentSpec], + selector: &str, + this_host: &str, + runner: &dyn Runner, + verify_hooks: V, +) -> anyhow::Result +where + V: FnOnce() -> anyhow::Result<()>, +{ + crate::reconcile::resolve_task(specs, selector, this_host)?; + let sessions = runner + .list_sessions() + .map_err(|e| anyhow::anyhow!("list sessions: {e}"))?; + let mut plan = crate::reconcile::reconcile_selected(specs, &sessions, this_host, selector)?; + let mut report = UpReport::default(); + gate_codex_launches_on_hooks(&mut plan, catalog_root, &mut report, verify_hooks); + execute(&plan, runner, &mut FlappingCap::default(), &mut report); + Ok(report) +} + /// Supervise an in-memory spec team: keep-alive + respawn on a timer, behaving exactly like /// [`up_loop`] over a catalog (same /// FlappingCap, LivenessDebounce, crash-loop surfacing, and "stop leaves sessions running"). Timer-only @@ -1227,7 +1326,8 @@ pub fn detect_host() -> String { #[cfg(test)] mod tests { use super::*; - use agent_spec::spec::TaskKind; + use agent_spec::spec::{AgentSpec, JobType, Task, TaskKind}; + use std::cell::Cell; use std::collections::BTreeMap; use std::ffi::OsStr; @@ -1246,28 +1346,74 @@ mod tests { } } - #[cfg(target_os = "linux")] - struct EmptyRunner; + struct GateRunner { + list_calls: Cell, + } - #[cfg(target_os = "linux")] - impl Runner for EmptyRunner { + impl Runner for GateRunner { fn list_sessions(&self) -> anyhow::Result> { + self.list_calls.set(self.list_calls.get() + 1); Ok(Vec::new()) } fn spawn(&self, _target: &TaskTarget, _spec_dir: &Path) -> anyhow::Result<()> { - unreachable!("an empty catalog cannot launch") + panic!("gate runner must not spawn") } fn kill(&self, _pty_id: &str) -> anyhow::Result<()> { - unreachable!("an empty catalog cannot kill") + panic!("gate runner must not kill") } fn remove(&self, _pty_id: &str) -> anyhow::Result<()> { - unreachable!("an empty catalog cannot remove") + panic!("gate runner must not remove") } } + #[test] + fn selected_codex_gate_suppresses_launch_on_stale_hooks() { + let spec = AgentSpec { + identity: "codex".into(), + host: None, + role: None, + job_type: JobType::Service, + workspace: None, + supervisor: None, + retired: false, + keep: false, + restart: None, + tasks: vec![Task { + kind: TaskKind::Pty, + derived: false, + name: "agent".into(), + id: Some("test.codex.agent".into()), + command: None, + argv: Some(vec!["$CATALOG/bin/codex".into(), "--version".into()]), + cwd: None, + tags: BTreeMap::new(), + env: BTreeMap::new(), + keep: false, + }], + path: "/tmp/spec.kdl".into(), + }; + let runner = GateRunner { + list_calls: Cell::new(0), + }; + let report = up_once_selected_specs_with_gates( + Path::new("/tmp"), + &[spec], + "test.codex.agent", + "test", + &runner, + || anyhow::bail!("stale receipt"), + ) + .unwrap(); + assert_eq!(runner.list_calls.get(), 1); + assert!(report.launched.is_empty()); + assert!(report.errors.iter().any(|error| { + error.contains("stale receipt") && error.contains("launch suppressed") + })); + } + #[cfg(target_os = "linux")] #[test] fn idle_supervisor_does_not_spin_on_its_own_catalog_reads() { @@ -1283,7 +1429,9 @@ mod tests { up_loop_until( catalog.path(), "test-host", - &EmptyRunner, + &GateRunner { + list_calls: Cell::new(0), + }, Duration::from_secs(60), &stop, |_| passes += 1, @@ -1301,8 +1449,6 @@ mod tests { // destructively GC/relaunch a HEALTHY agent; a stable death must still be reaped ────────────── use crate::reconcile::Launch; - use agent_spec::spec::{AgentSpec, JobType}; - fn sess(id: &str, alive: bool) -> Session { Session { pty_id: id.to_string(), diff --git a/tests/materialize.rs b/tests/materialize.rs index 9c80095..d80e647 100644 --- a/tests/materialize.rs +++ b/tests/materialize.rs @@ -10,6 +10,203 @@ fn write(path: &Path, contents: impl AsRef<[u8]>) { fs::write(path, contents).unwrap(); } +#[test] +fn task_selector_refusal_is_nonzero_before_catalog_mutation() { + let tmp = tempfile::tempdir().unwrap(); + let before = fs::read_dir(tmp.path()).unwrap().count(); + let out = Command::new(env!("CARGO_BIN_EXE_st2")) + .args(["up", "--catalog"]) + .arg(tmp.path()) + .args([ + "--host", + "host", + "--materialize-only", + "--task", + "host.missing.task", + ]) + .output() + .unwrap(); + assert!(!out.status.success()); + assert_eq!(fs::read_dir(tmp.path()).unwrap().count(), before); +} + +#[test] +fn task_selector_materializes_only_owning_agent() { + let tmp = tempfile::tempdir().unwrap(); + let catalog = tmp.path().join("catalog"); + let owner = tmp.path().join("owner"); + let sibling = tmp.path().join("sibling"); + fs::create_dir_all(&owner).unwrap(); + fs::create_dir_all(&sibling).unwrap(); + write( + &catalog.join("agents/Silber/cos/agent.kdl"), + agent_kdl(&owner, r#" copy "_templates/owner" "OWNER.txt""#), + ); + write( + &catalog.join("agents/Silber/pty/agent.kdl"), + agent_kdl(&sibling, r#" copy "_templates/sibling" "SIBLING.txt""#) + .replace("agent \"cos\"", "agent \"pty\"") + .replace("Silber.cos", "Silber.pty"), + ); + write(&catalog.join("_templates/owner"), "owner"); + write(&catalog.join("_templates/sibling"), "sibling"); + let out = Command::new(env!("CARGO_BIN_EXE_st2")) + .args(["up", "--catalog"]) + .arg(&catalog) + .args([ + "--host", + "Silber", + "--materialize-only", + "--task", + "Silber.cos.agent", + ]) + .output() + .unwrap(); + assert!( + out.status.success(), + "{}", + String::from_utf8_lossy(&out.stderr) + ); + assert_eq!( + fs::read_to_string(owner.join("OWNER.txt")).unwrap(), + "owner" + ); + assert!(!sibling.join("SIBLING.txt").exists()); + assert!(String::from_utf8_lossy(&out.stdout).contains("materialized 1 operation")); +} + +#[test] +fn task_selector_ambiguous_refuses_without_mutation() { + let tmp = tempfile::tempdir().unwrap(); + let catalog = tmp.path().join("catalog"); + let a = tmp.path().join("a"); + let b = tmp.path().join("b"); + fs::create_dir_all(&a).unwrap(); + fs::create_dir_all(&b).unwrap(); + let kdl = |id: &str, ws: &Path, marker: &str| { + format!( + "agent \"{id}\" {{\n host \"Silber\"\n type \"service\"\n workspace \"{}\"\n pty \"agent\" {{\n id \"dup\"\n command \"true\"\n }}\n render {{ file \"MARKER.txt\" \"{marker}\" }}\n}}\n", + ws.display() + ) + }; + write( + &catalog.join("agents/Silber/a/agent.kdl"), + kdl("a", &a, "a"), + ); + write( + &catalog.join("agents/Silber/b/agent.kdl"), + kdl("b", &b, "b"), + ); + let found = st2::discover(&catalog); + assert!(found.errors.is_empty(), "{:?}", found.errors); + assert_eq!(found.specs.len(), 2); + assert!( + found + .specs + .iter() + .all(|s| s.tasks.len() == 1 && s.tasks[0].id.as_deref() == Some("dup")) + ); + let out = Command::new(env!("CARGO_BIN_EXE_st2")) + .args(["up", "--catalog"]) + .arg(&catalog) + .args(["--host", "Silber", "--materialize-only", "--task", "dup"]) + .output() + .unwrap(); + assert!(!out.status.success()); + assert!(String::from_utf8_lossy(&out.stderr).contains("ambiguous")); + assert!(!a.join("MARKER.txt").exists() && !b.join("MARKER.txt").exists()); +} + +#[test] +fn task_selector_wrong_host_refuses_without_mutation() { + let tmp = tempfile::tempdir().unwrap(); + let catalog = tmp.path().join("catalog"); + let target = tmp.path().join("hetz-target"); + fs::create_dir_all(&target).unwrap(); + let text = format!( + "agent \"remote\" {{\n host \"Hetz\"\n type \"service\"\n workspace \"{}\"\n pty \"agent\" {{ id \"hetz.task\" command \"true\" }}\n render {{ file \"MARKER.txt\" \"remote\" }}\n}}\n", + target.display() + ); + write(&catalog.join("agents/Hetz/remote/agent.kdl"), text); + let found = st2::discover(&catalog); + assert!(found.errors.is_empty()); + assert_eq!(found.specs[0].tasks[0].id.as_deref(), Some("hetz.task")); + let out = Command::new(env!("CARGO_BIN_EXE_st2")) + .args(["up", "--catalog"]) + .arg(&catalog) + .args([ + "--host", + "Silber", + "--materialize-only", + "--task", + "hetz.task", + ]) + .output() + .unwrap(); + assert!(!out.status.success()); + assert!(String::from_utf8_lossy(&out.stderr).contains("did not resolve")); + assert!(!target.join("MARKER.txt").exists()); +} + +#[test] +fn task_selector_cli_modes_fail_closed() { + let tmp = tempfile::tempdir().unwrap(); + for args in [ + vec![ + "up", + "--catalog", + tmp.path().to_str().unwrap(), + "--task", + "host.a.x", + ], + vec![ + "up", + "--catalog", + tmp.path().to_str().unwrap(), + "--materialize-only", + "--task", + "host.a.x", + "--agent", + "a", + ], + vec![ + "up", + "--catalog", + tmp.path().to_str().unwrap(), + "--agent", + "a", + ], + ] { + let out = Command::new(env!("CARGO_BIN_EXE_st2")) + .args(args) + .output() + .unwrap(); + assert!(!out.status.success()); + assert!(!String::from_utf8_lossy(&out.stderr).trim().is_empty()); + } +} + +#[test] +fn task_selector_single_file_modes_refuse_unchanged() { + let tmp = tempfile::tempdir().unwrap(); + let spec = tmp.path().join("spec.kdl"); + fs::write(&spec, "agent \"a\" { host \"Silber\" command \"true\" }\n").unwrap(); + let before = fs::read_to_string(&spec).unwrap(); + for extra in [ + ["--materialize-only", "--task", "Silber.a.agent"], + ["--once", "--task", "Silber.a.agent"], + ] { + let out = Command::new(env!("CARGO_BIN_EXE_st2")) + .args(["up", spec.to_str().unwrap()]) + .args(extra) + .output() + .unwrap(); + assert!(!out.status.success()); + assert!(!String::from_utf8_lossy(&out.stderr).is_empty()); + } + assert_eq!(fs::read_to_string(&spec).unwrap(), before); +} + fn spec(catalog: &Path, identity: &str) -> AgentSpec { discover(catalog) .specs diff --git a/tests/nomad_survival.rs b/tests/nomad_survival.rs index f3fa938..92bb26f 100644 --- a/tests/nomad_survival.rs +++ b/tests/nomad_survival.rs @@ -191,7 +191,10 @@ impl Fixture { let bin_dir = self.xdg.join("bin"); std::fs::create_dir_all(&bin_dir).unwrap(); let installed = bin_dir.join("st2"); - std::fs::copy(env!("CARGO_BIN_EXE_st2"), &installed).unwrap(); + let staged = installed.with_extension("installing"); + std::fs::copy(env!("CARGO_BIN_EXE_st2"), &staged).unwrap(); + std::fs::File::open(&staged).unwrap().sync_all().unwrap(); + std::fs::rename(&staged, &installed).unwrap(); installed } @@ -216,14 +219,27 @@ impl Fixture { } fn spawn_loop_from(&self, binary: &Path) -> Child { - self.st2_from(binary) - .arg("up") - .arg(&self.catalog) - .args(["--host", HOST, "--interval", "60"]) - .stdout(Stdio::null()) - .stderr(Stdio::null()) - .spawn() - .unwrap() + for attempt in 0..5 { + let result = self + .st2_from(binary) + .arg("up") + .arg(&self.catalog) + .args(["--host", HOST, "--interval", "60"]) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn(); + match result { + Ok(child) => return child, + Err(error) if error.raw_os_error() == Some(libc::ETXTBSY) && attempt + 1 < 5 => { + // Some Linux filesystems briefly retain the writer exclusion after installing + // a copied executable. Retry only that transient; every other spawn error stays + // loud and immediate. + std::thread::sleep(Duration::from_millis(20)); + } + Err(error) => panic!("spawning fixture control plane failed: {error}"), + } + } + unreachable!("the bounded spawn loop always returns or panics") } /// One `st2 up --once` pass; returns its stdout (where launched/adopted/torn-down is reported). diff --git a/tests/reconcile.rs b/tests/reconcile.rs index 2e2c7d7..0b2defd 100644 --- a/tests/reconcile.rs +++ b/tests/reconcile.rs @@ -3,9 +3,324 @@ use std::collections::BTreeMap; use std::path::PathBuf; +use st2::reconcile::reconcile_selected; +use st2::reconcile::resolve_task; use st2::spec::{AgentSpec, JobType, Task, TaskKind}; use st2::{Session, reconcile}; +#[test] +fn exact_task_selector_matrix() { + let specs = vec![ + svc( + "a", + None, + vec![ + task(TaskKind::Pty, "agent", None, Some("run")), + task(TaskKind::Exec, "ding", Some("host.a.ding"), Some("ding")), + ], + ), + svc( + "b", + None, + vec![task( + TaskKind::Pty, + "agent", + Some("host.a.agent"), + Some("other"), + )], + ), + svc( + "remote", + Some("other"), + vec![task(TaskKind::Pty, "agent", None, Some("remote"))], + ), + ]; + let (_, selected, runtime) = resolve_task(&specs, "host.a.ding", "host").unwrap(); + assert_eq!(selected.name, "ding"); + assert_eq!(runtime, "host.a.ding"); + assert!(resolve_task(&specs, "host.a.agent", "host").is_err()); // explicit-id collision is ambiguous + assert!(resolve_task(&specs, "a", "host").is_err()); + assert!(resolve_task(&specs, "other.remote.agent", "host").is_err()); + assert!(resolve_task(&specs, "host.a.missing", "host").is_err()); +} + +#[test] +fn selector_derived_id_success() { + let s = vec![svc( + "plain", + None, + vec![task(TaskKind::Exec, "work", None, Some("cmd"))], + )]; + let (owner, t, id) = resolve_task(&s, "host.plain.work", "host").unwrap(); + assert_eq!( + ( + owner.identity.as_str(), + t.name.as_str(), + t.command.as_deref(), + id.as_str() + ), + ("plain", "work", Some("cmd"), "host.plain.work") + ); +} + +#[test] +fn selector_explicit_id_and_qualified_alias_return_explicit_runtime() { + let s = vec![svc( + "agent", + None, + vec![task(TaskKind::Exec, "work", Some("custom"), Some("cmd"))], + )]; + for selector in ["custom", "host.agent.work"] { + let (o, t, id) = resolve_task(&s, selector, "host").unwrap(); + assert_eq!( + ( + o.identity.as_str(), + t.name.as_str(), + t.command.as_deref(), + id.as_str() + ), + ("agent", "work", Some("cmd"), "custom") + ); + } +} + +#[test] +fn selector_agent_command_runtime_id_and_inputs_immutable() { + let s = vec![svc( + "agent", + None, + vec![task( + TaskKind::Pty, + "agent", + Some("host.agent"), + Some("run"), + )], + )]; + let before = s.clone(); + let (o, t, id) = resolve_task(&s, "host.agent", "host").unwrap(); + assert_eq!( + ( + o.identity.as_str(), + t.name.as_str(), + t.command.as_deref(), + id.as_str() + ), + ("agent", "agent", Some("run"), "host.agent") + ); + assert_eq!(s, before); + assert!(resolve_task(&s, "host.missing", "host").is_err()); + assert_eq!(s, before); +} + +#[test] +fn selected_reconcile_launches_missing_and_adopts_live_without_siblings() { + let specs = vec![ + svc( + "a", + None, + vec![ + task(TaskKind::Exec, "x", None, Some("a")), + task(TaskKind::Exec, "y", None, Some("b")), + ], + ), + svc("b", None, vec![task(TaskKind::Exec, "z", None, Some("c"))]), + ]; + let plan = reconcile_selected(&specs, &[], "host", "host.a.x").unwrap(); + assert_eq!(plan.launch.len(), 1); + assert_eq!(plan.launch[0].tasks.len(), 1); + assert_eq!(plan.launch[0].tasks[0].pty_id, "host.a.x"); + let plan2 = reconcile_selected(&specs, &[live("host.a.x"), live("host.a.y"), live("host.b.z")], "host", "host.a.x").unwrap(); + assert!(plan2.launch.is_empty() && plan2.gc.is_empty() && plan2.teardown.is_empty()); + assert_eq!(plan2.adopt.iter().map(|s| s.identity.as_str()).collect::>(), vec!["a"]); +} + +#[test] +fn selected_reconcile_freezes_dead_keep_and_retired_task_keep() { + let keep = svc( + "a", + None, + vec![{ + let mut t = task(TaskKind::Exec, "x", None, Some("a")); + t.keep = true; + t + }], + ); + let specs = [keep]; + let p = reconcile_selected( + &specs, + &[Session { + pty_id: "host.a.x".into(), + alive: false, + exit_code: Some(7), + }], + "host", + "host.a.x", + ) + .unwrap(); + assert!(p.launch.is_empty() && p.gc.is_empty() && p.adopt.len() == 1); + let mut retired = svc( + "b", + None, + vec![{ + let mut t = task(TaskKind::Exec, "x", None, Some("b")); + t.keep = true; + t + }], + ); + retired.retired = true; + let specs = [retired]; + let p = reconcile_selected( + &specs, + &[Session { + pty_id: "host.b.x".into(), + alive: false, + exit_code: Some(7), + }], + "host", + "host.b.x", + ) + .unwrap(); + assert!(p.teardown.is_empty() && p.gc.is_empty()); +} + +#[test] +fn selected_reconcile_action_ids_are_exact_and_refusals_immutable() { + let specs = vec![ + svc( + "a", + None, + vec![ + task(TaskKind::Exec, "x", None, Some("a")), + task(TaskKind::Exec, "y", None, Some("b")), + ], + ), + svc("b", None, vec![task(TaskKind::Exec, "z", None, Some("c"))]), + ]; + let sessions = vec![live("host.a.y"), live("host.b.z")]; + let before = (specs.clone(), sessions.clone()); + let p = reconcile_selected(&specs, &sessions, "host", "host.a.x").unwrap(); + assert_eq!( + p.launch + .iter() + .flat_map(|l| l.tasks.iter().map(|t| t.pty_id.as_str())) + .collect::>(), + vec!["host.a.x"] + ); + assert!(p.gc.is_empty() && p.teardown.is_empty()); + assert!(reconcile_selected(&specs, &sessions, "host", "host.a.missing").is_err()); + assert_eq!((specs, sessions), before); +} + +#[test] +fn selected_dead_non_keep_gc_and_relaunch_only_selected() { + let specs = vec![ + svc( + "a", + None, + vec![ + task(TaskKind::Exec, "x", None, Some("a")), + task(TaskKind::Exec, "y", None, Some("b")), + ], + ), + svc("b", None, vec![task(TaskKind::Exec, "z", None, Some("c"))]), + ]; + let p = reconcile_selected( + &specs, + &[ + Session { + pty_id: "host.a.x".into(), + alive: false, + exit_code: Some(1), + }, + Session { + pty_id: "host.a.y".into(), + alive: false, + exit_code: Some(1), + }, + Session { + pty_id: "host.b.z".into(), + alive: false, + exit_code: Some(1), + }, + ], + "host", + "host.a.x", + ) + .unwrap(); + assert_eq!(p.gc, vec!["host.a.x"]); + assert_eq!(p.launch.iter().flat_map(|l| l.tasks.iter().map(|t| t.pty_id.as_str())).collect::>(), vec!["host.a.x"]); + assert!(p.teardown.is_empty()); +} +#[test] +fn selected_retired_live_tears_down_only_selected() { + let mut s = svc("a", None, vec![task(TaskKind::Exec, "x", None, Some("a")), task(TaskKind::Exec, "sib", None, Some("b"))]); + s.retired = true; + let specs = [s, svc("b", None, vec![task(TaskKind::Exec, "z", None, Some("c"))])]; + let p = reconcile_selected( + &specs, + &[live("host.a.x"), live("host.a.sib"), live("host.b.z")], + "host", + "host.a.x", + ) + .unwrap(); + assert_eq!(p.teardown.iter().flat_map(|t| t.pty_ids.iter().map(String::as_str)).collect::>(), vec!["host.a.x"]); + assert!(p.launch.is_empty() && p.gc.is_empty()); +} +#[test] +fn selected_refusals_ambiguous_and_unknown_preserve_inputs() { + let specs = vec![ + svc( + "a", + None, + vec![task(TaskKind::Exec, "x", Some("dup"), Some("a"))], + ), + svc( + "b", + None, + vec![task(TaskKind::Exec, "y", Some("dup"), Some("b"))], + ), + ]; + let sessions = vec![live("dup")]; + let before = (specs.clone(), sessions.clone()); + assert!(reconcile_selected(&specs, &sessions, "host", "dup").is_err()); + assert!(reconcile_selected(&specs, &sessions, "host", "none").is_err()); + assert_eq!((specs, sessions), before); +} +#[test] +fn selected_unrunnable_is_runner_action_free() { + let specs = vec![svc("a", None, vec![task(TaskKind::Exec, "x", None, None)])]; + let before = specs.clone(); + let p = reconcile_selected(&specs, &[], "host", "host.a.x").unwrap(); + assert!(p.launch.is_empty() && p.gc.is_empty() && p.teardown.is_empty()); + assert_eq!(specs, before); +} + +#[test] +fn selector_rejects_duplicate_explicit_and_runtime_collision() { + let dup = vec![ + svc( + "a", + None, + vec![task(TaskKind::Exec, "x", Some("dup"), Some("a"))], + ), + svc( + "b", + None, + vec![task(TaskKind::Exec, "y", Some("dup"), Some("b"))], + ), + ]; + assert!(resolve_task(&dup, "dup", "host").is_err()); + let collision = vec![ + svc("a", None, vec![task(TaskKind::Exec, "x", None, Some("a"))]), + svc( + "b", + None, + vec![task(TaskKind::Exec, "y", Some("host.a.x"), Some("b"))], + ), + ]; + assert!(resolve_task(&collision, "host.a.x", "host").is_err()); +} + fn task(kind: TaskKind, name: &str, id: Option<&str>, command: Option<&str>) -> Task { Task { kind, @@ -21,7 +336,13 @@ fn task(kind: TaskKind, name: &str, id: Option<&str>, command: Option<&str>) -> } } -fn spec(identity: &str, host: Option<&str>, job_type: JobType, retired: bool, tasks: Vec) -> AgentSpec { +fn spec( + identity: &str, + host: Option<&str>, + job_type: JobType, + retired: bool, + tasks: Vec, +) -> AgentSpec { AgentSpec { identity: identity.to_string(), host: host.map(String::from), @@ -33,7 +354,10 @@ fn spec(identity: &str, host: Option<&str>, job_type: JobType, retired: bool, ta keep: false, restart: None, tasks, - path: PathBuf::from(format!("/cat/agents/{}/{identity}/agent.kdl", host.unwrap_or("this"))), + path: PathBuf::from(format!( + "/cat/agents/{}/{identity}/agent.kdl", + host.unwrap_or("this") + )), } } @@ -42,10 +366,18 @@ fn svc(identity: &str, host: Option<&str>, tasks: Vec) -> AgentSpec { } fn live(id: &str) -> Session { - Session { pty_id: id.to_string(), alive: true, exit_code: None } + Session { + pty_id: id.to_string(), + alive: true, + exit_code: None, + } } fn dead(id: &str) -> Session { - Session { pty_id: id.to_string(), alive: false, exit_code: None } + Session { + pty_id: id.to_string(), + alive: false, + exit_code: None, + } } const HOST: &str = "hetz"; @@ -56,8 +388,18 @@ fn fresh_service_launches_all_tasks_pty_and_exec() { "st2-claude", Some(HOST), vec![ - task(TaskKind::Pty, "agent", Some("hetz.st2-claude"), Some("exec claude 'boot'")), - task(TaskKind::Exec, "ding", Some("hetz.st2.ding"), Some("st2 ding hetz.st2")), + task( + TaskKind::Pty, + "agent", + Some("hetz.st2-claude"), + Some("exec claude 'boot'"), + ), + task( + TaskKind::Exec, + "ding", + Some("hetz.st2.ding"), + Some("st2 ding hetz.st2"), + ), ], )]; let plan = reconcile(&specs, &[], HOST); @@ -102,7 +444,11 @@ fn one_dead_task_launches_only_the_missing_one() { #[test] fn exited_session_is_reaped_and_relaunched() { - let specs = vec![svc("a", Some(HOST), vec![task(TaskKind::Pty, "agent", Some("hetz.a"), Some("x"))])]; + let specs = vec![svc( + "a", + Some(HOST), + vec![task(TaskKind::Pty, "agent", Some("hetz.a"), Some("x"))], + )]; let plan = reconcile(&specs, &[dead("hetz.a")], HOST); assert_eq!(plan.launch.len(), 1); assert_eq!(plan.gc, vec!["hetz.a"]); // reap the corpse, then respawn @@ -136,8 +482,21 @@ fn retired_with_live_sessions_is_torn_down() { #[test] fn other_host_specs_are_skipped() { let specs = vec![ - svc("here", Some(HOST), vec![task(TaskKind::Pty, "agent", Some("hetz.here"), Some("x"))]), - svc("there", Some("silber"), vec![task(TaskKind::Pty, "agent", Some("silber.there"), Some("y"))]), + svc( + "here", + Some(HOST), + vec![task(TaskKind::Pty, "agent", Some("hetz.here"), Some("x"))], + ), + svc( + "there", + Some("silber"), + vec![task( + TaskKind::Pty, + "agent", + Some("silber.there"), + Some("y"), + )], + ), ]; let plan = reconcile(&specs, &[], HOST); assert_eq!(plan.launch.len(), 1); @@ -148,7 +507,11 @@ fn other_host_specs_are_skipped() { #[test] fn host_none_defaults_to_this_host_with_fallback_id() { - let specs = vec![svc("local", None, vec![task(TaskKind::Pty, "agent", None, Some("x"))])]; + let specs = vec![svc( + "local", + None, + vec![task(TaskKind::Pty, "agent", None, Some("x"))], + )]; let plan = reconcile(&specs, &[], HOST); assert_eq!(plan.launch.len(), 1); assert_eq!(plan.launch[0].tasks[0].pty_id, "hetz.local.agent"); // . @@ -156,7 +519,11 @@ fn host_none_defaults_to_this_host_with_fallback_id() { #[test] fn unrendered_job_without_commands_is_unrunnable() { - let specs = vec![svc("nr", Some(HOST), vec![task(TaskKind::Pty, "agent", None, None)])]; + let specs = vec![svc( + "nr", + Some(HOST), + vec![task(TaskKind::Pty, "agent", None, None)], + )]; let plan = reconcile(&specs, &[], HOST); assert!(plan.launch.is_empty()); assert_eq!(plan.unrunnable.len(), 1); @@ -179,12 +546,7 @@ fn generated_ding_only_job_is_unrunnable_and_does_not_launch() { #[test] fn generated_ding_launches_alongside_authored_work() { - let agent = task( - TaskKind::Pty, - "agent", - Some("hetz.runnable"), - Some("codex"), - ); + let agent = task(TaskKind::Pty, "agent", Some("hetz.runnable"), Some("codex")); let mut ding = task( TaskKind::Exec, "ding", @@ -215,10 +577,17 @@ fn agent_level_keep_pins_all_task_targets() { #[test] fn workspace_is_carried_into_task_targets_for_cwd_defaulting() { - let mut s = svc("w", Some(HOST), vec![task(TaskKind::Pty, "agent", Some("hetz.w"), Some("x"))]); + let mut s = svc( + "w", + Some(HOST), + vec![task(TaskKind::Pty, "agent", Some("hetz.w"), Some("x"))], + ); s.workspace = Some("/repos/w".into()); let plan = reconcile(std::slice::from_ref(&s), &[], HOST); - assert_eq!(plan.launch[0].tasks[0].workspace.as_deref(), Some("/repos/w")); + assert_eq!( + plan.launch[0].tasks[0].workspace.as_deref(), + Some("/repos/w") + ); } #[test] diff --git a/tests/run.rs b/tests/run.rs index 21f5c4c..8762db3 100644 --- a/tests/run.rs +++ b/tests/run.rs @@ -1,17 +1,414 @@ //! M1 correctness net: plan execution against a fake Runner (no real processes spawned). -use std::cell::RefCell; +use std::cell::{Cell, RefCell}; +use std::collections::BTreeMap; use std::fs; use std::path::Path; +use st2::message; use st2::reconcile::{Session, TaskTarget}; use st2::run::Runner; -use st2::message; -use st2::run::{CrashLoop, surface_crash_loop}; +use st2::run::{CrashLoop, surface_crash_loop, up_once_selected, up_once_selected_specs}; +use st2::spec::{AgentSpec, JobType, Task, TaskKind}; + +fn selected_catalog_agent(identity: &str, workspace: &Path, render: &str) -> String { + format!( + r#"agent "{identity}" {{ + host "host" + type "service" + workspace "{}" + pty "work" {{ + id "host.{identity}.work" + command "true" + }} + render {{ + {render} + }} +}} +"#, + workspace.display() + ) +} + +fn write_selected_catalog( + catalog: &Path, + owner_workspace: &Path, + sibling_workspace: &Path, + owner_render: &str, +) { + fs::create_dir_all(owner_workspace).unwrap(); + fs::create_dir_all(sibling_workspace).unwrap(); + write( + catalog, + "agents/host/owner/agent.kdl", + &selected_catalog_agent("owner", owner_workspace, owner_render), + ); + write( + catalog, + "agents/host/sibling/agent.kdl", + &selected_catalog_agent( + "sibling", + sibling_workspace, + r#"file "SIBLING.txt" "sibling""#, + ), + ); +} + +#[test] +fn selected_catalog_two_agent_kdl_recording_runner_matrix() { + enum Actual { + Missing, + Live, + Dead, + } + + for actual in [Actual::Missing, Actual::Live, Actual::Dead] { + let tmp = tempfile::tempdir().unwrap(); + let catalog = tmp.path().join("catalog"); + let owner_workspace = tmp.path().join("owner-workspace"); + let sibling_workspace = tmp.path().join("sibling-workspace"); + write_selected_catalog( + &catalog, + &owner_workspace, + &sibling_workspace, + r#"file "OWNER.txt" "owner""#, + ); + + let mut sessions = vec![live("host.sibling.work")]; + match actual { + Actual::Missing => {} + Actual::Live => sessions.push(live("host.owner.work")), + Actual::Dead => sessions.push(dead("host.owner.work")), + } + let runner = FakeRunner { + sessions, + ..Default::default() + }; + + let report = up_once_selected(&catalog, "host.owner.work", "host", &runner).unwrap(); + + assert_eq!(runner.list_calls.get(), 1); + assert_eq!( + fs::read_to_string(owner_workspace.join("OWNER.txt")).unwrap(), + "owner" + ); + assert!( + !sibling_workspace.join("SIBLING.txt").exists(), + "the unrelated owner must not be materialized" + ); + assert!(runner.killed.borrow().is_empty()); + assert!(runner.removed.borrow().is_empty()); + match actual { + Actual::Missing => { + assert_eq!(runner.spawned.borrow().as_slice(), ["host.owner.work"]); + assert!(runner.reaped.borrow().is_empty()); + assert_eq!(report.launched, ["host.owner.work"]); + } + Actual::Live => { + assert!(runner.spawned.borrow().is_empty()); + assert!(runner.reaped.borrow().is_empty()); + assert_eq!(report.adopted, ["owner"]); + } + Actual::Dead => { + assert_eq!(runner.reaped.borrow().as_slice(), ["host.owner.work"]); + assert_eq!(runner.spawned.borrow().as_slice(), ["host.owner.work"]); + assert_eq!(report.gc, ["host.owner.work"]); + assert_eq!(report.launched, ["host.owner.work"]); + } + } + assert!( + runner + .spawned + .borrow() + .iter() + .chain(runner.reaped.borrow().iter()) + .all(|id| id == "host.owner.work") + ); + } +} + +#[test] +fn selected_catalog_surfaces_unrelated_malformed_diagnostics_without_blocking_owner() { + let tmp = tempfile::tempdir().unwrap(); + let catalog = tmp.path().join("catalog"); + let owner_workspace = tmp.path().join("owner-workspace"); + let sibling_workspace = tmp.path().join("sibling-workspace"); + write_selected_catalog( + &catalog, + &owner_workspace, + &sibling_workspace, + r#"file "OWNER.txt" "owner""#, + ); + write( + &catalog, + "agents/host/sibling/broken.kdl", + r#"agent "broken" {"#, + ); + let runner = FakeRunner { + sessions: vec![live("host.sibling.work")], + ..Default::default() + }; + + let report = up_once_selected(&catalog, "host.owner.work", "host", &runner).unwrap(); + + assert_eq!(runner.spawned.borrow().as_slice(), ["host.owner.work"]); + assert_eq!( + fs::read_to_string(owner_workspace.join("OWNER.txt")).unwrap(), + "owner" + ); + assert!(!sibling_workspace.join("SIBLING.txt").exists()); + assert!( + report + .errors + .iter() + .any(|error| error.contains("broken.kdl") && error.contains("KDL")), + "{:?}", + report.errors + ); +} + +#[test] +fn selected_catalog_owner_render_failure_refuses_runner_actions() { + let tmp = tempfile::tempdir().unwrap(); + let catalog = tmp.path().join("catalog"); + let owner_workspace = tmp.path().join("owner-workspace"); + let sibling_workspace = tmp.path().join("sibling-workspace"); + write_selected_catalog( + &catalog, + &owner_workspace, + &sibling_workspace, + r#"copy "_templates/missing" "OWNER.txt""#, + ); + let runner = FakeRunner { + sessions: vec![live("host.sibling.work")], + ..Default::default() + }; + + let report = up_once_selected(&catalog, "host.owner.work", "host", &runner).unwrap(); + + assert_eq!(runner.list_calls.get(), 0); + assert_refusal(&runner); + assert!(!owner_workspace.join("OWNER.txt").exists()); + assert!(!sibling_workspace.join("SIBLING.txt").exists()); + assert!( + report + .errors + .iter() + .any(|error| error.contains("_templates/missing")), + "{:?}", + report.errors + ); +} + +#[test] +fn selected_one_shot_unknown_refuses_before_runner_list() { + let runner = FakeRunner::default(); + let error = + up_once_selected_specs(Path::new("/tmp"), &[], "host.missing.task", "host", &runner) + .unwrap_err(); + assert!(error.to_string().contains("did not resolve")); + assert_eq!(runner.list_calls.get(), 0); + assert!(runner.spawned.borrow().is_empty()); + assert!(runner.killed.borrow().is_empty()); + assert!(runner.reaped.borrow().is_empty()); + assert!(runner.removed.borrow().is_empty()); +} + +fn task_spec(identity: &str, host: Option<&str>, id: &str) -> AgentSpec { + AgentSpec { + identity: identity.into(), + host: host.map(str::to_owned), + role: None, + job_type: JobType::Service, + workspace: None, + supervisor: None, + retired: false, + keep: false, + restart: None, + tasks: vec![Task { + kind: TaskKind::Exec, + derived: false, + name: "work".into(), + id: Some(id.into()), + command: Some("true".into()), + argv: None, + cwd: None, + tags: BTreeMap::new(), + env: BTreeMap::new(), + keep: false, + }], + path: "/tmp/spec.kdl".into(), + } +} + +fn assert_refusal(runner: &FakeRunner) { + assert_eq!(runner.list_calls.get(), 0); + assert!(runner.spawned.borrow().is_empty()); + assert!(runner.killed.borrow().is_empty()); + assert!(runner.reaped.borrow().is_empty()); + assert!(runner.removed.borrow().is_empty()); +} + +fn two_task_spec(identity: &str, first: &str, second: &str) -> AgentSpec { + let mut spec = task_spec(identity, None, first); + spec.tasks.push(Task { + kind: TaskKind::Exec, + derived: false, + name: "side".into(), + id: Some(second.into()), + command: Some("true".into()), + argv: None, + cwd: None, + tags: BTreeMap::new(), + env: BTreeMap::new(), + keep: false, + }); + spec +} + +#[test] +fn selected_one_shot_missing_spawns_only_selected_task() { + let runner = FakeRunner { + sessions: vec![live("host.agent.side"), live("host.sibling.work")], + ..Default::default() + }; + let specs = vec![ + two_task_spec("agent", "host.agent.work", "host.agent.side"), + task_spec("sibling", None, "host.sibling.work"), + ]; + let report = up_once_selected_specs( + Path::new("/tmp"), + &specs, + "host.agent.work", + "host", + &runner, + ) + .unwrap(); + assert_eq!(runner.list_calls.get(), 1); + assert_eq!(runner.spawned.borrow().as_slice(), ["host.agent.work"]); + assert!(runner.killed.borrow().is_empty()); + assert!(runner.reaped.borrow().is_empty()); + assert!(runner.removed.borrow().is_empty()); + assert_eq!(report.launched, ["host.agent.work"]); +} + +#[test] +fn selected_one_shot_live_adopts_without_actions() { + let runner = FakeRunner { + sessions: vec![ + live("host.agent.work"), + live("host.agent.side"), + live("host.sibling.work"), + ], + ..Default::default() + }; + let specs = vec![ + two_task_spec("agent", "host.agent.work", "host.agent.side"), + task_spec("sibling", None, "host.sibling.work"), + ]; + let report = up_once_selected_specs( + Path::new("/tmp"), + &specs, + "host.agent.work", + "host", + &runner, + ) + .unwrap(); + assert_eq!(runner.list_calls.get(), 1); + assert!(runner.spawned.borrow().is_empty()); + assert!(runner.killed.borrow().is_empty()); + assert!(runner.reaped.borrow().is_empty()); + assert!(runner.removed.borrow().is_empty()); + assert_eq!(report.adopted, ["agent"]); +} + +#[test] +fn selected_one_shot_dead_reaps_and_relaunches_only_selected() { + let runner = FakeRunner { + sessions: vec![ + dead("host.agent.work"), + live("host.agent.side"), + live("host.sibling.work"), + ], + ..Default::default() + }; + let specs = vec![ + two_task_spec("agent", "host.agent.work", "host.agent.side"), + task_spec("sibling", None, "host.sibling.work"), + ]; + let report = up_once_selected_specs( + Path::new("/tmp"), + &specs, + "host.agent.work", + "host", + &runner, + ) + .unwrap(); + assert_eq!(runner.list_calls.get(), 1); + assert_eq!(runner.reaped.borrow().as_slice(), ["host.agent.work"]); + assert_eq!(runner.spawned.borrow().as_slice(), ["host.agent.work"]); + assert!(runner.killed.borrow().is_empty()); + assert!(runner.removed.borrow().is_empty()); + assert_eq!(report.gc, ["host.agent.work"]); + assert_eq!(report.launched, ["host.agent.work"]); +} + +#[test] +fn selected_one_shot_second_live_pass_is_a_noop() { + let runner = FakeRunner { + sessions: vec![live("host.agent.work"), live("host.sibling.work")], + ..Default::default() + }; + let specs = vec![ + task_spec("agent", None, "host.agent.work"), + task_spec("sibling", None, "host.sibling.work"), + ]; + let report = up_once_selected_specs( + Path::new("/tmp"), + &specs, + "host.agent.work", + "host", + &runner, + ) + .unwrap(); + assert_eq!(runner.list_calls.get(), 1); + assert!(runner.spawned.borrow().is_empty()); + assert!(runner.killed.borrow().is_empty()); + assert!(runner.reaped.borrow().is_empty()); + assert!(runner.removed.borrow().is_empty()); + assert_eq!(report.adopted, ["agent"]); +} + +#[test] +fn selected_one_shot_ambiguous_refuses_before_runner_list() { + let runner = FakeRunner::default(); + let specs = vec![task_spec("one", None, "dup"), task_spec("two", None, "dup")]; + let error = + up_once_selected_specs(Path::new("/tmp"), &specs, "dup", "host", &runner).unwrap_err(); + assert!(error.to_string().contains("ambiguous"), "{error}"); + assert_refusal(&runner); +} + +#[test] +fn selected_one_shot_wrong_host_refuses_before_runner_list() { + let runner = FakeRunner::default(); + let specs = vec![task_spec("remote", Some("other"), "other.remote.work")]; + let error = up_once_selected_specs( + Path::new("/tmp"), + &specs, + "other.remote.work", + "host", + &runner, + ) + .unwrap_err(); + assert!(error.to_string().contains("did not resolve")); + assert_refusal(&runner); +} use st2::{FlappingCap, UpReport, discover, down, execute, reconcile, up_once}; #[derive(Default)] struct FakeRunner { + list_calls: Cell, sessions: Vec, fail_list: bool, fail_spawn: Option, @@ -25,6 +422,7 @@ struct FakeRunner { impl Runner for FakeRunner { fn list_sessions(&self) -> anyhow::Result> { + self.list_calls.set(self.list_calls.get() + 1); if self.fail_list { anyhow::bail!("simulated list failure"); } @@ -64,10 +462,18 @@ fn write(root: &Path, rel: &str, contents: &str) { } fn live(id: &str) -> Session { - Session { pty_id: id.to_string(), alive: true, exit_code: None } + Session { + pty_id: id.to_string(), + alive: true, + exit_code: None, + } } fn dead(id: &str) -> Session { - Session { pty_id: id.to_string(), alive: false, exit_code: None } + Session { + pty_id: id.to_string(), + alive: false, + exit_code: None, + } } /// A v2 service job: a pty agent + an exec ding. @@ -115,7 +521,10 @@ fn up_once_adopts_when_all_tasks_already_live() { fn up_once_launches_only_the_missing_task() { let tmp = tempfile::tempdir().unwrap(); write(tmp.path(), "agents/hetz/demo/agent.toml", AGENT); - let runner = FakeRunner { sessions: vec![live("hetz.demo-claude")], ..Default::default() }; + let runner = FakeRunner { + sessions: vec![live("hetz.demo-claude")], + ..Default::default() + }; let report = up_once(tmp.path(), "hetz", &runner).unwrap(); assert_eq!(report.launched, vec!["hetz.demo.ding"]); } @@ -148,8 +557,16 @@ command = "st2 ding hetz.demo" #[test] fn up_once_skips_other_host_specs() { let tmp = tempfile::tempdir().unwrap(); - write(tmp.path(), "agents/hetz/here/agent.toml", "identity=\"here\"\n[pty.agent]\ncommand=\"x\"\n"); - write(tmp.path(), "agents/silber/there/agent.toml", "identity=\"there\"\n[pty.agent]\ncommand=\"y\"\n"); + write( + tmp.path(), + "agents/hetz/here/agent.toml", + "identity=\"here\"\n[pty.agent]\ncommand=\"x\"\n", + ); + write( + tmp.path(), + "agents/silber/there/agent.toml", + "identity=\"there\"\n[pty.agent]\ncommand=\"y\"\n", + ); let runner = FakeRunner::default(); let report = up_once(tmp.path(), "hetz", &runner).unwrap(); assert_eq!(report.launched.len(), 1); @@ -160,7 +577,10 @@ fn up_once_skips_other_host_specs() { fn up_once_collects_spawn_errors_without_aborting() { let tmp = tempfile::tempdir().unwrap(); write(tmp.path(), "agents/hetz/demo/agent.toml", AGENT); - let runner = FakeRunner { fail_spawn: Some("hetz.demo-claude".into()), ..Default::default() }; + let runner = FakeRunner { + fail_spawn: Some("hetz.demo-claude".into()), + ..Default::default() + }; let report = up_once(tmp.path(), "hetz", &runner).unwrap(); assert_eq!(report.launched, vec!["hetz.demo.ding"]); assert_eq!(report.errors.len(), 1); @@ -242,7 +662,10 @@ fn flapping_cap_parks_a_fail_mode_task_that_keeps_dying() { "identity=\"demo\"\nsupervisor=\"cos-claude\"\n[restart]\nattempts=3\ninterval=\"60s\"\nmode=\"fail\"\n[pty.agent]\nid=\"hetz.demo-claude\"\ncommand=\"x\"\n", ); let found = discover(tmp.path()); - let runner = FakeRunner { sessions: vec![dead("hetz.demo-claude")], ..Default::default() }; + let runner = FakeRunner { + sessions: vec![dead("hetz.demo-claude")], + ..Default::default() + }; let mut cap = FlappingCap::default(); let mut last = UpReport::default(); @@ -298,12 +721,19 @@ fn surface_crash_loop_notifies_the_supervisor_over_the_bus() { let inbox = message::inbox_dir(&tmp.path().join("agents/hetz/cos-claude")); let msgs = message::list_dir(&inbox).unwrap(); - assert_eq!(msgs.len(), 1, "supervisor gets exactly one crash-loop message"); + assert_eq!( + msgs.len(), + 1, + "supervisor gets exactly one crash-loop message" + ); let m = &msgs[0]; assert_eq!(m.from.as_deref(), Some("st2.hetz")); // the runner is the sender assert_eq!(m.subject.as_deref(), Some("crash-loop: hetz.demo parked")); assert!(m.tags.contains(&"crash-loop".to_string())); - assert!(m.body.contains("hetz.demo-claude"), "body names the parked task"); + assert!( + m.body.contains("hetz.demo-claude"), + "body names the parked task" + ); } /// `st2 down` kills every LIVE task of THIS host's catalog agents (the explicit teardown), skips @@ -311,13 +741,29 @@ fn surface_crash_loop_notifies_the_supervisor_over_the_bus() { #[test] fn down_tears_down_this_hosts_live_tasks_only() { let tmp = tempfile::tempdir().unwrap(); - write(tmp.path(), "agents/hetz/demo/agent.toml", "identity=\"demo\"\n[pty.agent]\nid=\"hetz.demo-claude\"\ncommand=\"x\"\n"); - write(tmp.path(), "agents/hetz/dead/agent.toml", "identity=\"dead\"\n[pty.agent]\nid=\"hetz.dead\"\ncommand=\"x\"\n"); - write(tmp.path(), "agents/silber/other/agent.toml", "identity=\"other\"\nhost=\"silber\"\n[pty.agent]\nid=\"silber.other\"\ncommand=\"x\"\n"); + write( + tmp.path(), + "agents/hetz/demo/agent.toml", + "identity=\"demo\"\n[pty.agent]\nid=\"hetz.demo-claude\"\ncommand=\"x\"\n", + ); + write( + tmp.path(), + "agents/hetz/dead/agent.toml", + "identity=\"dead\"\n[pty.agent]\nid=\"hetz.dead\"\ncommand=\"x\"\n", + ); + write( + tmp.path(), + "agents/silber/other/agent.toml", + "identity=\"other\"\nhost=\"silber\"\n[pty.agent]\nid=\"silber.other\"\ncommand=\"x\"\n", + ); // demo is live, dead is dead, other belongs to another host + is live. let runner = FakeRunner { - sessions: vec![live("hetz.demo-claude"), dead("hetz.dead"), live("silber.other")], + sessions: vec![ + live("hetz.demo-claude"), + dead("hetz.dead"), + live("silber.other"), + ], ..Default::default() }; let report = down(tmp.path(), "hetz", &runner).unwrap(); @@ -351,9 +797,21 @@ fn surface_crash_loop_without_supervisor_sends_nothing() { #[test] fn up_once_surfaces_discovery_errors_and_unrunnable() { let tmp = tempfile::tempdir().unwrap(); - write(tmp.path(), "agents/hetz/good/agent.toml", "identity=\"good\"\n[pty.agent]\ncommand=\"x\"\n"); - write(tmp.path(), "agents/hetz/bad/agent.toml", "identity=\"b\"\nnot valid ="); - write(tmp.path(), "agents/hetz/nr/agent.toml", "identity=\"nr\"\ntype=\"service\"\n"); + write( + tmp.path(), + "agents/hetz/good/agent.toml", + "identity=\"good\"\n[pty.agent]\ncommand=\"x\"\n", + ); + write( + tmp.path(), + "agents/hetz/bad/agent.toml", + "identity=\"b\"\nnot valid =", + ); + write( + tmp.path(), + "agents/hetz/nr/agent.toml", + "identity=\"nr\"\ntype=\"service\"\n", + ); let runner = FakeRunner::default(); let report = up_once(tmp.path(), "hetz", &runner).unwrap(); assert_eq!(report.launched, vec!["hetz.good.agent"]); diff --git a/tests/targeted_reconcile.rs b/tests/targeted_reconcile.rs new file mode 100644 index 0000000..85b1b06 --- /dev/null +++ b/tests/targeted_reconcile.rs @@ -0,0 +1,437 @@ +//! R19 executable proofs for the `st2 up --once --task ` CLI path. +//! +//! The exec-backed tests use a recording `pty` shim to prove refusal/listing order without touching +//! a real registry. The lifecycle test uses an isolated real `PTY_ROOT` and keeps an unrelated +//! sibling's PID + creation generation fixed across selected missing/live/replacement passes. + +use std::fs; +use std::os::unix::fs::PermissionsExt; +use std::path::{Path, PathBuf}; +use std::process::{Command, Output, Stdio}; +use std::time::{Duration, Instant}; + +const HOST: &str = "targeted"; +const OWNER: &str = "targeted.owner.work"; +const SIBLING: &str = "targeted.sibling.work"; + +fn write(path: &Path, contents: impl AsRef<[u8]>) { + fs::create_dir_all(path.parent().unwrap()).unwrap(); + fs::write(path, contents).unwrap(); +} + +fn executable(path: &Path, body: &str) { + fs::write(path, body).unwrap(); + fs::set_permissions(path, fs::Permissions::from_mode(0o755)).unwrap(); +} + +fn agent_kdl(identity: &str, kind: &str, task_id: &str, workspace: &Path, marker: &str) -> String { + format!( + r#"agent "{identity}" {{ + host "{HOST}" + type "service" + workspace "{}" + {kind} "work" {{ + id "{task_id}" + command "sleep 120" + }} + render {{ + file "{marker}" "{identity}" + }} +}} +"#, + workspace.display() + ) +} + +fn write_two_agent_catalog( + catalog: &Path, + kind: &str, + owner_workspace: &Path, + sibling_workspace: &Path, +) { + fs::create_dir_all(owner_workspace).unwrap(); + fs::create_dir_all(sibling_workspace).unwrap(); + write( + &catalog.join("agents/targeted/owner/agent.kdl"), + agent_kdl("owner", kind, OWNER, owner_workspace, "OWNER.txt"), + ); + write( + &catalog.join("agents/targeted/sibling/agent.kdl"), + agent_kdl("sibling", kind, SIBLING, sibling_workspace, "SIBLING.txt"), + ); +} + +fn prepend_path(directory: &Path) -> String { + format!( + "{}:{}", + directory.display(), + std::env::var("PATH").unwrap_or_default() + ) +} + +fn selected_once(catalog: &Path, xdg: &Path, pty_root: &Path, selector: &str) -> Output { + Command::new(env!("CARGO_BIN_EXE_st2")) + .args(["up", "--catalog"]) + .arg(catalog) + .args(["--host", HOST, "--once", "--task", selector]) + .env("XDG_STATE_HOME", xdg) + .env("PTY_ROOT", pty_root) + .output() + .unwrap() +} + +fn assert_success(output: &Output, context: &str) { + assert!( + output.status.success(), + "{context}\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); +} + +fn read_pid(path: &Path) -> Option { + fs::read_to_string(path).ok()?.trim().parse().ok() +} + +fn kill_process_group(pid: i32) { + for target in [format!("-{pid}"), pid.to_string()] { + let _ = Command::new("kill") + .arg("-KILL") + .arg(target) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status(); + } +} + +struct ExecCleanup { + pidfiles: Vec, +} + +impl Drop for ExecCleanup { + fn drop(&mut self) { + for pidfile in &self.pidfiles { + if let Some(pid) = read_pid(pidfile) { + kill_process_group(pid); + } + } + } +} + +#[test] +fn targeted_once_cli_resolves_before_listing_and_runs_only_the_selected_exec() { + let tmp = tempfile::tempdir().unwrap(); + let catalog = tmp.path().join("catalog"); + let xdg = tmp.path().join("xdg"); + let pty_root = tmp.path().join("pty"); + let owner_workspace = tmp.path().join("owner-workspace"); + let sibling_workspace = tmp.path().join("sibling-workspace"); + let shim_bin = tmp.path().join("bin"); + let pty_calls = tmp.path().join("pty-calls"); + fs::create_dir_all(&shim_bin).unwrap(); + executable( + &shim_bin.join("pty"), + "#!/bin/sh\nprintf '%s\\n' \"$*\" >> \"$PTY_CALLS\"\n[ \"$1\" = list ] && printf '[]\\n' && exit 0\nexit 97\n", + ); + write_two_agent_catalog(&catalog, "exec", &owner_workspace, &sibling_workspace); + + let owner_pidfile = xdg + .join("st2") + .join(HOST) + .join("exec") + .join(format!("{OWNER}.pid")); + let sibling_pidfile = xdg + .join("st2") + .join(HOST) + .join("exec") + .join(format!("{SIBLING}.pid")); + let _cleanup = ExecCleanup { + pidfiles: vec![owner_pidfile.clone(), sibling_pidfile.clone()], + }; + let invoke = |selector: &str| { + Command::new(env!("CARGO_BIN_EXE_st2")) + .args(["up", "--catalog"]) + .arg(&catalog) + .args(["--host", HOST, "--once", "--task", selector]) + .env("PATH", prepend_path(&shim_bin)) + .env("PTY_CALLS", &pty_calls) + .env("XDG_STATE_HOME", &xdg) + .env("PTY_ROOT", &pty_root) + .output() + .unwrap() + }; + + let refused = invoke("targeted.missing.work"); + assert!(!refused.status.success()); + assert!( + !pty_calls.exists(), + "an unknown selector must refuse before `pty list`" + ); + assert!(!owner_workspace.join("OWNER.txt").exists()); + assert!(!sibling_workspace.join("SIBLING.txt").exists()); + + let launched = invoke(OWNER); + assert_success(&launched, "selected exec launch failed"); + let stdout = String::from_utf8_lossy(&launched.stdout); + assert!( + stdout.contains(&format!("launched (1): {OWNER}")), + "{stdout}" + ); + assert_eq!( + fs::read_to_string(owner_workspace.join("OWNER.txt")).unwrap(), + "owner" + ); + assert!(!sibling_workspace.join("SIBLING.txt").exists()); + assert!(owner_pidfile.exists()); + assert!(!sibling_pidfile.exists()); + assert_eq!( + fs::read_to_string(&pty_calls) + .unwrap() + .lines() + .collect::>(), + ["list --json"] + ); + + let owner_pid = read_pid(&owner_pidfile).unwrap(); + let adopted = invoke(OWNER); + assert_success(&adopted, "selected live exec adoption failed"); + assert_eq!(read_pid(&owner_pidfile), Some(owner_pid)); + assert!(!sibling_pidfile.exists()); + assert!( + String::from_utf8_lossy(&adopted.stdout).contains("adopted (1): owner"), + "{}", + String::from_utf8_lossy(&adopted.stdout) + ); +} + +#[test] +fn targeted_once_cli_owner_render_failure_is_nonzero_before_listing() { + let tmp = tempfile::tempdir().unwrap(); + let catalog = tmp.path().join("catalog"); + let xdg = tmp.path().join("xdg"); + let pty_root = tmp.path().join("pty"); + let workspace = tmp.path().join("workspace"); + let shim_bin = tmp.path().join("bin"); + let pty_calls = tmp.path().join("pty-calls"); + fs::create_dir_all(&workspace).unwrap(); + fs::create_dir_all(&shim_bin).unwrap(); + executable( + &shim_bin.join("pty"), + "#!/bin/sh\nprintf '%s\\n' \"$*\" >> \"$PTY_CALLS\"\nprintf '[]\\n'\n", + ); + write( + &catalog.join("agents/targeted/owner/agent.kdl"), + format!( + r#"agent "owner" {{ + host "{HOST}" + type "service" + workspace "{}" + exec "work" {{ + id "{OWNER}" + command "sleep 120" + }} + render {{ + copy "_templates/missing" "OWNER.txt" + }} +}} +"#, + workspace.display() + ), + ); + + let output = Command::new(env!("CARGO_BIN_EXE_st2")) + .args(["up", "--catalog"]) + .arg(&catalog) + .args(["--host", HOST, "--once", "--task", OWNER]) + .env("PATH", prepend_path(&shim_bin)) + .env("PTY_CALLS", &pty_calls) + .env("XDG_STATE_HOME", &xdg) + .env("PTY_ROOT", &pty_root) + .output() + .unwrap(); + + assert!(!output.status.success()); + assert!( + !pty_calls.exists(), + "owner render refusal must occur before `pty list`" + ); + assert!(!workspace.join("OWNER.txt").exists()); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!(stderr.contains("_templates/missing"), "{stderr}"); + assert!( + stderr.contains("targeted one-shot reconcile pass reported errors"), + "{stderr}" + ); +} + +fn pty_available() -> bool { + Command::new("pty") + .arg("--help") + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status() + .map(|status| status.success()) + .unwrap_or(false) +} + +fn pty_gate(test: &str) -> bool { + if pty_available() { + return true; + } + assert!( + std::env::var_os("ST2_ALLOW_PTY_SKIP").is_some(), + "{test}: `pty` is not on PATH, so the real-PTY targeted reconcile gate is unproven. \ + Install `pty`, or set ST2_ALLOW_PTY_SKIP=1 for a local opt-out." + ); + eprintln!("SKIP {test}: `pty` not on PATH (ST2_ALLOW_PTY_SKIP set)"); + false +} + +#[derive(Clone, Debug, PartialEq, Eq)] +struct PtyGeneration { + pid: i64, + created_at: String, +} + +fn pty_generation(pty_root: &Path, id: &str) -> Option { + let output = Command::new("pty") + .args(["list", "--json"]) + .env("PTY_ROOT", pty_root) + .output() + .ok()?; + if !output.status.success() { + return None; + } + let rows: serde_json::Value = serde_json::from_slice(&output.stdout).ok()?; + rows.as_array()? + .iter() + .find(|row| row["name"].as_str() == Some(id) && row["status"].as_str() == Some("running")) + .and_then(|row| { + Some(PtyGeneration { + pid: row["pid"].as_i64()?, + created_at: row["createdAt"].as_str()?.to_owned(), + }) + }) +} + +fn wait_for_generation(pty_root: &Path, id: &str) -> PtyGeneration { + let deadline = Instant::now() + Duration::from_secs(5); + loop { + if let Some(generation) = pty_generation(pty_root, id) { + return generation; + } + assert!( + Instant::now() < deadline, + "session {id} did not become running in {}", + pty_root.display() + ); + std::thread::sleep(Duration::from_millis(50)); + } +} + +struct PtyCleanup { + root: PathBuf, + ids: Vec<&'static str>, +} + +impl Drop for PtyCleanup { + fn drop(&mut self) { + for id in &self.ids { + let _ = Command::new("pty") + .args(["kill", id]) + .env("PTY_ROOT", &self.root) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status(); + } + std::thread::sleep(Duration::from_millis(600)); + for id in &self.ids { + let _ = Command::new("pty") + .args(["rm", id]) + .env("PTY_ROOT", &self.root) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status(); + } + } +} + +#[test] +fn targeted_once_real_pty_preserves_sibling_generation_across_selected_lifecycle() { + if !pty_gate("targeted_once_real_pty_preserves_sibling_generation_across_selected_lifecycle") { + return; + } + + let tmp = tempfile::tempdir().unwrap(); + let catalog = tmp.path().join("catalog"); + let xdg = tmp.path().join("xdg"); + let pty_root = tmp.path().join("pty"); + let owner_workspace = tmp.path().join("owner-workspace"); + let sibling_workspace = tmp.path().join("sibling-workspace"); + fs::create_dir_all(&pty_root).unwrap(); + write_two_agent_catalog(&catalog, "pty", &owner_workspace, &sibling_workspace); + let _cleanup = PtyCleanup { + root: pty_root.clone(), + ids: vec![OWNER, SIBLING], + }; + + let sibling_boot = Command::new("pty") + .args(["run", "-d", "--id", SIBLING, "--", "sleep", "120"]) + .env("PTY_ROOT", &pty_root) + .output() + .unwrap(); + assert_success(&sibling_boot, "failed to seed the unrelated sibling PTY"); + let sibling_generation = wait_for_generation(&pty_root, SIBLING); + + let launched = selected_once(&catalog, &xdg, &pty_root, OWNER); + assert_success(&launched, "selected missing PTY launch failed"); + let owner_generation = wait_for_generation(&pty_root, OWNER); + assert_eq!( + pty_generation(&pty_root, SIBLING), + Some(sibling_generation.clone()) + ); + assert_eq!( + fs::read_to_string(owner_workspace.join("OWNER.txt")).unwrap(), + "owner" + ); + assert!(!sibling_workspace.join("SIBLING.txt").exists()); + let stdout = String::from_utf8_lossy(&launched.stdout); + assert!( + stdout.contains(&format!("launched (1): {OWNER}")), + "{stdout}" + ); + + let adopted = selected_once(&catalog, &xdg, &pty_root, OWNER); + assert_success(&adopted, "selected live PTY adoption failed"); + assert_eq!( + pty_generation(&pty_root, OWNER), + Some(owner_generation.clone()) + ); + assert_eq!( + pty_generation(&pty_root, SIBLING), + Some(sibling_generation.clone()) + ); + assert!( + String::from_utf8_lossy(&adopted.stdout).contains("adopted (1): owner"), + "{}", + String::from_utf8_lossy(&adopted.stdout) + ); + + let killed = Command::new("kill") + .args(["-KILL", &owner_generation.pid.to_string()]) + .status() + .unwrap(); + assert!(killed.success(), "failed to hard-kill selected owner PTY"); + let relaunched = selected_once(&catalog, &xdg, &pty_root, OWNER); + assert_success(&relaunched, "selected dead PTY relaunch failed"); + let replacement_generation = wait_for_generation(&pty_root, OWNER); + assert_ne!(replacement_generation, owner_generation); + assert_eq!(pty_generation(&pty_root, SIBLING), Some(sibling_generation)); + assert!(!sibling_workspace.join("SIBLING.txt").exists()); + let stdout = String::from_utf8_lossy(&relaunched.stdout); + assert!( + stdout.contains(&format!("launched (1): {OWNER}")), + "{stdout}" + ); + assert!(!stdout.contains(SIBLING), "{stdout}"); +} diff --git a/tests/validate.rs b/tests/validate.rs index 0e0783a..0894f7a 100644 --- a/tests/validate.rs +++ b/tests/validate.rs @@ -200,7 +200,7 @@ fn ls_marks_a_generated_ding_only_agent_as_unrendered() { .unwrap(); assert!(output.status.success()); assert!( - String::from_utf8_lossy(&output.stdout).contains("[UNRENDERED: no task command]"), + String::from_utf8_lossy(&output.stdout).contains("[UNRENDERED: no task launch]"), "stdout:\n{}", String::from_utf8_lossy(&output.stdout) );