feat(scheduled_action): support remote agents as a schedulable ActionKind

apps/web/src/features/block-automation/component/automationUtils.ts

@@ -255,9 +255,11 @@ export function createEmptyDraft(): ScheduleDraft {
   };
 }
 
-function getAgentTask(schedule: ScheduledAction): AgentTask {
-  // Backend stores task as a JSON object; for kind === "Agent" it is shaped
-  // like AgentTask. Cast through unknown to satisfy the open-ended type.
+function getAgentTask(schedule: ScheduledAction): AgentTask | null {
+  // Backend stores task as a JSON object and `kind` is its discriminant: only
+  // kind === "Agent" is shaped like AgentTask. Other kinds (RemoteAgent) carry
+  // a different payload, so the cast is only sound behind this check.
+  if (schedule.kind !== 'Agent') return null;
   return schedule.task as unknown as AgentTask;
 }
 
@@ -268,12 +270,12 @@ export function draftFromSchedule(schedule: ScheduledAction): ScheduleDraft {
   return {
     id: schedule.id ?? undefined,
     name: schedule.name,
-    prompt: task.user_prompt ?? '',
+    prompt: task?.user_prompt ?? '',
     frequency: parsed.frequency,
     time: parsed.time,
     daysOfWeek: parsed.daysOfWeek,
     dayOfMonth: parsed.dayOfMonth,
-    model: (task.model as Model) ?? undefined,
+    model: (task?.model as Model) ?? undefined,
     enabled: schedule.enabled,
   };
 }
@@ -304,9 +306,16 @@ export function draftToUpdateBody(
   return {
     name: draft.name.trim() || deriveScheduleName(draft.prompt),
     schedule: buildCron(draft),
-    kind: 'Agent',
+    // Preserve the stored kind and, for kinds this editor cannot build, the
+    // stored task: an update is a full overwrite, so writing an Agent-shaped
+    // task over a RemoteAgent action would drop its endpoint_url. Name,
+    // schedule and enabled still apply to every kind.
+    kind: previous.kind,
     timezone: previous.timezone || getDefaultTimezone(),
-    task: buildAgentTask(draft) as unknown as UpdateScheduledAction['task'],
+    task:
+      previous.kind === 'Agent'
+        ? (buildAgentTask(draft) as unknown as UpdateScheduledAction['task'])
+        : previous.task,
     enabled: draft.enabled,
   };
 }

services/scheduled_action/Cargo.toml

@@ -54,6 +54,7 @@ model = { path = "../../crates/model" }
 model-entity = { path = "../../crates/model-entity" }
 model_notifications = { path = "../../crates/model_notifications" }
 notification = { path = "../../crates/notification" }
+reqwest = { workspace = true, features = ["stream"] }
 utoipa = { workspace = true, features = ["axum_extras", "chrono"] }
 utoipa-swagger-ui = { workspace = true }
 

services/scheduled_action/src/bins/service.rs

@@ -26,6 +26,7 @@ use scheduled_action::outbound::pg_polling_dispatcher::{
     PgPollingDispatcher, PgPollingDispatcherLifecycle,
 };
 use scheduled_action::outbound::pg_scheduled_action_repo::PgScheduledActionRepo;
+use scheduled_action::outbound::remote_agent_http::ReqwestRemoteAgentClient;
 use scheduled_action::swagger::ApiDoc;
 use sqlx::postgres::PgPoolOptions;
 use tokio_util::{sync::CancellationToken, task::TaskTracker};
@@ -76,6 +77,10 @@ async fn main() -> Result<()> {
 
     let repo = Arc::new(PgScheduledActionRepo::new(db.clone()));
 
+    let remote_agent_client = Arc::new(
+        ReqwestRemoteAgentClient::new().context("failed to build remote agent http client")?,
+    );
+
     // The dispatcher consumes its executor, so build a second executor for the
     // service to use when handling execute-now requests. Both executors share
     // the underlying repo/pool/tool-context via cheap Arc/PgPool clones.
@@ -85,13 +90,15 @@ async fn main() -> Result<()> {
         tool_context.clone(),
         Arc::clone(&notification_ingress),
         Arc::clone(&live_updates),
+        Arc::clone(&remote_agent_client),
     );
     let service_executor = Arc::new(InProcessExecutor::new(
         Arc::clone(&repo),
         db.clone(),
         tool_context,
         notification_ingress,
         live_updates,
+        remote_agent_client,
     ));
 
     let dispatcher_cancellation_token = CancellationToken::new();

services/scheduled_action/src/domain/models.rs

@@ -51,7 +51,11 @@ impl<'de> Deserialize<'de> for Schedule {
 
 #[derive(Debug, Serialize, Deserialize, Clone, ToSchema)]
 pub enum ActionKind {
+    /// Executed by Macro's own agent loop against a Macro model.
     Agent,
+    /// Executed by an agent running outside Macro, reached over HTTPS. Macro
+    /// remains the scheduler, the system of record and the UI.
+    RemoteAgent,
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone, ToSchema)]
@@ -61,6 +65,56 @@ pub struct AgentTask {
     pub user_prompt: String,
 }
 
+/// Label attributed to run messages when a [`RemoteAgentTask`] does not name
+/// the agent. Stored on the chat message in place of a Macro model name.
+pub const DEFAULT_REMOTE_AGENT_LABEL: &str = "remote-agent";
+
+/// Task for an agent executing outside Macro, stored in the untyped `task`
+/// column of a [`ScheduledAction`] whose kind is [`ActionKind::RemoteAgent`].
+#[derive(Serialize, Deserialize, Debug, Clone, ToSchema)]
+pub struct RemoteAgentTask {
+    /// HTTPS endpoint invoked on each run.
+    pub endpoint_url: String,
+    /// Opaque prompt forwarded to the remote agent.
+    pub user_prompt: String,
+    /// Optional label shown in place of a model name.
+    #[serde(default)]
+    pub agent_label: Option<String>,
+}
+
+impl RemoteAgentTask {
+    /// Label to attribute this task's chat messages to.
+    pub fn label(&self) -> &str {
+        match self.agent_label.as_deref() {
+            Some(label) if !label.trim().is_empty() => label,
+            _ => DEFAULT_REMOTE_AGENT_LABEL,
+        }
+    }
+}
+
+/// Payload POSTed to a remote agent endpoint on each run. The remote agent is
+/// given the identifiers of the run so it can correlate its own logs; it is not
+/// expected to write to Macro itself.
+#[derive(Serialize, Deserialize, Debug, Clone, ToSchema)]
+pub struct RemoteAgentRunRequest {
+    /// Scheduled action being run.
+    #[schema(value_type = String, format = Uuid)]
+    pub action_id: Uuid,
+    /// Chat the run transcript is written to.
+    pub chat_id: String,
+    /// Name of the scheduled action, as shown in the UI.
+    pub action_name: String,
+    /// Prompt configured on the action.
+    pub user_prompt: String,
+}
+
+/// Response a remote agent endpoint is expected to return.
+#[derive(Serialize, Deserialize, Debug, Clone, ToSchema)]
+pub struct RemoteAgentRunResponse {
+    /// Text written back into the run chat as the assistant reply.
+    pub output: String,
+}
+
 /// Client-supplied payload for creating a scheduled action. The server fills
 /// in `id`, `owner` (from the authenticated user), timestamps, `claimed`, and
 /// `next_run_at` (derived from the cron).

services/scheduled_action/src/domain/ports.rs

@@ -1,6 +1,6 @@
 use super::models::{
-    ActionExecutionRecord, DispatchEvent, InProgressExecution, ScheduledAction,
-    ScheduledActionUpdate,
+    ActionExecutionRecord, DispatchEvent, InProgressExecution, RemoteAgentRunRequest,
+    RemoteAgentRunResponse, RemoteAgentTask, ScheduledAction, ScheduledActionUpdate,
 };
 use anyhow::Result;
 use chrono::{DateTime, Utc};
@@ -111,3 +111,14 @@ pub trait ScheduledActionExecutor {
 pub trait ScheduledActionLiveUpdate: Send + Sync + 'static {
     fn publish_update(&self, update: ScheduledActionUpdate) -> impl Future<Output = ()> + Send;
 }
+
+/// Invokes an agent that runs outside Macro. Implementations own transport
+/// concerns — endpoint validation, timeouts, redirect policy and any future
+/// request signing — so the executor stays transport-agnostic.
+pub trait RemoteAgentClient: Send + Sync + 'static {
+    fn run(
+        &self,
+        task: &RemoteAgentTask,
+        request: &RemoteAgentRunRequest,
+    ) -> impl Future<Output = Result<RemoteAgentRunResponse>> + Send;
+}

services/scheduled_action/src/outbound.rs

@@ -2,4 +2,5 @@ pub mod conn_gateway_live_updates;
 pub mod inprocess_executor;
 pub mod pg_polling_dispatcher;
 pub mod pg_scheduled_action_repo;
+pub mod remote_agent_http;
 pub mod tokio_dispatcher;

services/scheduled_action/src/outbound/inprocess_executor/mod.rs

@@ -1,5 +1,6 @@
 mod agent_task;
 mod notify;
+mod remote_agent_task;
 
 use std::sync::Arc;
 
@@ -16,31 +17,40 @@ use crate::domain::models::{
     ScheduledAction, ScheduledActionUpdate,
 };
 use crate::domain::ports::{
-    ScheduledActionExecutor, ScheduledActionLiveUpdate, ScheduledActionRepo,
+    RemoteAgentClient, ScheduledActionExecutor, ScheduledActionLiveUpdate, ScheduledActionRepo,
 };
 
-pub struct InProcessExecutor<Rpo: ScheduledActionRepo, Live: ScheduledActionLiveUpdate> {
+pub struct InProcessExecutor<
+    Rpo: ScheduledActionRepo,
+    Live: ScheduledActionLiveUpdate,
+    Remote: RemoteAgentClient,
+> {
     repo: Arc<Rpo>,
     db: PgPool,
     tool_context: ToolServiceContext,
     notification_ingress: Arc<SqsNotificationIngress<SqsQueue>>,
     live_updates: Arc<Live>,
+    remote_agent_client: Arc<Remote>,
 }
 
-impl<Rpo: ScheduledActionRepo, Live: ScheduledActionLiveUpdate> InProcessExecutor<Rpo, Live> {
+impl<Rpo: ScheduledActionRepo, Live: ScheduledActionLiveUpdate, Remote: RemoteAgentClient>
+    InProcessExecutor<Rpo, Live, Remote>
+{
     pub fn new(
         repo: Arc<Rpo>,
         db: PgPool,
         tool_context: ToolServiceContext,
         notification_ingress: Arc<SqsNotificationIngress<SqsQueue>>,
         live_updates: Arc<Live>,
+        remote_agent_client: Arc<Remote>,
     ) -> Self {
         Self {
             repo,
             db,
             tool_context,
             notification_ingress,
             live_updates,
+            remote_agent_client,
         }
     }
 }
@@ -57,10 +67,11 @@ fn try_claim(action: &ScheduledAction) -> Result<()> {
     Ok(())
 }
 
-impl<Rpo, Live> ScheduledActionExecutor for InProcessExecutor<Rpo, Live>
+impl<Rpo, Live, Remote> ScheduledActionExecutor for InProcessExecutor<Rpo, Live, Remote>
 where
     Rpo: ScheduledActionRepo + Send + Sync + 'static,
     Live: ScheduledActionLiveUpdate,
+    Remote: RemoteAgentClient,
 {
     async fn execute_action(&self, action: ScheduledAction) -> Result<InProgressExecution> {
         try_claim(&action)?;
@@ -70,8 +81,29 @@ where
 
         // Create the chat up front so the caller gets a chat_id synchronously
         // and the eventual execution record can link back to it.
-        let chat_id = match action.kind {
-            ActionKind::Agent => agent_task::create_run_chat(&self.db, &action).await?,
+        // Both kinds write their run transcript to a chat, so the UI (which
+        // navigates by chat_id) is identical for a remote agent.
+        let created_chat = match action.kind {
+            ActionKind::Agent | ActionKind::RemoteAgent => {
+                agent_task::create_run_chat(&self.db, &action).await
+            }
+        };
+
+        // The claim is taken before setup, and only the spawned run releases
+        // it. Failing here without releasing would leave the action unrunnable
+        // until the claim goes stale (MAX_ACTION_TIME).
+        let chat_id = match created_chat {
+            Ok(chat_id) => chat_id,
+            Err(e) => {
+                if let Err(release_error) = self.repo.release_action(&id).await {
+                    tracing::error!(
+                        error=?release_error,
+                        action_id=?id,
+                        "failed to release action claim after run setup failed"
+                    );
+                }
+                return Err(e);
+            }
         };
 
         self.live_updates
@@ -92,11 +124,19 @@ where
         let tool_context = self.tool_context.clone();
         let notification_ingress = Arc::clone(&self.notification_ingress);
         let live_updates = Arc::clone(&self.live_updates);
+        let remote_agent_client = Arc::clone(&self.remote_agent_client);
         let start_time = Utc::now();
         let record_resource_id = chat_id.clone();
         tokio::spawn(async move {
-            let result =
-                run_job(&db, &tool_context, &notification_ingress, &action, &chat_id).await;
+            let result = run_job(
+                &db,
+                &tool_context,
+                &notification_ingress,
+                remote_agent_client.as_ref(),
+                &action,
+                &chat_id,
+            )
+            .await;
             let end_time = Utc::now();
             let is_success = result.is_ok();
 
@@ -154,10 +194,11 @@ where
     }
 }
 
-async fn run_job(
+async fn run_job<Remote: RemoteAgentClient>(
     db: &PgPool,
     tool_context: &ToolServiceContext,
     notification_ingress: &Arc<SqsNotificationIngress<SqsQueue>>,
+    remote_agent_client: &Remote,
     action: &ScheduledAction,
     chat_id: &str,
 ) -> Result<()> {
@@ -167,5 +208,16 @@ async fn run_job(
                 .await?;
             Ok(())
         }
+        ActionKind::RemoteAgent => {
+            remote_agent_task::run_remote_agent_task(
+                db,
+                remote_agent_client,
+                notification_ingress,
+                action,
+                chat_id,
+            )
+            .await?;
+            Ok(())
+        }
     }
 }