diff --git a/src/agent/routine_engine.rs b/src/agent/routine_engine.rs index da22ffc1..79b86079 100644 --- a/src/agent/routine_engine.rs +++ b/src/agent/routine_engine.rs @@ -25,6 +25,7 @@ use crate::agent::routine::{ }; use crate::channels::{IncomingMessage, OutgoingResponse}; use crate::config::RoutineConfig; +use crate::context::JobState; use crate::db::Database; use crate::error::RoutineError; use crate::llm::{ChatMessage, CompletionRequest, FinishReason, LlmProvider}; @@ -180,6 +181,130 @@ impl RoutineEngine { } } + /// Sync dispatched routine runs with their linked background job status. + /// + /// Full-job routines are fire-and-forget: the routine run is created with + /// `Running` status when the job is dispatched, but the run record is never + /// updated when the background job completes or fails. This method checks + /// all `Running` routine runs that have a linked job, queries the job's + /// current state, and updates the routine run accordingly. It also sends + /// failure/success notifications that would otherwise be lost. + pub async fn sync_dispatched_runs(&self) { + let runs = match self.store.list_dispatched_routine_runs().await { + Ok(r) => r, + Err(e) => { + tracing::debug!("Failed to list dispatched routine runs: {}", e); + return; + } + }; + + for run in runs { + let Some(job_id) = run.job_id else { + continue; + }; + + // Check the linked job's current state + let job = match self.store.get_job(job_id).await { + Ok(Some(j)) => j, + Ok(None) => { + // Job was deleted — mark the routine run as failed + tracing::warn!( + run_id = %run.id, + job_id = %job_id, + "Linked job not found, marking routine run as failed" + ); + self.complete_dispatched_run( + &run, + RunStatus::Failed, + "Linked job not found (may have been deleted)", + ) + .await; + continue; + } + Err(e) => { + tracing::debug!( + run_id = %run.id, + job_id = %job_id, + "Failed to query linked job: {}", e + ); + continue; + } + }; + + // Extract the reason from the most recent state transition + let last_reason = job.transitions.last().and_then(|t| t.reason.clone()); + + // Map job state to routine run status + let (new_status, summary) = match job.state { + JobState::Completed | JobState::Submitted | JobState::Accepted => { + let summary = + last_reason.unwrap_or_else(|| "Job completed successfully".to_string()); + (RunStatus::Ok, summary) + } + JobState::Failed => { + let summary = last_reason + .unwrap_or_else(|| "Job failed (no error message recorded)".to_string()); + (RunStatus::Failed, summary) + } + JobState::Cancelled => (RunStatus::Failed, "Job was cancelled".to_string()), + // Still in progress — skip + JobState::Pending | JobState::InProgress | JobState::Stuck => continue, + }; + + tracing::info!( + run_id = %run.id, + job_id = %job_id, + status = %new_status, + "Syncing dispatched routine run with completed job" + ); + + self.complete_dispatched_run(&run, new_status, &summary) + .await; + } + } + + /// Complete a dispatched routine run and send the appropriate notification. + async fn complete_dispatched_run(&self, run: &RoutineRun, status: RunStatus, summary: &str) { + if let Err(e) = self + .store + .complete_routine_run(run.id, status, Some(summary), None) + .await + { + tracing::error!( + run_id = %run.id, + "Failed to update dispatched routine run: {}", e + ); + return; + } + + // Look up the routine to get its notify config and name + match self.store.get_routine(run.routine_id).await { + Ok(Some(routine)) => { + send_notification( + &self.notify_tx, + &routine.notify, + &routine.name, + status, + Some(summary), + None, + ) + .await; + } + Ok(None) => { + tracing::debug!( + routine_id = %run.routine_id, + "Routine not found for notification (may have been deleted)" + ); + } + Err(e) => { + tracing::debug!( + routine_id = %run.routine_id, + "Failed to look up routine for notification: {}", e + ); + } + } + } + /// Fire a routine manually (from tool call or CLI). /// /// Bypasses cooldown checks (those only apply to cron/event triggers). @@ -534,9 +659,10 @@ async fn execute_full_job( ); let summary = format!( - "Dispatched job {job_id} for full execution with tool access (max_iterations: {max_iterations})" + "Dispatched job {job_id} for full execution with tool access (max_iterations: {max_iterations}). \ + Status will be updated when the job completes." ); - Ok((RunStatus::Ok, Some(summary), None)) + Ok((RunStatus::Running, Some(summary), None)) } /// Execute a lightweight routine (single LLM call). @@ -712,6 +838,7 @@ pub fn spawn_cron_ticker( loop { ticker.tick().await; engine.check_cron_triggers().await; + engine.sync_dispatched_runs().await; } }) } @@ -756,4 +883,30 @@ mod tests { let _ = status.to_string(); } } + + #[test] + fn test_running_status_does_not_notify() { + // Running status should not trigger notifications (job still in progress) + let config = NotifyConfig { + on_success: true, + on_failure: true, + on_attention: true, + ..Default::default() + }; + + // RunStatus::Running maps to false in send_notification's match + let should_notify = match RunStatus::Running { + RunStatus::Ok => config.on_success, + RunStatus::Attention => config.on_attention, + RunStatus::Failed => config.on_failure, + RunStatus::Running => false, + }; + assert!(!should_notify); + } + + #[test] + fn test_full_job_dispatch_returns_running_status() { + // Verify the status text for Running is "running" + assert_eq!(RunStatus::Running.to_string(), "running"); + } } diff --git a/src/db/libsql/routines.rs b/src/db/libsql/routines.rs index f85ba0e3..7107619a 100644 --- a/src/db/libsql/routines.rs +++ b/src/db/libsql/routines.rs @@ -423,4 +423,29 @@ impl RoutineStore for LibSqlBackend { .map_err(|e| DatabaseError::Query(e.to_string()))?; Ok(()) } + + async fn list_dispatched_routine_runs(&self) -> Result, DatabaseError> { + let conn = self.connect().await?; + let mut rows = conn + .query( + &format!( + "SELECT {} FROM routine_runs \ + WHERE status = 'running' AND job_id IS NOT NULL", + ROUTINE_RUN_COLUMNS + ), + params![], + ) + .await + .map_err(|e| DatabaseError::Query(e.to_string()))?; + + let mut runs = Vec::new(); + while let Some(row) = rows + .next() + .await + .map_err(|e| DatabaseError::Query(e.to_string()))? + { + runs.push(row_to_routine_run_libsql(&row)?); + } + Ok(runs) + } } diff --git a/src/db/mod.rs b/src/db/mod.rs index 560d682a..f8e1d637 100644 --- a/src/db/mod.rs +++ b/src/db/mod.rs @@ -303,6 +303,10 @@ pub trait RoutineStore: Send + Sync { run_id: Uuid, job_id: Uuid, ) -> Result<(), DatabaseError>; + /// List routine runs that were dispatched as full_job (status = 'running' + /// with a linked job_id). Used by the routine engine to sync completion + /// status from the background job. + async fn list_dispatched_routine_runs(&self) -> Result, DatabaseError>; } #[async_trait] diff --git a/src/db/postgres.rs b/src/db/postgres.rs index 9dd988bc..b5ef3ff8 100644 --- a/src/db/postgres.rs +++ b/src/db/postgres.rs @@ -494,6 +494,10 @@ impl RoutineStore for PgBackend { ) -> Result<(), DatabaseError> { self.store.link_routine_run_to_job(run_id, job_id).await } + + async fn list_dispatched_routine_runs(&self) -> Result, DatabaseError> { + self.store.list_dispatched_routine_runs().await + } } // ==================== ToolFailureStore ==================== diff --git a/src/history/store.rs b/src/history/store.rs index 2a46aaea..45a6ae8f 100644 --- a/src/history/store.rs +++ b/src/history/store.rs @@ -1295,6 +1295,17 @@ impl Store { .await?; Ok(()) } + + pub async fn list_dispatched_routine_runs(&self) -> Result, DatabaseError> { + let conn = self.conn().await?; + let rows = conn + .query( + "SELECT * FROM routine_runs WHERE status = 'running' AND job_id IS NOT NULL", + &[], + ) + .await?; + rows.iter().map(row_to_routine_run).collect() + } } #[cfg(feature = "postgres")]