fix: job token budget, iteration cap → Failed, web cancel stops worker (#788)

* feat: persist user_id in save_job and expose job_id on routine runs (#709)

* feat: persist worker events to DB and fix activity tab rendering

In-process Worker (used by Scheduler::dispatch_job) now persists events
via save_job_event at key execution points: plan creation, LLM
responses, tool_use, tool_result, and job completion/failure/stuck.
Event data shapes match the container worker format so the gateway
activity tab renders them correctly.

Frontend: tool_result errors now show a red X icon with danger styling
instead of a silent empty output. The result event falls back to the
error field when message is absent.

Co-Authored-By: Claude Opus 4.6 <[email protected]>

* feat: wire RoutineEngine into gateway for direct manual trigger firing

Replace the message-channel hack in routines_trigger_handler with a
direct call to RoutineEngine::fire_manual(), ensuring FullJob routines
dispatch correctly when triggered from the web UI. Inject the engine
into GatewayState from Agent::run after construction.

Also persists user_id in save_job for both PG and libSQL backends,
removes the source='sandbox' filter so all jobs are visible, and
exposes job_id on RoutineRunInfo for the frontend job link.

Co-Authored-By: Claude Opus 4.6 <[email protected]>

* fix: remove stale gateway_state argument from Agent::new test call sites

The gateway_state parameter was removed from Agent::new during rebase
(replaced by post-construction set_routine_engine_slot), but three test
call sites still passed the extra None argument.

Co-Authored-By: Claude Opus 4.6 <[email protected]>

* fix: address PR review — restore sandbox source filter, remove blank lines

- Revert removal of `source = 'sandbox'` filter in all SandboxStore
  queries (8 sites across PG and libSQL). Sandbox-specific APIs should
  stay scoped to sandbox jobs; unified job listing for the Jobs tab
  should use a separate query path.
- Remove extra blank lines in agent_loop.rs and worker.rs that caused
  formatting CI failure.

[skip-regression-check]

Co-Authored-By: Claude Opus 4.6 <[email protected]>

* fix: address review — regenerate Cargo.lock, add user_id regression test

- Regenerate Cargo.lock from main's lockfile to eliminate dependency
  version downgrades (anyhow, syn, etc.) that were churn from rebase.
- Add regression test verifying user_id round-trips through save_job
  and get_job in the libSQL backend.

Co-Authored-By: Claude Opus 4.6 <[email protected]>

* style: remove trailing blank line in libsql jobs.rs

[skip-regression-check]

Co-Authored-By: Claude Opus 4.6 <[email protected]>

* test: add Postgres-side regression test for user_id persistence in save_job

Mirrors the existing libSQL test (test_save_job_persists_user_id) for the
Postgres backend. Gated behind #[cfg(feature = "postgres")] + #[ignore]
since it requires a running PostgreSQL instance (integration tier).

Co-Authored-By: Claude Opus 4.6 <[email protected]>

---------

Co-authored-by: Claude Opus 4.6 <[email protected]>

* fix: add job token budget, change iteration cap to Failed, fix web cancel (#698)

Jobs could enter infinite retry loops because: (1) no token budget was
enforced, (2) iteration cap marked jobs as Stuck (allowing self-repair to
restart them), and (3) the web UI cancel button only updated the DB without
stopping the running worker.

- Add `max_tokens_per_job` config (settings.json + AGENT_MAX_TOKENS_PER_JOB
  env var, default 0 = unlimited) with per-job metadata override
- Track token usage after respond_with_tools() and fail the job on budget
  exceeded
- Change iteration cap and persistent rate limiting from mark_stuck to
  mark_failed, preventing self-repair restart loops
- Fix web cancel handler to call scheduler.stop() which updates in-memory
  state AND aborts the worker task, falling back to DB-only update

Co-Authored-By: Claude Opus 4.6 <[email protected]>

* fix: address PR review — always persist cancel to DB, simplify token check

- Cancel handler now always persists Cancelled to DB regardless of whether
  scheduler.stop() ran, fixing the edge case where stop() returns Ok(())
  for jobs not in the scheduler map
- Collapse nested ifs per clippy (let-chains)
- Add NOTE comment about select_tools() not exposing TokenUsage

[skip-regression-check]

Co-Authored-By: Claude Opus 4.6 <[email protected]>

* fix: rustfmt formatting in wizard.rs (pre-existing)

[skip-regression-check]

Co-Authored-By: Claude Opus 4.6 <[email protected]>

---------

Co-authored-by: Claude Opus 4.6 <[email protected]>

.env.example

@@ -115,6 +115,8 @@ AGENT_NAME=ironclaw
 AGENT_MAX_PARALLEL_JOBS=5
 AGENT_JOB_TIMEOUT_SECS=3600
 AGENT_STUCK_THRESHOLD_SECS=300
+# Maximum tokens per job (0 = unlimited, also settable via settings.json agent.max_tokens_per_job)
+# AGENT_MAX_TOKENS_PER_JOB=0
 # Enable planning phase before tool execution (default: true)
 AGENT_USE_PLANNING=true
 

src/agent/dispatcher.rs

@@ -1205,6 +1205,7 @@ mod tests {
                 max_tool_iterations: 50,
                 auto_approve_tools: false,
                 default_timezone: "UTC".to_string(),
+                max_tokens_per_job: 0,
             },
             deps,
             Arc::new(ChannelManager::new()),
@@ -2043,6 +2044,7 @@ mod tests {
                 max_tool_iterations,
                 auto_approve_tools: true,
                 default_timezone: "UTC".to_string(),
+                max_tokens_per_job: 0,
             },
             deps,
             Arc::new(ChannelManager::new()),
@@ -2159,6 +2161,7 @@ mod tests {
                     max_tool_iterations: max_iter,
                     auto_approve_tools: true,
                     default_timezone: "UTC".to_string(),
+                    max_tokens_per_job: 0,
                 },
                 deps,
                 Arc::new(ChannelManager::new()),

src/agent/scheduler.rs

@@ -160,6 +160,13 @@ impl Scheduler {
             .create_job_for_user(user_id, title, description)
             .await?;
 
+        // Apply token budget from config, allowing per-job metadata override.
+        let max_tokens = metadata
+            .as_ref()
+            .and_then(|m| m.get("max_tokens"))
+            .and_then(|v| v.as_u64())
+            .unwrap_or(self.config.max_tokens_per_job);
+
         // Apply metadata if provided
         if let Some(meta) = metadata {
             self.context_manager
@@ -169,6 +176,15 @@ impl Scheduler {
                 .await?;
         }
 
+        // Set token budget (separate update to avoid overwriting metadata)
+        if max_tokens > 0 {
+            self.context_manager
+                .update_context(job_id, |ctx| {
+                    ctx.max_tokens = max_tokens;
+                })
+                .await?;
+        }
+
         // Persist to DB before scheduling so the worker's FK references are valid
         if let Some(ref store) = self.store {
             let ctx = self.context_manager.get_context(job_id).await?;

src/agent/worker.rs

@@ -417,7 +417,8 @@ Report when the job is complete or if you encounter issues you cannot resolve."#
 
             iteration += 1;
             if iteration > max_iterations {
-                self.mark_stuck("Maximum iterations exceeded").await?;
+                self.mark_failed("Maximum iterations exceeded: job hit the iteration cap")
+                    .await?;
                 return Ok(());
             }
 
@@ -437,7 +438,8 @@ Report when the job is complete or if you encounter issues you cannot resolve."#
                         "LLM rate limited during tool selection, backing off"
                     );
                     if consecutive_rate_limits >= MAX_CONSECUTIVE_RATE_LIMITS {
-                        self.mark_stuck("Persistent rate limiting").await?;
+                        self.mark_failed("Persistent rate limiting: exceeded retry limit")
+                            .await?;
                         return Ok(());
                     }
                     self.log_event(
@@ -467,7 +469,8 @@ Report when the job is complete or if you encounter issues you cannot resolve."#
                             "LLM rate limited during respond_with_tools, backing off"
                         );
                         if consecutive_rate_limits >= MAX_CONSECUTIVE_RATE_LIMITS {
-                            self.mark_stuck("Persistent rate limiting").await?;
+                            self.mark_failed("Persistent rate limiting: exceeded retry limit")
+                                .await?;
                             return Ok(());
                         }
                         self.log_event(
@@ -483,6 +486,20 @@ Report when the job is complete or if you encounter issues you cannot resolve."#
                     Err(e) => return Err(e.into()),
                 };
 
+                // Track token usage from LLM call against the job budget.
+                // NOTE: select_tools() also makes LLM calls but doesn't expose
+                // TokenUsage; only respond_with_tools() usage is tracked here.
+                let total_tokens = respond_output.usage.total() as u64;
+                if total_tokens > 0
+                    && let Err(msg) = self
+                        .context_manager()
+                        .update_context(self.job_id, |ctx| ctx.add_tokens(total_tokens))
+                        .await?
+                {
+                    self.mark_failed(&msg).await?;
+                    return Ok(());
+                }
+
                 match respond_output.result {
                     RespondResult::Text(response) => {
                         // Check for explicit completion phrases. Use word-boundary
@@ -1762,4 +1779,84 @@ mod tests {
             "Always tool should be allowed with permission"
         );
     }
+
+    #[tokio::test]
+    async fn test_token_budget_exceeded_fails_job() {
+        let worker = make_worker(vec![]).await;
+
+        // Transition to InProgress (required for mark_failed)
+        worker
+            .context_manager()
+            .update_context(worker.job_id, |ctx| {
+                ctx.transition_to(JobState::InProgress, None)
+            })
+            .await
+            .unwrap()
+            .unwrap();
+
+        // Set a token budget
+        worker
+            .context_manager()
+            .update_context(worker.job_id, |ctx| {
+                ctx.max_tokens = 100;
+            })
+            .await
+            .unwrap();
+
+        // Simulate adding tokens that exceed the budget
+        let budget_result = worker
+            .context_manager()
+            .update_context(worker.job_id, |ctx| ctx.add_tokens(200))
+            .await
+            .unwrap();
+
+        assert!(
+            budget_result.is_err(),
+            "Should return error when token budget exceeded"
+        );
+
+        // Verify that mark_failed transitions job to Failed
+        worker
+            .mark_failed(&budget_result.unwrap_err())
+            .await
+            .unwrap();
+        let ctx = worker
+            .context_manager()
+            .get_context(worker.job_id)
+            .await
+            .unwrap();
+        assert_eq!(ctx.state, JobState::Failed);
+    }
+
+    #[tokio::test]
+    async fn test_iteration_cap_marks_failed_not_stuck() {
+        let worker = make_worker(vec![]).await;
+
+        // Transition to InProgress (required for mark_failed)
+        worker
+            .context_manager()
+            .update_context(worker.job_id, |ctx| {
+                ctx.transition_to(JobState::InProgress, None)
+            })
+            .await
+            .unwrap()
+            .unwrap();
+
+        // Simulate what the execution loop does when max_iterations is exceeded
+        worker
+            .mark_failed("Maximum iterations exceeded: job hit the iteration cap")
+            .await
+            .unwrap();
+
+        let ctx = worker
+            .context_manager()
+            .get_context(worker.job_id)
+            .await
+            .unwrap();
+        assert_eq!(
+            ctx.state,
+            JobState::Failed,
+            "Iteration cap should transition to Failed, not Stuck"
+        );
+    }
 }

src/channels/web/handlers/jobs.rs

@@ -276,11 +276,25 @@ pub async fn jobs_cancel_handler(
         })));
     }
 
-    // Fall back to agent job cancellation via DB status update.
+    // Fall back to agent job cancellation: stop the worker via the scheduler
+    // (which updates the in-memory ContextManager AND aborts the task handle),
+    // then persist the status to the DB as a fallback.
     if let Some(ref store) = state.store
         && let Ok(Some(job)) = store.get_job(job_id).await
     {
         if job.state.is_active() {
+            // Try to stop via scheduler (aborts the worker task + updates
+            // in-memory ContextManager). This is best-effort — the job may
+            // not be in the scheduler map if it already finished.
+            if let Some(ref slot) = state.scheduler
+                && let Some(ref scheduler) = *slot.read().await
+            {
+                let _ = scheduler.stop(job_id).await;
+            }
+
+            // Always persist cancellation to the DB so the state is
+            // consistent even if the scheduler wasn't available or the
+            // job wasn't in its in-memory map.
             store
                 .update_job_status(
                     job_id,

src/channels/web/handlers/routines.rs

@@ -108,6 +108,7 @@ pub async fn routines_detail_handler(
             status: format!("{:?}", run.status),
             result_summary: run.result_summary.clone(),
             tokens_used: run.tokens_used,
+            job_id: run.job_id,
         })
         .collect();
 
@@ -252,6 +253,7 @@ pub async fn routines_runs_handler(
             status: format!("{:?}", run.status),
             result_summary: run.result_summary.clone(),
             tokens_used: run.tokens_used,
+            job_id: run.job_id,
         })
         .collect();
 

src/channels/web/server.rs

@@ -2017,6 +2017,7 @@ async fn routines_detail_handler(
             status: format!("{:?}", run.status),
             result_summary: run.result_summary.clone(),
             tokens_used: run.tokens_used,
+            job_id: run.job_id,
         })
         .collect();
 
@@ -2169,6 +2170,7 @@ async fn routines_runs_handler(
             status: format!("{:?}", run.status),
             result_summary: run.result_summary.clone(),
             tokens_used: run.tokens_used,
+            job_id: run.job_id,
         })
         .collect();
 

src/channels/web/types.rs

@@ -776,6 +776,7 @@ pub struct RoutineRunInfo {
     pub status: String,
     pub result_summary: Option<String>,
     pub tokens_used: Option<i32>,
+    pub job_id: Option<Uuid>,
 }
 
 // --- Settings ---

src/config/agent.rs

@@ -29,6 +29,8 @@ pub struct AgentConfig {
     pub auto_approve_tools: bool,
     /// Default timezone for new sessions (IANA name, e.g. "America/New_York").
     pub default_timezone: String,
+    /// Maximum tokens per job (0 = unlimited).
+    pub max_tokens_per_job: u64,
 }
 
 impl AgentConfig {
@@ -50,6 +52,7 @@ impl AgentConfig {
             max_tool_iterations: 10,
             auto_approve_tools: true,
             default_timezone: "UTC".to_string(),
+            max_tokens_per_job: 0,
         }
     }
 
@@ -105,6 +108,10 @@ impl AgentConfig {
                 }
                 tz
             },
+            max_tokens_per_job: parse_optional_env(
+                "AGENT_MAX_TOKENS_PER_JOB",
+                settings.agent.max_tokens_per_job,
+            )?,
         })
     }
 }

src/db/libsql/jobs.rs

@@ -28,14 +28,16 @@ impl JobStore for LibSqlBackend {
                 r#"
                 INSERT INTO agent_jobs (
                     id, conversation_id, title, description, category, status, source,
+                    user_id,
                     budget_amount, budget_token, bid_amount, estimated_cost, estimated_time_secs,
                     actual_cost, repair_attempts, created_at, started_at, completed_at
-                ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17)
+                ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18)
                 ON CONFLICT (id) DO UPDATE SET
                     title = excluded.title,
                     description = excluded.description,
                     category = excluded.category,
                     status = excluded.status,
+                    user_id = excluded.user_id,
                     estimated_cost = excluded.estimated_cost,
                     estimated_time_secs = excluded.estimated_time_secs,
                     actual_cost = excluded.actual_cost,
@@ -51,6 +53,7 @@ impl JobStore for LibSqlBackend {
                     opt_text(ctx.category.as_deref()),
                     status,
                     "direct",
+                    ctx.user_id.as_str(),
                     opt_text_owned(ctx.budget.map(|d| d.to_string())),
                     opt_text(ctx.budget_token.as_deref()),
                     opt_text_owned(ctx.bid_amount.map(|d| d.to_string())),

src/db/libsql/mod.rs

@@ -482,6 +482,24 @@ mod tests {
         assert_eq!(timeout, 5000);
     }
 
+    /// Regression test: save_job must persist user_id and get_job must return it.
+    #[tokio::test]
+    async fn test_save_job_persists_user_id() {
+        use crate::context::JobContext;
+        use crate::db::JobStore;
+
+        let dir = tempfile::tempdir().unwrap();
+        let db_path = dir.path().join("test_user_id.db");
+        let backend = LibSqlBackend::new_local(&db_path).await.unwrap();
+        backend.run_migrations().await.unwrap();
+
+        let ctx = JobContext::with_user("test-user-42", "Test Job", "A test job");
+        backend.save_job(&ctx).await.unwrap();
+
+        let loaded = backend.get_job(ctx.job_id).await.unwrap().unwrap();
+        assert_eq!(loaded.user_id, "test-user-42");
+    }
+
     #[tokio::test]
     async fn test_concurrent_writes_succeed() {
         // Use a temp file so connections share state (in-memory DBs are connection-local)

src/history/store.rs

@@ -149,14 +149,16 @@ impl Store {
             r#"
             INSERT INTO agent_jobs (
                 id, conversation_id, title, description, category, status, source,
+                user_id,
                 budget_amount, budget_token, bid_amount, estimated_cost, estimated_time_secs,
                 actual_cost, repair_attempts, created_at, started_at, completed_at
-            ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17)
+            ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18)
             ON CONFLICT (id) DO UPDATE SET
                 title = EXCLUDED.title,
                 description = EXCLUDED.description,
                 category = EXCLUDED.category,
                 status = EXCLUDED.status,
+                user_id = EXCLUDED.user_id,
                 estimated_cost = EXCLUDED.estimated_cost,
                 estimated_time_secs = EXCLUDED.estimated_time_secs,
                 actual_cost = EXCLUDED.actual_cost,
@@ -172,6 +174,7 @@ impl Store {
                 &ctx.category,
                 &status,
                 &"direct", // source
+                &ctx.user_id,
                 &ctx.budget,
                 &ctx.budget_token,
                 &ctx.bid_amount,
@@ -2133,4 +2136,36 @@ mod tests {
             assert_eq!(summary.channel, ch);
         }
     }
+
+    /// Regression test: save_job must persist user_id and get_job must return it.
+    /// Requires a running PostgreSQL instance (integration tier).
+    #[cfg(feature = "postgres")]
+    #[tokio::test]
+    #[ignore]
+    async fn test_save_job_persists_user_id() {
+        use crate::config::Config;
+        use crate::context::JobContext;
+
+        let _ = dotenvy::dotenv();
+        let config = Config::from_env().await.expect("Failed to load config");
+        let store = Store::new(&config.database)
+            .await
+            .expect("Failed to connect to database");
+        store
+            .run_migrations()
+            .await
+            .expect("Failed to run migrations");
+
+        let ctx = JobContext::with_user("test-user-42", "PG user_id test", "regression test");
+        store.save_job(&ctx).await.unwrap();
+
+        let loaded = store.get_job(ctx.job_id).await.unwrap().unwrap();
+        assert_eq!(loaded.user_id, "test-user-42");
+
+        // Clean up
+        let conn = store.conn().await.unwrap();
+        conn.execute("DELETE FROM agent_jobs WHERE id = $1", &[&ctx.job_id])
+            .await
+            .unwrap();
+    }
 }

src/settings.rs

@@ -386,6 +386,10 @@ pub struct AgentSettings {
     /// Default timezone for new sessions (IANA name, e.g. "America/New_York").
     #[serde(default = "default_timezone")]
     pub default_timezone: String,
+
+    /// Maximum tokens per job (0 = unlimited).
+    #[serde(default)]
+    pub max_tokens_per_job: u64,
 }
 
 fn default_agent_name() -> String {
@@ -442,6 +446,7 @@ impl Default for AgentSettings {
             max_tool_iterations: default_max_tool_iterations(),
             auto_approve_tools: false,
             default_timezone: default_timezone(),
+            max_tokens_per_job: 0,
         }
     }
 }