From ffcd63f0b90e4f5937c733cfa1e57385651ad011 Mon Sep 17 00:00:00 2001 From: schickling-assistant <261620128+schickling-assistant@users.noreply.github.com> Date: Sun, 6 Sep 2026 20:15:50 +0200 Subject: [PATCH] feat: orchestrate on-demand runtime residency agent-identity: mbp2025.direct.omp.bn5hrz39 agent-persona: generalist agent-supervisor: unavailable agent-tool: OMP agent-tool-version: 18.1.7 agent-runtime: OMP 18.1.7 tooling-profile: dotfiles@4816a7a --- docs/vrs/spec.md | 14 + src/claude_session.rs | 103 +++- src/codex_app_server.rs | 256 +++++++++- src/lib.rs | 6 +- src/main.rs | 280 ++++++++--- src/omp_session.rs | 88 +++- src/provider_session.rs | 49 +- src/residency_host.rs | 1049 +++++++++++++++++++++++++++++++++++++++ src/run.rs | 301 +++++++---- 9 files changed, 1938 insertions(+), 208 deletions(-) create mode 100644 src/residency_host.rs diff --git a/docs/vrs/spec.md b/docs/vrs/spec.md index 784a29ad..c03be1c9 100644 --- a/docs/vrs/spec.md +++ b/docs/vrs/spec.md @@ -716,6 +716,20 @@ state, and Resource observation are not demand. Host policy owns idle thresholds and warm capacity. Telemetry measures resume-to-ready and demand-to-delivery latency before the project assigns a latency objective. +The folder-catalog supervisor enables this policy only when `st2 up` receives +both `--residency-idle-after ` and +`--residency-warm-capacity `. If a catalog contains a running on-demand +agent and the host omits this policy, reconciliation holds that agent +`adopt-only` and reports an error instead of launching it. `st2 wake` writes an +explicit durable wake request. `st2 pty attach ` writes the same +request before attaching. Each launch attempt has a host-minted incarnation +fence. A retry after a missing provider rotates this fence. A live provider +retains the fence while reconciliation completes its sidecars. Provider +readiness must match the independent fence, so a binding left by an earlier +failed attempt cannot make a later attempt active. A wake request that races +with checkpoint or stop remains pending until the complete task group is absent +and exact native resume can start. + ## Host-local scheduling and supervision ```text diff --git a/src/claude_session.rs b/src/claude_session.rs index c25ba79b..f0cb7410 100644 --- a/src/claude_session.rs +++ b/src/claude_session.rs @@ -34,7 +34,7 @@ pub fn run( runtime_id: String, claude_argv: Vec, ) -> Result<()> { - run_with_required_resume(catalog_root, identity, runtime_id, claude_argv, None) + run_with_required_resume(catalog_root, identity, runtime_id, claude_argv, None, None) } pub fn run_residency( @@ -44,12 +44,36 @@ pub fn run_residency( claude_argv: Vec, resume_generation: crate::residency::Generation, ) -> Result<()> { + run_residency_attempt( + catalog_root, + identity, + runtime_id, + claude_argv, + resume_generation, + harness_state::session_token(), + ) +} + +/// Run one host-owned cold-residency attempt under its exact incarnation. +pub fn run_residency_attempt( + catalog_root: &Path, + identity: String, + runtime_id: String, + claude_argv: Vec, + resume_generation: crate::residency::Generation, + required_incarnation: String, +) -> Result<()> { + anyhow::ensure!( + !required_incarnation.is_empty(), + "Claude required runtime incarnation is empty" + ); run_with_required_resume( catalog_root, identity, runtime_id, claude_argv, Some(resume_generation), + Some(required_incarnation), ) } @@ -59,11 +83,16 @@ fn run_with_required_resume( runtime_id: String, claude_argv: Vec, required_resume_generation: Option, + required_incarnation: Option, ) -> Result<()> { anyhow::ensure!( !claude_argv.is_empty(), "Claude driver '{runtime_id}' has no provider argv" ); + anyhow::ensure!( + required_resume_generation.is_some() == required_incarnation.is_some(), + "Claude residency launch has an incomplete attempt fence" + ); let workspace = std::env::current_dir().context("reading the Claude driver workspace")?; let required_native_session = match required_resume_generation { Some(generation) => Some(required_residency_resume( @@ -85,7 +114,12 @@ fn run_with_required_resume( crate::pretrust::pretrust_claude(std::slice::from_ref(&workspace)) .with_context(|| format!("admitting Claude driver workspace {}", workspace.display()))?; install_signal_handler(); - let observer = SessionObserver::new(&agent_dir, &identity, "claude", &runtime_id)?; + let observer = match required_incarnation { + Some(session) => { + SessionObserver::with_session(&agent_dir, &identity, "claude", &runtime_id, session)? + } + None => SessionObserver::new(&agent_dir, &identity, "claude", &runtime_id)?, + }; let mut env = vec![ (RUNTIME_ID_ENV.to_string(), runtime_id.clone()), (SESSION_ENV.to_string(), observer.session().to_string()), @@ -689,6 +723,13 @@ fn load_checkpoint( && checkpoint.binding.runtime_id == runtime_id, "Claude residency checkpoint belongs to a different agent runtime" ); + anyhow::ensure!( + !checkpoint.binding.runtime_incarnation.is_empty() + && valid_uuid(&checkpoint.binding.native_session_id) + && checkpoint.binding.canonical_workspace.is_absolute() + && checkpoint.binding.transcript_path.is_absolute(), + "Claude residency checkpoint has an incomplete native session binding" + ); anyhow::ensure!( checkpoint.transcript_sha256.len() == 64 && checkpoint @@ -700,6 +741,25 @@ fn load_checkpoint( Ok(checkpoint) } +fn validate_residency_candidate( + checkpoint: &ClaudeResidencyCheckpoint, + candidate: &ClaudeSessionBinding, + resume_generation: crate::residency::Generation, +) -> Result<()> { + anyhow::ensure!( + candidate.resume_generation == Some(resume_generation) + && candidate.runtime_incarnation != checkpoint.binding.runtime_incarnation, + "Claude SessionStart does not prove the required residency generation and a new wrapper incarnation" + ); + anyhow::ensure!( + candidate.native_session_id == checkpoint.binding.native_session_id + && candidate.canonical_workspace == checkpoint.binding.canonical_workspace + && candidate.transcript_path == checkpoint.binding.transcript_path, + "Claude resumed a different native session than the residency checkpoint" + ); + Ok(()) +} + pub fn required_residency_resume( state_dir: &Path, agent: &str, @@ -739,15 +799,11 @@ pub fn residency_ready( let current = load_binding(state_dir, agent, runtime_id)?; if let Some(current) = ¤t && current.resume_generation == Some(resume_generation) - && current.runtime_incarnation == expected_runtime_incarnation { - anyhow::ensure!( - current.native_session_id == checkpoint.binding.native_session_id - && current.canonical_workspace == checkpoint.binding.canonical_workspace - && current.transcript_path == checkpoint.binding.transcript_path, - "Claude ready binding does not match the residency checkpoint" - ); - return Ok(true); + validate_residency_candidate(&checkpoint, current, resume_generation)?; + if current.runtime_incarnation == expected_runtime_incarnation { + return Ok(true); + } } let Some(candidate) = load_pending_binding(state_dir, agent, runtime_id)? else { return Ok(false); @@ -759,17 +815,10 @@ pub fn residency_ready( "Claude native session changed after this residency generation became ready" ); } + validate_residency_candidate(&checkpoint, &candidate, resume_generation)?; anyhow::ensure!( - candidate.resume_generation == Some(resume_generation) - && candidate.runtime_incarnation != checkpoint.binding.runtime_incarnation - && candidate.runtime_incarnation == expected_runtime_incarnation, - "Claude SessionStart does not prove the required residency generation and wrapper incarnation" - ); - anyhow::ensure!( - candidate.native_session_id == checkpoint.binding.native_session_id - && candidate.canonical_workspace == checkpoint.binding.canonical_workspace - && candidate.transcript_path == checkpoint.binding.transcript_path, - "Claude resumed a different native session than the residency checkpoint" + candidate.runtime_incarnation == expected_runtime_incarnation, + "Claude SessionStart does not prove the expected wrapper incarnation" ); validate_transcript( &candidate.transcript_path, @@ -2481,6 +2530,20 @@ mod tests { } let temp = tempfile::tempdir().unwrap(); + let empty_incarnation = run_residency_attempt( + temp.path(), + "no-spawn.worker".into(), + "no-spawn.worker".into(), + vec!["claude".into()], + crate::residency::Generation(2), + String::new(), + ) + .unwrap_err(); + assert!( + empty_incarnation + .to_string() + .contains("required runtime incarnation is empty") + ); let marker = temp.path().join("provider-started"); let provider = temp.path().join("claude"); fs::write( diff --git a/src/codex_app_server.rs b/src/codex_app_server.rs index 7f7ddf92..2e64c542 100644 --- a/src/codex_app_server.rs +++ b/src/codex_app_server.rs @@ -161,11 +161,19 @@ pub struct CodexRuntime { impl CodexRuntime { fn fresh(agent: String, runtime_id: String) -> Result { + Self::with_incarnation(agent, runtime_id, random_token()?) + } + + fn with_incarnation(agent: String, runtime_id: String, incarnation: String) -> Result { + anyhow::ensure!( + !incarnation.is_empty(), + "Codex runtime incarnation is empty" + ); Ok(Self { schema: RUNTIME_SCHEMA.to_string(), agent, runtime_id, - incarnation: random_token()?, + incarnation, }) } @@ -1248,8 +1256,7 @@ impl CodexInboxDelivery { .entries() .iter() .filter(|entry| { - entry.binding == state.thread_id() - && entry.phase < delivery_ledger::Phase::Consumed + entry.binding == state.thread_id() && entry.phase < delivery_ledger::Phase::Consumed }) .map(|entry| (entry.filename.clone(), entry.correlation.value.clone())) .collect(); @@ -1738,7 +1745,14 @@ pub fn run_controlled( runtime_id: String, codex_argv: Vec, ) -> Result<()> { - run_controlled_with_required_resume(catalog_root, identity, runtime_id, codex_argv, None) + run_controlled_with_required_resume( + catalog_root, + identity, + runtime_id, + codex_argv, + None, + None, + ) } /// Run a cold residency generation with an exact checkpointed native thread. @@ -1749,12 +1763,36 @@ pub fn run_controlled_residency( codex_argv: Vec, resume_generation: crate::residency::Generation, ) -> Result<()> { + run_controlled_residency_attempt( + catalog_root, + identity, + runtime_id, + codex_argv, + resume_generation, + random_token()?, + ) +} + +/// Run one host-owned cold-residency attempt under its exact incarnation. +pub fn run_controlled_residency_attempt( + catalog_root: &Path, + identity: String, + runtime_id: String, + codex_argv: Vec, + resume_generation: crate::residency::Generation, + required_incarnation: String, +) -> Result<()> { + anyhow::ensure!( + !required_incarnation.is_empty(), + "Codex required runtime incarnation is empty" + ); run_controlled_with_required_resume( catalog_root, identity, runtime_id, codex_argv, Some(resume_generation), + Some(required_incarnation), ) } @@ -1764,11 +1802,16 @@ fn run_controlled_with_required_resume( runtime_id: String, codex_argv: Vec, required_resume_generation: Option, + required_incarnation: Option, ) -> Result<()> { anyhow::ensure!( !codex_argv.is_empty(), "Codex controlled launch argv is empty" ); + anyhow::ensure!( + required_resume_generation.is_some() == required_incarnation.is_some(), + "Codex residency launch has an incomplete attempt fence" + ); // Install before any protocol preflight child exists. A SIGTERM in the preflight window must // set the stop flag instead of leaving a detached app server and stale socket behind. crate::provider_session::install_signal_handler(); @@ -1809,6 +1852,7 @@ fn run_controlled_with_required_resume( codex_argv, delivery, resume_thread, + required_incarnation, &mut diagnostics, ); match result { @@ -1838,6 +1882,7 @@ fn run_controlled_owned( codex_argv: Vec, delivery: CodexDeliveryConfig, resume_thread: Option, + required_incarnation: Option, diagnostics: &mut WrapperDiagnostics, ) -> Result<()> { let socket_path = socket_path(catalog_root, &identity)?; @@ -1849,7 +1894,10 @@ fn run_controlled_owned( // Publish a new incarnation only after this process holds the owner lock and has proved that no // older daemon is live. A rejected second owner must not invalidate the first owner's binding. - let runtime = CodexRuntime::fresh(identity, runtime_id)?; + let runtime = match required_incarnation { + Some(incarnation) => CodexRuntime::with_incarnation(identity, runtime_id, incarnation)?, + None => CodexRuntime::fresh(identity, runtime_id)?, + }; atomic_json(&state_dir.join("runtime.json"), &runtime)?; diagnostics.record( "runtimePublished", @@ -2847,8 +2895,7 @@ fn pump_control( let mut control_state: Option = None; let mut subscription_pending = false; let mut peer_closed = false; - let delivery_ledger_path = - control_state_path.with_file_name(delivery_ledger::LEDGER_FILE); + let delivery_ledger_path = control_state_path.with_file_name(delivery_ledger::LEDGER_FILE); let mut delivery = delivery .map(|config| { CodexInboxDelivery::new(config, delivery_ledger_path.clone(), runtime.clone()) @@ -4100,6 +4147,34 @@ pub fn required_residency_resume( Ok(current.thread_id) } +/// Whether the initialized binding proves an exact cold-residency resume. +/// +/// A missing binding means the replacement provider has not initialized yet. Every binding that +/// does exist must belong to the requested runtime, retain the checkpointed native thread, and +/// come from the independently expected wrapper incarnation. +pub fn residency_ready( + state_dir: &Path, + agent: &str, + runtime_id: &str, + resume_generation: crate::residency::Generation, + expected_runtime_incarnation: &str, +) -> Result { + let checkpoint = load_residency_checkpoint(state_dir, agent, runtime_id, resume_generation)?; + let Some(current) = load_thread_binding(&state_dir.join("binding.json"), agent, runtime_id)? + else { + return Ok(false); + }; + anyhow::ensure!( + current.runtime_incarnation != checkpoint.binding.runtime_incarnation, + "Codex residency binding did not advance to a new runtime incarnation" + ); + anyhow::ensure!( + current.thread_id == checkpoint.binding.thread_id, + "Codex residency binding does not match the checkpointed native thread" + ); + Ok(current.runtime_incarnation == expected_runtime_incarnation) +} + fn load_residency_checkpoint( state_dir: &Path, agent: &str, @@ -4120,12 +4195,15 @@ fn load_residency_checkpoint( "Codex residency checkpoint belongs to a different generation" ); anyhow::ensure!( - checkpoint.binding.agent == agent && checkpoint.binding.runtime_id == runtime_id, + checkpoint.binding.schema == BINDING_SCHEMA + && checkpoint.binding.agent == agent + && checkpoint.binding.runtime_id == runtime_id, "Codex residency checkpoint belongs to a different agent runtime" ); anyhow::ensure!( - !checkpoint.binding.thread_id.is_empty(), - "Codex residency checkpoint has an empty native thread id" + !checkpoint.binding.runtime_incarnation.is_empty() + && !checkpoint.binding.thread_id.is_empty(), + "Codex residency checkpoint has an incomplete native thread binding" ); Ok(checkpoint) } @@ -4150,8 +4228,8 @@ fn load_thread_binding( "Codex resume binding belongs to a different agent runtime" ); anyhow::ensure!( - !binding.thread_id.is_empty(), - "Codex resume binding has an empty thread id" + !binding.runtime_incarnation.is_empty() && !binding.thread_id.is_empty(), + "Codex resume binding is incomplete" ); Ok(Some(binding)) } @@ -6284,11 +6362,7 @@ mod tests { ) .unwrap(); assert_eq!( - replacement - .ledger - .entry(&absent_filename) - .unwrap() - .negative, + replacement.ledger.entry(&absent_filename).unwrap().negative, Some(delivery_ledger::NegativeReceipt::Absent), "the absence is retained as evidence, not erased" ); @@ -6296,7 +6370,6 @@ mod tests { assert_eq!(retry["params"]["clientUserMessageId"], absent_client_id); } - #[test] fn subscribed_control_pump_delivers_a_typed_reference_to_the_real_fifo_head() { let tmp = tempfile::tempdir().unwrap(); @@ -6515,7 +6588,6 @@ mod tests { assert_eq!(observed.used_percent, Some(33.0)); } - #[test] fn control_initializes_before_recording_the_first_thread_only() { let tmp = tempfile::tempdir().unwrap(); @@ -7145,6 +7217,138 @@ mod tests { ); } + #[test] + fn residency_readiness_requires_the_exact_thread_under_the_expected_incarnation() { + let tmp = tempfile::tempdir().unwrap(); + let state = tmp.path().join("state"); + let prior = CodexRuntime::fresh("h.worker".into(), "h.worker".into()).unwrap(); + let prior_binding = CodexThreadBinding::new(&prior, "thread-prior".into()); + atomic_json(&state.join("binding.json"), &prior_binding).unwrap(); + checkpoint_residency( + &state, + "h.worker", + "h.worker", + crate::residency::Generation(1), + crate::residency::Generation(2), + ) + .unwrap(); + + fs::remove_file(state.join("binding.json")).unwrap(); + assert!( + !residency_ready( + &state, + "h.worker", + "h.worker", + crate::residency::Generation(2), + "attempt-next", + ) + .unwrap() + ); + + atomic_json(&state.join("binding.json"), &prior_binding).unwrap(); + let stale = residency_ready( + &state, + "h.worker", + "h.worker", + crate::residency::Generation(2), + "attempt-next", + ) + .unwrap_err(); + assert!(stale.to_string().contains("new runtime incarnation")); + + let foreign = CodexRuntime::fresh("h.other".into(), "h.other".into()).unwrap(); + atomic_json( + &state.join("binding.json"), + &CodexThreadBinding::new(&foreign, "thread-prior".into()), + ) + .unwrap(); + let foreign = residency_ready( + &state, + "h.worker", + "h.worker", + crate::residency::Generation(2), + "attempt-next", + ) + .unwrap_err(); + assert!(foreign.to_string().contains("different agent runtime")); + + let stale_attempt = CodexRuntime::with_incarnation( + "h.worker".into(), + "h.worker".into(), + "attempt-stale".into(), + ) + .unwrap(); + atomic_json( + &state.join("binding.json"), + &CodexThreadBinding::new(&stale_attempt, "thread-other".into()), + ) + .unwrap(); + let mismatched = residency_ready( + &state, + "h.worker", + "h.worker", + crate::residency::Generation(2), + "attempt-next", + ) + .unwrap_err(); + assert!( + mismatched + .to_string() + .contains("checkpointed native thread") + ); + + atomic_json( + &state.join("binding.json"), + &CodexThreadBinding::new(&stale_attempt, "thread-prior".into()), + ) + .unwrap(); + assert!( + !residency_ready( + &state, + "h.worker", + "h.worker", + crate::residency::Generation(2), + "attempt-next", + ) + .unwrap(), + "a stale prior-attempt binding became ready" + ); + + let replacement = CodexRuntime::with_incarnation( + "h.worker".into(), + "h.worker".into(), + "attempt-next".into(), + ) + .unwrap(); + atomic_json( + &state.join("binding.json"), + &CodexThreadBinding::new(&replacement, "thread-prior".into()), + ) + .unwrap(); + assert!( + residency_ready( + &state, + "h.worker", + "h.worker", + crate::residency::Generation(2), + "attempt-next", + ) + .unwrap() + ); + assert!( + residency_ready( + &state, + "h.worker", + "h.worker", + crate::residency::Generation(3), + "attempt-next", + ) + .unwrap_err() + .to_string() + .contains("different generation") + ); + } + #[test] fn mandatory_residency_resume_refuses_missing_or_authored_session_selection() { let tmp = tempfile::tempdir().unwrap(); @@ -7271,6 +7475,20 @@ mod tests { fn mandatory_residency_failure_does_not_spawn_a_provider_child() { let _stop_exclusive = stop_flag_tests(); let tmp = tempfile::tempdir().unwrap(); + let empty_incarnation = run_controlled_residency_attempt( + tmp.path(), + "h.worker".into(), + "h.worker".into(), + vec!["codex".into()], + crate::residency::Generation(2), + String::new(), + ) + .unwrap_err(); + assert!( + empty_incarnation + .to_string() + .contains("required runtime incarnation is empty") + ); let marker = tmp.path().join("provider-started"); let codex = tmp.path().join("codex"); fs::write( diff --git a/src/lib.rs b/src/lib.rs index 120aa8ae..da80be07 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -36,9 +36,9 @@ pub mod identity; pub mod isolate; pub mod materialize; pub mod message; -pub mod omp_session; pub mod metrics; pub mod migrations; +pub mod omp_session; pub mod opencode_session; pub mod park; pub mod pi_channel; @@ -48,6 +48,7 @@ pub mod provider_session; pub mod reconcile; pub mod request; pub mod residency; +pub mod residency_host; pub mod resource_observe; pub mod resource_profile; pub mod resource_profile_supervisor; @@ -83,5 +84,6 @@ pub use 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_selected, up_once_selected_specs, up_once_specs, + up_loop, up_loop_specs, up_loop_with_residency, up_once, up_once_selected, + up_once_selected_specs, up_once_specs, up_once_with_residency, }; diff --git a/src/main.rs b/src/main.rs index 710b1884..18cd13d6 100644 --- a/src/main.rs +++ b/src/main.rs @@ -64,6 +64,28 @@ enum Command { /// immediately regardless). #[arg(long, default_value_t = 30)] interval: u64, + /// Host idle threshold before an eligible on-demand agent may become cold. + #[arg( + long, + value_parser = st2::parse_duration, + requires = "residency_warm_capacity" + )] + residency_idle_after: Option, + /// Minimum number of eligible on-demand agents that this host keeps warm. + #[arg(long, requires = "residency_idle_after")] + residency_warm_capacity: Option, + }, + /// Durably request that the host supervisor wake one on-demand agent. + Wake { + /// Agent bus address or unique local identity. + #[arg(required_unless_present = "agent_id")] + identity: Option, + /// Exact immutable agent ID. + #[arg(long = "agent-id", conflicts_with = "identity")] + agent_id: Option, + /// Host whose supervisor owns residency. Defaults to this host. + #[arg(long)] + host: Option, }, /// Native message bus: send/list/read/archive/reply over agents' `resources/inbox`. /// The stable wire format is a `-.md` Markdown file. @@ -138,8 +160,11 @@ enum Command { #[arg(long)] runtime_id: String, /// Internal cold-residency generation that must resume its exact checkpointed thread. - #[arg(long, hide = true)] + #[arg(long, hide = true, requires = "required_resume_incarnation")] required_resume_generation: Option, + /// Host-minted incarnation that fences this exact launch attempt. + #[arg(long, hide = true, requires = "required_resume_generation")] + required_resume_incarnation: Option, /// Original structured Codex invocation, including its provider executable. #[arg(required = true, trailing_var_arg = true, allow_hyphen_values = true)] codex_argv: Vec, @@ -347,8 +372,11 @@ enum DriverCmd { #[arg(long)] runtime_id: String, /// Internal cold-residency generation that must resume its exact checkpointed thread. - #[arg(long, hide = true)] + #[arg(long, hide = true, requires = "required_resume_incarnation")] required_resume_generation: Option, + /// Host-minted incarnation that fences this exact launch attempt. + #[arg(long, hide = true, requires = "required_resume_generation")] + required_resume_incarnation: Option, #[arg(required = true, trailing_var_arg = true, allow_hyphen_values = true)] argv: Vec, }, @@ -371,8 +399,11 @@ enum DriverCmd { #[arg(long)] runtime_id: String, /// Internal cold-residency generation that must resume its exact checkpointed session. - #[arg(long, hide = true)] + #[arg(long, hide = true, requires = "required_resume_incarnation")] required_resume_generation: Option, + /// Host-minted incarnation that fences this exact launch attempt. + #[arg(long, hide = true, requires = "required_resume_generation")] + required_resume_incarnation: Option, #[arg(required = true, trailing_var_arg = true, allow_hyphen_values = true)] argv: Vec, }, @@ -414,8 +445,11 @@ enum DriverCmd { #[arg(long)] runtime_id: String, /// Internal cold-residency generation that must resume its exact checkpointed session. - #[arg(long, hide = true)] + #[arg(long, hide = true, requires = "required_resume_incarnation")] required_resume_generation: Option, + /// Host-minted incarnation that fences this exact launch attempt. + #[arg(long, hide = true, requires = "required_resume_generation")] + required_resume_incarnation: Option, #[arg(required = true, trailing_var_arg = true, allow_hyphen_values = true)] argv: Vec, }, @@ -724,8 +758,14 @@ struct PresentationArgs { impl PresentationArgs { /// The selected subject and the requested value, where `None` is the cleared representation. - fn selection(self) -> Result<(st2::identity::AgentSelector, Option, bool, Option)> - { + fn selection( + self, + ) -> Result<( + st2::identity::AgentSelector, + Option, + bool, + Option, + )> { let (selector, value) = match self.agent_id { Some(id) => (st2::identity::AgentSelector::Id(id), self.first), None => ( @@ -1132,7 +1172,10 @@ enum EventCmd { /// Emit one producer-identified event into a declared agent stream. Emit { /// Owning agent: a bus address (`.
`) or a bare local address. - #[arg(required_unless_present = "recipient_id", conflicts_with = "recipient_id")] + #[arg( + required_unless_present = "recipient_id", + conflicts_with = "recipient_id" + )] recipient: Option, /// Owning agent by exact immutable agent ID (R24). #[arg(long = "recipient-id")] @@ -1307,6 +1350,8 @@ fn dispatch(command: Command, catalog_path: Option<&std::path::Path>) -> Result< interval, agent, task, + residency_idle_after, + residency_warm_capacity, } => { let root = catalog_arg(root)?; if task.is_some() && !materialize_only && !once { @@ -1315,9 +1360,23 @@ fn dispatch(command: Command, catalog_path: Option<&std::path::Path>) -> Result< if agent.is_some() && !materialize_only { anyhow::bail!("--agent requires --materialize-only"); } - up(&root, host, once, materialize_only, interval, agent, task) + up( + &root, + host, + once, + materialize_only, + interval, + agent, + task, + residency_idle_after.zip(residency_warm_capacity), + ) } Command::Message(cmd) => message_cmd(cmd), + Command::Wake { + identity, + agent_id, + host, + } => wake_cmd(identity, agent_id, host), Command::Event(cmd) => event_cmd(cmd), Command::Stream(cmd) => stream_cmd(cmd), Command::Request(cmd) => request_cmd(cmd), @@ -1338,24 +1397,26 @@ fn dispatch(command: Command, catalog_path: Option<&std::path::Path>) -> Result< identity, runtime_id, required_resume_generation, + required_resume_incarnation, codex_argv, } => { let catalog = catalog_arg(None)?; let catalog = catalog.canonicalize().unwrap_or(catalog); - match required_resume_generation { - Some(generation) => st2::codex_app_server::run_controlled_residency( - &catalog, - identity, - runtime_id, - codex_argv, - st2::residency::Generation(generation), - ), - None => st2::codex_app_server::run_controlled( - &catalog, - identity, - runtime_id, - codex_argv, + match (required_resume_generation, required_resume_incarnation) { + (Some(generation), Some(incarnation)) => { + st2::codex_app_server::run_controlled_residency_attempt( + &catalog, + identity, + runtime_id, + codex_argv, + st2::residency::Generation(generation), + incarnation, + ) + } + (None, None) => st2::codex_app_server::run_controlled( + &catalog, identity, runtime_id, codex_argv, ), + _ => unreachable!("clap requires the complete residency launch fence"), } } Command::ClaudeMcp { identity } => { @@ -1367,21 +1428,26 @@ fn dispatch(command: Command, catalog_path: Option<&std::path::Path>) -> Result< identity, runtime_id, required_resume_generation, + required_resume_incarnation, argv, }) => { let catalog = catalog_arg(None)?; let catalog = catalog.canonicalize().unwrap_or(catalog); - match required_resume_generation { - Some(generation) => st2::codex_app_server::run_controlled_residency( - &catalog, - identity, - runtime_id, - argv, - st2::residency::Generation(generation), - ), - None => { + match (required_resume_generation, required_resume_incarnation) { + (Some(generation), Some(incarnation)) => { + st2::codex_app_server::run_controlled_residency_attempt( + &catalog, + identity, + runtime_id, + argv, + st2::residency::Generation(generation), + incarnation, + ) + } + (None, None) => { st2::codex_app_server::run_controlled(&catalog, identity, runtime_id, argv) } + _ => unreachable!("clap requires the complete residency launch fence"), } } Command::Driver(DriverCmd::PiChannel { identity }) => { @@ -1415,19 +1481,22 @@ fn dispatch(command: Command, catalog_path: Option<&std::path::Path>) -> Result< identity, runtime_id, required_resume_generation, + required_resume_incarnation, argv, }) => { let catalog = catalog_arg(None)?; let catalog = catalog.canonicalize().unwrap_or(catalog); - match required_resume_generation { - Some(generation) => st2::omp_session::run_residency( + match (required_resume_generation, required_resume_incarnation) { + (Some(generation), Some(incarnation)) => st2::omp_session::run_residency_attempt( &catalog, identity, runtime_id, argv, st2::residency::Generation(generation), + incarnation, ), - None => st2::omp_session::run(&catalog, identity, runtime_id, argv), + (None, None) => st2::omp_session::run(&catalog, identity, runtime_id, argv), + _ => unreachable!("clap requires the complete residency launch fence"), } } Command::Driver(DriverCmd::Claude { identity }) => { @@ -1440,19 +1509,24 @@ fn dispatch(command: Command, catalog_path: Option<&std::path::Path>) -> Result< identity, runtime_id, required_resume_generation, + required_resume_incarnation, argv, }) => { let catalog = catalog_arg(None)?; let catalog = catalog.canonicalize().unwrap_or(catalog); - match required_resume_generation { - Some(generation) => st2::claude_session::run_residency( - &catalog, - identity, - runtime_id, - argv, - st2::residency::Generation(generation), - ), - None => st2::claude_session::run(&catalog, identity, runtime_id, argv), + match (required_resume_generation, required_resume_incarnation) { + (Some(generation), Some(incarnation)) => { + st2::claude_session::run_residency_attempt( + &catalog, + identity, + runtime_id, + argv, + st2::residency::Generation(generation), + incarnation, + ) + } + (None, None) => st2::claude_session::run(&catalog, identity, runtime_id, argv), + _ => unreachable!("clap requires the complete residency launch fence"), } } Command::Driver(DriverCmd::ClaudeObserve { @@ -1506,9 +1580,8 @@ fn dispatch(command: Command, catalog_path: Option<&std::path::Path>) -> Result< Some(_) => (None, first), None => (first, second), }; - let state = state.context( - "a desired state is required: `running`, `suspended`, or `retired`", - )?; + let state = state + .context("a desired state is required: `running`, `suspended`, or `retired`")?; anyhow::ensure!( matches!(state.as_str(), "running" | "suspended" | "retired"), "desired state must be `running`, `suspended`, or `retired`, not '{state}'" @@ -2120,6 +2193,47 @@ fn catalog_root_for_env() -> Result { )?; absolute_catalog_path(&root) } +fn wake_cmd( + identity: Option, + agent_id: Option, + host: Option, +) -> Result<()> { + let root = catalog_root_for_env()?; + let host = host.unwrap_or_else(detect_host); + let declaration = selected_declaration(&root, &host, identity, agent_id)? + .context("wake requires an agent reference")?; + let found = discover(&root); + let spec = found + .specs + .iter() + .find(|spec| spec.bus_id(&host) == declaration) + .context("selected wake declaration disappeared")?; + anyhow::ensure!( + spec.residency_policy == st2::ResidencyPolicy::OnDemand && spec.desired_state.is_running(), + "wake target is not a running on-demand agent" + ); + let agent_id = spec.effective_id(&host); + let path = st2::residency_host::request_wake(&root, &host, &agent_id)?; + println!( + "wake requested for '{}' on {host}; request {}", + spec.bus_id(&host), + path.display() + ); + Ok(()) +} + +fn request_attach_wake(root: &Path, target: &str) -> Result<()> { + let host = detect_host(); + let found = discover(root); + if let Some(spec) = found.specs.iter().find(|spec| { + spec.residency_policy == st2::ResidencyPolicy::OnDemand + && spec.desired_state.is_running() + && spec.bus_id(&host) == target + }) { + st2::residency_host::request_wake(root, &host, &spec.effective_id(&host))?; + } + Ok(()) +} /// Set the same native catalog environment that `st2 env` prints. The catalog's own declared session /// registry is used, not the caller's ambient one: these hand `pty` the roots of the *catalog*. @@ -2136,6 +2250,11 @@ fn pty_cmd(args: &[String]) -> Result<()> { use std::os::unix::process::CommandExt; let root = catalog_root_for_env()?; + if args.first().is_some_and(|argument| argument == "attach") + && let Some(target) = args.get(1) + { + request_attach_wake(&root, target)?; + } let mut cmd = std::process::Command::new("pty"); cmd.args(args); with_bus_env(&mut cmd, &root); @@ -3337,10 +3456,12 @@ fn message_cmd(cmd: MessageCmd) -> Result<()> { Some(id) => id, None => acting_route(&root, &host, &ctx)?, }; - let mut view = - message::with_resolved_agent_dir(&root, &route_selector(&id), &host, |agent_dir| { - message::list_sent(agent_dir, include_body) - })?; + let mut view = message::with_resolved_agent_dir( + &root, + &route_selector(&id), + &host, + |agent_dir| message::list_sent(agent_dir, include_body), + )?; if let Some(recipient) = &to { view.messages.retain(|message| message.to == *recipient); } @@ -3570,8 +3691,7 @@ fn event_cmd(cmd: EventCmd) -> Result<()> { ctx, } => { let (root, host) = resolve_ctx(&ctx)?; - let recipient = - resolve_route(&root, &host, one_selector(recipient, recipient_id)?)?; + let recipient = resolve_route(&root, &host, one_selector(recipient, recipient_id)?)?; let body = body_or_stdin(body)?; let receipt = st2::event::emit( &root, @@ -4431,6 +4551,7 @@ fn up( interval: u64, agent: Option, task: Option, + residency_policy: Option<(Duration, usize)>, ) -> 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. @@ -4438,6 +4559,9 @@ fn up( if task.is_some() { anyhow::bail!("--task is for folder catalogs, not single-file specs"); } + if residency_policy.is_some() { + anyhow::bail!("host residency policy is supported only for folder catalogs"); + } if materialize_only { anyhow::bail!( "--materialize-only is for folder catalogs with agent render{{}} blocks, not single-file specs" @@ -4519,11 +4643,23 @@ fn up( if once { let targeted = task.is_some(); - let report = match task.as_deref() { - Some(selector) => { + let report = match (task.as_deref(), residency_policy) { + (Some(_), Some(_)) => { + anyhow::bail!("--task cannot be combined with host residency policy") + } + (Some(selector), None) => { st2::run::up_once_selected(&catalog_root, selector, &this_host, &runner)? } - None => up_once(&catalog_root, &this_host, &runner)?, + (None, Some((idle_after, warm_capacity))) => st2::up_once_with_residency( + &catalog_root, + &this_host, + &runner, + st2::residency_host::HostPolicy { + idle_after, + warm_capacity, + }, + )?, + (None, None) => up_once(&catalog_root, &this_host, &runner)?, }; println!("reconcile pass on host '{this_host}':"); print_report(&report); @@ -4548,17 +4684,31 @@ fn up( "st2: supervising {} on host '{this_host}' (reconcile every {interval}s + on change; Ctrl-C to stop)", root.display() ); - let result = up_loop( - &catalog_root, - &this_host, - &runner, - Duration::from_secs(interval), - |report| { - if report.is_noteworthy() { - print_report(report); - } - }, - ); + let on_report = |report: &UpReport| { + if report.is_noteworthy() { + print_report(report); + } + }; + let result = match residency_policy { + Some((idle_after, warm_capacity)) => st2::up_loop_with_residency( + &catalog_root, + &this_host, + &runner, + Duration::from_secs(interval), + st2::residency_host::HostPolicy { + idle_after, + warm_capacity, + }, + on_report, + ), + None => up_loop( + &catalog_root, + &this_host, + &runner, + Duration::from_secs(interval), + on_report, + ), + }; lock.release(); result } diff --git a/src/omp_session.rs b/src/omp_session.rs index 4ca2e6ca..8b2dea92 100644 --- a/src/omp_session.rs +++ b/src/omp_session.rs @@ -241,6 +241,7 @@ pub fn residency_ready( agent: &str, runtime_id: &str, resume_generation: crate::residency::Generation, + expected_runtime_incarnation: &str, ) -> Result { let checkpoint = load_checkpoint(state_dir, agent, runtime_id, resume_generation)?; let Some(current) = load_binding(state_dir, agent, runtime_id)? else { @@ -260,7 +261,7 @@ pub fn residency_ready( current.native_session_id == checkpoint.binding.native_session_id, "OMP resumed a different native session than the residency checkpoint" ); - Ok(true) + Ok(current.runtime_incarnation == expected_runtime_incarnation) } fn load_binding( @@ -381,7 +382,7 @@ pub fn run( runtime_id: String, omp_argv: Vec, ) -> Result<()> { - run_with_required_resume(catalog_root, identity, runtime_id, omp_argv, None) + run_with_required_resume(catalog_root, identity, runtime_id, omp_argv, None, None) } pub fn run_residency( @@ -391,12 +392,36 @@ pub fn run_residency( omp_argv: Vec, resume_generation: crate::residency::Generation, ) -> Result<()> { + run_residency_attempt( + catalog_root, + identity, + runtime_id, + omp_argv, + resume_generation, + harness_state::session_token(), + ) +} + +/// Run one host-owned cold-residency attempt under its exact incarnation. +pub fn run_residency_attempt( + catalog_root: &Path, + identity: String, + runtime_id: String, + omp_argv: Vec, + resume_generation: crate::residency::Generation, + required_incarnation: String, +) -> Result<()> { + anyhow::ensure!( + !required_incarnation.is_empty(), + "OMP required runtime incarnation is empty" + ); run_with_required_resume( catalog_root, identity, runtime_id, omp_argv, Some(resume_generation), + Some(required_incarnation), ) } @@ -406,11 +431,16 @@ fn run_with_required_resume( runtime_id: String, omp_argv: Vec, required_resume_generation: Option, + required_incarnation: Option, ) -> Result<()> { anyhow::ensure!( !omp_argv.is_empty(), "omp driver '{runtime_id}' has no provider argv" ); + anyhow::ensure!( + required_resume_generation.is_some() == required_incarnation.is_some(), + "OMP residency launch has an incomplete attempt fence" + ); let required_native_session = match required_resume_generation { Some(generation) => Some(required_residency_resume( &state_dir(catalog_root, &identity), @@ -431,7 +461,7 @@ fn run_with_required_resume( verify_supported_version(&omp_argv[0])?; let executable = std::env::current_exe().context("resolving st2 executable for the omp channel")?; - let session = harness_state::session_token(); + let session = required_incarnation.unwrap_or_else(harness_state::session_token); // The claim is written: it supersedes whatever the predecessor left — including a // still-fresh live record — before the channel or terminal writer act under it. let seq = harness_state::claim(&agent_dir, identity.clone(), "omp", &session)?; @@ -875,6 +905,20 @@ mod tests { #[test] fn mandatory_residency_validation_precedes_any_provider_child() { let temp = tempfile::tempdir().unwrap(); + let empty_incarnation = run_residency_attempt( + temp.path(), + "residency-no-spawn.worker".into(), + "residency-no-spawn.worker".into(), + vec!["omp".into()], + crate::residency::Generation(2), + String::new(), + ) + .unwrap_err(); + assert!( + empty_incarnation + .to_string() + .contains("required runtime incarnation is empty") + ); let marker = temp.path().join("provider-started"); let fake = FakeExecutable::new(&format!( "#!/bin/sh\ntouch '{}'\nprintf 'omp v18.1.7\\n'\n", @@ -898,7 +942,7 @@ mod tests { } #[test] - fn residency_resume_binds_the_exact_native_session_and_generation() { + fn residency_resume_binds_the_exact_native_session_generation_and_incarnation() { let temp = tempfile::tempdir().unwrap(); let state = temp.path().join("state"); record_channel_binding( @@ -975,6 +1019,7 @@ mod tests { "h.worker", "h.worker", crate::residency::Generation(2), + "runtime-next", ) .unwrap() ); @@ -983,7 +1028,7 @@ mod tests { &state, "h.worker", "h.worker", - "runtime-next", + "runtime-stale", "session-exact", Some(crate::residency::Generation(2)), Some("session-exact"), @@ -995,6 +1040,7 @@ mod tests { "h.worker", "h.worker", crate::residency::Generation(2), + "runtime-next", ) .unwrap(), "binding publication alone is not channel readiness" @@ -1011,6 +1057,37 @@ mod tests { "session-exact", "an unconfirmed channel attempt must leave the checkpoint retryable" ); + confirm_channel_binding( + &state, + "h.worker", + "h.worker", + "runtime-stale", + "session-exact", + Some(crate::residency::Generation(2)), + ) + .unwrap(); + assert!( + !residency_ready( + &state, + "h.worker", + "h.worker", + crate::residency::Generation(2), + "runtime-next", + ) + .unwrap(), + "a stale prior-attempt binding became ready" + ); + + record_channel_binding( + &state, + "h.worker", + "h.worker", + "runtime-next", + "session-exact", + Some(crate::residency::Generation(2)), + Some("session-exact"), + ) + .unwrap(); confirm_channel_binding( &state, "h.worker", @@ -1026,6 +1103,7 @@ mod tests { "h.worker", "h.worker", crate::residency::Generation(2), + "runtime-next", ) .unwrap() ); diff --git a/src/provider_session.rs b/src/provider_session.rs index 834804f5..04f9621d 100644 --- a/src/provider_session.rs +++ b/src/provider_session.rs @@ -81,7 +81,24 @@ impl SessionObserver { harness: &'static str, pty_session: &str, ) -> anyhow::Result { - let session = harness_state::session_token(); + Self::with_session( + agent_dir, + identity, + harness, + pty_session, + harness_state::session_token(), + ) + } + + /// Claim an explicitly supplied session incarnation for a host-owned launch attempt. + pub(crate) fn with_session( + agent_dir: &Path, + identity: &str, + harness: &'static str, + pty_session: &str, + session: String, + ) -> anyhow::Result { + anyhow::ensure!(!session.is_empty(), "provider session incarnation is empty"); let seq = harness_state::claim(agent_dir, identity, harness, &session)?; Ok(Self { seq, @@ -176,6 +193,7 @@ fn describe_exit(exit: ExitStatus) -> String { /// `refresh_interval` for exactly as long as the spawned child lives. Fails on a nonzero exit; /// wrappers that need the exit itself use [`run_provider_observed`]. With an observer, the /// terminal record lands on every exit path this process survives. +#[cfg(test)] #[allow(clippy::too_many_arguments)] pub(crate) fn run_provider( provider: &str, @@ -411,6 +429,35 @@ mod tests { assert_eq!(record.reason.as_deref(), Some("launch-error")); } + #[test] + fn an_explicit_session_incarnation_is_claimed_exactly() { + let tmp = tempfile::tempdir().unwrap(); + let _observer = SessionObserver::with_session( + tmp.path(), + "hetz.worker", + "claude", + "hetz.worker", + "attempt-exact".into(), + ) + .unwrap(); + let record: serde_json::Value = serde_json::from_slice( + &std::fs::read(harness_state::harness_state_path(tmp.path())).unwrap(), + ) + .unwrap(); + assert_eq!(record["incarnation"], "attempt-exact"); + + let error = SessionObserver::with_session( + tmp.path(), + "hetz.worker", + "claude", + "hetz.worker", + String::new(), + ) + .err() + .unwrap(); + assert!(error.to_string().contains("incarnation is empty")); + } + #[test] fn explicit_environment_removal_precedes_managed_values() { const FENCE: &str = "ST2_TEST_PROVIDER_FENCE"; diff --git a/src/residency_host.rs b/src/residency_host.rs new file mode 100644 index 00000000..f45f324e --- /dev/null +++ b/src/residency_host.rs @@ -0,0 +1,1049 @@ +use std::collections::BTreeSet; +use std::fs; +use std::path::{Path, PathBuf}; +use std::time::Duration; + +use anyhow::{Context, Result}; +use serde::{Deserialize, Serialize}; + +use crate::harness_state::{Activity, SessionLiveness}; +use crate::reconcile::Session; +use crate::residency::{ + Event, Generation, Ledger, RefusalReason, RuntimePresence, RuntimeResidency, +}; +use crate::{AgentSpec, ResidencyPolicy, Runner, SessionDriver, TaskLifecycle, UpReport}; + +const WAKE_SCHEMA: &str = "st2.residency-wake.v1"; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct HostPolicy { + pub idle_after: Duration, + pub warm_capacity: usize, +} + +#[derive(Debug, Serialize, Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +struct WakeRequest { + schema: String, + agent_id: String, + host: String, + requested_at_ms: u64, +} + +const LAUNCH_SCHEMA: &str = "st2.residency-launch.v1"; + +#[derive(Debug, Serialize, Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +struct LaunchAttempt { + schema: String, + agent_id: String, + host: String, + generation: Generation, + incarnation: String, +} + +#[derive(Debug)] +struct LaunchReceipt { + path: PathBuf, + agent_id: String, + host: String, + driver: SessionDriver, + generation: Generation, + task_ids: Vec, +} + +#[derive(Debug, Default)] +pub struct Pass { + launches: Vec, +} + +fn wake_dir(catalog: &Path) -> PathBuf { + catalog.join(".st2").join("control").join("residency-wake") +} + +fn wake_path(catalog: &Path, host: &str, agent_id: &str) -> PathBuf { + let mut key = sha2::Sha256::new(); + use sha2::Digest as _; + for value in [host.as_bytes(), agent_id.as_bytes()] { + key.update((value.len() as u64).to_be_bytes()); + key.update(value); + } + wake_dir(catalog).join(format!("{:x}.json", key.finalize())) +} + +fn launch_path(catalog: &Path, host: &str, agent_id: &str) -> PathBuf { + let file = wake_path(catalog, host, agent_id) + .file_name() + .expect("hashed wake path has a file name") + .to_owned(); + catalog + .join(".st2") + .join("control") + .join("residency-launch") + .join(file) +} + +fn load_launch_attempt( + catalog: &Path, + host: &str, + agent_id: &str, + generation: Generation, +) -> Result> { + let path = launch_path(catalog, host, agent_id); + let bytes = match fs::read(path) { + Ok(bytes) => bytes, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None), + Err(error) => return Err(error.into()), + }; + let attempt: LaunchAttempt = serde_json::from_slice(&bytes)?; + anyhow::ensure!( + attempt.schema == LAUNCH_SCHEMA + && attempt.agent_id == agent_id + && attempt.host == host + && attempt.generation == generation + && !attempt.incarnation.is_empty(), + "residency launch attempt ownership, generation, or schema mismatch" + ); + Ok(Some(attempt)) +} + +fn prepare_launch_attempt( + catalog: &Path, + host: &str, + agent_id: &str, + generation: Generation, + canonical_present: bool, +) -> Result { + if canonical_present { + return load_launch_attempt(catalog, host, agent_id, generation)? + .context("present residency provider has no launch-attempt fence"); + } + let attempt = LaunchAttempt { + schema: LAUNCH_SCHEMA.to_owned(), + agent_id: agent_id.to_owned(), + host: host.to_owned(), + generation, + incarnation: crate::harness_state::session_token(), + }; + crate::residency::atomic_json(&launch_path(catalog, host, agent_id), &attempt)?; + Ok(attempt) +} + +pub fn request_wake(catalog: &Path, host: &str, agent_id: &str) -> Result { + let path = wake_path(catalog, host, agent_id); + let request = WakeRequest { + schema: WAKE_SCHEMA.to_owned(), + agent_id: agent_id.to_owned(), + host: host.to_owned(), + requested_at_ms: crate::message::now_ms(), + }; + crate::residency::atomic_json(&path, &request)?; + tracing::info!( + target: "st2", + agent_id, + host, + requested_at_ms = request.requested_at_ms, + "residency wake requested" + ); + Ok(path) +} + +fn has_wake_request(catalog: &Path, host: &str, agent_id: &str) -> Result { + let path = wake_path(catalog, host, agent_id); + let bytes = match fs::read(&path) { + Ok(bytes) => bytes, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(false), + Err(error) => return Err(error.into()), + }; + let request: WakeRequest = serde_json::from_slice(&bytes)?; + anyhow::ensure!( + request.schema == WAKE_SCHEMA && request.agent_id == agent_id && request.host == host, + "residency wake request ownership or schema mismatch" + ); + Ok(true) +} + +fn task_ids(spec: &AgentSpec, host: &str) -> Vec { + let bus_id = spec.bus_id(host); + spec.tasks + .iter() + .map(|task| crate::reconcile::resolve_task_id(&bus_id, &task.name, task.id.as_deref())) + .collect() +} + +fn canonical_task(spec: &AgentSpec) -> Result<&crate::Task> { + spec.tasks + .iter() + .find(|task| task.name == "agent") + .context("on-demand agent has no canonical agent task") +} + +fn canonical_runtime_id(spec: &AgentSpec, host: &str) -> Result { + let task = canonical_task(spec)?; + Ok(crate::reconcile::resolve_task_id( + &spec.bus_id(host), + &task.name, + task.id.as_deref(), + )) +} + +fn provider_argv(task: &crate::Task) -> Result<&[String]> { + let argv = task + .argv + .as_deref() + .context("on-demand canonical task has no compiled argv")?; + let runtime = argv + .iter() + .position(|argument| argument == "--runtime-id") + .context("on-demand wrapper argv has no --runtime-id")?; + anyhow::ensure!( + runtime + 2 < argv.len(), + "on-demand wrapper has no provider argv" + ); + Ok(&argv[runtime + 2..]) +} + +fn workspace(spec: &AgentSpec, task: &crate::Task, catalog: &Path) -> PathBuf { + let spec_dir = spec.path.parent().unwrap_or_else(|| Path::new(".")); + match task.cwd.as_deref().or(spec.workspace.as_deref()) { + Some(path) => spec_dir.join(crate::expand::expand_catalog(path, catalog)), + None => spec_dir.to_path_buf(), + } +} + +fn inject_resume_fence( + spec: &mut AgentSpec, + generation: Generation, + incarnation: &str, +) -> Result<()> { + let task = spec + .tasks + .iter_mut() + .find(|task| task.name == "agent") + .context("on-demand agent has no canonical agent task")?; + let argv = task + .argv + .as_mut() + .context("on-demand canonical task has no compiled argv")?; + anyhow::ensure!( + !argv.iter().any(|argument| { + argument == "--required-resume-generation" + || argument == "--required-resume-incarnation" + }), + "on-demand wrapper already carries a resume fence" + ); + let runtime = argv + .iter() + .position(|argument| argument == "--runtime-id") + .context("on-demand wrapper argv has no --runtime-id")?; + anyhow::ensure!( + runtime + 1 < argv.len(), + "on-demand wrapper has no runtime id value" + ); + argv.splice( + runtime + 2..runtime + 2, + [ + "--required-resume-generation".to_owned(), + generation.0.to_string(), + "--required-resume-incarnation".to_owned(), + incarnation.to_owned(), + ], + ); + Ok(()) +} + +fn group_present(ids: &[String], sessions: &[Session]) -> bool { + ids.iter().all(|id| { + sessions + .iter() + .any(|session| session.pty_id == *id && session.alive) + }) +} + +fn group_absent(ids: &[String], sessions: &[Session]) -> bool { + ids.iter().all(|id| { + !sessions + .iter() + .any(|session| session.pty_id == *id && session.alive) + }) +} + +fn group_presence(ids: &[String], sessions: &[Session]) -> RuntimePresence { + if group_absent(ids, sessions) { + RuntimePresence::Absent + } else { + RuntimePresence::Present + } +} + +fn store_event(path: &Path, ledger: &mut Ledger, event: Event) -> Result<()> { + let mut next = ledger.clone(); + let transition = next.apply(event); + crate::residency::store(path, &next).map_err(anyhow::Error::from)?; + tracing::info!( + target: "st2", + agent_id = next.agent_id(), + host = next.host(), + generation = next.generation().0, + event = ?event, + outcome = ?transition.outcome, + "residency transition" + ); + *ledger = next; + Ok(()) +} + +fn refuse( + path: &Path, + ledger: &mut Ledger, + presence: RuntimePresence, + report: &mut UpReport, + context: &str, + error: anyhow::Error, +) { + let generation = ledger.generation(); + ledger.apply(Event::Refused { + generation, + reason: RefusalReason::InvalidFence, + presence, + }); + if let Err(store_error) = crate::residency::store(path, ledger) { + report.errors.push(format!( + "{context}: {error:#}; store residency refusal: {store_error}" + )); + } else { + report.errors.push(format!("{context}: {error:#}")); + } +} + +fn checkpoint( + catalog: &Path, + spec: &AgentSpec, + host: &str, + driver: SessionDriver, + source: Generation, + resume: Generation, +) -> Result<()> { + let bus_id = spec.bus_id(host); + let runtime_id = canonical_runtime_id(spec, host)?; + match driver { + SessionDriver::Codex => crate::codex_app_server::checkpoint_residency( + &crate::codex_app_server::state_dir(catalog, &bus_id), + &bus_id, + &runtime_id, + source, + resume, + ) + .map(|_| ()), + SessionDriver::Omp => crate::omp_session::checkpoint_residency( + &crate::omp_session::state_dir(catalog, &bus_id), + &bus_id, + &runtime_id, + source, + resume, + ) + .map(|_| ()), + SessionDriver::Claude => crate::claude_session::checkpoint_residency( + &crate::claude_session::state_dir(catalog, &bus_id), + &bus_id, + &runtime_id, + source, + resume, + ) + .map(|_| ()), + _ => anyhow::bail!("native residency resume is unsupported for {driver:?}"), + } +} + +fn prepare( + catalog: &Path, + spec: &AgentSpec, + host: &str, + driver: SessionDriver, + generation: Generation, +) -> Result<()> { + let bus_id = spec.bus_id(host); + let runtime_id = canonical_runtime_id(spec, host)?; + let task = canonical_task(spec)?; + let argv = provider_argv(task)?; + match driver { + SessionDriver::Codex => crate::codex_app_server::required_residency_resume( + &crate::codex_app_server::state_dir(catalog, &bus_id), + &bus_id, + &runtime_id, + generation, + &argv[1..], + ) + .map(|_| ()), + SessionDriver::Omp => crate::omp_session::required_residency_resume( + &crate::omp_session::state_dir(catalog, &bus_id), + &bus_id, + &runtime_id, + generation, + &argv[1..], + ) + .map(|_| ()), + SessionDriver::Claude => crate::claude_session::required_residency_resume( + &crate::claude_session::state_dir(catalog, &bus_id), + &bus_id, + &runtime_id, + &workspace(spec, task, catalog), + generation, + argv, + ) + .map(|_| ()), + _ => anyhow::bail!("native residency resume is unsupported for {driver:?}"), + } +} + +fn verify_native( + catalog: &Path, + spec: &AgentSpec, + host: &str, + driver: SessionDriver, + generation: Generation, + expected_incarnation: &str, +) -> Result { + let bus_id = spec.bus_id(host); + let runtime_id = canonical_runtime_id(spec, host)?; + match driver { + SessionDriver::Codex => crate::codex_app_server::residency_ready( + &crate::codex_app_server::state_dir(catalog, &bus_id), + &bus_id, + &runtime_id, + generation, + expected_incarnation, + ), + SessionDriver::Omp => crate::omp_session::residency_ready( + &crate::omp_session::state_dir(catalog, &bus_id), + &bus_id, + &runtime_id, + generation, + expected_incarnation, + ), + SessionDriver::Claude => crate::claude_session::residency_ready( + &crate::claude_session::state_dir(catalog, &bus_id), + &bus_id, + &runtime_id, + generation, + expected_incarnation, + ), + _ => anyhow::bail!("native residency resume is unsupported for {driver:?}"), + } +} + +fn demand(catalog: &Path, spec: &AgentSpec, host: &str) -> Result<(bool, bool)> { + let agent_id = spec.effective_id(host); + let requested = has_wake_request(catalog, host, &agent_id)?; + let agent_dir = spec.path.parent().unwrap_or_else(|| Path::new(".")); + let inbox = !crate::message::list_inbox(&crate::message::inbox_dir(agent_dir))?.is_empty(); + Ok((requested || inbox, requested)) +} + +fn idle_since(spec: &AgentSpec, sessions: &[Session]) -> Option { + let expected = spec.effective_session_driver()?.as_str(); + let alive = |id: &str| { + if sessions + .iter() + .any(|session| session.pty_id == id && session.alive) + { + SessionLiveness::Alive + } else { + SessionLiveness::Dead + } + }; + let agent_dir = spec.path.parent().unwrap_or_else(|| Path::new(".")); + let observed = crate::harness_state::read( + &crate::harness_state::harness_state_path(agent_dir), + Some(&alive), + )?; + (observed.state == Activity::Idle && observed.harness.as_deref() == Some(expected)) + .then_some(observed.since_ms) + .flatten() +} + +pub fn before_reconcile( + catalog: &Path, + host: &str, + specs: &mut [AgentSpec], + sessions: &[Session], + runner: &dyn Runner, + policy: HostPolicy, + report: &mut UpReport, +) -> Pass { + let mut pass = Pass::default(); + let mut ledgers = Vec::new(); + let mut failed_specs = BTreeSet::new(); + for (index, spec) in specs.iter().enumerate() { + if spec.residency_policy != ResidencyPolicy::OnDemand + || !spec.desired_state.is_running() + || spec.resolved_host(host) != host + { + continue; + } + let Some(driver) = spec.effective_session_driver() else { + report.errors.push(format!( + "on-demand agent {} has no native session driver", + spec.bus_id(host) + )); + failed_specs.insert(index); + continue; + }; + let agent_id = spec.effective_id(host); + let path = crate::residency::ledger_path(catalog, host, &agent_id); + let ledger = match crate::residency::load(&path, &agent_id, host, driver) { + Ok(Some(ledger)) => ledger, + Ok(None) => match Ledger::active(&agent_id, host, driver, Generation(1)) { + Ok(ledger) => { + if let Err(error) = crate::residency::store(&path, &ledger) { + report + .errors + .push(format!("initialize residency for {agent_id}: {error}")); + failed_specs.insert(index); + continue; + } + ledger + } + Err(error) => { + report + .errors + .push(format!("initialize residency for {agent_id}: {error}")); + failed_specs.insert(index); + continue; + } + }, + Err(error) => { + report + .errors + .push(format!("load residency for {agent_id}: {error}")); + failed_specs.insert(index); + continue; + } + }; + ledgers.push((index, path, driver, ledger)); + } + + for index in failed_specs { + for task in &mut specs[index].tasks { + task.lifecycle = TaskLifecycle::AdoptOnly; + } + } + let active_count = ledgers + .iter() + .filter(|(_, _, _, ledger)| ledger.runtime_residency() == &RuntimeResidency::Active) + .count(); + let excess = active_count.saturating_sub(policy.warm_capacity); + let mut idle = ledgers + .iter() + .filter_map(|(index, _, _, ledger)| { + (ledger.runtime_residency() == &RuntimeResidency::Active) + .then(|| idle_since(&specs[*index], sessions).map(|since| (*index, since))) + .flatten() + }) + .filter(|(_, since)| { + crate::message::now_ms().saturating_sub(*since) + >= u64::try_from(policy.idle_after.as_millis()).unwrap_or(u64::MAX) + }) + .collect::>(); + idle.sort_by_key(|(_, since)| *since); + let suspend = idle + .into_iter() + .take(excess) + .map(|(index, _)| index) + .collect::>(); + + for (index, path, driver, mut ledger) in ledgers { + let ids = task_ids(&specs[index], host); + let presence = group_presence(&ids, sessions); + let (has_demand, requested, demand_observed) = match demand(catalog, &specs[index], host) { + Ok((has_demand, requested)) => (has_demand, requested, true), + Err(error) => { + report.errors.push(format!( + "observe residency demand for {}: {error:#}", + specs[index].bus_id(host) + )); + (false, false, false) + } + }; + if has_demand { + if let Err(error) = store_event(&path, &mut ledger, Event::WakeDemandObserved) { + report.errors.push(format!( + "record residency demand for {}: {error:#}", + specs[index].bus_id(host) + )); + } else if requested { + let _ = fs::remove_file(wake_path(catalog, host, &specs[index].effective_id(host))); + } + } + if suspend.contains(&index) && !has_demand && demand_observed { + let generation = ledger.generation(); + if let Err(error) = store_event(&path, &mut ledger, Event::IdleConfirmed { generation }) + { + report.errors.push(format!( + "record idle residency for {}: {error:#}", + specs[index].bus_id(host) + )); + } + } + + loop { + let Some(action) = ledger.next_action() else { + break; + }; + let result = match action { + crate::residency::Action::CheckpointNativeSession { source, resume } => { + checkpoint(catalog, &specs[index], host, driver, source, resume).and_then( + |()| { + store_event( + &path, + &mut ledger, + Event::CheckpointStored { source, resume }, + ) + }, + ) + } + crate::residency::Action::StopOwnedGroup { generation } => { + let mut failed = None; + for id in &ids { + if sessions + .iter() + .any(|session| session.pty_id == *id && session.alive) + && let Err(error) = runner.kill(id) + { + let error = error.context(format!("stopping residency task {id}")); + if failed.is_none() { + failed = Some(error); + } + } + } + match failed { + Some(error) => Err(error), + None => { + store_event(&path, &mut ledger, Event::OwnedGroupStopped { generation }) + } + } + } + crate::residency::Action::VerifyAbsent { generation } => { + if group_absent(&ids, sessions) { + store_event(&path, &mut ledger, Event::AbsenceVerified { generation }) + } else { + break; + } + } + crate::residency::Action::PrepareNativeResume { generation } => { + prepare(catalog, &specs[index], host, driver, generation).and_then(|()| { + store_event( + &path, + &mut ledger, + Event::NativeResumePrepared { generation }, + ) + }) + } + crate::residency::Action::LaunchOwnedGroup { generation } => { + let agent_id = specs[index].effective_id(host); + let canonical_id = canonical_runtime_id(&specs[index], host); + let canonical_present = canonical_id.as_ref().is_ok_and(|canonical_id| { + sessions + .iter() + .any(|session| session.pty_id == *canonical_id && session.alive) + }); + match canonical_id + .and_then(|_| { + prepare_launch_attempt( + catalog, + host, + &agent_id, + generation, + canonical_present, + ) + }) + .and_then(|attempt| { + inject_resume_fence(&mut specs[index], generation, &attempt.incarnation) + }) + { + Ok(()) => pass.launches.push(LaunchReceipt { + path: path.clone(), + agent_id, + host: host.to_owned(), + driver, + generation, + task_ids: ids.clone(), + }), + Err(error) => refuse( + &path, + &mut ledger, + presence, + report, + "prepare residency launch", + error, + ), + } + break; + } + crate::residency::Action::VerifyNativeSession { generation } => { + if !group_present(&ids, sessions) { + break; + } + let agent_id = specs[index].effective_id(host); + let verified = load_launch_attempt(catalog, host, &agent_id, generation) + .and_then(|attempt| { + attempt.context("residency launch-attempt fence disappeared") + }) + .and_then(|attempt| { + if !verify_native( + catalog, + &specs[index], + host, + driver, + generation, + &attempt.incarnation, + )? { + return Ok(false); + } + store_event( + &path, + &mut ledger, + Event::NativeSessionVerified { generation }, + )?; + let _ = fs::remove_file(launch_path(catalog, host, &agent_id)); + Ok(true) + }); + match verified { + Ok(true) => Ok(()), + Ok(false) => break, + Err(error) => Err(error), + } + } + }; + if let Err(error) = result { + if matches!(action, crate::residency::Action::StopOwnedGroup { .. }) { + report.errors.push(format!( + "residency action for {}: {error:#}", + specs[index].bus_id(host) + )); + } else { + refuse( + &path, + &mut ledger, + presence, + report, + &format!("residency action for {}", specs[index].bus_id(host)), + error, + ); + } + break; + } + } + + if !matches!( + ledger.runtime_residency(), + RuntimeResidency::Active + | RuntimeResidency::Starting { + step: crate::residency::StartStep::LaunchOwnedGroup, + .. + } + ) { + for task in &mut specs[index].tasks { + task.lifecycle = TaskLifecycle::AdoptOnly; + } + } + } + pass +} + +pub fn after_reconcile(runner: &dyn Runner, pass: Pass, report: &mut UpReport) { + if pass.launches.is_empty() { + return; + } + let sessions = match runner.list_sessions() { + Ok(sessions) => sessions, + Err(error) => { + report + .errors + .push(format!("verify launched residency groups: {error}")); + return; + } + }; + for launch in pass.launches { + if !group_present(&launch.task_ids, &sessions) { + continue; + } + let mut ledger = match crate::residency::load( + &launch.path, + &launch.agent_id, + &launch.host, + launch.driver, + ) { + Ok(Some(ledger)) => ledger, + Ok(None) => { + report.errors.push(format!( + "verify residency launch for {}: ledger disappeared", + launch.agent_id + )); + continue; + } + Err(error) => { + report.errors.push(format!( + "verify residency launch for {}: {error}", + launch.agent_id + )); + continue; + } + }; + if let Err(error) = store_event( + &launch.path, + &mut ledger, + Event::OwnedGroupLaunched { + generation: launch.generation, + }, + ) { + report.errors.push(format!( + "record residency launch for {}: {error:#}", + launch.agent_id + )); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn session(id: &str, alive: bool) -> Session { + Session { + pty_id: id.to_owned(), + alive, + exit_code: None, + presentation: None, + } + } + + struct FailingKillRunner { + killed: std::cell::RefCell>, + } + + impl Runner for FailingKillRunner { + fn list_sessions(&self) -> Result> { + unreachable!() + } + + fn spawn(&self, _target: &crate::reconcile::TaskTarget, _spec_dir: &Path) -> Result<()> { + unreachable!() + } + + fn kill(&self, pty_id: &str) -> Result<()> { + let first = self.killed.borrow().is_empty(); + self.killed.borrow_mut().push(pty_id.to_owned()); + if first { + anyhow::bail!("injected kill failure") + } + Ok(()) + } + + fn remove(&self, _pty_id: &str) -> Result<()> { + unreachable!() + } + } + + fn on_demand_spec(path: PathBuf) -> AgentSpec { + let task = |name: &str| agent_spec::spec::Task { + kind: agent_spec::spec::TaskKind::Pty, + derived: name != "agent", + name: name.to_owned(), + id: None, + command: None, + argv: Some(vec![ + "wrapper".to_owned(), + "--runtime-id".to_owned(), + "runtime".to_owned(), + ]), + cwd: None, + tags: Default::default(), + env: Default::default(), + keep: false, + lifecycle: TaskLifecycle::Service, + }; + AgentSpec { + id: None, + address: None, + identity: "worker".to_owned(), + name: None, + description: None, + host: Some("host-a".to_owned()), + role: None, + job_type: agent_spec::spec::JobType::Service, + workspace: None, + supervisor: None, + desired_state: agent_spec::spec::AgentDesiredState::Running, + residency_policy: ResidencyPolicy::OnDemand, + keep: false, + restart: None, + delivery: None, + session_driver: Some(SessionDriver::Omp), + driver: None, + delivery_readiness: None, + resources: Vec::new(), + streams: Vec::new(), + tasks: vec![task("agent"), task("sidecar")], + path, + } + } + + #[test] + fn missing_provider_retries_rotate_the_attempt_fence_and_live_providers_retain_it() { + let catalog = tempfile::tempdir().unwrap(); + let mut spec = on_demand_spec(catalog.path().join("agent.kdl")); + let first = prepare_launch_attempt( + catalog.path(), + "host-a", + "host-a.worker", + Generation(2), + false, + ) + .unwrap(); + let second = prepare_launch_attempt( + catalog.path(), + "host-a", + "host-a.worker", + Generation(2), + false, + ) + .unwrap(); + let retained = prepare_launch_attempt( + catalog.path(), + "host-a", + "host-a.worker", + Generation(2), + true, + ) + .unwrap(); + + assert_ne!(first.incarnation, second.incarnation); + assert_eq!(retained.incarnation, second.incarnation); + inject_resume_fence(&mut spec, Generation(2), &second.incarnation).unwrap(); + let argv = spec.tasks[0].argv.as_ref().unwrap(); + assert!( + argv.windows(2) + .any(|pair| { pair[0] == "--required-resume-generation" && pair[1] == "2" }) + ); + assert!(argv.windows(2).any(|pair| { + pair[0] == "--required-resume-incarnation" && pair[1] == second.incarnation + })); + } + + #[test] + fn stop_failures_retry_without_refusing_and_attempt_the_complete_group() { + let catalog = tempfile::tempdir().unwrap(); + let agent_dir = catalog.path().join("agents/host-a/worker"); + std::fs::create_dir_all(agent_dir.join("inbox")).unwrap(); + let mut spec = on_demand_spec(agent_dir.join("agent.kdl")); + let agent_id = spec.effective_id("host-a"); + let path = crate::residency::ledger_path(catalog.path(), "host-a", &agent_id); + let mut ledger = + Ledger::active(&agent_id, "host-a", SessionDriver::Omp, Generation(1)).unwrap(); + ledger.apply(Event::IdleConfirmed { + generation: Generation(1), + }); + ledger.apply(Event::CheckpointStored { + source: Generation(1), + resume: Generation(2), + }); + crate::residency::store(&path, &ledger).unwrap(); + let ids = task_ids(&spec, "host-a"); + let sessions = ids.iter().map(|id| session(id, true)).collect::>(); + let runner = FailingKillRunner { + killed: std::cell::RefCell::new(Vec::new()), + }; + let mut report = UpReport::default(); + + before_reconcile( + catalog.path(), + "host-a", + std::slice::from_mut(&mut spec), + &sessions, + &runner, + HostPolicy { + idle_after: Duration::ZERO, + warm_capacity: 0, + }, + &mut report, + ); + + assert_eq!(&*runner.killed.borrow(), &ids); + assert!( + spec.tasks + .iter() + .all(|task| task.lifecycle == TaskLifecycle::AdoptOnly) + ); + let stored = crate::residency::load(&path, &agent_id, "host-a", SessionDriver::Omp) + .unwrap() + .unwrap(); + assert!(matches!( + stored.runtime_residency(), + RuntimeResidency::Stopping { + step: crate::residency::StopStep::StopOwnedGroup, + .. + } + )); + assert!( + report + .errors + .iter() + .any(|error| error.contains("injected kill failure")) + ); + } + + #[test] + fn failed_persistence_does_not_advance_the_in_memory_ledger() { + let temp = tempfile::tempdir().unwrap(); + let blocker = temp.path().join("not-a-directory"); + std::fs::write(&blocker, "block").unwrap(); + let mut ledger = + Ledger::active("agent-a", "host-a", SessionDriver::Omp, Generation(1)).unwrap(); + + assert!( + store_event( + &blocker.join("ledger.json"), + &mut ledger, + Event::IdleConfirmed { + generation: Generation(1), + }, + ) + .is_err() + ); + assert_eq!(ledger.runtime_residency(), &RuntimeResidency::Active); + } + + #[test] + fn wake_requests_are_scoped_to_exact_host_and_agent_ownership() { + let catalog = tempfile::tempdir().unwrap(); + let path = request_wake(catalog.path(), "host-a", "agent-a").unwrap(); + + assert!(has_wake_request(catalog.path(), "host-a", "agent-a").unwrap()); + assert!(!has_wake_request(catalog.path(), "host-b", "agent-a").unwrap()); + assert!(!has_wake_request(catalog.path(), "host-a", "agent-b").unwrap()); + assert!(!path.to_string_lossy().contains("agent-a")); + } + + #[test] + fn malformed_wake_requests_fail_closed() { + let catalog = tempfile::tempdir().unwrap(); + let path = request_wake(catalog.path(), "host-a", "agent-a").unwrap(); + std::fs::write(path, br#"{"schema":"wrong"}"#).unwrap(); + + assert!(has_wake_request(catalog.path(), "host-a", "agent-a").is_err()); + } + + #[test] + fn group_inventory_requires_every_task_for_presence_and_none_for_absence() { + let ids = vec!["agent".to_owned(), "sidecar".to_owned()]; + let partial = vec![session("agent", true), session("sidecar", false)]; + let complete = vec![session("agent", true), session("sidecar", true)]; + + assert!(!group_present(&ids, &partial)); + assert!(!group_absent(&ids, &partial)); + assert!(group_present(&ids, &complete)); + assert!(!group_absent(&ids, &complete)); + assert!(group_absent(&ids, &[])); + } +} diff --git a/src/run.rs b/src/run.rs index 67d2b5ad..69566edc 100644 --- a/src/run.rs +++ b/src/run.rs @@ -1900,6 +1900,7 @@ fn live_resync_specs( /// pass is SKIPPED (the error is recorded but nothing is reconciled) — treating a transient list /// failure as "no sessions" would double-spawn everything. `cap` carries flapping state across passes; /// `debounce` carries per-id liveness so a transient not-alive flicker isn't destructively reaped. +#[cfg(test)] fn reconcile_pass( root: &Path, this_host: &str, @@ -1910,6 +1911,32 @@ fn reconcile_pass( presentation_cursor: &mut PresentationPatchCursor, resync: Option<&crate::resync::ResyncSupervisor>, resource_profiles: Option<&crate::resource_profile_supervisor::ResourceProfileSupervisor>, +) -> UpReport { + reconcile_pass_with_residency( + root, + this_host, + task_context, + runner, + cap, + debounce, + presentation_cursor, + resync, + resource_profiles, + None, + ) +} + +fn reconcile_pass_with_residency( + root: &Path, + this_host: &str, + task_context: &TaskCompileContext, + runner: &dyn Runner, + cap: &mut FlappingCap, + debounce: &mut LivenessDebounce, + presentation_cursor: &mut PresentationPatchCursor, + resync: Option<&crate::resync::ResyncSupervisor>, + resource_profiles: Option<&crate::resource_profile_supervisor::ResourceProfileSupervisor>, + residency_policy: Option, ) -> UpReport { let catalog_lock = { let span = catalog_lock_span(); @@ -2062,7 +2089,7 @@ fn reconcile_pass( } compiled_specs.push(spec); } - let eligible_specs = compiled_specs + let mut eligible_specs = compiled_specs .iter() .filter(|spec| !materialized.failed_agents.contains(&spec.bus_id(this_host))) .cloned() @@ -2090,6 +2117,37 @@ fn reconcile_pass( }; let now = Instant::now(); debounce.observe(&sessions, now); + if residency_policy.is_none() { + let mut gated = Vec::new(); + for spec in &mut eligible_specs { + if spec.residency_policy == crate::ResidencyPolicy::OnDemand + && spec.desired_state.is_running() + && spec.resolved_host(this_host) == this_host + { + for task in &mut spec.tasks { + task.lifecycle = crate::TaskLifecycle::AdoptOnly; + } + gated.push(spec.bus_id(this_host)); + } + } + if !gated.is_empty() { + report.errors.push(format!( + "on-demand agents require host --residency-idle-after and --residency-warm-capacity: {}; launches suppressed", + gated.join(", ") + )); + } + } + let residency_pass = residency_policy.map(|policy| { + crate::residency_host::before_reconcile( + root, + this_host, + &mut eligible_specs, + &sessions, + runner, + policy, + &mut report, + ) + }); let mut plan = match crate::reconcile(&eligible_specs, &sessions, this_host) { Ok(plan) => plan, Err(error) => { @@ -2141,6 +2199,9 @@ fn reconcile_pass( &mut report, &mut install_new_live_seat, ); + if let Some(pass) = residency_pass { + crate::residency_host::after_reconcile(runner, pass, &mut report); + } report.warnings.extend(boundary_warnings); if resync.is_some() || resource_profiles.is_some() { let loaded = crate::catalog::declared_profile_catalog(root) @@ -2512,20 +2573,42 @@ fn finish_failed_reconcile_pass(span: &tracing::Span) { /// never `Err` — all failures are collected in `report.errors`. The debounce is throwaway too: a /// single pass has no prior liveness history, so it defers nothing (correct — one-shot has no flicker). pub fn up_once(root: &Path, this_host: &str, runner: &dyn Runner) -> anyhow::Result { + up_once_with_optional_residency(root, this_host, runner, None) +} + +pub fn up_once_with_residency( + root: &Path, + this_host: &str, + runner: &dyn Runner, + policy: crate::residency_host::HostPolicy, +) -> anyhow::Result { + up_once_with_optional_residency(root, this_host, runner, Some(policy)) +} + +fn up_once_with_optional_residency( + root: &Path, + this_host: &str, + runner: &dyn Runner, + policy: Option, +) -> anyhow::Result { let task_context = TaskCompileContext::current(root.to_path_buf())?; let mut debounce = LivenessDebounce::new(DEBOUNCE_GRACE); let started = Instant::now(); let span = reconcile_span(this_host, "catalog"); let report = { let _entered = span.enter(); - let report = reconcile_pass(root, - this_host, - &task_context, - runner, - &mut FlappingCap::default(), - &mut debounce, - &mut PresentationPatchCursor::default(), - None, None); + let report = reconcile_pass_with_residency( + root, + this_host, + &task_context, + runner, + &mut FlappingCap::default(), + &mut debounce, + &mut PresentationPatchCursor::default(), + None, + None, + policy, + ); finish_reconcile_pass(&span, &report); report }; @@ -3124,6 +3207,27 @@ pub fn up_loop( ) } +pub fn up_loop_with_residency( + root: &Path, + this_host: &str, + runner: &dyn Runner, + interval: Duration, + policy: crate::residency_host::HostPolicy, + on_report: impl FnMut(&UpReport), +) -> anyhow::Result<()> { + install_signal_handler(); + up_loop_until_with_residency( + root, + this_host, + runner, + interval, + &STOP, + best_effort_catalog_watcher, + Some(policy), + on_report, + ) +} + fn up_loop_until( root: &Path, this_host: &str, @@ -3131,6 +3235,28 @@ fn up_loop_until( interval: Duration, stop: &AtomicBool, install_watcher: impl FnOnce(&Path, Sender<()>) -> Option, + on_report: impl FnMut(&UpReport), +) -> anyhow::Result<()> { + up_loop_until_with_residency( + root, + this_host, + runner, + interval, + stop, + install_watcher, + None, + on_report, + ) +} + +fn up_loop_until_with_residency( + root: &Path, + this_host: &str, + runner: &dyn Runner, + interval: Duration, + stop: &AtomicBool, + install_watcher: impl FnOnce(&Path, Sender<()>) -> Option, + residency_policy: Option, mut on_report: impl FnMut(&UpReport), ) -> anyhow::Result<()> { let task_context = TaskCompileContext::current(root.to_path_buf())?; @@ -3199,15 +3325,18 @@ fn up_loop_until( let span = reconcile_span(this_host, "catalog"); let pass = { let _entered = span.enter(); - let pass = reconcile_pass(root, - this_host, - &task_context, - runner, - &mut cap, - &mut debounce, - &mut presentation_cursor, - resync.as_ref(), - resource_profiles.as_ref()); + let pass = reconcile_pass_with_residency( + root, + this_host, + &task_context, + runner, + &mut cap, + &mut debounce, + &mut presentation_cursor, + resync.as_ref(), + resource_profiles.as_ref(), + residency_policy, + ); finish_reconcile_pass(&span, &pass); pass }; @@ -3347,16 +3476,13 @@ pub fn detect_host() -> String { mod tests { use super::*; - use agent_spec::spec::{ - AgentSpec, Driver, JobType, OmpDriver, Task, TaskKind, TaskLifecycle, - }; + use agent_spec::spec::{AgentSpec, Driver, JobType, OmpDriver, Task, TaskKind, TaskLifecycle}; use std::cell::{Cell, RefCell}; use std::collections::{BTreeMap, BTreeSet}; use std::ffi::OsStr; use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering}; use std::sync::mpsc; - /// The bound is derived from the resolved pty root, never a fixed maximum identity length. /// /// `pty` binds `/.sock`, so the separator plus the five-byte suffix is @@ -4037,6 +4163,33 @@ mod tests { } } + #[test] + fn on_demand_agents_do_not_launch_without_host_policy() { + let catalog = tempfile::tempdir().unwrap(); + let agent_dir = catalog.path().join("agents/hetz/worker"); + std::fs::create_dir_all(&agent_dir).unwrap(); + std::fs::write( + agent_dir.join("agent.kdl"), + r#"agent "worker" { + host "hetz" + residency-policy "on-demand" + session-driver "omp" + argv "omp" "--prompt" "continue" +} +"#, + ) + .unwrap(); + let runner = SpawnCountingRunner::default(); + + let report = up_once(catalog.path(), "hetz", &runner).unwrap(); + + assert!(runner.spawned.borrow().is_empty()); + assert!(report.errors.iter().any(|error| { + error.contains("on-demand agents require host --residency-idle-after") + && error.contains("hetz.worker") + })); + } + /// `execute` must close every pass, or a task that recovers is never forgiven and eventually /// parks even though it is healthy — the opposite of the crash-loop bug the cap exists for. /// @@ -4254,11 +4407,8 @@ mod tests { live_derived: Vec::new(), }; let mut omp_argv = target("hetz.demo.agent", "unused"); - omp_argv.launch = TaskLaunch::Argv(vec![ - "st2".into(), - "driver".into(), - "omp-session".into(), - ]); + omp_argv.launch = + TaskLaunch::Argv(vec!["st2".into(), "driver".into(), "omp-session".into()]); let mut exec = target("hetz.demo.agent", "codex"); exec.kind = TaskKind::Exec; let targets = [ @@ -4570,19 +4720,6 @@ mod tests { (agent_dir, goal) } - #[cfg(all(test, feature = "wasm-resolver"))] - fn current_resync_event_for_key(agent_dir: &Path, key: &str) -> Option { - let expected = format!("key: {key}"); - std::fs::read_dir(agent_dir.join("resources/inbox")) - .ok()? - .filter_map(Result::ok) - .filter_map(|entry| std::fs::read_to_string(entry.path()).ok()) - .find(|body| { - body.lines().any(|line| line == "stream: resync") - && body.lines().any(|line| line == expected) - }) - } - #[cfg(all(test, feature = "wasm-resolver"))] fn wait_for_resync_event_for_key(agent_dir: &Path, key: &str) -> Option { let deadline = Instant::now() + Duration::from_secs(5); @@ -4658,8 +4795,7 @@ mod tests { fn up_loop_keeps_complete_notify_chain_sets_during_steady_reconcile() { let catalog = tempfile::tempdir().unwrap(); write_notify_chain_profile(catalog.path()); - let (root_dir, root_goal) = - write_notify_chain_agent(catalog.path(), "root", None, false); + let (root_dir, root_goal) = write_notify_chain_agent(catalog.path(), "root", None, false); let (lead_dir, lead_goal) = write_notify_chain_agent(catalog.path(), "lead", Some("hetz.root"), false); let (worker_dir, _worker_goal) = @@ -4695,17 +4831,10 @@ mod tests { std::fs::write(&root_goal, "steady baseline transition\n").unwrap(); let root_initial = wait_for_resync_event_for_key(&root_dir, "goal"); - let lead_initial = - wait_for_resync_event_for_key(&lead_dir, "goal@hetz.root"); - let worker_initial = - wait_for_resync_event_for_key(&worker_dir, "goal@hetz.root"); - - write_notify_chain_agent( - &observer_catalog, - "worker", - Some("hetz.lead"), - true, - ); + let lead_initial = wait_for_resync_event_for_key(&lead_dir, "goal@hetz.root"); + let worker_initial = wait_for_resync_event_for_key(&worker_dir, "goal@hetz.root"); + + write_notify_chain_agent(&observer_catalog, "worker", Some("hetz.lead"), true); let entered = entered_rx.recv_timeout(Duration::from_secs(5)).is_ok(); let root_after_reconcile = root_initial.as_deref().and_then(|prior| { std::fs::write(&root_goal, "transition during steady reconcile\n").unwrap(); @@ -4720,8 +4849,7 @@ mod tests { std::fs::write(&lead_goal, "lead transition during steady reconcile\n").unwrap(); let lead_own = wait_for_resync_event_for_key(&lead_dir, "goal"); - let worker_from_lead = - wait_for_resync_event_for_key(&worker_dir, "goal@hetz.lead"); + let worker_from_lead = wait_for_resync_event_for_key(&worker_dir, "goal@hetz.lead"); let _ = release_tx.send(()); observer_stop.store(true, Ordering::SeqCst); @@ -4752,7 +4880,10 @@ mod tests { observer.join().unwrap() }); - assert!(evidence.0, "the steady-state reconcile must reach its later task"); + assert!( + evidence.0, + "the steady-state reconcile must reach its later task" + ); assert!(evidence.1.is_some(), "root must receive its own transition"); assert!( evidence.2.is_some() && evidence.3.is_some(), @@ -4823,8 +4954,7 @@ mod tests { std::thread::sleep(Duration::from_millis(300)); std::fs::write(&root_goal, "root transition after full refresh\n").unwrap(); let root_event = wait_for_resync_event_for_key(&root_dir, "goal"); - let child_event = - wait_for_resync_event_for_key(&child_dir, "goal@hetz.root"); + let child_event = wait_for_resync_event_for_key(&child_dir, "goal@hetz.root"); let middle_event = current_resync_event_for_key(&middle_dir, "goal@hetz.root"); observer_stop.store(true, Ordering::SeqCst); (root_event, child_event, middle_event) @@ -4972,12 +5102,18 @@ mod tests { ); assert!(corrected.launched.is_empty(), "{corrected:#?}"); assert!( - corrected.adopted.iter().any(|identity| identity == "broken"), + corrected + .adopted + .iter() + .any(|identity| identity == "broken"), "the corrected already-live seat should be adopted: {corrected:#?}" ); let corrected_event = wait_for_resync_event_change(&live_dir, &first_event) .expect("correcting another declaration must not reseed and hide the live transition"); - assert!(corrected_event.contains(r#""binding":"goal""#), "{corrected_event}"); + assert!( + corrected_event.contains(r#""binding":"goal""#), + "{corrected_event}" + ); } #[test] @@ -5119,17 +5255,13 @@ mod tests { fn notify_chain_launch_boundary_installs_ancestors_before_a_later_task_finishes() { let catalog = tempfile::tempdir().unwrap(); write_notify_chain_profile(catalog.path()); - let (_root_dir, root_goal) = - write_notify_chain_agent(catalog.path(), "root", None, false); + let (_root_dir, root_goal) = write_notify_chain_agent(catalog.path(), "root", None, false); write_notify_chain_agent(catalog.path(), "lead", Some("hetz.root"), false); let (worker_dir, _worker_goal) = write_notify_chain_agent(catalog.path(), "worker", Some("hetz.lead"), false); crate::event::publish_owner_binding_for_test(catalog.path(), "hetz").unwrap(); let specs = crate::discover_strict(catalog.path()).specs; - let worker = specs - .iter() - .find(|spec| spec.identity == "worker") - .unwrap(); + let worker = specs.iter().find(|spec| spec.identity == "worker").unwrap(); let mut later = target("hetz.worker.later", "later"); later.name = "later".into(); later.derived = true; @@ -5160,16 +5292,12 @@ mod tests { let observer = scope.spawn(move || { entered_rx.recv().unwrap(); std::fs::write(&root_goal, "changed while later task launches\n").unwrap(); - let event = - wait_for_resync_event_for_key(&worker_dir, "goal@hetz.root"); + let event = wait_for_resync_event_for_key(&worker_dir, "goal@hetz.root"); release_tx.send(()).unwrap(); event }); let report = execute_resync_plan(&plan, &runner, &specs, &resync); - assert_eq!( - report.launched, - ["hetz.worker.agent", "hetz.worker.later"] - ); + assert_eq!(report.launched, ["hetz.worker.agent", "hetz.worker.later"]); observer.join().unwrap() }) .expect("the fresh worker must receive its ancestor transition before full refresh"); @@ -5220,10 +5348,7 @@ mod tests { release_tx.send(()).unwrap(); }); let report = execute_resync_plan(&plan, &runner, &specs, &resync); - assert_eq!( - report.launched, - ["hetz.first.agent", "hetz.second.agent"] - ); + assert_eq!(report.launched, ["hetz.first.agent", "hetz.second.agent"]); }); let event = wait_for_resync_event(&first_dir) @@ -6789,11 +6914,7 @@ while [ ! -e "$0.release" ]; do sleep 0.01; done bin: fake.display().to_string(), catalog_root: tmp.path().join("catalog"), on_command_spawn: Some(std::sync::Arc::new(move |pid| { - release_ready_fixture( - pid, - &observed_fake, - "fake PTY inventory was not published", - ); + release_ready_fixture(pid, &observed_fake, "fake PTY inventory was not published"); })), } .task_observations_at_root(&HashSet::from(["h.worker"]), &root); @@ -6895,11 +7016,7 @@ esac std::fs::read_to_string(invocations).unwrap(), "list --json\nstats --json\nlist --json\nstats --json\n" ); - for (before, after) in unavailable - .observations - .iter() - .zip(&available.observations) - { + for (before, after) in unavailable.observations.iter().zip(&available.observations) { let (ObservedState::Running(before), ObservedState::Running(after)) = (&before.state, &after.state) else { @@ -6958,11 +7075,7 @@ while [ ! -e "$0.release" ]; do sleep 0.01; done bin: fake.display().to_string(), catalog_root: catalog, on_command_spawn: Some(std::sync::Arc::new(move |pid| { - release_ready_fixture( - pid, - &observed_fake, - "fake PTY inventory was not published", - ); + release_ready_fixture(pid, &observed_fake, "fake PTY inventory was not published"); })), }; let desired = HashSet::from(["h.live", "h.exit", "h.gone"]); @@ -7035,11 +7148,7 @@ while [ ! -e "$0.release" ]; do sleep 0.01; done bin: fake.display().to_string(), catalog_root: catalog, on_command_spawn: Some(std::sync::Arc::new(move |pid| { - release_ready_fixture( - pid, - &observed_fake, - "fake PTY inventory was not published", - ); + release_ready_fixture(pid, &observed_fake, "fake PTY inventory was not published"); })), } .task_observations(&HashSet::from(["h.live"]));