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

Add experimental thread job scheduler for /loop

Browse Source
This commit is contained in:
Eric Traut
2026-03-30 15:30:34 -06:00
parent 716f7b0428
commit a02c2cd50c
52 changed files with 2291 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/agent/control_tests.rs
+2 -2
View File
@@ -495,7 +495,7 @@ async fn append_message_records_assistant_message() {
.expect("append_message should succeed");
assert!(!submission_id.is_empty());
timeout(Duration::from_secs(5), async {
timeout(Duration::from_secs(15), async {
loop {
let history_items = thread
.codex
@@ -1478,7 +1478,7 @@ async fn resume_thread_subagent_restores_stored_nickname_and_role() {
let state_db = child_thread
.state_db()
.expect("sqlite state db should be available for nickname resume test");
timeout(Duration::from_secs(5), async {
timeout(Duration::from_secs(15), async {
loop {
if let Ok(Some(metadata)) = state_db.get_thread(child_thread_id).await
&& metadata.agent_nickname.is_some()
codex-rs/core/src/codex.rs
+159 -2
View File
@@ -259,6 +259,13 @@ use crate::injection::ToolMentionKind;
use crate::injection::app_id_from_path;
use crate::injection::tool_kind_for_path;
use crate::instructions::UserInstructions;
use crate::jobs::ClaimedJob;
use crate::jobs::JOB_UPDATED_BACKGROUND_EVENT_PREFIX;
use crate::jobs::JobTurnContext;
use crate::jobs::JobsState;
use crate::jobs::ThreadJob;
use crate::jobs::job_prompt_input_item;
use crate::jobs::job_turn_developer_instructions;
use crate::mcp::CODEX_APPS_MCP_SERVER_NAME;
use crate::mcp::McpManager;
use crate::mcp::auth::compute_auth_statuses;
@@ -812,6 +819,7 @@ pub(crate) struct Session {
mailbox: Mailbox,
mailbox_rx: Mutex<MailboxReceiver>,
idle_pending_input: Mutex<Vec<ResponseInputItem>>, // TODO (jif) merge with mailbox!
jobs: Mutex<JobsState>,
pub(crate) guardian_review_session: GuardianReviewSessionManager,
pub(crate) services: SessionServices,
js_repl: Arc<JsReplHandle>,
@@ -856,6 +864,7 @@ pub(crate) struct TurnContext {
pub(crate) timezone: Option<String>,
pub(crate) app_server_client_name: Option<String>,
pub(crate) developer_instructions: Option<String>,
pub(crate) job_context: Option<JobTurnContext>,
pub(crate) compact_prompt: Option<String>,
pub(crate) user_instructions: Option<String>,
pub(crate) collaboration_mode: CollaborationMode,
@@ -964,6 +973,7 @@ impl TurnContext {
timezone: self.timezone.clone(),
app_server_client_name: self.app_server_client_name.clone(),
developer_instructions: self.developer_instructions.clone(),
job_context: self.job_context.clone(),
compact_prompt: self.compact_prompt.clone(),
user_instructions: self.user_instructions.clone(),
collaboration_mode,
@@ -1430,6 +1440,7 @@ impl Session {
timezone: Some(timezone),
app_server_client_name: session_configuration.app_server_client_name.clone(),
developer_instructions: session_configuration.developer_instructions.clone(),
job_context: None,
compact_prompt: session_configuration.compact_prompt.clone(),
user_instructions: session_configuration.user_instructions.clone(),
collaboration_mode: session_configuration.collaboration_mode.clone(),
@@ -1926,6 +1937,7 @@ impl Session {
mailbox,
mailbox_rx: Mutex::new(mailbox_rx),
idle_pending_input: Mutex::new(Vec::new()),
jobs: Mutex::new(JobsState::default()),
guardian_review_session: GuardianReviewSessionManager::default(),
services,
js_repl,
@@ -2413,6 +2425,7 @@ impl Session {
session_configuration,
updates.final_output_json_schema,
sandbox_policy_changed,
/*job_context*/ None,
)
.await)
}
@@ -2423,6 +2436,7 @@ impl Session {
session_configuration: SessionConfiguration,
final_output_json_schema: Option<Option<Value>>,
sandbox_policy_changed: bool,
job_context: Option<JobTurnContext>,
) -> Arc<TurnContext> {
let per_turn_config = Self::build_per_turn_config(&session_configuration);
self.services
@@ -2495,6 +2509,7 @@ impl Session {
if let Some(final_schema) = final_output_json_schema {
turn_context.final_output_json_schema = final_schema;
}
turn_context.job_context = job_context;
let turn_context = Arc::new(turn_context);
turn_context.turn_metadata_state.spawn_git_enrichment_task();
turn_context
@@ -2600,10 +2615,150 @@ impl Session {
session_configuration,
/*final_output_json_schema*/ None,
/*sandbox_policy_changed*/ false,
/*job_context*/ None,
)
.await
}
pub(crate) async fn new_job_turn_with_sub_id(
&self,
sub_id: String,
job_context: JobTurnContext,
) -> Arc<TurnContext> {
let session_configuration = {
let state = self.state.lock().await;
state.session_configuration.clone()
};
self.new_turn_from_configuration(
sub_id,
session_configuration,
/*final_output_json_schema*/ None,
/*sandbox_policy_changed*/ false,
Some(job_context),
)
.await
}
pub(crate) async fn list_jobs(&self) -> Vec<ThreadJob> {
self.jobs.lock().await.list_jobs()
}
pub(crate) async fn create_job(
self: &Arc<Self>,
cron_expression: String,
prompt: String,
run_once: bool,
) -> Result<ThreadJob, String> {
let schedule = crate::jobs::JobSchedule::parse(&cron_expression)?;
let timer_cancel = matches!(schedule, crate::jobs::JobSchedule::EverySeconds(_))
.then(CancellationToken::new);
let id = uuid::Uuid::new_v4().to_string();
let job = {
let mut jobs = self.jobs.lock().await;
jobs.create_job(
id.clone(),
cron_expression,
prompt,
run_once,
Utc::now(),
timer_cancel.clone(),
)?
};
if let (crate::jobs::JobSchedule::EverySeconds(seconds), Some(cancel)) =
(schedule, timer_cancel)
{
self.spawn_job_timer(id, seconds, cancel);
}
self.emit_job_updated_notification().await;
self.maybe_start_pending_job().await;
Ok(job)
}
pub(crate) async fn delete_job(&self, id: &str) -> bool {
let deleted = self.jobs.lock().await.delete_job(id);
if deleted {
self.emit_job_updated_notification().await;
}
deleted
}
pub(crate) async fn mark_after_turn_jobs_due(&self) {
self.jobs.lock().await.mark_after_turn_jobs_due();
}
pub(crate) async fn maybe_start_pending_job(self: &Arc<Self>) {
if self.active_turn.lock().await.is_some() {
return;
}
if self.has_queued_response_items_for_next_turn().await
|| self.has_trigger_turn_mailbox_items().await
{
return;
}
let Some(ClaimedJob {
job: _,
context,
deleted_run_once_job,
}) = self.jobs.lock().await.claim_next_job(Utc::now())
else {
return;
};
if deleted_run_once_job {
self.emit_job_updated_notification().await;
}
self.queue_response_items_for_next_turn(vec![job_prompt_input_item(&context.prompt)])
.await;
let turn_context = self
.new_job_turn_with_sub_id(uuid::Uuid::new_v4().to_string(), context)
.await;
self.maybe_emit_unknown_model_warning_for_turn(turn_context.as_ref())
.await;
tokio::task::block_in_place(|| {
tokio::runtime::Handle::current().block_on(self.start_task(
turn_context,
Vec::new(),
crate::tasks::RegularTask::new(),
));
});
}
fn spawn_job_timer(
self: &Arc<Self>,
id: String,
seconds: u64,
cancellation_token: CancellationToken,
) {
let weak = Arc::downgrade(self);
tokio::spawn(async move {
loop {
tokio::select! {
_ = cancellation_token.cancelled() => break,
_ = tokio::time::sleep(std::time::Duration::from_secs(seconds)) => {}
}
let Some(session) = weak.upgrade() else {
break;
};
session.jobs.lock().await.mark_job_due(&id, Utc::now());
session.maybe_start_pending_job().await;
}
});
}
async fn emit_job_updated_notification(&self) {
let jobs = self.list_jobs().await;
let Ok(payload) = serde_json::to_string(&jobs) else {
warn!("failed to serialize job update payload");
return;
};
self.send_event_raw(Event {
id: INITIAL_SUBMIT_ID.to_owned(),
msg: EventMsg::BackgroundEvent(BackgroundEventEvent {
message: format!("{JOB_UPDATED_BACKGROUND_EVENT_PREFIX}{payload}"),
}),
})
.await;
}
async fn build_settings_update_items(
&self,
reference_context_item: Option<&TurnContextItem>,
@@ -3519,6 +3674,9 @@ impl Session {
{
developer_sections.push(developer_instructions.to_string());
}
if let Some(job_context) = turn_context.job_context.as_ref() {
developer_sections.push(job_turn_developer_instructions(job_context));
}
// Add developer instructions for memories.
if turn_context.features.enabled(Feature::MemoryTool)
&& turn_context.config.memories.use_memories
@@ -4019,8 +4177,6 @@ impl Session {
}
}
/// Queue response items to be injected into the next active turn created for this session.
#[cfg(test)]
pub(crate) async fn queue_response_items_for_next_turn(&self, items: Vec<ResponseInputItem>) {
if items.is_empty() {
return;
@@ -5466,6 +5622,7 @@ async fn spawn_review_thread(
timezone: parent_turn_context.timezone.clone(),
app_server_client_name: parent_turn_context.app_server_client_name.clone(),
developer_instructions: None,
job_context: None,
user_instructions: None,
compact_prompt: parent_turn_context.compact_prompt.clone(),
collaboration_mode: parent_turn_context.collaboration_mode.clone(),
codex-rs/core/src/codex_tests.rs
+3
View File
@@ -12,6 +12,7 @@ use crate::config_loader::Sourced;
use crate::exec::ExecCapturePolicy;
use crate::exec::ExecToolCallOutput;
use crate::function_tool::FunctionCallError;
use crate::jobs::JobsState;
use crate::mcp_connection_manager::ToolInfo;
use crate::models_manager::model_info;
use crate::shell::default_user_shell;
@@ -2751,6 +2752,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) {
mailbox,
mailbox_rx: Mutex::new(mailbox_rx),
idle_pending_input: Mutex::new(Vec::new()),
jobs: Mutex::new(JobsState::default()),
guardian_review_session: crate::guardian::GuardianReviewSessionManager::default(),
services,
js_repl,
@@ -3594,6 +3596,7 @@ pub(crate) async fn make_session_and_context_with_dynamic_tools_and_rx(
mailbox,
mailbox_rx: Mutex::new(mailbox_rx),
idle_pending_input: Mutex::new(Vec::new()),
jobs: Mutex::new(JobsState::default()),
guardian_review_session: crate::guardian::GuardianReviewSessionManager::default(),
services,
js_repl,
codex-rs/core/src/codex_thread.rs
+21
View File
@@ -26,6 +26,7 @@ use std::path::PathBuf;
use tokio::sync::Mutex;
use tokio::sync::watch;
use crate::jobs::ThreadJob;
use crate::state_db::StateDbHandle;
#[derive(Clone, Debug)]
@@ -236,6 +237,26 @@ impl CodexThread {
Ok(*guard)
}
pub async fn create_job(
&self,
cron_expression: String,
prompt: String,
run_once: bool,
) -> Result<ThreadJob, String> {
self.codex
.session
.create_job(cron_expression, prompt, run_once)
.await
}
pub async fn delete_job(&self, id: &str) -> bool {
self.codex.session.delete_job(id).await
}
pub async fn list_jobs(&self) -> Vec<ThreadJob> {
self.codex.session.list_jobs().await
}
}
fn pending_message_input_item(message: &ResponseItem) -> CodexResult<ResponseInputItem> {
codex-rs/core/src/config/config_tests.rs
+80 -55
View File
@@ -34,6 +34,7 @@ use core_test_support::PathExt;
use core_test_support::TempDirExt;
use core_test_support::test_absolute_path;
use pretty_assertions::assert_eq;
use serial_test::serial;
use std::collections::BTreeMap;
use std::collections::HashMap;
@@ -5724,79 +5725,95 @@ shell_tool = true
}
#[tokio::test]
#[serial]
async fn approvals_reviewer_defaults_to_manual_only_without_guardian_feature() -> std::io::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 config = Config::load_from_base_config_with_overrides(
ConfigToml::default(),
ConfigOverrides {
cwd: Some(codex_home.path().to_path_buf()),
..Default::default()
},
codex_home.path().to_path_buf(),
)?;
assert_eq!(config.approvals_reviewer, ApprovalsReviewer::User);
Ok(())
}
#[tokio::test]
#[serial]
async fn approvals_reviewer_stays_manual_only_when_guardian_feature_is_enabled()
-> std::io::Result<()> {
let codex_home = TempDir::new()?;
std::fs::write(
codex_home.path().join(CONFIG_TOML_FILE),
r#"[features]
let raw_config = r#"[features]
guardian_approval = true
"#,
)?;
"#;
std::fs::write(codex_home.path().join(CONFIG_TOML_FILE), raw_config)?;
let config_toml = toml::from_str::<ConfigToml>(raw_config)
.map_err(|err| std::io::Error::new(std::io::ErrorKind::InvalidData, err))?;
let config = ConfigBuilder::default()
.codex_home(codex_home.path().to_path_buf())
.fallback_cwd(Some(codex_home.path().to_path_buf()))
.build()
.await?;
let config = Config::load_from_base_config_with_overrides(
config_toml,
ConfigOverrides {
cwd: Some(codex_home.path().to_path_buf()),
..Default::default()
},
codex_home.path().to_path_buf(),
)?;
assert_eq!(config.approvals_reviewer, ApprovalsReviewer::User);
Ok(())
}
#[tokio::test]
#[serial]
async fn approvals_reviewer_can_be_set_in_config_without_guardian_approval() -> std::io::Result<()>
{
let codex_home = TempDir::new()?;
std::fs::write(
codex_home.path().join(CONFIG_TOML_FILE),
r#"approvals_reviewer = "user"
"#,
)?;
let raw_config = r#"approvals_reviewer = "user"
"#;
std::fs::write(codex_home.path().join(CONFIG_TOML_FILE), raw_config)?;
let config_toml = toml::from_str::<ConfigToml>(raw_config)
.map_err(|err| std::io::Error::new(std::io::ErrorKind::InvalidData, err))?;
let config = ConfigBuilder::default()
.codex_home(codex_home.path().to_path_buf())
.fallback_cwd(Some(codex_home.path().to_path_buf()))
.build()
.await?;
let config = Config::load_from_base_config_with_overrides(
config_toml,
ConfigOverrides {
cwd: Some(codex_home.path().to_path_buf()),
..Default::default()
},
codex_home.path().to_path_buf(),
)?;
assert_eq!(config.approvals_reviewer, ApprovalsReviewer::User);
Ok(())
}
#[tokio::test]
#[serial]
async fn approvals_reviewer_can_be_set_in_profile_without_guardian_approval() -> std::io::Result<()>
{
let codex_home = TempDir::new()?;
std::fs::write(
codex_home.path().join(CONFIG_TOML_FILE),
r#"profile = "guardian"
let raw_config = r#"profile = "guardian"
[profiles.guardian]
approvals_reviewer = "guardian_subagent"
"#,
)?;
"#;
std::fs::write(codex_home.path().join(CONFIG_TOML_FILE), raw_config)?;
let config_toml = toml::from_str::<ConfigToml>(raw_config)
.map_err(|err| std::io::Error::new(std::io::ErrorKind::InvalidData, err))?;
let config = ConfigBuilder::default()
.codex_home(codex_home.path().to_path_buf())
.fallback_cwd(Some(codex_home.path().to_path_buf()))
.build()
.await?;
let config = Config::load_from_base_config_with_overrides(
config_toml,
ConfigOverrides {
cwd: Some(codex_home.path().to_path_buf()),
..Default::default()
},
codex_home.path().to_path_buf(),
)?;
assert_eq!(
config.approvals_reviewer,
@@ -5806,20 +5823,24 @@ approvals_reviewer = "guardian_subagent"
}
#[tokio::test]
#[serial]
async fn smart_approvals_alias_is_ignored() -> std::io::Result<()> {
let codex_home = TempDir::new()?;
std::fs::write(
codex_home.path().join(CONFIG_TOML_FILE),
r#"[features]
let raw_config = r#"[features]
smart_approvals = true
"#,
)?;
"#;
std::fs::write(codex_home.path().join(CONFIG_TOML_FILE), raw_config)?;
let config_toml = toml::from_str::<ConfigToml>(raw_config)
.map_err(|err| std::io::Error::new(std::io::ErrorKind::InvalidData, err))?;
let config = ConfigBuilder::default()
.codex_home(codex_home.path().to_path_buf())
.fallback_cwd(Some(codex_home.path().to_path_buf()))
.build()
.await?;
let config = Config::load_from_base_config_with_overrides(
config_toml,
ConfigOverrides {
cwd: Some(codex_home.path().to_path_buf()),
..Default::default()
},
codex_home.path().to_path_buf(),
)?;
assert!(!config.features.enabled(Feature::GuardianApproval));
assert_eq!(config.approvals_reviewer, ApprovalsReviewer::User);
@@ -5833,22 +5854,26 @@ smart_approvals = true
}
#[tokio::test]
#[serial]
async fn smart_approvals_alias_is_ignored_in_profiles() -> std::io::Result<()> {
let codex_home = TempDir::new()?;
std::fs::write(
codex_home.path().join(CONFIG_TOML_FILE),
r#"profile = "guardian"
let raw_config = r#"profile = "guardian"
[profiles.guardian.features]
smart_approvals = true
"#,
)?;
"#;
std::fs::write(codex_home.path().join(CONFIG_TOML_FILE), raw_config)?;
let config_toml = toml::from_str::<ConfigToml>(raw_config)
.map_err(|err| std::io::Error::new(std::io::ErrorKind::InvalidData, err))?;
let config = ConfigBuilder::default()
.codex_home(codex_home.path().to_path_buf())
.fallback_cwd(Some(codex_home.path().to_path_buf()))
.build()
.await?;
let config = Config::load_from_base_config_with_overrides(
config_toml,
ConfigOverrides {
cwd: Some(codex_home.path().to_path_buf()),
..Default::default()
},
codex_home.path().to_path_buf(),
)?;
assert!(!config.features.enabled(Feature::GuardianApproval));
assert_eq!(config.approvals_reviewer, ApprovalsReviewer::User);
codex-rs/core/src/jobs.rs
+323
View File
@@ -0,0 +1,323 @@
//! Runtime-only thread-local job scheduling for follow-on turns.
//!
//! This module owns the in-memory job registry, limited schedule parsing, and
//! the hidden turn context injected when a job fires so the model can act on
//! the scheduled prompt and deschedule itself via `JobDelete(currentJobId)`.
use chrono::DateTime;
use chrono::Duration as ChronoDuration;
use chrono::Utc;
use codex_protocol::models::ContentItem;
use codex_protocol::models::ResponseInputItem;
use serde::Deserialize;
use serde::Serialize;
use std::collections::HashMap;
use tokio_util::sync::CancellationToken;
pub const AFTER_TURN_CRON_EXPRESSION: &str = "@after-turn";
const EVERY_PREFIX: &str = "@every ";
const EVERY_SECONDS_PREFIX: &str = "@every:";
pub const JOB_UPDATED_BACKGROUND_EVENT_PREFIX: &str = "job_updated:";
pub const JOB_FIRED_BACKGROUND_EVENT_PREFIX: &str = "job_fired:";
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ThreadJob {
pub id: String,
pub cron_expression: String,
pub prompt: String,
pub run_once: bool,
pub created_at: i64,
pub next_run_at: Option<i64>,
pub last_run_at: Option<i64>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct JobTurnContext {
pub(crate) current_job_id: String,
pub(crate) prompt: String,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct ClaimedJob {
pub(crate) job: ThreadJob,
pub(crate) context: JobTurnContext,
pub(crate) deleted_run_once_job: bool,
}
#[derive(Debug, Default)]
pub(crate) struct JobsState {
jobs: HashMap<String, JobRuntime>,
}
#[derive(Debug)]
struct JobRuntime {
job: ThreadJob,
schedule: JobSchedule,
pending_run: bool,
timer_cancel: Option<CancellationToken>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum JobSchedule {
AfterTurn,
EverySeconds(u64),
}
impl JobSchedule {
pub(crate) fn parse(cron_expression: &str) -> Result<Self, String> {
if cron_expression == AFTER_TURN_CRON_EXPRESSION {
return Ok(Self::AfterTurn);
}
if let Some(seconds) = cron_expression
.strip_prefix(EVERY_PREFIX)
.map(str::trim)
.and_then(parse_duration_literal)
{
return Ok(Self::EverySeconds(seconds));
}
if let Some(seconds) = cron_expression
.strip_prefix(EVERY_SECONDS_PREFIX)
.and_then(|raw| raw.parse::<u64>().ok())
.filter(|seconds| *seconds > 0)
{
return Ok(Self::EverySeconds(seconds));
}
Err(format!(
"unsupported cron_expression `{cron_expression}`; supported values are `{AFTER_TURN_CRON_EXPRESSION}`, `@every 5m`, or `@every:300`"
))
}
fn next_run_at(self, now: DateTime<Utc>) -> Option<DateTime<Utc>> {
match self {
Self::AfterTurn => None,
Self::EverySeconds(seconds) => Some(now + ChronoDuration::seconds(seconds as i64)),
}
}
}
impl JobsState {
pub(crate) fn list_jobs(&self) -> Vec<ThreadJob> {
let mut jobs = self
.jobs
.values()
.map(|runtime| runtime.job.clone())
.collect::<Vec<_>>();
jobs.sort_by(|left, right| {
left.created_at
.cmp(&right.created_at)
.then_with(|| left.id.cmp(&right.id))
});
jobs
}
pub(crate) fn create_job(
&mut self,
id: String,
cron_expression: String,
prompt: String,
run_once: bool,
now: DateTime<Utc>,
timer_cancel: Option<CancellationToken>,
) -> Result<ThreadJob, String> {
let schedule = JobSchedule::parse(&cron_expression)?;
let job = ThreadJob {
id: id.clone(),
cron_expression,
prompt,
run_once,
created_at: now.timestamp(),
next_run_at: schedule.next_run_at(now).map(|value| value.timestamp()),
last_run_at: None,
};
self.jobs.insert(
id,
JobRuntime {
job: job.clone(),
schedule,
pending_run: matches!(schedule, JobSchedule::AfterTurn),
timer_cancel,
},
);
Ok(job)
}
pub(crate) fn delete_job(&mut self, id: &str) -> bool {
let Some(runtime) = self.jobs.remove(id) else {
return false;
};
if let Some(cancel) = runtime.timer_cancel {
cancel.cancel();
}
true
}
pub(crate) fn mark_after_turn_jobs_due(&mut self) {
for runtime in self.jobs.values_mut() {
if matches!(runtime.schedule, JobSchedule::AfterTurn) {
runtime.pending_run = true;
}
}
}
pub(crate) fn mark_job_due(&mut self, id: &str, now: DateTime<Utc>) {
let Some(runtime) = self.jobs.get_mut(id) else {
return;
};
runtime.pending_run = true;
runtime.job.next_run_at = runtime
.schedule
.next_run_at(now)
.map(|value| value.timestamp());
}
pub(crate) fn claim_next_job(&mut self, now: DateTime<Utc>) -> Option<ClaimedJob> {
let next_job_id = self
.jobs
.values()
.filter(|runtime| runtime.pending_run)
.min_by(|left, right| {
left.job
.created_at
.cmp(&right.job.created_at)
.then_with(|| left.job.id.cmp(&right.job.id))
})
.map(|runtime| runtime.job.id.clone())?;
let runtime = self.jobs.remove(&next_job_id)?;
let JobRuntime {
mut job,
schedule,
pending_run: _,
timer_cancel,
} = runtime;
let deleted_run_once_job = job.run_once;
if deleted_run_once_job {
if let Some(cancel) = timer_cancel {
cancel.cancel();
}
} else {
job.last_run_at = Some(now.timestamp());
self.jobs.insert(
job.id.clone(),
JobRuntime {
job: job.clone(),
schedule,
pending_run: false,
timer_cancel,
},
);
}
Some(ClaimedJob {
job: job.clone(),
context: JobTurnContext {
current_job_id: job.id,
prompt: job.prompt,
},
deleted_run_once_job,
})
}
}
pub(crate) fn job_turn_developer_instructions(job: &JobTurnContext) -> String {
format!(
"This turn was triggered by a thread job.\ncurrentJobId: {}\n\nThe job prompt has already been injected as hidden context for this turn.\nIf you determine the job should stop, call JobDelete with currentJobId. Do not expose scheduler internals unless they matter to the user.",
job.current_job_id
)
}
pub(crate) fn job_prompt_input_item(prompt: &str) -> ResponseInputItem {
ResponseInputItem::Message {
role: "developer".to_string(),
content: vec![ContentItem::InputText {
text: prompt.to_string(),
}],
}
}
fn parse_duration_literal(raw: &str) -> Option<u64> {
let mut digits = String::new();
let mut unit = String::new();
for ch in raw.chars() {
if ch.is_ascii_digit() && unit.is_empty() {
digits.push(ch);
} else if !ch.is_whitespace() {
unit.push(ch);
}
}
let value = digits.parse::<u64>().ok().filter(|value| *value > 0)?;
match unit.as_str() {
"s" | "sec" | "secs" | "second" | "seconds" => Some(value),
"m" | "min" | "mins" | "minute" | "minutes" => value.checked_mul(60),
"h" | "hr" | "hrs" | "hour" | "hours" => value.checked_mul(60 * 60),
"d" | "day" | "days" => value.checked_mul(60 * 60 * 24),
_ => None,
}
}
#[cfg(test)]
mod tests {
use super::AFTER_TURN_CRON_EXPRESSION;
use super::JobSchedule;
use super::JobsState;
use super::job_prompt_input_item;
use chrono::TimeZone;
use chrono::Utc;
use codex_protocol::models::ContentItem;
use codex_protocol::models::ResponseInputItem;
use pretty_assertions::assert_eq;
#[test]
fn parses_supported_job_schedules() {
assert_eq!(
JobSchedule::parse(AFTER_TURN_CRON_EXPRESSION),
Ok(JobSchedule::AfterTurn)
);
assert_eq!(
JobSchedule::parse("@every 5m"),
Ok(JobSchedule::EverySeconds(300))
);
assert_eq!(
JobSchedule::parse("@every:3600"),
Ok(JobSchedule::EverySeconds(3600))
);
}
#[test]
fn claim_run_once_job_removes_it() {
let now = Utc.timestamp_opt(100, 0).single().expect("valid timestamp");
let mut jobs = JobsState::default();
let job = jobs
.create_job(
"job-1".to_string(),
AFTER_TURN_CRON_EXPRESSION.to_string(),
"run tests".to_string(),
true,
now,
None,
)
.expect("job should be created");
assert_eq!(jobs.list_jobs(), vec![job]);
let claimed = jobs.claim_next_job(now).expect("job should be claimed");
assert_eq!(claimed.context.current_job_id, "job-1");
assert!(claimed.deleted_run_once_job);
assert!(jobs.list_jobs().is_empty());
}
#[test]
fn job_prompt_input_is_hidden_developer_input() {
let item = job_prompt_input_item("run tests");
assert_eq!(
item,
ResponseInputItem::Message {
role: "developer".to_string(),
content: vec![ContentItem::InputText {
text: "run tests".to_string(),
}],
}
);
}
}
codex-rs/core/src/lib.rs
+1
View File
@@ -44,6 +44,7 @@ mod git_info_tests;
mod guardian;
mod hook_runtime;
pub mod instructions;
pub mod jobs;
pub mod landlock;
pub mod mcp;
mod mcp_connection_manager;
codex-rs/core/src/shell_snapshot.rs
+1 -1
View File
@@ -29,7 +29,7 @@ pub struct ShellSnapshot {
pub cwd: PathBuf,
}
const SNAPSHOT_TIMEOUT: Duration = Duration::from_secs(10);
const SNAPSHOT_TIMEOUT: Duration = Duration::from_secs(20);
const SNAPSHOT_RETENTION: Duration = Duration::from_secs(60 * 60 * 24 * 3); // 3 days retention.
const SNAPSHOT_DIR: &str = "shell_snapshots";
const EXCLUDED_EXPORT_VARS: &[&str] = &["PWD", "OLDPWD"];
codex-rs/core/src/state/turn.rs
+1 -1
View File
@@ -48,7 +48,7 @@ pub(crate) enum TaskKind {
pub(crate) struct RunningTask {
pub(crate) done: Arc<Notify>,
pub(crate) kind: TaskKind,
pub(crate) task: Arc<dyn SessionTask>,
pub(crate) task: Arc<dyn SessionTask + Send + Sync>,
pub(crate) cancellation_token: CancellationToken,
pub(crate) handle: Arc<AbortOnDropHandle<()>>,
pub(crate) turn_context: Arc<TurnContext>,
codex-rs/core/src/tasks/mod.rs
+4 -2
View File
@@ -172,13 +172,13 @@ impl Session {
self.start_task(turn_context, input, task).await;
}
async fn start_task<T: SessionTask>(
pub(crate) async fn start_task<T: SessionTask>(
self: &Arc<Self>,
turn_context: Arc<TurnContext>,
input: Vec<UserInput>,
task: T,
) {
let task: Arc<dyn SessionTask> = Arc::new(task);
let task: Arc<dyn SessionTask + Send + Sync> = Arc::new(task);
let task_kind = task.kind();
let span_name = task.span_name();
let started_at = Instant::now();
@@ -436,6 +436,8 @@ impl Session {
last_agent_message,
});
self.send_event(turn_context.as_ref(), event).await;
self.mark_after_turn_jobs_due().await;
self.maybe_start_pending_job().await;
}
async fn take_active_turn(&self) -> Option<ActiveTurn> {
codex-rs/core/src/tools/handlers/jobs.rs
+105
View File
@@ -0,0 +1,105 @@
use async_trait::async_trait;
use serde::Deserialize;
use crate::function_tool::FunctionCallError;
use crate::tools::context::FunctionToolOutput;
use crate::tools::context::ToolInvocation;
use crate::tools::context::ToolPayload;
use crate::tools::handlers::parse_arguments;
use crate::tools::registry::ToolHandler;
use crate::tools::registry::ToolKind;
#[derive(Deserialize)]
struct JobCreateArgs {
cron_expression: String,
prompt: String,
run_once: Option<bool>,
}
#[derive(Deserialize)]
struct JobDeleteArgs {
id: String,
}
pub struct JobCreateHandler;
#[async_trait]
impl ToolHandler for JobCreateHandler {
type Output = FunctionToolOutput;
fn kind(&self) -> ToolKind {
ToolKind::Function
}
async fn handle(&self, invocation: ToolInvocation) -> Result<Self::Output, FunctionCallError> {
let ToolPayload::Function { arguments } = invocation.payload else {
return Err(FunctionCallError::RespondToModel(
"JobCreate received unsupported payload".to_string(),
));
};
let args: JobCreateArgs = parse_arguments(&arguments)?;
let job = invocation
.session
.create_job(
args.cron_expression,
args.prompt,
args.run_once.unwrap_or(false),
)
.await
.map_err(FunctionCallError::RespondToModel)?;
let content = serde_json::to_string(&job).map_err(|err| {
FunctionCallError::Fatal(format!("failed to serialize JobCreate response: {err}"))
})?;
Ok(FunctionToolOutput::from_text(content, Some(true)))
}
}
pub struct JobDeleteHandler;
#[async_trait]
impl ToolHandler for JobDeleteHandler {
type Output = FunctionToolOutput;
fn kind(&self) -> ToolKind {
ToolKind::Function
}
async fn handle(&self, invocation: ToolInvocation) -> Result<Self::Output, FunctionCallError> {
let ToolPayload::Function { arguments } = invocation.payload else {
return Err(FunctionCallError::RespondToModel(
"JobDelete received unsupported payload".to_string(),
));
};
let args: JobDeleteArgs = parse_arguments(&arguments)?;
let deleted = invocation.session.delete_job(&args.id).await;
let content = serde_json::json!({ "deleted": deleted }).to_string();
Ok(FunctionToolOutput::from_text(content, Some(deleted)))
}
}
pub struct JobListHandler;
#[async_trait]
impl ToolHandler for JobListHandler {
type Output = FunctionToolOutput;
fn kind(&self) -> ToolKind {
ToolKind::Function
}
async fn handle(&self, invocation: ToolInvocation) -> Result<Self::Output, FunctionCallError> {
match invocation.payload {
ToolPayload::Function { .. } => {}
_ => {
return Err(FunctionCallError::RespondToModel(
"JobList received unsupported payload".to_string(),
));
}
}
let jobs = invocation.session.list_jobs().await;
let content = serde_json::to_string(&jobs).map_err(|err| {
FunctionCallError::Fatal(format!("failed to serialize JobList response: {err}"))
})?;
Ok(FunctionToolOutput::from_text(content, Some(true)))
}
}
codex-rs/core/src/tools/handlers/mod.rs
+4
View File
@@ -1,6 +1,7 @@
pub(crate) mod agent_jobs;
pub mod apply_patch;
mod dynamic;
mod jobs;
mod js_repl;
mod list_dir;
mod mcp;
@@ -37,6 +38,9 @@ pub use apply_patch::ApplyPatchHandler;
use codex_protocol::models::PermissionProfile;
use codex_protocol::protocol::AskForApproval;
pub use dynamic::DynamicToolHandler;
pub use jobs::JobCreateHandler;
pub use jobs::JobDeleteHandler;
pub use jobs::JobListHandler;
pub use js_repl::JsReplHandler;
pub use js_repl::JsReplResetHandler;
pub use list_dir::ListDirHandler;
codex-rs/core/src/tools/handlers/multi_agents_tests.rs
+1 -1
View File
@@ -1074,7 +1074,7 @@ async fn multi_agent_v2_assign_task_interrupts_busy_child_without_losing_message
)
}));
timeout(Duration::from_secs(5), async {
timeout(Duration::from_secs(15), async {
loop {
let history_items = thread
.codex
codex-rs/core/src/tools/spec.rs
+33
View File
@@ -8,6 +8,9 @@ use crate::shell::ShellType;
use crate::tools::code_mode::PUBLIC_TOOL_NAME;
use crate::tools::code_mode::WAIT_TOOL_NAME;
use crate::tools::discoverable::DiscoverableTool;
use crate::tools::handlers::JobCreateHandler;
use crate::tools::handlers::JobDeleteHandler;
use crate::tools::handlers::JobListHandler;
use crate::tools::handlers::PLAN_TOOL;
use crate::tools::handlers::TOOL_SEARCH_DEFAULT_LIMIT;
use crate::tools::handlers::TOOL_SEARCH_TOOL_NAME;
@@ -51,6 +54,9 @@ use codex_tools::create_close_agent_tool_v1;
use codex_tools::create_close_agent_tool_v2;
use codex_tools::create_code_mode_tool;
use codex_tools::create_exec_command_tool;
use codex_tools::create_job_create_tool;
use codex_tools::create_job_delete_tool;
use codex_tools::create_job_list_tool;
use codex_tools::create_js_repl_reset_tool;
use codex_tools::create_js_repl_tool;
use codex_tools::create_list_agents_tool;
@@ -167,6 +173,7 @@ pub(crate) struct ToolsConfig {
pub collab_tools: bool,
pub multi_agent_v2: bool,
pub request_user_input: bool,
pub job_scheduler: bool,
pub default_mode_request_user_input: bool,
pub experimental_supported_tools: Vec<String>,
pub agent_jobs_tools: bool,
@@ -217,6 +224,7 @@ impl ToolsConfig {
let include_multi_agent_v2 = features.enabled(Feature::MultiAgentV2);
let include_agent_jobs = features.enabled(Feature::SpawnCsv);
let include_request_user_input = !matches!(session_source, SessionSource::SubAgent(_));
let include_job_scheduler = features.enabled(Feature::JobScheduler);
let include_default_mode_request_user_input =
include_request_user_input && features.enabled(Feature::DefaultModeRequestUserInput);
let include_search_tool =
@@ -301,6 +309,7 @@ impl ToolsConfig {
collab_tools: include_collab_tools,
multi_agent_v2: include_multi_agent_v2,
request_user_input: include_request_user_input,
job_scheduler: include_job_scheduler,
default_mode_request_user_input: include_default_mode_request_user_input,
experimental_supported_tools: model_info.experimental_supported_tools.clone(),
agent_jobs_tools: include_agent_jobs,
@@ -626,6 +635,30 @@ pub(crate) fn build_specs_with_discoverable_tools(
builder.register_handler("request_permissions", request_permissions_handler);
}
if config.job_scheduler {
push_tool_spec(
&mut builder,
create_job_create_tool(),
/*supports_parallel_tool_calls*/ false,
config.code_mode_enabled,
);
push_tool_spec(
&mut builder,
create_job_delete_tool(),
/*supports_parallel_tool_calls*/ false,
config.code_mode_enabled,
);
push_tool_spec(
&mut builder,
create_job_list_tool(),
/*supports_parallel_tool_calls*/ false,
config.code_mode_enabled,
);
builder.register_handler("JobCreate", Arc::new(JobCreateHandler));
builder.register_handler("JobDelete", Arc::new(JobDeleteHandler));
builder.register_handler("JobList", Arc::new(JobListHandler));
}
if config.search_tool
&& let Some(app_tools) = app_tools
{