Remove CSV-backed agent jobs (#34413)

## What changed

- Remove the `spawn_agents_on_csv` and `report_agent_job_result` tools and their agent-job runtime and state models.
- Drop the legacy `agent_jobs` and `agent_job_items` tables during state database migration.
- Keep `features.enable_fanout` and `agents.job_max_runtime_seconds` accepted as no-op compatibility settings while omitting them from the generated configuration schema.

## Testing

- Verify upgrades remove both legacy agent-job tables.
- Verify the removed feature and configuration keys still parse without taking effect.

GitOrigin-RevId: 8cc3337da78c67162229f02f40a747f503542646
This commit is contained in:
jif
2026-07-20 21:00:17 +00:00
committed by copyberry
parent 81e89fa5af
commit 687f05cb94
39 changed files with 138 additions and 3170 deletions
-1
View File
@@ -2685,7 +2685,6 @@ dependencies = [
"codex-web-search-extension",
"codex-windows-sandbox",
"core_test_support",
"csv",
"ctor 0.6.3",
"dirs",
"dunce",
-1
View File
@@ -299,7 +299,6 @@ constant_time_eq = "0.3.1"
crossbeam-channel = "0.5.15"
crypto_box = { version = "0.9.1", features = ["seal"] }
crossterm = "0.28.1"
csv = "1.3.1"
ctor = "0.6.3"
deno_core_icudata = "0.77.0"
derive_more = "2"
@@ -1017,7 +1017,7 @@ async fn thread_list_relation_filters_read_spawn_graph_from_state_db() -> Result
(
older_child_id,
"2025-02-01T10:00:00Z",
CoreSessionSource::SubAgent(SubAgentSource::Other("agent_job:job-1".to_string())),
CoreSessionSource::SubAgent(SubAgentSource::Other("custom:worker-1".to_string())),
"other_provider",
),
(
@@ -1029,7 +1029,7 @@ async fn thread_list_relation_filters_read_spawn_graph_from_state_db() -> Result
(
grandchild_id,
"2025-02-01T12:00:00Z",
CoreSessionSource::SubAgent(SubAgentSource::Other("agent_job:job-2".to_string())),
CoreSessionSource::SubAgent(SubAgentSource::Other("custom:worker-2".to_string())),
"mock_provider",
),
] {
+10
View File
@@ -4039,6 +4039,16 @@ mod tests {
);
}
#[test]
fn feature_toggles_accept_removed_enable_fanout_flag() {
let toggles = FeatureToggles {
enable: vec!["enable_fanout".to_string()],
disable: Vec::new(),
};
let overrides = toggles.to_overrides().expect("valid features");
assert_eq!(overrides, vec!["features.enable_fanout=true".to_string(),]);
}
#[test]
fn feature_toggles_unknown_feature_errors() {
let toggles = FeatureToggles {
+2 -2
View File
@@ -692,8 +692,8 @@ pub struct AgentsToml {
pub default_subagent_model: Option<String>,
/// Default reasoning effort for spawned subagents when the spawn call does not select one.
pub default_subagent_reasoning_effort: Option<ReasoningEffort>,
/// Default maximum runtime in seconds for agent job workers.
#[schemars(range(min = 1))]
/// Removed agent-job setting retained as a no-op for compatibility.
#[schemars(skip)]
pub job_max_runtime_seconds: Option<u64>,
/// Whether to record a model-visible message when an agent turn is interrupted.
/// Defaults to true.
-1
View File
@@ -79,7 +79,6 @@ codex-utils-pty = { workspace = true }
codex-utils-string = { workspace = true }
codex-utils-stream-parser = { workspace = true }
codex-windows-sandbox = { package = "codex-windows-sandbox", path = "../windows-sandbox-rs" }
csv = { workspace = true }
dirs = { workspace = true }
dunce = { workspace = true }
eventsource-stream = { workspace = true }
-6
View File
@@ -56,12 +56,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_concurrent_threads_per_session": {
"description": "Maximum number of spawned agent threads that can be open concurrently per session. When unset, the selected multi-agent backend uses its default.",
"format": "uint",
+19
View File
@@ -8359,6 +8359,25 @@ model = "gpt-5-mini"
Ok(())
}
#[test]
fn legacy_agent_job_max_runtime_seconds_is_accepted_as_noop() {
let parsed = toml::from_str::<ConfigToml>(
r#"
[agents]
job_max_runtime_seconds = 900
"#,
)
.expect("legacy agent job setting should deserialize");
assert_eq!(
parsed.agents,
Some(AgentsToml {
job_max_runtime_seconds: Some(900),
..Default::default()
})
);
}
#[tokio::test]
async fn load_config_resolves_agent_controls() -> std::io::Result<()> {
let codex_home = TempDir::new()?;
-24
View File
@@ -266,7 +266,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";
@@ -874,9 +873,6 @@ pub struct Config {
/// Default reasoning effort for spawned subagents when the spawn call does not select one.
pub agent_default_subagent_reasoning_effort: Option<ReasoningEffort>,
/// 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,
@@ -3616,25 +3612,6 @@ impl Config {
.agents
.as_ref()
.and_then(|agents| agents.default_subagent_reasoning_effort.clone());
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()
@@ -3973,7 +3950,6 @@ impl Config {
agent_max_depth,
agent_roles,
memories: memories_config,
agent_job_max_runtime_seconds,
agent_interrupt_message_enabled,
codex_home,
sqlite_home,
@@ -1050,7 +1050,6 @@ pub(crate) fn build_guardian_review_session_config(
)?);
}
for feature in [
Feature::SpawnCsv,
Feature::Collab,
Feature::MultiAgentV2,
Feature::CodexHooks,
-1
View File
@@ -183,7 +183,6 @@ fn save_config_resolved_fields(
agents.default_subagent_model = config.agent_default_subagent_model.clone();
agents.default_subagent_reasoning_effort =
config.agent_default_subagent_reasoning_effort.clone();
agents.job_max_runtime_seconds = config.agent_job_max_runtime_seconds;
agents.interrupt_message = Some(config.agent_interrupt_message_enabled);
lock_config
-1
View File
@@ -110,7 +110,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);
@@ -1,771 +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::multi_agents_common::apply_requested_spawn_agent_model_overrides;
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 mut spawn_config = build_agent_spawn_config(&base_instructions, turn.as_ref())?;
apply_requested_spawn_agent_model_overrides(
session,
turn,
&mut spawn_config,
/*requested_model*/ None,
/*requested_reasoning_effort*/ None,
)
.await?;
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,
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;
@@ -1,98 +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;
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()
}
fn handle(&self, invocation: ToolInvocation) -> codex_tools::ToolExecutorFuture<'_> {
Box::pin(self.handle_call(invocation))
}
}
impl ReportAgentJobResultHandler {
async fn handle_call(
&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)))
}
@@ -1,329 +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;
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()
}
fn handle(&self, invocation: ToolInvocation) -> codex_tools::ToolExecutorFuture<'_> {
Box::pin(self.handle_call(invocation))
}
}
impl SpawnAgentsOnCsvHandler {
async fn handle_call(
&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 mut environments = turn.environments.turn_environments();
let Some(turn_environment) = environments.next() else {
return Err(FunctionCallError::RespondToModel(
"spawn_agents_on_csv requires exactly one local environment".to_string(),
));
};
if environments.next().is_some() {
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(),
));
}
// TODO(anp): Migrate spawn_agents_on_csv filesystem access to PathUri before enabling it for
// remote environments.
turn_environment.cwd().to_abs_path().map_err(|err| {
FunctionCallError::RespondToModel(format!(
"spawn_agents_on_csv cwd `{}` is not native to the Codex host: {err}",
turn_environment.cwd()
))
})
}
@@ -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;
@@ -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,
})
);
}
@@ -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())
);
}
-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 current_time;
-29
View File
@@ -30,8 +30,6 @@ use crate::tools::handlers::ToolSearchHandlerCache;
use crate::tools::handlers::ViewImageHandler;
use crate::tools::handlers::WaitForEnvironmentHandler;
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;
@@ -67,8 +65,6 @@ use codex_protocol::openai_models::ConfigShellToolType;
use codex_protocol::openai_models::InputModality;
use codex_protocol::openai_models::ToolMode;
use codex_protocol::protocol::MultiAgentVersion;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SubAgentSource;
use codex_tools::ResponsesApiNamespace;
use codex_tools::ResponsesApiNamespaceTool;
use codex_tools::TOOL_SEARCH_TOOL_NAME;
@@ -346,24 +342,6 @@ fn collab_tools_enabled(turn_context: &TurnContext) -> bool {
}
}
fn agent_jobs_tools_enabled(turn_context: &TurnContext) -> bool {
turn_context
.config
.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_available(turn_context: &TurnContext) -> bool {
if !turn_context
.config
@@ -870,13 +848,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);
}
}
}
#[instrument(level = "trace", skip_all, fields(dynamic_tool_count = context.dynamic_tools.len()))]
@@ -15,8 +15,6 @@ use codex_protocol::openai_models::ConfigShellToolType;
use codex_protocol::openai_models::InputModality;
use codex_protocol::openai_models::ToolMode;
use codex_protocol::openai_models::WebSearchToolType;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SubAgentSource;
use codex_tools::DiscoverablePluginInfo;
use codex_tools::DiscoverableTool;
use codex_tools::ResponsesApiNamespaceTool;
@@ -704,24 +702,6 @@ async fn environment_tools_follow_the_step_context() {
plan.assert_visible_contains(&["exec_command", "apply_patch", "view_image"]);
}
#[tokio::test]
async fn host_context_gates_agent_job_tools() {
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]
async fn sleep_tool_follows_current_time_config() {
let disabled = probe(|turn| {
-4
View File
@@ -341,10 +341,6 @@ fn turn_metadata_state_includes_known_parent_for_non_thread_spawn_subagents_with
let sources = [
(SubAgentSource::Review, "review"),
(SubAgentSource::Other("guardian".to_string()), "guardian"),
(
SubAgentSource::Other("agent_job:job-1".to_string()),
"agent_job:job-1",
),
];
for (subagent_source, subagent_kind) in sources {
-469
View File
@@ -1,469 +0,0 @@
use anyhow::Result;
use codex_features::Feature;
use codex_protocol::openai_models::ReasoningEffort;
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 pretty_assertions::assert_eq;
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).expect("worker args should serialize");
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).expect("worker args should serialize");
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");
config.agent_default_subagent_model = Some("gpt-5.4".to_string());
config.agent_default_subagent_reasoning_effort = Some(ReasoningEffort::High);
});
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\""));
let worker_settings = server
.received_requests()
.await
.unwrap_or_default()
.into_iter()
.filter_map(|request| {
let body = serde_json::from_slice::<Value>(&decode_body_bytes(&request)).ok()?;
if has_function_call_output(&body) {
return None;
}
extract_job_and_item(&body)?;
Some((
body["model"].as_str().map(str::to_string),
body["reasoning"]["effort"].as_str().map(str::to_string),
))
})
.collect::<Vec<_>>();
assert_eq!(
worker_settings,
vec![(Some("gpt-5.4".to_string()), Some("high".to_string())); 2]
);
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(())
}
-1
View File
@@ -31,7 +31,6 @@ pub static CODEX_ALIASES_TEMP_DIR: Option<TestBinaryDispatchGuard> = {
mod abort_tasks;
mod additional_context;
mod agent_execution;
mod agent_jobs;
mod agent_websocket;
mod agents_md;
mod apply_patch_cli;
+2 -5
View File
@@ -152,7 +152,7 @@ pub enum Feature {
MultiAgentV2,
/// Removed compatibility flag retained as a no-op.
MultiAgentMode,
/// Enable CSV-backed agent job tools.
/// Removed compatibility flag for the deleted agent-job tools.
SpawnCsv,
/// Enable apps.
Apps,
@@ -555,9 +555,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);
}
@@ -1061,7 +1058,7 @@ pub const FEATURES: &[FeatureSpec] = &[
FeatureSpec {
id: Feature::SpawnCsv,
key: "enable_fanout",
stage: Stage::UnderDevelopment,
stage: Stage::Removed,
default_enabled: false,
},
FeatureSpec {
-15
View File
@@ -191,21 +191,6 @@ fn codex_hooks_is_legacy_alias_for_hooks() {
assert_eq!(feature_for_key("codex_hooks"), Some(Feature::CodexHooks));
}
#[test]
fn enable_fanout_normalization_enables_multi_agent_one_way() {
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]
fn apps_require_feature_flag_and_chatgpt_auth() {
let mut features = Features::with_defaults();
-1
View File
@@ -326,7 +326,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);
@@ -0,0 +1,2 @@
DROP TABLE IF EXISTS agent_job_items;
DROP TABLE IF EXISTS agent_jobs;
-7
View File
@@ -34,13 +34,6 @@ pub use audit::read_thread_state_audit_rows;
/// Most consumers should prefer [`StateRuntime`].
pub use extract::apply_rollout_item;
pub use extract::rollout_item_affects_thread_metadata;
pub use model::AgentJob;
pub use model::AgentJobCreateParams;
pub use model::AgentJobItem;
pub use model::AgentJobItemCreateParams;
pub use model::AgentJobItemStatus;
pub use model::AgentJobProgress;
pub use model::AgentJobStatus;
pub use model::Anchor;
pub use model::BackfillState;
pub use model::BackfillStats;
+93
View File
@@ -27,6 +27,99 @@ fn migrator_through(version: i64) -> Migrator {
}
}
#[tokio::test]
async fn agent_job_tables_are_dropped_when_upgrading() {
let sqlite_home = crate::runtime::test_support::unique_temp_dir();
tokio::fs::create_dir_all(&sqlite_home)
.await
.expect("sqlite home should be created");
let _cleanup = scopeguard::guard(sqlite_home.clone(), |sqlite_home| {
let _ = std::fs::remove_dir_all(sqlite_home);
});
let sqlite = crate::SqliteConfig::new_for_testing(sqlite_home.clone());
let pool = sqlite
.open_read_write_pool(&state_db_path(&sqlite_home))
.await
.expect("sqlite database should open");
migrator_through(/*version*/ 15)
.run(&pool)
.await
.expect("agent job migrations should apply");
sqlx::query(
r#"
INSERT INTO agent_jobs (
id,
name,
status,
instruction,
input_headers_json,
input_csv_path,
output_csv_path,
created_at,
updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
"#,
)
.bind("job-1")
.bind("legacy job")
.bind("running")
.bind("process rows")
.bind(r#"["path"]"#)
.bind("/tmp/input.csv")
.bind("/tmp/output.csv")
.bind(1_700_000_000_i64)
.bind(1_700_000_000_i64)
.execute(&pool)
.await
.expect("legacy agent job should insert");
sqlx::query(
r#"
INSERT INTO agent_job_items (
job_id,
item_id,
row_index,
row_json,
status,
result_json,
created_at,
updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
"#,
)
.bind("job-1")
.bind("item-1")
.bind(0_i64)
.bind(r#"{"path":"secret.csv"}"#)
.bind("completed")
.bind(r#"{"result":"legacy"}"#)
.bind(1_700_000_000_i64)
.bind(1_700_000_000_i64)
.execute(&pool)
.await
.expect("legacy agent job item should insert");
STATE_MIGRATOR
.run(&pool)
.await
.expect("current migrations should apply");
let agent_job_tables = sqlx::query_scalar::<_, String>(
r#"
SELECT name
FROM sqlite_master
WHERE type = 'table' AND name IN ('agent_jobs', 'agent_job_items')
ORDER BY name
"#,
)
.fetch_all(&pool)
.await
.expect("remaining agent job tables should load");
assert_eq!(agent_job_tables, Vec::<String>::new());
pool.close().await;
}
#[tokio::test]
async fn recency_migration_backfills_and_seeds_old_binary_inserts() {
let sqlite_home = crate::runtime::test_support::unique_temp_dir();
-256
View File
@@ -1,256 +0,0 @@
use anyhow::Result;
use chrono::DateTime;
use chrono::Utc;
use serde_json::Value;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AgentJobStatus {
Pending,
Running,
Completed,
Failed,
Cancelled,
}
impl AgentJobStatus {
pub const fn as_str(self) -> &'static str {
match self {
AgentJobStatus::Pending => "pending",
AgentJobStatus::Running => "running",
AgentJobStatus::Completed => "completed",
AgentJobStatus::Failed => "failed",
AgentJobStatus::Cancelled => "cancelled",
}
}
pub fn parse(value: &str) -> Result<Self> {
match value {
"pending" => Ok(Self::Pending),
"running" => Ok(Self::Running),
"completed" => Ok(Self::Completed),
"failed" => Ok(Self::Failed),
"cancelled" => Ok(Self::Cancelled),
_ => Err(anyhow::anyhow!("invalid agent job status: {value}")),
}
}
pub fn is_final(self) -> bool {
matches!(
self,
AgentJobStatus::Completed | AgentJobStatus::Failed | AgentJobStatus::Cancelled
)
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AgentJobItemStatus {
Pending,
Running,
Completed,
Failed,
}
impl AgentJobItemStatus {
pub const fn as_str(self) -> &'static str {
match self {
AgentJobItemStatus::Pending => "pending",
AgentJobItemStatus::Running => "running",
AgentJobItemStatus::Completed => "completed",
AgentJobItemStatus::Failed => "failed",
}
}
pub fn parse(value: &str) -> Result<Self> {
match value {
"pending" => Ok(Self::Pending),
"running" => Ok(Self::Running),
"completed" => Ok(Self::Completed),
"failed" => Ok(Self::Failed),
_ => Err(anyhow::anyhow!("invalid agent job item status: {value}")),
}
}
}
#[derive(Debug, Clone, PartialEq)]
pub struct AgentJob {
pub id: String,
pub name: String,
pub status: AgentJobStatus,
pub instruction: String,
pub auto_export: bool,
pub max_runtime_seconds: Option<u64>,
// TODO(jif-oai): Convert to JSON Schema and enforce structured outputs.
pub output_schema_json: Option<Value>,
pub input_headers: Vec<String>,
pub input_csv_path: String,
pub output_csv_path: String,
pub created_at: DateTime<Utc>,
pub updated_at: DateTime<Utc>,
pub started_at: Option<DateTime<Utc>>,
pub completed_at: Option<DateTime<Utc>>,
pub last_error: Option<String>,
}
#[derive(Debug, Clone, PartialEq)]
pub struct AgentJobItem {
pub job_id: String,
pub item_id: String,
pub row_index: i64,
pub source_id: Option<String>,
pub row_json: Value,
pub status: AgentJobItemStatus,
pub assigned_thread_id: Option<String>,
pub attempt_count: i64,
pub result_json: Option<Value>,
pub last_error: Option<String>,
pub created_at: DateTime<Utc>,
pub updated_at: DateTime<Utc>,
pub completed_at: Option<DateTime<Utc>>,
pub reported_at: Option<DateTime<Utc>>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AgentJobProgress {
pub total_items: usize,
pub pending_items: usize,
pub running_items: usize,
pub completed_items: usize,
pub failed_items: usize,
}
#[derive(Debug, Clone)]
pub struct AgentJobCreateParams {
pub id: String,
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,
pub output_csv_path: String,
}
#[derive(Debug, Clone)]
pub struct AgentJobItemCreateParams {
pub item_id: String,
pub row_index: i64,
pub source_id: Option<String>,
pub row_json: Value,
}
#[derive(Debug, sqlx::FromRow)]
pub(crate) struct AgentJobRow {
pub(crate) id: String,
pub(crate) name: String,
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,
pub(crate) output_csv_path: String,
pub(crate) created_at: i64,
pub(crate) updated_at: i64,
pub(crate) started_at: Option<i64>,
pub(crate) completed_at: Option<i64>,
pub(crate) last_error: Option<String>,
}
impl TryFrom<AgentJobRow> for AgentJob {
type Error = anyhow::Error;
fn try_from(value: AgentJobRow) -> Result<Self, Self::Error> {
let output_schema_json = value
.output_schema_json
.as_deref()
.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,
output_csv_path: value.output_csv_path,
created_at: epoch_seconds_to_datetime(value.created_at)?,
updated_at: epoch_seconds_to_datetime(value.updated_at)?,
started_at: value
.started_at
.map(epoch_seconds_to_datetime)
.transpose()?,
completed_at: value
.completed_at
.map(epoch_seconds_to_datetime)
.transpose()?,
last_error: value.last_error,
})
}
}
#[derive(Debug, sqlx::FromRow)]
pub(crate) struct AgentJobItemRow {
pub(crate) job_id: String,
pub(crate) item_id: String,
pub(crate) row_index: i64,
pub(crate) source_id: Option<String>,
pub(crate) row_json: String,
pub(crate) status: String,
pub(crate) assigned_thread_id: Option<String>,
pub(crate) attempt_count: i64,
pub(crate) result_json: Option<String>,
pub(crate) last_error: Option<String>,
pub(crate) created_at: i64,
pub(crate) updated_at: i64,
pub(crate) completed_at: Option<i64>,
pub(crate) reported_at: Option<i64>,
}
impl TryFrom<AgentJobItemRow> for AgentJobItem {
type Error = anyhow::Error;
fn try_from(value: AgentJobItemRow) -> Result<Self, Self::Error> {
Ok(Self {
job_id: value.job_id,
item_id: value.item_id,
row_index: value.row_index,
source_id: value.source_id,
row_json: serde_json::from_str(value.row_json.as_str())?,
status: AgentJobItemStatus::parse(value.status.as_str())?,
assigned_thread_id: value.assigned_thread_id,
attempt_count: value.attempt_count,
result_json: value
.result_json
.as_deref()
.map(serde_json::from_str)
.transpose()?,
last_error: value.last_error,
created_at: epoch_seconds_to_datetime(value.created_at)?,
updated_at: epoch_seconds_to_datetime(value.updated_at)?,
completed_at: value
.completed_at
.map(epoch_seconds_to_datetime)
.transpose()?,
reported_at: value
.reported_at
.map(epoch_seconds_to_datetime)
.transpose()?,
})
}
}
fn epoch_seconds_to_datetime(secs: i64) -> Result<DateTime<Utc>> {
DateTime::<Utc>::from_timestamp(secs, 0)
.ok_or_else(|| anyhow::anyhow!("invalid unix timestamp: {secs}"))
}
-10
View File
@@ -1,4 +1,3 @@
mod agent_job;
mod backfill_state;
mod graph;
mod log;
@@ -6,13 +5,6 @@ mod memories;
mod thread_goal;
mod thread_metadata;
pub use agent_job::AgentJob;
pub use agent_job::AgentJobCreateParams;
pub use agent_job::AgentJobItem;
pub use agent_job::AgentJobItemCreateParams;
pub use agent_job::AgentJobItemStatus;
pub use agent_job::AgentJobProgress;
pub use agent_job::AgentJobStatus;
pub use backfill_state::BackfillState;
pub use backfill_state::BackfillStatus;
pub use graph::DirectionalThreadSpawnEdgeStatus;
@@ -36,8 +28,6 @@ pub use thread_metadata::ThreadMetadataBuilder;
pub use thread_metadata::ThreadRelationFilter;
pub use thread_metadata::ThreadsPage;
pub(crate) use agent_job::AgentJobItemRow;
pub(crate) use agent_job::AgentJobRow;
pub(crate) use thread_goal::ThreadGoalRow;
pub(crate) use thread_metadata::ThreadRow;
pub(crate) use thread_metadata::anchor_from_item;
-9
View File
@@ -1,10 +1,3 @@
use crate::AgentJob;
use crate::AgentJobCreateParams;
use crate::AgentJobItem;
use crate::AgentJobItemCreateParams;
use crate::AgentJobItemStatus;
use crate::AgentJobProgress;
use crate::AgentJobStatus;
use crate::GOALS_DB_FILENAME;
use crate::LOGS_DB_FILENAME;
use crate::LogEntry;
@@ -25,7 +18,6 @@ use crate::migrations::runtime_logs_migrator;
use crate::migrations::runtime_memories_migrator;
use crate::migrations::runtime_state_migrator;
use crate::migrations::runtime_thread_history_migrator;
use crate::model::AgentJobRow;
use crate::model::ThreadRow;
use crate::model::anchor_from_item;
use crate::model::datetime_to_epoch_millis;
@@ -54,7 +46,6 @@ use std::sync::atomic::AtomicI64;
use std::time::Instant;
use tracing::warn;
mod agent_jobs;
mod backfill;
mod external_agent_config_imports;
mod goals;
-684
View File
@@ -1,684 +0,0 @@
use super::*;
use crate::model::AgentJobItemRow;
impl StateRuntime {
pub async fn create_agent_job(
&self,
params: &AgentJobCreateParams,
items: &[AgentJobItemCreateParams],
) -> anyhow::Result<AgentJob> {
let now = Utc::now().timestamp();
let input_headers_json = serde_json::to_string(&params.input_headers)?;
let output_schema_json = params
.output_schema_json
.as_ref()
.map(serde_json::to_string)
.transpose()?;
let max_runtime_seconds = params
.max_runtime_seconds
.map(i64::try_from)
.transpose()
.map_err(|_| anyhow::anyhow!("invalid max_runtime_seconds value"))?;
let mut tx = self.pool.begin().await?;
sqlx::query(
r#"
INSERT INTO agent_jobs (
id,
name,
status,
instruction,
auto_export,
max_runtime_seconds,
output_schema_json,
input_headers_json,
input_csv_path,
output_csv_path,
created_at,
updated_at,
started_at,
completed_at,
last_error
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, NULL, NULL)
"#,
)
.bind(params.id.as_str())
.bind(params.name.as_str())
.bind(AgentJobStatus::Pending.as_str())
.bind(params.instruction.as_str())
.bind(i64::from(params.auto_export))
.bind(max_runtime_seconds)
.bind(output_schema_json)
.bind(input_headers_json)
.bind(params.input_csv_path.as_str())
.bind(params.output_csv_path.as_str())
.bind(now)
.bind(now)
.execute(&mut *tx)
.await?;
for item in items {
let row_json = serde_json::to_string(&item.row_json)?;
sqlx::query(
r#"
INSERT INTO agent_job_items (
job_id,
item_id,
row_index,
source_id,
row_json,
status,
assigned_thread_id,
attempt_count,
result_json,
last_error,
created_at,
updated_at,
completed_at,
reported_at
) VALUES (?, ?, ?, ?, ?, ?, NULL, 0, NULL, NULL, ?, ?, NULL, NULL)
"#,
)
.bind(params.id.as_str())
.bind(item.item_id.as_str())
.bind(item.row_index)
.bind(item.source_id.as_deref())
.bind(row_json)
.bind(AgentJobItemStatus::Pending.as_str())
.bind(now)
.bind(now)
.execute(&mut *tx)
.await?;
}
tx.commit().await?;
let job_id = params.id.as_str();
self.get_agent_job(job_id)
.await?
.ok_or_else(|| anyhow::anyhow!("failed to load created agent job {job_id}"))
}
pub async fn get_agent_job(&self, job_id: &str) -> anyhow::Result<Option<AgentJob>> {
let row = sqlx::query_as::<_, AgentJobRow>(
r#"
SELECT
id,
name,
status,
instruction,
auto_export,
max_runtime_seconds,
output_schema_json,
input_headers_json,
input_csv_path,
output_csv_path,
created_at,
updated_at,
started_at,
completed_at,
last_error
FROM agent_jobs
WHERE id = ?
"#,
)
.bind(job_id)
.fetch_optional(self.pool.as_ref())
.await?;
row.map(AgentJob::try_from).transpose()
}
pub async fn list_agent_job_items(
&self,
job_id: &str,
status: Option<AgentJobItemStatus>,
limit: Option<usize>,
) -> anyhow::Result<Vec<AgentJobItem>> {
let mut builder = QueryBuilder::<Sqlite>::new(
r#"
SELECT
job_id,
item_id,
row_index,
source_id,
row_json,
status,
assigned_thread_id,
attempt_count,
result_json,
last_error,
created_at,
updated_at,
completed_at,
reported_at
FROM agent_job_items
WHERE job_id =
"#,
);
builder.push_bind(job_id);
if let Some(status) = status {
builder.push(" AND status = ");
builder.push_bind(status.as_str());
}
builder.push(" ORDER BY row_index ASC");
if let Some(limit) = limit {
builder.push(" LIMIT ");
builder.push_bind(limit as i64);
}
let rows: Vec<AgentJobItemRow> = builder
.build_query_as::<AgentJobItemRow>()
.fetch_all(self.pool.as_ref())
.await?;
rows.into_iter().map(AgentJobItem::try_from).collect()
}
pub async fn get_agent_job_item(
&self,
job_id: &str,
item_id: &str,
) -> anyhow::Result<Option<AgentJobItem>> {
let row: Option<AgentJobItemRow> = sqlx::query_as::<_, AgentJobItemRow>(
r#"
SELECT
job_id,
item_id,
row_index,
source_id,
row_json,
status,
assigned_thread_id,
attempt_count,
result_json,
last_error,
created_at,
updated_at,
completed_at,
reported_at
FROM agent_job_items
WHERE job_id = ? AND item_id = ?
"#,
)
.bind(job_id)
.bind(item_id)
.fetch_optional(self.pool.as_ref())
.await?;
row.map(AgentJobItem::try_from).transpose()
}
pub async fn mark_agent_job_running(&self, job_id: &str) -> anyhow::Result<()> {
let now = Utc::now().timestamp();
sqlx::query(
r#"
UPDATE agent_jobs
SET
status = ?,
updated_at = ?,
started_at = COALESCE(started_at, ?),
completed_at = NULL,
last_error = NULL
WHERE id = ?
"#,
)
.bind(AgentJobStatus::Running.as_str())
.bind(now)
.bind(now)
.bind(job_id)
.execute(self.pool.as_ref())
.await?;
Ok(())
}
pub async fn mark_agent_job_completed(&self, job_id: &str) -> anyhow::Result<()> {
let now = Utc::now().timestamp();
sqlx::query(
r#"
UPDATE agent_jobs
SET status = ?, updated_at = ?, completed_at = ?, last_error = NULL
WHERE id = ?
"#,
)
.bind(AgentJobStatus::Completed.as_str())
.bind(now)
.bind(now)
.bind(job_id)
.execute(self.pool.as_ref())
.await?;
Ok(())
}
pub async fn mark_agent_job_failed(
&self,
job_id: &str,
error_message: &str,
) -> anyhow::Result<()> {
let now = Utc::now().timestamp();
sqlx::query(
r#"
UPDATE agent_jobs
SET status = ?, updated_at = ?, completed_at = ?, last_error = ?
WHERE id = ?
"#,
)
.bind(AgentJobStatus::Failed.as_str())
.bind(now)
.bind(now)
.bind(error_message)
.bind(job_id)
.execute(self.pool.as_ref())
.await?;
Ok(())
}
pub async fn mark_agent_job_cancelled(
&self,
job_id: &str,
reason: &str,
) -> anyhow::Result<bool> {
let now = Utc::now().timestamp();
let result = sqlx::query(
r#"
UPDATE agent_jobs
SET status = ?, updated_at = ?, completed_at = ?, last_error = ?
WHERE id = ? AND status IN (?, ?)
"#,
)
.bind(AgentJobStatus::Cancelled.as_str())
.bind(now)
.bind(now)
.bind(reason)
.bind(job_id)
.bind(AgentJobStatus::Pending.as_str())
.bind(AgentJobStatus::Running.as_str())
.execute(self.pool.as_ref())
.await?;
Ok(result.rows_affected() > 0)
}
pub async fn is_agent_job_cancelled(&self, job_id: &str) -> anyhow::Result<bool> {
let row = sqlx::query(
r#"
SELECT status
FROM agent_jobs
WHERE id = ?
"#,
)
.bind(job_id)
.fetch_optional(self.pool.as_ref())
.await?;
let Some(row) = row else {
return Ok(false);
};
let status: String = row.try_get("status")?;
Ok(AgentJobStatus::parse(status.as_str())? == AgentJobStatus::Cancelled)
}
pub async fn mark_agent_job_item_running(
&self,
job_id: &str,
item_id: &str,
) -> anyhow::Result<bool> {
let now = Utc::now().timestamp();
let result = sqlx::query(
r#"
UPDATE agent_job_items
SET
status = ?,
assigned_thread_id = NULL,
attempt_count = attempt_count + 1,
updated_at = ?,
last_error = NULL
WHERE job_id = ? AND item_id = ? AND status = ?
"#,
)
.bind(AgentJobItemStatus::Running.as_str())
.bind(now)
.bind(job_id)
.bind(item_id)
.bind(AgentJobItemStatus::Pending.as_str())
.execute(self.pool.as_ref())
.await?;
Ok(result.rows_affected() > 0)
}
pub async fn mark_agent_job_item_running_with_thread(
&self,
job_id: &str,
item_id: &str,
thread_id: &str,
) -> anyhow::Result<bool> {
let now = Utc::now().timestamp();
let result = sqlx::query(
r#"
UPDATE agent_job_items
SET
status = ?,
assigned_thread_id = ?,
attempt_count = attempt_count + 1,
updated_at = ?,
last_error = NULL
WHERE job_id = ? AND item_id = ? AND status = ?
"#,
)
.bind(AgentJobItemStatus::Running.as_str())
.bind(thread_id)
.bind(now)
.bind(job_id)
.bind(item_id)
.bind(AgentJobItemStatus::Pending.as_str())
.execute(self.pool.as_ref())
.await?;
Ok(result.rows_affected() > 0)
}
pub async fn mark_agent_job_item_pending(
&self,
job_id: &str,
item_id: &str,
error_message: Option<&str>,
) -> anyhow::Result<bool> {
let now = Utc::now().timestamp();
let result = sqlx::query(
r#"
UPDATE agent_job_items
SET
status = ?,
assigned_thread_id = NULL,
updated_at = ?,
last_error = ?
WHERE job_id = ? AND item_id = ? AND status = ?
"#,
)
.bind(AgentJobItemStatus::Pending.as_str())
.bind(now)
.bind(error_message)
.bind(job_id)
.bind(item_id)
.bind(AgentJobItemStatus::Running.as_str())
.execute(self.pool.as_ref())
.await?;
Ok(result.rows_affected() > 0)
}
pub async fn set_agent_job_item_thread(
&self,
job_id: &str,
item_id: &str,
thread_id: &str,
) -> anyhow::Result<bool> {
let now = Utc::now().timestamp();
let result = sqlx::query(
r#"
UPDATE agent_job_items
SET assigned_thread_id = ?, updated_at = ?
WHERE job_id = ? AND item_id = ? AND status = ?
"#,
)
.bind(thread_id)
.bind(now)
.bind(job_id)
.bind(item_id)
.bind(AgentJobItemStatus::Running.as_str())
.execute(self.pool.as_ref())
.await?;
Ok(result.rows_affected() > 0)
}
pub async fn report_agent_job_item_result(
&self,
job_id: &str,
item_id: &str,
reporting_thread_id: &str,
result_json: &Value,
) -> anyhow::Result<bool> {
let now = Utc::now().timestamp();
let serialized = serde_json::to_string(result_json)?;
let result = sqlx::query(
r#"
UPDATE agent_job_items
SET
status = ?,
result_json = ?,
reported_at = ?,
completed_at = ?,
updated_at = ?,
last_error = NULL,
assigned_thread_id = NULL
WHERE
job_id = ?
AND item_id = ?
AND status = ?
AND assigned_thread_id = ?
"#,
)
.bind(AgentJobItemStatus::Completed.as_str())
.bind(serialized)
.bind(now)
.bind(now)
.bind(now)
.bind(job_id)
.bind(item_id)
.bind(AgentJobItemStatus::Running.as_str())
.bind(reporting_thread_id)
.execute(self.pool.as_ref())
.await?;
Ok(result.rows_affected() > 0)
}
pub async fn mark_agent_job_item_completed(
&self,
job_id: &str,
item_id: &str,
) -> anyhow::Result<bool> {
let now = Utc::now().timestamp();
let result = sqlx::query(
r#"
UPDATE agent_job_items
SET
status = ?,
completed_at = ?,
updated_at = ?,
assigned_thread_id = NULL
WHERE
job_id = ?
AND item_id = ?
AND status = ?
AND result_json IS NOT NULL
"#,
)
.bind(AgentJobItemStatus::Completed.as_str())
.bind(now)
.bind(now)
.bind(job_id)
.bind(item_id)
.bind(AgentJobItemStatus::Running.as_str())
.execute(self.pool.as_ref())
.await?;
Ok(result.rows_affected() > 0)
}
pub async fn mark_agent_job_item_failed(
&self,
job_id: &str,
item_id: &str,
error_message: &str,
) -> anyhow::Result<bool> {
let now = Utc::now().timestamp();
let result = sqlx::query(
r#"
UPDATE agent_job_items
SET
status = ?,
completed_at = ?,
updated_at = ?,
last_error = ?,
assigned_thread_id = NULL
WHERE
job_id = ?
AND item_id = ?
AND status = ?
"#,
)
.bind(AgentJobItemStatus::Failed.as_str())
.bind(now)
.bind(now)
.bind(error_message)
.bind(job_id)
.bind(item_id)
.bind(AgentJobItemStatus::Running.as_str())
.execute(self.pool.as_ref())
.await?;
Ok(result.rows_affected() > 0)
}
pub async fn get_agent_job_progress(&self, job_id: &str) -> anyhow::Result<AgentJobProgress> {
let row = sqlx::query(
r#"
SELECT
COUNT(*) AS total_items,
SUM(CASE WHEN status = ? THEN 1 ELSE 0 END) AS pending_items,
SUM(CASE WHEN status = ? THEN 1 ELSE 0 END) AS running_items,
SUM(CASE WHEN status = ? THEN 1 ELSE 0 END) AS completed_items,
SUM(CASE WHEN status = ? THEN 1 ELSE 0 END) AS failed_items
FROM agent_job_items
WHERE job_id = ?
"#,
)
.bind(AgentJobItemStatus::Pending.as_str())
.bind(AgentJobItemStatus::Running.as_str())
.bind(AgentJobItemStatus::Completed.as_str())
.bind(AgentJobItemStatus::Failed.as_str())
.bind(job_id)
.fetch_one(self.pool.as_ref())
.await?;
let total_items: i64 = row.try_get("total_items")?;
let pending_items: Option<i64> = row.try_get("pending_items")?;
let running_items: Option<i64> = row.try_get("running_items")?;
let completed_items: Option<i64> = row.try_get("completed_items")?;
let failed_items: Option<i64> = row.try_get("failed_items")?;
Ok(AgentJobProgress {
total_items: usize::try_from(total_items).unwrap_or_default(),
pending_items: usize::try_from(pending_items.unwrap_or_default()).unwrap_or_default(),
running_items: usize::try_from(running_items.unwrap_or_default()).unwrap_or_default(),
completed_items: usize::try_from(completed_items.unwrap_or_default())
.unwrap_or_default(),
failed_items: usize::try_from(failed_items.unwrap_or_default()).unwrap_or_default(),
})
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::runtime::test_support::unique_temp_dir;
use pretty_assertions::assert_eq;
use serde_json::json;
async fn create_running_single_item_job(
runtime: &StateRuntime,
) -> anyhow::Result<(String, String, String)> {
let job_id = "job-1".to_string();
let item_id = "item-1".to_string();
let thread_id = "thread-1".to_string();
runtime
.create_agent_job(
&AgentJobCreateParams {
id: job_id.clone(),
name: "test-job".to_string(),
instruction: "Return a result".to_string(),
auto_export: true,
max_runtime_seconds: None,
output_schema_json: None,
input_headers: vec!["path".to_string()],
input_csv_path: "/tmp/in.csv".to_string(),
output_csv_path: "/tmp/out.csv".to_string(),
},
&[AgentJobItemCreateParams {
item_id: item_id.clone(),
row_index: 0,
source_id: None,
row_json: json!({"path":"file-1"}),
}],
)
.await?;
runtime.mark_agent_job_running(job_id.as_str()).await?;
let marked_running = runtime
.mark_agent_job_item_running_with_thread(
job_id.as_str(),
item_id.as_str(),
thread_id.as_str(),
)
.await?;
assert!(marked_running);
Ok((job_id, item_id, thread_id))
}
#[tokio::test]
async fn report_agent_job_item_result_completes_item_atomically() -> anyhow::Result<()> {
let codex_home = unique_temp_dir();
let runtime = StateRuntime::init(codex_home, "test-provider".to_string()).await?;
let (job_id, item_id, thread_id) = create_running_single_item_job(runtime.as_ref()).await?;
let accepted = runtime
.report_agent_job_item_result(
job_id.as_str(),
item_id.as_str(),
thread_id.as_str(),
&json!({"ok": true}),
)
.await?;
assert!(accepted);
let item = runtime
.get_agent_job_item(job_id.as_str(), item_id.as_str())
.await?
.expect("job item should exist");
assert_eq!(item.status, AgentJobItemStatus::Completed);
assert_eq!(item.result_json, Some(json!({"ok": true})));
assert_eq!(item.assigned_thread_id, None);
assert_eq!(item.last_error, None);
assert!(item.reported_at.is_some());
assert!(item.completed_at.is_some());
let progress = runtime.get_agent_job_progress(job_id.as_str()).await?;
assert_eq!(
progress,
AgentJobProgress {
total_items: 1,
pending_items: 0,
running_items: 0,
completed_items: 1,
failed_items: 0,
}
);
Ok(())
}
#[tokio::test]
async fn report_agent_job_item_result_rejects_late_reports() -> anyhow::Result<()> {
let codex_home = unique_temp_dir();
let runtime = StateRuntime::init(codex_home, "test-provider".to_string()).await?;
let (job_id, item_id, thread_id) = create_running_single_item_job(runtime.as_ref()).await?;
let marked_failed = runtime
.mark_agent_job_item_failed(job_id.as_str(), item_id.as_str(), "missing report")
.await?;
assert!(marked_failed);
let accepted = runtime
.report_agent_job_item_result(
job_id.as_str(),
item_id.as_str(),
thread_id.as_str(),
&json!({"late": true}),
)
.await?;
assert!(!accepted);
let item = runtime
.get_agent_job_item(job_id.as_str(), item_id.as_str())
.await?
.expect("job item should exist");
assert_eq!(item.status, AgentJobItemStatus::Failed);
assert_eq!(item.result_json, None);
assert_eq!(item.last_error, Some("missing report".to_string()));
Ok(())
}
}
+8 -1
View File
@@ -330,7 +330,14 @@ mod tests {
.await
.expect("create codex home");
let old_state_migrator = Migrator {
migrations: Cow::Owned(STATE_MIGRATOR.migrations[..36].to_vec()),
migrations: Cow::Owned(
STATE_MIGRATOR
.migrations
.iter()
.filter(|migration| migration.version <= 36)
.cloned()
.collect(),
),
ignore_missing: false,
locking: true,
no_tx: false,
-104
View File
@@ -1040,70 +1040,12 @@ ON CONFLICT(id) DO UPDATE SET
self.thread_goals.delete_thread_goal(*thread_id).await?;
}
let now = Utc::now().timestamp();
let mut tx = self.pool.begin().await?;
for thread_id_string in &thread_id_strings {
for parent_thread_id_string in &thread_id_strings {
// If both the job runner and worker are being deleted, requeueing
// the worker item would leave a running job with no loop to consume it.
sqlx::query(
r#"
UPDATE agent_jobs
SET status = ?, updated_at = ?, completed_at = ?, last_error = ?
WHERE status IN (?, ?)
AND id IN (
SELECT item.job_id
FROM agent_job_items AS item
JOIN thread_spawn_edges AS edge ON edge.child_thread_id = item.assigned_thread_id
WHERE item.status = ? AND item.assigned_thread_id = ? AND edge.parent_thread_id = ?
)
"#,
)
.bind(AgentJobStatus::Cancelled.as_str())
.bind(now)
.bind(now)
.bind("agent job runner thread was deleted")
.bind(AgentJobStatus::Pending.as_str())
.bind(AgentJobStatus::Running.as_str())
.bind(AgentJobItemStatus::Running.as_str())
.bind(thread_id_string)
.bind(parent_thread_id_string)
.execute(&mut *tx)
.await?;
}
sqlx::query("DELETE FROM thread_dynamic_tools WHERE thread_id = ?")
.bind(thread_id_string)
.execute(&mut *tx)
.await?;
sqlx::query(
r#"
UPDATE agent_job_items
SET
status = ?,
assigned_thread_id = NULL,
updated_at = ?,
last_error = ?
WHERE assigned_thread_id = ? AND status = ?
"#,
)
.bind(AgentJobItemStatus::Pending.as_str())
.bind(now)
.bind("assigned thread was deleted")
.bind(thread_id_string)
.bind(AgentJobItemStatus::Running.as_str())
.execute(&mut *tx)
.await?;
sqlx::query(
r#"
UPDATE agent_job_items
SET assigned_thread_id = NULL, updated_at = ?
WHERE assigned_thread_id = ?
"#,
)
.bind(now)
.bind(thread_id_string)
.execute(&mut *tx)
.await?;
}
for thread_id_string in &thread_id_strings {
sqlx::query(
@@ -1449,7 +1391,6 @@ mod tests {
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::ThreadHistoryMode;
use pretty_assertions::assert_eq;
use serde_json::json;
use std::path::PathBuf;
#[tokio::test]
@@ -1536,36 +1477,6 @@ mod tests {
.bind("{}")
.execute(runtime.pool.as_ref())
.await?;
runtime
.create_agent_job(
&AgentJobCreateParams {
id: "job-1".to_string(),
name: "test-job".to_string(),
instruction: "Return a result".to_string(),
auto_export: true,
max_runtime_seconds: None,
output_schema_json: None,
input_headers: vec!["path".to_string()],
input_csv_path: "/tmp/in.csv".to_string(),
output_csv_path: "/tmp/out.csv".to_string(),
},
&[AgentJobItemCreateParams {
item_id: "item-1".to_string(),
row_index: 0,
source_id: None,
row_json: json!({"path": "file-1"}),
}],
)
.await?;
runtime.mark_agent_job_running("job-1").await?;
runtime
.mark_agent_job_item_running_with_thread(
"job-1",
"item-1",
&child_thread_id.to_string(),
)
.await?;
let rows = runtime
.delete_threads_strict(&[thread_id, child_thread_id])
.await?;
@@ -1579,21 +1490,6 @@ mod tests {
.await?;
assert_eq!(dynamic_tool_count, 0);
assert_thread_cleanup_state(&runtime, thread_id).await?;
let job_item = runtime
.get_agent_job_item("job-1", "item-1")
.await?
.expect("job item should exist");
assert_eq!(job_item.status, AgentJobItemStatus::Pending);
assert_eq!(job_item.assigned_thread_id, None);
assert_eq!(
job_item.last_error,
Some("assigned thread was deleted".to_string())
);
let job = runtime
.get_agent_job("job-1")
.await?
.expect("job should exist");
assert_eq!(job.status, AgentJobStatus::Cancelled);
let missing_thread_id = ThreadId::from_string("00000000-0000-0000-0000-000000000403")?;
let missing_child_thread_id =
@@ -245,7 +245,6 @@ fn new_config(model: Option<String>, arg0_paths: Arg0DispatchPaths) -> anyhow::R
agent_max_threads: Some(6),
agent_default_subagent_model: None,
agent_default_subagent_reasoning_effort: None,
agent_job_max_runtime_seconds: None,
agent_interrupt_message_enabled: false,
agent_max_depth: 1,
agent_roles: BTreeMap::new(),
-6
View File
@@ -80,11 +80,6 @@ fn render_agents_config_lines(config: &Config) -> Vec<Line<'static>> {
format_optional(config.agent_default_subagent_reasoning_effort.as_ref())
)
.into(),
format!(
" - job_max_runtime_seconds = {}",
format_optional(config.agent_job_max_runtime_seconds)
)
.into(),
format!(
" - interrupt_message = {}",
config.agent_interrupt_message_enabled
@@ -651,7 +646,6 @@ max_concurrent_threads_per_session = 7
max_depth = -2
default_subagent_model = "gpt-5.6-terra"
default_subagent_reasoning_effort = "high"
job_max_runtime_seconds = 900
interrupt_message = false
"#,
)
@@ -10,5 +10,4 @@ expression: render_to_text(&render_agents_config_lines(&config))
- max_depth = -2 (V1 only; ignored by V2)
- default_subagent_model = gpt-5.6-terra
- default_subagent_reasoning_effort = high
- job_max_runtime_seconds = 900
- interrupt_message = false