diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index 6bc41c1a3f..612854ce40 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -8,7 +8,6 @@ use crate::attestation::app_server_attestation_provider; use crate::config_manager::ConfigManager; use crate::connection_rpc_gate::ConnectionRpcGate; use crate::error_code::invalid_request; -use crate::error_code::method_not_found; use crate::extensions::ThreadExtensionDependencies; use crate::extensions::app_server_extension_event_sink; use crate::extensions::guardian_agent_spawner; @@ -1218,8 +1217,10 @@ impl MessageProcessor { .automation_delete(connection_id, params) .await } - ClientRequest::AutomationRunNow { .. } => { - Err(automation_api_not_implemented("automation/runNow")) + ClientRequest::AutomationRunNow { params, .. } => { + self.thread_processor + .automation_run_now(connection_id, params) + .await } ClientRequest::ThreadSearch { params, .. } => { self.thread_processor.thread_search(params).await @@ -1523,10 +1524,6 @@ impl MessageProcessor { } } -fn automation_api_not_implemented(method: &str) -> JSONRPCErrorError { - method_not_found(format!("{method} is not supported yet")) -} - #[cfg(test)] #[path = "message_processor_tracing_tests.rs"] mod message_processor_tracing_tests; diff --git a/codex-rs/app-server/src/request_processors.rs b/codex-rs/app-server/src/request_processors.rs index 62dd49467c..ce892d1490 100644 --- a/codex-rs/app-server/src/request_processors.rs +++ b/codex-rs/app-server/src/request_processors.rs @@ -477,6 +477,7 @@ use codex_app_server_protocol::ServerRequest; mod account_processor; mod apps_processor; +mod automation_dispatch; mod automation_processor; mod catalog_processor; mod command_exec_processor; diff --git a/codex-rs/app-server/src/request_processors/automation_dispatch.rs b/codex-rs/app-server/src/request_processors/automation_dispatch.rs new file mode 100644 index 0000000000..58b4ec0780 --- /dev/null +++ b/codex-rs/app-server/src/request_processors/automation_dispatch.rs @@ -0,0 +1,411 @@ +use std::collections::BTreeMap; +use std::path::Path; +use std::path::PathBuf; + +use crate::error_code::internal_error; +use crate::error_code::invalid_request; +use crate::outgoing_message::ConnectionId; +use crate::request_processors::thread_processor::ThreadRequestProcessor; +use crate::request_processors::thread_processor::build_thread_from_snapshot; +use crate::request_processors::thread_summary::thread_started_notification; +use crate::thread_status::resolve_thread_status; +use codex_app_server_protocol::AutomationRunNowParams; +use codex_app_server_protocol::AutomationRunNowResponse; +use codex_app_server_protocol::ClientResponsePayload; +use codex_app_server_protocol::JSONRPCErrorError; +use codex_app_server_protocol::ServerNotification; +use codex_app_server_protocol::ThreadStatus; +use codex_core::StartThreadOptions; +use codex_core::config::ConfigOverrides; +use codex_extension_api::ExtensionDataInit; +use codex_features::Feature; +use codex_protocol::ThreadId; +use codex_protocol::protocol::InitialHistory; +use codex_protocol::protocol::Op; +use codex_protocol::protocol::SessionSource; +use codex_protocol::protocol::ThreadSettingsOverrides; +use codex_protocol::protocol::ThreadSource; +use codex_protocol::user_input::UserInput; +use codex_rollout::StateDbHandle; +use codex_state::Automation; +use codex_state::AutomationDispatchClaim; +use codex_state::AutomationDispatchOutcome; +use codex_state::AutomationDispatchSettings; +use codex_state::AutomationTarget; +use codex_utils_absolute_path::AbsolutePathBuf; +use uuid::Uuid; + +impl ThreadRequestProcessor { + pub(crate) async fn automation_run_now( + &self, + connection_id: ConnectionId, + params: AutomationRunNowParams, + ) -> Result, JSONRPCErrorError> { + if !self.config.features.enabled(Feature::Automations) { + return Err(invalid_request("automations feature is disabled")); + } + let state_db = self + .state_db + .as_ref() + .ok_or_else(|| internal_error("sqlite state db unavailable for automations"))?; + + let Some(automation) = state_db + .get_automation(params.automation_id.as_str()) + .await + .map_err(|err| internal_error(format!("failed to read automation: {err}")))? + else { + return Ok(Some(run_now_response( + /*found*/ false, /*started_count*/ 0, + ))); + }; + if !self + .connection_can_run_automation(connection_id, &automation, params.thread_id.as_deref()) + .await? + { + return Ok(Some(run_now_response( + /*found*/ false, /*started_count*/ 0, + ))); + } + + let claimed_by = format!("app-server-run-now:{}", Uuid::now_v7()); + let outcome = state_db + .claim_automation_run_now_if_owner( + params.automation_id.as_str(), + automation.owner_thread_id, + claimed_by.as_str(), + ) + .await + .map_err(|err| internal_error(format!("failed to claim automation run: {err}")))?; + let claim = match outcome { + AutomationDispatchOutcome::NotFound => { + return Ok(Some(run_now_response( + /*found*/ false, /*started_count*/ 0, + ))); + } + AutomationDispatchOutcome::AlreadyClaimed => { + return Ok(Some(run_now_response( + /*found*/ true, /*started_count*/ 0, + ))); + } + AutomationDispatchOutcome::Claimed(claim) => claim, + }; + + let dispatch_result = match self.dispatch_automation_run_now(state_db, &claim).await { + Ok(dispatch_result) => dispatch_result, + Err(err) => { + let _ = state_db + .mark_automation_dispatch_failed_terminal( + claim.automation.id.as_str(), + claim.ownership_token.as_str(), + err.message.as_str(), + ) + .await; + return Err(err); + } + }; + let completed = state_db + .mark_automation_dispatch_completed(&claim, dispatch_result.last_error.as_deref()) + .await + .map_err(|err| internal_error(format!("failed to complete automation run: {err}")))?; + if !completed { + return Err(internal_error( + "automation run claim was lost before completion", + )); + } + + Ok(Some(run_now_response( + /*found*/ true, + u32::try_from(dispatch_result.started_count).unwrap_or(u32::MAX), + ))) + } + + async fn connection_can_run_automation( + &self, + connection_id: ConnectionId, + automation: &Automation, + requested_thread_id: Option<&str>, + ) -> Result { + if let Some(requested_thread_id) = requested_thread_id { + let requested_thread_id = + ThreadId::from_string(requested_thread_id).map_err(|err| { + invalid_request(format!("invalid automation runNow threadId: {err}")) + })?; + if requested_thread_id != automation.owner_thread_id { + return Ok(false); + } + } + Ok(self + .thread_state_manager + .thread_ids_for_connection(connection_id) + .await + .into_iter() + .any(|thread_id| thread_id == automation.owner_thread_id)) + } + + async fn dispatch_automation_run_now( + &self, + state_db: &StateDbHandle, + claim: &AutomationDispatchClaim, + ) -> Result { + let started = state_db + .mark_automation_dispatch_started( + claim.automation.id.as_str(), + claim.ownership_token.as_str(), + ) + .await + .map_err(|err| internal_error(format!("failed to start automation dispatch: {err}")))?; + if !started { + return Err(internal_error( + "automation run claim was lost before dispatch", + )); + } + + match &claim.automation.target { + AutomationTarget::Cron { cwds } => self.dispatch_cron_run_now(claim, cwds).await, + AutomationTarget::Heartbeat { thread_id } => { + self.dispatch_heartbeat_run_now(claim, *thread_id).await + } + } + } + + async fn dispatch_cron_run_now( + &self, + claim: &AutomationDispatchClaim, + cwds: &[PathBuf], + ) -> Result { + let mut started_count = 0_usize; + let mut last_error = None; + for cwd in cwds.iter().skip(claim.dispatch_cwd_index) { + match self + .spawn_and_submit_automation_thread(&claim.automation, cwd) + .await + { + Ok(()) => { + started_count += 1; + } + Err(err) => { + last_error = Some(err.message); + } + } + } + if started_count == 0 + && let Some(last_error) = last_error + { + return Err(internal_error(last_error)); + } + Ok(RunNowDispatchResult { + started_count, + last_error, + }) + } + + async fn dispatch_heartbeat_run_now( + &self, + claim: &AutomationDispatchClaim, + thread_id: ThreadId, + ) -> Result { + let loaded_status = self + .thread_watch_manager + .loaded_status_for_thread(&thread_id.to_string()) + .await; + if matches!(loaded_status, ThreadStatus::Active { .. }) { + return Err(invalid_request("heartbeat target thread is busy")); + } + + let thread = self + .thread_manager + .get_thread(thread_id) + .await + .map_err(|_| invalid_request("heartbeat target thread is not loaded"))?; + let thread_state = self.thread_state_manager.thread_state(thread_id).await; + self.ensure_listener_task_running(thread_id, thread.clone(), thread_state) + .await?; + self.submit_automation_prompt(thread.as_ref(), build_heartbeat_prompt(&claim.automation)) + .await?; + Ok(RunNowDispatchResult { + started_count: 1, + last_error: None, + }) + } + + async fn spawn_and_submit_automation_thread( + &self, + automation: &Automation, + cwd: &Path, + ) -> Result<(), JSONRPCErrorError> { + let dispatch_settings = automation + .dispatch_settings + .as_ref() + .ok_or_else(|| invalid_request("cron automation is missing dispatch settings"))?; + ensure_cwd_in_dispatch_scope(cwd, dispatch_settings)?; + + let config = self + .config_manager + .load_for_cwd( + /*request_overrides*/ None, + ConfigOverrides { + cwd: Some(cwd.to_path_buf()), + model: automation.model.clone(), + approval_policy: Some(dispatch_settings.approval_policy), + approvals_reviewer: Some(dispatch_settings.approvals_reviewer), + permission_profile: Some(dispatch_settings.permission_profile.clone()), + workspace_roots: Some(absolute_roots(&dispatch_settings.workspace_roots)?), + ..ConfigOverrides::default() + }, + Some(cwd.to_path_buf()), + ) + .await + .map_err(|err| invalid_request(format!("failed to load automation config: {err}")))?; + let environments = self + .thread_manager + .default_environment_selections(&config.cwd); + let codex_core::NewThread { + thread_id, + thread, + session_configured, + } = self + .thread_manager + .start_thread_with_options(StartThreadOptions { + config, + initial_history: InitialHistory::New, + session_source: Some(SessionSource::automation()), + thread_source: Some(ThreadSource::User), + dynamic_tools: Vec::new(), + metrics_service_name: None, + parent_trace: None, + environments, + thread_extension_init: ExtensionDataInit::default(), + }) + .await + .map_err(|err| internal_error(format!("failed to start automation thread: {err}")))?; + + let thread_state = self.thread_state_manager.thread_state(thread_id).await; + self.ensure_listener_task_running(thread_id, thread.clone(), thread_state) + .await?; + let config_snapshot = thread.config_snapshot().await; + let mut api_thread = build_thread_from_snapshot( + thread_id, + session_configured.session_id.to_string(), + &config_snapshot, + session_configured.rollout_path.clone(), + ); + self.thread_watch_manager + .upsert_thread_silently(api_thread.clone()) + .await; + api_thread.status = resolve_thread_status( + self.thread_watch_manager + .loaded_status_for_thread(api_thread.id.as_str()) + .await, + /*has_in_progress_turn*/ false, + ); + self.outgoing + .send_server_notification(ServerNotification::ThreadStarted( + thread_started_notification(api_thread), + )) + .await; + + self.submit_automation_prompt(thread.as_ref(), build_cron_prompt(automation)) + .await + } + + async fn submit_automation_prompt( + &self, + thread: &codex_core::CodexThread, + prompt: String, + ) -> Result<(), JSONRPCErrorError> { + thread + .submit(Op::UserInput { + items: vec![UserInput::Text { + text: prompt, + text_elements: Vec::new(), + }], + final_output_json_schema: None, + responsesapi_client_metadata: None, + additional_context: BTreeMap::new(), + thread_settings: ThreadSettingsOverrides::default(), + }) + .await + .map(|_| ()) + .map_err(|err| internal_error(format!("failed to submit automation prompt: {err}"))) + } +} + +fn run_now_response(found: bool, started_count: u32) -> ClientResponsePayload { + AutomationRunNowResponse { + found, + started_count, + } + .into() +} + +struct RunNowDispatchResult { + started_count: usize, + last_error: Option, +} + +fn ensure_cwd_in_dispatch_scope( + cwd: &Path, + dispatch_settings: &AutomationDispatchSettings, +) -> Result<(), JSONRPCErrorError> { + if dispatch_settings + .workspace_roots + .iter() + .any(|root| cwd.starts_with(root)) + { + Ok(()) + } else { + Err(invalid_request( + "automation cwd is outside its approved workspace roots", + )) + } +} + +fn absolute_roots(paths: &[PathBuf]) -> Result, JSONRPCErrorError> { + paths + .iter() + .map(|path| { + AbsolutePathBuf::from_absolute_path(path).map_err(|err| { + invalid_request(format!( + "automation dispatch root must be absolute: {} ({err})", + path.display() + )) + }) + }) + .collect() +} + +fn build_cron_prompt(automation: &Automation) -> String { + let last_run_label = automation + .last_run_at + .map(|value| format!("{} ({})", value.to_rfc3339(), value.timestamp())) + .unwrap_or_else(|| "never".to_string()); + format!( + "Automation: {name}\nAutomation ID: {id}\nAutomation memory: $CODEX_HOME/automations/{id}/memory.md\nLast run: {last_run_label}\n\n{prompt}", + name = automation.name, + id = automation.id, + prompt = automation.prompt, + ) +} + +fn build_heartbeat_prompt(automation: &Automation) -> String { + format!( + r#" + {automation_id} + +{instructions} + + + +This heartbeat automation is being delivered over the normal user-input channel. +"#, + automation_id = automation.id, + instructions = escape_xml_text(&automation.prompt), + ) +} + +fn escape_xml_text(input: &str) -> String { + input + .replace('&', "&") + .replace('<', "<") + .replace('>', ">") +} diff --git a/codex-rs/app-server/src/request_processors/thread_processor.rs b/codex-rs/app-server/src/request_processors/thread_processor.rs index a5299f2c72..541db6c082 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -853,7 +853,7 @@ impl ThreadRequestProcessor { .await } - async fn ensure_listener_task_running( + pub(super) async fn ensure_listener_task_running( &self, conversation_id: ThreadId, conversation: Arc, @@ -4327,7 +4327,7 @@ fn permission_profile_trusts_project( } } -fn build_thread_from_snapshot( +pub(super) fn build_thread_from_snapshot( thread_id: ThreadId, session_id: String, config_snapshot: &ThreadConfigSnapshot, diff --git a/codex-rs/app-server/tests/suite/v2/automation.rs b/codex-rs/app-server/tests/suite/v2/automation.rs index d9044d59df..92c07194a4 100644 --- a/codex-rs/app-server/tests/suite/v2/automation.rs +++ b/codex-rs/app-server/tests/suite/v2/automation.rs @@ -10,6 +10,8 @@ use codex_app_server_protocol::AutomationDeleteResponse; use codex_app_server_protocol::AutomationListParams; use codex_app_server_protocol::AutomationListResponse; use codex_app_server_protocol::AutomationReadResponse; +use codex_app_server_protocol::AutomationRunNowParams; +use codex_app_server_protocol::AutomationRunNowResponse; use codex_app_server_protocol::AutomationStatus; use codex_app_server_protocol::AutomationTarget; use codex_app_server_protocol::AutomationUpdateResponse; @@ -53,6 +55,23 @@ async fn automation_api_rejects_disabled_feature() -> Result<()> { assert_eq!(err.error.code, -32600); assert_eq!(err.error.message, "automations feature is disabled"); + let request_id = app_server + .send_raw_request( + "automation/runNow", + Some(serde_json::to_value(AutomationRunNowParams { + automation_id: "automation-missing".to_string(), + thread_id: None, + })?), + ) + .await?; + let err: JSONRPCError = timeout( + DEFAULT_READ_TIMEOUT, + app_server.read_stream_until_error_message(RequestId::Integer(request_id)), + ) + .await??; + assert_eq!(err.error.code, -32600); + assert_eq!(err.error.message, "automations feature is disabled"); + Ok(()) } @@ -108,6 +127,40 @@ async fn automation_crud_is_scoped_to_subscribed_threads() -> Result<()> { Ok(()) } +#[tokio::test] +async fn automation_run_now_dispatches_loaded_heartbeat() -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; + let codex_home = TempDir::new()?; + write_config( + codex_home.path(), + &server.uri(), + /*automations_enabled*/ true, + )?; + let cwd = codex_home.path().join("workspace"); + std::fs::create_dir(&cwd)?; + + let mut app_server = init_app_server(codex_home.path()).await?; + let thread = start_thread(&mut app_server, &cwd).await?.thread; + let automation = create_heartbeat_automation(&mut app_server, thread.id.as_str()).await?; + + let run_now = + run_now_automation(&mut app_server, automation.id.as_str(), Some(&thread.id)).await?; + assert_eq!( + run_now, + AutomationRunNowResponse { + found: true, + started_count: 1, + } + ); + timeout( + DEFAULT_READ_TIMEOUT, + app_server.read_stream_until_notification_message("turn/completed"), + ) + .await??; + + Ok(()) +} + async fn init_app_server(codex_home: &Path) -> Result { let mut app_server = TestAppServer::new(codex_home).await?; timeout(DEFAULT_READ_TIMEOUT, app_server.initialize()).await??; @@ -159,6 +212,48 @@ async fn create_cron_automation(app_server: &mut TestAppServer, cwd: &Path) -> R Ok(to_response::(resp)?.automation) } +async fn create_heartbeat_automation( + app_server: &mut TestAppServer, + thread_id: &str, +) -> Result { + let request_id = app_server + .send_raw_request( + "automation/create", + Some(serde_json::to_value(AutomationCreateParams { + name: "gentle nudge".to_string(), + prompt: "check whether anything needs attention".to_string(), + rrule: Some("FREQ=MINUTELY;INTERVAL=30".to_string()), + model: None, + reasoning_effort: None, + status: Some(AutomationStatus::Active), + target: AutomationTarget::Heartbeat { + thread_id: thread_id.to_string(), + }, + })?), + ) + .await?; + let resp = response_for(app_server, request_id).await?; + Ok(to_response::(resp)?.automation) +} + +async fn run_now_automation( + app_server: &mut TestAppServer, + automation_id: &str, + thread_id: Option<&str>, +) -> Result { + let request_id = app_server + .send_raw_request( + "automation/runNow", + Some(serde_json::to_value(AutomationRunNowParams { + automation_id: automation_id.to_string(), + thread_id: thread_id.map(ToString::to_string), + })?), + ) + .await?; + let resp = response_for(app_server, request_id).await?; + to_response::(resp) +} + async fn list_automations(app_server: &mut TestAppServer) -> Result { let request_id = app_server .send_raw_request(