Fleet: actionable executions — rules engine + metrics-aware placement (Phase 2)

Turn the Beszel-tapped metrics into a self-managing loop.

- migration node_rules (workspace/node-scoped: metric op threshold, for_seconds,
  action JSONB, last_fired).
- cm-db: repo/node_rules.rs (CRUD + list_enabled); node_metrics::eval_all merges
  Beszel + heartbeat scalars per node + a headroom() heuristic; nodes::status_of;
  heartbeat now PRESERVES a `draining` status across heartbeats (so a cordon sticks).
- cm-api: node_rules.rs evaluator (spawn_evaluator, 20s) — when a metric condition
  holds for the rule's window it fires drain / undrain / alert (in-memory sustained
  + cooldown tracking, modeled on the node sweeper); routes/beszel.rs rules CRUD
  (GET/POST/PATCH/DELETE /api/fleet/rules); spawned in clawmates-server.
- cm-runtime: placement_node() is metrics-aware — a `draining` node stops receiving
  new agent sandboxes (falls back to local), so the drain rule is actionable.
- frontend: FleetRules section in the Local view — build rules (node · metric · op ·
  threshold · duration → action), toggle/delete, with fired-history.

The loop: hot/overloaded node → rule drains it → placement avoids it → recovers →
undrain rule brings it back. Deployed; node_rules migration applied.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

crates/cm-db/src/repo/mod.rs

@@ -10,6 +10,7 @@ pub mod fleet_beszel;
 pub mod fleet_tailscale;
 pub mod messages;
 pub mod node_metrics;
+pub mod node_rules;
 pub mod nodes;
 pub mod orgs;
 pub mod outbox;

crates/cm-db/src/repo/node_metrics.rs

@@ -2,9 +2,10 @@
 //! per node: scalar columns for the fleet cards + rules engine, plus a JSONB blob
 //! for the per-node monitor page.
 
-use cm_domain::NodeId;
+use cm_domain::{NodeId, WorkspaceId};
 use serde_json::Value;
 use sqlx::{PgPool, Row};
+use uuid::Uuid;
 
 use crate::DbError;
 
@@ -57,6 +58,72 @@ pub async fn upsert(pool: &PgPool, node_id: NodeId, m: &NodeMetrics) -> Result<(
     Ok(())
 }
 
+/// A node's current evaluatable scalars (Beszel-tapped, falling back to the
+/// heartbeat health), for the rules engine + metrics-aware placement.
+#[derive(Debug, Clone)]
+pub struct EvalRow {
+    pub node_id: NodeId,
+    pub workspace_id: WorkspaceId,
+    pub status: String,
+    pub cpu_pct: Option<f64>,
+    pub mem_pct: Option<f64>,
+    pub disk_pct: Option<f64>,
+    pub gpu_pct: Option<f64>,
+    pub temp_max: Option<f64>,
+    pub load1: Option<f64>,
+}
+
+impl EvalRow {
+    /// Look up a metric by rule name.
+    pub fn metric(&self, name: &str) -> Option<f64> {
+        match name {
+            "cpu_pct" => self.cpu_pct,
+            "mem_pct" => self.mem_pct,
+            "disk_pct" => self.disk_pct,
+            "gpu_pct" => self.gpu_pct,
+            "temp_max" => self.temp_max,
+            "load1" => self.load1,
+            _ => None,
+        }
+    }
+    /// Free headroom heuristic (higher = more capacity) for placement ranking.
+    pub fn headroom(&self) -> f64 {
+        let used = self.cpu_pct.unwrap_or(0.0).max(self.mem_pct.unwrap_or(0.0));
+        100.0 - used
+    }
+}
+
+/// Every node's current metric scalars (merged Beszel + heartbeat health).
+pub async fn eval_all(pool: &PgPool) -> Result<Vec<EvalRow>, DbError> {
+    let rows = sqlx::query(
+        "SELECT n.id, n.workspace_id, n.status,
+            COALESCE(m.cpu_pct, h.cpu_pct) AS cpu_pct,
+            COALESCE(m.mem_pct, CASE WHEN h.mem_total > 0 THEN h.mem_used::float8 / h.mem_total * 100 END) AS mem_pct,
+            COALESCE(m.disk_pct, CASE WHEN h.disk_total > 0 THEN (h.disk_total - h.disk_free)::float8 / h.disk_total * 100 END) AS disk_pct,
+            m.gpu_pct, m.temp_max,
+            COALESCE(m.load1, h.load1) AS load1
+         FROM nodes n
+         LEFT JOIN node_health h ON h.node_id = n.id
+         LEFT JOIN node_metrics m ON m.node_id = n.id",
+    )
+    .fetch_all(pool)
+    .await?;
+    Ok(rows
+        .into_iter()
+        .map(|r| EvalRow {
+            node_id: NodeId::from(r.get::<Uuid, _>("id")),
+            workspace_id: WorkspaceId::from(r.get::<Uuid, _>("workspace_id")),
+            status: r.get("status"),
+            cpu_pct: r.get("cpu_pct"),
+            mem_pct: r.get("mem_pct"),
+            disk_pct: r.get("disk_pct"),
+            gpu_pct: r.get("gpu_pct"),
+            temp_max: r.get("temp_max"),
+            load1: r.get("load1"),
+        })
+        .collect())
+}
+
 /// The latest metrics blob for a node (the full snapshot for the monitor page).
 pub async fn latest(pool: &PgPool, node_id: NodeId) -> Result<Option<Value>, DbError> {
     let row = sqlx::query(

crates/cm-db/src/repo/node_rules.rs

@@ -0,0 +1,127 @@
+//! Fleet automation rules: metric thresholds → actions (drain/undrain/alert),
+//! evaluated by the background rules engine.
+
+use cm_domain::{NodeId, WorkspaceId};
+use serde_json::Value;
+use sqlx::{PgPool, Row};
+use time::OffsetDateTime;
+use uuid::Uuid;
+
+use crate::DbError;
+
+#[derive(Debug, Clone)]
+pub struct NodeRule {
+    pub id: Uuid,
+    pub workspace_id: WorkspaceId,
+    pub node_id: Option<NodeId>,
+    pub name: String,
+    pub enabled: bool,
+    pub metric: String,
+    pub op: String,
+    pub threshold: f64,
+    pub for_seconds: i32,
+    pub action: Value,
+    pub last_fired_at: Option<OffsetDateTime>,
+}
+
+fn map_rule(r: sqlx::postgres::PgRow) -> NodeRule {
+    NodeRule {
+        id: r.get("id"),
+        workspace_id: WorkspaceId::from(r.get::<Uuid, _>("workspace_id")),
+        node_id: r.get::<Option<Uuid>, _>("node_id").map(NodeId::from),
+        name: r.get("name"),
+        enabled: r.get("enabled"),
+        metric: r.get("metric"),
+        op: r.get("op"),
+        threshold: r.get("threshold"),
+        for_seconds: r.get("for_seconds"),
+        action: r.get("action"),
+        last_fired_at: r.get("last_fired_at"),
+    }
+}
+
+const COLS: &str =
+    "id, workspace_id, node_id, name, enabled, metric, op, threshold, for_seconds, action, last_fired_at";
+
+#[allow(clippy::too_many_arguments)]
+pub async fn create(
+    pool: &PgPool,
+    workspace_id: WorkspaceId,
+    node_id: Option<NodeId>,
+    name: &str,
+    metric: &str,
+    op: &str,
+    threshold: f64,
+    for_seconds: i32,
+    action: &Value,
+) -> Result<Uuid, DbError> {
+    let id = Uuid::now_v7();
+    sqlx::query(
+        "INSERT INTO node_rules (id, workspace_id, node_id, name, metric, op, threshold, for_seconds, action)
+         VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)",
+    )
+    .bind(id)
+    .bind(workspace_id.as_uuid())
+    .bind(node_id.map(|n| n.as_uuid()))
+    .bind(name)
+    .bind(metric)
+    .bind(op)
+    .bind(threshold)
+    .bind(for_seconds)
+    .bind(action)
+    .execute(pool)
+    .await?;
+    Ok(id)
+}
+
+/// List a workspace's rules (newest last).
+pub async fn list(pool: &PgPool, workspace_id: WorkspaceId) -> Result<Vec<NodeRule>, DbError> {
+    let rows = sqlx::query(&format!(
+        "SELECT {COLS} FROM node_rules WHERE workspace_id = $1 ORDER BY created_at"
+    ))
+    .bind(workspace_id.as_uuid())
+    .fetch_all(pool)
+    .await?;
+    Ok(rows.into_iter().map(map_rule).collect())
+}
+
+/// All enabled rules across workspaces (for the evaluator).
+pub async fn list_enabled(pool: &PgPool) -> Result<Vec<NodeRule>, DbError> {
+    let rows = sqlx::query(&format!("SELECT {COLS} FROM node_rules WHERE enabled"))
+        .fetch_all(pool)
+        .await?;
+    Ok(rows.into_iter().map(map_rule).collect())
+}
+
+pub async fn set_enabled(
+    pool: &PgPool,
+    id: Uuid,
+    workspace_id: WorkspaceId,
+    enabled: bool,
+) -> Result<(), DbError> {
+    sqlx::query("UPDATE node_rules SET enabled = $3 WHERE id = $1 AND workspace_id = $2")
+        .bind(id)
+        .bind(workspace_id.as_uuid())
+        .bind(enabled)
+        .execute(pool)
+        .await?;
+    Ok(())
+}
+
+pub async fn delete(pool: &PgPool, id: Uuid, workspace_id: WorkspaceId) -> Result<(), DbError> {
+    sqlx::query("DELETE FROM node_rules WHERE id = $1 AND workspace_id = $2")
+        .bind(id)
+        .bind(workspace_id.as_uuid())
+        .execute(pool)
+        .await?;
+    Ok(())
+}
+
+/// Stamp a rule as just-fired (debounce + UI).
+pub async fn mark_fired(pool: &PgPool, id: Uuid) -> Result<(), DbError> {
+    sqlx::query("UPDATE node_rules SET last_fired_at = now() WHERE id = $1")
+        .bind(id)
+        .execute(pool)
+        .await?;
+    Ok(())
+}

crates/cm-db/src/repo/nodes.rs

@@ -121,7 +121,10 @@ pub async fn heartbeat(
     h: &NodeHealth,
 ) -> Result<(), DbError> {
     sqlx::query(
-        "UPDATE nodes SET status = 'online', last_seen = now(),
+        // Preserve a rule-/operator-set `draining` state across heartbeats; a node
+        // is only un-drained by an explicit set_status.
+        "UPDATE nodes SET status = CASE WHEN status = 'draining' THEN 'draining' ELSE 'online' END,
+            last_seen = now(),
             agent_version = COALESCE($2, agent_version),
             tailscale_ip = COALESCE($3, tailscale_ip),
             hostname = COALESCE($4, hostname),
@@ -198,6 +201,16 @@ pub async fn delete(pool: &PgPool, id: NodeId, workspace_id: WorkspaceId) -> Res
     Ok(())
 }
 
+/// A node's current status (for metrics-aware placement: skip 'draining').
+pub async fn status_of(pool: &PgPool, id: NodeId) -> Result<Option<String>, DbError> {
+    Ok(
+        sqlx::query_scalar::<_, String>("SELECT status FROM nodes WHERE id = $1")
+            .bind(id.as_uuid())
+            .fetch_optional(pool)
+            .await?,
+    )
+}
+
 fn map_node(r: sqlx::postgres::PgRow) -> NodeRow {
     let health = r
         .get::<Option<uuid::Uuid>, _>("health_node")