feat(agents): classify agents by lifecycle and reap the finished and orphaned

A mission mints a crew, and the only thing that reaped one was DELETING the
mission. A mission that merely completed left its agents in the roster forever,
and a crew whose reap was skipped or failed left agents bound to nothing —
indistinguishable in the UI from the operator's own staff.

Four states, from one query:

  owned      no agent_template_link row   → hand-created. NEVER reaped.
  active     on a running/draft mission   → working right now. Kept.
  completed  every mission terminal       → reaped after a 24h grace.
  orphaned   minted, bound to nothing     → reaped.

The discriminator is `agent_template_link`, which mission_orchestrator writes
per minted claw. This matters more than it looks: verified on live data, a
hand-created agent and an orphaned crew member both have ZERO team links and are
structurally identical by binding alone. Judging orphanhood by "no team" would
delete the user's workforce. Provenance is the only honest signal.

The grace window exists because the results view, the World's 24h replay and
"who did this work?" all read the crew AFTER the run ends; reaping on the
terminal transition deletes the answer exactly when the question gets asked. A
completed crew with no usable timestamp is KEPT — a missing date must never read
as "old enough to delete".

Also fixes the delete summary, which reported how many claws were FOUND rather
than purged: "reaped 4 claw(s)" was printed by a delete that purged none, which
is precisely the log you would read while wondering why the agents are still
there. It now reports purged / kept / FAILED, and failed > 0 is the orphan case.

Verified against live data — all four states observed, including the two that
look alike.

Co-Authored-By: Claude Opus 5 <[email protected]>

crates/bins/clawmates-server/src/main.rs

@@ -401,6 +401,16 @@ async fn run() -> Result<(), String> {
     // row has never deleted a directory — which is why the gateway, the smallest
     // disk in the fleet, accumulates mission trees that nothing reclaims.
     cm_api::mission_gc::spawn(pool.clone(), std::time::Duration::from_secs(3600));
+    // Agent lifecycle: reap crews whose missions finished (after a 24h grace so
+    // the results view can still show who did the work) and crews left bound to
+    // nothing. Never touches an agent without an `agent_template_link` row —
+    // that is the operator's own staff, which looks identical to an orphan if
+    // you judge by team membership alone.
+    cm_api::agent_lifecycle::spawn(
+        pool.clone(),
+        runtime.clone(),
+        std::time::Duration::from_secs(3600),
+    );
     // Fleet backstop: a node whose heartbeats stop (without a clean channel
     // close) goes offline within ~28s even if its control channel hangs.
     cm_api::fleet::spawn_node_sweeper(pool.clone(), std::time::Duration::from_secs(8), 20);

crates/cm-api/src/agent_lifecycle.rs

@@ -0,0 +1,267 @@
+//! Which agents are working, which are finished, and which are orphaned.
+//!
+//! A mission mints a crew, and until now the only thing that reaped that crew
+//! was deleting the mission. A mission that merely *completed* left its agents
+//! in the roster forever, and a crew whose reap was skipped or failed left
+//! agents bound to nothing at all — indistinguishable, in the UI, from the
+//! operator's own staff.
+//!
+//! The discriminator is `agent_template_link`. `mission_orchestrator` writes one
+//! row per claw it mints, recording the template and role slot it was minted
+//! for. An agent WITHOUT that row was created by a human (or the planner) and is
+//! part of the workforce: it is never touched here, whatever it is bound to.
+//! Verified against live data — the two hand-created agents on this deployment
+//! have no link row and no team membership, while every mission crew member has
+//! both.
+//!
+//! ```text
+//!   owned      no template link            → the operator's own agent. KEEP.
+//!   active     on a running/draft mission  → doing work right now. KEEP.
+//!   completed  every mission terminal      → reapable once past the grace window.
+//!   orphaned   minted, bound to nothing    → reap.
+//! ```
+//!
+//! `completed` waits out a grace window rather than reaping the moment a mission
+//! finishes: the results view, the World's 24h replay and "who did this work?"
+//! all read the crew AFTER the run ends. Reaping on the terminal transition
+//! would delete the answer at the moment the question gets asked.
+
+use std::time::Duration;
+
+use sqlx::{PgPool, Row};
+use uuid::Uuid;
+
+/// How long a finished crew is kept before it is reaped. Matches the World's
+/// 24h window for finished missions, so nothing the UI can still show is
+/// collected out from under it.
+pub const COMPLETED_GRACE_HOURS: i64 = 24;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum AgentState {
+    Owned,
+    Active,
+    Completed,
+    Orphaned,
+}
+
+impl AgentState {
+    pub fn as_str(self) -> &'static str {
+        match self {
+            AgentState::Owned => "owned",
+            AgentState::Active => "active",
+            AgentState::Completed => "completed",
+            AgentState::Orphaned => "orphaned",
+        }
+    }
+    /// Only these two are ever collected. `owned` and `active` are never
+    /// touched, and that is the whole safety property of this module.
+    pub fn reapable(self) -> bool {
+        matches!(self, AgentState::Completed | AgentState::Orphaned)
+    }
+}
+
+pub struct Classified {
+    pub id: Uuid,
+    pub name: String,
+    pub state: AgentState,
+    /// When the newest mission this agent served reached a terminal state.
+    /// `None` for owned/active/orphaned.
+    pub finished_hours_ago: Option<f64>,
+}
+
+/// The classification, as one query.
+///
+/// `deleted_at IS NULL` throughout: a soft-deleted agent is already gone as far
+/// as every surface is concerned, and re-reaping it would double-count.
+const CENSUS_SQL: &str = r#"
+SELECT a.id,
+       a.name,
+       CASE
+         WHEN atl.agent_id IS NULL THEN 'owned'
+         WHEN EXISTS (
+           SELECT 1 FROM team_members tm
+             JOIN mission_teams mt ON mt.team_id = tm.team_id
+             JOIN missions m       ON m.id       = mt.mission_id
+            WHERE tm.claw_id = a.id AND m.status IN ('running', 'draft')
+         ) THEN 'active'
+         WHEN EXISTS (
+           SELECT 1 FROM team_members tm
+             JOIN mission_teams mt ON mt.team_id = tm.team_id
+            WHERE tm.claw_id = a.id
+         ) THEN 'completed'
+         ELSE 'orphaned'
+       END AS state,
+       (SELECT EXTRACT(EPOCH FROM (now() - MAX(COALESCE(m.completed_at, m.updated_at)))) / 3600.0
+          FROM team_members tm
+          JOIN mission_teams mt ON mt.team_id = tm.team_id
+          JOIN missions m       ON m.id       = mt.mission_id
+         WHERE tm.claw_id = a.id) AS finished_hours_ago
+  FROM agents a
+  LEFT JOIN agent_template_link atl ON atl.agent_id = a.id
+ WHERE a.workspace_id = $1
+   AND a.deleted_at IS NULL
+ ORDER BY a.created_at, a.id
+"#;
+
+pub async fn census(pool: &PgPool, workspace_id: Uuid) -> Result<Vec<Classified>, String> {
+    let rows = sqlx::query(CENSUS_SQL)
+        .bind(workspace_id)
+        .fetch_all(pool)
+        .await
+        .map_err(|e| format!("agent census: {e}"))?;
+    Ok(rows
+        .into_iter()
+        .map(|r| {
+            let state = match r.get::<String, _>("state").as_str() {
+                "owned" => AgentState::Owned,
+                "active" => AgentState::Active,
+                "completed" => AgentState::Completed,
+                _ => AgentState::Orphaned,
+            };
+            Classified {
+                id: r.get("id"),
+                name: r.get("name"),
+                state,
+                finished_hours_ago: r.get::<Option<f64>, _>("finished_hours_ago"),
+            }
+        })
+        .collect())
+}
+
+/// What one sweep did.
+#[derive(Debug, Default, PartialEq, Eq)]
+pub struct Swept {
+    pub reaped: usize,
+    pub failed: usize,
+    pub kept_in_grace: usize,
+}
+
+/// Decide, without touching the database, whether a classified agent should be
+/// collected on this pass. Split out so the policy is testable on its own —
+/// the expensive half is the purge, and the half that can silently delete a
+/// workforce is this one.
+pub fn should_reap(c: &Classified, grace_hours: i64) -> bool {
+    match c.state {
+        AgentState::Owned | AgentState::Active => false,
+        AgentState::Orphaned => true,
+        AgentState::Completed => c
+            .finished_hours_ago
+            // No timestamp means we cannot prove the grace has elapsed, so keep
+            // it. A missing date must never read as "old enough to delete".
+            .is_some_and(|h| h >= grace_hours as f64),
+    }
+}
+
+/// Reap finished and orphaned crews across every workspace.
+pub async fn sweep(
+    pool: &PgPool,
+    runtime: &cm_runtime::Runtime,
+    grace_hours: i64,
+) -> Result<Swept, String> {
+    let workspaces: Vec<Uuid> = sqlx::query_scalar("SELECT id FROM workspaces")
+        .fetch_all(pool)
+        .await
+        .map_err(|e| format!("list workspaces: {e}"))?;
+
+    let provisioner = crate::runtime_provision::RuntimeProvisioner::from_env();
+    let mut out = Swept::default();
+    for ws in workspaces {
+        for c in census(pool, ws).await? {
+            if !c.state.reapable() {
+                continue;
+            }
+            if !should_reap(&c, grace_hours) {
+                out.kept_in_grace += 1;
+                continue;
+            }
+            let report = crate::routes::claws::purge_agent(
+                pool,
+                runtime,
+                provisioner.as_ref(),
+                cm_domain::AgentId::from(c.id),
+            )
+            .await;
+            match report.counts {
+                Ok(_) => {
+                    out.reaped += 1;
+                    eprintln!(
+                        "agent_lifecycle: reaped {} claw {} ({})",
+                        c.state.as_str(),
+                        c.name,
+                        c.id
+                    );
+                }
+                Err(e) => {
+                    out.failed += 1;
+                    eprintln!("agent_lifecycle: purge {} failed (continuing): {e}", c.id);
+                }
+            }
+        }
+    }
+    Ok(out)
+}
+
+/// Spawn the sweeper.
+pub fn spawn(pool: PgPool, runtime: cm_runtime::Runtime, interval: Duration) {
+    tokio::spawn(async move {
+        let mut tick = tokio::time::interval(interval);
+        // The first tick fires immediately; skip it so a restart loop cannot
+        // turn into a reap loop.
+        tick.tick().await;
+        loop {
+            tick.tick().await;
+            match sweep(&pool, &runtime, COMPLETED_GRACE_HOURS).await {
+                Ok(s) if s.reaped > 0 || s.failed > 0 => eprintln!(
+                    "agent_lifecycle: swept — {} reaped, {} failed, {} still in grace",
+                    s.reaped, s.failed, s.kept_in_grace
+                ),
+                Ok(_) => {}
+                Err(e) => eprintln!("agent_lifecycle: sweep failed: {e}"),
+            }
+        }
+    });
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn c(state: AgentState, hours: Option<f64>) -> Classified {
+        Classified {
+            id: Uuid::now_v7(),
+            name: "x".into(),
+            state,
+            finished_hours_ago: hours,
+        }
+    }
+
+    /// The property that matters most: this sweeper must never be able to
+    /// delete the operator's own staff, no matter what it is bound to.
+    #[test]
+    fn owned_and_active_are_never_reaped() {
+        for hours in [None, Some(0.0), Some(1_000_000.0)] {
+            assert!(!should_reap(&c(AgentState::Owned, hours), 24));
+            assert!(!should_reap(&c(AgentState::Active, hours), 24));
+        }
+    }
+
+    #[test]
+    fn orphans_go_immediately() {
+        assert!(should_reap(&c(AgentState::Orphaned, None), 24));
+    }
+
+    #[test]
+    fn a_finished_crew_waits_out_the_grace_window() {
+        assert!(!should_reap(&c(AgentState::Completed, Some(1.0)), 24));
+        assert!(!should_reap(&c(AgentState::Completed, Some(23.9)), 24));
+        assert!(should_reap(&c(AgentState::Completed, Some(24.0)), 24));
+    }
+
+    /// A completed crew with no usable timestamp must be KEPT. Treating a
+    /// missing date as "old" is how a sweeper deletes something it was never
+    /// able to prove was finished.
+    #[test]
+    fn a_missing_finish_time_is_not_treated_as_old() {
+        assert!(!should_reap(&c(AgentState::Completed, None), 24));
+    }
+}

crates/cm-api/src/lib.rs

@@ -1,6 +1,7 @@
 //! REST API for Clawmates (spec §13). One route resource per module.
 
 pub mod benchmark_runner;
+pub mod agent_lifecycle;
 pub mod beszel;
 pub mod brain_seed;
 pub mod cleanup_sweeper;

crates/cm-api/src/routes/missions.rs

@@ -414,7 +414,9 @@ pub async fn artifact_download(
         [
             (
                 axum::http::header::CONTENT_TYPE,
-                artifact.mime.unwrap_or_else(|| "application/octet-stream".into()),
+                artifact
+                    .mime
+                    .unwrap_or_else(|| "application/octet-stream".into()),
             ),
             (
                 axum::http::header::CONTENT_DISPOSITION,
@@ -898,6 +900,14 @@ async fn reap_mission_resources(state: &AppState, mission_id: Uuid) {
     //    all DB rows. Shared with the batch-delete reaper so this path cannot
     //    drift back into skipping the container teardown.
     let provisioner = crate::runtime_provision::RuntimeProvisioner::from_env();
+    // Counted, not assumed. The summary below used to report `claw_ids.len()`,
+    // which is how many claws were FOUND — including every one skipped as still
+    // employed and every one whose purge failed. So "reaped 4 claw(s)" was
+    // printed by a delete that purged none, which is exactly the log you would
+    // read while wondering why the agents are still there.
+    let mut purged = 0usize;
+    let mut kept = 0usize;
+    let mut failed = 0usize;
     for cid in &claw_ids {
         // Only claws this mission is the LAST holder of.
         //
@@ -921,6 +931,7 @@ async fn reap_mission_resources(state: &AppState, mission_id: Uuid) {
             eprintln!(
                 "missions::delete: keeping claw {cid} — {shared} other mission(s) still employ it"
             );
+            kept += 1;
             continue;
         }
 
@@ -931,10 +942,14 @@ async fn reap_mission_resources(state: &AppState, mission_id: Uuid) {
             cm_domain::AgentId::from(*cid),
         )
         .await;
-        if let Err(e) = report.counts {
+        match report.counts {
+            Ok(_) => purged += 1,
+            Err(e) => {
+                failed += 1;
                 eprintln!("missions::delete: hard_purge claw {cid} failed (continuing): {e}");
             }
         }
+    }
 
     // 3. Delete the (permanent-lifecycle) teams — no mission FK cascades them.
     //    team_members cascades from teams.
@@ -969,10 +984,14 @@ async fn reap_mission_resources(state: &AppState, mission_id: Uuid) {
         }
     }
 
+    // Say what actually happened. `failed > 0` means the mission row is about to
+    // be deleted while its claws survive with nothing left pointing at them —
+    // the orphan case, and the only way to notice it after the fact.
     eprintln!(
-        "missions::delete: reaped {} claw(s), {} team(s) for mission {mission_id}",
+        "missions::delete: mission {mission_id}: {purged} claw(s) purged, {kept} kept (still \
-        claw_ids.len(),
+         employed), {failed} FAILED, {} team(s) deleted, {} claw(s) considered",
-        team_ids.len()
+        team_ids.len(),
+        claw_ids.len()
     );
 }
 
@@ -1727,7 +1746,8 @@ pub async fn workforce(
     .await?;
 
     let mut missions: Vec<Value> = Vec::new();
-    let mut seen_mission: std::collections::HashMap<String, usize> = std::collections::HashMap::new();
+    let mut seen_mission: std::collections::HashMap<String, usize> =
+        std::collections::HashMap::new();
     for r in rows {
         let mid: String = r.get("mission_id");
         let idx = match seen_mission.get(&mid) {
@@ -1755,7 +1775,9 @@ pub async fn workforce(
             "status": r.get::<String, _>("agent_status"),
         });
         // A claw bound to two NODES of the same mission is still one colleague.
-        let list = missions[idx]["agents"].as_array_mut().expect("agents array");
+        let list = missions[idx]["agents"]
+            .as_array_mut()
+            .expect("agents array");
         let id = agent["id"].clone();
         if !list.iter().any(|a| a["id"] == id) {
             list.push(agent);
@@ -1863,7 +1885,8 @@ mod artifact_tests {
             "one resolver"
         );
         assert_eq!(
-            src.matches(concat!("resolve_", "artifact_path(&artifact.path)")).count(),
+            src.matches(concat!("resolve_", "artifact_path(&artifact.path)"))
+                .count(),
             2,
             "and both routes must go through it"
         );