diff --git a/codex-rs/core/src/tools/handlers/agent_jobs.rs b/codex-rs/core/src/tools/handlers/agent_jobs.rs index 783d5a71a1..a47ffa12bd 100644 --- a/codex-rs/core/src/tools/handlers/agent_jobs.rs +++ b/codex-rs/core/src/tools/handlers/agent_jobs.rs @@ -612,10 +612,7 @@ fn job_runtime_timeout(job: &codex_state::AgentJob) -> Duration { } fn started_at_from_item(item: &codex_state::AgentJobItem) -> Instant { - let now = chrono::Utc::now(); - let age = now.signed_duration_since(item.updated_at); - if let Ok(age) = age.to_std() { - let age = age.saturating_sub(Duration::from_secs(1)); + if let Some(age) = age_since_item_update(item) { Instant::now().checked_sub(age).unwrap_or_else(Instant::now) } else { Instant::now() @@ -623,12 +620,15 @@ fn started_at_from_item(item: &codex_state::AgentJobItem) -> Instant { } fn is_item_stale(item: &codex_state::AgentJobItem, runtime_timeout: Duration) -> bool { + age_since_item_update(item).is_some_and(|age| age > runtime_timeout) +} + +fn age_since_item_update(item: &codex_state::AgentJobItem) -> Option { let now = chrono::Utc::now(); - if let Ok(age) = now.signed_duration_since(item.updated_at).to_std() { - age > runtime_timeout - } else { - false - } + now.signed_duration_since(item.updated_at) + .to_std() + .ok() + .map(|age| age.saturating_sub(Duration::from_secs(1))) } fn default_output_csv_path(input_csv_path: &AbsolutePathBuf, job_id: &str) -> AbsolutePathBuf { diff --git a/codex-rs/core/src/tools/handlers/agent_jobs_tests.rs b/codex-rs/core/src/tools/handlers/agent_jobs_tests.rs index a2dbe6a480..1478a1ab70 100644 --- a/codex-rs/core/src/tools/handlers/agent_jobs_tests.rs +++ b/codex-rs/core/src/tools/handlers/agent_jobs_tests.rs @@ -1,6 +1,8 @@ use super::*; +use crate::session::tests::make_session_and_context; use pretty_assertions::assert_eq; use serde_json::json; +use std::time::Duration; #[test] fn parse_csv_supports_quotes_and_commas() { @@ -60,3 +62,121 @@ fn ensure_unique_headers_rejects_duplicates() { FunctionCallError::RespondToModel("csv header path is duplicated".to_string()) ); } + +#[test] +fn is_item_stale_allows_second_precision_grace() { + let now = chrono::Utc::now(); + let item = codex_state::AgentJobItem { + job_id: "job-1".to_string(), + item_id: "item-1".to_string(), + row_index: 0, + source_id: None, + row_json: json!({"path": "file-1"}), + status: codex_state::AgentJobItemStatus::Running, + assigned_thread_id: Some("00000000-0000-0000-0000-000000000001".to_string()), + attempt_count: 1, + result_json: None, + last_error: None, + created_at: now, + updated_at: now - chrono::Duration::milliseconds(1100), + completed_at: None, + reported_at: None, + }; + + assert!(!is_item_stale(&item, Duration::from_secs(1))); +} + +#[tokio::test] +async fn run_agent_job_loop_fails_stale_state_item_outside_active_map() -> anyhow::Result<()> { + let (mut session, turn) = make_session_and_context().await; + let codex_home = tempfile::tempdir()?; + let db = codex_state::StateRuntime::init( + codex_home.path().to_path_buf(), + "test-provider".to_string(), + ) + .await?; + session.services.state_db = Some(db.clone()); + + let job_id = "job-watchdog".to_string(); + let keepalive_item_id = "keepalive".to_string(); + let target_item_id = "target".to_string(); + let output_path = codex_home.path().join("out.csv"); + db.create_agent_job( + &codex_state::AgentJobCreateParams { + id: job_id.clone(), + name: "test-job".to_string(), + instruction: "Return {path}".to_string(), + auto_export: true, + max_runtime_seconds: Some(1), + output_schema_json: None, + input_headers: vec!["path".to_string()], + input_csv_path: codex_home.path().join("in.csv").display().to_string(), + output_csv_path: output_path.display().to_string(), + }, + &[ + codex_state::AgentJobItemCreateParams { + item_id: keepalive_item_id, + row_index: 0, + source_id: None, + row_json: json!({"path": "keepalive"}), + }, + codex_state::AgentJobItemCreateParams { + item_id: target_item_id.clone(), + row_index: 1, + source_id: None, + row_json: json!({"path": "target"}), + }, + ], + ) + .await?; + db.mark_agent_job_running(job_id.as_str()).await?; + + let session = std::sync::Arc::new(session); + let turn = std::sync::Arc::new(turn); + let runner = tokio::spawn(run_agent_job_loop( + std::sync::Arc::clone(&session), + std::sync::Arc::clone(&turn), + db.clone(), + job_id.clone(), + JobRunnerOptions { + max_concurrency: 0, + spawn_config: (*turn.config).clone(), + }, + )); + + tokio::time::sleep(STATUS_POLL_INTERVAL * 2).await; + assert!( + db.mark_agent_job_item_running_with_thread( + job_id.as_str(), + target_item_id.as_str(), + "00000000-0000-0000-0000-000000000001", + ) + .await? + ); + + let failed_item = tokio::time::timeout(Duration::from_secs(5), async { + loop { + let item = db + .get_agent_job_item(job_id.as_str(), target_item_id.as_str()) + .await? + .expect("target item should exist"); + if item.status == codex_state::AgentJobItemStatus::Failed { + break anyhow::Ok(item); + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + }) + .await??; + assert_eq!( + failed_item.last_error, + Some("worker exceeded max runtime of 1s".to_string()) + ); + + assert!( + db.mark_agent_job_cancelled(job_id.as_str(), "test complete") + .await? + ); + let runner_result = tokio::time::timeout(Duration::from_secs(2), runner).await?; + runner_result??; + Ok(()) +}