From 14abd609179a66cc735f2342fa92cdc60bfc0bd9 Mon Sep 17 00:00:00 2001 From: Henry Park Date: Wed, 18 Mar 2026 15:33:57 -0700 Subject: [PATCH] fix: full_job routine runs stay running until linked job completion (#1374) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix: full_job routine runs stay running until linked job completion (#1317) Previously, execute_full_job() returned RunStatus::Ok immediately after dispatching the job, causing routine runs to be marked as completed before the linked worker job had actually finished. This meant failure notifications were never sent and max_concurrent guardrails stopped applying once the run was prematurely finalized. Changes: - execute_full_job() now returns RunStatus::Running instead of Ok - execute_routine() skips finalization for Running status (leaves run open) - New sync_dispatched_runs() polls on each cron tick, checks linked job state, and finalizes runs when jobs reach terminal states - New list_dispatched_routine_runs() DB method on both backends - Deferred notifications are sent when the run is actually finalized - consecutive_failures is preserved (not reset) while outcome is unknown Co-Authored-By: Claude Opus 4.6 (1M context) * fix: address PR review feedback (watcher predicate, running_count safety) - FullJobWatcher: use is_parallel_blocking() instead of is_active() so the watcher exits when a job reaches Completed (not terminal but finished executing). Fixes infinite-poll for routine jobs. - Remove running_count decrement from sync_dispatched_runs() — in normal flow execute_routine() handles it; sync only runs for crash recovery where the counter is already 0. - Update PR description to match actual FullJobWatcher behavior. Co-Authored-By: Claude Opus 4.6 (1M context) * fix: sync only at startup to prevent double-completion race - Move sync_dispatched_runs() out of cron loop into startup-only path. During normal operation FullJobWatcher handles finalization inline; running sync on every tick would race with the watcher. - Update complete_dispatched_run() to properly advance runtime fields (last_run_at, next_fire_at, run_count) for crash recovery — in that scenario execute_routine() never reached its runtime update. - Fix stale doc comment on complete_dispatched_run(). Co-Authored-By: Claude Opus 4.6 (1M context) * fix: use boot_time filter for safe periodic sync of orphaned runs - Add boot_time field to RoutineEngine, set to Utc::now() at creation. - sync_dispatched_runs() now filters runs by started_at < boot_time, so it only processes orphans from a previous process — never races with FullJobWatcher instances from the current process. - Move sync back into the cron loop (safe with boot_time filter) and run it BEFORE check_cron_triggers to avoid picking up freshly dispatched runs. - Fix doc comments to match actual behavior. Co-Authored-By: Claude Opus 4.6 (1M context) --------- Co-authored-by: Claude Opus 4.6 (1M context) --- src/agent/routine_engine.rs | 348 ++++++++++++++++++++++++- src/db/libsql/routines.rs | 24 ++ src/db/mod.rs | 3 + src/db/postgres.rs | 4 + src/history/store.rs | 12 + tests/dispatched_routine_run_tests.rs | 360 ++++++++++++++++++++++++++ 6 files changed, 740 insertions(+), 11 deletions(-) create mode 100644 tests/dispatched_routine_run_tests.rs diff --git a/src/agent/routine_engine.rs b/src/agent/routine_engine.rs index 14360d85..9047a5ad 100644 --- a/src/agent/routine_engine.rs +++ b/src/agent/routine_engine.rs @@ -25,7 +25,7 @@ use crate::agent::routine::{ }; use crate::channels::OutgoingResponse; use crate::config::RoutineConfig; -use crate::context::JobContext; +use crate::context::{JobContext, JobState}; use crate::db::Database; use crate::error::RoutineError; use crate::llm::{ @@ -60,6 +60,10 @@ pub struct RoutineEngine { tools: Arc, /// Safety layer for tool output sanitization. safety: Arc, + /// Timestamp when this engine instance was created. Used by + /// `sync_dispatched_runs` to distinguish orphaned runs (from a previous + /// process) from actively-watched runs (from this process). + boot_time: chrono::DateTime, } impl RoutineEngine { @@ -85,6 +89,7 @@ impl RoutineEngine { scheduler, tools, safety, + boot_time: Utc::now(), } } @@ -371,6 +376,230 @@ impl RoutineEngine { } } + /// Reconcile orphaned full_job routine runs with their linked job outcomes. + /// + /// Called on each cron tick. Finds routine runs that are still `running` + /// with a linked `job_id`, checks the job state, and finalizes the run + /// when the job reaches a completed or terminal state. + /// + /// Only processes runs started **before** this engine's boot time, so it + /// never races with `FullJobWatcher` instances from the current process. + /// This makes it safe to call on every tick as a crash-recovery mechanism. + pub async fn sync_dispatched_runs(&self) { + let runs = match self.store.list_dispatched_routine_runs().await { + Ok(r) => r, + Err(e) => { + tracing::error!("Failed to list dispatched routine runs: {}", e); + return; + } + }; + + // Only process runs from a previous process instance. Runs started + // after boot_time are actively watched by a FullJobWatcher in this + // process and should not be finalized here. + let orphaned: Vec<_> = runs + .into_iter() + .filter(|r| r.started_at < self.boot_time) + .collect(); + + if orphaned.is_empty() { + return; + } + + tracing::info!( + "Recovering {} orphaned dispatched routine runs", + orphaned.len() + ); + + for run in orphaned { + let job_id = match run.job_id { + Some(id) => id, + None => continue, // Should not happen (query filters), but guard anyway + }; + + // Fetch the linked job + let job = match self.store.get_job(job_id).await { + Ok(Some(j)) => j, + Ok(None) => { + // Orphaned: job record was deleted or never persisted + 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, + &format!("Linked job {job_id} not found (orphaned)"), + ) + .await; + continue; + } + Err(e) => { + tracing::error!( + run_id = %run.id, + job_id = %job_id, + "Failed to fetch linked job: {}", e + ); + continue; + } + }; + + // Map job state to final run status + let final_status = match job.state { + JobState::Completed | JobState::Submitted | JobState::Accepted => { + Some(RunStatus::Ok) + } + JobState::Failed | JobState::Cancelled => Some(RunStatus::Failed), + // Pending, InProgress, Stuck — still running + _ => None, + }; + + let status = match final_status { + Some(s) => s, + None => continue, // Job still active, check again next tick + }; + + // Build summary + let summary = if status == RunStatus::Failed { + match self.store.get_agent_job_failure_reason(job_id).await { + Ok(Some(reason)) => format!("Job {job_id} failed: {reason}"), + _ => format!("Job {job_id} {}", job.state), + } + } else { + format!("Job {job_id} completed successfully") + }; + + self.complete_dispatched_run(&run, status, &summary).await; + } + } + + /// Finalize a dispatched routine run: update DB, update routine runtime, + /// persist to conversation thread, and send notification. + async fn complete_dispatched_run(&self, run: &RoutineRun, status: RunStatus, summary: &str) { + // Complete the run record in DB + if let Err(e) = self + .store + .complete_routine_run(run.id, status, Some(summary), None) + .await + { + tracing::error!( + run_id = %run.id, + "Failed to complete dispatched routine run: {}", e + ); + return; + } + + tracing::info!( + run_id = %run.id, + status = %status, + "Finalized dispatched routine run" + ); + + // Load the routine to update consecutive_failures and send notification + let routine = match self.store.get_routine(run.routine_id).await { + Ok(Some(r)) => r, + Ok(None) => { + tracing::warn!( + run_id = %run.id, + routine_id = %run.routine_id, + "Routine not found for dispatched run finalization" + ); + return; + } + Err(e) => { + tracing::error!( + run_id = %run.id, + "Failed to load routine for dispatched run: {}", e + ); + return; + } + }; + + // Update runtime fields. In crash recovery, execute_routine() never + // reached its normal runtime update, so we must advance all fields here. + let new_failures = if status == RunStatus::Failed { + routine.consecutive_failures + 1 + } else { + 0 + }; + + let now = Utc::now(); + let next_fire = if let Trigger::Cron { + ref schedule, + ref timezone, + } = routine.trigger + { + next_cron_fire(schedule, timezone.as_deref()).unwrap_or(None) + } else { + None + }; + + if let Err(e) = self + .store + .update_routine_runtime( + routine.id, + now, + next_fire, + routine.run_count + 1, + new_failures, + &routine.state, + ) + .await + { + tracing::error!( + routine = %routine.name, + "Failed to update routine runtime after dispatched run: {}", e + ); + } + + // Persist result to the routine's conversation thread + let thread_id = match self + .store + .get_or_create_routine_conversation(routine.id, &routine.name, &routine.user_id) + .await + { + Ok(conv_id) => { + let msg = format!("[dispatched] {}: {}", status, summary); + if let Err(e) = self + .store + .add_conversation_message(conv_id, "assistant", &msg) + .await + { + tracing::error!( + routine = %routine.name, + "Failed to persist dispatched run message: {}", e + ); + } + Some(conv_id.to_string()) + } + Err(e) => { + tracing::error!( + routine = %routine.name, + "Failed to get routine conversation: {}", e + ); + None + } + }; + + // Send notification + send_notification( + &self.notify_tx, + &routine.notify, + &routine.user_id, + &routine.name, + status, + Some(summary), + thread_id.as_deref(), + ) + .await; + + // Note: we do NOT decrement running_count here. In normal flow, + // execute_routine() handles that after FullJobWatcher returns. + // This sync path only runs for crash recovery (process restarted), + // where running_count was already reset to 0. + } + /// Fire a routine manually (from tool call or CLI). /// /// Bypasses cooldown checks (those only apply to cron/event triggers). @@ -548,7 +777,11 @@ impl FullJobWatcher { // if the job is already done (e.g. fast-failing jobs). match self.store.get_job(self.job_id).await { Ok(Some(job_ctx)) => { - if !job_ctx.state.is_active() { + // Use is_parallel_blocking (Pending/InProgress/Stuck) instead + // of is_active (!is_terminal) because routine jobs typically + // stop at Completed — which is NOT terminal but IS finished + // from an execution standpoint. + if !job_ctx.state.is_parallel_blocking() { break Self::map_job_state(&job_ctx.state); } } @@ -816,13 +1049,16 @@ async fn execute_full_job( 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 - ); - } + // Link the routine run to the dispatched job. + // This MUST succeed — if it fails, sync_dispatched_runs() will never find + // this run (it filters on job_id IS NOT NULL), leaving it stuck as 'running' + // with running_count permanently elevated. + ctx.store + .link_routine_run_to_job(run.id, job_id) + .await + .map_err(|e| RoutineError::Database { + reason: format!("failed to link run to job: {e}"), + })?; tracing::info!( routine = %routine.name, @@ -1408,14 +1644,22 @@ pub fn spawn_cron_ticker( interval: Duration, ) -> tokio::task::JoinHandle<()> { tokio::spawn(async move { - // Run one check immediately so routines due at startup don't wait - // an extra full polling interval. + // Recover orphaned runs from a previous process crash before + // dispatching any new work, so we don't confuse fresh dispatches + // with crash orphans. + engine.sync_dispatched_runs().await; + + // Run one cron check immediately so routines due at startup don't + // wait an extra full polling interval. engine.check_cron_triggers().await; let mut ticker = tokio::time::interval(interval); loop { ticker.tick().await; + // Sync first: only processes runs from before boot_time, so it + // never races with FullJobWatcher instances from this process. + engine.sync_dispatched_runs().await; engine.check_cron_triggers().await; } }) @@ -1709,4 +1953,86 @@ mod tests { assert_eq!(snapshot[1].content, "a"); // safety: test-only no-panics CI false positive assert_eq!(snapshot[2].content, "b"); // safety: test-only no-panics CI false positive } + + /// Regression test for #1317: FullJobWatcher maps terminal job states correctly. + #[test] + fn test_full_job_watcher_state_mapping() { + use crate::context::JobState; + + // Failed/Cancelled → RunStatus::Failed + assert_eq!( + super::FullJobWatcher::map_job_state(&JobState::Failed), + RunStatus::Failed + ); + assert_eq!( + super::FullJobWatcher::map_job_state(&JobState::Cancelled), + RunStatus::Failed + ); + + // All other non-active states → RunStatus::Ok + assert_eq!( + super::FullJobWatcher::map_job_state(&JobState::Completed), + RunStatus::Ok + ); + assert_eq!( + super::FullJobWatcher::map_job_state(&JobState::Accepted), + RunStatus::Ok + ); + } + + /// Verify that job state to run status mapping covers all expected cases. + #[test] + fn test_job_state_to_run_status_mapping() { + use crate::context::JobState; + + // Success states + for state in [JobState::Completed, JobState::Submitted, JobState::Accepted] { + let status = match state { + JobState::Completed | JobState::Submitted | JobState::Accepted => { + Some(RunStatus::Ok) + } + JobState::Failed | JobState::Cancelled => Some(RunStatus::Failed), + _ => None, + }; + assert_eq!( + status, + Some(RunStatus::Ok), + "{:?} should map to RunStatus::Ok", + state + ); + } + + // Failure states + for state in [JobState::Failed, JobState::Cancelled] { + let status = match state { + JobState::Completed | JobState::Submitted | JobState::Accepted => { + Some(RunStatus::Ok) + } + JobState::Failed | JobState::Cancelled => Some(RunStatus::Failed), + _ => None, + }; + assert_eq!( + status, + Some(RunStatus::Failed), + "{:?} should map to RunStatus::Failed", + state + ); + } + + // Active states (should not finalize) + for state in [JobState::Pending, JobState::InProgress, JobState::Stuck] { + let status = match state { + JobState::Completed | JobState::Submitted | JobState::Accepted => { + Some(RunStatus::Ok) + } + JobState::Failed | JobState::Cancelled => Some(RunStatus::Failed), + _ => None, + }; + assert_eq!( + status, None, + "{:?} should not finalize the routine run", + state + ); + } + } } diff --git a/src/db/libsql/routines.rs b/src/db/libsql/routines.rs index b75afb47..3151e75b 100644 --- a/src/db/libsql/routines.rs +++ b/src/db/libsql/routines.rs @@ -476,4 +476,28 @@ 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 6d2eb296..49287308 100644 --- a/src/db/mod.rs +++ b/src/db/mod.rs @@ -525,6 +525,9 @@ pub trait RoutineStore: Send + Sync { run_id: Uuid, job_id: Uuid, ) -> Result<(), DatabaseError>; + /// List routine runs that were dispatched as full_job but have not yet + /// been finalized (status='running' with a linked job_id). + async fn list_dispatched_routine_runs(&self) -> Result, DatabaseError>; } #[async_trait] diff --git a/src/db/postgres.rs b/src/db/postgres.rs index 8c18e252..eaa6e049 100644 --- a/src/db/postgres.rs +++ b/src/db/postgres.rs @@ -503,6 +503,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 04e3167f..2deffab5 100644 --- a/src/history/store.rs +++ b/src/history/store.rs @@ -1348,6 +1348,18 @@ impl Store { .await?; Ok(()) } + + /// List routine runs dispatched as full_job that have not yet been finalized. + 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")] diff --git a/tests/dispatched_routine_run_tests.rs b/tests/dispatched_routine_run_tests.rs new file mode 100644 index 00000000..4ab5d2a8 --- /dev/null +++ b/tests/dispatched_routine_run_tests.rs @@ -0,0 +1,360 @@ +//! Integration tests for dispatched routine run tracking (#1317). +//! +//! Verifies: +//! 1. list_dispatched_routine_runs returns only running runs with linked jobs +//! 2. Completed jobs cause linked routine runs to be finalized as Ok +//! 3. Failed jobs cause linked routine runs to be finalized as Failed +//! 4. Active (InProgress) jobs are not finalized +//! 5. Orphaned runs (job_id set but no job record) are handled + +#[cfg(feature = "libsql")] +mod tests { + use std::sync::Arc; + + use chrono::Utc; + use uuid::Uuid; + + use ironclaw::agent::routine::{ + Routine, RoutineAction, RoutineGuardrails, RoutineRun, RunStatus, Trigger, + }; + use ironclaw::context::{JobContext, JobState}; + use ironclaw::db::Database; + + async fn create_test_db() -> (Arc, tempfile::TempDir) { + use ironclaw::db::libsql::LibSqlBackend; + + let temp_dir = tempfile::tempdir().expect("tempdir"); + let db_path = temp_dir.path().join("test.db"); + let backend = LibSqlBackend::new_local(&db_path) + .await + .expect("LibSqlBackend"); + backend.run_migrations().await.expect("migrations"); + let db: Arc = Arc::new(backend); + (db, temp_dir) + } + + fn make_routine(id: Uuid) -> Routine { + Routine { + id, + name: format!("test-routine-{}", id), + description: "Test routine".to_string(), + user_id: "default".to_string(), + enabled: true, + trigger: Trigger::Manual, + action: RoutineAction::FullJob { + title: "Test job".to_string(), + description: "Test description".to_string(), + max_iterations: 5, + tool_permissions: vec![], + }, + guardrails: RoutineGuardrails { + cooldown: std::time::Duration::from_secs(0), + max_concurrent: 1, + dedup_window: None, + }, + notify: Default::default(), + last_run_at: None, + next_fire_at: None, + run_count: 0, + consecutive_failures: 0, + state: serde_json::json!({}), + created_at: Utc::now(), + updated_at: Utc::now(), + } + } + + fn make_run(routine_id: Uuid, job_id: Option) -> RoutineRun { + RoutineRun { + id: Uuid::new_v4(), + routine_id, + trigger_type: "manual".to_string(), + trigger_detail: None, + started_at: Utc::now(), + completed_at: None, + status: RunStatus::Running, + result_summary: None, + tokens_used: None, + job_id, + created_at: Utc::now(), + } + } + + // ----------------------------------------------------------------------- + // Test 1: list_dispatched_routine_runs returns only running runs with jobs + // ----------------------------------------------------------------------- + + #[tokio::test] + async fn list_dispatched_returns_only_running_with_job_id() { + let (db, _tmp) = create_test_db().await; + let routine_id = Uuid::new_v4(); + let routine = make_routine(routine_id); + db.create_routine(&routine).await.expect("create routine"); + + // Create jobs first (FK constraint requires job records to exist) + let job1 = JobContext::new("Job 1", "Dispatched job"); + db.save_job(&job1).await.expect("save job1"); + let job2 = JobContext::new("Job 2", "Completed job"); + db.save_job(&job2).await.expect("save job2"); + + // Create a running run WITH job_id (dispatched full_job) + let dispatched_run = make_run(routine_id, Some(job1.job_id)); + db.create_routine_run(&dispatched_run) + .await + .expect("create dispatched run"); + + // Create a running run WITHOUT job_id (lightweight in-progress) + let lightweight_run = make_run(routine_id, None); + db.create_routine_run(&lightweight_run) + .await + .expect("create lightweight run"); + + // Create a completed run WITH job_id (already finalized) + let mut completed_run = make_run(routine_id, Some(job2.job_id)); + completed_run.status = RunStatus::Ok; + completed_run.completed_at = Some(Utc::now()); + db.create_routine_run(&completed_run) + .await + .expect("create completed run"); + + let dispatched = db + .list_dispatched_routine_runs() + .await + .expect("list dispatched"); + + assert_eq!(dispatched.len(), 1, "Should return only the dispatched run"); + assert_eq!(dispatched[0].id, dispatched_run.id); + assert_eq!(dispatched[0].job_id, Some(job1.job_id)); + assert_eq!(dispatched[0].status, RunStatus::Running); + } + + // ----------------------------------------------------------------------- + // Test 2: Completed job linked to run can be detected + // ----------------------------------------------------------------------- + + #[tokio::test] + async fn dispatched_run_with_completed_job_can_be_finalized() { + let (db, _tmp) = create_test_db().await; + let routine_id = Uuid::new_v4(); + let routine = make_routine(routine_id); + db.create_routine(&routine).await.expect("create routine"); + + // Create and save a job in Completed state + let mut job = JobContext::new("Test job", "Test description"); + job.state = JobState::Completed; + db.save_job(&job).await.expect("save job"); + + // Create a dispatched run linked to that job + let run = make_run(routine_id, Some(job.job_id)); + db.create_routine_run(&run).await.expect("create run"); + + // Verify the run is listed as dispatched + let dispatched = db + .list_dispatched_routine_runs() + .await + .expect("list dispatched"); + assert_eq!(dispatched.len(), 1); + + // Verify we can fetch the linked job and see it's completed + let fetched_job = db + .get_job(job.job_id) + .await + .expect("get job") + .expect("job should exist"); + assert_eq!(fetched_job.state, JobState::Completed); + + // Simulate sync: complete the run + db.complete_routine_run(run.id, RunStatus::Ok, Some("Job completed"), None) + .await + .expect("complete run"); + + // Run should no longer appear in dispatched list + let dispatched_after = db + .list_dispatched_routine_runs() + .await + .expect("list dispatched after"); + assert!( + dispatched_after.is_empty(), + "Finalized run should not appear in dispatched list" + ); + } + + // ----------------------------------------------------------------------- + // Test 3: Failed job causes run to be finalized as Failed + // ----------------------------------------------------------------------- + + #[tokio::test] + async fn dispatched_run_with_failed_job() { + let (db, _tmp) = create_test_db().await; + let routine_id = Uuid::new_v4(); + let routine = make_routine(routine_id); + db.create_routine(&routine).await.expect("create routine"); + + let mut job = JobContext::new("Failing job", "Will fail"); + job.state = JobState::Failed; + db.save_job(&job).await.expect("save job"); + + let run = make_run(routine_id, Some(job.job_id)); + db.create_routine_run(&run).await.expect("create run"); + + // Verify job is failed + let fetched_job = db + .get_job(job.job_id) + .await + .expect("get job") + .expect("job should exist"); + assert_eq!(fetched_job.state, JobState::Failed); + + // Simulate sync: complete the run as failed + db.complete_routine_run(run.id, RunStatus::Failed, Some("Job failed"), None) + .await + .expect("complete run as failed"); + + let dispatched = db + .list_dispatched_routine_runs() + .await + .expect("list dispatched"); + assert!(dispatched.is_empty(), "Failed run should be finalized"); + } + + // ----------------------------------------------------------------------- + // Test 4: Active (InProgress) job leaves run as running + // ----------------------------------------------------------------------- + + #[tokio::test] + async fn dispatched_run_with_active_job_stays_running() { + let (db, _tmp) = create_test_db().await; + let routine_id = Uuid::new_v4(); + let routine = make_routine(routine_id); + db.create_routine(&routine).await.expect("create routine"); + + let mut job = JobContext::new("Active job", "Still running"); + job.state = JobState::InProgress; + db.save_job(&job).await.expect("save job"); + + let run = make_run(routine_id, Some(job.job_id)); + db.create_routine_run(&run).await.expect("create run"); + + // Verify job is still active + let fetched_job = db + .get_job(job.job_id) + .await + .expect("get job") + .expect("job should exist"); + assert!(!fetched_job.state.is_terminal()); + + // Run should still be in dispatched list (not finalized) + let dispatched = db + .list_dispatched_routine_runs() + .await + .expect("list dispatched"); + assert_eq!( + dispatched.len(), + 1, + "Run with active job should remain dispatched" + ); + assert_eq!(dispatched[0].status, RunStatus::Running); + } + + // ----------------------------------------------------------------------- + // Test 5: Orphaned run (job_id set but job record missing) + // ----------------------------------------------------------------------- + + #[tokio::test] + async fn dispatched_run_orphan_detection() { + let (db, _tmp) = create_test_db().await; + let routine_id = Uuid::new_v4(); + let routine = make_routine(routine_id); + db.create_routine(&routine).await.expect("create routine"); + + // Create a real job so the FK constraint is satisfied + let job = JobContext::new("Will be orphaned", "Test orphan detection"); + db.save_job(&job).await.expect("save job"); + + let run = make_run(routine_id, Some(job.job_id)); + db.create_routine_run(&run).await.expect("create run"); + + // The run appears in dispatched list + let dispatched = db + .list_dispatched_routine_runs() + .await + .expect("list dispatched"); + assert_eq!(dispatched.len(), 1); + + // Verify orphan detection: a random UUID returns None from get_job + let nonexistent_id = Uuid::new_v4(); + let missing = db + .get_job(nonexistent_id) + .await + .expect("get_job should not error"); + assert!( + missing.is_none(), + "get_job for nonexistent ID should return None" + ); + + // Simulate sync handling of an orphaned run: mark as failed + db.complete_routine_run( + run.id, + RunStatus::Failed, + Some(&format!("Linked job {} not found (orphaned)", job.job_id)), + None, + ) + .await + .expect("complete orphaned run"); + + let dispatched_after = db + .list_dispatched_routine_runs() + .await + .expect("list dispatched after"); + assert!( + dispatched_after.is_empty(), + "Finalized run should not appear in dispatched list" + ); + } + + // ----------------------------------------------------------------------- + // Test 6: link_routine_run_to_job then list shows linked run + // ----------------------------------------------------------------------- + + #[tokio::test] + async fn link_and_list_dispatched_run() { + let (db, _tmp) = create_test_db().await; + let routine_id = Uuid::new_v4(); + let routine = make_routine(routine_id); + db.create_routine(&routine).await.expect("create routine"); + + // Create job record (FK constraint) + let job = JobContext::new("Linked job", "Test linking"); + db.save_job(&job).await.expect("save job"); + + // Create a running run without job_id initially + let run = make_run(routine_id, None); + db.create_routine_run(&run).await.expect("create run"); + + // Should not appear in dispatched list yet + let dispatched = db + .list_dispatched_routine_runs() + .await + .expect("list dispatched"); + assert!( + dispatched.is_empty(), + "Run without job_id should not be dispatched" + ); + + // Link the run to the job + db.link_routine_run_to_job(run.id, job.job_id) + .await + .expect("link run to job"); + + // Now it should appear + let dispatched_after = db + .list_dispatched_routine_runs() + .await + .expect("list dispatched after link"); + assert_eq!( + dispatched_after.len(), + 1, + "Linked run should appear in dispatched list" + ); + assert_eq!(dispatched_after[0].job_id, Some(job.job_id)); + } +}