Adding failing unit tests

codex-rs/core/src/tools/handlers/agent_jobs_tests.rs

@@ -60,3 +60,133 @@ fn ensure_unique_headers_rejects_duplicates() {
         FunctionCallError::RespondToModel("csv header path is duplicated".to_string())
     );
 }
+
+#[tokio::test]
+async fn assigning_thread_before_prompt_delivery_does_not_mark_item_running() -> anyhow::Result<()>
+{
+    let codex_home = tempfile::tempdir()?;
+    let runtime =
+        codex_state::StateRuntime::init(codex_home.path().to_path_buf(), "test-provider".into())
+            .await?;
+    let job_id = "job-before-prompt";
+    let item_id = "row-1";
+
+    runtime
+        .create_agent_job(
+            &codex_state::AgentJobCreateParams {
+                id: job_id.to_string(),
+                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: "/tmp/input.csv".to_string(),
+                output_csv_path: "/tmp/output.csv".to_string(),
+            },
+            &[codex_state::AgentJobItemCreateParams {
+                item_id: item_id.to_string(),
+                row_index: 0,
+                source_id: None,
+                row_json: json!({"path": "file-1"}),
+            }],
+        )
+        .await?;
+    runtime.mark_agent_job_running(job_id).await?;
+
+    let marked_running = runtime
+        .mark_agent_job_item_running_with_thread(job_id, item_id, "child-thread-without-prompt")
+        .await?;
+    assert!(
+        !marked_running,
+        "assigning a child thread is not enough to mark an item running; \
+         the worker prompt must be delivered first"
+    );
+
+    let item = runtime
+        .get_agent_job_item(job_id, item_id)
+        .await?
+        .expect("job item should exist");
+    assert_eq!(item.status, codex_state::AgentJobItemStatus::Pending);
+    assert_eq!(item.assigned_thread_id, None);
+    assert_eq!(
+        runtime.get_agent_job_progress(job_id).await?,
+        codex_state::AgentJobProgress {
+            total_items: 1,
+            pending_items: 1,
+            running_items: 0,
+            completed_items: 0,
+            failed_items: 0,
+        }
+    );
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn running_item_past_max_runtime_fails_even_if_worker_never_returns() -> anyhow::Result<()> {
+    let codex_home = tempfile::tempdir()?;
+    let runtime =
+        codex_state::StateRuntime::init(codex_home.path().to_path_buf(), "test-provider".into())
+            .await?;
+    let job_id = "job-timeout";
+    let item_id = "row-1";
+
+    runtime
+        .create_agent_job(
+            &codex_state::AgentJobCreateParams {
+                id: job_id.to_string(),
+                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: "/tmp/input.csv".to_string(),
+                output_csv_path: "/tmp/output.csv".to_string(),
+            },
+            &[codex_state::AgentJobItemCreateParams {
+                item_id: item_id.to_string(),
+                row_index: 0,
+                source_id: None,
+                row_json: json!({"path": "file-1"}),
+            }],
+        )
+        .await?;
+    runtime.mark_agent_job_running(job_id).await?;
+    let marked_running = runtime
+        .mark_agent_job_item_running_with_thread(
+            job_id,
+            item_id,
+            "00000000-0000-0000-0000-000000000001",
+        )
+        .await?;
+    assert!(marked_running);
+
+    tokio::time::sleep(std::time::Duration::from_millis(1_100)).await;
+
+    let item = runtime
+        .get_agent_job_item(job_id, item_id)
+        .await?
+        .expect("job item should exist");
+    assert_eq!(item.status, codex_state::AgentJobItemStatus::Failed);
+    assert_eq!(item.assigned_thread_id, None);
+    assert!(
+        item.last_error
+            .as_deref()
+            .is_some_and(|error| error.contains("worker exceeded max runtime")),
+        "timed out running item should record a max-runtime failure"
+    );
+    assert_eq!(
+        runtime.get_agent_job_progress(job_id).await?,
+        codex_state::AgentJobProgress {
+            total_items: 1,
+            pending_items: 0,
+            running_items: 0,
+            completed_items: 0,
+            failed_items: 1,
+        }
+    );
+
+    Ok(())
+}

codex-rs/core/tests/suite/agent_jobs.rs

@@ -15,6 +15,7 @@ use std::sync::Arc;
 use std::sync::atomic::AtomicBool;
 use std::sync::atomic::AtomicUsize;
 use std::sync::atomic::Ordering;
+use std::time::Duration;
 use wiremock::Mock;
 use wiremock::Respond;
 use wiremock::ResponseTemplate;
@@ -53,6 +54,20 @@ impl StopAfterFirstResponder {
     }
 }
 
+struct NeverTerminalWorkerResponder {
+    spawn_args_json: String,
+    seen_main: AtomicBool,
+}
+
+impl NeverTerminalWorkerResponder {
+    fn new(spawn_args_json: String) -> Self {
+        Self {
+            spawn_args_json,
+            seen_main: AtomicBool::new(false),
+        }
+    }
+}
+
 impl Respond for StopAfterFirstResponder {
     fn respond(&self, request: &wiremock::Request) -> ResponseTemplate {
         let body_bytes = decode_body_bytes(request);
@@ -100,6 +115,38 @@ impl Respond for StopAfterFirstResponder {
     }
 }
 
+impl Respond for NeverTerminalWorkerResponder {
+    fn respond(&self, request: &wiremock::Request) -> ResponseTemplate {
+        let body_bytes = decode_body_bytes(request);
+        let body: Value = serde_json::from_slice(&body_bytes).unwrap_or(Value::Null);
+
+        if has_function_call_output(&body) {
+            return sse_response(sse(vec![
+                ev_response_created("resp-tool"),
+                ev_completed("resp-tool"),
+            ]));
+        }
+
+        if extract_job_and_item(&body).is_some() {
+            return sse_response(sse(vec![ev_response_created("resp-worker")]))
+                .set_delay(Duration::from_secs(60));
+        }
+
+        if !self.seen_main.swap(true, Ordering::SeqCst) {
+            return sse_response(sse(vec![
+                ev_response_created("resp-main"),
+                ev_function_call("call-spawn", "spawn_agents_on_csv", &self.spawn_args_json),
+                ev_completed("resp-main"),
+            ]));
+        }
+
+        sse_response(sse(vec![
+            ev_response_created("resp-default"),
+            ev_completed("resp-default"),
+        ]))
+    }
+}
+
 impl Respond for AgentJobsResponder {
     fn respond(&self, request: &wiremock::Request) -> ResponseTemplate {
         let body_bytes = decode_body_bytes(request);
@@ -326,6 +373,53 @@ async fn spawn_agents_on_csv_runs_and_exports() -> Result<()> {
     Ok(())
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn spawn_agents_on_csv_parent_does_not_wait_forever_for_non_terminal_item() -> Result<()> {
+    let server = start_mock_server().await;
+    let mut builder = test_codex().with_config(|config| {
+        config
+            .features
+            .enable(Feature::SpawnCsv)
+            .expect("test config should allow feature update");
+        config
+            .features
+            .enable(Feature::Sqlite)
+            .expect("test config should allow feature update");
+    });
+    let test = builder.build(&server).await?;
+
+    let input_path = test.cwd_path().join("agent_jobs_non_terminal.csv");
+    let output_path = test.cwd_path().join("agent_jobs_non_terminal_out.csv");
+    fs::write(&input_path, "path\nfile-1\n")?;
+
+    let args = json!({
+        "csv_path": input_path.display().to_string(),
+        "instruction": "Return {path}",
+        "output_csv_path": output_path.display().to_string(),
+    });
+    let args_json = serde_json::to_string(&args)?;
+
+    let responder = NeverTerminalWorkerResponder::new(args_json);
+    Mock::given(method("POST"))
+        .and(path_regex(".*/responses$"))
+        .respond_with(responder)
+        .mount(&server)
+        .await;
+
+    let completed = tokio::time::timeout(Duration::from_secs(2), test.submit_turn("run job")).await;
+    assert!(
+        completed.is_ok(),
+        "parent spawn_agents_on_csv call should not wait forever when a worker item never reaches a terminal state"
+    );
+    completed??;
+    assert!(
+        output_path.exists(),
+        "parent should finalize and export a CSV snapshot even when an item fails to reach a terminal state"
+    );
+
+    Ok(())
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn spawn_agents_on_csv_dedupes_item_ids() -> Result<()> {
     let server = start_mock_server().await;