Add automation runNow dispatch

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;

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;

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<Option<ClientResponsePayload>, 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<bool, JSONRPCErrorError> {
+        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<RunNowDispatchResult, JSONRPCErrorError> {
+        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<RunNowDispatchResult, JSONRPCErrorError> {
+        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<RunNowDispatchResult, JSONRPCErrorError> {
+        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<String>,
+}
+
+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<Vec<AbsolutePathBuf>, 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#"<heartbeat>
+  <automation_id>{automation_id}</automation_id>
+  <instructions>
+{instructions}
+  </instructions>
+</heartbeat>
+
+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('&', "&amp;")
+        .replace('<', "&lt;")
+        .replace('>', "&gt;")
+}

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<CodexThread>,
@@ -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,

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<TestAppServer> {
     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::<AutomationCreateResponse>(resp)?.automation)
 }
 
+async fn create_heartbeat_automation(
+    app_server: &mut TestAppServer,
+    thread_id: &str,
+) -> Result<Automation> {
+    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::<AutomationCreateResponse>(resp)?.automation)
+}
+
+async fn run_now_automation(
+    app_server: &mut TestAppServer,
+    automation_id: &str,
+    thread_id: Option<&str>,
+) -> Result<AutomationRunNowResponse> {
+    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::<AutomationRunNowResponse>(resp)
+}
+
 async fn list_automations(app_server: &mut TestAppServer) -> Result<AutomationListResponse> {
     let request_id = app_server
         .send_raw_request(