diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 8669268591..5255bacf7d 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -2685,7 +2685,6 @@ dependencies = [ "codex-web-search-extension", "codex-windows-sandbox", "core_test_support", - "csv", "ctor 0.6.3", "dirs", "dunce", diff --git a/codex-rs/Cargo.toml b/codex-rs/Cargo.toml index f43faa319e..2fc037aeba 100644 --- a/codex-rs/Cargo.toml +++ b/codex-rs/Cargo.toml @@ -299,7 +299,6 @@ constant_time_eq = "0.3.1" crossbeam-channel = "0.5.15" crypto_box = { version = "0.9.1", features = ["seal"] } crossterm = "0.28.1" -csv = "1.3.1" ctor = "0.6.3" deno_core_icudata = "0.77.0" derive_more = "2" diff --git a/codex-rs/app-server/tests/suite/v2/thread_list.rs b/codex-rs/app-server/tests/suite/v2/thread_list.rs index 8176118ecf..bfc076fe52 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_list.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_list.rs @@ -1017,7 +1017,7 @@ async fn thread_list_relation_filters_read_spawn_graph_from_state_db() -> Result ( older_child_id, "2025-02-01T10:00:00Z", - CoreSessionSource::SubAgent(SubAgentSource::Other("agent_job:job-1".to_string())), + CoreSessionSource::SubAgent(SubAgentSource::Other("custom:worker-1".to_string())), "other_provider", ), ( @@ -1029,7 +1029,7 @@ async fn thread_list_relation_filters_read_spawn_graph_from_state_db() -> Result ( grandchild_id, "2025-02-01T12:00:00Z", - CoreSessionSource::SubAgent(SubAgentSource::Other("agent_job:job-2".to_string())), + CoreSessionSource::SubAgent(SubAgentSource::Other("custom:worker-2".to_string())), "mock_provider", ), ] { diff --git a/codex-rs/cli/src/main.rs b/codex-rs/cli/src/main.rs index 433c317e0a..baaa57ce0b 100644 --- a/codex-rs/cli/src/main.rs +++ b/codex-rs/cli/src/main.rs @@ -4039,6 +4039,16 @@ mod tests { ); } + #[test] + fn feature_toggles_accept_removed_enable_fanout_flag() { + let toggles = FeatureToggles { + enable: vec!["enable_fanout".to_string()], + disable: Vec::new(), + }; + let overrides = toggles.to_overrides().expect("valid features"); + assert_eq!(overrides, vec!["features.enable_fanout=true".to_string(),]); + } + #[test] fn feature_toggles_unknown_feature_errors() { let toggles = FeatureToggles { diff --git a/codex-rs/config/src/config_toml.rs b/codex-rs/config/src/config_toml.rs index 62b5606bd3..40a027f33d 100644 --- a/codex-rs/config/src/config_toml.rs +++ b/codex-rs/config/src/config_toml.rs @@ -692,8 +692,8 @@ pub struct AgentsToml { pub default_subagent_model: Option, /// Default reasoning effort for spawned subagents when the spawn call does not select one. pub default_subagent_reasoning_effort: Option, - /// Default maximum runtime in seconds for agent job workers. - #[schemars(range(min = 1))] + /// Removed agent-job setting retained as a no-op for compatibility. + #[schemars(skip)] pub job_max_runtime_seconds: Option, /// Whether to record a model-visible message when an agent turn is interrupted. /// Defaults to true. diff --git a/codex-rs/core/Cargo.toml b/codex-rs/core/Cargo.toml index 88f5369b7c..f3767b57c2 100644 --- a/codex-rs/core/Cargo.toml +++ b/codex-rs/core/Cargo.toml @@ -79,7 +79,6 @@ codex-utils-pty = { workspace = true } codex-utils-string = { workspace = true } codex-utils-stream-parser = { workspace = true } codex-windows-sandbox = { package = "codex-windows-sandbox", path = "../windows-sandbox-rs" } -csv = { workspace = true } dirs = { workspace = true } dunce = { workspace = true } eventsource-stream = { workspace = true } diff --git a/codex-rs/core/config.schema.json b/codex-rs/core/config.schema.json index cd9d5507d6..4b3549744f 100644 --- a/codex-rs/core/config.schema.json +++ b/codex-rs/core/config.schema.json @@ -56,12 +56,6 @@ "description": "Whether to record a model-visible message when an agent turn is interrupted. Defaults to true.", "type": "boolean" }, - "job_max_runtime_seconds": { - "description": "Default maximum runtime in seconds for agent job workers.", - "format": "uint64", - "minimum": 1.0, - "type": "integer" - }, "max_concurrent_threads_per_session": { "description": "Maximum number of spawned agent threads that can be open concurrently per session. When unset, the selected multi-agent backend uses its default.", "format": "uint", diff --git a/codex-rs/core/src/config/config_tests.rs b/codex-rs/core/src/config/config_tests.rs index ac021f2fe6..a52838b2ac 100644 --- a/codex-rs/core/src/config/config_tests.rs +++ b/codex-rs/core/src/config/config_tests.rs @@ -8359,6 +8359,25 @@ model = "gpt-5-mini" Ok(()) } +#[test] +fn legacy_agent_job_max_runtime_seconds_is_accepted_as_noop() { + let parsed = toml::from_str::( + r#" +[agents] +job_max_runtime_seconds = 900 +"#, + ) + .expect("legacy agent job setting should deserialize"); + + assert_eq!( + parsed.agents, + Some(AgentsToml { + job_max_runtime_seconds: Some(900), + ..Default::default() + }) + ); +} + #[tokio::test] async fn load_config_resolves_agent_controls() -> std::io::Result<()> { let codex_home = TempDir::new()?; diff --git a/codex-rs/core/src/config/mod.rs b/codex-rs/core/src/config/mod.rs index e9b1353a6d..48ca61c0c9 100644 --- a/codex-rs/core/src/config/mod.rs +++ b/codex-rs/core/src/config/mod.rs @@ -266,7 +266,6 @@ pub(crate) const HARD_MIN_MULTI_AGENT_V2_TIMEOUT_MS: i64 = 0; pub(crate) const HARD_MAX_MULTI_AGENT_V2_TIMEOUT_MS: i64 = DEFAULT_MULTI_AGENT_V2_MAX_WAIT_TIMEOUT_MS; pub(crate) const DEFAULT_AGENT_MAX_DEPTH: i32 = 1; -pub(crate) const DEFAULT_AGENT_JOB_MAX_RUNTIME_SECONDS: Option = None; const LOCAL_DEV_BUILD_VERSION: &str = "0.0.0"; pub const CONFIG_TOML_FILE: &str = "config.toml"; @@ -874,9 +873,6 @@ pub struct Config { /// Default reasoning effort for spawned subagents when the spawn call does not select one. pub agent_default_subagent_reasoning_effort: Option, - /// Maximum runtime in seconds for agent job workers before they are failed. - pub agent_job_max_runtime_seconds: Option, - /// Whether to record a model-visible message when an agent turn is interrupted. pub agent_interrupt_message_enabled: bool, @@ -3616,25 +3612,6 @@ impl Config { .agents .as_ref() .and_then(|agents| agents.default_subagent_reasoning_effort.clone()); - let agent_job_max_runtime_seconds = cfg - .agents - .as_ref() - .and_then(|agents| agents.job_max_runtime_seconds) - .or(DEFAULT_AGENT_JOB_MAX_RUNTIME_SECONDS); - if agent_job_max_runtime_seconds == Some(0) { - return Err(std::io::Error::new( - std::io::ErrorKind::InvalidInput, - "agents.job_max_runtime_seconds must be at least 1", - )); - } - if let Some(max_runtime_seconds) = agent_job_max_runtime_seconds - && max_runtime_seconds > i64::MAX as u64 - { - return Err(std::io::Error::new( - std::io::ErrorKind::InvalidInput, - "agents.job_max_runtime_seconds must fit within a 64-bit signed integer", - )); - } let agent_interrupt_message_enabled = cfg .agents .as_ref() @@ -3973,7 +3950,6 @@ impl Config { agent_max_depth, agent_roles, memories: memories_config, - agent_job_max_runtime_seconds, agent_interrupt_message_enabled, codex_home, sqlite_home, diff --git a/codex-rs/core/src/guardian/review_session.rs b/codex-rs/core/src/guardian/review_session.rs index a5e7506d3a..054d28ac56 100644 --- a/codex-rs/core/src/guardian/review_session.rs +++ b/codex-rs/core/src/guardian/review_session.rs @@ -1050,7 +1050,6 @@ pub(crate) fn build_guardian_review_session_config( )?); } for feature in [ - Feature::SpawnCsv, Feature::Collab, Feature::MultiAgentV2, Feature::CodexHooks, diff --git a/codex-rs/core/src/session/config_lock.rs b/codex-rs/core/src/session/config_lock.rs index da0c7a43f6..c18ceb2b3e 100644 --- a/codex-rs/core/src/session/config_lock.rs +++ b/codex-rs/core/src/session/config_lock.rs @@ -183,7 +183,6 @@ fn save_config_resolved_fields( agents.default_subagent_model = config.agent_default_subagent_model.clone(); agents.default_subagent_reasoning_effort = config.agent_default_subagent_reasoning_effort.clone(); - agents.job_max_runtime_seconds = config.agent_job_max_runtime_seconds; agents.interrupt_message = Some(config.agent_interrupt_message_enabled); lock_config diff --git a/codex-rs/core/src/tasks/review.rs b/codex-rs/core/src/tasks/review.rs index 550eaa329d..0e2e2712e0 100644 --- a/codex-rs/core/src/tasks/review.rs +++ b/codex-rs/core/src/tasks/review.rs @@ -110,7 +110,6 @@ async fn start_review_conversation( { panic!("by construction Constrained must always support Disabled: {err}"); } - let _ = sub_agent_config.features.disable(Feature::SpawnCsv); let _ = sub_agent_config.features.disable(Feature::Collab); let _ = sub_agent_config.features.disable(Feature::MultiAgentV2); diff --git a/codex-rs/core/src/tools/handlers/agent_jobs.rs b/codex-rs/core/src/tools/handlers/agent_jobs.rs deleted file mode 100644 index 2398889691..0000000000 --- a/codex-rs/core/src/tools/handlers/agent_jobs.rs +++ /dev/null @@ -1,771 +0,0 @@ -use crate::agent::control::SpawnAgentOptions; -use crate::agent::status::is_final; -use crate::config::Config; -use crate::function_tool::FunctionCallError; -use crate::session::session::Session; -use crate::session::turn_context::TurnContext; -use crate::tools::handlers::multi_agents::build_agent_spawn_config; -use crate::tools::handlers::multi_agents_common::apply_requested_spawn_agent_model_overrides; -use crate::tools::handlers::parse_arguments; -use codex_protocol::ThreadId; -use codex_protocol::error::CodexErr; -use codex_protocol::protocol::AgentStatus; -use codex_protocol::protocol::MultiAgentVersion; -use codex_protocol::protocol::SessionSource; -use codex_protocol::protocol::SubAgentSource; -use codex_protocol::user_input::UserInput; -use codex_utils_absolute_path::AbsolutePathBuf; -use futures::StreamExt; -use futures::stream::FuturesUnordered; -use serde::Deserialize; -use serde::Serialize; -use serde_json::Value; -use std::collections::HashMap; -use std::collections::HashSet; -use std::path::PathBuf; -use std::sync::Arc; -use tokio::sync::watch::Receiver; -use tokio::time::Duration; -use tokio::time::Instant; -use tokio::time::timeout; -use uuid::Uuid; - -mod report_agent_job_result; -mod spawn_agents_on_csv; - -pub use report_agent_job_result::ReportAgentJobResultHandler; -pub use spawn_agents_on_csv::SpawnAgentsOnCsvHandler; - -const DEFAULT_AGENT_JOB_CONCURRENCY: usize = 16; -const MAX_AGENT_JOB_CONCURRENCY: usize = 64; -const STATUS_POLL_INTERVAL: Duration = Duration::from_millis(250); -const DEFAULT_AGENT_JOB_ITEM_TIMEOUT: Duration = Duration::from_secs(60 * 30); - -#[derive(Debug, Deserialize)] -struct SpawnAgentsOnCsvArgs { - csv_path: String, - instruction: String, - id_column: Option, - output_csv_path: Option, - output_schema: Option, - max_concurrency: Option, - max_workers: Option, - max_runtime_seconds: Option, -} - -#[derive(Debug, Deserialize)] -struct ReportAgentJobResultArgs { - job_id: String, - item_id: String, - result: Value, - stop: Option, -} - -#[derive(Debug, Serialize)] -struct SpawnAgentsOnCsvResult { - job_id: String, - status: String, - output_csv_path: String, - total_items: usize, - completed_items: usize, - failed_items: usize, - job_error: Option, - failed_item_errors: Option>, -} - -#[derive(Debug, Serialize)] -struct AgentJobFailureSummary { - item_id: String, - source_id: Option, - last_error: String, -} - -#[derive(Debug, Serialize)] -struct ReportAgentJobResultToolResult { - accepted: bool, -} - -#[derive(Debug, Clone)] -struct JobRunnerOptions { - max_concurrency: usize, - spawn_config: Config, -} - -#[derive(Debug, Clone)] -struct ActiveJobItem { - item_id: String, - started_at: Instant, - status_rx: Option>, -} - -fn required_state_db( - session: &Arc, -) -> Result, FunctionCallError> { - session.state_db().ok_or_else(|| { - FunctionCallError::Fatal("sqlite state db is unavailable for this session".to_string()) - }) -} - -async fn build_runner_options( - session: &Arc, - turn: &Arc, - requested_concurrency: Option, -) -> Result { - let multi_agent_version = turn.multi_agent_version; - if multi_agent_version == MultiAgentVersion::Disabled { - return Err(FunctionCallError::RespondToModel( - "multi-agent runtime is disabled; this session cannot spawn workers".to_string(), - )); - } - let agent_max_threads = turn.config.effective_agent_max_threads(multi_agent_version); - if agent_max_threads == Some(0) { - return Err(FunctionCallError::RespondToModel( - "agent thread limit reached; this session cannot spawn more subagents".to_string(), - )); - } - let max_concurrency = normalize_concurrency(requested_concurrency, agent_max_threads); - let base_instructions = session.get_base_instructions().await; - let mut spawn_config = build_agent_spawn_config(&base_instructions, turn.as_ref())?; - apply_requested_spawn_agent_model_overrides( - session, - turn, - &mut spawn_config, - /*requested_model*/ None, - /*requested_reasoning_effort*/ None, - ) - .await?; - Ok(JobRunnerOptions { - max_concurrency, - spawn_config, - }) -} - -fn normalize_concurrency(requested: Option, max_threads: Option) -> usize { - let requested = requested.unwrap_or(DEFAULT_AGENT_JOB_CONCURRENCY).max(1); - let requested = requested.min(MAX_AGENT_JOB_CONCURRENCY); - if let Some(max_threads) = max_threads { - requested.min(max_threads.max(1)) - } else { - requested - } -} - -fn normalize_max_runtime_seconds(requested: Option) -> Result, FunctionCallError> { - let Some(requested) = requested else { - return Ok(None); - }; - if requested == 0 { - return Err(FunctionCallError::RespondToModel( - "max_runtime_seconds must be >= 1".to_string(), - )); - } - Ok(Some(requested)) -} - -async fn run_agent_job_loop( - session: Arc, - turn: Arc, - db: Arc, - job_id: String, - options: JobRunnerOptions, -) -> anyhow::Result<()> { - let job = db - .get_agent_job(job_id.as_str()) - .await? - .ok_or_else(|| anyhow::anyhow!("agent job {job_id} was not found"))?; - let runtime_timeout = job_runtime_timeout(&job); - let mut active_items: HashMap = HashMap::new(); - recover_running_items( - session.clone(), - db.clone(), - job_id.as_str(), - &mut active_items, - runtime_timeout, - ) - .await?; - - let mut cancel_requested = db.is_agent_job_cancelled(job_id.as_str()).await?; - loop { - let mut progressed = false; - - if !cancel_requested && db.is_agent_job_cancelled(job_id.as_str()).await? { - cancel_requested = true; - } - - if !cancel_requested && active_items.len() < options.max_concurrency { - let slots = options.max_concurrency - active_items.len(); - let pending_items = db - .list_agent_job_items( - job_id.as_str(), - Some(codex_state::AgentJobItemStatus::Pending), - Some(slots), - ) - .await?; - for item in pending_items { - let prompt = build_worker_prompt(&job, &item)?; - let items = vec![UserInput::Text { - text: prompt, - text_elements: Vec::new(), - }]; - let thread_id = match session - .services - .agent_control - .spawn_agent_with_metadata( - options.spawn_config.clone(), - items, - Some(SessionSource::SubAgent(SubAgentSource::Other(format!( - "agent_job:{job_id}" - )))), - SpawnAgentOptions { - parent_thread_id: Some(session.thread_id), - environments: Some(turn.environments.to_selections()), - ..Default::default() - }, - ) - .await - { - Ok(spawned_agent) => spawned_agent.thread_id, - Err(CodexErr::AgentLimitReached { .. }) => { - db.mark_agent_job_item_pending( - job_id.as_str(), - item.item_id.as_str(), - /*error_message*/ None, - ) - .await?; - break; - } - Err(err) => { - let error_message = format!("failed to spawn worker: {err}"); - db.mark_agent_job_item_failed( - job_id.as_str(), - item.item_id.as_str(), - error_message.as_str(), - ) - .await?; - progressed = true; - continue; - } - }; - let assigned = db - .mark_agent_job_item_running_with_thread( - job_id.as_str(), - item.item_id.as_str(), - thread_id.to_string().as_str(), - ) - .await?; - if !assigned { - let _ = session - .services - .agent_control - .shutdown_live_agent(thread_id) - .await; - continue; - } - active_items.insert( - thread_id, - ActiveJobItem { - item_id: item.item_id.clone(), - started_at: Instant::now(), - status_rx: session - .services - .agent_control - .subscribe_status(thread_id) - .await - .ok(), - }, - ); - progressed = true; - } - } - - if reap_stale_active_items( - session.clone(), - db.clone(), - job_id.as_str(), - &mut active_items, - runtime_timeout, - ) - .await? - { - progressed = true; - } - - let finished = find_finished_threads(session.clone(), &active_items).await; - if finished.is_empty() { - let progress = db.get_agent_job_progress(job_id.as_str()).await?; - if cancel_requested { - if progress.running_items == 0 && active_items.is_empty() { - break; - } - } else if progress.pending_items == 0 - && progress.running_items == 0 - && active_items.is_empty() - { - break; - } - if !progressed { - wait_for_status_change(&active_items).await; - } - continue; - } - - for (thread_id, item_id) in finished { - finalize_finished_item( - session.clone(), - db.clone(), - job_id.as_str(), - item_id.as_str(), - thread_id, - ) - .await?; - active_items.remove(&thread_id); - } - } - - if let Err(err) = export_job_csv_snapshot(db.clone(), &job).await { - let message = format!("auto-export failed: {err}"); - db.mark_agent_job_failed(job_id.as_str(), message.as_str()) - .await?; - return Ok(()); - } - let cancelled = cancel_requested || db.is_agent_job_cancelled(job_id.as_str()).await?; - if cancelled { - return Ok(()); - } - db.mark_agent_job_completed(job_id.as_str()).await?; - Ok(()) -} - -async fn export_job_csv_snapshot( - db: Arc, - job: &codex_state::AgentJob, -) -> anyhow::Result<()> { - let items = db - .list_agent_job_items(job.id.as_str(), /*status*/ None, /*limit*/ None) - .await?; - let csv_content = render_job_csv(job.input_headers.as_slice(), items.as_slice()) - .map_err(|err| anyhow::anyhow!("failed to render job csv for auto-export: {err}"))?; - let output_path = PathBuf::from(job.output_csv_path.clone()); - if let Some(parent) = output_path.parent() { - tokio::fs::create_dir_all(parent).await?; - } - tokio::fs::write(&output_path, csv_content).await?; - Ok(()) -} - -async fn recover_running_items( - session: Arc, - db: Arc, - job_id: &str, - active_items: &mut HashMap, - runtime_timeout: Duration, -) -> anyhow::Result<()> { - let running_items = db - .list_agent_job_items( - job_id, - Some(codex_state::AgentJobItemStatus::Running), - /*limit*/ None, - ) - .await?; - for item in running_items { - if is_item_stale(&item, runtime_timeout) { - let error_message = format!("worker exceeded max runtime of {runtime_timeout:?}"); - db.mark_agent_job_item_failed(job_id, item.item_id.as_str(), error_message.as_str()) - .await?; - if let Some(assigned_thread_id) = item.assigned_thread_id.as_ref() - && let Ok(thread_id) = ThreadId::from_string(assigned_thread_id.as_str()) - { - let _ = session - .services - .agent_control - .shutdown_live_agent(thread_id) - .await; - } - continue; - } - let Some(assigned_thread_id) = item.assigned_thread_id.clone() else { - db.mark_agent_job_item_failed( - job_id, - item.item_id.as_str(), - "running item is missing assigned_thread_id", - ) - .await?; - continue; - }; - let thread_id = match ThreadId::from_string(assigned_thread_id.as_str()) { - Ok(thread_id) => thread_id, - Err(err) => { - let error_message = format!("invalid assigned_thread_id: {err:?}"); - db.mark_agent_job_item_failed( - job_id, - item.item_id.as_str(), - error_message.as_str(), - ) - .await?; - continue; - } - }; - if is_final(&session.services.agent_control.get_status(thread_id).await) { - finalize_finished_item( - session.clone(), - db.clone(), - job_id, - item.item_id.as_str(), - thread_id, - ) - .await?; - } else { - active_items.insert( - thread_id, - ActiveJobItem { - item_id: item.item_id.clone(), - started_at: started_at_from_item(&item), - status_rx: session - .services - .agent_control - .subscribe_status(thread_id) - .await - .ok(), - }, - ); - } - } - Ok(()) -} - -async fn find_finished_threads( - session: Arc, - active_items: &HashMap, -) -> Vec<(ThreadId, String)> { - let mut finished = Vec::new(); - for (thread_id, item) in active_items { - let status = active_item_status(session.as_ref(), *thread_id, item).await; - if is_final(&status) { - finished.push((*thread_id, item.item_id.clone())); - } - } - finished -} - -async fn active_item_status( - session: &Session, - thread_id: ThreadId, - item: &ActiveJobItem, -) -> AgentStatus { - if let Some(status_rx) = item.status_rx.as_ref() - && status_rx.has_changed().is_ok() - { - return status_rx.borrow().clone(); - } - session.services.agent_control.get_status(thread_id).await -} - -async fn wait_for_status_change(active_items: &HashMap) { - let mut waiters = FuturesUnordered::new(); - for item in active_items.values() { - if let Some(status_rx) = item.status_rx.as_ref() { - let mut status_rx = status_rx.clone(); - waiters.push(async move { - let _ = status_rx.changed().await; - }); - } - } - if waiters.is_empty() { - tokio::time::sleep(STATUS_POLL_INTERVAL).await; - return; - } - let _ = timeout(STATUS_POLL_INTERVAL, waiters.next()).await; -} - -async fn reap_stale_active_items( - session: Arc, - db: Arc, - job_id: &str, - active_items: &mut HashMap, - runtime_timeout: Duration, -) -> anyhow::Result { - let mut stale = Vec::new(); - for (thread_id, item) in active_items.iter() { - if item.started_at.elapsed() >= runtime_timeout { - stale.push((*thread_id, item.item_id.clone())); - } - } - if stale.is_empty() { - return Ok(false); - } - for (thread_id, item_id) in stale { - let error_message = format!("worker exceeded max runtime of {runtime_timeout:?}"); - db.mark_agent_job_item_failed(job_id, item_id.as_str(), error_message.as_str()) - .await?; - let _ = session - .services - .agent_control - .shutdown_live_agent(thread_id) - .await; - active_items.remove(&thread_id); - } - Ok(true) -} - -async fn finalize_finished_item( - session: Arc, - db: Arc, - job_id: &str, - item_id: &str, - thread_id: ThreadId, -) -> anyhow::Result<()> { - let item = db - .get_agent_job_item(job_id, item_id) - .await? - .ok_or_else(|| { - anyhow::anyhow!("job item not found for finalization: {job_id}/{item_id}") - })?; - if matches!(item.status, codex_state::AgentJobItemStatus::Running) { - if item.result_json.is_some() { - let _ = db.mark_agent_job_item_completed(job_id, item_id).await?; - } else { - let _ = db - .mark_agent_job_item_failed( - job_id, - item_id, - "worker finished without calling report_agent_job_result", - ) - .await?; - } - } - let _ = session - .services - .agent_control - .shutdown_live_agent(thread_id) - .await; - Ok(()) -} - -fn build_worker_prompt( - job: &codex_state::AgentJob, - item: &codex_state::AgentJobItem, -) -> anyhow::Result { - let job_id = job.id.as_str(); - let item_id = item.item_id.as_str(); - let instruction = render_instruction_template(job.instruction.as_str(), &item.row_json); - let output_schema = job - .output_schema_json - .as_ref() - .map(serde_json::to_string_pretty) - .transpose()? - .unwrap_or_else(|| "{}".to_string()); - let row_json = serde_json::to_string_pretty(&item.row_json)?; - Ok(format!( - "You are processing one item for a generic agent job.\n\ -Job ID: {job_id}\n\ -Item ID: {item_id}\n\n\ -Task instruction:\n\ -{instruction}\n\n\ -Input row (JSON):\n\ -{row_json}\n\n\ -Expected result schema (JSON Schema or {{}}):\n\ -{output_schema}\n\n\ -You MUST call the `report_agent_job_result` tool exactly once with:\n\ -1. `job_id` = \"{job_id}\"\n\ -2. `item_id` = \"{item_id}\"\n\ -3. `result` = a JSON object that contains your analysis result for this row.\n\n\ -If you need to stop the job early, include `stop` = true in the tool call.\n\n\ -After the tool call succeeds, stop.", - )) -} - -fn render_instruction_template(instruction: &str, row_json: &Value) -> String { - const OPEN_BRACE_SENTINEL: &str = "__CODEX_OPEN_BRACE__"; - const CLOSE_BRACE_SENTINEL: &str = "__CODEX_CLOSE_BRACE__"; - - let mut rendered = instruction - .replace("{{", OPEN_BRACE_SENTINEL) - .replace("}}", CLOSE_BRACE_SENTINEL); - let Some(row) = row_json.as_object() else { - return rendered - .replace(OPEN_BRACE_SENTINEL, "{") - .replace(CLOSE_BRACE_SENTINEL, "}"); - }; - for (key, value) in row { - let placeholder = format!("{{{key}}}"); - let replacement = value - .as_str() - .map(str::to_string) - .unwrap_or_else(|| value.to_string()); - rendered = rendered.replace(placeholder.as_str(), replacement.as_str()); - } - rendered - .replace(OPEN_BRACE_SENTINEL, "{") - .replace(CLOSE_BRACE_SENTINEL, "}") -} - -fn ensure_unique_headers(headers: &[String]) -> Result<(), FunctionCallError> { - let mut seen = HashSet::new(); - for header in headers { - if !seen.insert(header) { - return Err(FunctionCallError::RespondToModel(format!( - "csv header {header} is duplicated" - ))); - } - } - Ok(()) -} - -fn job_runtime_timeout(job: &codex_state::AgentJob) -> Duration { - job.max_runtime_seconds - .map(Duration::from_secs) - .unwrap_or(DEFAULT_AGENT_JOB_ITEM_TIMEOUT) -} - -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() { - Instant::now().checked_sub(age).unwrap_or_else(Instant::now) - } else { - Instant::now() - } -} - -fn is_item_stale(item: &codex_state::AgentJobItem, runtime_timeout: Duration) -> bool { - let now = chrono::Utc::now(); - if let Ok(age) = now.signed_duration_since(item.updated_at).to_std() { - age >= runtime_timeout - } else { - false - } -} - -fn default_output_csv_path(input_csv_path: &AbsolutePathBuf, job_id: &str) -> AbsolutePathBuf { - let stem = input_csv_path - .as_path() - .file_stem() - .and_then(|stem| stem.to_str()) - .unwrap_or("agent_job_output"); - let job_suffix = &job_id[..8]; - let output_dir = input_csv_path - .parent() - .unwrap_or_else(|| input_csv_path.clone()); - output_dir.join(format!("{stem}.agent-job-{job_suffix}.csv")) -} - -fn parse_csv(content: &str) -> Result<(Vec, Vec>), String> { - let mut reader = csv::ReaderBuilder::new() - .has_headers(true) - .flexible(true) - .from_reader(content.as_bytes()); - let headers_record = reader.headers().map_err(|err| err.to_string())?; - let mut headers: Vec = headers_record.iter().map(str::to_string).collect(); - if let Some(first) = headers.first_mut() { - *first = first.trim_start_matches('\u{feff}').to_string(); - } - let mut rows = Vec::new(); - for record in reader.records() { - let record = record.map_err(|err| err.to_string())?; - let row: Vec = record.iter().map(str::to_string).collect(); - if row.iter().all(std::string::String::is_empty) { - continue; - } - rows.push(row); - } - Ok((headers, rows)) -} - -fn render_job_csv( - headers: &[String], - items: &[codex_state::AgentJobItem], -) -> Result { - let mut csv = String::new(); - let mut output_headers = headers.to_vec(); - output_headers.extend([ - "job_id".to_string(), - "item_id".to_string(), - "row_index".to_string(), - "source_id".to_string(), - "status".to_string(), - "attempt_count".to_string(), - "last_error".to_string(), - "result_json".to_string(), - "reported_at".to_string(), - "completed_at".to_string(), - ]); - csv.push_str( - output_headers - .iter() - .map(|header| csv_escape(header.as_str())) - .collect::>() - .join(",") - .as_str(), - ); - csv.push('\n'); - for item in items { - let row_object = item.row_json.as_object().ok_or_else(|| { - let item_id = item.item_id.as_str(); - FunctionCallError::RespondToModel(format!( - "row_json for item {item_id} is not a JSON object" - )) - })?; - let mut row_values = Vec::new(); - for header in headers { - let value = row_object - .get(header) - .map_or_else(String::new, value_to_csv_string); - row_values.push(csv_escape(value.as_str())); - } - row_values.push(csv_escape(item.job_id.as_str())); - row_values.push(csv_escape(item.item_id.as_str())); - row_values.push(csv_escape(item.row_index.to_string().as_str())); - row_values.push(csv_escape( - item.source_id.clone().unwrap_or_default().as_str(), - )); - row_values.push(csv_escape(item.status.as_str())); - row_values.push(csv_escape(item.attempt_count.to_string().as_str())); - row_values.push(csv_escape( - item.last_error.clone().unwrap_or_default().as_str(), - )); - row_values.push(csv_escape( - item.result_json - .as_ref() - .map_or_else(String::new, std::string::ToString::to_string) - .as_str(), - )); - row_values.push(csv_escape( - item.reported_at - .map(|value| value.to_rfc3339()) - .unwrap_or_default() - .as_str(), - )); - row_values.push(csv_escape( - item.completed_at - .map(|value| value.to_rfc3339()) - .unwrap_or_default() - .as_str(), - )); - csv.push_str(row_values.join(",").as_str()); - csv.push('\n'); - } - Ok(csv) -} - -fn value_to_csv_string(value: &Value) -> String { - match value { - Value::Null => String::new(), - Value::String(s) => s.clone(), - Value::Bool(b) => b.to_string(), - Value::Number(n) => n.to_string(), - Value::Array(_) | Value::Object(_) => value.to_string(), - } -} - -fn csv_escape(value: &str) -> String { - if value.contains(',') || value.contains('\n') || value.contains('\r') || value.contains('"') { - let escaped = value.replace('"', "\"\""); - format!("\"{escaped}\"") - } else { - value.to_string() - } -} - -#[cfg(test)] -#[path = "agent_jobs_tests.rs"] -mod tests; diff --git a/codex-rs/core/src/tools/handlers/agent_jobs/report_agent_job_result.rs b/codex-rs/core/src/tools/handlers/agent_jobs/report_agent_job_result.rs deleted file mode 100644 index 579cc25ffb..0000000000 --- a/codex-rs/core/src/tools/handlers/agent_jobs/report_agent_job_result.rs +++ /dev/null @@ -1,98 +0,0 @@ -use crate::function_tool::FunctionCallError; -use crate::tools::context::FunctionToolOutput; -use crate::tools::context::ToolInvocation; -use crate::tools::context::ToolPayload; -use crate::tools::context::boxed_tool_output; -use crate::tools::handlers::agent_jobs_spec::create_report_agent_job_result_tool; -use crate::tools::registry::CoreToolRuntime; -use crate::tools::registry::ToolExecutor; -use codex_tools::ToolName; -use codex_tools::ToolSpec; - -use super::*; - -pub struct ReportAgentJobResultHandler; - -impl ToolExecutor for ReportAgentJobResultHandler { - fn tool_name(&self) -> ToolName { - ToolName::plain("report_agent_job_result") - } - - fn spec(&self) -> ToolSpec { - create_report_agent_job_result_tool() - } - - fn handle(&self, invocation: ToolInvocation) -> codex_tools::ToolExecutorFuture<'_> { - Box::pin(self.handle_call(invocation)) - } -} - -impl ReportAgentJobResultHandler { - async fn handle_call( - &self, - invocation: ToolInvocation, - ) -> Result, FunctionCallError> { - let ToolInvocation { - session, payload, .. - } = invocation; - - let arguments = match payload { - ToolPayload::Function { arguments } => arguments, - _ => { - return Err(FunctionCallError::RespondToModel( - "report_agent_job_result handler received unsupported payload".to_string(), - )); - } - }; - - handle(session, arguments).await.map(boxed_tool_output) - } -} - -impl CoreToolRuntime for ReportAgentJobResultHandler { - fn matches_kind(&self, payload: &ToolPayload) -> bool { - matches!(payload, ToolPayload::Function { .. }) - } -} - -pub async fn handle( - session: Arc, - arguments: String, -) -> Result { - let args: ReportAgentJobResultArgs = parse_arguments(arguments.as_str())?; - if !args.result.is_object() { - return Err(FunctionCallError::RespondToModel( - "result must be a JSON object".to_string(), - )); - } - let db = required_state_db(&session)?; - let reporting_thread_id = session.thread_id.to_string(); - let accepted = db - .report_agent_job_item_result( - args.job_id.as_str(), - args.item_id.as_str(), - reporting_thread_id.as_str(), - &args.result, - ) - .await - .map_err(|err| { - let job_id = args.job_id.as_str(); - let item_id = args.item_id.as_str(); - FunctionCallError::RespondToModel(format!( - "failed to record agent job result for {job_id} / {item_id}: {err}" - )) - })?; - if accepted && args.stop.unwrap_or(false) { - let message = "cancelled by worker request"; - let _ = db - .mark_agent_job_cancelled(args.job_id.as_str(), message) - .await; - } - let content = - serde_json::to_string(&ReportAgentJobResultToolResult { accepted }).map_err(|err| { - FunctionCallError::Fatal(format!( - "failed to serialize report_agent_job_result result: {err}" - )) - })?; - Ok(FunctionToolOutput::from_text(content, Some(true))) -} diff --git a/codex-rs/core/src/tools/handlers/agent_jobs/spawn_agents_on_csv.rs b/codex-rs/core/src/tools/handlers/agent_jobs/spawn_agents_on_csv.rs deleted file mode 100644 index 2ad0e22fb8..0000000000 --- a/codex-rs/core/src/tools/handlers/agent_jobs/spawn_agents_on_csv.rs +++ /dev/null @@ -1,329 +0,0 @@ -use crate::function_tool::FunctionCallError; -use crate::tools::context::FunctionToolOutput; -use crate::tools::context::ToolInvocation; -use crate::tools::context::ToolPayload; -use crate::tools::context::boxed_tool_output; -use crate::tools::handlers::agent_jobs_spec::create_spawn_agents_on_csv_tool; -use crate::tools::registry::CoreToolRuntime; -use crate::tools::registry::ToolExecutor; -use codex_tools::ToolName; -use codex_tools::ToolSpec; -use codex_utils_absolute_path::AbsolutePathBuf; - -use super::*; - -pub struct SpawnAgentsOnCsvHandler; - -impl ToolExecutor for SpawnAgentsOnCsvHandler { - fn tool_name(&self) -> ToolName { - ToolName::plain("spawn_agents_on_csv") - } - - fn spec(&self) -> ToolSpec { - create_spawn_agents_on_csv_tool() - } - - fn handle(&self, invocation: ToolInvocation) -> codex_tools::ToolExecutorFuture<'_> { - Box::pin(self.handle_call(invocation)) - } -} - -impl SpawnAgentsOnCsvHandler { - async fn handle_call( - &self, - invocation: ToolInvocation, - ) -> Result, FunctionCallError> { - let ToolInvocation { - session, - turn, - payload, - .. - } = invocation; - - let arguments = match payload { - ToolPayload::Function { arguments } => arguments, - _ => { - return Err(FunctionCallError::RespondToModel( - "agent jobs handler received unsupported payload".to_string(), - )); - } - }; - - handle(session, turn, arguments) - .await - .map(boxed_tool_output) - } -} - -impl CoreToolRuntime for SpawnAgentsOnCsvHandler { - fn matches_kind(&self, payload: &ToolPayload) -> bool { - matches!(payload, ToolPayload::Function { .. }) - } -} - -/// Create a new agent job from a CSV and run it to completion. -/// -/// Each CSV row becomes a job item. The instruction string is a template where `{column}` -/// placeholders are filled with values from that row. Results are reported by workers via -/// `report_agent_job_result`, then exported to CSV on completion. -pub async fn handle( - session: Arc, - turn: Arc, - arguments: String, -) -> Result { - let args: SpawnAgentsOnCsvArgs = parse_arguments(arguments.as_str())?; - if args.instruction.trim().is_empty() { - return Err(FunctionCallError::RespondToModel( - "instruction must be non-empty".to_string(), - )); - } - - let cwd = single_local_environment_cwd(&turn)?; - let db = required_state_db(&session)?; - let input_path = cwd.join(args.csv_path); - let input_path_display = input_path.display().to_string(); - let csv_content = tokio::fs::read_to_string(&input_path) - .await - .map_err(|err| { - FunctionCallError::RespondToModel(format!( - "failed to read csv input {input_path_display}: {err}" - )) - })?; - let (headers, rows) = parse_csv(csv_content.as_str()).map_err(|err| { - FunctionCallError::RespondToModel(format!("failed to parse csv input: {err}")) - })?; - if headers.is_empty() { - return Err(FunctionCallError::RespondToModel( - "csv input must include a header row".to_string(), - )); - } - ensure_unique_headers(headers.as_slice())?; - - let id_column_index = args.id_column.as_ref().map_or(Ok(None), |column_name| { - headers - .iter() - .position(|header| header == column_name) - .map(Some) - .ok_or_else(|| { - FunctionCallError::RespondToModel(format!( - "id_column {column_name} was not found in csv headers" - )) - }) - })?; - - let mut items = Vec::with_capacity(rows.len()); - let mut seen_ids = HashSet::new(); - for (idx, row) in rows.into_iter().enumerate() { - if row.len() != headers.len() { - let row_index = idx + 2; - let row_len = row.len(); - let header_len = headers.len(); - return Err(FunctionCallError::RespondToModel(format!( - "csv row {row_index} has {row_len} fields but header has {header_len}" - ))); - } - - let source_id = id_column_index - .and_then(|index| row.get(index).cloned()) - .filter(|value| !value.trim().is_empty()); - let row_index = idx + 1; - let base_item_id = source_id - .clone() - .unwrap_or_else(|| format!("row-{row_index}")); - let mut item_id = base_item_id.clone(); - let mut suffix = 2usize; - while !seen_ids.insert(item_id.clone()) { - item_id = format!("{base_item_id}-{suffix}"); - suffix = suffix.saturating_add(1); - } - - let row_object = headers - .iter() - .zip(row.iter()) - .map(|(header, value)| (header.clone(), Value::String(value.clone()))) - .collect::>(); - items.push(codex_state::AgentJobItemCreateParams { - item_id, - row_index: idx as i64, - source_id, - row_json: Value::Object(row_object), - }); - } - - let job_id = Uuid::new_v4().to_string(); - let output_csv_path = args.output_csv_path.map_or_else( - || default_output_csv_path(&input_path, job_id.as_str()), - |path| cwd.join(path), - ); - let job_suffix = &job_id[..8]; - let job_name = format!("agent-job-{job_suffix}"); - let max_runtime_seconds = normalize_max_runtime_seconds( - args.max_runtime_seconds - .or(turn.config.agent_job_max_runtime_seconds), - )?; - let _job = db - .create_agent_job( - &codex_state::AgentJobCreateParams { - id: job_id.clone(), - name: job_name, - instruction: args.instruction, - auto_export: true, - max_runtime_seconds, - output_schema_json: args.output_schema, - input_headers: headers, - input_csv_path: input_path.display().to_string(), - output_csv_path: output_csv_path.display().to_string(), - }, - items.as_slice(), - ) - .await - .map_err(|err| { - FunctionCallError::RespondToModel(format!("failed to create agent job: {err}")) - })?; - - let requested_concurrency = args.max_concurrency.or(args.max_workers); - let options = match build_runner_options(&session, &turn, requested_concurrency).await { - Ok(options) => options, - Err(err) => { - let error_message = err.to_string(); - let _ = db - .mark_agent_job_failed(job_id.as_str(), error_message.as_str()) - .await; - return Err(err); - } - }; - db.mark_agent_job_running(job_id.as_str()) - .await - .map_err(|err| { - FunctionCallError::RespondToModel(format!( - "failed to transition agent job {job_id} to running: {err}" - )) - })?; - if let Err(err) = run_agent_job_loop( - session.clone(), - turn.clone(), - db.clone(), - job_id.clone(), - options, - ) - .await - { - let error_message = format!("job runner failed: {err}"); - let _ = db - .mark_agent_job_failed(job_id.as_str(), error_message.as_str()) - .await; - return Err(FunctionCallError::RespondToModel(format!( - "agent job {job_id} failed: {err}" - ))); - } - - let job = db - .get_agent_job(job_id.as_str()) - .await - .map_err(|err| { - FunctionCallError::RespondToModel(format!("failed to load agent job {job_id}: {err}")) - })? - .ok_or_else(|| { - FunctionCallError::RespondToModel(format!("agent job {job_id} not found")) - })?; - let output_path = PathBuf::from(job.output_csv_path.clone()); - if !tokio::fs::try_exists(&output_path).await.unwrap_or(false) { - export_job_csv_snapshot(db.clone(), &job) - .await - .map_err(|err| { - FunctionCallError::RespondToModel(format!( - "failed to export output csv {job_id}: {err}" - )) - })?; - } - let progress = db - .get_agent_job_progress(job_id.as_str()) - .await - .map_err(|err| { - FunctionCallError::RespondToModel(format!( - "failed to load agent job progress {job_id}: {err}" - )) - })?; - let mut job_error = job.last_error.clone().filter(|err| !err.trim().is_empty()); - let failed_item_errors = if progress.failed_items > 0 { - let items = db - .list_agent_job_items( - job_id.as_str(), - Some(codex_state::AgentJobItemStatus::Failed), - Some(5), - ) - .await - .unwrap_or_default(); - let summaries: Vec<_> = items - .into_iter() - .filter_map(|item| { - let last_error = item.last_error.unwrap_or_default(); - if last_error.trim().is_empty() { - return None; - } - Some(AgentJobFailureSummary { - item_id: item.item_id, - source_id: item.source_id, - last_error, - }) - }) - .collect(); - if summaries.is_empty() { - if job_error.is_none() { - job_error = Some( - "agent job has failed items but no error details were recorded".to_string(), - ); - } - None - } else { - Some(summaries) - } - } else { - None - }; - let content = serde_json::to_string(&SpawnAgentsOnCsvResult { - job_id, - status: job.status.as_str().to_string(), - output_csv_path: job.output_csv_path, - total_items: progress.total_items, - completed_items: progress.completed_items, - failed_items: progress.failed_items, - job_error, - failed_item_errors, - }) - .map_err(|err| { - FunctionCallError::Fatal(format!( - "failed to serialize spawn_agents_on_csv result: {err}" - )) - })?; - Ok(FunctionToolOutput::from_text(content, Some(true))) -} - -fn single_local_environment_cwd(turn: &TurnContext) -> Result { - let mut environments = turn.environments.turn_environments(); - let Some(turn_environment) = environments.next() else { - return Err(FunctionCallError::RespondToModel( - "spawn_agents_on_csv requires exactly one local environment".to_string(), - )); - }; - if environments.next().is_some() { - return Err(FunctionCallError::RespondToModel( - "spawn_agents_on_csv requires exactly one local environment".to_string(), - )); - }; - - if turn_environment.environment.is_remote() { - return Err(FunctionCallError::RespondToModel( - "spawn_agents_on_csv is not supported for remote environments".to_string(), - )); - } - - // TODO(anp): Migrate spawn_agents_on_csv filesystem access to PathUri before enabling it for - // remote environments. - turn_environment.cwd().to_abs_path().map_err(|err| { - FunctionCallError::RespondToModel(format!( - "spawn_agents_on_csv cwd `{}` is not native to the Codex host: {err}", - turn_environment.cwd() - )) - }) -} diff --git a/codex-rs/core/src/tools/handlers/agent_jobs_spec.rs b/codex-rs/core/src/tools/handlers/agent_jobs_spec.rs deleted file mode 100644 index 437549a929..0000000000 --- a/codex-rs/core/src/tools/handlers/agent_jobs_spec.rs +++ /dev/null @@ -1,119 +0,0 @@ -use codex_tools::JsonSchema; -use codex_tools::ResponsesApiTool; -use codex_tools::ToolSpec; -use std::collections::BTreeMap; - -pub fn create_spawn_agents_on_csv_tool() -> ToolSpec { - let mut output_schema = JsonSchema::object( - BTreeMap::new(), - /*required*/ None, - /*additional_properties*/ None, - ); - output_schema.description = - Some("JSON Schema for each worker result. Omit to accept any result object.".to_string()); - - let properties = BTreeMap::from([ - ( - "csv_path".to_string(), - JsonSchema::string(Some("Path to the CSV file containing input rows.".to_string())), - ), - ( - "instruction".to_string(), - JsonSchema::string(Some( - "Instruction template to apply to each CSV row. Use {column_name} placeholders to inject values from the row." - .to_string(), - )), - ), - ( - "id_column".to_string(), - JsonSchema::string(Some( - "CSV column to use as stable item id. Omit to use row numbers.".to_string(), - )), - ), - ( - "output_csv_path".to_string(), - JsonSchema::string(Some( - "Output CSV path for exported results. Omit to create one next to the input CSV." - .to_string(), - )), - ), - ( - "max_concurrency".to_string(), - JsonSchema::number(Some( - "Maximum concurrent workers for this job. Defaults to 16 and is capped by config." - .to_string(), - )), - ), - ( - "max_workers".to_string(), - JsonSchema::number(Some( - "Alias for max_concurrency. Defaults to 16 and is capped by config.".to_string(), - )), - ), - ( - "max_runtime_seconds".to_string(), - JsonSchema::number(Some( - "Maximum runtime per worker before failure. Defaults to 1800 seconds; config may set a different default." - .to_string(), - )), - ), - ("output_schema".to_string(), output_schema), - ]); - - ToolSpec::Function(ResponsesApiTool { - name: "spawn_agents_on_csv".to_string(), - description: "Process a CSV by spawning one worker sub-agent per row. The instruction string is a template where `{column}` placeholders are replaced with row values. Each worker must call `report_agent_job_result` with a JSON object (matching `output_schema` when provided); missing reports are treated as failures. This call blocks until all rows finish and automatically exports results to `output_csv_path` (or a default path)." - .to_string(), - strict: false, - defer_loading: None, - parameters: JsonSchema::object(properties, Some(vec!["csv_path".to_string(), "instruction".to_string()]), Some(false.into())), - output_schema: None, - }) -} - -pub fn create_report_agent_job_result_tool() -> ToolSpec { - let mut result_schema = JsonSchema::object( - BTreeMap::new(), - /*required*/ None, - /*additional_properties*/ None, - ); - result_schema.description = Some("Result object for this job item.".to_string()); - - let properties = BTreeMap::from([ - ( - "job_id".to_string(), - JsonSchema::string(Some("Identifier of the job.".to_string())), - ), - ( - "item_id".to_string(), - JsonSchema::string(Some("Identifier of the job item.".to_string())), - ), - ("result".to_string(), result_schema), - ( - "stop".to_string(), - JsonSchema::boolean(Some( - "True cancels remaining job items after this result is recorded; false or omitted continues the job." - .to_string(), - )), - ), - ]); - - ToolSpec::Function(ResponsesApiTool { - name: "report_agent_job_result".to_string(), - description: - "Worker-only tool to report a result for an agent job item. Main agents should not call this." - .to_string(), - strict: false, - defer_loading: None, - parameters: JsonSchema::object(properties, Some(vec![ - "job_id".to_string(), - "item_id".to_string(), - "result".to_string(), - ]), Some(false.into())), - output_schema: None, - }) -} - -#[cfg(test)] -#[path = "agent_jobs_spec_tests.rs"] -mod tests; diff --git a/codex-rs/core/src/tools/handlers/agent_jobs_spec_tests.rs b/codex-rs/core/src/tools/handlers/agent_jobs_spec_tests.rs deleted file mode 100644 index b087094095..0000000000 --- a/codex-rs/core/src/tools/handlers/agent_jobs_spec_tests.rs +++ /dev/null @@ -1,126 +0,0 @@ -use super::*; -use codex_tools::JsonSchema; -use pretty_assertions::assert_eq; -use std::collections::BTreeMap; - -fn described_object(description: &str) -> JsonSchema { - let mut schema = JsonSchema::object( - BTreeMap::new(), - /*required*/ None, - /*additional_properties*/ None, - ); - schema.description = Some(description.to_string()); - schema -} - -#[test] -fn spawn_agents_on_csv_tool_requires_csv_and_instruction() { - assert_eq!( - create_spawn_agents_on_csv_tool(), - ToolSpec::Function(ResponsesApiTool { - name: "spawn_agents_on_csv".to_string(), - description: "Process a CSV by spawning one worker sub-agent per row. The instruction string is a template where `{column}` placeholders are replaced with row values. Each worker must call `report_agent_job_result` with a JSON object (matching `output_schema` when provided); missing reports are treated as failures. This call blocks until all rows finish and automatically exports results to `output_csv_path` (or a default path)." - .to_string(), - strict: false, - defer_loading: None, - parameters: JsonSchema::object(BTreeMap::from([ - ( - "csv_path".to_string(), - JsonSchema::string(Some( - "Path to the CSV file containing input rows.".to_string(), - )), - ), - ( - "instruction".to_string(), - JsonSchema::string(Some( - "Instruction template to apply to each CSV row. Use {column_name} placeholders to inject values from the row." - .to_string(), - )), - ), - ( - "id_column".to_string(), - JsonSchema::string(Some( - "CSV column to use as stable item id. Omit to use row numbers." - .to_string(), - )), - ), - ( - "output_csv_path".to_string(), - JsonSchema::string(Some( - "Output CSV path for exported results. Omit to create one next to the input CSV." - .to_string(), - )), - ), - ( - "max_concurrency".to_string(), - JsonSchema::number(Some( - "Maximum concurrent workers for this job. Defaults to 16 and is capped by config." - .to_string(), - )), - ), - ( - "max_workers".to_string(), - JsonSchema::number(Some( - "Alias for max_concurrency. Defaults to 16 and is capped by config." - .to_string(), - )), - ), - ( - "max_runtime_seconds".to_string(), - JsonSchema::number(Some( - "Maximum runtime per worker before failure. Defaults to 1800 seconds; config may set a different default." - .to_string(), - )), - ), - ( - "output_schema".to_string(), - described_object( - "JSON Schema for each worker result. Omit to accept any result object.", - ), - ), - ]), Some(vec!["csv_path".to_string(), "instruction".to_string()]), Some(false.into())), - output_schema: None, - }) - ); -} - -#[test] -fn report_agent_job_result_tool_requires_result_payload() { - assert_eq!( - create_report_agent_job_result_tool(), - ToolSpec::Function(ResponsesApiTool { - name: "report_agent_job_result".to_string(), - description: - "Worker-only tool to report a result for an agent job item. Main agents should not call this." - .to_string(), - strict: false, - defer_loading: None, - parameters: JsonSchema::object(BTreeMap::from([ - ( - "job_id".to_string(), - JsonSchema::string(Some("Identifier of the job.".to_string())), - ), - ( - "item_id".to_string(), - JsonSchema::string(Some("Identifier of the job item.".to_string())), - ), - ( - "result".to_string(), - described_object("Result object for this job item."), - ), - ( - "stop".to_string(), - JsonSchema::boolean(Some( - "True cancels remaining job items after this result is recorded; false or omitted continues the job." - .to_string(), - )), - ), - ]), Some(vec![ - "job_id".to_string(), - "item_id".to_string(), - "result".to_string(), - ]), Some(false.into())), - output_schema: None, - }) - ); -} diff --git a/codex-rs/core/src/tools/handlers/agent_jobs_tests.rs b/codex-rs/core/src/tools/handlers/agent_jobs_tests.rs deleted file mode 100644 index a2dbe6a480..0000000000 --- a/codex-rs/core/src/tools/handlers/agent_jobs_tests.rs +++ /dev/null @@ -1,62 +0,0 @@ -use super::*; -use pretty_assertions::assert_eq; -use serde_json::json; - -#[test] -fn parse_csv_supports_quotes_and_commas() { - let input = "id,name\n1,\"alpha, beta\"\n2,gamma\n"; - let (headers, rows) = parse_csv(input).expect("csv parse"); - assert_eq!(headers, vec!["id".to_string(), "name".to_string()]); - assert_eq!( - rows, - vec![ - vec!["1".to_string(), "alpha, beta".to_string()], - vec!["2".to_string(), "gamma".to_string()] - ] - ); -} - -#[test] -fn csv_escape_quotes_when_needed() { - assert_eq!(csv_escape("simple"), "simple"); - assert_eq!(csv_escape("a,b"), "\"a,b\""); - assert_eq!(csv_escape("a\"b"), "\"a\"\"b\""); -} - -#[test] -fn render_instruction_template_expands_placeholders_and_escapes_braces() { - let row = json!({ - "path": "src/lib.rs", - "area": "test", - "file path": "docs/readme.md", - }); - let rendered = render_instruction_template( - "Review {path} in {area}. Also see {file path}. Use {{literal}}.", - &row, - ); - assert_eq!( - rendered, - "Review src/lib.rs in test. Also see docs/readme.md. Use {literal}." - ); -} - -#[test] -fn render_instruction_template_leaves_unknown_placeholders() { - let row = json!({ - "path": "src/lib.rs", - }); - let rendered = render_instruction_template("Check {path} then {missing}", &row); - assert_eq!(rendered, "Check src/lib.rs then {missing}"); -} - -#[test] -fn ensure_unique_headers_rejects_duplicates() { - let headers = vec!["path".to_string(), "path".to_string()]; - let Err(err) = ensure_unique_headers(headers.as_slice()) else { - panic!("expected duplicate header error"); - }; - assert_eq!( - err, - FunctionCallError::RespondToModel("csv header path is duplicated".to_string()) - ); -} diff --git a/codex-rs/core/src/tools/handlers/mod.rs b/codex-rs/core/src/tools/handlers/mod.rs index 2b5559fb27..6ae144663a 100644 --- a/codex-rs/core/src/tools/handlers/mod.rs +++ b/codex-rs/core/src/tools/handlers/mod.rs @@ -1,5 +1,3 @@ -pub(crate) mod agent_jobs; -pub(crate) mod agent_jobs_spec; pub(crate) mod apply_patch; pub(crate) mod apply_patch_spec; mod current_time; diff --git a/codex-rs/core/src/tools/spec_plan.rs b/codex-rs/core/src/tools/spec_plan.rs index 570106764e..fb58089485 100644 --- a/codex-rs/core/src/tools/spec_plan.rs +++ b/codex-rs/core/src/tools/spec_plan.rs @@ -30,8 +30,6 @@ use crate::tools::handlers::ToolSearchHandlerCache; use crate::tools::handlers::ViewImageHandler; use crate::tools::handlers::WaitForEnvironmentHandler; use crate::tools::handlers::WriteStdinHandler; -use crate::tools::handlers::agent_jobs::ReportAgentJobResultHandler; -use crate::tools::handlers::agent_jobs::SpawnAgentsOnCsvHandler; use crate::tools::handlers::extension_tools::ExtensionToolAdapter; use crate::tools::handlers::multi_agents::CloseAgentHandler; use crate::tools::handlers::multi_agents::ResumeAgentHandler; @@ -67,8 +65,6 @@ use codex_protocol::openai_models::ConfigShellToolType; use codex_protocol::openai_models::InputModality; use codex_protocol::openai_models::ToolMode; use codex_protocol::protocol::MultiAgentVersion; -use codex_protocol::protocol::SessionSource; -use codex_protocol::protocol::SubAgentSource; use codex_tools::ResponsesApiNamespace; use codex_tools::ResponsesApiNamespaceTool; use codex_tools::TOOL_SEARCH_TOOL_NAME; @@ -346,24 +342,6 @@ fn collab_tools_enabled(turn_context: &TurnContext) -> bool { } } -fn agent_jobs_tools_enabled(turn_context: &TurnContext) -> bool { - turn_context - .config - .features - .get() - .enabled(Feature::SpawnCsv) - && collab_tools_enabled(turn_context) -} - -fn agent_jobs_worker_tools_enabled(turn_context: &TurnContext) -> bool { - agent_jobs_tools_enabled(turn_context) - && matches!( - &turn_context.session_source, - SessionSource::SubAgent(SubAgentSource::Other(label)) - if label.starts_with("agent_job:") - ) -} - fn image_generation_available(turn_context: &TurnContext) -> bool { if !turn_context .config @@ -870,13 +848,6 @@ fn add_collaboration_tools(context: &CoreToolPlanContext<'_>, planned_tools: &mu planned_tools.add_with_exposure(CloseAgentHandler, exposure); } } - - if agent_jobs_tools_enabled(turn_context) { - planned_tools.add(SpawnAgentsOnCsvHandler); - if agent_jobs_worker_tools_enabled(turn_context) { - planned_tools.add(ReportAgentJobResultHandler); - } - } } #[instrument(level = "trace", skip_all, fields(dynamic_tool_count = context.dynamic_tools.len()))] diff --git a/codex-rs/core/src/tools/spec_plan_tests.rs b/codex-rs/core/src/tools/spec_plan_tests.rs index e8b7ba732b..1422c61b84 100644 --- a/codex-rs/core/src/tools/spec_plan_tests.rs +++ b/codex-rs/core/src/tools/spec_plan_tests.rs @@ -15,8 +15,6 @@ use codex_protocol::openai_models::ConfigShellToolType; use codex_protocol::openai_models::InputModality; use codex_protocol::openai_models::ToolMode; use codex_protocol::openai_models::WebSearchToolType; -use codex_protocol::protocol::SessionSource; -use codex_protocol::protocol::SubAgentSource; use codex_tools::DiscoverablePluginInfo; use codex_tools::DiscoverableTool; use codex_tools::ResponsesApiNamespaceTool; @@ -704,24 +702,6 @@ async fn environment_tools_follow_the_step_context() { plan.assert_visible_contains(&["exec_command", "apply_patch", "view_image"]); } -#[tokio::test] -async fn host_context_gates_agent_job_tools() { - let normal_agent_job = probe(|turn| { - set_feature(turn, Feature::SpawnCsv, /*enabled*/ true); - }) - .await; - normal_agent_job.assert_visible_contains(&["spawn_agents_on_csv"]); - normal_agent_job.assert_visible_lacks(&["report_agent_job_result"]); - - let worker_agent_job = probe(|turn| { - set_feature(turn, Feature::SpawnCsv, /*enabled*/ true); - turn.session_source = - SessionSource::SubAgent(SubAgentSource::Other("agent_job:42".to_string())); - }) - .await; - worker_agent_job.assert_visible_contains(&["spawn_agents_on_csv", "report_agent_job_result"]); -} - #[tokio::test] async fn sleep_tool_follows_current_time_config() { let disabled = probe(|turn| { diff --git a/codex-rs/core/src/turn_metadata_tests.rs b/codex-rs/core/src/turn_metadata_tests.rs index 83649b9187..788a785184 100644 --- a/codex-rs/core/src/turn_metadata_tests.rs +++ b/codex-rs/core/src/turn_metadata_tests.rs @@ -341,10 +341,6 @@ fn turn_metadata_state_includes_known_parent_for_non_thread_spawn_subagents_with let sources = [ (SubAgentSource::Review, "review"), (SubAgentSource::Other("guardian".to_string()), "guardian"), - ( - SubAgentSource::Other("agent_job:job-1".to_string()), - "agent_job:job-1", - ), ]; for (subagent_source, subagent_kind) in sources { diff --git a/codex-rs/core/tests/suite/agent_jobs.rs b/codex-rs/core/tests/suite/agent_jobs.rs deleted file mode 100644 index 3601cd9d7f..0000000000 --- a/codex-rs/core/tests/suite/agent_jobs.rs +++ /dev/null @@ -1,469 +0,0 @@ -use anyhow::Result; -use codex_features::Feature; -use codex_protocol::openai_models::ReasoningEffort; -use core_test_support::responses::ev_completed; -use core_test_support::responses::ev_function_call; -use core_test_support::responses::ev_response_created; -use core_test_support::responses::sse; -use core_test_support::responses::sse_response; -use core_test_support::responses::start_mock_server; -use core_test_support::test_codex::test_codex; -use pretty_assertions::assert_eq; -use regex_lite::Regex; -use serde_json::Value; -use serde_json::json; -use std::fs; -use std::sync::Arc; -use std::sync::atomic::AtomicBool; -use std::sync::atomic::AtomicUsize; -use std::sync::atomic::Ordering; -use wiremock::Mock; -use wiremock::Respond; -use wiremock::ResponseTemplate; -use wiremock::matchers::method; -use wiremock::matchers::path_regex; - -struct AgentJobsResponder { - spawn_args_json: String, - seen_main: AtomicBool, - call_counter: AtomicUsize, -} - -impl AgentJobsResponder { - fn new(spawn_args_json: String) -> Self { - Self { - spawn_args_json, - seen_main: AtomicBool::new(false), - call_counter: AtomicUsize::new(0), - } - } -} - -struct StopAfterFirstResponder { - spawn_args_json: String, - seen_main: AtomicBool, - worker_calls: Arc, -} - -impl StopAfterFirstResponder { - fn new(spawn_args_json: String, worker_calls: Arc) -> Self { - Self { - spawn_args_json, - seen_main: AtomicBool::new(false), - worker_calls, - } - } -} - -impl Respond for StopAfterFirstResponder { - 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 let Some((job_id, item_id)) = extract_job_and_item(&body) { - let call_index = self.worker_calls.fetch_add(1, Ordering::SeqCst); - let call_id = format!("call-worker-{call_index}"); - let stop = call_index == 0; - let args = json!({ - "job_id": job_id, - "item_id": item_id, - "result": { "item_id": item_id }, - "stop": stop, - }); - let args_json = serde_json::to_string(&args).expect("worker args should serialize"); - return sse_response(sse(vec![ - ev_response_created("resp-worker"), - ev_function_call(&call_id, "report_agent_job_result", &args_json), - ev_completed("resp-worker"), - ])); - } - - 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); - 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 let Some((job_id, item_id)) = extract_job_and_item(&body) { - let call_id = format!( - "call-worker-{}", - self.call_counter.fetch_add(1, Ordering::SeqCst) - ); - let args = json!({ - "job_id": job_id, - "item_id": item_id, - "result": { "item_id": item_id } - }); - let args_json = serde_json::to_string(&args).expect("worker args should serialize"); - return sse_response(sse(vec![ - ev_response_created("resp-worker"), - ev_function_call(&call_id, "report_agent_job_result", &args_json), - ev_completed("resp-worker"), - ])); - } - - 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"), - ])) - } -} - -fn decode_body_bytes(request: &wiremock::Request) -> Vec { - let Some(encoding) = request - .headers - .get("content-encoding") - .and_then(|value| value.to_str().ok()) - else { - return request.body.clone(); - }; - if encoding - .split(',') - .any(|entry| entry.trim().eq_ignore_ascii_case("zstd")) - { - zstd::stream::decode_all(std::io::Cursor::new(&request.body)) - .unwrap_or_else(|_| request.body.clone()) - } else { - request.body.clone() - } -} - -fn has_function_call_output(body: &Value) -> bool { - body.get("input") - .and_then(Value::as_array) - .is_some_and(|items| { - items.iter().any(|item| { - item.get("type").and_then(Value::as_str) == Some("function_call_output") - }) - }) -} - -fn extract_job_and_item(body: &Value) -> Option<(String, String)> { - let texts = message_input_texts(body); - let mut combined = texts.join("\n"); - if let Some(instructions) = body.get("instructions").and_then(Value::as_str) { - combined.push('\n'); - combined.push_str(instructions); - } - if !combined.contains("You are processing one item for a generic agent job.") { - return None; - } - let job_id = Regex::new(r"Job ID:\s*([^\n]+)") - .ok()? - .captures(&combined) - .and_then(|caps| caps.get(1)) - .map(|m| m.as_str().trim().to_string())?; - let item_id = Regex::new(r"Item ID:\s*([^\n]+)") - .ok()? - .captures(&combined) - .and_then(|caps| caps.get(1)) - .map(|m| m.as_str().trim().to_string())?; - Some((job_id, item_id)) -} - -fn message_input_texts(body: &Value) -> Vec { - let Some(items) = body.get("input").and_then(Value::as_array) else { - return Vec::new(); - }; - items - .iter() - .filter(|item| item.get("type").and_then(Value::as_str) == Some("message")) - .filter_map(|item| item.get("content").and_then(Value::as_array)) - .flatten() - .filter(|span| span.get("type").and_then(Value::as_str) == Some("input_text")) - .filter_map(|span| span.get("text").and_then(Value::as_str)) - .map(str::to_string) - .collect() -} - -fn parse_simple_csv_line(line: &str) -> Vec { - line.split(',').map(str::to_string).collect() -} - -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn report_agent_job_result_rejects_wrong_thread() -> 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_wrong_thread.csv"); - let output_path = test.cwd_path().join("agent_jobs_wrong_thread_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 = AgentJobsResponder::new(args_json); - Mock::given(method("POST")) - .and(path_regex(".*/responses$")) - .respond_with(responder) - .mount(&server) - .await; - - test.submit_turn("run job").await?; - - let db = test.codex.state_db().expect("state db"); - let output = fs::read_to_string(&output_path)?; - let rows: Vec<&str> = output.lines().skip(1).collect(); - assert_eq!(rows.len(), 1); - let job_id = rows - .first() - .and_then(|line| { - parse_simple_csv_line(line) - .iter() - .find(|value| value.len() == 36) - .cloned() - }) - .expect("job_id from csv"); - let job = db.get_agent_job(job_id.as_str()).await?.expect("job"); - let items = db - .list_agent_job_items(job.id.as_str(), /*status*/ None, Some(10)) - .await?; - let item = items.first().expect("item"); - let wrong_thread_id = "00000000-0000-0000-0000-000000000000"; - let accepted = db - .report_agent_job_item_result( - job.id.as_str(), - item.item_id.as_str(), - wrong_thread_id, - &json!({ "wrong": true }), - ) - .await?; - assert!(!accepted); - Ok(()) -} - -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn spawn_agents_on_csv_runs_and_exports() -> 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"); - config.agent_default_subagent_model = Some("gpt-5.4".to_string()); - config.agent_default_subagent_reasoning_effort = Some(ReasoningEffort::High); - }); - let test = builder.build(&server).await?; - - let input_path = test.cwd_path().join("agent_jobs_input.csv"); - let output_path = test.cwd_path().join("agent_jobs_output.csv"); - fs::write(&input_path, "path,area\nfile-1,test\nfile-2,test\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 = AgentJobsResponder::new(args_json); - Mock::given(method("POST")) - .and(path_regex(".*/responses$")) - .respond_with(responder) - .mount(&server) - .await; - - test.submit_turn("run batch job").await?; - - let output = fs::read_to_string(&output_path)?; - assert!(output.contains("result_json")); - assert!(output.contains("item_id")); - assert!(output.contains("\"item_id\"")); - let worker_settings = server - .received_requests() - .await - .unwrap_or_default() - .into_iter() - .filter_map(|request| { - let body = serde_json::from_slice::(&decode_body_bytes(&request)).ok()?; - if has_function_call_output(&body) { - return None; - } - extract_job_and_item(&body)?; - Some(( - body["model"].as_str().map(str::to_string), - body["reasoning"]["effort"].as_str().map(str::to_string), - )) - }) - .collect::>(); - assert_eq!( - worker_settings, - vec![(Some("gpt-5.4".to_string()), Some("high".to_string())); 2] - ); - 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; - - 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_dupe.csv"); - let output_path = test.cwd_path().join("agent_jobs_dupe_out.csv"); - fs::write(&input_path, "id,path\nfoo,alpha\nfoo,beta\n")?; - - let args = json!({ - "csv_path": input_path.display().to_string(), - "instruction": "Return {path}", - "id_column": "id", - "output_csv_path": output_path.display().to_string(), - }); - let args_json = serde_json::to_string(&args)?; - - let responder = AgentJobsResponder::new(args_json); - Mock::given(method("POST")) - .and(path_regex(".*/responses$")) - .respond_with(responder) - .mount(&server) - .await; - - test.submit_turn("run batch job with duplicate ids").await?; - - let output = fs::read_to_string(&output_path)?; - let mut lines = output.lines(); - let headers = lines.next().expect("csv headers"); - let header_cols = parse_simple_csv_line(headers); - let item_id_index = header_cols - .iter() - .position(|header| header == "item_id") - .expect("item_id column"); - - let mut item_ids = Vec::new(); - for line in lines { - let cols = parse_simple_csv_line(line); - item_ids.push(cols[item_id_index].clone()); - } - item_ids.sort(); - item_ids.dedup(); - assert_eq!(item_ids.len(), 2); - assert!(item_ids.contains(&"foo".to_string())); - assert!(item_ids.contains(&"foo-2".to_string())); - Ok(()) -} - -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn spawn_agents_on_csv_stop_halts_future_items() -> 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_stop.csv"); - let output_path = test.cwd_path().join("agent_jobs_stop_out.csv"); - fs::write(&input_path, "path\nfile-1\nfile-2\nfile-3\n")?; - - let args = json!({ - "csv_path": input_path.display().to_string(), - "instruction": "Return {path}", - "output_csv_path": output_path.display().to_string(), - "max_concurrency": 1, - }); - let args_json = serde_json::to_string(&args)?; - - let worker_calls = Arc::new(AtomicUsize::new(0)); - let responder = StopAfterFirstResponder::new(args_json, worker_calls.clone()); - Mock::given(method("POST")) - .and(path_regex(".*/responses$")) - .respond_with(responder) - .mount(&server) - .await; - - test.submit_turn("run job").await?; - - let output = fs::read_to_string(&output_path)?; - let rows: Vec<&str> = output.lines().skip(1).collect(); - assert_eq!(rows.len(), 3); - let job_id = rows - .first() - .and_then(|line| { - parse_simple_csv_line(line) - .iter() - .find(|value| value.len() == 36) - .cloned() - }) - .expect("job_id from csv"); - let db = test.codex.state_db().expect("state db"); - let job = db.get_agent_job(job_id.as_str()).await?.expect("job"); - assert_eq!(job.status, codex_state::AgentJobStatus::Cancelled); - let progress = db.get_agent_job_progress(job_id.as_str()).await?; - assert_eq!(progress.total_items, 3); - assert_eq!(progress.completed_items, 1); - assert_eq!(progress.failed_items, 0); - assert_eq!(progress.running_items, 0); - assert_eq!(progress.pending_items, 2); - assert_eq!(worker_calls.load(Ordering::SeqCst), 1); - Ok(()) -} diff --git a/codex-rs/core/tests/suite/mod.rs b/codex-rs/core/tests/suite/mod.rs index 42695e9bc2..20dc354c85 100644 --- a/codex-rs/core/tests/suite/mod.rs +++ b/codex-rs/core/tests/suite/mod.rs @@ -31,7 +31,6 @@ pub static CODEX_ALIASES_TEMP_DIR: Option = { mod abort_tasks; mod additional_context; mod agent_execution; -mod agent_jobs; mod agent_websocket; mod agents_md; mod apply_patch_cli; diff --git a/codex-rs/features/src/lib.rs b/codex-rs/features/src/lib.rs index 65bd7ae5d5..0b0f30a223 100644 --- a/codex-rs/features/src/lib.rs +++ b/codex-rs/features/src/lib.rs @@ -152,7 +152,7 @@ pub enum Feature { MultiAgentV2, /// Removed compatibility flag retained as a no-op. MultiAgentMode, - /// Enable CSV-backed agent job tools. + /// Removed compatibility flag for the deleted agent-job tools. SpawnCsv, /// Enable apps. Apps, @@ -555,9 +555,6 @@ impl Features { } pub fn normalize_dependencies(&mut self) { - if self.enabled(Feature::SpawnCsv) && !self.enabled(Feature::Collab) { - self.enable(Feature::Collab); - } if self.enabled(Feature::CodeModeOnly) && !self.enabled(Feature::CodeMode) { self.enable(Feature::CodeMode); } @@ -1061,7 +1058,7 @@ pub const FEATURES: &[FeatureSpec] = &[ FeatureSpec { id: Feature::SpawnCsv, key: "enable_fanout", - stage: Stage::UnderDevelopment, + stage: Stage::Removed, default_enabled: false, }, FeatureSpec { diff --git a/codex-rs/features/src/tests.rs b/codex-rs/features/src/tests.rs index 8aabb63cda..898e5db268 100644 --- a/codex-rs/features/src/tests.rs +++ b/codex-rs/features/src/tests.rs @@ -191,21 +191,6 @@ fn codex_hooks_is_legacy_alias_for_hooks() { assert_eq!(feature_for_key("codex_hooks"), Some(Feature::CodexHooks)); } -#[test] -fn enable_fanout_normalization_enables_multi_agent_one_way() { - let mut enable_fanout_features = Features::with_defaults(); - enable_fanout_features.enable(Feature::SpawnCsv); - enable_fanout_features.normalize_dependencies(); - assert_eq!(enable_fanout_features.enabled(Feature::SpawnCsv), true); - assert_eq!(enable_fanout_features.enabled(Feature::Collab), true); - - let mut collab_features = Features::with_defaults(); - collab_features.enable(Feature::Collab); - collab_features.normalize_dependencies(); - assert_eq!(collab_features.enabled(Feature::Collab), true); - assert_eq!(collab_features.enabled(Feature::SpawnCsv), false); -} - #[test] fn apps_require_feature_flag_and_chatgpt_auth() { let mut features = Features::with_defaults(); diff --git a/codex-rs/memories/write/src/phase2.rs b/codex-rs/memories/write/src/phase2.rs index d6673f22ca..76e3435a76 100644 --- a/codex-rs/memories/write/src/phase2.rs +++ b/codex-rs/memories/write/src/phase2.rs @@ -326,7 +326,6 @@ mod agent { // Approval policy agent_config.permissions.approval_policy = Constrained::allow_only(AskForApproval::Never); // Consolidation runs as an internal worker and must not recursively delegate. - let _ = agent_config.features.disable(Feature::SpawnCsv); let _ = agent_config.features.disable(Feature::Collab); let _ = agent_config.features.disable(Feature::MemoryTool); let _ = agent_config.features.disable(Feature::Apps); diff --git a/codex-rs/state/migrations/0042_drop_agent_jobs.sql b/codex-rs/state/migrations/0042_drop_agent_jobs.sql new file mode 100644 index 0000000000..921006b1a6 --- /dev/null +++ b/codex-rs/state/migrations/0042_drop_agent_jobs.sql @@ -0,0 +1,2 @@ +DROP TABLE IF EXISTS agent_job_items; +DROP TABLE IF EXISTS agent_jobs; diff --git a/codex-rs/state/src/lib.rs b/codex-rs/state/src/lib.rs index e0e8e44565..5df56e96ef 100644 --- a/codex-rs/state/src/lib.rs +++ b/codex-rs/state/src/lib.rs @@ -34,13 +34,6 @@ pub use audit::read_thread_state_audit_rows; /// Most consumers should prefer [`StateRuntime`]. pub use extract::apply_rollout_item; pub use extract::rollout_item_affects_thread_metadata; -pub use model::AgentJob; -pub use model::AgentJobCreateParams; -pub use model::AgentJobItem; -pub use model::AgentJobItemCreateParams; -pub use model::AgentJobItemStatus; -pub use model::AgentJobProgress; -pub use model::AgentJobStatus; pub use model::Anchor; pub use model::BackfillState; pub use model::BackfillStats; diff --git a/codex-rs/state/src/migrations_tests.rs b/codex-rs/state/src/migrations_tests.rs index 4243211929..05c30e2d1a 100644 --- a/codex-rs/state/src/migrations_tests.rs +++ b/codex-rs/state/src/migrations_tests.rs @@ -27,6 +27,99 @@ fn migrator_through(version: i64) -> Migrator { } } +#[tokio::test] +async fn agent_job_tables_are_dropped_when_upgrading() { + let sqlite_home = crate::runtime::test_support::unique_temp_dir(); + tokio::fs::create_dir_all(&sqlite_home) + .await + .expect("sqlite home should be created"); + let _cleanup = scopeguard::guard(sqlite_home.clone(), |sqlite_home| { + let _ = std::fs::remove_dir_all(sqlite_home); + }); + let sqlite = crate::SqliteConfig::new_for_testing(sqlite_home.clone()); + let pool = sqlite + .open_read_write_pool(&state_db_path(&sqlite_home)) + .await + .expect("sqlite database should open"); + migrator_through(/*version*/ 15) + .run(&pool) + .await + .expect("agent job migrations should apply"); + + sqlx::query( + r#" +INSERT INTO agent_jobs ( + id, + name, + status, + instruction, + input_headers_json, + input_csv_path, + output_csv_path, + created_at, + updated_at +) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + "#, + ) + .bind("job-1") + .bind("legacy job") + .bind("running") + .bind("process rows") + .bind(r#"["path"]"#) + .bind("/tmp/input.csv") + .bind("/tmp/output.csv") + .bind(1_700_000_000_i64) + .bind(1_700_000_000_i64) + .execute(&pool) + .await + .expect("legacy agent job should insert"); + sqlx::query( + r#" +INSERT INTO agent_job_items ( + job_id, + item_id, + row_index, + row_json, + status, + result_json, + created_at, + updated_at +) VALUES (?, ?, ?, ?, ?, ?, ?, ?) + "#, + ) + .bind("job-1") + .bind("item-1") + .bind(0_i64) + .bind(r#"{"path":"secret.csv"}"#) + .bind("completed") + .bind(r#"{"result":"legacy"}"#) + .bind(1_700_000_000_i64) + .bind(1_700_000_000_i64) + .execute(&pool) + .await + .expect("legacy agent job item should insert"); + + STATE_MIGRATOR + .run(&pool) + .await + .expect("current migrations should apply"); + + let agent_job_tables = sqlx::query_scalar::<_, String>( + r#" +SELECT name +FROM sqlite_master +WHERE type = 'table' AND name IN ('agent_jobs', 'agent_job_items') +ORDER BY name + "#, + ) + .fetch_all(&pool) + .await + .expect("remaining agent job tables should load"); + assert_eq!(agent_job_tables, Vec::::new()); + + pool.close().await; +} + #[tokio::test] async fn recency_migration_backfills_and_seeds_old_binary_inserts() { let sqlite_home = crate::runtime::test_support::unique_temp_dir(); diff --git a/codex-rs/state/src/model/agent_job.rs b/codex-rs/state/src/model/agent_job.rs deleted file mode 100644 index 36f9ff12da..0000000000 --- a/codex-rs/state/src/model/agent_job.rs +++ /dev/null @@ -1,256 +0,0 @@ -use anyhow::Result; -use chrono::DateTime; -use chrono::Utc; -use serde_json::Value; - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum AgentJobStatus { - Pending, - Running, - Completed, - Failed, - Cancelled, -} - -impl AgentJobStatus { - pub const fn as_str(self) -> &'static str { - match self { - AgentJobStatus::Pending => "pending", - AgentJobStatus::Running => "running", - AgentJobStatus::Completed => "completed", - AgentJobStatus::Failed => "failed", - AgentJobStatus::Cancelled => "cancelled", - } - } - - pub fn parse(value: &str) -> Result { - match value { - "pending" => Ok(Self::Pending), - "running" => Ok(Self::Running), - "completed" => Ok(Self::Completed), - "failed" => Ok(Self::Failed), - "cancelled" => Ok(Self::Cancelled), - _ => Err(anyhow::anyhow!("invalid agent job status: {value}")), - } - } - - pub fn is_final(self) -> bool { - matches!( - self, - AgentJobStatus::Completed | AgentJobStatus::Failed | AgentJobStatus::Cancelled - ) - } -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum AgentJobItemStatus { - Pending, - Running, - Completed, - Failed, -} - -impl AgentJobItemStatus { - pub const fn as_str(self) -> &'static str { - match self { - AgentJobItemStatus::Pending => "pending", - AgentJobItemStatus::Running => "running", - AgentJobItemStatus::Completed => "completed", - AgentJobItemStatus::Failed => "failed", - } - } - - pub fn parse(value: &str) -> Result { - match value { - "pending" => Ok(Self::Pending), - "running" => Ok(Self::Running), - "completed" => Ok(Self::Completed), - "failed" => Ok(Self::Failed), - _ => Err(anyhow::anyhow!("invalid agent job item status: {value}")), - } - } -} - -#[derive(Debug, Clone, PartialEq)] -pub struct AgentJob { - pub id: String, - pub name: String, - pub status: AgentJobStatus, - pub instruction: String, - pub auto_export: bool, - pub max_runtime_seconds: Option, - // TODO(jif-oai): Convert to JSON Schema and enforce structured outputs. - pub output_schema_json: Option, - pub input_headers: Vec, - pub input_csv_path: String, - pub output_csv_path: String, - pub created_at: DateTime, - pub updated_at: DateTime, - pub started_at: Option>, - pub completed_at: Option>, - pub last_error: Option, -} - -#[derive(Debug, Clone, PartialEq)] -pub struct AgentJobItem { - pub job_id: String, - pub item_id: String, - pub row_index: i64, - pub source_id: Option, - pub row_json: Value, - pub status: AgentJobItemStatus, - pub assigned_thread_id: Option, - pub attempt_count: i64, - pub result_json: Option, - pub last_error: Option, - pub created_at: DateTime, - pub updated_at: DateTime, - pub completed_at: Option>, - pub reported_at: Option>, -} - -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct AgentJobProgress { - pub total_items: usize, - pub pending_items: usize, - pub running_items: usize, - pub completed_items: usize, - pub failed_items: usize, -} - -#[derive(Debug, Clone)] -pub struct AgentJobCreateParams { - pub id: String, - pub name: String, - pub instruction: String, - pub auto_export: bool, - pub max_runtime_seconds: Option, - pub output_schema_json: Option, - pub input_headers: Vec, - pub input_csv_path: String, - pub output_csv_path: String, -} - -#[derive(Debug, Clone)] -pub struct AgentJobItemCreateParams { - pub item_id: String, - pub row_index: i64, - pub source_id: Option, - pub row_json: Value, -} - -#[derive(Debug, sqlx::FromRow)] -pub(crate) struct AgentJobRow { - pub(crate) id: String, - pub(crate) name: String, - pub(crate) status: String, - pub(crate) instruction: String, - pub(crate) auto_export: i64, - pub(crate) max_runtime_seconds: Option, - pub(crate) output_schema_json: Option, - pub(crate) input_headers_json: String, - pub(crate) input_csv_path: String, - pub(crate) output_csv_path: String, - pub(crate) created_at: i64, - pub(crate) updated_at: i64, - pub(crate) started_at: Option, - pub(crate) completed_at: Option, - pub(crate) last_error: Option, -} - -impl TryFrom for AgentJob { - type Error = anyhow::Error; - - fn try_from(value: AgentJobRow) -> Result { - let output_schema_json = value - .output_schema_json - .as_deref() - .map(serde_json::from_str) - .transpose()?; - let input_headers = serde_json::from_str(value.input_headers_json.as_str())?; - let max_runtime_seconds = value - .max_runtime_seconds - .map(u64::try_from) - .transpose() - .map_err(|_| anyhow::anyhow!("invalid max_runtime_seconds value"))?; - Ok(Self { - id: value.id, - name: value.name, - status: AgentJobStatus::parse(value.status.as_str())?, - instruction: value.instruction, - auto_export: value.auto_export != 0, - max_runtime_seconds, - output_schema_json, - input_headers, - input_csv_path: value.input_csv_path, - output_csv_path: value.output_csv_path, - created_at: epoch_seconds_to_datetime(value.created_at)?, - updated_at: epoch_seconds_to_datetime(value.updated_at)?, - started_at: value - .started_at - .map(epoch_seconds_to_datetime) - .transpose()?, - completed_at: value - .completed_at - .map(epoch_seconds_to_datetime) - .transpose()?, - last_error: value.last_error, - }) - } -} - -#[derive(Debug, sqlx::FromRow)] -pub(crate) struct AgentJobItemRow { - pub(crate) job_id: String, - pub(crate) item_id: String, - pub(crate) row_index: i64, - pub(crate) source_id: Option, - pub(crate) row_json: String, - pub(crate) status: String, - pub(crate) assigned_thread_id: Option, - pub(crate) attempt_count: i64, - pub(crate) result_json: Option, - pub(crate) last_error: Option, - pub(crate) created_at: i64, - pub(crate) updated_at: i64, - pub(crate) completed_at: Option, - pub(crate) reported_at: Option, -} - -impl TryFrom for AgentJobItem { - type Error = anyhow::Error; - - fn try_from(value: AgentJobItemRow) -> Result { - Ok(Self { - job_id: value.job_id, - item_id: value.item_id, - row_index: value.row_index, - source_id: value.source_id, - row_json: serde_json::from_str(value.row_json.as_str())?, - status: AgentJobItemStatus::parse(value.status.as_str())?, - assigned_thread_id: value.assigned_thread_id, - attempt_count: value.attempt_count, - result_json: value - .result_json - .as_deref() - .map(serde_json::from_str) - .transpose()?, - last_error: value.last_error, - created_at: epoch_seconds_to_datetime(value.created_at)?, - updated_at: epoch_seconds_to_datetime(value.updated_at)?, - completed_at: value - .completed_at - .map(epoch_seconds_to_datetime) - .transpose()?, - reported_at: value - .reported_at - .map(epoch_seconds_to_datetime) - .transpose()?, - }) - } -} - -fn epoch_seconds_to_datetime(secs: i64) -> Result> { - DateTime::::from_timestamp(secs, 0) - .ok_or_else(|| anyhow::anyhow!("invalid unix timestamp: {secs}")) -} diff --git a/codex-rs/state/src/model/mod.rs b/codex-rs/state/src/model/mod.rs index 544ef0b8d5..f99aa7cf35 100644 --- a/codex-rs/state/src/model/mod.rs +++ b/codex-rs/state/src/model/mod.rs @@ -1,4 +1,3 @@ -mod agent_job; mod backfill_state; mod graph; mod log; @@ -6,13 +5,6 @@ mod memories; mod thread_goal; mod thread_metadata; -pub use agent_job::AgentJob; -pub use agent_job::AgentJobCreateParams; -pub use agent_job::AgentJobItem; -pub use agent_job::AgentJobItemCreateParams; -pub use agent_job::AgentJobItemStatus; -pub use agent_job::AgentJobProgress; -pub use agent_job::AgentJobStatus; pub use backfill_state::BackfillState; pub use backfill_state::BackfillStatus; pub use graph::DirectionalThreadSpawnEdgeStatus; @@ -36,8 +28,6 @@ pub use thread_metadata::ThreadMetadataBuilder; pub use thread_metadata::ThreadRelationFilter; pub use thread_metadata::ThreadsPage; -pub(crate) use agent_job::AgentJobItemRow; -pub(crate) use agent_job::AgentJobRow; pub(crate) use thread_goal::ThreadGoalRow; pub(crate) use thread_metadata::ThreadRow; pub(crate) use thread_metadata::anchor_from_item; diff --git a/codex-rs/state/src/runtime.rs b/codex-rs/state/src/runtime.rs index a80b59a4f8..aa6d72a6d7 100644 --- a/codex-rs/state/src/runtime.rs +++ b/codex-rs/state/src/runtime.rs @@ -1,10 +1,3 @@ -use crate::AgentJob; -use crate::AgentJobCreateParams; -use crate::AgentJobItem; -use crate::AgentJobItemCreateParams; -use crate::AgentJobItemStatus; -use crate::AgentJobProgress; -use crate::AgentJobStatus; use crate::GOALS_DB_FILENAME; use crate::LOGS_DB_FILENAME; use crate::LogEntry; @@ -25,7 +18,6 @@ use crate::migrations::runtime_logs_migrator; use crate::migrations::runtime_memories_migrator; use crate::migrations::runtime_state_migrator; use crate::migrations::runtime_thread_history_migrator; -use crate::model::AgentJobRow; use crate::model::ThreadRow; use crate::model::anchor_from_item; use crate::model::datetime_to_epoch_millis; @@ -54,7 +46,6 @@ use std::sync::atomic::AtomicI64; use std::time::Instant; use tracing::warn; -mod agent_jobs; mod backfill; mod external_agent_config_imports; mod goals; diff --git a/codex-rs/state/src/runtime/agent_jobs.rs b/codex-rs/state/src/runtime/agent_jobs.rs deleted file mode 100644 index 3f5526c58d..0000000000 --- a/codex-rs/state/src/runtime/agent_jobs.rs +++ /dev/null @@ -1,684 +0,0 @@ -use super::*; -use crate::model::AgentJobItemRow; - -impl StateRuntime { - pub async fn create_agent_job( - &self, - params: &AgentJobCreateParams, - items: &[AgentJobItemCreateParams], - ) -> anyhow::Result { - let now = Utc::now().timestamp(); - let input_headers_json = serde_json::to_string(¶ms.input_headers)?; - let output_schema_json = params - .output_schema_json - .as_ref() - .map(serde_json::to_string) - .transpose()?; - let max_runtime_seconds = params - .max_runtime_seconds - .map(i64::try_from) - .transpose() - .map_err(|_| anyhow::anyhow!("invalid max_runtime_seconds value"))?; - let mut tx = self.pool.begin().await?; - sqlx::query( - r#" -INSERT INTO agent_jobs ( - id, - name, - status, - instruction, - auto_export, - max_runtime_seconds, - output_schema_json, - input_headers_json, - input_csv_path, - output_csv_path, - created_at, - updated_at, - started_at, - completed_at, - last_error -) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, NULL, NULL) - "#, - ) - .bind(params.id.as_str()) - .bind(params.name.as_str()) - .bind(AgentJobStatus::Pending.as_str()) - .bind(params.instruction.as_str()) - .bind(i64::from(params.auto_export)) - .bind(max_runtime_seconds) - .bind(output_schema_json) - .bind(input_headers_json) - .bind(params.input_csv_path.as_str()) - .bind(params.output_csv_path.as_str()) - .bind(now) - .bind(now) - .execute(&mut *tx) - .await?; - - for item in items { - let row_json = serde_json::to_string(&item.row_json)?; - sqlx::query( - r#" -INSERT INTO agent_job_items ( - job_id, - item_id, - row_index, - source_id, - row_json, - status, - assigned_thread_id, - attempt_count, - result_json, - last_error, - created_at, - updated_at, - completed_at, - reported_at -) VALUES (?, ?, ?, ?, ?, ?, NULL, 0, NULL, NULL, ?, ?, NULL, NULL) - "#, - ) - .bind(params.id.as_str()) - .bind(item.item_id.as_str()) - .bind(item.row_index) - .bind(item.source_id.as_deref()) - .bind(row_json) - .bind(AgentJobItemStatus::Pending.as_str()) - .bind(now) - .bind(now) - .execute(&mut *tx) - .await?; - } - - tx.commit().await?; - - let job_id = params.id.as_str(); - self.get_agent_job(job_id) - .await? - .ok_or_else(|| anyhow::anyhow!("failed to load created agent job {job_id}")) - } - - pub async fn get_agent_job(&self, job_id: &str) -> anyhow::Result> { - let row = sqlx::query_as::<_, AgentJobRow>( - r#" -SELECT - id, - name, - status, - instruction, - auto_export, - max_runtime_seconds, - output_schema_json, - input_headers_json, - input_csv_path, - output_csv_path, - created_at, - updated_at, - started_at, - completed_at, - last_error -FROM agent_jobs -WHERE id = ? - "#, - ) - .bind(job_id) - .fetch_optional(self.pool.as_ref()) - .await?; - row.map(AgentJob::try_from).transpose() - } - - pub async fn list_agent_job_items( - &self, - job_id: &str, - status: Option, - limit: Option, - ) -> anyhow::Result> { - let mut builder = QueryBuilder::::new( - r#" -SELECT - job_id, - item_id, - row_index, - source_id, - row_json, - status, - assigned_thread_id, - attempt_count, - result_json, - last_error, - created_at, - updated_at, - completed_at, - reported_at -FROM agent_job_items -WHERE job_id = - "#, - ); - builder.push_bind(job_id); - if let Some(status) = status { - builder.push(" AND status = "); - builder.push_bind(status.as_str()); - } - builder.push(" ORDER BY row_index ASC"); - if let Some(limit) = limit { - builder.push(" LIMIT "); - builder.push_bind(limit as i64); - } - let rows: Vec = builder - .build_query_as::() - .fetch_all(self.pool.as_ref()) - .await?; - rows.into_iter().map(AgentJobItem::try_from).collect() - } - - pub async fn get_agent_job_item( - &self, - job_id: &str, - item_id: &str, - ) -> anyhow::Result> { - let row: Option = sqlx::query_as::<_, AgentJobItemRow>( - r#" -SELECT - job_id, - item_id, - row_index, - source_id, - row_json, - status, - assigned_thread_id, - attempt_count, - result_json, - last_error, - created_at, - updated_at, - completed_at, - reported_at -FROM agent_job_items -WHERE job_id = ? AND item_id = ? - "#, - ) - .bind(job_id) - .bind(item_id) - .fetch_optional(self.pool.as_ref()) - .await?; - row.map(AgentJobItem::try_from).transpose() - } - - pub async fn mark_agent_job_running(&self, job_id: &str) -> anyhow::Result<()> { - let now = Utc::now().timestamp(); - sqlx::query( - r#" -UPDATE agent_jobs -SET - status = ?, - updated_at = ?, - started_at = COALESCE(started_at, ?), - completed_at = NULL, - last_error = NULL -WHERE id = ? - "#, - ) - .bind(AgentJobStatus::Running.as_str()) - .bind(now) - .bind(now) - .bind(job_id) - .execute(self.pool.as_ref()) - .await?; - Ok(()) - } - - pub async fn mark_agent_job_completed(&self, job_id: &str) -> anyhow::Result<()> { - let now = Utc::now().timestamp(); - sqlx::query( - r#" -UPDATE agent_jobs -SET status = ?, updated_at = ?, completed_at = ?, last_error = NULL -WHERE id = ? - "#, - ) - .bind(AgentJobStatus::Completed.as_str()) - .bind(now) - .bind(now) - .bind(job_id) - .execute(self.pool.as_ref()) - .await?; - Ok(()) - } - - pub async fn mark_agent_job_failed( - &self, - job_id: &str, - error_message: &str, - ) -> anyhow::Result<()> { - let now = Utc::now().timestamp(); - sqlx::query( - r#" -UPDATE agent_jobs -SET status = ?, updated_at = ?, completed_at = ?, last_error = ? -WHERE id = ? - "#, - ) - .bind(AgentJobStatus::Failed.as_str()) - .bind(now) - .bind(now) - .bind(error_message) - .bind(job_id) - .execute(self.pool.as_ref()) - .await?; - Ok(()) - } - - pub async fn mark_agent_job_cancelled( - &self, - job_id: &str, - reason: &str, - ) -> anyhow::Result { - let now = Utc::now().timestamp(); - let result = sqlx::query( - r#" -UPDATE agent_jobs -SET status = ?, updated_at = ?, completed_at = ?, last_error = ? -WHERE id = ? AND status IN (?, ?) - "#, - ) - .bind(AgentJobStatus::Cancelled.as_str()) - .bind(now) - .bind(now) - .bind(reason) - .bind(job_id) - .bind(AgentJobStatus::Pending.as_str()) - .bind(AgentJobStatus::Running.as_str()) - .execute(self.pool.as_ref()) - .await?; - Ok(result.rows_affected() > 0) - } - - pub async fn is_agent_job_cancelled(&self, job_id: &str) -> anyhow::Result { - let row = sqlx::query( - r#" -SELECT status -FROM agent_jobs -WHERE id = ? - "#, - ) - .bind(job_id) - .fetch_optional(self.pool.as_ref()) - .await?; - let Some(row) = row else { - return Ok(false); - }; - let status: String = row.try_get("status")?; - Ok(AgentJobStatus::parse(status.as_str())? == AgentJobStatus::Cancelled) - } - - pub async fn mark_agent_job_item_running( - &self, - job_id: &str, - item_id: &str, - ) -> anyhow::Result { - let now = Utc::now().timestamp(); - let result = sqlx::query( - r#" -UPDATE agent_job_items -SET - status = ?, - assigned_thread_id = NULL, - attempt_count = attempt_count + 1, - updated_at = ?, - last_error = NULL -WHERE job_id = ? AND item_id = ? AND status = ? - "#, - ) - .bind(AgentJobItemStatus::Running.as_str()) - .bind(now) - .bind(job_id) - .bind(item_id) - .bind(AgentJobItemStatus::Pending.as_str()) - .execute(self.pool.as_ref()) - .await?; - Ok(result.rows_affected() > 0) - } - - pub async fn mark_agent_job_item_running_with_thread( - &self, - job_id: &str, - item_id: &str, - thread_id: &str, - ) -> anyhow::Result { - let now = Utc::now().timestamp(); - let result = sqlx::query( - r#" -UPDATE agent_job_items -SET - status = ?, - assigned_thread_id = ?, - attempt_count = attempt_count + 1, - updated_at = ?, - last_error = NULL -WHERE job_id = ? AND item_id = ? AND status = ? - "#, - ) - .bind(AgentJobItemStatus::Running.as_str()) - .bind(thread_id) - .bind(now) - .bind(job_id) - .bind(item_id) - .bind(AgentJobItemStatus::Pending.as_str()) - .execute(self.pool.as_ref()) - .await?; - Ok(result.rows_affected() > 0) - } - - pub async fn mark_agent_job_item_pending( - &self, - job_id: &str, - item_id: &str, - error_message: Option<&str>, - ) -> anyhow::Result { - let now = Utc::now().timestamp(); - let result = sqlx::query( - r#" -UPDATE agent_job_items -SET - status = ?, - assigned_thread_id = NULL, - updated_at = ?, - last_error = ? -WHERE job_id = ? AND item_id = ? AND status = ? - "#, - ) - .bind(AgentJobItemStatus::Pending.as_str()) - .bind(now) - .bind(error_message) - .bind(job_id) - .bind(item_id) - .bind(AgentJobItemStatus::Running.as_str()) - .execute(self.pool.as_ref()) - .await?; - Ok(result.rows_affected() > 0) - } - - pub async fn set_agent_job_item_thread( - &self, - job_id: &str, - item_id: &str, - thread_id: &str, - ) -> anyhow::Result { - let now = Utc::now().timestamp(); - let result = sqlx::query( - r#" -UPDATE agent_job_items -SET assigned_thread_id = ?, updated_at = ? -WHERE job_id = ? AND item_id = ? AND status = ? - "#, - ) - .bind(thread_id) - .bind(now) - .bind(job_id) - .bind(item_id) - .bind(AgentJobItemStatus::Running.as_str()) - .execute(self.pool.as_ref()) - .await?; - Ok(result.rows_affected() > 0) - } - - pub async fn report_agent_job_item_result( - &self, - job_id: &str, - item_id: &str, - reporting_thread_id: &str, - result_json: &Value, - ) -> anyhow::Result { - let now = Utc::now().timestamp(); - let serialized = serde_json::to_string(result_json)?; - let result = sqlx::query( - r#" -UPDATE agent_job_items -SET - status = ?, - result_json = ?, - reported_at = ?, - completed_at = ?, - updated_at = ?, - last_error = NULL, - assigned_thread_id = NULL -WHERE - job_id = ? - AND item_id = ? - AND status = ? - AND assigned_thread_id = ? - "#, - ) - .bind(AgentJobItemStatus::Completed.as_str()) - .bind(serialized) - .bind(now) - .bind(now) - .bind(now) - .bind(job_id) - .bind(item_id) - .bind(AgentJobItemStatus::Running.as_str()) - .bind(reporting_thread_id) - .execute(self.pool.as_ref()) - .await?; - Ok(result.rows_affected() > 0) - } - - pub async fn mark_agent_job_item_completed( - &self, - job_id: &str, - item_id: &str, - ) -> anyhow::Result { - let now = Utc::now().timestamp(); - let result = sqlx::query( - r#" -UPDATE agent_job_items -SET - status = ?, - completed_at = ?, - updated_at = ?, - assigned_thread_id = NULL -WHERE - job_id = ? - AND item_id = ? - AND status = ? - AND result_json IS NOT NULL - "#, - ) - .bind(AgentJobItemStatus::Completed.as_str()) - .bind(now) - .bind(now) - .bind(job_id) - .bind(item_id) - .bind(AgentJobItemStatus::Running.as_str()) - .execute(self.pool.as_ref()) - .await?; - Ok(result.rows_affected() > 0) - } - - pub async fn mark_agent_job_item_failed( - &self, - job_id: &str, - item_id: &str, - error_message: &str, - ) -> anyhow::Result { - let now = Utc::now().timestamp(); - let result = sqlx::query( - r#" -UPDATE agent_job_items -SET - status = ?, - completed_at = ?, - updated_at = ?, - last_error = ?, - assigned_thread_id = NULL -WHERE - job_id = ? - AND item_id = ? - AND status = ? - "#, - ) - .bind(AgentJobItemStatus::Failed.as_str()) - .bind(now) - .bind(now) - .bind(error_message) - .bind(job_id) - .bind(item_id) - .bind(AgentJobItemStatus::Running.as_str()) - .execute(self.pool.as_ref()) - .await?; - Ok(result.rows_affected() > 0) - } - - pub async fn get_agent_job_progress(&self, job_id: &str) -> anyhow::Result { - let row = sqlx::query( - r#" -SELECT - COUNT(*) AS total_items, - SUM(CASE WHEN status = ? THEN 1 ELSE 0 END) AS pending_items, - SUM(CASE WHEN status = ? THEN 1 ELSE 0 END) AS running_items, - SUM(CASE WHEN status = ? THEN 1 ELSE 0 END) AS completed_items, - SUM(CASE WHEN status = ? THEN 1 ELSE 0 END) AS failed_items -FROM agent_job_items -WHERE job_id = ? - "#, - ) - .bind(AgentJobItemStatus::Pending.as_str()) - .bind(AgentJobItemStatus::Running.as_str()) - .bind(AgentJobItemStatus::Completed.as_str()) - .bind(AgentJobItemStatus::Failed.as_str()) - .bind(job_id) - .fetch_one(self.pool.as_ref()) - .await?; - - let total_items: i64 = row.try_get("total_items")?; - let pending_items: Option = row.try_get("pending_items")?; - let running_items: Option = row.try_get("running_items")?; - let completed_items: Option = row.try_get("completed_items")?; - let failed_items: Option = row.try_get("failed_items")?; - Ok(AgentJobProgress { - total_items: usize::try_from(total_items).unwrap_or_default(), - pending_items: usize::try_from(pending_items.unwrap_or_default()).unwrap_or_default(), - running_items: usize::try_from(running_items.unwrap_or_default()).unwrap_or_default(), - completed_items: usize::try_from(completed_items.unwrap_or_default()) - .unwrap_or_default(), - failed_items: usize::try_from(failed_items.unwrap_or_default()).unwrap_or_default(), - }) - } -} - -#[cfg(test)] -mod tests { - use super::*; - use crate::runtime::test_support::unique_temp_dir; - use pretty_assertions::assert_eq; - use serde_json::json; - - async fn create_running_single_item_job( - runtime: &StateRuntime, - ) -> anyhow::Result<(String, String, String)> { - let job_id = "job-1".to_string(); - let item_id = "item-1".to_string(); - let thread_id = "thread-1".to_string(); - runtime - .create_agent_job( - &AgentJobCreateParams { - id: job_id.clone(), - name: "test-job".to_string(), - instruction: "Return a result".to_string(), - auto_export: true, - max_runtime_seconds: None, - output_schema_json: None, - input_headers: vec!["path".to_string()], - input_csv_path: "/tmp/in.csv".to_string(), - output_csv_path: "/tmp/out.csv".to_string(), - }, - &[AgentJobItemCreateParams { - item_id: item_id.clone(), - row_index: 0, - source_id: None, - row_json: json!({"path":"file-1"}), - }], - ) - .await?; - runtime.mark_agent_job_running(job_id.as_str()).await?; - let marked_running = runtime - .mark_agent_job_item_running_with_thread( - job_id.as_str(), - item_id.as_str(), - thread_id.as_str(), - ) - .await?; - assert!(marked_running); - Ok((job_id, item_id, thread_id)) - } - - #[tokio::test] - async fn report_agent_job_item_result_completes_item_atomically() -> anyhow::Result<()> { - let codex_home = unique_temp_dir(); - let runtime = StateRuntime::init(codex_home, "test-provider".to_string()).await?; - let (job_id, item_id, thread_id) = create_running_single_item_job(runtime.as_ref()).await?; - - let accepted = runtime - .report_agent_job_item_result( - job_id.as_str(), - item_id.as_str(), - thread_id.as_str(), - &json!({"ok": true}), - ) - .await?; - assert!(accepted); - - let item = runtime - .get_agent_job_item(job_id.as_str(), item_id.as_str()) - .await? - .expect("job item should exist"); - assert_eq!(item.status, AgentJobItemStatus::Completed); - assert_eq!(item.result_json, Some(json!({"ok": true}))); - assert_eq!(item.assigned_thread_id, None); - assert_eq!(item.last_error, None); - assert!(item.reported_at.is_some()); - assert!(item.completed_at.is_some()); - let progress = runtime.get_agent_job_progress(job_id.as_str()).await?; - assert_eq!( - progress, - AgentJobProgress { - total_items: 1, - pending_items: 0, - running_items: 0, - completed_items: 1, - failed_items: 0, - } - ); - Ok(()) - } - - #[tokio::test] - async fn report_agent_job_item_result_rejects_late_reports() -> anyhow::Result<()> { - let codex_home = unique_temp_dir(); - let runtime = StateRuntime::init(codex_home, "test-provider".to_string()).await?; - let (job_id, item_id, thread_id) = create_running_single_item_job(runtime.as_ref()).await?; - - let marked_failed = runtime - .mark_agent_job_item_failed(job_id.as_str(), item_id.as_str(), "missing report") - .await?; - assert!(marked_failed); - let accepted = runtime - .report_agent_job_item_result( - job_id.as_str(), - item_id.as_str(), - thread_id.as_str(), - &json!({"late": true}), - ) - .await?; - assert!(!accepted); - - let item = runtime - .get_agent_job_item(job_id.as_str(), item_id.as_str()) - .await? - .expect("job item should exist"); - assert_eq!(item.status, AgentJobItemStatus::Failed); - assert_eq!(item.result_json, None); - assert_eq!(item.last_error, Some("missing report".to_string())); - Ok(()) - } -} diff --git a/codex-rs/state/src/runtime/remote_control.rs b/codex-rs/state/src/runtime/remote_control.rs index 28133f4440..eda2adc3b3 100644 --- a/codex-rs/state/src/runtime/remote_control.rs +++ b/codex-rs/state/src/runtime/remote_control.rs @@ -330,7 +330,14 @@ mod tests { .await .expect("create codex home"); let old_state_migrator = Migrator { - migrations: Cow::Owned(STATE_MIGRATOR.migrations[..36].to_vec()), + migrations: Cow::Owned( + STATE_MIGRATOR + .migrations + .iter() + .filter(|migration| migration.version <= 36) + .cloned() + .collect(), + ), ignore_missing: false, locking: true, no_tx: false, diff --git a/codex-rs/state/src/runtime/threads.rs b/codex-rs/state/src/runtime/threads.rs index 4f2b8b5fed..2cc7b10ebb 100644 --- a/codex-rs/state/src/runtime/threads.rs +++ b/codex-rs/state/src/runtime/threads.rs @@ -1040,70 +1040,12 @@ ON CONFLICT(id) DO UPDATE SET self.thread_goals.delete_thread_goal(*thread_id).await?; } - let now = Utc::now().timestamp(); let mut tx = self.pool.begin().await?; for thread_id_string in &thread_id_strings { - for parent_thread_id_string in &thread_id_strings { - // If both the job runner and worker are being deleted, requeueing - // the worker item would leave a running job with no loop to consume it. - sqlx::query( - r#" -UPDATE agent_jobs -SET status = ?, updated_at = ?, completed_at = ?, last_error = ? -WHERE status IN (?, ?) - AND id IN ( - SELECT item.job_id - FROM agent_job_items AS item - JOIN thread_spawn_edges AS edge ON edge.child_thread_id = item.assigned_thread_id - WHERE item.status = ? AND item.assigned_thread_id = ? AND edge.parent_thread_id = ? - ) - "#, - ) - .bind(AgentJobStatus::Cancelled.as_str()) - .bind(now) - .bind(now) - .bind("agent job runner thread was deleted") - .bind(AgentJobStatus::Pending.as_str()) - .bind(AgentJobStatus::Running.as_str()) - .bind(AgentJobItemStatus::Running.as_str()) - .bind(thread_id_string) - .bind(parent_thread_id_string) - .execute(&mut *tx) - .await?; - } sqlx::query("DELETE FROM thread_dynamic_tools WHERE thread_id = ?") .bind(thread_id_string) .execute(&mut *tx) .await?; - sqlx::query( - r#" -UPDATE agent_job_items -SET - status = ?, - assigned_thread_id = NULL, - updated_at = ?, - last_error = ? -WHERE assigned_thread_id = ? AND status = ? - "#, - ) - .bind(AgentJobItemStatus::Pending.as_str()) - .bind(now) - .bind("assigned thread was deleted") - .bind(thread_id_string) - .bind(AgentJobItemStatus::Running.as_str()) - .execute(&mut *tx) - .await?; - sqlx::query( - r#" -UPDATE agent_job_items -SET assigned_thread_id = NULL, updated_at = ? -WHERE assigned_thread_id = ? - "#, - ) - .bind(now) - .bind(thread_id_string) - .execute(&mut *tx) - .await?; } for thread_id_string in &thread_id_strings { sqlx::query( @@ -1449,7 +1391,6 @@ mod tests { use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::ThreadHistoryMode; use pretty_assertions::assert_eq; - use serde_json::json; use std::path::PathBuf; #[tokio::test] @@ -1536,36 +1477,6 @@ mod tests { .bind("{}") .execute(runtime.pool.as_ref()) .await?; - runtime - .create_agent_job( - &AgentJobCreateParams { - id: "job-1".to_string(), - name: "test-job".to_string(), - instruction: "Return a result".to_string(), - auto_export: true, - max_runtime_seconds: None, - output_schema_json: None, - input_headers: vec!["path".to_string()], - input_csv_path: "/tmp/in.csv".to_string(), - output_csv_path: "/tmp/out.csv".to_string(), - }, - &[AgentJobItemCreateParams { - item_id: "item-1".to_string(), - row_index: 0, - source_id: None, - row_json: json!({"path": "file-1"}), - }], - ) - .await?; - runtime.mark_agent_job_running("job-1").await?; - runtime - .mark_agent_job_item_running_with_thread( - "job-1", - "item-1", - &child_thread_id.to_string(), - ) - .await?; - let rows = runtime .delete_threads_strict(&[thread_id, child_thread_id]) .await?; @@ -1579,21 +1490,6 @@ mod tests { .await?; assert_eq!(dynamic_tool_count, 0); assert_thread_cleanup_state(&runtime, thread_id).await?; - let job_item = runtime - .get_agent_job_item("job-1", "item-1") - .await? - .expect("job item should exist"); - assert_eq!(job_item.status, AgentJobItemStatus::Pending); - assert_eq!(job_item.assigned_thread_id, None); - assert_eq!( - job_item.last_error, - Some("assigned thread was deleted".to_string()) - ); - let job = runtime - .get_agent_job("job-1") - .await? - .expect("job should exist"); - assert_eq!(job.status, AgentJobStatus::Cancelled); let missing_thread_id = ThreadId::from_string("00000000-0000-0000-0000-000000000403")?; let missing_child_thread_id = diff --git a/codex-rs/thread-manager-sample/src/main.rs b/codex-rs/thread-manager-sample/src/main.rs index 63b33e335a..0ab63a271b 100644 --- a/codex-rs/thread-manager-sample/src/main.rs +++ b/codex-rs/thread-manager-sample/src/main.rs @@ -245,7 +245,6 @@ fn new_config(model: Option, arg0_paths: Arg0DispatchPaths) -> anyhow::R agent_max_threads: Some(6), agent_default_subagent_model: None, agent_default_subagent_reasoning_effort: None, - agent_job_max_runtime_seconds: None, agent_interrupt_message_enabled: false, agent_max_depth: 1, agent_roles: BTreeMap::new(), diff --git a/codex-rs/tui/src/debug_config.rs b/codex-rs/tui/src/debug_config.rs index 83f136cc24..088b5c008e 100644 --- a/codex-rs/tui/src/debug_config.rs +++ b/codex-rs/tui/src/debug_config.rs @@ -80,11 +80,6 @@ fn render_agents_config_lines(config: &Config) -> Vec> { format_optional(config.agent_default_subagent_reasoning_effort.as_ref()) ) .into(), - format!( - " - job_max_runtime_seconds = {}", - format_optional(config.agent_job_max_runtime_seconds) - ) - .into(), format!( " - interrupt_message = {}", config.agent_interrupt_message_enabled @@ -651,7 +646,6 @@ max_concurrent_threads_per_session = 7 max_depth = -2 default_subagent_model = "gpt-5.6-terra" default_subagent_reasoning_effort = "high" -job_max_runtime_seconds = 900 interrupt_message = false "#, ) diff --git a/codex-rs/tui/src/snapshots/codex_tui__debug_config__tests__debug_config_output_lists_agents_fields.snap b/codex-rs/tui/src/snapshots/codex_tui__debug_config__tests__debug_config_output_lists_agents_fields.snap index d0e2f25471..e2d272f210 100644 --- a/codex-rs/tui/src/snapshots/codex_tui__debug_config__tests__debug_config_output_lists_agents_fields.snap +++ b/codex-rs/tui/src/snapshots/codex_tui__debug_config__tests__debug_config_output_lists_agents_fields.snap @@ -10,5 +10,4 @@ expression: render_to_text(&render_agents_config_lines(&config)) - max_depth = -2 (V1 only; ignored by V2) - default_subagent_model = gpt-5.6-terra - default_subagent_reasoning_effort = high - - job_max_runtime_seconds = 900 - interrupt_message = false