Make agent job runtime timeout configurable

daveaitel-openai · daveaitel-openai · commit 8eef34dab24d · 2026-02-10T13:57:43.000-05:00
diff --git a/codex-rs/core/src/tools/handlers/agent_jobs.rs b/codex-rs/core/src/tools/handlers/agent_jobs.rs
@@ -37,7 +37,7 @@ const DEFAULT_AGENT_JOB_CONCURRENCY: usize = 16;
 const MAX_AGENT_JOB_CONCURRENCY: usize = 64;
 const STATUS_POLL_INTERVAL: Duration = Duration::from_millis(250);
 const PROGRESS_EMIT_INTERVAL: Duration = Duration::from_secs(1);
-const RUNNING_ITEM_TIMEOUT: Duration = Duration::from_secs(60 * 30);
+const DEFAULT_AGENT_JOB_ITEM_TIMEOUT: Duration = Duration::from_secs(60 * 30);
 
 #[derive(Debug, Deserialize)]
 struct SpawnAgentsOnCsvArgs {
@@ -49,6 +49,7 @@ struct SpawnAgentsOnCsvArgs {
     output_schema: Option<Value>,
     max_concurrency: Option<usize>,
     max_workers: Option<usize>,
+    max_runtime_seconds: Option<u64>,
 }
 
 #[derive(Debug, Deserialize)]
@@ -310,13 +311,15 @@ mod spawn_agents_on_csv {
         let job_name = args
             .job_name
             .unwrap_or_else(|| format!("agent-job-{job_suffix}"));
+        let max_runtime_seconds = normalize_max_runtime_seconds(args.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(),
@@ -527,6 +530,18 @@ fn normalize_concurrency(requested: Option<usize>, max_threads: Option<usize>) -
     }
 }
 
+fn normalize_max_runtime_seconds(requested: Option<u64>) -> Result<Option<u64>, 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<Session>,
     turn: Arc<TurnContext>,
@@ -536,18 +551,19 @@ async fn run_agent_job_loop(
 ) -> anyhow::Result<()> {
     let mut active_items: HashMap<ThreadId, ActiveJobItem> = HashMap::new();
     let mut progress_emitter = JobProgressEmitter::new();
+    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);
     recover_running_items(
         session.clone(),
         db.clone(),
         job_id.as_str(),
         &mut active_items,
+        runtime_timeout,
     )
     .await?;
-
-    let job = db
-        .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)
@@ -632,6 +648,7 @@ async fn run_agent_job_loop(
             db.clone(),
             job_id.as_str(),
             &mut active_items,
+            runtime_timeout,
         )
         .await?
         {
@@ -708,14 +725,14 @@ async fn recover_running_items(
     db: Arc<codex_state::StateRuntime>,
     job_id: &str,
     active_items: &mut HashMap<ThreadId, ActiveJobItem>,
+    runtime_timeout: Duration,
 ) -> anyhow::Result<()> {
     let running_items = db
         .list_agent_job_items(job_id, Some(codex_state::AgentJobItemStatus::Running), None)
         .await?;
     for item in running_items {
-        if is_item_stale(&item) {
-            let error_message =
-                format!("worker exceeded max runtime of {:?}", RUNNING_ITEM_TIMEOUT);
+        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?;
             continue;
@@ -782,18 +799,19 @@ async fn reap_stale_active_items(
     db: Arc<codex_state::StateRuntime>,
     job_id: &str,
     active_items: &mut HashMap<ThreadId, ActiveJobItem>,
+    runtime_timeout: Duration,
 ) -> anyhow::Result<bool> {
     let mut stale = Vec::new();
     for (thread_id, item) in active_items.iter() {
-        if item.started_at.elapsed() >= RUNNING_ITEM_TIMEOUT {
+        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 {:?}", RUNNING_ITEM_TIMEOUT);
+        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
@@ -922,6 +940,12 @@ fn ensure_unique_headers(headers: &[String]) -> Result<(), FunctionCallError> {
     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);
@@ -932,10 +956,10 @@ fn started_at_from_item(item: &codex_state::AgentJobItem) -> Instant {
     }
 }
 
-fn is_item_stale(item: &codex_state::AgentJobItem) -> bool {
+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 >= RUNNING_ITEM_TIMEOUT
+        age >= runtime_timeout
     } else {
         false
     }
diff --git a/codex-rs/core/src/tools/spec.rs b/codex-rs/core/src/tools/spec.rs
@@ -549,6 +549,15 @@ fn create_spawn_agents_on_csv_tool() -> ToolSpec {
             ),
         },
     );
+    properties.insert(
+        "max_runtime_seconds".to_string(),
+        JsonSchema::Number {
+            description: Some(
+                "Maximum runtime per worker before it is failed. Defaults to 1800 seconds."
+                    .to_string(),
+            ),
+        },
+    );
     properties.insert(
         "output_schema".to_string(),
         JsonSchema::Object {
diff --git a/codex-rs/state/migrations/0009_agent_jobs.sql b/codex-rs/state/migrations/0009_agent_jobs.sql
@@ -8,6 +8,7 @@ CREATE TABLE agent_jobs (
     input_csv_path TEXT NOT NULL,
     output_csv_path TEXT NOT NULL,
     auto_export INTEGER NOT NULL DEFAULT 1,
+    max_runtime_seconds INTEGER,
     created_at INTEGER NOT NULL,
     updated_at INTEGER NOT NULL,
     started_at INTEGER,
diff --git a/codex-rs/state/src/model/agent_job.rs b/codex-rs/state/src/model/agent_job.rs
@@ -80,6 +80,7 @@ pub struct AgentJob {
     pub status: AgentJobStatus,
     pub instruction: String,
     pub auto_export: bool,
+    pub max_runtime_seconds: Option<u64>,
     pub output_schema_json: Option<Value>,
     pub input_headers: Vec<String>,
     pub input_csv_path: String,
@@ -124,6 +125,7 @@ pub struct AgentJobCreateParams {
     pub name: String,
     pub instruction: String,
     pub auto_export: bool,
+    pub max_runtime_seconds: Option<u64>,
     pub output_schema_json: Option<Value>,
     pub input_headers: Vec<String>,
     pub input_csv_path: String,
@@ -145,6 +147,7 @@ pub(crate) struct AgentJobRow {
     pub(crate) status: String,
     pub(crate) instruction: String,
     pub(crate) auto_export: i64,
+    pub(crate) max_runtime_seconds: Option<i64>,
     pub(crate) output_schema_json: Option<String>,
     pub(crate) input_headers_json: String,
     pub(crate) input_csv_path: String,
@@ -164,6 +167,7 @@ impl AgentJobRow {
             status: row.try_get("status")?,
             instruction: row.try_get("instruction")?,
             auto_export: row.try_get("auto_export")?,
+            max_runtime_seconds: row.try_get("max_runtime_seconds")?,
             output_schema_json: row.try_get("output_schema_json")?,
             input_headers_json: row.try_get("input_headers_json")?,
             input_csv_path: row.try_get("input_csv_path")?,
@@ -187,12 +191,18 @@ impl TryFrom<AgentJobRow> for AgentJob {
             .map(serde_json::from_str)
             .transpose()?;
         let input_headers = serde_json::from_str(value.input_headers_json.as_str())?;
+        let max_runtime_seconds = value
+            .max_runtime_seconds
+            .map(u64::try_from)
+            .transpose()
+            .map_err(|_| anyhow::anyhow!("invalid max_runtime_seconds value"))?;
         Ok(Self {
             id: value.id,
             name: value.name,
             status: AgentJobStatus::parse(value.status.as_str())?,
             instruction: value.instruction,
             auto_export: value.auto_export != 0,
+            max_runtime_seconds,
             output_schema_json,
             input_headers,
             input_csv_path: value.input_csv_path,
diff --git a/codex-rs/state/src/runtime.rs b/codex-rs/state/src/runtime.rs
@@ -747,6 +747,7 @@ INSERT INTO agent_jobs (
     status,
     instruction,
     auto_export,
+    max_runtime_seconds,
     output_schema_json,
     input_headers_json,
     input_csv_path,
@@ -764,6 +765,7 @@ INSERT INTO agent_jobs (
         .bind(AgentJobStatus::Pending.as_str())
         .bind(params.instruction.as_str())
         .bind(i64::from(params.auto_export))
+        .bind(params.max_runtime_seconds.map(i64::try_from).transpose()?)
         .bind(output_schema_json)
         .bind(input_headers_json)
         .bind(params.input_csv_path.as_str())
@@ -824,6 +826,7 @@ SELECT
     status,
     instruction,
     auto_export,
+    max_runtime_seconds,
     output_schema_json,
     input_headers_json,
     input_csv_path,
