mirror of
https://github.com/outbackdingo/optimclaw.git
synced 2026-08-25 14:53:34 +00:00
fix: full_job routine runs stay running until linked job completion (#1374)
* 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) <[email protected]> * 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) <[email protected]> * 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) <[email protected]> * 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) <[email protected]> --------- Co-authored-by: Claude Opus 4.6 (1M context) <[email protected]>
This commit is contained in:
Henry Park
co-authored by
Claude Opus 4.6
Claude Opus 4.6
parent
6831bb4d7b
commit
14abd60917
+337
-11
@@ -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<ToolRegistry>,
|
||||
/// Safety layer for tool output sanitization.
|
||||
safety: Arc<SafetyLayer>,
|
||||
/// 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<Utc>,
|
||||
}
|
||||
|
||||
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
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<Vec<RoutineRun>, 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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<Vec<RoutineRun>, DatabaseError>;
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
|
||||
@@ -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<Vec<RoutineRun>, DatabaseError> {
|
||||
self.store.list_dispatched_routine_runs().await
|
||||
}
|
||||
}
|
||||
|
||||
// ==================== ToolFailureStore ====================
|
||||
|
||||
@@ -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<Vec<RoutineRun>, 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")]
|
||||
|
||||
@@ -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<dyn Database>, 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<dyn Database> = 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<Uuid>) -> 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));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user