Address agent job watchdog review

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<Duration> {
     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 {

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(())
+}