diff --git a/codex-rs/config/src/config_toml.rs b/codex-rs/config/src/config_toml.rs index 7c060cadb7..12c46b10a4 100644 --- a/codex-rs/config/src/config_toml.rs +++ b/codex-rs/config/src/config_toml.rs @@ -667,9 +667,6 @@ pub struct AgentsToml { /// Root sessions start at depth 0. #[schemars(range(min = 1))] pub max_depth: Option, - /// Default maximum runtime in seconds for agent job workers. - #[schemars(range(min = 1))] - pub job_max_runtime_seconds: Option, /// Whether to record a model-visible message when an agent turn is interrupted. /// Defaults to true. pub interrupt_message: Option, diff --git a/codex-rs/core/config.schema.json b/codex-rs/core/config.schema.json index f5a97b142a..3e1734c4a6 100644 --- a/codex-rs/core/config.schema.json +++ b/codex-rs/core/config.schema.json @@ -40,12 +40,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_depth": { "description": "Maximum nesting depth allowed for spawned agent threads. Root sessions start at depth 0.", "format": "int32", @@ -5218,4 +5212,4 @@ }, "title": "ConfigToml", "type": "object" -} +} \ No newline at end of file diff --git a/codex-rs/core/src/config/config_tests.rs b/codex-rs/core/src/config/config_tests.rs index 588eb7ac1b..855bec8090 100644 --- a/codex-rs/core/src/config/config_tests.rs +++ b/codex-rs/core/src/config/config_tests.rs @@ -6776,7 +6776,6 @@ async fn load_config_rejects_missing_agent_role_config_file() -> std::io::Result agents: Some(AgentsToml { max_threads: None, max_depth: None, - job_max_runtime_seconds: None, interrupt_message: None, roles: BTreeMap::from([( "researcher".to_string(), @@ -7724,7 +7723,6 @@ async fn load_config_normalizes_agent_role_nickname_candidates() -> std::io::Res agents: Some(AgentsToml { max_threads: None, max_depth: None, - job_max_runtime_seconds: None, interrupt_message: None, roles: BTreeMap::from([( "researcher".to_string(), @@ -7767,7 +7765,6 @@ async fn load_config_rejects_empty_agent_role_nickname_candidates() -> std::io:: agents: Some(AgentsToml { max_threads: None, max_depth: None, - job_max_runtime_seconds: None, interrupt_message: None, roles: BTreeMap::from([( "researcher".to_string(), @@ -7804,7 +7801,6 @@ async fn load_config_rejects_duplicate_agent_role_nickname_candidates() -> std:: agents: Some(AgentsToml { max_threads: None, max_depth: None, - job_max_runtime_seconds: None, interrupt_message: None, roles: BTreeMap::from([( "researcher".to_string(), @@ -7841,7 +7837,6 @@ async fn load_config_rejects_unsafe_agent_role_nickname_candidates() -> std::io: agents: Some(AgentsToml { max_threads: None, max_depth: None, - job_max_runtime_seconds: None, interrupt_message: None, roles: BTreeMap::from([( "researcher".to_string(), diff --git a/codex-rs/core/src/config/mod.rs b/codex-rs/core/src/config/mod.rs index b74f2feefe..02892c83d6 100644 --- a/codex-rs/core/src/config/mod.rs +++ b/codex-rs/core/src/config/mod.rs @@ -236,7 +236,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"; @@ -824,9 +823,6 @@ pub struct Config { /// User-configured maximum number of agent threads that can be open concurrently. pub agent_max_threads: 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, @@ -3140,25 +3136,6 @@ impl Config { "agents.max_depth must be at least 1", )); } - 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() @@ -3503,7 +3480,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 d81f84590b..64fe1323e4 100644 --- a/codex-rs/core/src/guardian/review_session.rs +++ b/codex-rs/core/src/guardian/review_session.rs @@ -952,7 +952,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 c081a2aee6..7fd263f224 100644 --- a/codex-rs/core/src/session/config_lock.rs +++ b/codex-rs/core/src/session/config_lock.rs @@ -167,7 +167,6 @@ fn save_config_resolved_fields( config.agent_max_threads }; agents.max_depth = Some(config.agent_max_depth); - 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 9f7eba9573..79e9f8aaab 100644 --- a/codex-rs/core/src/tasks/review.rs +++ b/codex-rs/core/src/tasks/review.rs @@ -108,7 +108,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 360744973c..0000000000 --- a/codex-rs/core/src/tools/handlers/agent_jobs.rs +++ /dev/null @@ -1,762 +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::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 spawn_config = build_agent_spawn_config(&base_instructions, turn.as_ref())?; - 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.into(), - 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 c7adb27610..0000000000 --- a/codex-rs/core/src/tools/handlers/agent_jobs/report_agent_job_result.rs +++ /dev/null @@ -1,93 +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; - -#[async_trait::async_trait] -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() - } - - async fn handle( - &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 26438ca7ba..0000000000 --- a/codex-rs/core/src/tools/handlers/agent_jobs/spawn_agents_on_csv.rs +++ /dev/null @@ -1,311 +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; - -#[async_trait::async_trait] -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() - } - - async fn handle( - &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<&AbsolutePathBuf, FunctionCallError> { - let [turn_environment] = turn.environments.turn_environments.as_slice() else { - 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(), - )); - } - - Ok(&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 d6963f0761..98632566cc 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 dynamic; diff --git a/codex-rs/core/src/tools/spec_plan.rs b/codex-rs/core/src/tools/spec_plan.rs index 47d2ac6726..690ab2184f 100644 --- a/codex-rs/core/src/tools/spec_plan.rs +++ b/codex-rs/core/src/tools/spec_plan.rs @@ -27,8 +27,6 @@ use crate::tools::handlers::ToolSearchHandler; use crate::tools::handlers::UpdateGoalHandler; use crate::tools::handlers::ViewImageHandler; 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; @@ -313,19 +311,6 @@ fn goal_tools_enabled(turn_context: &TurnContext) -> bool { ) } -fn agent_jobs_tools_enabled(turn_context: &TurnContext) -> bool { - turn_context.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_tool_enabled(turn_context: &TurnContext) -> bool { image_generation_runtime_enabled(turn_context) && turn_context @@ -778,13 +763,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); - } - } } fn add_mcp_runtime_tools(context: &CoreToolPlanContext<'_>, planned_tools: &mut PlannedTools) { diff --git a/codex-rs/core/src/tools/spec_plan_tests.rs b/codex-rs/core/src/tools/spec_plan_tests.rs index 17dfdc7a74..ce4136c7cd 100644 --- a/codex-rs/core/src/tools/spec_plan_tests.rs +++ b/codex-rs/core/src/tools/spec_plan_tests.rs @@ -619,7 +619,7 @@ async fn environment_count_controls_environment_backed_tools() { } #[tokio::test] -async fn host_context_gates_goal_and_agent_job_tools() { +async fn host_context_gates_goal_tools() { let feature_disabled = probe(|turn| { set_feature(turn, Feature::Goals, /*enabled*/ false); turn.goal_tools_supported = true; @@ -648,21 +648,6 @@ async fn host_context_gates_goal_and_agent_job_tools() { }) .await; review_thread.assert_visible_lacks(&["get_goal", "create_goal", "update_goal"]); - - 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] diff --git a/codex-rs/core/src/turn_metadata_tests.rs b/codex-rs/core/src/turn_metadata_tests.rs index 32a559848a..0d536dc1fe 100644 --- a/codex-rs/core/src/turn_metadata_tests.rs +++ b/codex-rs/core/src/turn_metadata_tests.rs @@ -306,8 +306,8 @@ fn turn_metadata_state_includes_known_parent_for_non_thread_spawn_subagents_with (SubAgentSource::Review, "review"), (SubAgentSource::Other("guardian".to_string()), "guardian"), ( - SubAgentSource::Other("agent_job:job-1".to_string()), - "agent_job:job-1", + SubAgentSource::Other("custom_worker:job-1".to_string()), + "custom_worker:job-1", ), ]; 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 b275b1878a..0000000000 --- a/codex-rs/core/tests/suite/agent_jobs.rs +++ /dev/null @@ -1,448 +0,0 @@ -use anyhow::Result; -use codex_features::Feature; -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 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).unwrap_or_else(|err| { - panic!("worker args serialize: {err}"); - }); - 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).unwrap_or_else(|err| { - panic!("worker args serialize: {err}"); - }); - 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"); - }); - 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\"")); - 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 1e4a5f9501..66726eee87 100644 --- a/codex-rs/core/tests/suite/mod.rs +++ b/codex-rs/core/tests/suite/mod.rs @@ -30,7 +30,6 @@ pub static CODEX_ALIASES_TEMP_DIR: Option = { #[cfg(not(target_os = "windows"))] mod abort_tasks; mod additional_context; -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 3ba973b804..7289aa0fa5 100644 --- a/codex-rs/features/src/lib.rs +++ b/codex-rs/features/src/lib.rs @@ -135,7 +135,7 @@ pub enum Feature { Collab, /// Enable task-path-based multi-agent routing. MultiAgentV2, - /// Enable CSV-backed agent job tools. + /// Removed compatibility flag for deleted CSV-backed agent job tools. SpawnCsv, /// Enable apps. Apps, @@ -511,9 +511,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); } @@ -967,7 +964,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 aa5e0174ba..e6a48aaf36 100644 --- a/codex-rs/features/src/tests.rs +++ b/codex-rs/features/src/tests.rs @@ -360,24 +360,17 @@ fn multi_agent_is_stable_and_enabled_by_default() { } #[test] -fn enable_fanout_is_under_development() { - assert_eq!(Feature::SpawnCsv.stage(), Stage::UnderDevelopment); +fn enable_fanout_is_removed() { + assert_eq!(Feature::SpawnCsv.stage(), Stage::Removed); assert_eq!(Feature::SpawnCsv.default_enabled(), false); } #[test] -fn enable_fanout_normalization_enables_multi_agent_one_way() { +fn enable_fanout_normalization_is_noop() { 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] diff --git a/codex-rs/memories/write/src/phase2.rs b/codex-rs/memories/write/src/phase2.rs index c78032d9c2..2e0acd0a0f 100644 --- a/codex-rs/memories/write/src/phase2.rs +++ b/codex-rs/memories/write/src/phase2.rs @@ -311,7 +311,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/thread-manager-sample/src/main.rs b/codex-rs/thread-manager-sample/src/main.rs index 47e0fc3911..c89ada3f26 100644 --- a/codex-rs/thread-manager-sample/src/main.rs +++ b/codex-rs/thread-manager-sample/src/main.rs @@ -224,7 +224,6 @@ fn new_config(model: Option, arg0_paths: Arg0DispatchPaths) -> anyhow::R project_doc_fallback_filenames: Vec::new(), tool_output_token_limit: None, agent_max_threads: Some(6), - agent_job_max_runtime_seconds: None, agent_interrupt_message_enabled: false, agent_max_depth: 1, agent_roles: BTreeMap::new(),