openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-09 20:50:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

Remove agent job core tools

Browse Source
This commit is contained in:
jif-oai
2026-06-05 16:13:49 +02:00
parent 66232220e2
commit 96fd94e558
23 changed files with 9 additions and 2023 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/config/src/config_toml.rs
-3
View File
@@ -667,9 +667,6 @@ pub struct AgentsToml {
/// Root sessions start at depth 0.
#[schemars(range(min = 1))]
pub max_depth: Option<i32>,
/// Default maximum runtime in seconds for agent job workers.
#[schemars(range(min = 1))]
pub job_max_runtime_seconds: Option<u64>,
/// Whether to record a model-visible message when an agent turn is interrupted.
/// Defaults to true.
pub interrupt_message: Option<bool>,
codex-rs/core/config.schema.json
+1 -7
View File
@@ -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"
}
}
codex-rs/core/src/config/config_tests.rs
-5
View File
@@ -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(),
codex-rs/core/src/config/mod.rs
-24
View File
@@ -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<u64> = 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<usize>,
/// Maximum runtime in seconds for agent job workers before they are failed.
pub agent_job_max_runtime_seconds: Option<u64>,
/// 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,
codex-rs/core/src/guardian/review_session.rs
-1
View File
@@ -952,7 +952,6 @@ pub(crate) fn build_guardian_review_session_config(
)?);
}
for feature in [
Feature::SpawnCsv,
Feature::Collab,
Feature::MultiAgentV2,
Feature::CodexHooks,
codex-rs/core/src/session/config_lock.rs
-1
View File
@@ -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
codex-rs/core/src/tasks/review.rs
-1
View File
@@ -108,7 +108,6 @@ async fn start_review_conversation(
{
panic!("by construction Constrained<WebSearchMode> 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);
codex-rs/core/src/tools/handlers/agent_jobs.rs
-762
View File
@@ -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<String>,
output_csv_path: Option<String>,
output_schema: Option<Value>,
max_concurrency: Option<usize>,
max_workers: Option<usize>,
max_runtime_seconds: Option<u64>,
}
#[derive(Debug, Deserialize)]
struct ReportAgentJobResultArgs {
job_id: String,
item_id: String,
result: Value,
stop: Option<bool>,
}
#[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<String>,
failed_item_errors: Option<Vec<AgentJobFailureSummary>>,
}
#[derive(Debug, Serialize)]
struct AgentJobFailureSummary {
item_id: String,
source_id: Option<String>,
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<Receiver<AgentStatus>>,
}
fn required_state_db(
session: &Arc<Session>,
) -> Result<Arc<codex_state::StateRuntime>, 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<Session>,
turn: &Arc<TurnContext>,
requested_concurrency: Option<usize>,
) -> Result<JobRunnerOptions, FunctionCallError> {
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<usize>, max_threads: Option<usize>) -> 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<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>,
db: Arc<codex_state::StateRuntime>,
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<ThreadId, ActiveJobItem> = 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<codex_state::StateRuntime>,
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<Session>,
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),
/*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<Session>,
active_items: &HashMap<ThreadId, ActiveJobItem>,
) -> 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<ThreadId, ActiveJobItem>) {
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<Session>,
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() >= 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<Session>,
db: Arc<codex_state::StateRuntime>,
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<String> {
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<String>, Vec<Vec<String>>), 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<String> = 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<String> = 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<String, FunctionCallError> {
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::<Vec<_>>()
.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;
codex-rs/core/src/tools/handlers/agent_jobs/report_agent_job_result.rs
-93
View File
@@ -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<ToolInvocation> 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<Box<dyn crate::tools::context::ToolOutput>, 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<Session>,
arguments: String,
) -> Result<FunctionToolOutput, FunctionCallError> {
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)))
}
codex-rs/core/src/tools/handlers/agent_jobs/spawn_agents_on_csv.rs
-311
View File
@@ -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<ToolInvocation> 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<Box<dyn crate::tools::context::ToolOutput>, 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<Session>,
turn: Arc<TurnContext>,
arguments: String,
) -> Result<FunctionToolOutput, FunctionCallError> {
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::<serde_json::Map<_, _>>();
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)
}
codex-rs/core/src/tools/handlers/agent_jobs_spec.rs
-119
View File
@@ -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;
codex-rs/core/src/tools/handlers/agent_jobs_spec_tests.rs
-126
View File
@@ -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,
})
);
}
codex-rs/core/src/tools/handlers/agent_jobs_tests.rs
-62
View File
@@ -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())
);
}
codex-rs/core/src/tools/handlers/mod.rs
-2
View File
@@ -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;
codex-rs/core/src/tools/spec_plan.rs
-22
View File
@@ -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) {
codex-rs/core/src/tools/spec_plan_tests.rs
+1 -16
View File
@@ -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]
codex-rs/core/src/turn_metadata_tests.rs
+2 -2
View File
@@ -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",
),
];
codex-rs/core/tests/suite/agent_jobs.rs
-448
View File
@@ -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<AtomicUsize>,
}
impl StopAfterFirstResponder {
fn new(spawn_args_json: String, worker_calls: Arc<AtomicUsize>) -> 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<u8> {
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<String> {
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<String> {
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(())
}
codex-rs/core/tests/suite/mod.rs
-1
View File
@@ -30,7 +30,6 @@ pub static CODEX_ALIASES_TEMP_DIR: Option<TestBinaryDispatchGuard> = {
#[cfg(not(target_os = "windows"))]
mod abort_tasks;
mod additional_context;
mod agent_jobs;
mod agent_websocket;
mod agents_md;
mod apply_patch_cli;
codex-rs/features/src/lib.rs
+2 -5
View File
@@ -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 {
codex-rs/features/src/tests.rs
+3 -10
View File
@@ -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]
codex-rs/memories/write/src/phase2.rs
-1
View File
@@ -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);
codex-rs/thread-manager-sample/src/main.rs
-1
View File
@@ -224,7 +224,6 @@ fn new_config(model: Option<String>, 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(),