diff --git a/src/core/runtime/builder.rs b/src/core/runtime/builder.rs index 312af93c5f..04f88548bc 100644 --- a/src/core/runtime/builder.rs +++ b/src/core/runtime/builder.rs @@ -506,6 +506,20 @@ impl CoreBuilder { let (ctx, has_operator_token, config) = CoreContext::init(self.host_kind, &self.token, self.domains).await?; + // Reap agent runs orphaned by a previous process (crash / restart / + // deploy). Here, and not with the other boot-once jobs, because those + // run from `serve()`: an embedder that only calls `build()` and then + // `invoke()` never reaches them, and `openhuman.agent_runs_active` is + // dispatchable the moment this returns. The core is a single in-process + // runtime, so a run left Pending/Running/Interrupted in the durable + // status store has no executor to advance it and would be listed as + // active forever. Best-effort — a store that cannot be read logs and + // reaps nothing rather than failing the build. + if let Some(cfg) = config.as_ref() { + crate::openhuman::agent::tinyagents::reaper::reap_orphaned_runs(&cfg.workspace_dir) + .await; + } + Ok(CoreRuntime { ctx, config, diff --git a/src/core/runtime/services.rs b/src/core/runtime/services.rs index 829758df3b..b181b211a9 100644 --- a/src/core/runtime/services.rs +++ b/src/core/runtime/services.rs @@ -335,6 +335,10 @@ pub fn start_bootstrap_jobs(services: ServiceSet, config: &Config) { pub async fn start_boot_once_jobs(services: ServiceSet, config: &Config) { run_legacy_migrations(config).await; + // The orphaned-run sweep does NOT live here. It runs in + // `CoreBuilder::build`, which every runtime goes through — these jobs only + // run from `serve()`, so a build-only embedder would never be swept. + if services.harness_init { let cfg_for_init = config.clone(); tokio::spawn(async move { diff --git a/src/openhuman/agent/tinyagents/mod.rs b/src/openhuman/agent/tinyagents/mod.rs index 0066c7c1d5..cbfa50bdc1 100644 --- a/src/openhuman/agent/tinyagents/mod.rs +++ b/src/openhuman/agent/tinyagents/mod.rs @@ -30,6 +30,7 @@ pub(crate) mod observability; pub(crate) mod orchestration; pub(crate) mod payload_summarizer; mod policy_denial; +pub(crate) mod reaper; pub(crate) mod replay; pub mod resolved_route; pub(crate) mod retriever; diff --git a/src/openhuman/agent/tinyagents/reaper.rs b/src/openhuman/agent/tinyagents/reaper.rs new file mode 100644 index 0000000000..ea97170c4e --- /dev/null +++ b/src/openhuman/agent/tinyagents/reaper.rs @@ -0,0 +1,279 @@ +//! Startup reconciliation for orphaned agent runs. +//! +//! The core is a single in-process runtime. When it exits — a crash, a restart, +//! or a deploy — any run still in a non-terminal state (`Pending` / `Running` / +//! `Interrupted`) in the durable status store is *orphaned*: no executor will +//! ever advance it, yet it lingers in the "active runs" listing +//! (`openhuman.agent_runs_active`) indefinitely. A long-lived instance was +//! observed carrying 50 such zombies, all invisible to any cancel path because +//! the task that owned each run no longer exists. +//! +//! On startup we sweep every still-active run to the terminal `Cancelled` state +//! with an explanatory error so the active listing reflects reality. This is +//! the one *writer* over the status seam that +//! [`crate::openhuman::agent::tinyagents::journal`] exposes; the replay/status +//! controllers ([`super::replay`]) stay strictly read-only. + +use std::path::Path; +use std::time::SystemTime; + +use tinyagents::harness::events::HarnessRunStatus; +use tinyagents::harness::ids::{ExecutionStatus, HarnessPhase}; +use tinyagents::harness::observability::HarnessStatusStore; + +use crate::openhuman::agent::session_import::ops::open_session_stores; +use crate::openhuman::agent::tinyagents::journal::FileStatusStore; + +/// Error recorded on a run reaped by the startup sweep. Stable + grep-friendly +/// so an operator (or a test) can tell a reaped run from a genuinely failed one. +pub(crate) const ORPHAN_REAP_REASON: &str = + "run orphaned: core restarted while the run was in flight"; + +/// Reap every run left non-terminal by a previous process. +/// +/// Opens the durable status store under `workspace`, lists the still-active +/// runs, and moves each to the terminal `Cancelled` state. Best-effort: a +/// per-run persistence failure is logged and does not abort the sweep, and a +/// failure to open/list the store logs and yields `0` rather than blocking +/// boot. Returns the number of runs reaped. +pub(crate) async fn reap_orphaned_runs(workspace: &Path) -> usize { + // One id for the whole sweep so every line below can be tied to the same + // boot — a restart loop otherwise interleaves two sweeps' lines with nothing + // to tell them apart. Derived from the clock rather than a uuid so it stays + // ordered in a log tail. + let sweep_id = SystemTime::now() + .duration_since(SystemTime::UNIX_EPOCH) + .map(|d| d.as_millis()) + .unwrap_or(0); + log::debug!( + "[agent] startup run sweep entry sweep_id={sweep_id} workspace={}", + workspace.display() + ); + + let stores = open_session_stores(workspace); + let store = FileStatusStore::new(stores.kv); + + let active = match store.list_active().await { + Ok(runs) => runs, + Err(err) => { + log::warn!( + "[agent] startup run sweep exit sweep_id={sweep_id} branch=list-failed: {err}" + ); + return 0; + } + }; + + if active.is_empty() { + log::debug!("[agent] startup run sweep exit sweep_id={sweep_id} branch=none-active"); + return 0; + } + log::debug!( + "[agent] startup run sweep sweep_id={sweep_id} active={} to reap", + active.len() + ); + + let mut reaped = 0usize; + for mut status in active { + let run_id = status.run_id.as_str().to_string(); + // The source state is read before the mutation, not assumed: `list_active` + // returns whatever non-terminal state a run was left in, and logging a + // presumed `Running` would misreport a run that died mid-`Queued`. + let from_status = format!("{:?}", status.status); + mark_orphaned(&mut status); + let to_status = format!("{:?}", status.status); + log::debug!( + "[agent] startup run sweep sweep_id={sweep_id} run_id={run_id} \ + transition={from_status}->{to_status} phase=Done persisting" + ); + match store.put_status(status).await { + Ok(()) => { + reaped += 1; + log::debug!( + "[agent] startup run sweep sweep_id={sweep_id} run_id={run_id} \ + transition={from_status}->{to_status} persisted" + ); + } + Err(err) => { + log::warn!( + "[agent] startup run sweep sweep_id={sweep_id} run_id={run_id} \ + transition={from_status}->{to_status} failed: {err}" + ); + } + } + } + // One operator-visible line for the whole sweep, and only when it did + // something: a clean boot is the normal case and says nothing. + if reaped > 0 { + log::info!("[agent] startup run sweep sweep_id={sweep_id} reaped {reaped} orphaned run(s)"); + } else { + log::debug!("[agent] startup run sweep exit sweep_id={sweep_id} branch=nothing-reaped"); + } + reaped +} + +/// Move a run to the terminal `Cancelled` state with the orphan reason. +/// +/// tinyagents exposes `mark_completed` / `mark_failed` / `mark_interrupted` but +/// no `mark_cancelled`; the `HarnessRunStatus` fields are public, so we set the +/// terminal state directly here rather than take a cross-repo dependency on a +/// new helper. `Cancelled` (not `Failed`) is deliberate: a restart is not a run +/// failure, and marking it `Failed` would render every reaped run as a red +/// error row in the UI. Mirrors the field writes `mark_failed` performs +/// (terminal status + `Done` phase + `error` + `ended_at`/`updated_at`). +fn mark_orphaned(status: &mut HarnessRunStatus) { + status.status = ExecutionStatus::Cancelled; + status.current_phase = HarnessPhase::Done; + status.error = Some(ORPHAN_REAP_REASON.to_string()); + let now = SystemTime::now(); + status.ended_at = Some(now); + status.updated_at = now; +} + +#[cfg(test)] +mod tests { + use super::*; + + use tinyagents::harness::ids::ComponentId; + + use crate::openhuman::agent::tinyagents::journal::mint_run_id; + + /// Build a fresh status in the given non-terminal state and persist it. + async fn seed_status(store: &FileStatusStore, status_kind: ExecutionStatus) -> String { + let run_id = mint_run_id(); + let mut status = + HarnessRunStatus::new(run_id.clone(), ComponentId::new("mock-model".to_string())); + match status_kind { + ExecutionStatus::Pending => { /* fresh status is already Pending */ } + ExecutionStatus::Running => status.mark_running(HarnessPhase::Model), + ExecutionStatus::Interrupted => status.mark_interrupted(), + other => panic!("seed_status only seeds non-terminal states, got {other:?}"), + } + store.put_status(status).await.unwrap(); + run_id.as_str().to_string() + } + + /// The sweep reaps every non-terminal run to `Cancelled` with the reason, + /// leaves terminal runs untouched, and empties the active listing. + #[tokio::test] + async fn reap_cancels_every_active_run_and_spares_terminal_ones() { + let tmp = std::env::temp_dir().join(format!("oh-reaper-{}", uuid::Uuid::new_v4())); + let store = FileStatusStore::new(open_session_stores(&tmp).kv); + + let pending = seed_status(&store, ExecutionStatus::Pending).await; + let running = seed_status(&store, ExecutionStatus::Running).await; + let interrupted = seed_status(&store, ExecutionStatus::Interrupted).await; + + // A run that already finished must survive the sweep unchanged. + let done = mint_run_id(); + let mut done_status = + HarnessRunStatus::new(done.clone(), ComponentId::new("mock-model".to_string())); + done_status.mark_running(HarnessPhase::Model); + done_status.mark_completed(); + store.put_status(done_status).await.unwrap(); + + let reaped = reap_orphaned_runs(&tmp).await; + assert_eq!(reaped, 3, "the three non-terminal runs were reaped"); + + // Every orphan is now terminal-cancelled with the stable reason. + for run_id in [&pending, &running, &interrupted] { + let status = store + .get_status(run_id) + .await + .unwrap() + .expect("status present"); + assert_eq!(status.status, ExecutionStatus::Cancelled); + assert_eq!(status.current_phase, HarnessPhase::Done); + assert_eq!(status.error.as_deref(), Some(ORPHAN_REAP_REASON)); + assert!(status.ended_at.is_some(), "reaped run has an end time"); + } + + // The completed run is left exactly as it was. + let done_after = store + .get_status(done.as_str()) + .await + .unwrap() + .expect("done present"); + assert_eq!(done_after.status, ExecutionStatus::Completed); + assert!(done_after.error.is_none()); + + // The active listing is now empty — a second sweep is a no-op. + assert!(store.list_active().await.unwrap().is_empty()); + assert_eq!( + reap_orphaned_runs(&tmp).await, + 0, + "idempotent: nothing left to reap" + ); + + let _ = std::fs::remove_dir_all(&tmp); + } + + /// A workspace that never hosted a run reaps nothing and does not error. + #[tokio::test] + async fn reap_on_empty_workspace_is_a_noop() { + let tmp = std::env::temp_dir().join(format!("oh-reaper-empty-{}", uuid::Uuid::new_v4())); + assert_eq!(reap_orphaned_runs(&tmp).await, 0); + let _ = std::fs::remove_dir_all(&tmp); + } + + /// The sweep has to happen on the path a **build-only** embedder takes. + /// + /// `CoreRuntime::invoke` dispatches `openhuman.agent_runs_active` the moment + /// `build()` returns, with no transport and no background services in + /// between, so a sweep that lived with the other boot-once jobs (which run + /// from `serve()`) would leave that caller reading the previous process's + /// graveyard. This pins the sweep to the build path with every optional + /// service off. + #[tokio::test] + async fn a_build_only_runtime_is_swept_before_it_can_be_invoked() { + use crate::core::runtime::{CoreBuilder, DomainSet, ServiceSet}; + use crate::core::types::HostKind; + + let tmp = std::env::temp_dir().join(format!("oh-reaper-boot-{}", uuid::Uuid::new_v4())); + // `OPENHUMAN_WORKSPACE` names the root; the resolved workspace is the + // `workspace` directory under it, which is what the sweep will read. + let workspace = tmp.join("workspace"); + let store = FileStatusStore::new(open_session_stores(&workspace).kv); + let orphan = seed_status(&store, ExecutionStatus::Running).await; + assert_eq!(store.list_active().await.unwrap().len(), 1); + + // Point the runtime at this workspace, then build it with no transport + // and no services — the shape `examples/embed_headless.rs` documents. + // `OPENHUMAN_WORKSPACE` is process-global; take the same lock the other + // env-mutating tests do so a parallel test cannot read ours. + let _env = crate::openhuman::config::TEST_ENV_LOCK + .lock() + .unwrap_or_else(|e| e.into_inner()); + let previous = std::env::var("OPENHUMAN_WORKSPACE").ok(); + std::env::set_var("OPENHUMAN_WORKSPACE", &tmp); + let built = CoreBuilder::new(HostKind::Cli) + .services(ServiceSet::none()) + .domains(DomainSet::harness()) + .build() + .await; + match previous { + Some(value) => std::env::set_var("OPENHUMAN_WORKSPACE", value), + None => std::env::remove_var("OPENHUMAN_WORKSPACE"), + } + let built = built.expect("a headless build succeeds"); + assert_eq!( + built.context().workspace_dir().ok(), + Some(workspace.clone()), + "the build must have resolved the workspace this test seeded" + ); + + let after = store + .get_status(&orphan) + .await + .unwrap() + .expect("the seeded run is still readable"); + assert_eq!( + after.status, + ExecutionStatus::Cancelled, + "build() must reap before any RPC can be dispatched" + ); + assert_eq!(after.error.as_deref(), Some(ORPHAN_REAP_REASON)); + assert!(store.list_active().await.unwrap().is_empty()); + + let _ = std::fs::remove_dir_all(&tmp); + } +}