Simplify agent job output and progress

2026-10-30 19:18:49 +00:00 · 2026-02-06 12:00:13 -05:00
parent 5f850f40c5
commit 903b8f4b0f
5 changed files with 310 additions and 433 deletions
@@ -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<Mutex<HashMap<String, JoinHandle<()>>>> =
-    Lazy::new(|| Mutex::new(HashMap::new()));
-
 #[derive(Debug, Deserialize)]
 struct SpawnAgentsOnCsvArgs {
    csv_path: String,
@@ -52,25 +43,6 @@ struct SpawnAgentsOnCsvArgs {
    output_csv_path: Option<String>,
    output_schema: Option<Value>,
    max_concurrency: Option<usize>,
-    auto_export: Option<bool>,
-}
-
-#[derive(Debug, Deserialize)]
-struct JobIdArgs {
-    job_id: String,
-}
-
-#[derive(Debug, Deserialize)]
-struct WaitAgentJobArgs {
-    job_id: String,
-    timeout_ms: Option<i64>,
-}
-
-#[derive(Debug, Deserialize)]
-struct ExportAgentJobCsvArgs {
-    job_id: String,
-    #[serde(alias = "path")]
-    output_csv_path: Option<String>,
 }

 #[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<u64>,
 }

 #[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<Session>,
-        arguments: String,
-    ) -> Result<ToolOutput, FunctionCallError> {
-        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<Session>,
-        arguments: String,
-    ) -> Result<ToolOutput, FunctionCallError> {
-        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<Session>,
-        turn: Arc<TurnContext>,
-        arguments: String,
-    ) -> Result<ToolOutput, FunctionCallError> {
-        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<Session>,
    turn: &Arc<TurnContext>,
    requested_concurrency: Option<usize>,
-    requested_auto_export: Option<bool>,
 ) -> Result<JobRunnerOptions, FunctionCallError> {
    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<usize>, max_threads: Option<usize>) -
    }
 }

-async fn start_job_runner(
-    session: Arc<Session>,
-    job_id: String,
-    options: JobRunnerOptions,
-) -> Result<bool, FunctionCallError> {
-    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<Session>,
+    turn: Arc<TurnContext>,
    db: Arc<codex_state::StateRuntime>,
    job_id: String,
    options: JobRunnerOptions,
 ) -> anyhow::Result<()> {
    let mut active_items: HashMap<ThreadId, String> = 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<codex_state::StateRuntime>,
-    job_id: &str,
-) -> Result<AgentJobToolResult, FunctionCallError> {
-    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()
@@ -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",
            ],
        );
@@ -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 `<input>.agent-job-<id>.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
 ```
@@ -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<codex_core::protocol::TokenUsageInfo>,
    final_message: Option<String>,
    last_proposed_plan: Option<String>,
+    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<u64>,
+}
+
 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<AgentJobProgressMessage> {
+        let payload = message.strip_prefix("agent_job_progress:")?;
+        serde_json::from_str::<AgentJobProgressMessage>(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::<String>();
+        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(" "))
 }
@@ -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<codex_core::CodexThread>,
    event: Event,
+    suppress_output: bool,
 }

 pub async fn run_main(cli: Cli, codex_linux_sandbox_exe: Option<PathBuf>) -> anyhow::Result<()> {
@@ -438,7 +440,7 @@ pub async fn run_main(cli: Cli, codex_linux_sandbox_exe: Option<PathBuf>) -> any

    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<ThreadEventEnvelope>();
    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<PathBuf>) -> 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<PathBuf>) -> 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<codex_core::CodexThread>,
    tx: tokio::sync::mpsc::UnboundedSender<ThreadEventEnvelope>,
+    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,