feat: auto-pause ecc2 sessions when budgets are exceeded

ecc2/src/session/manager.rs

@@ -353,6 +353,56 @@ pub async fn stop_session(db: &StateStore, id: &str) -> Result<()> {
     stop_session_with_options(db, id, true).await
 }
 
+#[derive(Debug, Clone, Default, Serialize, PartialEq)]
+pub struct BudgetEnforcementOutcome {
+    pub token_budget_exceeded: bool,
+    pub cost_budget_exceeded: bool,
+    pub paused_sessions: Vec<String>,
+}
+
+impl BudgetEnforcementOutcome {
+    pub fn hard_limit_exceeded(&self) -> bool {
+        self.token_budget_exceeded || self.cost_budget_exceeded
+    }
+}
+
+pub fn enforce_budget_hard_limits(
+    db: &StateStore,
+    cfg: &Config,
+) -> Result<BudgetEnforcementOutcome> {
+    let sessions = db.list_sessions()?;
+    let total_tokens = sessions
+        .iter()
+        .map(|session| session.metrics.tokens_used)
+        .sum::<u64>();
+    let total_cost = sessions
+        .iter()
+        .map(|session| session.metrics.cost_usd)
+        .sum::<f64>();
+
+    let mut outcome = BudgetEnforcementOutcome {
+        token_budget_exceeded: cfg.token_budget > 0 && total_tokens >= cfg.token_budget,
+        cost_budget_exceeded: cfg.cost_budget_usd > 0.0 && total_cost >= cfg.cost_budget_usd,
+        paused_sessions: Vec::new(),
+    };
+
+    if !outcome.hard_limit_exceeded() {
+        return Ok(outcome);
+    }
+
+    for session in sessions.into_iter().filter(|session| {
+        matches!(
+            session.state,
+            SessionState::Pending | SessionState::Running | SessionState::Idle
+        )
+    }) {
+        stop_session_recorded(db, &session, false)?;
+        outcome.paused_sessions.push(session.id);
+    }
+
+    Ok(outcome)
+}
+
 pub fn record_tool_call(
     db: &StateStore,
     session_id: &str,
@@ -1175,9 +1225,12 @@ async fn stop_session_with_options(
     cleanup_worktree: bool,
 ) -> Result<()> {
     let session = resolve_session(db, id)?;
+    stop_session_recorded(db, &session, cleanup_worktree)
+}
 
+fn stop_session_recorded(db: &StateStore, session: &Session, cleanup_worktree: bool) -> Result<()> {
     if let Some(pid) = session.pid {
-        kill_process(pid).await?;
+        kill_process(pid)?;
     }
 
     db.update_pid(&session.id, None)?;
@@ -1193,13 +1246,27 @@ async fn stop_session_with_options(
 }
 
 #[cfg(unix)]
-async fn kill_process(pid: u32) -> Result<()> {
+fn kill_process(pid: u32) -> Result<()> {
     send_signal(pid, libc::SIGTERM)?;
-    tokio::time::sleep(std::time::Duration::from_millis(1200)).await;
+    std::thread::sleep(std::time::Duration::from_millis(1200));
     send_signal(pid, libc::SIGKILL)?;
     Ok(())
 }
 
+#[cfg(windows)]
+fn kill_process(pid: u32) -> Result<()> {
+    let status = std::process::Command::new("taskkill")
+        .args(["/PID", &pid.to_string(), "/T", "/F"])
+        .status()
+        .with_context(|| format!("Failed to invoke taskkill for process {pid}"))?;
+
+    if status.success() {
+        Ok(())
+    } else {
+        Err(anyhow::anyhow!("taskkill exited with status {status}"))
+    }
+}
+
 #[cfg(unix)]
 fn send_signal(pid: u32, signal: i32) -> Result<()> {
     let outcome = unsafe { libc::kill(pid as i32, signal) };
@@ -1416,9 +1483,7 @@ impl fmt::Display for SessionStatus {
         writeln!(
             f,
             "Tokens:  {} total (in {} / out {})",
-            s.metrics.tokens_used,
-            s.metrics.input_tokens,
-            s.metrics.output_tokens
+            s.metrics.tokens_used, s.metrics.input_tokens, s.metrics.output_tokens
         )?;
         writeln!(f, "Tools:   {}", s.metrics.tool_calls)?;
         writeln!(f, "Files:   {}", s.metrics.files_changed)?;
@@ -1885,6 +1950,115 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn enforce_budget_hard_limits_stops_active_sessions_without_cleaning_worktrees() -> Result<()> {
+        let tempdir = TestDir::new("manager-budget-pause")?;
+        let mut cfg = build_config(tempdir.path());
+        cfg.token_budget = 100;
+        cfg.cost_budget_usd = 0.0;
+
+        let db = StateStore::open(&cfg.db_path)?;
+        let now = Utc::now();
+        let worktree_path = tempdir.path().join("keep-worktree");
+        fs::create_dir_all(&worktree_path)?;
+
+        db.insert_session(&Session {
+            id: "active-over-budget".to_string(),
+            task: "pause on hard limit".to_string(),
+            agent_type: "claude".to_string(),
+            working_dir: tempdir.path().to_path_buf(),
+            state: SessionState::Running,
+            pid: Some(999_999),
+            worktree: Some(crate::session::WorktreeInfo {
+                path: worktree_path.clone(),
+                branch: "ecc/active-over-budget".to_string(),
+                base_branch: "main".to_string(),
+            }),
+            created_at: now - Duration::minutes(1),
+            updated_at: now,
+            metrics: SessionMetrics::default(),
+        })?;
+        db.update_metrics(
+            "active-over-budget",
+            &SessionMetrics {
+                input_tokens: 90,
+                output_tokens: 30,
+                tokens_used: 120,
+                tool_calls: 0,
+                files_changed: 0,
+                duration_secs: 60,
+                cost_usd: 0.0,
+            },
+        )?;
+
+        let outcome = enforce_budget_hard_limits(&db, &cfg)?;
+        assert!(outcome.token_budget_exceeded);
+        assert!(!outcome.cost_budget_exceeded);
+        assert_eq!(
+            outcome.paused_sessions,
+            vec!["active-over-budget".to_string()]
+        );
+
+        let session = db
+            .get_session("active-over-budget")?
+            .context("session should still exist")?;
+        assert_eq!(session.state, SessionState::Stopped);
+        assert_eq!(session.pid, None);
+        assert!(
+            worktree_path.exists(),
+            "hard-limit pauses should preserve worktrees for resume"
+        );
+
+        Ok(())
+    }
+
+    #[test]
+    fn enforce_budget_hard_limits_ignores_inactive_sessions() -> Result<()> {
+        let tempdir = TestDir::new("manager-budget-ignore-inactive")?;
+        let mut cfg = build_config(tempdir.path());
+        cfg.token_budget = 100;
+        cfg.cost_budget_usd = 0.0;
+
+        let db = StateStore::open(&cfg.db_path)?;
+        let now = Utc::now();
+
+        db.insert_session(&Session {
+            id: "completed-over-budget".to_string(),
+            task: "already done".to_string(),
+            agent_type: "claude".to_string(),
+            working_dir: tempdir.path().to_path_buf(),
+            state: SessionState::Completed,
+            pid: None,
+            worktree: None,
+            created_at: now - Duration::minutes(2),
+            updated_at: now - Duration::minutes(1),
+            metrics: SessionMetrics::default(),
+        })?;
+        db.update_metrics(
+            "completed-over-budget",
+            &SessionMetrics {
+                input_tokens: 90,
+                output_tokens: 30,
+                tokens_used: 120,
+                tool_calls: 0,
+                files_changed: 0,
+                duration_secs: 60,
+                cost_usd: 0.0,
+            },
+        )?;
+
+        let outcome = enforce_budget_hard_limits(&db, &cfg)?;
+        assert!(outcome.token_budget_exceeded);
+        assert!(outcome.paused_sessions.is_empty());
+
+        let session = db
+            .get_session("completed-over-budget")?
+            .context("completed session should still exist")?;
+        assert_eq!(session.state, SessionState::Completed);
+
+        Ok(())
+    }
+
     #[tokio::test(flavor = "current_thread")]
     async fn resume_session_requeues_failed_session() -> Result<()> {
         let tempdir = TestDir::new("manager-resume-session")?;