fix(agent): sweep orphaned agent runs on startup

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,

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 {

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;

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);
+    }
+}