clawmates/crates/cm-api/src/agent_lifecycle.rs

//! 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,
    /// Soft-deleted by an operator. The `agents` row and its history survive.
    Deleted,
}

impl AgentState {
    pub fn as_str(self) -> &'static str {
        match self {
            AgentState::Owned => "owned",
            AgentState::Active => "active",
            AgentState::Completed => "completed",
            AgentState::Orphaned => "orphaned",
            AgentState::Deleted => "deleted",
        }
    }
    /// `owned` and `active` are NEVER collected, and that is the whole safety
    /// property of this module.
    pub fn reapable(self) -> bool {
        matches!(
            self,
            AgentState::Completed | AgentState::Orphaned | AgentState::Deleted
        )
    }
}

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.
///
/// Soft-deleted rows are INCLUDED, classified `deleted`, and collected: a soft
/// delete marks the row and leaves it, so "remove" never became permanent and
/// re-deleting did nothing. Purging takes `usage_events` with it — accepted
/// deliberately, since the alternative is rows that outlive the decision to
/// delete them.
const CENSUS_SQL: &str = r#"
SELECT a.id,
       a.name,
       CASE
         -- First, so a soft-deleted agent is never mistaken for live staff:
         -- these rows have no template link either, and would otherwise read
         -- as 'owned' and be kept forever.
         WHEN a.deleted_at IS NOT NULL THEN 'deleted'
         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
 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,
                "deleted" => AgentState::Deleted,
                _ => 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,
        // No grace: a human already decided. The soft delete IS the decision,
        // and these rows have sat for months waiting for something to honour it.
        AgentState::Deleted => true,
        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));
    }

    /// A soft delete is a decision that was never honoured — the row stayed,
    /// the agent kept appearing, and deleting it again did nothing. Collect it
    /// without a grace window: the human already waited.
    #[test]
    fn soft_deleted_agents_are_purged_without_a_grace_window() {
        assert!(should_reap(&c(AgentState::Deleted, None), 24));
        assert!(should_reap(&c(AgentState::Deleted, Some(0.0)), 24));
    }

    /// The safety property restated against the new state: `deleted` must not
    /// widen into anything that can take live staff with it.
    #[test]
    fn adding_deleted_did_not_make_owned_reapable() {
        assert!(!AgentState::Owned.reapable());
        assert!(!AgentState::Active.reapable());
        assert!(AgentState::Deleted.reapable());
    }

    #[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));
    }
}