feat: capacity governor / backoff for work autonomous job runner

crates/agentflare-jobs/src/executor.rs

@@ -17,5 +17,35 @@ pub trait InProcessExecutor: Send + Sync {
         job_id: &str,
         args: &[String],
         log: &mut dyn std::io::Write,
-    ) -> Result<(), String>;
+    ) -> Result<(), JobFailure>;
+}
+
+/// What went wrong, plus an optional hint for how long to wait before this
+/// job (if it still has retries left) becomes eligible again. Set by a
+/// caller that classifies the failure as transient/rate-limit shaped (see
+/// the main binary's `auth_runner`/`auth_db` — this crate stays agnostic to
+/// *why* a delay is warranted, it just carries the number through to
+/// `Queue::fail`). Left `None` for anything else, which keeps `Queue::fail`'s
+/// existing instant-requeue behavior.
+pub struct JobFailure {
+    pub message: String,
+    pub retry_after_secs: Option<u64>,
+}
+
+impl From<String> for JobFailure {
+    fn from(message: String) -> Self {
+        JobFailure {
+            message,
+            retry_after_secs: None,
+        }
+    }
+}
+
+impl From<&str> for JobFailure {
+    fn from(message: &str) -> Self {
+        JobFailure {
+            message: message.to_string(),
+            retry_after_secs: None,
+        }
+    }
 }

crates/agentflare-jobs/src/lib.rs

@@ -4,7 +4,7 @@ pub mod supervisor;
 pub mod types;
 pub mod worker;
 
-pub use executor::InProcessExecutor;
+pub use executor::{InProcessExecutor, JobFailure};
 pub use queue::Queue;
 pub use supervisor::Supervisor;
 pub use types::{AgentJob, JobInfo, JobOutput, JobState};

crates/agentflare-jobs/src/queue.rs

@@ -59,6 +59,7 @@ pub fn migrations() -> rusqlite_migration::Migrations<'static> {
             "ALTER TABLE agent_jobs ADD COLUMN stdout_bytes INTEGER NOT NULL DEFAULT 0;
             ALTER TABLE agent_jobs ADD COLUMN stderr_bytes INTEGER NOT NULL DEFAULT 0;",
         ),
+        rusqlite_migration::M::up("ALTER TABLE agent_jobs ADD COLUMN not_before INTEGER;"),
     ])
 }
 
@@ -145,21 +146,21 @@ impl Queue {
     pub fn dequeue(&self) -> Result<Option<(String, crate::types::AgentJob)>, Error> {
         let mut conn = self.conn.lock();
         let tx = conn.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
+        let now = db_kit::ids::now();
         let row: Option<(String, String)> = {
             let mut stmt = tx.prepare(
                 "SELECT id, payload FROM agent_jobs
-                 WHERE state = 'queued'
+                 WHERE state = 'queued' AND (not_before IS NULL OR not_before <= ?1)
                  ORDER BY created_at ASC
                  LIMIT 1",
             )?;
-            stmt.query_row([], |r| Ok((r.get(0)?, r.get(1)?)))
+            stmt.query_row(params![now], |r| Ok((r.get(0)?, r.get(1)?)))
                 .optional()?
         };
         let (id, payload_json) = match row {
             Some(r) => r,
             None => return Ok(None),
         };
-        let now = db_kit::ids::now();
         tx.execute(
             "UPDATE agent_jobs SET state = 'running', started_at = ?1 WHERE id = ?2",
             params![now, id],
@@ -199,7 +200,7 @@ impl Queue {
         Ok(())
     }
 
-    pub fn fail(&self, id: &str, error: &str) -> Result<(), Error> {
+    pub fn fail(&self, id: &str, error: &str, retry_after_secs: Option<u64>) -> Result<(), Error> {
         let now = db_kit::ids::now();
         let conn = self.conn.lock();
         let (retries, max_retries): (u32, u32) = conn.query_row(
@@ -209,11 +210,12 @@ impl Queue {
         )?;
         let retried = retries < max_retries;
         if retried {
+            let not_before = retry_after_secs.map(|s| now + s as i64);
             conn.execute(
                 "UPDATE agent_jobs
-                 SET state = 'queued', retries = retries + 1, error = ?1, started_at = NULL
-                 WHERE id = ?2",
-                params![error, id],
+                 SET state = 'queued', retries = retries + 1, error = ?1, started_at = NULL, not_before = ?2
+                 WHERE id = ?3",
+                params![error, not_before, id],
             )?;
         } else {
             conn.execute(
@@ -225,7 +227,11 @@ impl Queue {
         }
         drop(conn);
         // A retry goes back to 'queued' — wake workers so it's picked up
-        // promptly instead of waiting out the fallback poll interval.
+        // promptly instead of waiting out the fallback poll interval. Waking
+        // them even when `not_before` is in the future is harmless (their
+        // next `dequeue` simply finds nothing and re-parks) and keeps this
+        // function simple; the fallback poll in `worker_loop` is the real
+        // backstop for the delayed case regardless.
         if retried {
             self.wake_workers();
         }
@@ -391,3 +397,70 @@ impl<T> OptionalExt<T> for Result<T, rusqlite::Error> {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::types::AgentJob;
+
+    fn test_queue() -> (Queue, tempfile::TempDir) {
+        let dir = tempfile::tempdir().unwrap();
+        let queue = Queue::open_memory(dir.path().join("logs")).unwrap();
+        (queue, dir)
+    }
+
+    #[test]
+    fn fail_without_retry_delay_requeues_immediately() {
+        let (queue, _dir) = test_queue();
+        let job = AgentJob::new("true").max_retries(1);
+        let info = queue.enqueue(&job).unwrap();
+        queue.dequeue().unwrap();
+        queue.fail(&info.id, "boom", None).unwrap();
+        let (id, _) = queue
+            .dequeue()
+            .unwrap()
+            .expect("retried job should be immediately eligible");
+        assert_eq!(id, info.id);
+    }
+
+    #[test]
+    fn fail_with_retry_delay_hides_the_job_until_not_before_passes() {
+        let (queue, _dir) = test_queue();
+        let job = AgentJob::new("true").max_retries(1);
+        let info = queue.enqueue(&job).unwrap();
+        queue.dequeue().unwrap();
+        queue.fail(&info.id, "rate limited", Some(3600)).unwrap();
+
+        assert!(
+            queue.dequeue().unwrap().is_none(),
+            "a job with a future not_before must not be dequeued yet"
+        );
+
+        // Simulate the delay having elapsed, without a real sleep.
+        {
+            let conn = queue.conn.lock();
+            conn.execute(
+                "UPDATE agent_jobs SET not_before = 0 WHERE id = ?1",
+                params![info.id],
+            )
+            .unwrap();
+        }
+        let (id, _) = queue
+            .dequeue()
+            .unwrap()
+            .expect("job should be eligible once not_before has passed");
+        assert_eq!(id, info.id);
+    }
+
+    #[test]
+    fn fail_past_max_retries_marks_failed_regardless_of_retry_delay() {
+        let (queue, _dir) = test_queue();
+        let job = AgentJob::new("true").max_retries(0);
+        let info = queue.enqueue(&job).unwrap();
+        queue.dequeue().unwrap();
+        queue.fail(&info.id, "boom", Some(60)).unwrap();
+        let jobs = queue.list(Some(JobState::Failed)).unwrap();
+        assert_eq!(jobs.len(), 1);
+        assert_eq!(jobs[0].id, info.id);
+    }
+}

crates/agentflare-jobs/src/worker.rs

@@ -1,4 +1,4 @@
-use crate::executor::InProcessExecutor;
+use crate::executor::{InProcessExecutor, JobFailure};
 use crate::queue::Queue;
 use crate::supervisor::Supervisor;
 use crate::types::JobOutput;
@@ -99,7 +99,7 @@ fn worker_loop(queue: &Queue, running: &AtomicBool, executor: Option<&Arc<dyn In
                         }
                     }
                     Err(e) => {
-                        if let Err(qe) = queue.fail(&id, &e.to_string()) {
+                        if let Err(qe) = queue.fail(&id, &e.to_string(), None) {
                             eprintln!("agentflare-jobs: failed to record failure for {id}: {qe}");
                         }
                     }
@@ -147,6 +147,7 @@ fn run_in_process(
         if let Err(e) = queue.fail(
             id,
             "job is marked in_process but no InProcessExecutor is registered on this WorkerPool",
+            None,
         ) {
             eprintln!("agentflare-jobs: failed to record failure for {id}: {e}");
         }
@@ -158,14 +159,14 @@ fn run_in_process(
     let mut log_file = match std::fs::File::create(&stdout_path) {
         Ok(f) => f,
         Err(e) => {
-            if let Err(qe) = queue.fail(id, &format!("failed to open job log file: {e}")) {
+            if let Err(qe) = queue.fail(id, &format!("failed to open job log file: {e}"), None) {
                 eprintln!("agentflare-jobs: failed to record failure for {id}: {qe}");
             }
             return;
         }
     };
 
-    let (tx, rx) = std::sync::mpsc::channel::<Result<(), String>>();
+    let (tx, rx) = std::sync::mpsc::channel::<Result<(), JobFailure>>();
     let executor = executor.clone();
     let job_id = id.to_string();
     let args = job.args.clone();
@@ -192,8 +193,8 @@ fn run_in_process(
                 eprintln!("agentflare-jobs: failed to complete job {id}: {e}");
             }
         }
-        Ok(Err(msg)) => {
-            if let Err(e) = queue.fail(id, &msg) {
+        Ok(Err(failure)) => {
+            if let Err(e) = queue.fail(id, &failure.message, failure.retry_after_secs) {
                 eprintln!("agentflare-jobs: failed to record failure for {id}: {e}");
             }
         }
@@ -205,7 +206,7 @@ fn run_in_process(
                  what this timeout measures)",
                 job.timeout_secs
             );
-            if let Err(e) = queue.fail(id, &msg) {
+            if let Err(e) = queue.fail(id, &msg, None) {
                 eprintln!("agentflare-jobs: failed to record failure for {id}: {e}");
             }
         }

crates/agentflare-jobs/tests/in_process_test.rs

@@ -5,7 +5,9 @@
 //! separate from `worker_test.rs`'s existing subprocess-path coverage, which
 //! stays unchanged and passing to prove that path is untouched.
 
-use agentflare_jobs::{AgentJob, InProcessExecutor, JobInfo, JobState, Queue, WorkerPool};
+use agentflare_jobs::{
+    AgentJob, InProcessExecutor, JobFailure, JobInfo, JobState, Queue, WorkerPool,
+};
 use std::sync::Arc;
 
 fn test_queue() -> Queue {
@@ -34,7 +36,7 @@ impl InProcessExecutor for EchoExecutor {
         job_id: &str,
         args: &[String],
         log: &mut dyn std::io::Write,
-    ) -> Result<(), String> {
+    ) -> Result<(), JobFailure> {
         let _ = writeln!(log, "running job {job_id} with args {args:?}");
         Ok(())
     }
@@ -81,8 +83,8 @@ impl InProcessExecutor for FailingExecutor {
         _job_id: &str,
         _args: &[String],
         _log: &mut dyn std::io::Write,
-    ) -> Result<(), String> {
-        Err("deliberate failure".to_string())
+    ) -> Result<(), JobFailure> {
+        Err("deliberate failure".into())
     }
 }
 
@@ -135,7 +137,7 @@ impl InProcessExecutor for SlowExecutor {
         _job_id: &str,
         _args: &[String],
         _log: &mut dyn std::io::Write,
-    ) -> Result<(), String> {
+    ) -> Result<(), JobFailure> {
         std::thread::sleep(std::time::Duration::from_secs(5));
         Ok(())
     }

crates/agentflare-jobs/tests/queue_test.rs

@@ -122,19 +122,19 @@ fn fail_retries_then_permanent() {
     let id = info.id;
 
     // First failure → retry (queued again)
-    q.fail(&id, "err1").unwrap();
+    q.fail(&id, "err1", None).unwrap();
     let info = q.get(&id).unwrap();
     assert_eq!(info.state, JobState::Queued);
     assert_eq!(info.retries, 1);
 
     // Second failure → retry
-    q.fail(&id, "err2").unwrap();
+    q.fail(&id, "err2", None).unwrap();
     let info = q.get(&id).unwrap();
     assert_eq!(info.state, JobState::Queued);
     assert_eq!(info.retries, 2);
 
     // Third failure → permanent
-    q.fail(&id, "err3").unwrap();
+    q.fail(&id, "err3", None).unwrap();
     let info = q.get(&id).unwrap();
     assert_eq!(info.state, JobState::Failed);
     assert_eq!(info.error.as_deref(), Some("err3"));
@@ -177,7 +177,7 @@ fn cleanup_removes_old_jobs() {
     let q = test_queue();
     q.enqueue(&AgentJob::new("a")).unwrap();
     let (id, _) = q.dequeue().unwrap().unwrap();
-    q.fail(&id, "done").unwrap();
+    q.fail(&id, "done", None).unwrap();
 
     // Use a large negative cutoff to simulate "now = 0"
     // The jobs have created_at = now (positive), so anything older than

src/agent_launch.rs

@@ -375,7 +375,11 @@ pub fn run_headless(
                 diagnostic_suffix(&c)
             ))
         }
-        Ok(_) => HeadlessOutcome::Failed(format!("{} exited non-zero", spec.display_name)),
+        Ok(c) => HeadlessOutcome::Failed(format!(
+            "{} exited non-zero{}",
+            spec.display_name,
+            diagnostic_suffix(&c)
+        )),
         Err(e) => HeadlessOutcome::Failed(format!("failed to run {}: {e}", spec.display_name)),
     }
 }
@@ -862,4 +866,25 @@ mod tests {
         assert!(suffix.contains("last stderr before kill"));
         assert!(suffix.contains("panic: something broke"));
     }
+
+    #[test]
+    fn failed_non_zero_exit_includes_captured_output() {
+        let captured = Captured {
+            stdout: String::new(),
+            stderr: "HTTP 429 Too Many Requests".to_string(),
+            success: false,
+            timed_out: false,
+            idle_killed: false,
+        };
+        let msg = match Ok::<_, std::io::Error>(captured) {
+            Ok(c) if c.success => unreachable!(),
+            Ok(c) if c.timed_out => unreachable!(),
+            Ok(c) => format!("test exited non-zero{}", diagnostic_suffix(&c)),
+            Err(_) => unreachable!(),
+        };
+        assert!(
+            msg.contains("HTTP 429 Too Many Requests"),
+            "expected captured stderr in the failure message, got: {msg}"
+        );
+    }
 }

src/auth_db.rs

@@ -260,6 +260,19 @@ pub fn list_cooldowns(conn: &Connection, agent: Option<&str>) -> Vec<CooldownRow
     }
 }
 
+/// True if `agent` has at least one active (not-yet-expired) cooldown row,
+/// for any profile. Checked by agent only, not agent+profile: a caller
+/// deciding whether to dispatch a not-yet-started job doesn't know which
+/// profile that job will end up using, so this errs toward not dispatching
+/// rather than guessing a specific profile. Used by the autonomous
+/// work-dispatch path (`src/supervisor.rs`'s discovery tick, `src/cli/work.rs`'s
+/// `classify_and_cooldown`) to pause redispatch into a rate limit the
+/// interactive path (`auth_runner`) already recorded — or that this same
+/// autonomous path just recorded itself.
+pub fn is_cooling_down(conn: &Connection, agent: &str) -> bool {
+    !list_cooldowns(conn, Some(agent)).is_empty()
+}
+
 pub fn clear_cooldown(conn: &Connection, agent: &str, profile: &str) {
     conn.execute(
         "DELETE FROM cooldowns WHERE agent = ?1 AND profile = ?2",
@@ -447,6 +460,16 @@ mod tests {
         });
     }
 
+    #[test]
+    fn is_cooling_down_reflects_active_cooldowns() {
+        with_temp_home(|| {
+            let conn = open_or_rebuild();
+            assert!(!is_cooling_down(&conn, "claude-code"));
+            set_cooldown(&conn, "claude-code", "alice", 30, "rate limit");
+            assert!(is_cooling_down(&conn, "claude-code"));
+        });
+    }
+
     #[test]
     fn clear_cooldown_removes() {
         with_temp_home(|| {