cm-api: quota — enforce max_active_runs on run enqueue

Adds a Quota.max_active_runs ceiling (queued + running topology runs at
once, per workspace) to stop a single workspace flooding the shared queue.
Free tier: 10, Pro: 25, Team: 100. Enforced at every /run enqueue site:
run_org, run_company, run_team, and the webhook trigger. Webhooks return
429 rather than 402 so external callers can back off — the guard is what
stops a leaked webhook token from being weaponized into a queue flood.

A single team run also spawns a tier-tree of children, so the practical
cap grows with the topology — this counts the outer runs, not every step.

crates/cm-api/src/quota.rs

@@ -13,6 +13,10 @@ use crate::{ApiError, AppState, Authed};
 pub struct Quota {
     pub max_agents: i64,
     pub max_live_containers: i64,
+    /// Ceiling on `queued` + `running` topology runs at once. Prevents one
+    /// workspace flooding the shared queue (a single team run also spawns a
+    /// tier-tree of children, so the practical cap grows with the topology).
+    pub max_active_runs: i64,
 }
 
 /// Per-plan limits. Unknown plans fall back to the free tier.
@@ -21,14 +25,17 @@ pub fn plan_quota(plan: &str) -> Quota {
         "team" => Quota {
             max_agents: 50,
             max_live_containers: 50,
+            max_active_runs: 100,
         },
         "pro" => Quota {
             max_agents: 20,
             max_live_containers: 20,
+            max_active_runs: 25,
         },
         _ => Quota {
             max_agents: 3,
             max_live_containers: 3,
+            max_active_runs: 5,
         },
     }
 }
@@ -61,6 +68,23 @@ pub async fn enforce_new_agent(
     Ok(())
 }
 
+/// Reject enqueueing another topology run if the workspace is at its plan cap.
+pub async fn enforce_new_run(
+    state: &AppState,
+    workspace_id: WorkspaceId,
+) -> Result<(), ApiError> {
+    let plan = plan_of(state, workspace_id).await?;
+    let quota = plan_quota(&plan);
+    let used = cm_db::repo::topology_runs::count_active(&state.pool, workspace_id).await?;
+    if used >= quota.max_active_runs {
+        return Err(ApiError::Quota(format!(
+            "active-run limit reached ({} on the {plan} plan) — wait for a run to finish or upgrade",
+            quota.max_active_runs
+        )));
+    }
+    Ok(())
+}
+
 /// Reject spinning up another container if the workspace is at its plan cap.
 pub async fn enforce_new_container(
     state: &AppState,
@@ -86,6 +110,8 @@ pub struct QuotaUsage {
     max_agents: i64,
     containers_used: i64,
     max_live_containers: i64,
+    active_runs: i64,
+    max_active_runs: i64,
 }
 
 /// `GET /api/quota` — the caller's workspace usage + limits (for the UI).
@@ -98,11 +124,15 @@ pub async fn get_quota(
     let agents_used = cm_db::repo::agents::count_active(&state.pool, user.workspace_id).await?;
     let containers_used =
         cm_db::repo::agent_containers::count_for_workspace(&state.pool, user.workspace_id).await?;
+    let active_runs =
+        cm_db::repo::topology_runs::count_active(&state.pool, user.workspace_id).await?;
     Ok(Json(QuotaUsage {
         plan,
         agents_used,
         max_agents: quota.max_agents,
         containers_used,
         max_live_containers: quota.max_live_containers,
+        active_runs,
+        max_active_runs: quota.max_active_runs,
     }))
 }