From 1c10b5cb43f37dcc080cc10abdb40a8f5755685c Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Wed, 29 Jul 2026 05:29:11 +0200 Subject: [PATCH 01/37] feat: resolve exact task selectors purely --- src/reconcile.rs | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/src/reconcile.rs b/src/reconcile.rs index 1b08136..4a0c102 100644 --- a/src/reconcile.rs +++ b/src/reconcile.rs @@ -80,6 +80,24 @@ 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"), + } +} + /// The state of a declared task's session in the ACTUAL world. enum SessionState { Alive, From 37cad5868b3403df41c9308e931b2d303f08e7c9 Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Wed, 29 Jul 2026 05:31:05 +0200 Subject: [PATCH 02/37] test: cover exact task selector resolution --- tests/reconcile.rs | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/tests/reconcile.rs b/tests/reconcile.rs index 5eca7d1..49cafa8 100644 --- a/tests/reconcile.rs +++ b/tests/reconcile.rs @@ -5,6 +5,22 @@ use std::path::PathBuf; use st2::spec::{AgentSpec, JobType, Task, TaskKind}; use st2::{Session, reconcile}; +use st2::reconcile::resolve_task; + +#[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()); +} fn task(kind: TaskKind, name: &str, id: Option<&str>, command: Option<&str>) -> Task { Task { From 8533c56bfe8c2e3f8c5818bbe8cbd9cf7a5fb744 Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Wed, 29 Jul 2026 05:32:19 +0200 Subject: [PATCH 03/37] test: complete task selector matrix --- tests/reconcile.rs | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/tests/reconcile.rs b/tests/reconcile.rs index 49cafa8..d5261f3 100644 --- a/tests/reconcile.rs +++ b/tests/reconcile.rs @@ -22,6 +22,27 @@ fn exact_task_selector_matrix() { 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"] { assert_eq!(resolve_task(&s, selector, "host").unwrap().2, "custom"); } +} + +#[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, From 9d124b3cdc549198d52b1c0ff4e0f43ff071d2b6 Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Wed, 29 Jul 2026 05:34:16 +0200 Subject: [PATCH 04/37] test: prove selector owner fields and immutability --- tests/reconcile.rs | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/tests/reconcile.rs b/tests/reconcile.rs index d5261f3..e047dc4 100644 --- a/tests/reconcile.rs +++ b/tests/reconcile.rs @@ -32,7 +32,15 @@ fn selector_derived_id_success() { #[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"] { assert_eq!(resolve_task(&s, selector, "host").unwrap().2, "custom"); } + 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] From 934601c4b151e04429edacb370d81ab9c125d173 Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Wed, 29 Jul 2026 05:35:57 +0200 Subject: [PATCH 05/37] feat: add pure selected reconcile plan --- src/reconcile.rs | 13 +++++++++++++ tests/reconcile.rs | 9 +++++++++ 2 files changed, 22 insertions(+) diff --git a/src/reconcile.rs b/src/reconcile.rs index 4a0c102..91557b8 100644 --- a/src/reconcile.rs +++ b/src/reconcile.rs @@ -98,6 +98,19 @@ pub fn resolve_task<'a>(specs: &'a [AgentSpec], selector: &str, this_host: &str) } } +/// 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, _, runtime) = resolve_task(specs, selector, this_host)?; + let scoped_sessions: Vec<_> = sessions.iter().filter(|s| s.pty_id == runtime).cloned().collect(); + let mut plan = reconcile(specs, &scoped_sessions, this_host); + plan.launch.retain_mut(|launch| { launch.tasks.retain(|t| t.pty_id == runtime); !launch.tasks.is_empty() }); + plan.teardown.retain(|t| t.pty_ids.iter().any(|id| id == &runtime)); + plan.gc.retain(|id| id == &runtime); + plan.adopt.retain(|spec| spec.tasks.iter().any(|t| t.id.as_deref() == Some(&runtime) || format!("{}.{}", spec.bus_id(this_host), t.name) == runtime)); + if scoped_sessions.iter().any(|s| s.alive) && plan.launch.is_empty() && !plan.adopt.iter().any(|s| std::ptr::eq(*s, owner)) { plan.adopt.push(owner); } + Ok(plan) +} + /// The state of a declared task's session in the ACTUAL world. enum SessionState { Alive, diff --git a/tests/reconcile.rs b/tests/reconcile.rs index e047dc4..b3f68fe 100644 --- a/tests/reconcile.rs +++ b/tests/reconcile.rs @@ -6,6 +6,7 @@ use std::path::PathBuf; use st2::spec::{AgentSpec, JobType, Task, TaskKind}; use st2::{Session, reconcile}; use st2::reconcile::resolve_task; +use st2::reconcile::reconcile_selected; #[test] fn exact_task_selector_matrix() { @@ -43,6 +44,14 @@ fn selector_agent_command_runtime_id_and_inputs_immutable() { 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")], "host", "host.a.x").unwrap(); assert!(plan2.launch.is_empty()); assert_eq!(plan2.adopt.len(), 1); +} + #[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"))])]; From 2b1a587ae697687c75297fc09f32f24ba71370e3 Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Wed, 29 Jul 2026 05:55:26 +0200 Subject: [PATCH 06/37] fix: construct selected reconcile plan directly --- src/reconcile.rs | 17 +++++++++-------- 1 file changed, 9 insertions(+), 8 deletions(-) diff --git a/src/reconcile.rs b/src/reconcile.rs index 91557b8..d240080 100644 --- a/src/reconcile.rs +++ b/src/reconcile.rs @@ -100,14 +100,15 @@ pub fn resolve_task<'a>(specs: &'a [AgentSpec], selector: &str, this_host: &str) /// 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, _, runtime) = resolve_task(specs, selector, this_host)?; - let scoped_sessions: Vec<_> = sessions.iter().filter(|s| s.pty_id == runtime).cloned().collect(); - let mut plan = reconcile(specs, &scoped_sessions, this_host); - plan.launch.retain_mut(|launch| { launch.tasks.retain(|t| t.pty_id == runtime); !launch.tasks.is_empty() }); - plan.teardown.retain(|t| t.pty_ids.iter().any(|id| id == &runtime)); - plan.gc.retain(|id| id == &runtime); - plan.adopt.retain(|spec| spec.tasks.iter().any(|t| t.id.as_deref() == Some(&runtime) || format!("{}.{}", spec.bus_id(this_host), t.name) == runtime)); - if scoped_sessions.iter().any(|s| s.alive) && plan.launch.is_empty() && !plan.adopt.iter().any(|s| std::ptr::eq(*s, owner)) { plan.adopt.push(owner); } + 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 !owner.keep { plan.gc.push(runtime); } } return Ok(plan); } + let Some(command) = task.command.clone() else { plan.unrunnable.push(owner); return Ok(plan); }; + 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(), command, 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 || (s.exit_code.is_none() && target.keep) => plan.adopt.push(owner), Some(_) if !target.keep => { plan.gc.push(runtime); plan.launch.push(Launch { spec: owner, tasks: vec![target] }); }, _ => plan.launch.push(Launch { spec: owner, tasks: vec![target] }) } Ok(plan) } From 1d967251155e8732cc5e9643b8622c4f0bbd09bd Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Wed, 29 Jul 2026 05:57:37 +0200 Subject: [PATCH 07/37] fix: preserve selected keep task semantics --- src/reconcile.rs | 4 ++-- tests/reconcile.rs | 8 ++++++++ 2 files changed, 10 insertions(+), 2 deletions(-) diff --git a/src/reconcile.rs b/src/reconcile.rs index d240080..aaf3e16 100644 --- a/src/reconcile.rs +++ b/src/reconcile.rs @@ -103,12 +103,12 @@ pub fn reconcile_selected<'a>(specs: &'a [AgentSpec], sessions: &[Session], this 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 !owner.keep { plan.gc.push(runtime); } } return Ok(plan); } + 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 Some(command) = task.command.clone() else { plan.unrunnable.push(owner); return Ok(plan); }; 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(), command, 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 || (s.exit_code.is_none() && target.keep) => plan.adopt.push(owner), Some(_) if !target.keep => { plan.gc.push(runtime); plan.launch.push(Launch { spec: owner, tasks: vec![target] }); }, _ => plan.launch.push(Launch { spec: owner, tasks: vec![target] }) } + 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) } diff --git a/tests/reconcile.rs b/tests/reconcile.rs index b3f68fe..6ece55c 100644 --- a/tests/reconcile.rs +++ b/tests/reconcile.rs @@ -52,6 +52,14 @@ fn selected_reconcile_launches_missing_and_adopts_live_without_siblings() { let plan2 = reconcile_selected(&specs, &[live("host.a.x")], "host", "host.a.x").unwrap(); assert!(plan2.launch.is_empty()); assert_eq!(plan2.adopt.len(), 1); } +#[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 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"))])]; From 9577f0a221777f213e435943f81a6cbc2b89124f Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Wed, 29 Jul 2026 05:59:13 +0200 Subject: [PATCH 08/37] test: cover selected reconcile action matrix --- tests/reconcile.rs | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/tests/reconcile.rs b/tests/reconcile.rs index 6ece55c..14115de 100644 --- a/tests/reconcile.rs +++ b/tests/reconcile.rs @@ -60,6 +60,14 @@ fn selected_reconcile_freezes_dead_keep_and_retired_task_keep() { 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 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"))])]; From e7775113bc092255854ce35b0e3ed48b0a8e342a Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Wed, 29 Jul 2026 06:00:55 +0200 Subject: [PATCH 09/37] test: complete selected reconcile matrix --- tests/reconcile.rs | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/tests/reconcile.rs b/tests/reconcile.rs index 14115de..ad90676 100644 --- a/tests/reconcile.rs +++ b/tests/reconcile.rs @@ -68,6 +68,11 @@ fn selected_reconcile_action_ids_are_exact_and_refusals_immutable() { 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[0].tasks[0].pty_id,"host.a.x"); } +#[test] fn selected_retired_live_tears_down_only_selected() { let mut s=svc("a",None,vec![task(TaskKind::Exec,"x",None,Some("a"))]); s.retired=true; let specs=[s]; let p=reconcile_selected(&specs,&[live("host.a.x"),live("host.a.other")],"host","host.a.x").unwrap(); assert_eq!(p.teardown[0].pty_ids,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"))])]; From e35b6e7493659f11711f667eb03f72a658b2ecf6 Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Wed, 29 Jul 2026 06:02:40 +0200 Subject: [PATCH 10/37] test: format and strengthen selected reconcile proofs --- src/reconcile.rs | 104 ++++++++++-- tests/reconcile.rs | 394 +++++++++++++++++++++++++++++++++++++++------ 2 files changed, 432 insertions(+), 66 deletions(-) diff --git a/src/reconcile.rs b/src/reconcile.rs index aaf3e16..78fee48 100644 --- a/src/reconcile.rs +++ b/src/reconcile.rs @@ -81,14 +81,25 @@ pub struct ReconcilePlan<'a> { } /// 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)> { +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; } + 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 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)); } + if selector == runtime || selector == qualified { + matches.push((spec, task, runtime)); + } } } match matches.as_slice() { @@ -99,16 +110,65 @@ pub fn resolve_task<'a>(specs: &'a [AgentSpec], selector: &str, this_host: &str) } /// 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> { +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 Some(command) = task.command.clone() else { plan.unrunnable.push(owner); return Ok(plan); }; - 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(), command, 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] }) } + 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 Some(command) = task.command.clone() else { + plan.unrunnable.push(owner); + return Ok(plan); + }; + 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(), + command, + 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) } @@ -136,9 +196,15 @@ fn resolve_task_id(bus_id: &str, name: &str, explicit: Option<&str>) -> String { } /// Compute the reconcile plan for `specs` given observed `sessions`, filtering to `this_host`. -pub fn reconcile<'a>(specs: &'a [AgentSpec], sessions: &[Session], this_host: &str) -> ReconcilePlan<'a> { - let by_id: HashMap<&str, bool> = - sessions.iter().map(|s| (s.pty_id.as_str(), s.alive)).collect(); +pub fn reconcile<'a>( + specs: &'a [AgentSpec], + sessions: &[Session], + this_host: &str, +) -> ReconcilePlan<'a> { + let by_id: HashMap<&str, bool> = sessions + .iter() + .map(|s| (s.pty_id.as_str(), s.alive)) + .collect(); let mut plan = ReconcilePlan::default(); for spec in specs { @@ -160,7 +226,10 @@ pub fn reconcile<'a>(specs: &'a [AgentSpec], sessions: &[Session], this_host: &s } } if !teardown_ids.is_empty() { - plan.teardown.push(Teardown { spec, pty_ids: teardown_ids }); + plan.teardown.push(Teardown { + spec, + pty_ids: teardown_ids, + }); } continue; } @@ -217,7 +286,10 @@ pub fn reconcile<'a>(specs: &'a [AgentSpec], sessions: &[Session], this_host: &s if to_launch.is_empty() { plan.adopt.push(spec); } else { - plan.launch.push(Launch { spec, tasks: to_launch }); + plan.launch.push(Launch { + spec, + tasks: to_launch, + }); } } plan diff --git a/tests/reconcile.rs b/tests/reconcile.rs index ad90676..a6d8a02 100644 --- a/tests/reconcile.rs +++ b/tests/reconcile.rs @@ -3,20 +3,41 @@ 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}; -use st2::reconcile::resolve_task; -use st2::reconcile::reconcile_selected; #[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"))]), + 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_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()); @@ -25,59 +46,278 @@ fn exact_task_selector_matrix() { #[test] fn selector_derived_id_success() { - let s = vec![svc("plain", None, vec![task(TaskKind::Exec, "work", None, Some("cmd"))])]; + 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")); + 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")); } + 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); + 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 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")], "host", "host.a.x").unwrap(); assert!(plan2.launch.is_empty()); assert_eq!(plan2.adopt.len(), 1); + 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")], "host", "host.a.x").unwrap(); + assert!(plan2.launch.is_empty()); + assert_eq!(plan2.adopt.len(), 1); } #[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()); + 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); + 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[0].tasks[0].pty_id,"host.a.x"); } -#[test] fn selected_retired_live_tears_down_only_selected() { let mut s=svc("a",None,vec![task(TaskKind::Exec,"x",None,Some("a"))]); s.retired=true; let specs=[s]; let p=reconcile_selected(&specs,&[live("host.a.x"),live("host.a.other")],"host","host.a.x").unwrap(); assert_eq!(p.teardown[0].pty_ids,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 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"))])]; + 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"))])]; + 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()); } @@ -95,7 +335,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), @@ -107,7 +353,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") + )), } } @@ -116,10 +365,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"; @@ -130,8 +387,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); @@ -176,7 +443,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 @@ -210,8 +481,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); @@ -222,7 +506,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"); // . @@ -230,7 +518,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); @@ -253,12 +545,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", @@ -289,10 +576,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] From fc719b21f8b6d45f977e85d4c91f21ff5f96b97a Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Wed, 29 Jul 2026 06:04:09 +0200 Subject: [PATCH 11/37] test: prove selected second-pass sibling no-op --- tests/reconcile.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/tests/reconcile.rs b/tests/reconcile.rs index a6d8a02..c4f2ff2 100644 --- a/tests/reconcile.rs +++ b/tests/reconcile.rs @@ -129,9 +129,9 @@ fn selected_reconcile_launches_missing_and_adopts_live_without_siblings() { 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")], "host", "host.a.x").unwrap(); - assert!(plan2.launch.is_empty()); - assert_eq!(plan2.adopt.len(), 1); + 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] From 0ffee0c04b92d7264de6b53335d4320285df4efc Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Wed, 29 Jul 2026 06:24:21 +0200 Subject: [PATCH 12/37] feat: add exact task materialization selector --- src/main.rs | 13 ++++++++++++- 1 file changed, 12 insertions(+), 1 deletion(-) diff --git a/src/main.rs b/src/main.rs index e1ce616..12dfff8 100644 --- a/src/main.rs +++ b/src/main.rs @@ -56,6 +56,8 @@ enum Command { /// Limit materialization/reconciliation to one declared agent identity. #[arg(long)] agent: Option, + #[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)] @@ -529,9 +531,12 @@ 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() && !once && !materialize_only { 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), @@ -1723,6 +1728,7 @@ 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. @@ -1740,6 +1746,11 @@ 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 From 57322690a4a14da3e069d6bd6a755ddc191919bb Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Wed, 29 Jul 2026 06:25:38 +0200 Subject: [PATCH 13/37] test: reject unknown task selector before mutation --- tests/materialize.rs | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/tests/materialize.rs b/tests/materialize.rs index 9c80095..70e81ba 100644 --- a/tests/materialize.rs +++ b/tests/materialize.rs @@ -10,6 +10,18 @@ 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); +} + fn spec(catalog: &Path, identity: &str) -> AgentSpec { discover(catalog) .specs From 10b94d74c108d6de2488619d9f1b091425ac7115 Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Wed, 29 Jul 2026 06:25:56 +0200 Subject: [PATCH 14/37] fix: reject deferred task once selection --- src/main.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/main.rs b/src/main.rs index 12dfff8..5051cc2 100644 --- a/src/main.rs +++ b/src/main.rs @@ -534,7 +534,8 @@ fn main() -> Result<()> { task, } => { let root = catalog_arg(root)?; - if task.is_some() && !once && !materialize_only { anyhow::bail!("--task requires --once or --materialize-only"); } + if task.is_some() && once { anyhow::bail!("--task --once is deferred until selected reconcile support"); } + if task.is_some() && !materialize_only { anyhow::bail!("--task requires --materialize-only in this stage"); } if agent.is_some() && !materialize_only { anyhow::bail!("--agent requires --materialize-only"); } up(&root, host, once, materialize_only, interval, agent, task) } From b77a3f71d46af87aa720e42ed2260b07d24d5d7b Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Wed, 29 Jul 2026 06:27:46 +0200 Subject: [PATCH 15/37] test: cover task selector CLI refusal modes --- tests/materialize.rs | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/tests/materialize.rs b/tests/materialize.rs index 70e81ba..3dd7c4a 100644 --- a/tests/materialize.rs +++ b/tests/materialize.rs @@ -22,6 +22,14 @@ fn task_selector_refusal_is_nonzero_before_catalog_mutation() { assert_eq!(fs::read_dir(tmp.path()).unwrap().count(), before); } +#[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(), "--once", "--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()); + } +} + fn spec(catalog: &Path, identity: &str) -> AgentSpec { discover(catalog) .specs From fadadf35d4a812ff4ac82dccc7742a95624242d6 Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Wed, 29 Jul 2026 06:55:07 +0200 Subject: [PATCH 16/37] test: prove task selector owner-only materialization --- tests/materialize.rs | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) diff --git a/tests/materialize.rs b/tests/materialize.rs index 3dd7c4a..8bcba6b 100644 --- a/tests/materialize.rs +++ b/tests/materialize.rs @@ -22,11 +22,23 @@ fn task_selector_refusal_is_nonzero_before_catalog_mutation() { 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!(owner.join("OWNER.txt").exists()); assert!(!sibling.join("SIBLING.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(), "--once", "--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()); + 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()); } } From 110c665f70e12cfb244e4b2a37f9dedce81ead27 Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Wed, 29 Jul 2026 06:56:53 +0200 Subject: [PATCH 17/37] test: harden owner materialization receipt --- tests/materialize.rs | 105 +++++++++++++++++++++++++++++++++++++------ 1 file changed, 92 insertions(+), 13 deletions(-) diff --git a/tests/materialize.rs b/tests/materialize.rs index 8bcba6b..01a8027 100644 --- a/tests/materialize.rs +++ b/tests/materialize.rs @@ -15,30 +15,109 @@ 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(); + .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!(owner.join("OWNER.txt").exists()); assert!(!sibling.join("SIBLING.txt").exists()); + 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_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(), "--once", "--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()); + for args in [ + vec![ + "up", + "--catalog", + tmp.path().to_str().unwrap(), + "--task", + "host.a.x", + ], + vec![ + "up", + "--catalog", + tmp.path().to_str().unwrap(), + "--once", + "--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()); } } From e307ab4bd8350e0f282928ac36ac575a5142c2b0 Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Wed, 29 Jul 2026 07:00:13 +0200 Subject: [PATCH 18/37] test: prove ambiguous task selector refuses mutation --- tests/materialize.rs | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/tests/materialize.rs b/tests/materialize.rs index 01a8027..cdffebb 100644 --- a/tests/materialize.rs +++ b/tests/materialize.rs @@ -75,6 +75,15 @@ fn task_selector_materializes_only_owning_agent() { 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_cli_modes_fail_closed() { let tmp = tempfile::tempdir().unwrap(); From b7f7272c007c17e8ee11ed27402071f60803ea9b Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Wed, 29 Jul 2026 07:01:27 +0200 Subject: [PATCH 19/37] test: prove wrong-host task selector refusal --- tests/materialize.rs | 74 +++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 69 insertions(+), 5 deletions(-) diff --git a/tests/materialize.rs b/tests/materialize.rs index cdffebb..fb86e32 100644 --- a/tests/materialize.rs +++ b/tests/materialize.rs @@ -77,11 +77,75 @@ fn task_selector_materializes_only_owning_agent() { #[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()); + 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] From fdfa7b8a22bd5446006ed381637dc011e8d8020f Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Wed, 29 Jul 2026 07:03:32 +0200 Subject: [PATCH 20/37] test: refuse task selectors on single-file specs --- src/main.rs | 36 ++++++++++++++++++++++++++++-------- tests/materialize.rs | 21 +++++++++++++++++++++ 2 files changed, 49 insertions(+), 8 deletions(-) diff --git a/src/main.rs b/src/main.rs index 5051cc2..63244c2 100644 --- a/src/main.rs +++ b/src/main.rs @@ -534,9 +534,15 @@ fn main() -> Result<()> { task, } => { let root = catalog_arg(root)?; - if task.is_some() && once { anyhow::bail!("--task --once is deferred until selected reconcile support"); } - if task.is_some() && !materialize_only { anyhow::bail!("--task requires --materialize-only in this stage"); } - if agent.is_some() && !materialize_only { anyhow::bail!("--agent requires --materialize-only"); } + if task.is_some() && once { + anyhow::bail!("--task --once is deferred until selected reconcile support"); + } + if task.is_some() && !materialize_only { + anyhow::bail!("--task requires --materialize-only in this stage"); + } + 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), @@ -591,7 +597,12 @@ fn main() -> Result<()> { env_cmd(&root) } Command::Pretrust { dirs } => pretrust_cmd(&dirs), - Command::Eval { folder, host, keep, json } => eval_cmd(&folder, host, keep, json), + Command::Eval { + folder, + host, + keep, + json, + } => eval_cmd(&folder, host, keep, json), Command::Validate { root, host, @@ -719,12 +730,18 @@ fn eval_cmd(folder: &Path, host: Option, keep: bool, json: bool) -> Resu ) })?; let keep = keep || std::env::var_os("ST2_EVAL_KEEP").is_some(); - if json { unsafe { std::env::set_var("ST2_EVAL_JSON", "1"); } } + if json { + unsafe { + std::env::set_var("ST2_EVAL_JSON", "1"); + } + } let report = st2::eval_run::run_eval(&spec_file, host, keep)?; if json { println!("{}", serde_json::to_string_pretty(&report)?); - if report.passed() { return Ok(()); } + if report.passed() { + return Ok(()); + } anyhow::bail!("VERDICT: FAIL") } @@ -1749,8 +1766,11 @@ fn up( 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); + 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 diff --git a/tests/materialize.rs b/tests/materialize.rs index fb86e32..a1c7f19 100644 --- a/tests/materialize.rs +++ b/tests/materialize.rs @@ -194,6 +194,27 @@ fn task_selector_cli_modes_fail_closed() { } } +#[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 From d7a22b365dc410320917ea59e87c11b2c70a15f7 Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Wed, 29 Jul 2026 07:24:20 +0200 Subject: [PATCH 21/37] feat: add selected in-memory one-shot path --- src/run.rs | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/src/run.rs b/src/run.rs index 6785a53..cb3e102 100644 --- a/src/run.rs +++ b/src/run.rs @@ -954,6 +954,18 @@ 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( + specs: &[crate::spec::AgentSpec], selector: &str, this_host: &str, runner: &dyn Runner, +) -> anyhow::Result { + crate::reconcile::resolve_task(specs, selector, this_host)?; + let sessions = runner.list_sessions().map_err(|e| anyhow::anyhow!("list sessions: {e}"))?; + let plan = crate::reconcile::reconcile_selected(specs, &sessions, this_host, selector)?; + let mut report = UpReport::default(); + 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 From c26c8d832892dae37ef1a3e9d6f54bfac6c49701 Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Wed, 29 Jul 2026 07:26:14 +0200 Subject: [PATCH 22/37] fix: gate selected one-shot launches --- src/run.rs | 19 ++++++++++++++++--- 1 file changed, 16 insertions(+), 3 deletions(-) diff --git a/src/run.rs b/src/run.rs index cb3e102..72b06b9 100644 --- a/src/run.rs +++ b/src/run.rs @@ -956,12 +956,25 @@ 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( - specs: &[crate::spec::AgentSpec], selector: &str, this_host: &str, runner: &dyn Runner, + catalog_root: &Path, + specs: &[crate::spec::AgentSpec], + selector: &str, + this_host: &str, + runner: &dyn Runner, ) -> anyhow::Result { crate::reconcile::resolve_task(specs, selector, this_host)?; - let sessions = runner.list_sessions().map_err(|e| anyhow::anyhow!("list sessions: {e}"))?; - let plan = crate::reconcile::reconcile_selected(specs, &sessions, this_host, selector)?; + 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( + catalog_root, + &mut plan, + &mut report, + || Ok(()), + crate::pretrust::pretrust_codex, + ); execute(&plan, runner, &mut FlappingCap::default(), &mut report); Ok(report) } From 219d991857febe3bbc22b5c09598523e51d8d150 Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Wed, 29 Jul 2026 08:10:43 +0200 Subject: [PATCH 23/37] test: count selected one-shot runner listings --- tests/run.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/tests/run.rs b/tests/run.rs index 21f5c4c..03c38d2 100644 --- a/tests/run.rs +++ b/tests/run.rs @@ -1,6 +1,6 @@ //! M1 correctness net: plan execution against a fake Runner (no real processes spawned). -use std::cell::RefCell; +use std::cell::{Cell, RefCell}; use std::fs; use std::path::Path; @@ -12,6 +12,7 @@ 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 +26,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"); } From 7e7a8b30cc4c02bd3a2f64373fe60c5727d4a07c Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Wed, 29 Jul 2026 08:39:47 +0200 Subject: [PATCH 24/37] test: reject unknown selected one-shot before listing --- tests/run.rs | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/tests/run.rs b/tests/run.rs index 03c38d2..3b5f8aa 100644 --- a/tests/run.rs +++ b/tests/run.rs @@ -7,7 +7,15 @@ use std::path::Path; 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_specs}; + +#[test] +fn selected_one_shot_unknown_refuses_before_runner_list() { + let runner = FakeRunner::default(); + let error = up_once_selected_specs(Path::new("/tmp"), &[], "host", "host.missing.task", &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()); +} use st2::{FlappingCap, UpReport, discover, down, execute, reconcile, up_once}; #[derive(Default)] From 4b8071202f1123b4ec89dce140afb8ea0c30e08f Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Wed, 29 Jul 2026 08:43:40 +0200 Subject: [PATCH 25/37] test: cover selected one-shot refusal matrix --- tests/run.rs | 166 +++++++++++++++++++++++++++++++++++++++++++++------ 1 file changed, 147 insertions(+), 19 deletions(-) diff --git a/tests/run.rs b/tests/run.rs index 3b5f8aa..827937b 100644 --- a/tests/run.rs +++ b/tests/run.rs @@ -1,20 +1,88 @@ //! M1 correctness net: plan execution against a fake Runner (no real processes spawned). 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, up_once_selected_specs}; +use st2::spec::{AgentSpec, JobType, Task, TaskKind}; #[test] fn selected_one_shot_unknown_refuses_before_runner_list() { let runner = FakeRunner::default(); - let error = up_once_selected_specs(Path::new("/tmp"), &[], "host", "host.missing.task", &runner).unwrap_err(); + 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()), + 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()); +} + +#[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_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()); + assert_refusal(&runner); } use st2::{FlappingCap, UpReport, discover, down, execute, reconcile, up_once}; @@ -74,10 +142,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. @@ -125,7 +201,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"]); } @@ -158,8 +237,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); @@ -170,7 +257,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); @@ -252,7 +342,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(); @@ -308,12 +401,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 @@ -321,13 +421,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(); @@ -361,9 +477,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"]); From 0d4717195e9a509fc4d41062da9706d38bfab425 Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Wed, 29 Jul 2026 08:47:58 +0200 Subject: [PATCH 26/37] test: isolate selected reconcile actions --- tests/run.rs | 109 +++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 109 insertions(+) diff --git a/tests/run.rs b/tests/run.rs index 827937b..3900486 100644 --- a/tests/run.rs +++ b/tests/run.rs @@ -59,6 +59,115 @@ fn assert_refusal(runner: &FakeRunner) { 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()), + 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")], + ..Default::default() + }; + let specs = vec![two_task_spec("agent", "host.agent.work", "host.agent.side")]; + 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")], + ..Default::default() + }; + let specs = vec![two_task_spec("agent", "host.agent.work", "host.agent.side")]; + 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")], + ..Default::default() + }; + let specs = vec![two_task_spec("agent", "host.agent.work", "host.agent.side")]; + 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")], + ..Default::default() + }; + let specs = vec![task_spec("agent", None, "host.agent.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(); From 8aa9b7e5d5d733ccc56c0a5817ae23445e5809e1 Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Wed, 29 Jul 2026 08:49:54 +0200 Subject: [PATCH 27/37] test: prove selected sibling isolation --- tests/run.rs | 36 ++++++++++++++++++++++++++++-------- 1 file changed, 28 insertions(+), 8 deletions(-) diff --git a/tests/run.rs b/tests/run.rs index 3900486..d41d611 100644 --- a/tests/run.rs +++ b/tests/run.rs @@ -78,10 +78,13 @@ fn two_task_spec(identity: &str, first: &str, second: &str) -> AgentSpec { #[test] fn selected_one_shot_missing_spawns_only_selected_task() { let runner = FakeRunner { - sessions: vec![live("host.agent.side")], + 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")]; + 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, @@ -101,10 +104,17 @@ fn selected_one_shot_missing_spawns_only_selected_task() { #[test] fn selected_one_shot_live_adopts_without_actions() { let runner = FakeRunner { - sessions: vec![live("host.agent.work"), live("host.agent.side")], + 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")]; + 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, @@ -124,10 +134,17 @@ fn selected_one_shot_live_adopts_without_actions() { #[test] fn selected_one_shot_dead_reaps_and_relaunches_only_selected() { let runner = FakeRunner { - sessions: vec![dead("host.agent.work"), live("host.agent.side")], + 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")]; + 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, @@ -148,10 +165,13 @@ fn selected_one_shot_dead_reaps_and_relaunches_only_selected() { #[test] fn selected_one_shot_second_live_pass_is_a_noop() { let runner = FakeRunner { - sessions: vec![live("host.agent.work")], + sessions: vec![live("host.agent.work"), live("host.sibling.work")], ..Default::default() }; - let specs = vec![task_spec("agent", None, "host.agent.work")]; + 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, From 45fcf8ad65ea8eeffa2e81bf7410de8705a61567 Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Wed, 29 Jul 2026 08:54:26 +0200 Subject: [PATCH 28/37] test: inject selected codex gate hooks --- src/run.rs | 53 ++++++++++++++++++++++++++++++++++++++++++++++------- 1 file changed, 46 insertions(+), 7 deletions(-) diff --git a/src/run.rs b/src/run.rs index 72b06b9..1728311 100644 --- a/src/run.rs +++ b/src/run.rs @@ -962,6 +962,30 @@ pub fn up_once_selected_specs( 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(|_| ()), + crate::pretrust::pretrust_codex, + ) +} + +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, + pretrust: F, +) -> anyhow::Result +where + V: FnOnce() -> anyhow::Result<()>, + F: FnOnce(&[PathBuf]) -> anyhow::Result, +{ crate::reconcile::resolve_task(specs, selector, this_host)?; let sessions = runner .list_sessions() @@ -972,8 +996,8 @@ pub fn up_once_selected_specs( catalog_root, &mut plan, &mut report, - || Ok(()), - crate::pretrust::pretrust_codex, + verify_hooks, + pretrust, ); execute(&plan, runner, &mut FlappingCap::default(), &mut report); Ok(report) @@ -1263,7 +1287,7 @@ pub fn detect_host() -> String { #[cfg(test)] mod tests { use super::*; - use crate::spec::TaskKind; + use crate::spec::{AgentSpec, JobType, Task, TaskKind}; use std::collections::BTreeMap; use std::ffi::OsStr; @@ -1282,10 +1306,8 @@ mod tests { } } - #[cfg(target_os = "linux")] struct EmptyRunner; - #[cfg(target_os = "linux")] impl Runner for EmptyRunner { fn list_sessions(&self) -> anyhow::Result> { Ok(Vec::new()) @@ -1304,6 +1326,25 @@ mod tests { } } + #[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: Some("codex --version".into()), + cwd: None, tags: BTreeMap::new(), env: BTreeMap::new(), keep: false }], + path: "/tmp/spec.kdl".into(), + }; + let report = up_once_selected_specs_with_gates( + Path::new("/tmp"), &[spec], "test.codex.agent", "test", &EmptyRunner, + || anyhow::bail!("stale receipt"), + |_| anyhow::bail!("must not pretrust"), + ).unwrap(); + assert!(report.launched.is_empty()); + assert!(report.errors.iter().any(|e| e.contains("launch suppressed"))); + } + #[cfg(target_os = "linux")] #[test] fn idle_supervisor_does_not_spin_on_its_own_catalog_reads() { @@ -1337,8 +1378,6 @@ mod tests { // destructively GC/relaunch a HEALTHY agent; a stable death must still be reaped ────────────── use crate::reconcile::Launch; - use crate::spec::{AgentSpec, JobType}; - fn sess(id: &str, alive: bool) -> Session { Session { pty_id: id.to_string(), From c501ec3f3b96eb565d1a4ba230d50db6410a6ee0 Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Wed, 29 Jul 2026 08:57:53 +0200 Subject: [PATCH 29/37] test: harden selected codex gate regression --- src/run.rs | 47 ++++++++++++++++++++++++++++++++++------------- 1 file changed, 34 insertions(+), 13 deletions(-) diff --git a/src/run.rs b/src/run.rs index 1728311..f529c88 100644 --- a/src/run.rs +++ b/src/run.rs @@ -1288,6 +1288,7 @@ pub fn detect_host() -> String { mod tests { use super::*; use crate::spec::{AgentSpec, JobType, Task, TaskKind}; + use std::cell::Cell; use std::collections::BTreeMap; use std::ffi::OsStr; @@ -1306,43 +1307,63 @@ mod tests { } } - struct EmptyRunner; + struct GateRunner { + list_calls: Cell, + } - 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: Some("codex --version".into()), - cwd: None, tags: BTreeMap::new(), env: BTreeMap::new(), keep: false }], + 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: Some("codex --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", &EmptyRunner, + Path::new("/tmp"), &[spec], "test.codex.agent", "test", &runner, || anyhow::bail!("stale receipt"), - |_| anyhow::bail!("must not pretrust"), + |_| panic!("pretrust must not run"), ).unwrap(); + assert_eq!(runner.list_calls.get(), 1); assert!(report.launched.is_empty()); - assert!(report.errors.iter().any(|e| e.contains("launch suppressed"))); + assert!(report.errors.iter().any(|e| e.contains("stale receipt") && e.contains("launch suppressed"))); } #[cfg(target_os = "linux")] From a308086698bba6a4f45294fc19c7c785a521b2f1 Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Wed, 29 Jul 2026 09:00:37 +0200 Subject: [PATCH 30/37] fix: keep linux runner test compiling --- src/run.rs | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/src/run.rs b/src/run.rs index f529c88..ab2d4cb 100644 --- a/src/run.rs +++ b/src/run.rs @@ -1363,7 +1363,9 @@ mod tests { ).unwrap(); assert_eq!(runner.list_calls.get(), 1); assert!(report.launched.is_empty()); - assert!(report.errors.iter().any(|e| e.contains("stale receipt") && e.contains("launch suppressed"))); + assert!(report.errors.iter().any(|error| { + error.contains("stale receipt") && error.contains("launch suppressed") + })); } #[cfg(target_os = "linux")] @@ -1381,7 +1383,9 @@ mod tests { up_loop_until( catalog.path(), "test-host", - &EmptyRunner, + &GateRunner { + list_calls: Cell::new(0), + }, Duration::from_secs(60), &stop, |_| passes += 1, From eabdb2f0f3fb2b066a0bb364f728dc428483861a Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Wed, 29 Jul 2026 09:07:52 +0200 Subject: [PATCH 31/37] feat: add selected folder catalog runner helper --- src/run.rs | 27 +++++++++++++++++++++++++++ 1 file changed, 27 insertions(+) diff --git a/src/run.rs b/src/run.rs index ab2d4cb..e6fc7fd 100644 --- a/src/run.rs +++ b/src/run.rs @@ -973,6 +973,33 @@ pub fn up_once_selected_specs( ) } +/// 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( + 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| e.message)); + let owner = owner.clone(); + if crate::hooks::required_by_codex(std::slice::from_ref(&owner), this_host) { + crate::hooks::verify_installed().map_err(|e| anyhow::anyhow!("verify hooks: {e}"))?; + } + let materialized = + crate::materialize::materialize_catalog(catalog_root, std::slice::from_ref(&owner), this_host); + report.warnings.extend(materialized.warnings); + report.errors.extend(materialized.errors); + if !report.errors.is_empty() { + return Ok(report); + } + up_once_selected_specs(catalog_root, std::slice::from_ref(&owner), selector, this_host, runner) +} + fn up_once_selected_specs_with_gates( catalog_root: &Path, specs: &[crate::spec::AgentSpec], From 72dd916a0e92dfc28abcb1239d943ff072195a38 Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Wed, 29 Jul 2026 09:09:27 +0200 Subject: [PATCH 32/37] feat: complete selected catalog execution seam --- src/run.rs | 22 ++++++++++++++++++---- 1 file changed, 18 insertions(+), 4 deletions(-) diff --git a/src/run.rs b/src/run.rs index e6fc7fd..232c37a 100644 --- a/src/run.rs +++ b/src/run.rs @@ -975,7 +975,7 @@ pub fn up_once_selected_specs( /// 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( +pub fn up_once_selected( catalog_root: &Path, selector: &str, this_host: &str, @@ -983,12 +983,16 @@ pub fn up_once_selected_catalog( ) -> anyhow::Result { let found = crate::discovery::discover(catalog_root); let (owner, _, _) = crate::reconcile::resolve_task(&found.specs, selector, this_host)?; + let all_specs = found.specs.clone(); let mut report = UpReport::default(); report.warnings.extend(found.warnings); report.errors.extend(found.errors.into_iter().map(|e| e.message)); let owner = owner.clone(); - if crate::hooks::required_by_codex(std::slice::from_ref(&owner), this_host) { - crate::hooks::verify_installed().map_err(|e| anyhow::anyhow!("verify hooks: {e}"))?; + if crate::hooks::required_by_codex_agent(&owner, this_host) { + if 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); @@ -997,7 +1001,17 @@ pub fn up_once_selected_catalog( if !report.errors.is_empty() { return Ok(report); } - up_once_selected_specs(catalog_root, std::slice::from_ref(&owner), selector, this_host, runner) + let mut execution = up_once_selected_specs_with_gates( + catalog_root, &all_specs, selector, this_host, runner, + || Ok(()), crate::pretrust::pretrust_codex, + )?; + report.warnings.append(&mut execution.warnings); + report.errors.append(&mut execution.errors); + report.launched.append(&mut execution.launched); + report.torn_down.append(&mut execution.torn_down); + report.gc.append(&mut execution.gc); + report.adopted.append(&mut execution.adopted); + Ok(report) } fn up_once_selected_specs_with_gates( From ecd8f4a07abe0ca026c07a7fbeba35ac687b0440 Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Wed, 29 Jul 2026 09:12:06 +0200 Subject: [PATCH 33/37] fix: preserve selected catalog report semantics --- src/run.rs | 31 ++++++++++++++++++++++--------- 1 file changed, 22 insertions(+), 9 deletions(-) diff --git a/src/run.rs b/src/run.rs index 232c37a..4921c98 100644 --- a/src/run.rs +++ b/src/run.rs @@ -553,6 +553,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 { @@ -983,7 +997,6 @@ pub fn up_once_selected( ) -> anyhow::Result { let found = crate::discovery::discover(catalog_root); let (owner, _, _) = crate::reconcile::resolve_task(&found.specs, selector, this_host)?; - let all_specs = found.specs.clone(); let mut report = UpReport::default(); report.warnings.extend(found.warnings); report.errors.extend(found.errors.into_iter().map(|e| e.message)); @@ -997,20 +1010,20 @@ pub fn up_once_selected( 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 !report.errors.is_empty() { + if owner_materialization_failed { return Ok(report); } let mut execution = up_once_selected_specs_with_gates( - catalog_root, &all_specs, selector, this_host, runner, + catalog_root, + &found.specs, + selector, + this_host, + runner, || Ok(()), crate::pretrust::pretrust_codex, )?; - report.warnings.append(&mut execution.warnings); - report.errors.append(&mut execution.errors); - report.launched.append(&mut execution.launched); - report.torn_down.append(&mut execution.torn_down); - report.gc.append(&mut execution.gc); - report.adopted.append(&mut execution.adopted); + report.absorb(execution); Ok(report) } From b88619cfdf10c858020c76df166764a40c2310ee Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Wed, 29 Jul 2026 09:52:48 +0200 Subject: [PATCH 34/37] docs: require supervisors and durable error propagation --- src/run.rs | 19 ++++++++++++------- 1 file changed, 12 insertions(+), 7 deletions(-) diff --git a/src/run.rs b/src/run.rs index 4921c98..e71d04b 100644 --- a/src/run.rs +++ b/src/run.rs @@ -999,13 +999,18 @@ pub fn up_once_selected( 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| e.message)); + 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) { - if let Err(error) = crate::hooks::verify_installed() { - report.errors.push(format!("verify lifecycle hooks: {error}")); - return Ok(report); - } + if crate::hooks::required_by_codex_agent(&owner, this_host) + && 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); @@ -1015,7 +1020,7 @@ pub fn up_once_selected( if owner_materialization_failed { return Ok(report); } - let mut execution = up_once_selected_specs_with_gates( + let execution = up_once_selected_specs_with_gates( catalog_root, &found.specs, selector, From 92b0d5572774d093343a2a25ae40a3b3ea9f6ec2 Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Wed, 29 Jul 2026 11:03:06 +0200 Subject: [PATCH 35/37] docs: define targeted reconciliation contract --- docs/vrs/requirements.md | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/docs/vrs/requirements.md b/docs/vrs/requirements.md index 229e96a..9a2e85b 100644 --- a/docs/vrs/requirements.md +++ b/docs/vrs/requirements.md @@ -91,3 +91,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. +- **R18 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. From 5c77dcffc54044262a65cd6ae423f7a1867a733b Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Wed, 29 Jul 2026 11:03:44 +0200 Subject: [PATCH 36/37] docs: reserve R18 for hook resilience --- docs/vrs/requirements.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/vrs/requirements.md b/docs/vrs/requirements.md index 9a2e85b..fa75910 100644 --- a/docs/vrs/requirements.md +++ b/docs/vrs/requirements.md @@ -91,7 +91,7 @@ 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. -- **R18 Targeted reconciliation:** An exact agent/task selector resolves its +- **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 From 41c7ee5b92d2fff980c7dc01bf521e507527f257 Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Wed, 29 Jul 2026 12:40:43 +0200 Subject: [PATCH 37/37] feat: complete targeted reconcile CLI --- README.md | 9 + docs/vrs/spec.md | 19 +- src/lib.rs | 2 +- src/main.rs | 27 ++- tests/exec_backend.rs | 1 - tests/materialize.rs | 8 - tests/nomad_survival.rs | 34 ++- tests/run.rs | 191 +++++++++++++++- tests/targeted_reconcile.rs | 437 ++++++++++++++++++++++++++++++++++++ 9 files changed, 694 insertions(+), 34 deletions(-) create mode 100644 tests/targeted_reconcile.rs diff --git a/README.md b/README.md index c72c0b8..1b177dd 100644 --- a/README.md +++ b/README.md @@ -160,6 +160,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/spec.md b/docs/vrs/spec.md index cf16ccb..7589ba5 100644 --- a/docs/vrs/spec.md +++ b/docs/vrs/spec.md @@ -117,14 +117,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 9def522..d8e96d1 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -39,6 +39,6 @@ pub use host_lock::HostLock; pub use reconcile::{Launch, ReconcilePlan, Session, 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, }; pub use spec::{AgentSpec, JobType, Restart, RestartMode, Task, TaskKind, parse_duration}; diff --git a/src/main.rs b/src/main.rs index 63244c2..dac411b 100644 --- a/src/main.rs +++ b/src/main.rs @@ -53,9 +53,11 @@ 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 @@ -534,11 +536,8 @@ fn main() -> Result<()> { task, } => { let root = catalog_arg(root)?; - if task.is_some() && once { - anyhow::bail!("--task --once is deferred until selected reconcile support"); - } - if task.is_some() && !materialize_only { - anyhow::bail!("--task requires --materialize-only in this stage"); + 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"); @@ -1751,6 +1750,9 @@ fn up( // 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" @@ -1811,7 +1813,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 @@ -1823,12 +1825,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/tests/exec_backend.rs b/tests/exec_backend.rs index c36c477..309625a 100644 --- a/tests/exec_backend.rs +++ b/tests/exec_backend.rs @@ -2,7 +2,6 @@ use std::collections::BTreeMap; use std::fs; -use std::process::Command; use std::thread::sleep; use std::time::Duration; diff --git a/tests/materialize.rs b/tests/materialize.rs index a1c7f19..d80e647 100644 --- a/tests/materialize.rs +++ b/tests/materialize.rs @@ -159,14 +159,6 @@ fn task_selector_cli_modes_fail_closed() { "--task", "host.a.x", ], - vec![ - "up", - "--catalog", - tmp.path().to_str().unwrap(), - "--once", - "--task", - "host.a.x", - ], vec![ "up", "--catalog", 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/run.rs b/tests/run.rs index d41d611..e59f650 100644 --- a/tests/run.rs +++ b/tests/run.rs @@ -8,9 +8,198 @@ use std::path::Path; use st2::message; use st2::reconcile::{Session, TaskTarget}; use st2::run::Runner; -use st2::run::{CrashLoop, surface_crash_loop, up_once_selected_specs}; +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(); 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}"); +}