From 04d3b005b17cba5eb1822407fd6bfeadf6109d9e Mon Sep 17 00:00:00 2001 From: Illia Polosukhin Date: Sun, 22 Feb 2026 00:18:21 -0800 Subject: [PATCH] feat: implement FullJob routine mode with scheduler dispatch (#288) * feat: implement FullJob routine mode with scheduler dispatch FullJob routines previously fell back to lightweight mode (single LLM call, no tools) with a warning. This wires them to the existing Scheduler/Worker infrastructure so they dispatch real jobs with full tool access. Fire-and-forget model: the routine creates a job via ContextManager, schedules it, links the routine_run to the job_id, and completes immediately. The job runs independently with full tool access. - Add RoutineError::JobDispatchFailed variant - Add RoutineStore::link_routine_run_to_job (PostgreSQL + libSQL) - Add execute_full_job() in routine_engine with context_manager/scheduler - Wire context_manager + scheduler into RoutineEngine from agent_loop - Fix pre-existing clippy warnings in tests/html_to_markdown.rs Co-Authored-By: Claude Opus 4.6 * fix: persist job to DB before scheduling in execute_full_job The worker emits job_actions and llm_calls rows that reference agent_jobs via foreign key. Without persisting the job first, those inserts can fail. Match the pattern from commands.rs: fetch JobContext, save_job(), then schedule. Co-Authored-By: Claude Opus 4.6 * refactor: consolidate job dispatch into Scheduler::dispatch_job and wire max_iterations Move the create + persist + schedule sequence into a single Scheduler::dispatch_job() method so callers (commands.rs, routine_engine.rs) don't duplicate the logic. FullJob routines now pass max_iterations via job metadata, and the worker reads it (defaulting to 50 if unset). Also removes the context_manager field from RoutineEngine since dispatch_job handles everything internally. Co-Authored-By: Claude Opus 4.6 * fix: clamp max_iterations to 500 and log category update failures Address PR review feedback: - worker.rs: clamp max_iterations from metadata to MAX_WORKER_ITERATIONS (500) to prevent unbounded LLM token usage from malicious/buggy configs - commands.rs: log warning on category update failure instead of silently discarding the error Co-Authored-By: Claude Opus 4.6 * 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 --------- Co-authored-by: Claude Opus 4.6 --- src/agent/agent_loop.rs | 1 + src/agent/commands.rs | 29 +++------- src/agent/routine_engine.rs | 90 ++++++++++++++++++++--------- src/agent/scheduler.rs | 44 ++++++++++++++ src/agent/worker.rs | 95 ++++++++++++++++++++++++++++++- src/channels/web/static/app.js | 20 +++++-- src/channels/web/static/style.css | 8 +++ src/db/libsql/routines.rs | 15 +++++ src/db/mod.rs | 5 ++ src/db/postgres.rs | 8 +++ src/error.rs | 3 + src/history/store.rs | 15 +++++ 12 files changed, 279 insertions(+), 54 deletions(-) diff --git a/src/agent/agent_loop.rs b/src/agent/agent_loop.rs index bc0c1447..71108b6f 100644 --- a/src/agent/agent_loop.rs +++ b/src/agent/agent_loop.rs @@ -397,6 +397,7 @@ impl Agent { self.llm().clone(), Arc::clone(workspace), notify_tx, + Some(self.scheduler.clone()), )); // Register routine tools diff --git a/src/agent/commands.rs b/src/agent/commands.rs index 2661fed1..2b475727 100644 --- a/src/agent/commands.rs +++ b/src/agent/commands.rs @@ -73,36 +73,23 @@ impl Agent { description: String, category: Option, ) -> Result { - // Create job context let job_id = self - .context_manager - .create_job_for_user(user_id, &title, &description) + .scheduler + .dispatch_job(user_id, &title, &description, None) .await?; - // Update category if provided - if let Some(cat) = category { - self.context_manager + // Set the dedicated category field (not stored in metadata) + if let Some(cat) = category + && let Err(e) = self + .context_manager .update_context(job_id, |ctx| { ctx.category = Some(cat); }) - .await?; - } - - // Persist new job to database (fire-and-forget) - if let Some(store) = self.store() - && let Ok(ctx) = self.context_manager.get_context(job_id).await + .await { - let store = store.clone(); - tokio::spawn(async move { - if let Err(e) = store.save_job(&ctx).await { - tracing::warn!("Failed to persist new job {}: {}", job_id, e); - } - }); + tracing::warn!(job_id = %job_id, "Failed to set job category: {}", e); } - // Schedule for execution - self.scheduler.schedule(job_id).await?; - Ok(format!( "Created job: {}\nID: {}\n\nThe job has been scheduled and is now running.", title, job_id diff --git a/src/agent/routine_engine.rs b/src/agent/routine_engine.rs index 93e760f7..51e1e0ae 100644 --- a/src/agent/routine_engine.rs +++ b/src/agent/routine_engine.rs @@ -19,6 +19,7 @@ use regex::Regex; use tokio::sync::{RwLock, mpsc}; use uuid::Uuid; +use crate::agent::Scheduler; use crate::agent::routine::{ NotifyConfig, Routine, RoutineAction, RoutineRun, RunStatus, Trigger, next_cron_fire, }; @@ -41,6 +42,8 @@ pub struct RoutineEngine { running_count: Arc, /// Compiled event regex cache: routine_id -> compiled regex. event_cache: Arc>>, + /// Scheduler for dispatching jobs (FullJob mode). + scheduler: Option>, } impl RoutineEngine { @@ -50,6 +53,7 @@ impl RoutineEngine { llm: Arc, workspace: Arc, notify_tx: mpsc::Sender, + scheduler: Option>, ) -> Self { Self { config, @@ -59,6 +63,7 @@ impl RoutineEngine { notify_tx, running_count: Arc::new(AtomicUsize::new(0)), event_cache: Arc::new(RwLock::new(Vec::new())), + scheduler, } } @@ -225,7 +230,7 @@ impl RoutineEngine { workspace: self.workspace.clone(), notify_tx: self.notify_tx.clone(), running_count: self.running_count.clone(), - max_lightweight_tokens: self.config.max_lightweight_tokens, + scheduler: self.scheduler.clone(), }; tokio::spawn(async move { @@ -257,7 +262,7 @@ impl RoutineEngine { workspace: self.workspace.clone(), notify_tx: self.notify_tx.clone(), running_count: self.running_count.clone(), - max_lightweight_tokens: self.config.max_lightweight_tokens, + scheduler: self.scheduler.clone(), }; // Record the run in DB, then spawn execution @@ -304,7 +309,7 @@ struct EngineContext { workspace: Arc, notify_tx: mpsc::Sender, running_count: Arc, - max_lightweight_tokens: u32, + scheduler: Option>, } /// Execute a routine run. Handles both lightweight and full_job modes. @@ -318,29 +323,11 @@ async fn execute_routine(ctx: EngineContext, routine: Routine, run: RoutineRun) context_paths, max_tokens, } => execute_lightweight(&ctx, &routine, prompt, context_paths, *max_tokens).await, - RoutineAction::FullJob { description, .. } => { - // Full job mode: scheduler integration not yet implemented. - // Execute as lightweight and prepend a warning to the summary. - tracing::warn!( - routine = %routine.name, - "FullJob mode not yet implemented; falling back to lightweight execution" - ); - match execute_lightweight(&ctx, &routine, description, &[], ctx.max_lightweight_tokens) - .await - { - Ok((status, summary, tokens)) => { - let warning = "[Note: FullJob mode is not yet implemented. This routine ran as \ - a single LLM call without tool access. Configure as 'lightweight' \ - or wait for full scheduler integration.]"; - let summary = match summary { - Some(s) => Some(format!("{warning}\n\n{s}")), - None => Some(warning.to_string()), - }; - Ok((status, summary, tokens)) - } - Err(e) => Err(e), - } - } + RoutineAction::FullJob { + title, + description, + max_iterations, + } => execute_full_job(&ctx, &routine, &run, title, description, *max_iterations).await, }; // Decrement running count @@ -418,6 +405,57 @@ fn sanitize_routine_name(name: &str) -> String { .collect() } +/// Execute a full-job routine by dispatching to the scheduler. +/// +/// Fire-and-forget: creates a job via `Scheduler::dispatch_job` (which handles +/// creation, metadata, persistence, and scheduling), links the routine run to +/// the job, and returns immediately. The job runs independently via the +/// existing Worker/Scheduler with full tool access. +async fn execute_full_job( + ctx: &EngineContext, + routine: &Routine, + run: &RoutineRun, + title: &str, + description: &str, + max_iterations: u32, +) -> Result<(RunStatus, Option, Option), RoutineError> { + let scheduler = ctx + .scheduler + .as_ref() + .ok_or_else(|| RoutineError::JobDispatchFailed { + reason: "scheduler not available".to_string(), + })?; + + let metadata = serde_json::json!({ "max_iterations": max_iterations }); + + let job_id = scheduler + .dispatch_job(&routine.user_id, title, description, Some(metadata)) + .await + .map_err(|e| RoutineError::JobDispatchFailed { + reason: format!("failed to dispatch job: {e}"), + })?; + + // Link the routine run to the dispatched job + if let Err(e) = ctx.store.link_routine_run_to_job(run.id, job_id).await { + tracing::error!( + routine = %routine.name, + "Failed to link run to job: {}", e + ); + } + + tracing::info!( + routine = %routine.name, + job_id = %job_id, + max_iterations = max_iterations, + "Dispatched full job for routine" + ); + + let summary = format!( + "Dispatched job {job_id} for full execution with tool access (max_iterations: {max_iterations})" + ); + Ok((RunStatus::Ok, Some(summary), None)) +} + /// Execute a lightweight routine (single LLM call). async fn execute_lightweight( ctx: &EngineContext, diff --git a/src/agent/scheduler.rs b/src/agent/scheduler.rs index 048cf6e3..5950a8c7 100644 --- a/src/agent/scheduler.rs +++ b/src/agent/scheduler.rs @@ -81,6 +81,50 @@ impl Scheduler { } } + /// Create, persist, and schedule a job in one shot. + /// + /// This is the preferred entry point for dispatching new jobs. It: + /// 1. Creates the job context via `ContextManager` + /// 2. Optionally applies metadata (e.g. `max_iterations`) + /// 3. Persists the job to the database (so FK references from + /// `job_actions` / `llm_calls` work immediately) + /// 4. Schedules the job for worker execution + /// + /// Returns the new job ID. + pub async fn dispatch_job( + &self, + user_id: &str, + title: &str, + description: &str, + metadata: Option, + ) -> Result { + let job_id = self + .context_manager + .create_job_for_user(user_id, title, description) + .await?; + + // Apply metadata if provided + if let Some(meta) = metadata { + self.context_manager + .update_context(job_id, |ctx| { + ctx.metadata = meta; + }) + .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?; + store.save_job(&ctx).await.map_err(|e| JobError::Failed { + id: job_id, + reason: format!("failed to persist job: {e}"), + })?; + } + + self.schedule(job_id).await?; + Ok(job_id) + } + /// Schedule a job for execution. pub async fn schedule(&self, job_id: Uuid) -> Result<(), JobError> { // Hold write lock for the entire check-insert sequence to prevent diff --git a/src/agent/worker.rs b/src/agent/worker.rs index e953f705..87ee0ed4 100644 --- a/src/agent/worker.rs +++ b/src/agent/worker.rs @@ -98,6 +98,20 @@ impl Worker { } } + /// Fire-and-forget persistence of a job event. + fn log_event(&self, event_type: &str, data: serde_json::Value) { + if let Some(store) = self.store() { + let store = store.clone(); + let job_id = self.job_id; + let event_type = event_type.to_string(); + tokio::spawn(async move { + if let Err(e) = store.save_job_event(job_id, &event_type, &data).await { + tracing::warn!("Failed to persist event for job {}: {}", job_id, e); + } + }); + } + } + /// Run the worker until the job is complete or stopped. pub async fn run(self, mut rx: mpsc::Receiver) -> Result<(), Error> { tracing::info!("Worker starting for job {}", self.job_id); @@ -164,7 +178,15 @@ Report when the job is complete or if you encounter issues you cannot resolve."# reasoning: &Reasoning, reason_ctx: &mut ReasoningContext, ) -> Result<(), Error> { - let max_iterations = 50; + const MAX_WORKER_ITERATIONS: usize = 500; + let max_iterations = self + .context_manager() + .get_context(self.job_id) + .await + .ok() + .and_then(|ctx| ctx.metadata.get("max_iterations").and_then(|v| v.as_u64())) + .unwrap_or(50) as usize; + let max_iterations = max_iterations.min(MAX_WORKER_ITERATIONS); let mut iteration = 0; // Initial tool definitions for planning (will be refreshed in loop) @@ -193,6 +215,14 @@ Report when the job is complete or if you encounter issues you cannot resolve."# .join("\n") ))); + self.log_event("message", serde_json::json!({ + "role": "assistant", + "content": format!("Plan: {}\n\n{}", p.goal, + p.actions.iter().enumerate() + .map(|(i, a)| format!("{}. {} - {}", i + 1, a.tool_name, a.reasoning)) + .collect::>().join("\n")) + })); + Some(p) } Err(e) => { @@ -267,6 +297,14 @@ Report when the job is complete or if you encounter issues you cannot resolve."# // Add assistant response to context reason_ctx.messages.push(ChatMessage::assistant(&response)); + self.log_event( + "message", + serde_json::json!({ + "role": "assistant", + "content": response, + }), + ); + // Give it one more chance to select a tool if iteration > 3 && iteration % 5 == 0 { reason_ctx.messages.push(ChatMessage::user( @@ -285,6 +323,16 @@ Report when the job is complete or if you encounter issues you cannot resolve."# tool_calls.len() ); + if let Some(ref text) = content { + self.log_event( + "message", + serde_json::json!({ + "role": "assistant", + "content": text, + }), + ); + } + // Add assistant message with tool_calls (OpenAI protocol) reason_ctx .messages @@ -667,6 +715,15 @@ Report when the job is complete or if you encounter issues you cannot resolve."# selection: &ToolSelection, result: Result, ) -> Result { + self.log_event( + "tool_use", + serde_json::json!({ + "tool_name": selection.tool_name, + "input": crate::agent::agent_loop::truncate_for_preview( + &selection.parameters.to_string(), 500), + }), + ); + match result { Ok(output) => { // Sanitize output @@ -687,6 +744,12 @@ Report when the job is complete or if you encounter issues you cannot resolve."# wrapped, )); + self.log_event("tool_result", serde_json::json!({ + "tool_name": selection.tool_name, + "success": true, + "output": crate::agent::agent_loop::truncate_for_preview(&sanitized.content, 500), + })); + // Tool output never drives job completion. A malicious tool could // emit "TASK_COMPLETE" to force premature completion. Only the LLM's // own structured response (in execution_loop) can mark a job done. @@ -713,6 +776,15 @@ Report when the job is complete or if you encounter issues you cannot resolve."# }); } + self.log_event( + "tool_result", + serde_json::json!({ + "tool_name": selection.tool_name, + "success": false, + "output": format!("Error: {}", e), + }), + ); + reason_ctx.messages.push(ChatMessage::tool_result( &selection.tool_call_id, &selection.tool_name, @@ -834,6 +906,13 @@ Report when the job is complete or if you encounter issues you cannot resolve."# reason: s, })?; + self.log_event( + "result", + serde_json::json!({ + "success": true, + "message": "Job completed successfully", + }), + ); self.persist_status( JobState::Completed, Some("Job completed successfully".to_string()), @@ -852,6 +931,13 @@ Report when the job is complete or if you encounter issues you cannot resolve."# reason: s, })?; + self.log_event( + "result", + serde_json::json!({ + "success": false, + "message": format!("Execution failed: {}", reason), + }), + ); self.persist_status(JobState::Failed, Some(reason.to_string())); Ok(()) } @@ -865,6 +951,13 @@ Report when the job is complete or if you encounter issues you cannot resolve."# reason: s, })?; + self.log_event( + "result", + serde_json::json!({ + "success": false, + "message": format!("Job stuck: {}", reason), + }), + ); self.persist_status(JobState::Stuck, Some(reason.to_string())); Ok(()) } diff --git a/src/channels/web/static/app.js b/src/channels/web/static/app.js index fe225c0a..02a7ea73 100644 --- a/src/channels/web/static/app.js +++ b/src/channels/web/static/app.js @@ -2187,14 +2187,20 @@ function appendActivityEvent(terminal, eventType, data) { + escapeHtml(typeof data.input === 'string' ? data.input : JSON.stringify(data.input, null, 2)) + ''; break; - case 'tool_result': - el.innerHTML = '
' - + '✓ ' + case 'tool_result': { + const trSuccess = data.success !== false; + const trIcon = trSuccess ? '✓' : '✗'; + const trOutput = data.output || data.error || ''; + const trClass = 'activity-tool-block activity-tool-result' + + (trSuccess ? '' : ' activity-tool-error'); + el.innerHTML = '
' + + '' + trIcon + ' ' + escapeHtml(data.tool_name || 'result') + '
'
-        + escapeHtml(data.output || '')
+        + escapeHtml(trOutput)
         + '
'; break; + } case 'status': el.innerHTML = '' + escapeHtml(data.message || '') + ''; break; @@ -2202,7 +2208,7 @@ function appendActivityEvent(terminal, eventType, data) { el.className += ' activity-final'; const success = data.success !== false; el.innerHTML = '' - + escapeHtml(data.message || data.status || 'done') + ''; + + escapeHtml(data.message || data.error || data.status || 'done') + ''; if (data.session_id) { el.innerHTML += ' session: ' + escapeHtml(data.session_id) + ''; } @@ -2387,7 +2393,9 @@ function renderRoutineDetail(routine) { + '' + formatDate(run.started_at) + '' + '' + formatDate(run.completed_at) + '' + '' + escapeHtml(run.status) + '' - + '' + escapeHtml(run.result_summary || '-') + '' + + '' + escapeHtml(run.result_summary || '-') + + (run.job_id ? ' [view job]' : '') + + '' + '' + (run.tokens_used != null ? run.tokens_used : '-') + '' + ''; } diff --git a/src/channels/web/static/style.css b/src/channels/web/static/style.css index 7c263ab2..798505eb 100644 --- a/src/channels/web/static/style.css +++ b/src/channels/web/static/style.css @@ -2240,6 +2240,14 @@ body { color: var(--success); } +.activity-tool-error .activity-tool-icon { + color: var(--danger); +} + +.activity-tool-error summary { + color: var(--danger); +} + .activity-tool-input, .activity-tool-output { padding: 8px 10px; diff --git a/src/db/libsql/routines.rs b/src/db/libsql/routines.rs index 48028c13..3635dcb3 100644 --- a/src/db/libsql/routines.rs +++ b/src/db/libsql/routines.rs @@ -387,4 +387,19 @@ impl RoutineStore for LibSqlBackend { None => Ok(0), } } + + async fn link_routine_run_to_job( + &self, + run_id: Uuid, + job_id: Uuid, + ) -> Result<(), DatabaseError> { + let conn = self.connect().await?; + conn.execute( + "UPDATE routine_runs SET job_id = ?1 WHERE id = ?2", + params![job_id.to_string(), run_id.to_string()], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; + Ok(()) + } } diff --git a/src/db/mod.rs b/src/db/mod.rs index 86c9d568..dcf80e02 100644 --- a/src/db/mod.rs +++ b/src/db/mod.rs @@ -274,6 +274,11 @@ pub trait RoutineStore: Send + Sync { limit: i64, ) -> Result, DatabaseError>; async fn count_running_routine_runs(&self, routine_id: Uuid) -> Result; + async fn link_routine_run_to_job( + &self, + run_id: Uuid, + job_id: Uuid, + ) -> Result<(), DatabaseError>; } #[async_trait] diff --git a/src/db/postgres.rs b/src/db/postgres.rs index 9404dc7e..49c66f5b 100644 --- a/src/db/postgres.rs +++ b/src/db/postgres.rs @@ -437,6 +437,14 @@ impl RoutineStore for PgBackend { async fn count_running_routine_runs(&self, routine_id: Uuid) -> Result { self.store.count_running_routine_runs(routine_id).await } + + async fn link_routine_run_to_job( + &self, + run_id: Uuid, + job_id: Uuid, + ) -> Result<(), DatabaseError> { + self.store.link_routine_run_to_job(run_id, job_id).await + } } // ==================== ToolFailureStore ==================== diff --git a/src/error.rs b/src/error.rs index 8ff80704..c1d0072d 100644 --- a/src/error.rs +++ b/src/error.rs @@ -407,6 +407,9 @@ pub enum RoutineError { #[error("LLM call failed: {reason}")] LlmFailed { reason: String }, + #[error("Failed to dispatch full job: {reason}")] + JobDispatchFailed { reason: String }, + #[error("LLM returned empty content")] EmptyResponse, diff --git a/src/history/store.rs b/src/history/store.rs index 921b4725..f01c94d7 100644 --- a/src/history/store.rs +++ b/src/history/store.rs @@ -1167,6 +1167,21 @@ impl Store { .await?; Ok(row.get("cnt")) } + + /// Link a routine run to a dispatched job. + pub async fn link_routine_run_to_job( + &self, + run_id: Uuid, + job_id: Uuid, + ) -> Result<(), DatabaseError> { + let conn = self.conn().await?; + conn.execute( + "UPDATE routine_runs SET job_id = $1 WHERE id = $2", + &[&job_id, &run_id], + ) + .await?; + Ok(()) + } } #[cfg(feature = "postgres")]