core/core-agent-ide
8
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Agent jobs (spawn_agents_on_csv) + progress UI (#10935)

Browse source 
## Summary
- Add agent job support: spawn a batch of sub-agents from CSV, auto-run,
auto-export, and store results in SQLite.
- Simplify workflow: remove run/resume/get-status/export tools; spawn is
deterministic and completes in one call.
- Improve exec UX: stable, single-line progress bar with ETA; suppress
sub-agent chatter in exec.

## Why
Enables map-reduce style workflows over arbitrarily large repos using
the existing Codex orchestrator. This addresses review feedback about
overly complex job controls and non-deterministic monitoring.

## Demo (progress bar)
```
./codex-rs/target/debug/codex exec \
  --enable collab \
  --enable sqlite \
  --full-auto \
  --progress-cursor \
  -c agents.max_threads=16 \
  -C /Users/daveaitel/code/codex \
  - <<'PROMPT'
Create /tmp/agent_job_progress_demo.csv with columns: path,area and 30 rows:
path = item-01..item-30, area = test.

Then call spawn_agents_on_csv with:
- csv_path: /tmp/agent_job_progress_demo.csv
- instruction: "Run `python - <<'PY'` to sleep a random 0.3–1.2s, then output JSON with keys: path, score (int). Set score = 1."
- output_csv_path: /tmp/agent_job_progress_demo_out.csv
PROMPT
```

## Review feedback addressed
- Auto-start jobs on spawn; removed run/resume/status/export tools.
- Auto-export on success.
- More descriptive tool spec + clearer prompts.
- Avoid deadlocks on spawn failure; pending/running handled safely.
- Progress bar no longer scrolls; stable single-line redraw.

## Tests
- `cd codex-rs && cargo test -p codex-exec`
- `cd codex-rs && cargo build -p codex-cli`
This commit is contained in:
daveaitel-openai 2026-02-24 16:00:19 -05:00 • committed by GitHub
parent bd192b54cd
commit dcab40123f
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
36 changed files with 3370 additions and 50 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
codex-rs/Cargo.lock generated
View file

@ -1726,6 +1726,7 @@ dependencies = [
"codex-windows-sandbox",
"core-foundation 0.9.4",
"core_test_support",
"csv",
"ctor 0.6.3",
"dirs",
"dunce",

1
codex-rs/Cargo.toml
View file

@ -160,6 +160,7 @@ clap = "4"
clap_complete = "4"
color-eyre = "0.6.3"
crossbeam-channel = "0.5.15"
csv = "1.3.1"
crossterm = "0.28.1"
ctor = "0.6.3"
derive_more = "2"

1
codex-rs/app-server/tests/common/mcp_process.rs
View file

@ -105,6 +105,7 @@ impl McpProcess {
cmd.stdin(Stdio::piped());
cmd.stdout(Stdio::piped());
cmd.stderr(Stdio::piped());
cmd.current_dir(codex_home);
cmd.env("CODEX_HOME", codex_home);
cmd.env("RUST_LOG", "debug");
cmd.env_remove(CODEX_INTERNAL_ORIGINATOR_OVERRIDE_ENV_VAR);

8
codex-rs/app-server/tests/suite/v2/experimental_feature_list.rs
View file

@ -9,6 +9,7 @@ use codex_app_server_protocol::ExperimentalFeatureListResponse;
use codex_app_server_protocol::ExperimentalFeatureStage;
use codex_app_server_protocol::JSONRPCResponse;
use codex_app_server_protocol::RequestId;
use codex_core::config::ConfigBuilder;
use codex_core::features::FEATURES;
use codex_core::features::Stage;
use pretty_assertions::assert_eq;
@ -20,6 +21,11 @@ const DEFAULT_TIMEOUT: Duration = Duration::from_secs(10);
#[tokio::test]
async fn experimental_feature_list_returns_feature_metadata_with_stage() -> Result<()> {
let codex_home = TempDir::new()?;
let config = ConfigBuilder::default()
.codex_home(codex_home.path().to_path_buf())
.fallback_cwd(Some(codex_home.path().to_path_buf()))
.build()
.await?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??;
@ -63,7 +69,7 @@ async fn experimental_feature_list_returns_feature_metadata_with_stage() -> Resu
display_name,
description,
announcement,
enabled: spec.default_enabled,
enabled: config.features.enabled(spec.id),
default_enabled: spec.default_enabled,
}
})

1
codex-rs/core/Cargo.toml
View file

@ -51,6 +51,7 @@ codex-utils-readiness = { workspace = true }
codex-secrets = { workspace = true }
codex-utils-string = { workspace = true }
codex-windows-sandbox = { package = "codex-windows-sandbox", path = "../windows-sandbox-rs" }
csv = { workspace = true }
dirs = { workspace = true }
dunce = { workspace = true }
encoding_rs = { workspace = true }

20
codex-rs/core/config.schema.json
View file

@ -29,12 +29,24 @@
"$ref": "#/definitions/AgentRoleToml"
},
"properties": {
"job_max_runtime_seconds": {
"description": "Default maximum runtime in seconds for agent job workers.",
"format": "uint64",
"minimum": 1.0,
"type": "integer"
},
"max_depth": {
"description": "Maximum nesting depth allowed for spawned agent threads. Root sessions start at depth 0.",
"format": "int32",
"minimum": 1.0,
"type": "integer"
},
"max_spawn_depth": {
"description": "Maximum depth for thread-spawned subagents.",
"format": "uint",
"minimum": 1.0,
"type": "integer"
},
"max_threads": {
"description": "Maximum number of agent threads that can be open concurrently. When unset, no limit is enforced.",
"format": "uint",
@ -2040,6 +2052,14 @@
],
"description": "User-level skill config entries keyed by SKILL.md path."
},
"sqlite_home": {
"allOf": [
{
"$ref": "#/definitions/AbsolutePathBuf"
}
],
"description": "Directory where Codex stores the SQLite state DB. Defaults to `$CODEX_SQLITE_HOME` when set. Otherwise uses a temp dir under WorkspaceWrite sandboxing and `$CODEX_HOME` for other modes."
},
"suppress_unstable_features_warning": {
"description": "Suppress warnings about unstable (under development) features.",
"type": "boolean"

6
codex-rs/core/src/agent/guards.rs
View file

@ -1,3 +1,4 @@
use crate::config::DEFAULT_AGENT_MAX_SPAWN_DEPTH;
use crate::error::CodexErr;
use crate::error::Result;
use codex_protocol::ThreadId;
@ -30,7 +31,6 @@ struct ActiveAgents {
used_agent_nicknames: HashSet<String>,
nickname_reset_count: usize,
}
fn session_depth(session_source: &SessionSource) -> i32 {
match session_source {
SessionSource::SubAgent(SubAgentSource::ThreadSpawn { depth, .. }) => *depth,
@ -43,6 +43,10 @@ pub(crate) fn next_thread_spawn_depth(session_source: &SessionSource) -> i32 {
session_depth(session_source).saturating_add(1)
}
pub(crate) fn max_thread_spawn_depth(max_depth: Option<usize>) -> i32 {
let max_depth = max_depth.or(DEFAULT_AGENT_MAX_SPAWN_DEPTH).unwrap_or(1);
i32::try_from(max_depth).unwrap_or(i32::MAX)
}
pub(crate) fn exceeds_thread_spawn_depth_limit(depth: i32, max_depth: i32) -> bool {
depth > max_depth
}

1
codex-rs/core/src/agent/mod.rs
View file

@ -6,5 +6,6 @@ pub(crate) mod status;
pub(crate) use codex_protocol::protocol::AgentStatus;
pub(crate) use control::AgentControl;
pub(crate) use guards::exceeds_thread_spawn_depth_limit;
pub(crate) use guards::max_thread_spawn_depth;
pub(crate) use guards::next_thread_spawn_depth;
pub(crate) use status::agent_status_from_event;

6
codex-rs/core/src/codex.rs
View file

@ -640,6 +640,7 @@ impl TurnContext {
model_info: &model_info,
features: &features,
web_search_mode: self.tools_config.web_search_mode,
session_source: self.session_source.clone(),
})
.with_allow_login_shell(self.tools_config.allow_login_shell)
.with_agent_roles(config.agent_roles.clone());
@ -975,6 +976,7 @@ impl Session {
model_info: &model_info,
features: &per_turn_config.features,
web_search_mode: Some(per_turn_config.web_search_mode.value()),
session_source: session_source.clone(),
})
.with_allow_login_shell(per_turn_config.permissions.allow_login_shell)
.with_agent_roles(per_turn_config.agent_roles.clone());
@ -4592,6 +4594,7 @@ async fn spawn_review_thread(
model_info: &review_model_info,
features: &review_features,
web_search_mode: Some(review_web_search_mode),
session_source: parent_turn_context.session_source.clone(),
})
.with_allow_login_shell(config.permissions.allow_login_shell)
.with_agent_roles(config.agent_roles.clone());
@ -9267,6 +9270,7 @@ mod tests {
})
.to_string(),
},
source: ToolCallSource::Direct,
})
.await;
@ -9306,6 +9310,7 @@ mod tests {
})
.to_string(),
},
source: ToolCallSource::Direct,
})
.await;
@ -9365,6 +9370,7 @@ mod tests {
})
.to_string(),
},
source: ToolCallSource::Direct,
})
.await;

107
codex-rs/core/src/config/mod.rs
View file

@ -115,8 +115,35 @@ pub use codex_git::GhostSnapshotConfig;
/// the context window.
pub(crate) const PROJECT_DOC_MAX_BYTES: usize = 32 * 1024; // 32 KiB
pub(crate) const DEFAULT_AGENT_MAX_THREADS: Option<usize> = Some(6);
pub(crate) const DEFAULT_AGENT_MAX_SPAWN_DEPTH: Option<usize> = Some(2);
pub(crate) const DEFAULT_AGENT_MAX_DEPTH: i32 = 1;
pub(crate) const DEFAULT_AGENT_JOB_MAX_RUNTIME_SECONDS: Option<u64> = None;
pub const CONFIG_TOML_FILE: &str = "config.toml";
fn default_sqlite_home(sandbox_policy: &SandboxPolicy, codex_home: &Path) -> PathBuf {
if matches!(sandbox_policy, SandboxPolicy::WorkspaceWrite { .. }) {
let mut path = std::env::temp_dir();
path.push("codex-sqlite");
path
} else {
codex_home.to_path_buf()
}
}
fn resolve_sqlite_home_env(resolved_cwd: &Path) -> Option<PathBuf> {
let raw = std::env::var(codex_state::SQLITE_HOME_ENV).ok()?;
let trimmed = raw.trim();
if trimmed.is_empty() {
return None;
}
let path = PathBuf::from(trimmed);
if path.is_absolute() {
Some(path)
} else {
Some(resolved_cwd.join(path))
}
}
#[cfg(test)]
pub(crate) fn test_config() -> Config {
let codex_home = tempdir().expect("create temp dir");
@ -330,6 +357,10 @@ pub struct Config {
/// Maximum number of agent threads that can be open concurrently.
pub agent_max_threads: Option<usize>,
/// Maximum depth for thread-spawned subagents.
pub agent_max_spawn_depth: Option<usize>,
/// Maximum runtime in seconds for agent job workers before they are failed.
pub agent_job_max_runtime_seconds: Option<u64>,
/// Maximum nesting depth allowed for spawned agent threads.
pub agent_max_depth: i32,
@ -344,6 +375,9 @@ pub struct Config {
/// overridden by the `CODEX_HOME` environment variable).
pub codex_home: PathBuf,
/// Directory where Codex stores the SQLite state DB.
pub sqlite_home: PathBuf,
/// Directory where Codex writes log files (defaults to `$CODEX_HOME/log`).
pub log_dir: PathBuf,
@ -1108,6 +1142,11 @@ pub struct ConfigToml {
#[serde(default)]
pub history: Option<History>,
/// Directory where Codex stores the SQLite state DB.
/// Defaults to `$CODEX_SQLITE_HOME` when set. Otherwise uses a temp dir
/// under WorkspaceWrite sandboxing and `$CODEX_HOME` for other modes.
pub sqlite_home: Option<AbsolutePathBuf>,
/// Directory where Codex writes log files, for example `codex-tui.log`.
/// Defaults to `$CODEX_HOME/log`.
pub log_dir: Option<AbsolutePathBuf>,
@ -1295,11 +1334,16 @@ pub struct AgentsToml {
/// When unset, no limit is enforced.
#[schemars(range(min = 1))]
pub max_threads: Option<usize>,
/// Maximum depth for thread-spawned subagents.
#[schemars(range(min = 1))]
pub max_spawn_depth: Option<usize>,
/// Maximum nesting depth allowed for spawned agent threads.
/// Root sessions start at depth 0.
#[schemars(range(min = 1))]
pub max_depth: Option<i32>,
/// Default maximum runtime in seconds for agent job workers.
#[schemars(range(min = 1))]
pub job_max_runtime_seconds: Option<u64>,
/// User-defined role declarations keyed by role name.
///
@ -1813,6 +1857,44 @@ impl Config {
})
.transpose()?
.unwrap_or_default();
let agent_max_spawn_depth = cfg
.agents
.as_ref()
.and_then(|agents| agents.max_spawn_depth)
.or(DEFAULT_AGENT_MAX_SPAWN_DEPTH);
if agent_max_spawn_depth == Some(0) {
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidInput,
"agents.max_spawn_depth must be at least 1",
));
}
if let Some(max_spawn_depth) = agent_max_spawn_depth
&& max_spawn_depth > i32::MAX as usize
{
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidInput,
"agents.max_spawn_depth must fit within a 32-bit signed integer",
));
}
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 background_terminal_max_timeout = cfg
.background_terminal_max_timeout
.unwrap_or(DEFAULT_MAX_BACKGROUND_TERMINAL_TIMEOUT_MS)
@ -1937,6 +2019,12 @@ impl Config {
p.push("log");
p
});
let sqlite_home = cfg
.sqlite_home
.as_ref()
.map(AbsolutePathBuf::to_path_buf)
.or_else(|| resolve_sqlite_home_env(&resolved_cwd))
.unwrap_or_else(|| default_sqlite_home(&sandbox_policy, &codex_home));
// Ensure that every field of ConfigRequirements is applied to the final
// Config.
@ -2053,7 +2141,10 @@ impl Config {
agent_max_depth,
agent_roles,
memories: cfg.memories.unwrap_or_default().into(),
agent_max_spawn_depth,
agent_job_max_runtime_seconds,
codex_home,
sqlite_home,
log_dir,
config_layer_stack,
history,
@ -4387,7 +4478,9 @@ model = "gpt-5.1-codex"
let cfg = ConfigToml {
agents: Some(AgentsToml {
max_threads: None,
max_spawn_depth: None,
max_depth: None,
job_max_runtime_seconds: None,
roles: BTreeMap::from([(
"researcher".to_string(),
AgentRoleToml {
@ -4661,7 +4754,10 @@ model_verbosity = "high"
agent_max_depth: DEFAULT_AGENT_MAX_DEPTH,
agent_roles: BTreeMap::new(),
memories: MemoriesConfig::default(),
agent_max_spawn_depth: DEFAULT_AGENT_MAX_SPAWN_DEPTH,
agent_job_max_runtime_seconds: DEFAULT_AGENT_JOB_MAX_RUNTIME_SECONDS,
codex_home: fixture.codex_home(),
sqlite_home: fixture.codex_home(),
log_dir: fixture.codex_home().join("log"),
config_layer_stack: Default::default(),
startup_warnings: Vec::new(),
@ -4784,7 +4880,10 @@ model_verbosity = "high"
agent_max_depth: DEFAULT_AGENT_MAX_DEPTH,
agent_roles: BTreeMap::new(),
memories: MemoriesConfig::default(),
agent_max_spawn_depth: DEFAULT_AGENT_MAX_SPAWN_DEPTH,
agent_job_max_runtime_seconds: DEFAULT_AGENT_JOB_MAX_RUNTIME_SECONDS,
codex_home: fixture.codex_home(),
sqlite_home: fixture.codex_home(),
log_dir: fixture.codex_home().join("log"),
config_layer_stack: Default::default(),
startup_warnings: Vec::new(),
@ -4905,7 +5004,10 @@ model_verbosity = "high"
agent_max_depth: DEFAULT_AGENT_MAX_DEPTH,
agent_roles: BTreeMap::new(),
memories: MemoriesConfig::default(),
agent_max_spawn_depth: DEFAULT_AGENT_MAX_SPAWN_DEPTH,
agent_job_max_runtime_seconds: DEFAULT_AGENT_JOB_MAX_RUNTIME_SECONDS,
codex_home: fixture.codex_home(),
sqlite_home: fixture.codex_home(),
log_dir: fixture.codex_home().join("log"),
config_layer_stack: Default::default(),
startup_warnings: Vec::new(),
@ -5012,7 +5114,10 @@ model_verbosity = "high"
agent_max_depth: DEFAULT_AGENT_MAX_DEPTH,
agent_roles: BTreeMap::new(),
memories: MemoriesConfig::default(),
agent_max_spawn_depth: DEFAULT_AGENT_MAX_SPAWN_DEPTH,
agent_job_max_runtime_seconds: DEFAULT_AGENT_JOB_MAX_RUNTIME_SECONDS,
codex_home: fixture.codex_home(),
sqlite_home: fixture.codex_home(),
log_dir: fixture.codex_home().join("log"),
config_layer_stack: Default::default(),
startup_warnings: Vec::new(),

52
codex-rs/core/src/shell_snapshot.rs
View file

@ -123,6 +123,13 @@ impl ShellSnapshot {
let path = codex_home
.join(SNAPSHOT_DIR)
.join(format!("{session_id}.{extension}"));
let nonce = SystemTime::now()
.duration_since(SystemTime::UNIX_EPOCH)
.map(|duration| duration.as_nanos())
.unwrap_or(0);
let temp_path = codex_home
.join(SNAPSHOT_DIR)
.join(format!("{session_id}.tmp-{nonce}"));
// Clean the (unlikely) leaked snapshot files.
let codex_home = codex_home.to_path_buf();
@ -134,31 +141,42 @@ impl ShellSnapshot {
});
// Make the new snapshot.
let path = match write_shell_snapshot(shell.shell_type.clone(), &path, session_cwd).await {
Ok(path) => {
tracing::info!("Shell snapshot successfully created: {}", path.display());
path
}
Err(err) => {
tracing::warn!(
"Failed to create shell snapshot for {}: {err:?}",
shell.name()
);
return Err("write_failed");
}
};
let temp_path =
match write_shell_snapshot(shell.shell_type.clone(), &temp_path, session_cwd).await {
Ok(path) => {
tracing::info!("Shell snapshot successfully created: {}", path.display());
path
}
Err(err) => {
tracing::warn!(
"Failed to create shell snapshot for {}: {err:?}",
shell.name()
);
return Err("write_failed");
}
};
let snapshot = Self {
path,
let temp_snapshot = Self {
path: temp_path.clone(),
cwd: session_cwd.to_path_buf(),
};
if let Err(err) = validate_snapshot(shell, &snapshot.path, session_cwd).await {
if let Err(err) = validate_snapshot(shell, &temp_snapshot.path, session_cwd).await {
tracing::error!("Shell snapshot validation failed: {err:?}");
remove_snapshot_file(&temp_snapshot.path).await;
return Err("validation_failed");
}
Ok(snapshot)
if let Err(err) = fs::rename(&temp_snapshot.path, &path).await {
tracing::warn!("Failed to finalize shell snapshot: {err:?}");
remove_snapshot_file(&temp_snapshot.path).await;
return Err("write_failed");
}
Ok(Self {
path,
cwd: session_cwd.to_path_buf(),
})
}
}

10
codex-rs/core/src/state_db.rs
View file

@ -37,7 +37,7 @@ pub(crate) async fn init_if_enabled(
return None;
}
let runtime = match codex_state::StateRuntime::init(
config.codex_home.clone(),
config.sqlite_home.clone(),
config.model_provider_id.clone(),
otel.cloned(),
)
@ -47,7 +47,7 @@ pub(crate) async fn init_if_enabled(
Err(err) => {
warn!(
"failed to initialize state runtime at {}: {err}",
config.codex_home.display()
config.sqlite_home.display()
);
if let Some(otel) = otel {
otel.counter("codex.db.init", 1, &[("status", "init_error")]);
@ -79,20 +79,20 @@ pub(crate) async fn init_if_enabled(
/// Get the DB if the feature is enabled and the DB exists.
pub async fn get_state_db(config: &Config, otel: Option<&OtelManager>) -> Option<StateDbHandle> {
let state_path = codex_state::state_db_path(config.codex_home.as_path());
let state_path = codex_state::state_db_path(config.sqlite_home.as_path());
if !config.features.enabled(Feature::Sqlite)
|| !tokio::fs::try_exists(&state_path).await.unwrap_or(false)
{
return None;
}
let runtime = codex_state::StateRuntime::init(
config.codex_home.clone(),
config.sqlite_home.clone(),
config.model_provider_id.clone(),
otel.cloned(),
)
.await
.ok()?;
require_backfill_complete(runtime, config.codex_home.as_path()).await
require_backfill_complete(runtime, config.sqlite_home.as_path()).await
}
/// Open the state runtime when the SQLite file exists, without feature gating.

7
codex-rs/core/src/tools/context.rs
View file

@ -16,6 +16,12 @@ use tokio::sync::Mutex;
pub type SharedTurnDiffTracker = Arc<Mutex<TurnDiffTracker>>;
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ToolCallSource {
Direct,
JsRepl,
}
#[derive(Clone)]
pub struct ToolInvocation {
pub session: Arc<Session>,
@ -24,6 +30,7 @@ pub struct ToolInvocation {
pub call_id: String,
pub tool_name: String,
pub payload: ToolPayload,
pub source: ToolCallSource,
}
#[derive(Clone, Debug)]

1227
codex-rs/core/src/tools/handlers/agent_jobs.rs Normal file
View file

File diff suppressed because it is too large Load diff

1
codex-rs/core/src/tools/handlers/apply_patch.rs
View file

@ -86,6 +86,7 @@ impl ToolHandler for ApplyPatchHandler {
call_id,
tool_name,
payload,
..
} = invocation;
let patch_input = match payload {

1
codex-rs/core/src/tools/handlers/mod.rs
View file

@ -1,3 +1,4 @@
pub(crate) mod agent_jobs;
pub mod apply_patch;
mod dynamic;
mod grep_files;

21
codex-rs/core/src/tools/handlers/multi_agents.rs
View file

@ -1,5 +1,6 @@
use crate::agent::AgentStatus;
use crate::agent::exceeds_thread_spawn_depth_limit;
use crate::agent::max_thread_spawn_depth;
use crate::codex::Session;
use crate::codex::TurnContext;
use crate::config::Config;
@ -97,6 +98,7 @@ mod spawn {
use crate::agent::role::apply_role_to_config;
use crate::agent::exceeds_thread_spawn_depth_limit;
use crate::agent::max_thread_spawn_depth;
use crate::agent::next_thread_spawn_depth;
use std::sync::Arc;
@ -129,7 +131,8 @@ mod spawn {
let prompt = input_preview(&input_items);
let session_source = turn.session_source.clone();
let child_depth = next_thread_spawn_depth(&session_source);
if exceeds_thread_spawn_depth_limit(child_depth, turn.config.agent_max_depth) {
let max_depth = max_thread_spawn_depth(turn.config.agent_max_spawn_depth);
if exceeds_thread_spawn_depth_limit(child_depth, max_depth) {
return Err(FunctionCallError::RespondToModel(
"Agent depth limit reached. Solve the task yourself.".to_string(),
));
@ -345,7 +348,8 @@ mod resume_agent {
.await
.unwrap_or((None, None));
let child_depth = next_thread_spawn_depth(&turn.session_source);
if exceeds_thread_spawn_depth_limit(child_depth, turn.config.agent_max_depth) {
let max_depth = max_thread_spawn_depth(turn.config.agent_max_spawn_depth);
if exceeds_thread_spawn_depth_limit(child_depth, max_depth) {
return Err(FunctionCallError::RespondToModel(
"Agent depth limit reached. Solve the task yourself.".to_string(),
));
@ -891,7 +895,7 @@ fn input_preview(items: &[UserInput]) -> String {
parts.join("\n")
}
fn build_agent_spawn_config(
pub(crate) fn build_agent_spawn_config(
base_instructions: &BaseInstructions,
turn: &TurnContext,
child_depth: i32,
@ -948,7 +952,8 @@ fn apply_spawn_agent_runtime_overrides(
fn apply_spawn_agent_overrides(config: &mut Config, child_depth: i32) {
config.permissions.approval_policy = Constrained::allow_only(AskForApproval::Never);
if exceeds_thread_spawn_depth_limit(child_depth + 1, config.agent_max_depth) {
let max_depth = max_thread_spawn_depth(config.agent_max_spawn_depth);
if exceeds_thread_spawn_depth_limit(child_depth + 1, max_depth) {
config.features.disable(Feature::Collab);
}
}
@ -959,6 +964,7 @@ mod tests {
use crate::AuthManager;
use crate::CodexAuth;
use crate::ThreadManager;
use crate::agent::max_thread_spawn_depth;
use crate::built_in_model_providers;
use crate::codex::make_session_and_context;
use crate::config::DEFAULT_AGENT_MAX_DEPTH;
@ -998,6 +1004,7 @@ mod tests {
call_id: "call-1".to_string(),
tool_name: tool_name.to_string(),
payload,
source: crate::tools::router::ToolCallSource::Direct,
}
}
@ -1259,9 +1266,10 @@ mod tests {
let manager = thread_manager();
session.services.agent_control = manager.agent_control();
let max_depth = max_thread_spawn_depth(turn.config.agent_max_spawn_depth);
turn.session_source = SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id: session.conversation_id,
depth: DEFAULT_AGENT_MAX_DEPTH,
depth: max_depth,
agent_nickname: None,
agent_role: None,
});
@ -1689,9 +1697,10 @@ mod tests {
let manager = thread_manager();
session.services.agent_control = manager.agent_control();
let max_depth = max_thread_spawn_depth(turn.config.agent_max_spawn_depth);
turn.session_source = SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id: session.conversation_id,
depth: DEFAULT_AGENT_MAX_DEPTH,
depth: max_depth,
agent_nickname: None,
agent_role: None,
});

2
codex-rs/core/src/tools/handlers/shell.rs
View file

@ -176,6 +176,7 @@ impl ToolHandler for ShellHandler {
call_id,
tool_name,
payload,
..
} = invocation;
match payload {
@ -261,6 +262,7 @@ impl ToolHandler for ShellCommandHandler {
call_id,
tool_name,
payload,
..
} = invocation;
let ToolPayload::Function { arguments } = payload else {

24
codex-rs/core/src/tools/handlers/view_image.rs
View file

@ -8,6 +8,7 @@ use tokio::fs;
use crate::function_tool::FunctionCallError;
use crate::protocol::EventMsg;
use crate::protocol::ViewImageToolCallEvent;
use crate::tools::context::ToolCallSource;
use crate::tools::context::ToolInvocation;
use crate::tools::context::ToolOutput;
use crate::tools::context::ToolPayload;
@ -15,6 +16,7 @@ use crate::tools::handlers::parse_arguments;
use crate::tools::registry::ToolHandler;
use crate::tools::registry::ToolKind;
use codex_protocol::models::ContentItem;
use codex_protocol::models::ResponseInputItem;
use codex_protocol::models::local_image_content_items_with_label_number;
pub struct ViewImageHandler;
@ -50,6 +52,7 @@ impl ToolHandler for ViewImageHandler {
turn,
payload,
call_id,
source,
..
} = invocation;
@ -81,7 +84,26 @@ impl ToolHandler for ViewImageHandler {
}
let event_path = abs_path.clone();
let content = local_image_content_items_with_label_number(&abs_path, None)
let content = local_image_content_items_with_label_number(&abs_path, None);
if source == ToolCallSource::JsRepl
&& content
.iter()
.any(|item| matches!(item, ContentItem::InputImage { .. }))
{
let input_item = ResponseInputItem::Message {
role: "user".to_string(),
content: content.clone(),
};
if session
.inject_response_items(vec![input_item])
.await
.is_err()
{
tracing::warn!("view_image could not find an active turn to attach image input");
}
}
let content = content
.into_iter()
.map(|item| match item {
ContentItem::InputText { text } => {

9
codex-rs/core/src/tools/router.rs
View file

@ -22,6 +22,8 @@ use std::collections::HashMap;
use std::sync::Arc;
use tracing::instrument;
pub use crate::tools::context::ToolCallSource;
#[derive(Clone, Debug)]
pub struct ToolCall {
pub tool_name: String,
@ -29,12 +31,6 @@ pub struct ToolCall {
pub payload: ToolPayload,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ToolCallSource {
Direct,
JsRepl,
}
pub struct ToolRouter {
registry: ToolRegistry,
specs: Vec<ConfiguredToolSpec>,
@ -179,6 +175,7 @@ impl ToolRouter {
call_id,
tool_name,
payload,
source,
};
match self.registry.dispatch(invocation).await {

258
codex-rs/core/src/tools/spec.rs
View file

@ -9,6 +9,7 @@ use crate::mcp_connection_manager::ToolInfo;
use crate::tools::handlers::PLAN_TOOL;
use crate::tools::handlers::SEARCH_TOOL_BM25_DEFAULT_LIMIT;
use crate::tools::handlers::SEARCH_TOOL_BM25_TOOL_NAME;
use crate::tools::handlers::agent_jobs::BatchJobHandler;
use crate::tools::handlers::apply_patch::create_apply_patch_freeform_tool;
use crate::tools::handlers::apply_patch::create_apply_patch_json_tool;
use crate::tools::handlers::multi_agents::DEFAULT_WAIT_TIMEOUT_MS;
@ -22,6 +23,8 @@ use codex_protocol::models::VIEW_IMAGE_TOOL_NAME;
use codex_protocol::openai_models::ApplyPatchToolType;
use codex_protocol::openai_models::ConfigShellToolType;
use codex_protocol::openai_models::ModelInfo;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SubAgentSource;
use serde::Deserialize;
use serde::Serialize;
use serde_json::Value as JsonValue;
@ -53,12 +56,15 @@ pub(crate) struct ToolsConfig {
pub collab_tools: bool,
pub collaboration_modes_tools: bool,
pub experimental_supported_tools: Vec<String>,
pub agent_jobs_tools: bool,
pub agent_jobs_worker_tools: bool,
}
pub(crate) struct ToolsConfigParams<'a> {
pub(crate) model_info: &'a ModelInfo,
pub(crate) features: &'a Features,
pub(crate) web_search_mode: Option<WebSearchMode>,
pub(crate) session_source: SessionSource,
}
impl ToolsConfig {
@ -67,14 +73,16 @@ impl ToolsConfig {
model_info,
features,
web_search_mode,
session_source,
} = params;
let include_apply_patch_tool = features.enabled(Feature::ApplyPatchFreeform);
let include_js_repl = features.enabled(Feature::JsRepl);
let include_js_repl_tools_only =
include_js_repl && features.enabled(Feature::JsReplToolsOnly);
let include_collab_tools = features.enabled(Feature::Collab);
let include_collaboration_modes_tools = true;
let include_collaboration_modes_tools = features.enabled(Feature::CollaborationModes);
let include_search_tool = features.enabled(Feature::Apps);
let include_agent_jobs = include_collab_tools && features.enabled(Feature::Sqlite);
let request_permission_enabled = features.enabled(Feature::RequestPermissions);
let shell_command_backend =
if features.enabled(Feature::ShellTool) && features.enabled(Feature::ShellZshFork) {
@ -110,6 +118,13 @@ impl ToolsConfig {
}
};
let agent_jobs_worker_tools = include_agent_jobs
&& matches!(
session_source,
SessionSource::SubAgent(SubAgentSource::Other(label))
if label.starts_with("agent_job:")
);
Self {
shell_type,
shell_command_backend,
@ -124,6 +139,8 @@ impl ToolsConfig {
collab_tools: include_collab_tools,
collaboration_modes_tools: include_collaboration_modes_tools,
experimental_supported_tools: model_info.experimental_supported_tools.clone(),
agent_jobs_tools: include_agent_jobs,
agent_jobs_worker_tools,
}
}
@ -623,6 +640,131 @@ fn create_spawn_agent_tool(config: &ToolsConfig) -> ToolSpec {
})
}
fn create_spawn_agents_on_csv_tool() -> ToolSpec {
let mut properties = BTreeMap::new();
properties.insert(
"csv_path".to_string(),
JsonSchema::String {
description: Some("Path to the CSV file containing input rows.".to_string()),
},
);
properties.insert(
"instruction".to_string(),
JsonSchema::String {
description: Some(
"Instruction template to apply to each CSV row. Use {column_name} placeholders to inject values from the row."
.to_string(),
),
},
);
properties.insert(
"id_column".to_string(),
JsonSchema::String {
description: Some("Optional column name to use as stable item id.".to_string()),
},
);
properties.insert(
"output_csv_path".to_string(),
JsonSchema::String {
description: Some("Optional output CSV path for exported results.".to_string()),
},
);
properties.insert(
"max_concurrency".to_string(),
JsonSchema::Number {
description: Some(
"Maximum concurrent workers for this job. Defaults to 16 and is capped by config."
.to_string(),
),
},
);
properties.insert(
"max_workers".to_string(),
JsonSchema::Number {
description: Some(
"Alias for max_concurrency. Set to 1 to run sequentially.".to_string(),
),
},
);
properties.insert(
"max_runtime_seconds".to_string(),
JsonSchema::Number {
description: Some(
"Maximum runtime per worker before it is failed. Defaults to 1800 seconds."
.to_string(),
),
},
);
properties.insert(
"output_schema".to_string(),
JsonSchema::Object {
properties: BTreeMap::new(),
required: None,
additional_properties: None,
},
);
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,
parameters: JsonSchema::Object {
properties,
required: Some(vec!["csv_path".to_string(), "instruction".to_string()]),
additional_properties: Some(false.into()),
},
})
}
fn create_report_agent_job_result_tool() -> ToolSpec {
let mut properties = BTreeMap::new();
properties.insert(
"job_id".to_string(),
JsonSchema::String {
description: Some("Identifier of the job.".to_string()),
},
);
properties.insert(
"item_id".to_string(),
JsonSchema::String {
description: Some("Identifier of the job item.".to_string()),
},
);
properties.insert(
"result".to_string(),
JsonSchema::Object {
properties: BTreeMap::new(),
required: None,
additional_properties: None,
},
);
properties.insert(
"stop".to_string(),
JsonSchema::Boolean {
description: Some(
"Optional. When true, cancels the remaining job items after this result is recorded."
.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,
parameters: JsonSchema::Object {
properties,
required: Some(vec![
"job_id".to_string(),
"item_id".to_string(),
"result".to_string(),
]),
additional_properties: Some(false.into()),
},
})
}
fn create_send_input_tool() -> ToolSpec {
let properties = BTreeMap::from([
(
@ -1670,6 +1812,16 @@ pub(crate) fn build_specs(
builder.register_handler("close_agent", multi_agent_handler);
}
if config.agent_jobs_tools {
let agent_jobs_handler = Arc::new(BatchJobHandler);
builder.push_spec(create_spawn_agents_on_csv_tool());
builder.register_handler("spawn_agents_on_csv", agent_jobs_handler.clone());
if config.agent_jobs_worker_tools {
builder.push_spec(create_report_agent_job_result_tool());
builder.register_handler("report_agent_job_result", agent_jobs_handler);
}
}
if let Some(mcp_tools) = mcp_tools {
let mut entries: Vec<(String, rmcp::model::Tool)> = mcp_tools.into_iter().collect();
entries.sort_by(|a, b| a.0.cmp(&b.0));
@ -1870,6 +2022,7 @@ mod tests {
model_info: &model_info,
features: &features,
web_search_mode: Some(WebSearchMode::Live),
session_source: SessionSource::Cli,
});
let (tools, _) = build_specs(&config, None, None, &[]).build();
@ -1928,10 +2081,42 @@ mod tests {
let mut features = Features::with_defaults();
features.enable(Feature::Collab);
features.enable(Feature::CollaborationModes);
features.enable(Feature::Sqlite);
let tools_config = ToolsConfig::new(&ToolsConfigParams {
model_info: &model_info,
features: &features,
web_search_mode: Some(WebSearchMode::Cached),
session_source: SessionSource::Cli,
});
let (tools, _) = build_specs(&tools_config, None, None, &[]).build();
assert_contains_tool_names(
&tools,
&[
"spawn_agent",
"send_input",
"wait",
"close_agent",
"spawn_agents_on_csv",
],
);
}
#[test]
fn test_build_specs_agent_job_worker_tools_enabled() {
let config = test_config();
let model_info =
ModelsManager::construct_model_info_offline_for_tests("gpt-5-codex", &config);
let mut features = Features::with_defaults();
features.enable(Feature::Collab);
features.enable(Feature::CollaborationModes);
features.enable(Feature::Sqlite);
let tools_config = ToolsConfig::new(&ToolsConfigParams {
model_info: &model_info,
features: &features,
web_search_mode: Some(WebSearchMode::Cached),
session_source: SessionSource::SubAgent(SubAgentSource::Other(
"agent_job:test".to_string(),
)),
});
let (tools, _) = build_specs(&tools_config, None, None, &[]).build();
assert_contains_tool_names(
@ -1942,10 +2127,62 @@ mod tests {
"resume_agent",
"wait",
"close_agent",
"spawn_agents_on_csv",
"report_agent_job_result",
],
);
}
#[test]
fn request_user_input_requires_collaboration_modes_feature() {
let config = test_config();
let model_info =
ModelsManager::construct_model_info_offline_for_tests("gpt-5-codex", &config);
let mut features = Features::with_defaults();
features.disable(Feature::CollaborationModes);
let tools_config = ToolsConfig::new(&ToolsConfigParams {
model_info: &model_info,
features: &features,
web_search_mode: Some(WebSearchMode::Cached),
session_source: SessionSource::Cli,
});
let (tools, _) = build_specs(&tools_config, None, None, &[]).build();
assert!(
!tools.iter().any(|t| t.spec.name() == "request_user_input"),
"request_user_input should be disabled when collaboration_modes feature is off"
);
features.enable(Feature::CollaborationModes);
let tools_config = ToolsConfig::new(&ToolsConfigParams {
model_info: &model_info,
features: &features,
web_search_mode: Some(WebSearchMode::Cached),
session_source: SessionSource::Cli,
});
let (tools, _) = build_specs(&tools_config, None, None, &[]).build();
assert_contains_tool_names(&tools, &["request_user_input"]);
}
#[test]
fn get_memory_requires_feature_flag() {
let config = test_config();
let model_info =
ModelsManager::construct_model_info_offline_for_tests("gpt-5-codex", &config);
let mut features = Features::with_defaults();
features.disable(Feature::MemoryTool);
let tools_config = ToolsConfig::new(&ToolsConfigParams {
model_info: &model_info,
features: &features,
web_search_mode: Some(WebSearchMode::Cached),
session_source: SessionSource::Cli,
});
let (tools, _) = build_specs(&tools_config, None, None, &[]).build();
assert!(
!tools.iter().any(|t| t.spec.name() == "get_memory"),
"get_memory should be disabled when memory_tool feature is off"
);
}
#[test]
fn js_repl_requires_feature_flag() {
let config = test_config();
@ -1957,6 +2194,7 @@ mod tests {
model_info: &model_info,
features: &features,
web_search_mode: Some(WebSearchMode::Cached),
session_source: SessionSource::Cli,
});
let (tools, _) = build_specs(&tools_config, None, None, &[]).build();
@ -1982,6 +2220,7 @@ mod tests {
model_info: &model_info,
features: &features,
web_search_mode: Some(WebSearchMode::Cached),
session_source: SessionSource::Cli,
});
let (tools, _) = build_specs(&tools_config, None, None, &[]).build();
assert_contains_tool_names(&tools, &["js_repl", "js_repl_reset"]);
@ -2013,6 +2252,7 @@ mod tests {
model_info: &model_info,
features,
web_search_mode,
session_source: SessionSource::Cli,
});
let (tools, _) = build_specs(&tools_config, None, None, &[]).build();
let tool_names = tools.iter().map(|t| t.spec.name()).collect::<Vec<_>>();
@ -2046,6 +2286,7 @@ mod tests {
model_info: &model_info,
features: &features,
web_search_mode: Some(WebSearchMode::Cached),
session_source: SessionSource::Cli,
});
let (tools, _) = build_specs(&tools_config, None, None, &[]).build();
@ -2069,6 +2310,7 @@ mod tests {
model_info: &model_info,
features: &features,
web_search_mode: Some(WebSearchMode::Live),
session_source: SessionSource::Cli,
});
let (tools, _) = build_specs(&tools_config, None, None, &[]).build();
@ -2092,6 +2334,7 @@ mod tests {
model_info: &model_info,
features: &features,
web_search_mode: Some(WebSearchMode::Cached),
session_source: SessionSource::Cli,
});
let (tools, _) = build_specs(&tools_config, None, None, &[]).build();
@ -2115,6 +2358,7 @@ mod tests {
model_info: &model_info,
features: &features,
web_search_mode: Some(WebSearchMode::Cached),
session_source: SessionSource::Cli,
});
let (tools, _) = build_specs(&tools_config, Some(HashMap::new()), None, &[]).build();
@ -2314,6 +2558,7 @@ mod tests {
model_info: &model_info,
features: &features,
web_search_mode: Some(WebSearchMode::Live),
session_source: SessionSource::Cli,
});
let (tools, _) = build_specs(&tools_config, Some(HashMap::new()), None, &[]).build();
@ -2337,6 +2582,7 @@ mod tests {
model_info: &model_info,
features: &features,
web_search_mode: Some(WebSearchMode::Live),
session_source: SessionSource::Cli,
});
assert_eq!(tools_config.shell_type, ConfigShellToolType::ShellCommand);
@ -2358,6 +2604,7 @@ mod tests {
model_info: &model_info,
features: &features,
web_search_mode: Some(WebSearchMode::Cached),
session_source: SessionSource::Cli,
});
let (tools, _) = build_specs(&tools_config, None, None, &[]).build();
@ -2382,6 +2629,7 @@ mod tests {
model_info: &model_info,
features: &features,
web_search_mode: Some(WebSearchMode::Cached),
session_source: SessionSource::Cli,
});
let (tools, _) = build_specs(&tools_config, None, None, &[]).build();
@ -2413,6 +2661,7 @@ mod tests {
model_info: &model_info,
features: &features,
web_search_mode: Some(WebSearchMode::Live),
session_source: SessionSource::Cli,
});
let (tools, _) = build_specs(
&tools_config,
@ -2499,6 +2748,7 @@ mod tests {
model_info: &model_info,
features: &features,
web_search_mode: Some(WebSearchMode::Cached),
session_source: SessionSource::Cli,
});
// Intentionally construct a map with keys that would sort alphabetically.
@ -2544,6 +2794,7 @@ mod tests {
model_info: &model_info,
features: &features,
web_search_mode: Some(WebSearchMode::Cached),
session_source: SessionSource::Cli,
});
let (tools, _) = build_specs(
@ -2611,6 +2862,7 @@ mod tests {
model_info: &model_info,
features: &features,
web_search_mode: Some(WebSearchMode::Cached),
session_source: SessionSource::Cli,
});
let (tools, _) = build_specs(
@ -2665,6 +2917,7 @@ mod tests {
model_info: &model_info,
features: &features,
web_search_mode: Some(WebSearchMode::Cached),
session_source: SessionSource::Cli,
});
let (tools, _) = build_specs(
@ -2716,6 +2969,7 @@ mod tests {
model_info: &model_info,
features: &features,
web_search_mode: Some(WebSearchMode::Cached),
session_source: SessionSource::Cli,
});
let (tools, _) = build_specs(
@ -2769,6 +3023,7 @@ mod tests {
model_info: &model_info,
features: &features,
web_search_mode: Some(WebSearchMode::Cached),
session_source: SessionSource::Cli,
});
let (tools, _) = build_specs(
@ -2901,6 +3156,7 @@ Examples of valid command strings:
model_info: &model_info,
features: &features,
web_search_mode: Some(WebSearchMode::Cached),
session_source: SessionSource::Cli,
});
let (tools, _) = build_specs(
&tools_config,

8
codex-rs/core/tests/common/test_codex.rs
View file

@ -215,6 +215,14 @@ impl TestCodexBuilder {
}
if let Ok(path) = codex_utils_cargo_bin::cargo_bin("codex") {
config.codex_linux_sandbox_exe = Some(path);
} else if let Ok(exe) = std::env::current_exe()
&& let Some(path) = exe
.parent()
.and_then(|parent| parent.parent())
.map(|parent| parent.join("codex"))
&& path.is_file()
{
config.codex_linux_sandbox_exe = Some(path);
}
let mut mutators = vec![];

424
codex-rs/core/tests/suite/agent_jobs.rs Normal file
View file

@ -0,0 +1,424 @@
use anyhow::Result;
use codex_core::features::Feature;
use core_test_support::responses::ev_completed;
use core_test_support::responses::ev_function_call;
use core_test_support::responses::ev_response_created;
use core_test_support::responses::sse;
use core_test_support::responses::sse_response;
use core_test_support::responses::start_mock_server;
use core_test_support::test_codex::test_codex;
use regex_lite::Regex;
use serde_json::Value;
use serde_json::json;
use std::fs;
use std::sync::Arc;
use std::sync::atomic::AtomicBool;
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering;
use wiremock::Mock;
use wiremock::Respond;
use wiremock::ResponseTemplate;
use wiremock::matchers::method;
use wiremock::matchers::path_regex;
struct AgentJobsResponder {
spawn_args_json: String,
seen_main: AtomicBool,
call_counter: AtomicUsize,
}
impl AgentJobsResponder {
fn new(spawn_args_json: String) -> Self {
Self {
spawn_args_json,
seen_main: AtomicBool::new(false),
call_counter: AtomicUsize::new(0),
}
}
}
struct StopAfterFirstResponder {
spawn_args_json: String,
seen_main: AtomicBool,
worker_calls: Arc<AtomicUsize>,
}
impl StopAfterFirstResponder {
fn new(spawn_args_json: String, worker_calls: Arc<AtomicUsize>) -> Self {
Self {
spawn_args_json,
seen_main: AtomicBool::new(false),
worker_calls,
}
}
}
impl Respond for StopAfterFirstResponder {
fn respond(&self, request: &wiremock::Request) -> ResponseTemplate {
let body_bytes = decode_body_bytes(request);
let body: Value = serde_json::from_slice(&body_bytes).unwrap_or(Value::Null);
if has_function_call_output(&body) {
return sse_response(sse(vec![
ev_response_created("resp-tool"),
ev_completed("resp-tool"),
]));
}
if let Some((job_id, item_id)) = extract_job_and_item(&body) {
let call_index = self.worker_calls.fetch_add(1, Ordering::SeqCst);
let call_id = format!("call-worker-{call_index}");
let stop = call_index == 0;
let args = json!({
"job_id": job_id,
"item_id": item_id,
"result": { "item_id": item_id },
"stop": stop,
});
let args_json = serde_json::to_string(&args).unwrap_or_else(|err| {
panic!("worker args serialize: {err}");
});
return sse_response(sse(vec![
ev_response_created("resp-worker"),
ev_function_call(&call_id, "report_agent_job_result", &args_json),
ev_completed("resp-worker"),
]));
}
if !self.seen_main.swap(true, Ordering::SeqCst) {
return sse_response(sse(vec![
ev_response_created("resp-main"),
ev_function_call("call-spawn", "spawn_agents_on_csv", &self.spawn_args_json),
ev_completed("resp-main"),
]));
}
sse_response(sse(vec![
ev_response_created("resp-default"),
ev_completed("resp-default"),
]))
}
}
impl Respond for AgentJobsResponder {
fn respond(&self, request: &wiremock::Request) -> ResponseTemplate {
let body_bytes = decode_body_bytes(request);
let body: Value = serde_json::from_slice(&body_bytes).unwrap_or(Value::Null);
if has_function_call_output(&body) {
return sse_response(sse(vec![
ev_response_created("resp-tool"),
ev_completed("resp-tool"),
]));
}
if let Some((job_id, item_id)) = extract_job_and_item(&body) {
let call_id = format!(
"call-worker-{}",
self.call_counter.fetch_add(1, Ordering::SeqCst)
);
let args = json!({
"job_id": job_id,
"item_id": item_id,
"result": { "item_id": item_id }
});
let args_json = serde_json::to_string(&args).unwrap_or_else(|err| {
panic!("worker args serialize: {err}");
});
return sse_response(sse(vec![
ev_response_created("resp-worker"),
ev_function_call(&call_id, "report_agent_job_result", &args_json),
ev_completed("resp-worker"),
]));
}
if !self.seen_main.swap(true, Ordering::SeqCst) {
return sse_response(sse(vec![
ev_response_created("resp-main"),
ev_function_call("call-spawn", "spawn_agents_on_csv", &self.spawn_args_json),
ev_completed("resp-main"),
]));
}
sse_response(sse(vec![
ev_response_created("resp-default"),
ev_completed("resp-default"),
]))
}
}
fn decode_body_bytes(request: &wiremock::Request) -> Vec<u8> {
let Some(encoding) = request
.headers
.get("content-encoding")
.and_then(|value| value.to_str().ok())
else {
return request.body.clone();
};
if encoding
.split(',')
.any(|entry| entry.trim().eq_ignore_ascii_case("zstd"))
{
zstd::stream::decode_all(std::io::Cursor::new(&request.body))
.unwrap_or_else(|_| request.body.clone())
} else {
request.body.clone()
}
}
fn has_function_call_output(body: &Value) -> bool {
body.get("input")
.and_then(Value::as_array)
.is_some_and(|items| {
items.iter().any(|item| {
item.get("type").and_then(Value::as_str) == Some("function_call_output")
})
})
}
fn extract_job_and_item(body: &Value) -> Option<(String, String)> {
let texts = message_input_texts(body);
let mut combined = texts.join("\n");
if let Some(instructions) = body.get("instructions").and_then(Value::as_str) {
combined.push('\n');
combined.push_str(instructions);
}
if !combined.contains("You are processing one item for a generic agent job.") {
return None;
}
let job_id = Regex::new(r"Job ID:\s*([^\n]+)")
.ok()?
.captures(&combined)
.and_then(|caps| caps.get(1))
.map(|m| m.as_str().trim().to_string())?;
let item_id = Regex::new(r"Item ID:\s*([^\n]+)")
.ok()?
.captures(&combined)
.and_then(|caps| caps.get(1))
.map(|m| m.as_str().trim().to_string())?;
Some((job_id, item_id))
}
fn message_input_texts(body: &Value) -> Vec<String> {
let Some(items) = body.get("input").and_then(Value::as_array) else {
return Vec::new();
};
items
.iter()
.filter(|item| item.get("type").and_then(Value::as_str) == Some("message"))
.filter_map(|item| item.get("content").and_then(Value::as_array))
.flatten()
.filter(|span| span.get("type").and_then(Value::as_str) == Some("input_text"))
.filter_map(|span| span.get("text").and_then(Value::as_str))
.map(str::to_string)
.collect()
}
fn parse_simple_csv_line(line: &str) -> Vec<String> {
line.split(',').map(str::to_string).collect()
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn report_agent_job_result_rejects_wrong_thread() -> Result<()> {
let server = start_mock_server().await;
let mut builder = test_codex().with_config(|config| {
config.features.enable(Feature::Collab);
config.features.enable(Feature::Sqlite);
});
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(), 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::Collab);
config.features.enable(Feature::Sqlite);
});
let test = builder.build(&server).await?;
let input_path = test.cwd_path().join("agent_jobs_input.csv");
let output_path = test.cwd_path().join("agent_jobs_output.csv");
fs::write(&input_path, "path,area\nfile-1,test\nfile-2,test\n")?;
let args = json!({
"csv_path": input_path.display().to_string(),
"instruction": "Return {path}",
"output_csv_path": output_path.display().to_string(),
});
let args_json = serde_json::to_string(&args)?;
let responder = AgentJobsResponder::new(args_json);
Mock::given(method("POST"))
.and(path_regex(".*/responses$"))
.respond_with(responder)
.mount(&server)
.await;
test.submit_turn("run batch job").await?;
let output = fs::read_to_string(&output_path)?;
assert!(output.contains("result_json"));
assert!(output.contains("item_id"));
assert!(output.contains("\"item_id\""));
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn spawn_agents_on_csv_dedupes_item_ids() -> Result<()> {
let server = start_mock_server().await;
let mut builder = test_codex().with_config(|config| {
config.features.enable(Feature::Collab);
config.features.enable(Feature::Sqlite);
});
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::Collab);
config.features.enable(Feature::Sqlite);
});
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
codex-rs/core/tests/suite/mod.rs
View file

@ -56,6 +56,7 @@ pub static CODEX_ALIASES_TEMP_DIR: TestCodexAliasesGuard = unsafe {
#[cfg(not(target_os = "windows"))]
mod abort_tasks;
mod agent_jobs;
mod agent_websocket;
mod apply_patch_cli;
#[cfg(not(target_os = "windows"))]

14
codex-rs/core/tests/suite/shell_snapshot.rs
View file

@ -52,10 +52,16 @@ async fn wait_for_snapshot(codex_home: &Path) -> Result<PathBuf> {
let snapshot_dir = codex_home.join("shell_snapshots");
let deadline = Instant::now() + Duration::from_secs(5);
loop {
if let Ok(mut entries) = fs::read_dir(&snapshot_dir).await
&& let Some(entry) = entries.next_entry().await?
{
return Ok(entry.path());
if let Ok(mut entries) = fs::read_dir(&snapshot_dir).await {
while let Some(entry) = entries.next_entry().await? {
let path = entry.path();
let Some(extension) = path.extension().and_then(|ext| ext.to_str()) else {
continue;
};
if extension == "sh" || extension == "ps1" {
return Ok(path);
}
}
}
if Instant::now() >= deadline {

6
codex-rs/core/tests/suite/sqlite_state.rs
View file

@ -33,7 +33,7 @@ async fn new_thread_is_recorded_in_state_db() -> Result<()> {
let thread_id = test.session_configured.session_id;
let rollout_path = test.codex.rollout_path().expect("rollout path");
let db_path = codex_state::state_db_path(test.config.codex_home.as_path());
let db_path = codex_state::state_db_path(test.config.sqlite_home.as_path());
for _ in 0..100 {
if tokio::fs::try_exists(&db_path).await.unwrap_or(false) {
@ -161,7 +161,7 @@ async fn backfill_scans_existing_rollouts() -> Result<()> {
let test = builder.build(&server).await?;
let db_path = codex_state::state_db_path(test.config.codex_home.as_path());
let db_path = codex_state::state_db_path(test.config.sqlite_home.as_path());
let rollout_path = test.config.codex_home.join(&rollout_rel_path);
let default_provider = test.config.model_provider_id.clone();
@ -220,7 +220,7 @@ async fn user_messages_persist_in_state_db() -> Result<()> {
});
let test = builder.build(&server).await?;
let db_path = codex_state::state_db_path(test.config.codex_home.as_path());
let db_path = codex_state::state_db_path(test.config.sqlite_home.as_path());
for _ in 0..100 {
if tokio::fs::try_exists(&db_path).await.unwrap_or(false) {
break;

4
codex-rs/exec/src/cli.rs
View file

@ -86,6 +86,10 @@ pub struct Cli {
#[arg(long = "color", value_enum, default_value_t = Color::Auto)]
pub color: Color,
/// Force cursor-based progress updates in exec mode.
#[arg(long = "progress-cursor", default_value_t = false)]
pub progress_cursor: bool,
/// Print events to stdout as JSONL.
#[arg(
long = "json",

244
codex-rs/exec/src/event_processor_with_human_output.rs
View file

@ -38,9 +38,12 @@ use codex_utils_elapsed::format_duration;
use codex_utils_elapsed::format_elapsed;
use owo_colors::OwoColorize;
use owo_colors::Style;
use serde::Deserialize;
use shlex::try_join;
use std::collections::HashMap;
use std::io::Write;
use std::path::PathBuf;
use std::time::Duration;
use std::time::Instant;
use crate::event_processor::CodexStatus;
@ -76,11 +79,17 @@ pub(crate) struct EventProcessorWithHumanOutput {
last_total_token_usage: Option<codex_protocol::protocol::TokenUsageInfo>,
final_message: Option<String>,
last_proposed_plan: Option<String>,
progress_active: bool,
progress_last_len: usize,
use_ansi_cursor: bool,
progress_anchor: bool,
progress_done: bool,
}
impl EventProcessorWithHumanOutput {
pub(crate) fn create_with_ansi(
with_ansi: bool,
cursor_ansi: bool,
config: &Config,
last_message_path: Option<PathBuf>,
) -> Self {
@ -103,6 +112,11 @@ impl EventProcessorWithHumanOutput {
last_total_token_usage: None,
final_message: None,
last_proposed_plan: None,
progress_active: false,
progress_last_len: 0,
use_ansi_cursor: cursor_ansi,
progress_anchor: false,
progress_done: false,
}
} else {
Self {
@ -121,11 +135,27 @@ impl EventProcessorWithHumanOutput {
last_total_token_usage: None,
final_message: None,
last_proposed_plan: None,
progress_active: false,
progress_last_len: 0,
use_ansi_cursor: cursor_ansi,
progress_anchor: false,
progress_done: false,
}
}
}
}
#[derive(Debug, Deserialize)]
struct AgentJobProgressMessage {
job_id: String,
total_items: usize,
pending_items: usize,
running_items: usize,
completed_items: usize,
failed_items: usize,
eta_seconds: Option<u64>,
}
struct PatchApplyBegin {
start_time: Instant,
auto_approved: bool,
@ -176,6 +206,18 @@ impl EventProcessor for EventProcessorWithHumanOutput {
fn process_event(&mut self, event: Event) -> CodexStatus {
let Event { id: _, msg } = event;
if let EventMsg::BackgroundEvent(BackgroundEventEvent { message }) = &msg
&& let Some(update) = Self::parse_agent_job_progress(message)
{
self.render_agent_job_progress(update);
return CodexStatus::Running;
}
if self.progress_active && !Self::should_interrupt_progress(&msg) {
return CodexStatus::Running;
}
if !Self::is_silent_event(&msg) {
self.finish_progress_line();
}
match msg {
EventMsg::Error(ErrorEvent { message, .. }) => {
let prefix = "ERROR:".style(self.red);
@ -818,6 +860,7 @@ impl EventProcessor for EventProcessorWithHumanOutput {
}
fn print_final_output(&mut self) {
self.finish_progress_line();
if let Some(usage_info) = &self.last_total_token_usage {
eprintln!(
"{}\n{}",
@ -841,6 +884,207 @@ impl EventProcessor for EventProcessorWithHumanOutput {
}
}
impl EventProcessorWithHumanOutput {
fn parse_agent_job_progress(message: &str) -> Option<AgentJobProgressMessage> {
let payload = message.strip_prefix("agent_job_progress:")?;
serde_json::from_str::<AgentJobProgressMessage>(payload).ok()
}
fn is_silent_event(msg: &EventMsg) -> bool {
matches!(
msg,
EventMsg::ThreadNameUpdated(_)
| EventMsg::TokenCount(_)
| EventMsg::TurnStarted(_)
| EventMsg::ExecApprovalRequest(_)
| EventMsg::ApplyPatchApprovalRequest(_)
| EventMsg::TerminalInteraction(_)
| EventMsg::ExecCommandOutputDelta(_)
| EventMsg::GetHistoryEntryResponse(_)
| EventMsg::McpListToolsResponse(_)
| EventMsg::ListCustomPromptsResponse(_)
| EventMsg::ListSkillsResponse(_)
| EventMsg::ListRemoteSkillsResponse(_)
| EventMsg::RemoteSkillDownloaded(_)
| EventMsg::RawResponseItem(_)
| EventMsg::UserMessage(_)
| EventMsg::EnteredReviewMode(_)
| EventMsg::ExitedReviewMode(_)
| EventMsg::AgentMessageDelta(_)
| EventMsg::AgentReasoningDelta(_)
| EventMsg::AgentReasoningRawContentDelta(_)
| EventMsg::ItemStarted(_)
| EventMsg::ItemCompleted(_)
| EventMsg::AgentMessageContentDelta(_)
| EventMsg::PlanDelta(_)
| EventMsg::ReasoningContentDelta(_)
| EventMsg::ReasoningRawContentDelta(_)
| EventMsg::SkillsUpdateAvailable
| EventMsg::UndoCompleted(_)
| EventMsg::UndoStarted(_)
| EventMsg::ThreadRolledBack(_)
| EventMsg::RequestUserInput(_)
| EventMsg::DynamicToolCallRequest(_)
)
}
fn should_interrupt_progress(msg: &EventMsg) -> bool {
matches!(
msg,
EventMsg::Error(_)
| EventMsg::Warning(_)
| EventMsg::DeprecationNotice(_)
| EventMsg::StreamError(_)
| EventMsg::TurnComplete(_)
| EventMsg::ShutdownComplete
)
}
fn finish_progress_line(&mut self) {
if self.progress_active {
self.progress_active = false;
self.progress_last_len = 0;
self.progress_done = false;
if self.use_ansi_cursor {
if self.progress_anchor {
eprintln!("\u{1b}[1A\u{1b}[1G\u{1b}[2K");
} else {
eprintln!("\u{1b}[1G\u{1b}[2K");
}
} else {
eprintln!();
}
self.progress_anchor = false;
}
}
fn render_agent_job_progress(&mut self, update: AgentJobProgressMessage) {
let total = update.total_items.max(1);
let processed = update.completed_items + update.failed_items;
let percent = (processed as f64 / total as f64 * 100.0).round() as i64;
let job_label = update.job_id.chars().take(8).collect::<String>();
let eta = update
.eta_seconds
.map(|secs| format_duration(Duration::from_secs(secs)))
.unwrap_or_else(|| "--".to_string());
let columns = std::env::var("COLUMNS")
.ok()
.and_then(|value| value.parse::<usize>().ok())
.filter(|value| *value > 0);
let line = format_agent_job_progress_line(
columns,
job_label.as_str(),
AgentJobProgressStats {
processed,
total,
percent,
failed: update.failed_items,
running: update.running_items,
pending: update.pending_items,
},
eta.as_str(),
);
let done = processed >= update.total_items;
if !self.use_ansi_cursor {
eprintln!("{line}");
if done {
self.progress_active = false;
self.progress_last_len = 0;
}
return;
}
if done && self.progress_done {
return;
}
if !self.progress_active {
eprintln!();
self.progress_anchor = true;
self.progress_done = false;
}
let mut output = String::new();
if self.progress_anchor {
output.push_str("\u{1b}[1A\u{1b}[1G\u{1b}[2K");
} else {
output.push_str("\u{1b}[1G\u{1b}[2K");
}
output.push_str(&line);
if done {
output.push('\n');
eprint!("{output}");
self.progress_active = false;
self.progress_last_len = 0;
self.progress_anchor = false;
self.progress_done = true;
return;
}
eprint!("{output}");
let _ = std::io::stderr().flush();
self.progress_active = true;
self.progress_last_len = line.len();
}
}
struct AgentJobProgressStats {
processed: usize,
total: usize,
percent: i64,
failed: usize,
running: usize,
pending: usize,
}
fn format_agent_job_progress_line(
columns: Option<usize>,
job_label: &str,
stats: AgentJobProgressStats,
eta: &str,
) -> String {
let rest = format!(
"{processed}/{total} {percent}% f{failed} r{running} p{pending} eta {eta}",
processed = stats.processed,
total = stats.total,
percent = stats.percent,
failed = stats.failed,
running = stats.running,
pending = stats.pending
);
let prefix = format!("job {job_label}");
let base_len = prefix.len() + rest.len() + 4;
let mut bar_width = columns
.and_then(|columns| columns.checked_sub(base_len))
.filter(|available| *available > 0)
.unwrap_or(20usize);
let with_bar = |width: usize| {
let filled = ((stats.processed as f64 / stats.total as f64) * width as f64)
.round()
.clamp(0.0, width as f64) as usize;
let mut bar = "#".repeat(filled);
bar.push_str(&"-".repeat(width - filled));
format!("{prefix} [{bar}] {rest}")
};
let mut line = with_bar(bar_width);
if let Some(columns) = columns
&& line.len() > columns
{
let min_line = format!("{prefix} {rest}");
if min_line.len() > columns {
let mut truncated = min_line;
if columns > 2 && truncated.len() > columns {
truncated.truncate(columns - 2);
truncated.push_str("..");
}
return truncated;
}
let available = columns.saturating_sub(base_len);
if available == 0 {
return min_line;
}
bar_width = available.min(bar_width).max(1);
line = with_bar(bar_width);
}
line
}
fn escape_command(command: &[String]) -> String {
try_join(command.iter().map(String::as_str)).unwrap_or_else(|_| command.join(" "))
}

66
codex-rs/exec/src/lib.rs
View file

@ -41,6 +41,7 @@ use codex_protocol::protocol::Op;
use codex_protocol::protocol::ReviewRequest;
use codex_protocol::protocol::ReviewTarget;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SubAgentSource;
use codex_protocol::user_input::UserInput;
use codex_utils_absolute_path::AbsolutePathBuf;
use codex_utils_oss::ensure_oss_provider_ready;
@ -86,6 +87,7 @@ struct ThreadEventEnvelope {
thread_id: codex_protocol::ThreadId,
thread: Arc<codex_core::CodexThread>,
event: Event,
suppress_output: bool,
}
pub async fn run_main(cli: Cli, codex_linux_sandbox_exe: Option<PathBuf>) -> anyhow::Result<()> {
@ -113,9 +115,10 @@ pub async fn run_main(cli: Cli, codex_linux_sandbox_exe: Option<PathBuf>) -> any
prompt,
output_schema: output_schema_path,
config_overrides,
progress_cursor,
} = cli;
let (stdout_with_ansi, stderr_with_ansi) = match color {
let (_stdout_with_ansi, stderr_with_ansi) = match color {
cli::Color::Always => (true, true),
cli::Color::Never => (false, false),
cli::Color::Auto => (
@ -123,6 +126,24 @@ pub async fn run_main(cli: Cli, codex_linux_sandbox_exe: Option<PathBuf>) -> any
supports_color::on_cached(Stream::Stderr).is_some(),
),
};
let cursor_ansi = if progress_cursor {
true
} else {
match color {
cli::Color::Never => false,
cli::Color::Always => true,
cli::Color::Auto => {
if stderr_with_ansi || std::io::stderr().is_terminal() {
true
} else {
match std::env::var("TERM") {
Ok(term) => !term.is_empty() && term != "dumb",
Err(_) => false,
}
}
}
}
};
// Build fmt layer (existing logging) to compose with OTEL layer.
let default_level = "error";
@ -318,7 +339,8 @@ pub async fn run_main(cli: Cli, codex_linux_sandbox_exe: Option<PathBuf>) -> any
let mut event_processor: Box<dyn EventProcessor> = match json_mode {
true => Box::new(EventProcessorWithJsonOutput::new(last_message_file.clone())),
_ => Box::new(EventProcessorWithHumanOutput::create_with_ansi(
stdout_with_ansi,
stderr_with_ansi,
cursor_ansi,
&config,
last_message_file.clone(),
)),
@ -466,7 +488,7 @@ pub async fn run_main(cli: Cli, codex_linux_sandbox_exe: Option<PathBuf>) -> any
let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<ThreadEventEnvelope>();
let attached_threads = Arc::new(Mutex::new(HashSet::from([primary_thread_id])));
spawn_thread_listener(primary_thread_id, thread.clone(), tx.clone());
spawn_thread_listener(primary_thread_id, thread.clone(), tx.clone(), false);
{
let thread = thread.clone();
@ -494,7 +516,14 @@ pub async fn run_main(cli: Cli, codex_linux_sandbox_exe: Option<PathBuf>) -> any
match thread_manager.get_thread(thread_id).await {
Ok(thread) => {
attached_threads.lock().await.insert(thread_id);
spawn_thread_listener(thread_id, thread, tx.clone());
let suppress_output =
is_agent_job_subagent(&thread.config_snapshot().await);
spawn_thread_listener(
thread_id,
thread,
tx.clone(),
suppress_output,
);
}
Err(err) => {
warn!("failed to attach listener for thread {thread_id}: {err}")
@ -549,7 +578,11 @@ pub async fn run_main(cli: Cli, codex_linux_sandbox_exe: Option<PathBuf>) -> any
thread_id,
thread,
event,
suppress_output,
} = envelope;
if suppress_output && should_suppress_agent_job_event(&event.msg) {
continue;
}
if matches!(event.msg, EventMsg::Error(_)) {
error_seen = true;
}
@ -613,6 +646,7 @@ fn spawn_thread_listener(
thread_id: codex_protocol::ThreadId,
thread: Arc<codex_core::CodexThread>,
tx: tokio::sync::mpsc::UnboundedSender<ThreadEventEnvelope>,
suppress_output: bool,
) {
tokio::spawn(async move {
loop {
@ -625,6 +659,7 @@ fn spawn_thread_listener(
thread_id,
thread: Arc::clone(&thread),
event,
suppress_output,
}) {
error!("Error sending event: {err:?}");
break;
@ -645,6 +680,29 @@ fn spawn_thread_listener(
});
}
fn is_agent_job_subagent(config: &codex_core::ThreadConfigSnapshot) -> bool {
match &config.session_source {
SessionSource::SubAgent(SubAgentSource::Other(source)) => source.starts_with("agent_job:"),
_ => false,
}
}
fn should_suppress_agent_job_event(msg: &EventMsg) -> bool {
!matches!(
msg,
EventMsg::ExecApprovalRequest(_)
| EventMsg::ApplyPatchApprovalRequest(_)
| EventMsg::RequestUserInput(_)
| EventMsg::DynamicToolCallRequest(_)
| EventMsg::ElicitationRequest(_)
| EventMsg::Error(_)
| EventMsg::Warning(_)
| EventMsg::DeprecationNotice(_)
| EventMsg::StreamError(_)
| EventMsg::ShutdownComplete
)
}
async fn resolve_resume_path(
config: &Config,
args: &crate::cli::ResumeArgs,

38
codex-rs/state/migrations/0014_agent_jobs.sql Normal file
View file

@ -0,0 +1,38 @@
CREATE TABLE agent_jobs (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
status TEXT NOT NULL,
instruction TEXT NOT NULL,
output_schema_json TEXT,
input_headers_json TEXT NOT NULL,
input_csv_path TEXT NOT NULL,
output_csv_path TEXT NOT NULL,
auto_export INTEGER NOT NULL DEFAULT 1,
created_at INTEGER NOT NULL,
updated_at INTEGER NOT NULL,
started_at INTEGER,
completed_at INTEGER,
last_error TEXT
);
CREATE TABLE agent_job_items (
job_id TEXT NOT NULL,
item_id TEXT NOT NULL,
row_index INTEGER NOT NULL,
source_id TEXT,
row_json TEXT NOT NULL,
status TEXT NOT NULL,
assigned_thread_id TEXT,
attempt_count INTEGER NOT NULL DEFAULT 0,
result_json TEXT,
last_error TEXT,
created_at INTEGER NOT NULL,
updated_at INTEGER NOT NULL,
completed_at INTEGER,
reported_at INTEGER,
PRIMARY KEY (job_id, item_id),
FOREIGN KEY(job_id) REFERENCES agent_jobs(id) ON DELETE CASCADE
);
CREATE INDEX idx_agent_jobs_status ON agent_jobs(status, updated_at DESC);
CREATE INDEX idx_agent_job_items_status ON agent_job_items(job_id, status, row_index ASC);

2
codex-rs/state/migrations/0015_agent_jobs_max_runtime_seconds.sql Normal file
View file

@ -0,0 +1,2 @@
ALTER TABLE agent_jobs
ADD COLUMN max_runtime_seconds INTEGER;

10
codex-rs/state/src/lib.rs
View file

@ -22,6 +22,13 @@ pub use runtime::StateRuntime;
///
/// Most consumers should prefer [`StateRuntime`].
pub use extract::apply_rollout_item;
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;
@ -38,6 +45,9 @@ pub use model::ThreadsPage;
pub use runtime::state_db_filename;
pub use runtime::state_db_path;
/// Environment variable for overriding the SQLite state database home directory.
pub const SQLITE_HOME_ENV: &str = "CODEX_SQLITE_HOME";
pub const STATE_DB_FILENAME: &str = "state";
pub const STATE_DB_VERSION: u32 = 5;

256
codex-rs/state/src/model/agent_job.rs Normal file
View file

@ -0,0 +1,256 @@
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
codex-rs/state/src/model/mod.rs
View file

@ -1,8 +1,16 @@
mod agent_job;
mod backfill_state;
mod log;
mod memories;
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 log::LogEntry;
@ -21,6 +29,8 @@ pub use thread_metadata::ThreadMetadata;
pub use thread_metadata::ThreadMetadataBuilder;
pub use thread_metadata::ThreadsPage;
pub(crate) use agent_job::AgentJobItemRow;
pub(crate) use agent_job::AgentJobRow;
pub(crate) use memories::Stage1OutputRow;
pub(crate) use thread_metadata::ThreadRow;
pub(crate) use thread_metadata::anchor_from_item;

567
codex-rs/state/src/runtime.rs
View file

@ -1,3 +1,10 @@
use crate::AgentJob;
use crate::AgentJobCreateParams;
use crate::AgentJobItem;
use crate::AgentJobItemCreateParams;
use crate::AgentJobItemStatus;
use crate::AgentJobProgress;
use crate::AgentJobStatus;
use crate::DB_ERROR_METRIC;
use crate::LogEntry;
use crate::LogQuery;
@ -11,6 +18,8 @@ use crate::ThreadMetadataBuilder;
use crate::ThreadsPage;
use crate::apply_rollout_item;
use crate::migrations::MIGRATOR;
use crate::model::AgentJobItemRow;
use crate::model::AgentJobRow;
use crate::model::ThreadRow;
use crate::model::anchor_from_item;
use crate::model::datetime_to_epoch_seconds;
@ -901,6 +910,564 @@ ON CONFLICT(thread_id, position) DO NOTHING
Ok(result.rows_affected())
}
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 = 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 = 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
result_json = ?,
reported_at = ?,
updated_at = ?,
last_error = NULL
WHERE
job_id = ?
AND item_id = ?
AND status = ?
AND assigned_thread_id = ?
"#,
)
.bind(serialized)
.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(),
})
}
async fn ensure_backfill_state_row(&self) -> anyhow::Result<()> {
sqlx::query(
r#"

6
docs/config.md
View file

@ -28,6 +28,12 @@ Codex can run a notification hook when the agent finishes a turn. See the config
The generated JSON Schema for `config.toml` lives at `codex-rs/core/config.schema.json`.
## SQLite State DB
Codex stores the SQLite-backed state DB under `sqlite_home` (config key) or the
`CODEX_SQLITE_HOME` environment variable. When unset, WorkspaceWrite sandbox
sessions default to a temp directory; other modes default to `CODEX_HOME`.
## Notices
Codex stores "do not show again" flags for some UI prompts under the `[notice]` table.