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..b1bdb07e16 100644 --- a/codex-rs/core/src/tools/handlers/agent_jobs_tests.rs +++ b/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(()) +} diff --git a/codex-rs/core/tests/suite/agent_jobs.rs b/codex-rs/core/tests/suite/agent_jobs.rs index b275b1878a..39e25d1a63 100644 --- a/codex-rs/core/tests/suite/agent_jobs.rs +++ b/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;