feat(runtime): add channel_send outbox

Author: rexleimo

Cargo.lock

@@ -20,6 +20,11 @@ enum Command {
         #[command(subcommand)]
         command: AgentCommand,
     },
+    /// Outbound channels (outbox + dispatcher)
+    Channel {
+        #[command(subcommand)]
+        command: ChannelCommand,
+    },
     /// Long-running harness helpers (initializer + sessions)
     Harness {
         #[command(subcommand)]
@@ -81,6 +86,25 @@ enum AgentCommand {
     },
 }
 
+#[derive(Debug, clap::Subcommand)]
+enum ChannelCommand {
+    /// Drain queued outbox messages once
+    Drain {
+        /// Max messages to attempt in one run
+        #[arg(long, default_value_t = 50)]
+        limit: usize,
+    },
+    /// Run a long-lived worker that periodically drains the outbox
+    Worker {
+        /// Seconds between drain attempts
+        #[arg(long, default_value_t = 5)]
+        interval_secs: u64,
+        /// Max messages to attempt per drain cycle
+        #[arg(long, default_value_t = 50)]
+        limit: usize,
+    },
+}
+
 #[derive(Debug, Clone, Copy, clap::ValueEnum)]
 enum AgentKind {
     Planning,
@@ -152,8 +176,43 @@ async fn main() -> anyhow::Result<()> {
                 eprintln!("[rexos] session_id={session_id}");
             }
         },
+        Command::Channel { command } => match command {
+            ChannelCommand::Drain { limit } => {
+                let paths = RexosPaths::discover()?;
+                paths.ensure_dirs()?;
+                RexosConfig::ensure_default(&paths)?;
+                MemoryStore::open_or_create(&paths)?;
+
+                let dispatcher =
+                    rexos::agent::OutboxDispatcher::new(MemoryStore::open_or_create(&paths)?)?;
+                let summary = dispatcher.drain_once(limit).await?;
+                println!("drain: sent={} failed={}", summary.sent, summary.failed);
+            }
+            ChannelCommand::Worker {
+                interval_secs,
+                limit,
+            } => {
+                let paths = RexosPaths::discover()?;
+                paths.ensure_dirs()?;
+                RexosConfig::ensure_default(&paths)?;
+                MemoryStore::open_or_create(&paths)?;
+
+                let dispatcher =
+                    rexos::agent::OutboxDispatcher::new(MemoryStore::open_or_create(&paths)?)?;
+
+                loop {
+                    let summary = dispatcher.drain_once(limit).await?;
+                    println!("drain: sent={} failed={}", summary.sent, summary.failed);
+                    tokio::time::sleep(std::time::Duration::from_secs(interval_secs)).await;
+                }
+            }
+        },
         Command::Harness { command } => match command {
-            HarnessCommand::Init { dir, prompt, session } => {
+            HarnessCommand::Init {
+                dir,
+                prompt,
+                session,
+            } => {
                 if prompt.is_none() {
                     rexos::harness::init_workspace(&dir)?;
                     println!("Harness initialized in {}", dir.display());

crates/rexos-cli/src/main.rs

@@ -12,6 +12,7 @@ serde_json.workspace = true
 tokio.workspace = true
 toml.workspace = true
 uuid.workspace = true
+reqwest.workspace = true
 rexos-kernel = { path = "../rexos-kernel" }
 rexos-llm = { path = "../rexos-llm" }
 rexos-memory = { path = "../rexos-memory" }

crates/rexos-runtime/Cargo.toml

@@ -23,6 +23,130 @@ pub struct AgentRuntime {
     router: ModelRouter,
 }
 
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+pub struct OutboxDrainSummary {
+    pub sent: u32,
+    pub failed: u32,
+}
+
+#[derive(Debug)]
+pub struct OutboxDispatcher {
+    memory: MemoryStore,
+    http: reqwest::Client,
+}
+
+impl OutboxDispatcher {
+    pub fn new(memory: MemoryStore) -> anyhow::Result<Self> {
+        let http = reqwest::Client::builder()
+            .timeout(std::time::Duration::from_secs(15))
+            .build()
+            .context("build http client")?;
+        Ok(Self { memory, http })
+    }
+
+    pub async fn drain_once(&self, limit: usize) -> anyhow::Result<OutboxDrainSummary> {
+        let mut msgs = self.outbox_messages_get()?;
+        let mut summary = OutboxDrainSummary::default();
+
+        let mut processed = 0usize;
+        for msg in msgs.iter_mut() {
+            if processed >= limit.max(1) {
+                break;
+            }
+            if msg.status != OutboxStatus::Queued {
+                continue;
+            }
+            processed += 1;
+
+            let now = AgentRuntime::now_epoch_seconds();
+            msg.attempts = msg.attempts.saturating_add(1);
+            msg.updated_at = now;
+            msg.last_error = None;
+
+            let result = match msg.channel.as_str() {
+                "console" => {
+                    self.deliver_console(msg);
+                    Ok(())
+                }
+                "webhook" => self.deliver_webhook(msg).await,
+                other => Err(anyhow::anyhow!("unknown channel: {other}")),
+            };
+
+            match result {
+                Ok(()) => {
+                    msg.status = OutboxStatus::Sent;
+                    msg.sent_at = Some(now);
+                    summary.sent = summary.sent.saturating_add(1);
+                }
+                Err(e) => {
+                    msg.status = OutboxStatus::Failed;
+                    msg.last_error = Some(e.to_string());
+                    summary.failed = summary.failed.saturating_add(1);
+                }
+            }
+        }
+
+        if processed > 0 {
+            self.outbox_messages_set(&msgs)?;
+        }
+
+        Ok(summary)
+    }
+
+    fn outbox_messages_get(&self) -> anyhow::Result<Vec<OutboxMessageRecord>> {
+        let raw = self
+            .memory
+            .kv_get("rexos.outbox.messages")
+            .context("kv_get rexos.outbox.messages")?
+            .unwrap_or_else(|| "[]".to_string());
+        Ok(serde_json::from_str(&raw).unwrap_or_default())
+    }
+
+    fn outbox_messages_set(&self, msgs: &[OutboxMessageRecord]) -> anyhow::Result<()> {
+        let raw = serde_json::to_string(msgs).context("serialize rexos.outbox.messages")?;
+        self.memory
+            .kv_set("rexos.outbox.messages", &raw)
+            .context("kv_set rexos.outbox.messages")?;
+        Ok(())
+    }
+
+    fn deliver_console(&self, msg: &OutboxMessageRecord) {
+        let subject = msg.subject.as_deref().unwrap_or("");
+        println!(
+            "[rexos][channel_send][console] to={} subject={} message={}",
+            msg.recipient, subject, msg.message
+        );
+    }
+
+    async fn deliver_webhook(&self, msg: &OutboxMessageRecord) -> anyhow::Result<()> {
+        let url = std::env::var("REXOS_WEBHOOK_URL")
+            .ok()
+            .filter(|v| !v.trim().is_empty())
+            .ok_or_else(|| anyhow::anyhow!("REXOS_WEBHOOK_URL is not set"))?;
+
+        let payload = serde_json::json!({
+            "message_id": msg.message_id,
+            "recipient": msg.recipient,
+            "subject": msg.subject,
+            "message": msg.message,
+            "created_at": msg.created_at,
+        });
+
+        let resp = self
+            .http
+            .post(url)
+            .json(&payload)
+            .send()
+            .await
+            .context("send webhook request")?;
+
+        if !resp.status().is_success() {
+            bail!("webhook returned http {}", resp.status());
+        }
+        Ok(())
+    }
+}
+
 impl AgentRuntime {
     pub fn new(memory: MemoryStore, llms: LlmRegistry, router: ModelRouter) -> Self {
         Self {
@@ -209,6 +333,11 @@ impl AgentRuntime {
                             serde_json::from_str(&args_json).context("parse cron_cancel args")?;
                         self.cron_cancel(&args.job_id).context("cron_cancel")?
                     }
+                    "channel_send" => {
+                        let args: ChannelSendToolArgs =
+                            serde_json::from_str(&args_json).context("parse channel_send args")?;
+                        self.channel_send(args).context("channel_send")?
+                    }
                     "knowledge_add_entity" => {
                         let args: KnowledgeAddEntityToolArgs = serde_json::from_str(&args_json)
                             .context("parse knowledge_add_entity args")?;
@@ -694,7 +823,9 @@ impl AgentRuntime {
     }
 
     fn cron_create(&self, args: CronCreateToolArgs) -> anyhow::Result<String> {
-        let job_id = args.job_id.unwrap_or_else(|| uuid::Uuid::new_v4().to_string());
+        let job_id = args
+            .job_id
+            .unwrap_or_else(|| uuid::Uuid::new_v4().to_string());
         let mut jobs = self.cron_jobs_get()?;
         if let Some(existing) = jobs.iter().find(|j| j.job_id == job_id) {
             return Ok(serde_json::to_string(existing).unwrap_or_else(|_| "ok".to_string()));
@@ -736,6 +867,71 @@ impl AgentRuntime {
         Ok("ok".to_string())
     }
 
+    fn outbox_messages_get(&self) -> anyhow::Result<Vec<OutboxMessageRecord>> {
+        let key = "rexos.outbox.messages";
+        let raw = self
+            .memory
+            .kv_get(key)
+            .context("kv_get rexos.outbox.messages")?
+            .unwrap_or_else(|| "[]".to_string());
+        let msgs: Vec<OutboxMessageRecord> = serde_json::from_str(&raw).unwrap_or_default();
+        Ok(msgs)
+    }
+
+    fn outbox_messages_set(&self, msgs: &[OutboxMessageRecord]) -> anyhow::Result<()> {
+        let key = "rexos.outbox.messages";
+        let raw = serde_json::to_string(msgs).context("serialize rexos.outbox.messages")?;
+        self.memory
+            .kv_set(key, &raw)
+            .context("kv_set rexos.outbox.messages")?;
+        Ok(())
+    }
+
+    fn channel_send(&self, args: ChannelSendToolArgs) -> anyhow::Result<String> {
+        if args.channel.trim().is_empty() {
+            return Ok("error: channel is empty".to_string());
+        }
+        if args.recipient.trim().is_empty() {
+            return Ok("error: recipient is empty".to_string());
+        }
+        if args.message.trim().is_empty() {
+            return Ok("error: message is empty".to_string());
+        }
+
+        match args.channel.as_str() {
+            "console" | "webhook" => {}
+            other => return Ok(format!("error: unknown channel: {other}")),
+        }
+
+        let now = Self::now_epoch_seconds();
+        let record = OutboxMessageRecord {
+            message_id: uuid::Uuid::new_v4().to_string(),
+            channel: args.channel,
+            recipient: args.recipient,
+            subject: args.subject.filter(|s| !s.trim().is_empty()),
+            message: args.message,
+            status: OutboxStatus::Queued,
+            attempts: 0,
+            last_error: None,
+            created_at: now,
+            updated_at: now,
+            sent_at: None,
+        };
+
+        let mut msgs = self.outbox_messages_get()?;
+        msgs.push(record.clone());
+        if msgs.len() > 500 {
+            msgs.drain(0..(msgs.len() - 500));
+        }
+        self.outbox_messages_set(&msgs)?;
+
+        Ok(serde_json::json!({
+            "status": "queued",
+            "message_id": record.message_id,
+        })
+        .to_string())
+    }
+
     fn knowledge_entities_get(&self) -> anyhow::Result<Vec<KnowledgeEntityRecord>> {
         let key = "rexos.knowledge.entities";
         let raw = self
@@ -1059,6 +1255,41 @@ struct CronCancelToolArgs {
     job_id: String,
 }
 
+#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
+#[serde(rename_all = "snake_case")]
+enum OutboxStatus {
+    Queued,
+    Sent,
+    Failed,
+}
+
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+struct OutboxMessageRecord {
+    message_id: String,
+    channel: String,
+    recipient: String,
+    #[serde(default)]
+    subject: Option<String>,
+    message: String,
+    status: OutboxStatus,
+    attempts: u32,
+    #[serde(default)]
+    last_error: Option<String>,
+    created_at: i64,
+    updated_at: i64,
+    #[serde(default)]
+    sent_at: Option<i64>,
+}
+
+#[derive(Debug, serde::Deserialize)]
+struct ChannelSendToolArgs {
+    channel: String,
+    recipient: String,
+    #[serde(default)]
+    subject: Option<String>,
+    message: String,
+}
+
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 struct KnowledgeEntityRecord {
     id: String,