diff --git a/codex-rs/core/src/tools/handlers/agent_jobs.rs b/codex-rs/core/src/tools/handlers/agent_jobs.rs index ddf5e8f71f..fd5e54d72c 100644 --- a/codex-rs/core/src/tools/handlers/agent_jobs.rs +++ b/codex-rs/core/src/tools/handlers/agent_jobs.rs @@ -18,7 +18,6 @@ use codex_protocol::ThreadId; use codex_protocol::models::FunctionCallOutputBody; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::SubAgentSource; -use once_cell::sync::Lazy; use serde::Deserialize; use serde::Serialize; use serde_json::Value; @@ -27,22 +26,14 @@ use std::collections::HashSet; use std::path::Path; use std::path::PathBuf; use std::sync::Arc; -use tokio::sync::Mutex; -use tokio::task::JoinHandle; use tokio::time::Duration; use tokio::time::Instant; use uuid::Uuid; pub struct AgentJobsHandler; -const MIN_WAIT_TIMEOUT_MS: i64 = 1_000; -const DEFAULT_WAIT_TIMEOUT_MS: i64 = 30_000; -const MAX_WAIT_TIMEOUT_MS: i64 = 300_000; const STATUS_POLL_INTERVAL_MS: u64 = 250; -static ACTIVE_JOB_RUNNERS: Lazy>>> = - Lazy::new(|| Mutex::new(HashMap::new())); - #[derive(Debug, Deserialize)] struct SpawnAgentsOnCsvArgs { csv_path: String, @@ -52,25 +43,6 @@ struct SpawnAgentsOnCsvArgs { output_csv_path: Option, output_schema: Option, max_concurrency: Option, - auto_export: Option, -} - -#[derive(Debug, Deserialize)] -struct JobIdArgs { - job_id: String, -} - -#[derive(Debug, Deserialize)] -struct WaitAgentJobArgs { - job_id: String, - timeout_ms: Option, -} - -#[derive(Debug, Deserialize)] -struct ExportAgentJobCsvArgs { - job_id: String, - #[serde(alias = "path")] - output_csv_path: Option, } #[derive(Debug, Deserialize)] @@ -81,37 +53,24 @@ struct ReportAgentJobResultArgs { } #[derive(Debug, Serialize)] -struct AgentJobToolResult { +struct SpawnAgentsOnCsvResult { job_id: String, status: String, + output_csv_path: String, + total_items: usize, + completed_items: usize, + failed_items: usize, +} + +#[derive(Debug, Serialize)] +struct AgentJobProgressUpdate { + job_id: String, total_items: usize, pending_items: usize, running_items: usize, completed_items: usize, failed_items: usize, - output_csv_path: String, - runner_active: bool, -} - -#[derive(Debug, Serialize)] -struct SpawnAgentsOnCsvResult { - job_id: String, - started: bool, - output_csv_path: String, - total_items: usize, -} - -#[derive(Debug, Serialize)] -struct WaitAgentJobResult { - status: AgentJobToolResult, - timed_out: bool, -} - -#[derive(Debug, Serialize)] -struct ExportAgentJobCsvResult { - job_id: String, - output_csv_path: String, - row_count: usize, + eta_seconds: Option, } #[derive(Debug, Serialize)] @@ -123,8 +82,73 @@ struct ReportAgentJobResultToolResult { struct JobRunnerOptions { max_concurrency: usize, spawn_config: Config, - child_depth: i32, - auto_export: bool, +} + +struct JobProgressEmitter { + started_at: Instant, + last_emit_at: Instant, + last_processed: usize, + last_failed: usize, +} + +impl JobProgressEmitter { + fn new() -> Self { + let now = Instant::now(); + Self { + started_at: now, + last_emit_at: now, + last_processed: 0, + last_failed: 0, + } + } + + async fn maybe_emit( + &mut self, + session: &Session, + turn: &TurnContext, + job_id: &str, + progress: &codex_state::AgentJobProgress, + force: bool, + ) -> anyhow::Result<()> { + const EMIT_INTERVAL: Duration = Duration::from_secs(1); + let processed = progress.completed_items + progress.failed_items; + let should_emit = force + || processed != self.last_processed + || progress.failed_items != self.last_failed + || self.last_emit_at.elapsed() >= EMIT_INTERVAL; + if !should_emit { + return Ok(()); + } + let elapsed = self.started_at.elapsed().as_secs_f64(); + let eta_seconds = if processed > 0 && elapsed > 0.0 { + let remaining = progress.total_items.saturating_sub(processed) as f64; + let rate = processed as f64 / elapsed; + if rate > 0.0 { + Some((remaining / rate).round() as u64) + } else { + None + } + } else { + None + }; + let update = AgentJobProgressUpdate { + job_id: job_id.to_string(), + total_items: progress.total_items, + pending_items: progress.pending_items, + running_items: progress.running_items, + completed_items: progress.completed_items, + failed_items: progress.failed_items, + eta_seconds, + }; + let payload = serde_json::to_string(&update)?; + session + .notify_background_event(turn, format!("agent_job_progress:{payload}")) + .await; + self.last_emit_at = Instant::now(); + self.last_processed = processed; + self.last_failed = progress.failed_items; + Ok(()) + } } #[async_trait] @@ -157,9 +181,6 @@ impl ToolHandler for AgentJobsHandler { match tool_name.as_str() { "spawn_agents_on_csv" => spawn_agents_on_csv::handle(session, turn, arguments).await, - "get_agent_job_status" => get_agent_job_status::handle(session, arguments).await, - "wait_agent_job" => wait_agent_job::handle(session, arguments).await, - "export_agent_job_csv" => export_agent_job_csv::handle(session, turn, arguments).await, "report_agent_job_result" => report_agent_job_result::handle(session, arguments).await, other => Err(FunctionCallError::RespondToModel(format!( "unsupported agent job tool {other}" @@ -260,14 +281,13 @@ mod spawn_agents_on_csv { let job_name = args .job_name .unwrap_or_else(|| format!("agent-job-{job_suffix}")); - let auto_export = args.auto_export.unwrap_or(true); - let job = db + let _job = db .create_agent_job( &codex_state::AgentJobCreateParams { id: job_id.clone(), name: job_name, instruction: args.instruction, - auto_export, + auto_export: true, output_schema_json: args.output_schema, input_headers: headers, input_csv_path: input_path.display().to_string(), @@ -280,167 +300,62 @@ mod spawn_agents_on_csv { FunctionCallError::RespondToModel(format!("failed to create agent job: {err}")) })?; - let options = - build_runner_options(&session, &turn, args.max_concurrency, Some(auto_export)).await?; - let started = start_job_runner(session, job_id.clone(), options).await?; - - let content = serde_json::to_string(&SpawnAgentsOnCsvResult { - job_id, - started, - output_csv_path: job.output_csv_path, - total_items: items.len(), - }) - .map_err(|err| { - FunctionCallError::Fatal(format!( - "failed to serialize spawn_agents_on_csv result: {err}" - )) - })?; - Ok(ToolOutput::Function { - body: FunctionCallOutputBody::Text(content), - success: Some(true), - }) - } -} - -mod get_agent_job_status { - use super::*; - - pub async fn handle( - session: Arc, - arguments: String, - ) -> Result { - let args: JobIdArgs = parse_arguments(arguments.as_str())?; - let job_id = args.job_id; - let db = required_state_db(&session)?; - let status = render_job_status(db, job_id.as_str()).await?; - let content = serde_json::to_string(&status).map_err(|err| { - FunctionCallError::Fatal(format!( - "failed to serialize get_agent_job_status result: {err}" - )) - })?; - Ok(ToolOutput::Function { - body: FunctionCallOutputBody::Text(content), - success: Some(true), - }) - } -} - -mod wait_agent_job { - use super::*; - - pub async fn handle( - session: Arc, - arguments: String, - ) -> Result { - let args: WaitAgentJobArgs = parse_arguments(arguments.as_str())?; - let db = required_state_db(&session)?; - let timeout_ms = args.timeout_ms.unwrap_or(DEFAULT_WAIT_TIMEOUT_MS); - let timeout_ms = match timeout_ms { - ms if ms <= 0 => { - return Err(FunctionCallError::RespondToModel( - "timeout_ms must be greater than zero".to_string(), - )); - } - ms => ms.clamp(MIN_WAIT_TIMEOUT_MS, MAX_WAIT_TIMEOUT_MS), - }; - - let deadline = Instant::now() + Duration::from_millis(timeout_ms as u64); - loop { - let status = render_job_status(db.clone(), args.job_id.as_str()).await?; - if matches!(status.status.as_str(), "completed" | "failed" | "cancelled") { - let content = serde_json::to_string(&WaitAgentJobResult { - status, - timed_out: false, - }) - .map_err(|err| { - FunctionCallError::Fatal(format!( - "failed to serialize wait_agent_job result: {err}" - )) - })?; - return Ok(ToolOutput::Function { - body: FunctionCallOutputBody::Text(content), - success: Some(true), - }); - } - if Instant::now() >= deadline { - let content = serde_json::to_string(&WaitAgentJobResult { - status, - timed_out: true, - }) - .map_err(|err| { - FunctionCallError::Fatal(format!( - "failed to serialize wait_agent_job timeout result: {err}" - )) - })?; - return Ok(ToolOutput::Function { - body: FunctionCallOutputBody::Text(content), - success: Some(true), - }); - } - tokio::time::sleep(Duration::from_millis(STATUS_POLL_INTERVAL_MS)).await; + 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}" + )) + })?; + let options = build_runner_options(&session, &turn, args.max_concurrency).await?; + 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}" + ))); } - } -} -mod export_agent_job_csv { - use super::*; - - pub async fn handle( - session: Arc, - turn: Arc, - arguments: String, - ) -> Result { - let args: ExportAgentJobCsvArgs = parse_arguments(arguments.as_str())?; - let job_id = args.job_id; - let db = required_state_db(&session)?; let job = db .get_agent_job(job_id.as_str()) .await .map_err(|err| { FunctionCallError::RespondToModel(format!( - "failed to look up agent job {job_id}: {err}" + "failed to load agent job {job_id}: {err}" )) })? .ok_or_else(|| { FunctionCallError::RespondToModel(format!("agent job {job_id} not found")) })?; - let items = db - .list_agent_job_items(job_id.as_str(), None, None) + let progress = db + .get_agent_job_progress(job_id.as_str()) .await .map_err(|err| { FunctionCallError::RespondToModel(format!( - "failed to load items for agent job {job_id}: {err}" + "failed to load agent job progress {job_id}: {err}" )) })?; - let output_path = args.output_csv_path.map_or_else( - || PathBuf::from(job.output_csv_path.clone()), - |path| turn.resolve_path(Some(path)), - ); - if let Some(parent) = output_path.parent() { - let parent_display = parent.display().to_string(); - tokio::fs::create_dir_all(parent).await.map_err(|err| { - FunctionCallError::RespondToModel(format!( - "failed to create export directory {parent_display}: {err}" - )) - })?; - } - let csv_content = render_job_csv(job.input_headers.as_slice(), items.as_slice())?; - let output_display = output_path.display().to_string(); - tokio::fs::write(&output_path, csv_content) - .await - .map_err(|err| { - FunctionCallError::RespondToModel(format!( - "failed to write csv export {output_display}: {err}" - )) - })?; - let content = serde_json::to_string(&ExportAgentJobCsvResult { + let content = serde_json::to_string(&SpawnAgentsOnCsvResult { job_id, - output_csv_path: output_display, - row_count: items.len(), + 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, }) .map_err(|err| { FunctionCallError::Fatal(format!( - "failed to serialize export_agent_job_csv result: {err}" + "failed to serialize spawn_agents_on_csv result: {err}" )) })?; Ok(ToolOutput::Function { @@ -508,7 +423,6 @@ async fn build_runner_options( session: &Arc, turn: &Arc, requested_concurrency: Option, - requested_auto_export: Option, ) -> Result { let session_source = turn.session_source.clone(); let child_depth = next_thread_spawn_depth(&session_source); @@ -524,8 +438,6 @@ async fn build_runner_options( Ok(JobRunnerOptions { max_concurrency, spawn_config, - child_depth, - auto_export: requested_auto_export.unwrap_or(true), }) } @@ -539,81 +451,15 @@ fn normalize_concurrency(requested: Option, max_threads: Option) - } } -async fn start_job_runner( - session: Arc, - job_id: String, - options: JobRunnerOptions, -) -> Result { - cleanup_finished_runners().await; - let mut runners = ACTIVE_JOB_RUNNERS.lock().await; - if let Some(handle) = runners.get(job_id.as_str()) - && !handle.is_finished() - { - return Ok(false); - } - let db = required_state_db(&session)?; - 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 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}" - )) - })?; - if job.status.is_final() && progress.pending_items == 0 && progress.running_items == 0 { - return Ok(false); - } - 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}" - )) - })?; - let job_id_for_task = job_id.clone(); - let handle = tokio::spawn(async move { - if let Err(err) = - run_agent_job_loop(session, db.clone(), job_id_for_task.clone(), options).await - { - let error_message = format!("job runner failed: {err}"); - let _ = db - .mark_agent_job_failed(job_id_for_task.as_str(), error_message.as_str()) - .await; - } - }); - runners.insert(job_id, handle); - Ok(true) -} - -async fn cleanup_finished_runners() { - let mut runners = ACTIVE_JOB_RUNNERS.lock().await; - runners.retain(|_, handle| !handle.is_finished()); -} - -async fn job_runner_active(job_id: &str) -> bool { - cleanup_finished_runners().await; - let runners = ACTIVE_JOB_RUNNERS.lock().await; - runners - .get(job_id) - .is_some_and(|handle| !handle.is_finished()) -} - async fn run_agent_job_loop( session: Arc, + turn: Arc, db: Arc, job_id: String, options: JobRunnerOptions, ) -> anyhow::Result<()> { let mut active_items: HashMap = HashMap::new(); + let mut progress_emitter = JobProgressEmitter::new(); recover_running_items( session.clone(), db.clone(), @@ -626,6 +472,10 @@ async fn run_agent_job_loop( .get_agent_job(job_id.as_str()) .await? .ok_or_else(|| anyhow::anyhow!("agent job {job_id} was not found"))?; + let initial_progress = db.get_agent_job_progress(job_id.as_str()).await?; + progress_emitter + .maybe_emit(&session, &turn, job_id.as_str(), &initial_progress, true) + .await?; loop { let mut progressed = false; @@ -653,10 +503,9 @@ async fn run_agent_job_loop( .spawn_agent( options.spawn_config.clone(), prompt, - Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { - parent_thread_id: session.conversation_id, - depth: options.child_depth, - })), + Some(SessionSource::SubAgent(SubAgentSource::Other(format!( + "agent_job:{job_id}" + )))), ) .await { @@ -732,6 +581,10 @@ async fn run_agent_job_loop( ) .await?; active_items.remove(&thread_id); + let progress = db.get_agent_job_progress(job_id.as_str()).await?; + progress_emitter + .maybe_emit(&session, &turn, job_id.as_str(), &progress, false) + .await?; } } @@ -742,9 +595,7 @@ async fn run_agent_job_loop( db.mark_agent_job_failed(job_id.as_str(), message.as_str()) .await?; } else { - if options.auto_export - && let Err(err) = export_job_csv_snapshot(db.clone(), &job).await - { + 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?; @@ -752,6 +603,10 @@ async fn run_agent_job_loop( } db.mark_agent_job_completed(job_id.as_str()).await?; } + let progress = db.get_agent_job_progress(job_id.as_str()).await?; + progress_emitter + .maybe_emit(&session, &turn, job_id.as_str(), &progress, true) + .await?; Ok(()) } @@ -935,36 +790,6 @@ fn render_instruction_template(instruction: &str, row_json: &Value) -> String { .replace(CLOSE_BRACE_SENTINEL, "}") } -async fn render_job_status( - db: Arc, - job_id: &str, -) -> Result { - let job = db - .get_agent_job(job_id) - .await - .map_err(|err| { - FunctionCallError::RespondToModel(format!("failed to fetch agent job {job_id}: {err}")) - })? - .ok_or_else(|| { - FunctionCallError::RespondToModel(format!("agent job {job_id} not found")) - })?; - let progress = db.get_agent_job_progress(job_id).await.map_err(|err| { - FunctionCallError::RespondToModel(format!("failed to fetch progress for {job_id}: {err}")) - })?; - let runner_active = job_runner_active(job_id).await; - Ok(AgentJobToolResult { - job_id: job.id, - status: job.status.as_str().to_string(), - total_items: progress.total_items, - pending_items: progress.pending_items, - running_items: progress.running_items, - completed_items: progress.completed_items, - failed_items: progress.failed_items, - output_csv_path: job.output_csv_path, - runner_active, - }) -} - fn default_output_csv_path(input_csv_path: &Path, job_id: &str) -> PathBuf { let stem = input_csv_path .file_stem() diff --git a/codex-rs/core/src/tools/spec.rs b/codex-rs/core/src/tools/spec.rs index 6266b1d67a..1eb6654a32 100644 --- a/codex-rs/core/src/tools/spec.rs +++ b/codex-rs/core/src/tools/spec.rs @@ -517,14 +517,6 @@ fn create_spawn_agents_on_csv_tool() -> ToolSpec { description: Some("Optional output CSV path for exported results.".to_string()), }, ); - properties.insert( - "output_schema".to_string(), - JsonSchema::Object { - properties: BTreeMap::new(), - required: None, - additional_properties: None, - }, - ); properties.insert( "max_concurrency".to_string(), JsonSchema::Number { @@ -535,18 +527,18 @@ fn create_spawn_agents_on_csv_tool() -> ToolSpec { }, ); properties.insert( - "auto_export".to_string(), - JsonSchema::Boolean { - description: Some( - "Whether to auto-export CSV output on successful completion (default true)." - .to_string(), - ), + "output_schema".to_string(), + JsonSchema::Object { + properties: BTreeMap::new(), + required: None, + additional_properties: None, }, ); ToolSpec::Function(ResponsesApiTool { name: "spawn_agents_on_csv".to_string(), - description: "Create and start a job that spawns subagents to process each CSV row." - .to_string(), + description: + "Run a batch job that spawns subagents to process each CSV row, blocking until completion and auto-exporting results on success." + .to_string(), strict: false, parameters: JsonSchema::Object { properties, @@ -556,80 +548,6 @@ fn create_spawn_agents_on_csv_tool() -> ToolSpec { }) } -fn create_get_agent_job_status_tool() -> ToolSpec { - let mut properties = BTreeMap::new(); - properties.insert( - "job_id".to_string(), - JsonSchema::String { - description: Some("Identifier of the job to inspect.".to_string()), - }, - ); - ToolSpec::Function(ResponsesApiTool { - name: "get_agent_job_status".to_string(), - description: "Fetch job status and progress counters.".to_string(), - strict: false, - parameters: JsonSchema::Object { - properties, - required: Some(vec!["job_id".to_string()]), - additional_properties: Some(false.into()), - }, - }) -} - -fn create_wait_agent_job_tool() -> ToolSpec { - let mut properties = BTreeMap::new(); - properties.insert( - "job_id".to_string(), - JsonSchema::String { - description: Some("Identifier of the job to wait on.".to_string()), - }, - ); - properties.insert( - "timeout_ms".to_string(), - JsonSchema::Number { - description: Some( - "Maximum time in milliseconds to wait for job completion.".to_string(), - ), - }, - ); - ToolSpec::Function(ResponsesApiTool { - name: "wait_agent_job".to_string(), - description: "Wait for an agent job to complete or time out.".to_string(), - strict: false, - parameters: JsonSchema::Object { - properties, - required: Some(vec!["job_id".to_string()]), - additional_properties: Some(false.into()), - }, - }) -} - -fn create_export_agent_job_csv_tool() -> ToolSpec { - let mut properties = BTreeMap::new(); - properties.insert( - "job_id".to_string(), - JsonSchema::String { - description: Some("Identifier of the job to export.".to_string()), - }, - ); - properties.insert( - "output_csv_path".to_string(), - JsonSchema::String { - description: Some("Optional output CSV path override.".to_string()), - }, - ); - ToolSpec::Function(ResponsesApiTool { - name: "export_agent_job_csv".to_string(), - description: "Export job results to a CSV file.".to_string(), - strict: false, - parameters: JsonSchema::Object { - properties, - required: Some(vec!["job_id".to_string()]), - additional_properties: Some(false.into()), - }, - }) -} - fn create_report_agent_job_result_tool() -> ToolSpec { let mut properties = BTreeMap::new(); properties.insert( @@ -654,7 +572,9 @@ fn create_report_agent_job_result_tool() -> ToolSpec { ); ToolSpec::Function(ResponsesApiTool { name: "report_agent_job_result".to_string(), - description: "Report a worker result for an agent job item.".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, parameters: JsonSchema::Object { properties, @@ -1576,14 +1496,8 @@ pub(crate) fn build_specs( if config.collab_tools { let agent_jobs_handler = Arc::new(AgentJobsHandler); builder.push_spec(create_spawn_agents_on_csv_tool()); - builder.push_spec(create_get_agent_job_status_tool()); - builder.push_spec(create_wait_agent_job_tool()); - builder.push_spec(create_export_agent_job_csv_tool()); builder.push_spec(create_report_agent_job_result_tool()); builder.register_handler("spawn_agents_on_csv", agent_jobs_handler.clone()); - builder.register_handler("get_agent_job_status", agent_jobs_handler.clone()); - builder.register_handler("wait_agent_job", agent_jobs_handler.clone()); - builder.register_handler("export_agent_job_csv", agent_jobs_handler.clone()); builder.register_handler("report_agent_job_result", agent_jobs_handler); } @@ -1845,9 +1759,6 @@ mod tests { "wait", "close_agent", "spawn_agents_on_csv", - "get_agent_job_status", - "wait_agent_job", - "export_agent_job_csv", "report_agent_job_result", ], ); diff --git a/codex-rs/docs/agent_jobs.md b/codex-rs/docs/agent_jobs.md index 92f1fab789..2b72adfb27 100644 --- a/codex-rs/docs/agent_jobs.md +++ b/codex-rs/docs/agent_jobs.md @@ -5,8 +5,7 @@ Agent jobs are designed to be: 1. Durable via SQLite for progress tracking and export. 2. Bounded by configured concurrency and thread limits. -3. Observable via explicit status/progress tools. -4. Exportable to CSV at stage boundaries. +3. Exportable to CSV on successful completion. ## Tools @@ -15,6 +14,7 @@ All tools are function-style and gated by the `collab` feature. ### `spawn_agents_on_csv` Create a new job from a CSV input and immediately start it. +This tool blocks until the job completes and auto-exports on success. Required args: - `csv_path`: path to the CSV file (first row is headers). @@ -28,35 +28,19 @@ Optional args: - `output_csv_path`: destination for CSV export (defaults to `.agent-job-.csv`). - `output_schema`: JSON schema for result payloads (best-effort guidance). - `max_concurrency`: cap on parallel workers (defaults to 64, then capped by config). -- `auto_export`: auto-export CSV on successful completion (default true). - -### `get_agent_job_status` - -Return job status and progress counters. Most flows should prefer `wait_agent_job` -to deterministically block until completion. - -### `wait_agent_job` - -Wait for a job to complete, or return after a timeout. - -### `export_agent_job_csv` - -Export the current job results to CSV using the stored headers and results. You can -optionally override the destination path. ### `report_agent_job_result` -Workers must call this exactly once to report a JSON object for their assigned item. +Worker-only tool used internally for reporting JSON results. Main agents should not call this. ## Execution Model 1. Jobs are stored in SQLite with per-item state (pending/running/completed/failed). 2. The job runner spawns subagents up to `max_concurrency`. 3. The job instruction is rendered per row by substituting `{column_name}` placeholders. -4. Each worker processes one item and calls `report_agent_job_result`. +4. Each worker processes one item and reports results through `report_agent_job_result`. 5. The runner marks items completed after the worker finishes. -6. If `auto_export` is true, the runner writes a CSV snapshot on successful completion. -7. CSV export can also be generated manually by a single writer from the SQLite store. +6. The runner writes a CSV snapshot on successful completion. ## CSV Output @@ -97,7 +81,6 @@ csv_path: /tmp/security_rank_input_demo.csv instruction: read the file (relative to repo root), skim first 200 lines, score security relevance 1-10, output JSON with keys path, score, rationale, signals array. output_csv_path: /tmp/security_rank_output_demo.csv -Do NOT call export_agent_job_csv manually. -Wait for completion and then print the output path and `head -n 6` of the CSV. +After completion, print the output path and `head -n 6` of the CSV. PROMPT ``` diff --git a/codex-rs/exec/src/event_processor_with_human_output.rs b/codex-rs/exec/src/event_processor_with_human_output.rs index cbe45b92f1..49954b2652 100644 --- a/codex-rs/exec/src/event_processor_with_human_output.rs +++ b/codex-rs/exec/src/event_processor_with_human_output.rs @@ -38,9 +38,12 @@ use codex_protocol::items::TurnItem; use codex_protocol::num_format::format_with_separators; use owo_colors::OwoColorize; use owo_colors::Style; +use serde::Deserialize; use shlex::try_join; use std::collections::HashMap; +use std::io::Write; use std::path::PathBuf; +use std::time::Duration; use std::time::Instant; use crate::event_processor::CodexStatus; @@ -76,6 +79,8 @@ pub(crate) struct EventProcessorWithHumanOutput { last_total_token_usage: Option, final_message: Option, last_proposed_plan: Option, + progress_active: bool, + progress_last_len: usize, } impl EventProcessorWithHumanOutput { @@ -103,6 +108,8 @@ impl EventProcessorWithHumanOutput { last_total_token_usage: None, final_message: None, last_proposed_plan: None, + progress_active: false, + progress_last_len: 0, } } else { Self { @@ -121,11 +128,24 @@ impl EventProcessorWithHumanOutput { last_total_token_usage: None, final_message: None, last_proposed_plan: None, + progress_active: false, + progress_last_len: 0, } } } } +#[derive(Debug, Deserialize)] +struct AgentJobProgressMessage { + job_id: String, + total_items: usize, + pending_items: usize, + running_items: usize, + completed_items: usize, + failed_items: usize, + eta_seconds: Option, +} + struct PatchApplyBegin { start_time: Instant, auto_approved: bool, @@ -176,6 +196,15 @@ impl EventProcessor for EventProcessorWithHumanOutput { fn process_event(&mut self, event: Event) -> CodexStatus { let Event { id: _, msg } = event; + if let EventMsg::BackgroundEvent(BackgroundEventEvent { message }) = &msg + && let Some(update) = Self::parse_agent_job_progress(message) + { + self.render_agent_job_progress(update); + return CodexStatus::Running; + } + if !Self::is_silent_event(&msg) { + self.finish_progress_line(); + } match msg { EventMsg::Error(ErrorEvent { message, .. }) => { let prefix = "ERROR:".style(self.red); @@ -802,6 +831,7 @@ impl EventProcessor for EventProcessorWithHumanOutput { } fn print_final_output(&mut self) { + self.finish_progress_line(); if let Some(usage_info) = &self.last_total_token_usage { eprintln!( "{}\n{}", @@ -825,6 +855,96 @@ impl EventProcessor for EventProcessorWithHumanOutput { } } +impl EventProcessorWithHumanOutput { + fn parse_agent_job_progress(message: &str) -> Option { + let payload = message.strip_prefix("agent_job_progress:")?; + serde_json::from_str::(payload).ok() + } + + fn is_silent_event(msg: &EventMsg) -> bool { + matches!( + msg, + EventMsg::ThreadNameUpdated(_) + | EventMsg::ExecApprovalRequest(_) + | EventMsg::ApplyPatchApprovalRequest(_) + | EventMsg::TerminalInteraction(_) + | EventMsg::ExecCommandOutputDelta(_) + | EventMsg::GetHistoryEntryResponse(_) + | EventMsg::McpListToolsResponse(_) + | EventMsg::ListCustomPromptsResponse(_) + | EventMsg::ListSkillsResponse(_) + | EventMsg::ListRemoteSkillsResponse(_) + | EventMsg::RemoteSkillDownloaded(_) + | EventMsg::RawResponseItem(_) + | EventMsg::UserMessage(_) + | EventMsg::EnteredReviewMode(_) + | EventMsg::ExitedReviewMode(_) + | EventMsg::AgentMessageDelta(_) + | EventMsg::AgentReasoningDelta(_) + | EventMsg::AgentReasoningRawContentDelta(_) + | EventMsg::ItemStarted(_) + | EventMsg::ItemCompleted(_) + | EventMsg::AgentMessageContentDelta(_) + | EventMsg::PlanDelta(_) + | EventMsg::ReasoningContentDelta(_) + | EventMsg::ReasoningRawContentDelta(_) + | EventMsg::SkillsUpdateAvailable + | EventMsg::UndoCompleted(_) + | EventMsg::UndoStarted(_) + | EventMsg::ThreadRolledBack(_) + | EventMsg::RequestUserInput(_) + | EventMsg::DynamicToolCallRequest(_) + ) + } + + fn finish_progress_line(&mut self) { + if self.progress_active { + eprintln!(); + self.progress_active = false; + self.progress_last_len = 0; + } + } + + fn render_agent_job_progress(&mut self, update: AgentJobProgressMessage) { + let total = update.total_items.max(1); + let processed = update.completed_items + update.failed_items; + let percent = (processed as f64 / total as f64 * 100.0).round() as i64; + let bar_width = 28; + let filled = ((processed as f64 / total as f64) * bar_width as f64) + .round() + .clamp(0.0, bar_width as f64) as usize; + let mut bar = "#".repeat(filled); + bar.push_str(&"-".repeat(bar_width - filled)); + let job_label = update.job_id.chars().take(8).collect::(); + let eta = update + .eta_seconds + .map(|secs| format_duration(Duration::from_secs(secs))) + .unwrap_or_else(|| "--".to_string()); + let line = format!( + "agent job {job_label} [{bar}] {processed}/{total} ({percent}%) failed {} running {} pending {} eta {eta}", + update.failed_items, update.running_items, update.pending_items + ); + let done = processed >= update.total_items; + if done { + if self.progress_active { + eprint!("\r"); + } + eprintln!("{line}"); + self.progress_active = false; + self.progress_last_len = 0; + return; + } + let mut output = format!("\r{line}"); + if self.progress_last_len > line.len() { + output.push_str(&" ".repeat(self.progress_last_len - line.len())); + } + eprint!("{output}"); + let _ = std::io::stderr().flush(); + self.progress_active = true; + self.progress_last_len = line.len(); + } +} + fn escape_command(command: &[String]) -> String { try_join(command.iter().map(String::as_str)).unwrap_or_else(|_| command.join(" ")) } diff --git a/codex-rs/exec/src/lib.rs b/codex-rs/exec/src/lib.rs index a6ded3927a..5f5f40f578 100644 --- a/codex-rs/exec/src/lib.rs +++ b/codex-rs/exec/src/lib.rs @@ -41,6 +41,7 @@ use codex_core::protocol::ReviewTarget; use codex_core::protocol::SessionSource; use codex_protocol::approvals::ElicitationAction; use codex_protocol::config_types::SandboxMode; +use codex_protocol::protocol::SubAgentSource; use codex_protocol::user_input::UserInput; use codex_utils_absolute_path::AbsolutePathBuf; use event_processor_with_human_output::EventProcessorWithHumanOutput; @@ -84,6 +85,7 @@ struct ThreadEventEnvelope { thread_id: codex_protocol::ThreadId, thread: Arc, event: Event, + suppress_output: bool, } pub async fn run_main(cli: Cli, codex_linux_sandbox_exe: Option) -> anyhow::Result<()> { @@ -438,7 +440,7 @@ pub async fn run_main(cli: Cli, codex_linux_sandbox_exe: Option) -> any let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::(); let attached_threads = Arc::new(Mutex::new(HashSet::from([primary_thread_id]))); - spawn_thread_listener(primary_thread_id, thread.clone(), tx.clone()); + spawn_thread_listener(primary_thread_id, thread.clone(), tx.clone(), false); { let thread = thread.clone(); @@ -466,7 +468,14 @@ pub async fn run_main(cli: Cli, codex_linux_sandbox_exe: Option) -> any match thread_manager.get_thread(thread_id).await { Ok(thread) => { attached_threads.lock().await.insert(thread_id); - spawn_thread_listener(thread_id, thread, tx.clone()); + let suppress_output = + is_agent_job_subagent(&thread.config_snapshot().await); + spawn_thread_listener( + thread_id, + thread, + tx.clone(), + suppress_output, + ); } Err(err) => { warn!("failed to attach listener for thread {thread_id}: {err}") @@ -520,7 +529,11 @@ pub async fn run_main(cli: Cli, codex_linux_sandbox_exe: Option) -> any thread_id, thread, event, + suppress_output, } = envelope; + if suppress_output && should_suppress_agent_job_event(&event.msg) { + continue; + } if let EventMsg::ElicitationRequest(ev) = &event.msg { // Automatically cancel elicitation requests in exec mode. thread @@ -562,6 +575,7 @@ fn spawn_thread_listener( thread_id: codex_protocol::ThreadId, thread: Arc, tx: tokio::sync::mpsc::UnboundedSender, + suppress_output: bool, ) { tokio::spawn(async move { loop { @@ -574,6 +588,7 @@ fn spawn_thread_listener( thread_id, thread: Arc::clone(&thread), event, + suppress_output, }) { error!("Error sending event: {err:?}"); break; @@ -594,6 +609,29 @@ fn spawn_thread_listener( }); } +fn is_agent_job_subagent(config: &codex_core::ThreadConfigSnapshot) -> bool { + match &config.session_source { + SessionSource::SubAgent(SubAgentSource::Other(source)) => source.starts_with("agent_job:"), + _ => false, + } +} + +fn should_suppress_agent_job_event(msg: &EventMsg) -> bool { + !matches!( + msg, + EventMsg::ExecApprovalRequest(_) + | EventMsg::ApplyPatchApprovalRequest(_) + | EventMsg::RequestUserInput(_) + | EventMsg::DynamicToolCallRequest(_) + | EventMsg::ElicitationRequest(_) + | EventMsg::Error(_) + | EventMsg::Warning(_) + | EventMsg::DeprecationNotice(_) + | EventMsg::StreamError(_) + | EventMsg::ShutdownComplete + ) +} + async fn resolve_resume_path( config: &Config, args: &crate::cli::ResumeArgs,