fix(cron): route reminders from origin channel, gate announce by allowed_users (#1026)

src/openhuman/channels/runtime/dispatch.rs

@@ -44,6 +44,41 @@ fn starts_with_any(s: &str, prefixes: &[&str]) -> bool {
     prefixes.iter().any(|p| s.starts_with(p))
 }
 
+/// Build the per-turn `[Channel context]` block prepended to the user
+/// message for non-web inbound channels (e.g. Telegram, Discord, Slack).
+///
+/// Surfaces the active channel and reply target so the model knows
+/// where it is talking and can route any tool side-effects (notably
+/// `cron_add`) back to the same chat instead of defaulting to the
+/// in-app web stream. See issue #928.
+///
+/// Returns an empty string for web/cli turns (the desktop UI is the
+/// default delivery surface, no hint needed).
+fn build_channel_context_block(msg: &traits::ChannelMessage) -> String {
+    let channel = msg.channel.trim();
+    if channel.is_empty()
+        || channel.eq_ignore_ascii_case("web")
+        || channel.eq_ignore_ascii_case("cli")
+    {
+        return String::new();
+    }
+
+    let reply_target = msg.reply_target.trim();
+    if reply_target.is_empty() {
+        return String::new();
+    }
+
+    format!(
+        "[Channel context]\n\
+         You are responding via the \"{channel}\" channel. Reply target: \"{reply_target}\".\n\
+         For any cron/scheduled reminder you create with `cron_add`, set `delivery` to \
+         `{{ \"mode\": \"announce\", \"channel\": \"{channel}\", \"to\": \"{reply_target}\" }}` \
+         so the reminder is delivered back here instead of the in-app web stream. \
+         Only fall back to the default proactive delivery if the user explicitly asks for \
+         in-app/desktop notification.\n\n"
+    )
+}
+
 /// Pick a contextual acknowledgment emoji for an inbound message.
 ///
 /// Intent categories are checked in priority order. Within each category two
@@ -764,10 +799,12 @@ pub(crate) async fn process_channel_message(
             .await;
     }
 
-    let enriched_message = if memory_context.is_empty() {
-        msg.content.clone()
-    } else {
-        format!("{memory_context}{}", msg.content)
+    let channel_context = build_channel_context_block(&msg);
+    let enriched_message = match (memory_context.is_empty(), channel_context.is_empty()) {
+        (true, true) => msg.content.clone(),
+        (false, true) => format!("{memory_context}{}", msg.content),
+        (true, false) => format!("{channel_context}{}", msg.content),
+        (false, false) => format!("{memory_context}{channel_context}{}", msg.content),
     };
 
     println!("  ⏳ Processing message...");
@@ -1309,4 +1346,53 @@ mod tests {
         // Single "?" falls into question category (contains '?').
         assert!(is_in(r, &["🤔", "✍️"]));
     }
+
+    // ── build_channel_context_block (#928) ───────────────────────
+
+    fn cm(channel: &str, reply_target: &str) -> traits::ChannelMessage {
+        traits::ChannelMessage {
+            channel: channel.into(),
+            sender: "alice".into(),
+            content: "hi".into(),
+            id: "m1".into(),
+            reply_target: reply_target.into(),
+            thread_ts: None,
+            timestamp: 0,
+        }
+    }
+
+    #[test]
+    fn channel_context_block_omitted_for_web_and_cli() {
+        assert!(build_channel_context_block(&cm("web", "1")).is_empty());
+        assert!(build_channel_context_block(&cm("cli", "1")).is_empty());
+        assert!(build_channel_context_block(&cm("WEB", "1")).is_empty());
+        assert!(build_channel_context_block(&cm("", "1")).is_empty());
+    }
+
+    #[test]
+    fn channel_context_block_omitted_when_reply_target_missing() {
+        assert!(build_channel_context_block(&cm("telegram", "")).is_empty());
+        assert!(build_channel_context_block(&cm("telegram", "   ")).is_empty());
+    }
+
+    #[test]
+    fn channel_context_block_for_telegram_includes_routing_hint() {
+        let block = build_channel_context_block(&cm("telegram", "123456"));
+        assert!(block.contains("[Channel context]"));
+        assert!(block.contains("\"telegram\""));
+        assert!(block.contains("\"123456\""));
+        // Hint must steer the model toward announce mode with the same channel/target.
+        assert!(block.contains("announce"));
+        assert!(block.contains("cron_add"));
+    }
+
+    #[test]
+    fn channel_context_block_for_discord_and_slack_share_shape() {
+        for ch in ["discord", "slack", "matrix"] {
+            let block = build_channel_context_block(&cm(ch, "chan-42"));
+            assert!(block.contains(ch), "missing channel name in `{ch}` block");
+            assert!(block.contains("chan-42"));
+            assert!(block.contains("announce"));
+        }
+    }
 }

src/openhuman/cron/README.md

@@ -28,7 +28,30 @@ Scheduled-job runtime. Owns cron-expression and human-delay parsing, the persist
 - `src/core/all.rs` — controller registry wires `all_cron_*`.
 - Channel and agent runtimes consume `Cron` events via the bus.
 
+## Delivery modes
+
+A cron job's `DeliveryConfig.mode` decides where its output ends up:
+
+- **`proactive`** (default for agent jobs) — `deliver_if_configured` publishes
+  `DomainEvent::ProactiveMessageRequested`. The proactive subscriber
+  (`channels::proactive`) always pushes to the in-app web stream and additionally
+  mirrors to `channels_config.active_channel` when set. Use for jobs whose
+  natural surface is the desktop UI (briefings, app-pushed notifications).
+- **`announce`** — explicit channel-targeted delivery. Requires `channel` and
+  `to`; publishes `DomainEvent::CronDeliveryRequested` and lands only in that
+  channel. The agent layer should pick this mode when a cron is created from a
+  non-web channel (Telegram, Discord, Slack, …) so the reminder ends up where
+  the user asked for it. The `cron_add` tool validates `to` against the
+  channel's `allowed_users` to reject cross-tenant targets.
+- **`none`** — silent; output is stored in `last_output` only.
+
+The `[Channel context]` block injected by `channels::runtime::dispatch` for
+non-web inbound turns instructs the model to default to `announce` with the
+current channel + reply target — that is the routing path for the Telegram
+"remind me to drink water" use case in #928.
+
 ## Tests
 
 - Unit: `ops_tests.rs`, `scheduler_tests.rs`, `store_tests.rs`.
 - Schema/parsing coverage lives inside `schedule.rs` and `schemas.rs` `#[cfg(test)] mod tests` blocks.
+- Delivery validation: `tools::impl::cron::add::tests` (announce-mode `allowed_users` checks).

src/openhuman/tools/impl/cron/add.rs

@@ -6,6 +6,80 @@ use async_trait::async_trait;
 use serde_json::json;
 use std::sync::Arc;
 
+/// Look up the configured `allowed_users` list for a channel by name.
+/// Returns `None` if the channel is unknown or unconfigured. An empty
+/// `Some(&[])` means the channel is configured but accepts any sender.
+fn allowed_users_for_channel<'a>(config: &'a Config, channel: &str) -> Option<&'a [String]> {
+    let ch = channel.trim().to_ascii_lowercase();
+    let cc = &config.channels_config;
+    match ch.as_str() {
+        "telegram" => cc.telegram.as_ref().map(|c| c.allowed_users.as_slice()),
+        "discord" => cc.discord.as_ref().map(|c| c.allowed_users.as_slice()),
+        "slack" => cc.slack.as_ref().map(|c| c.allowed_users.as_slice()),
+        "mattermost" => cc.mattermost.as_ref().map(|c| c.allowed_users.as_slice()),
+        "matrix" => cc.matrix.as_ref().map(|c| c.allowed_users.as_slice()),
+        "irc" => cc.irc.as_ref().map(|c| c.allowed_users.as_slice()),
+        "lark" => cc.lark.as_ref().map(|c| c.allowed_users.as_slice()),
+        "dingtalk" => cc.dingtalk.as_ref().map(|c| c.allowed_users.as_slice()),
+        "qq" => cc.qq.as_ref().map(|c| c.allowed_users.as_slice()),
+        _ => None,
+    }
+}
+
+/// Validate a `DeliveryConfig` at cron-create time.
+///
+/// For `mode: "announce"` we require both `channel` and `to`, and we
+/// reject `to` values that are not in the channel's configured
+/// `allowed_users` list. This blocks an LLM (or RPC caller) from
+/// scheduling a cron whose output gets sent to an arbitrary chat id —
+/// see the "no cross-tenant `to`" acceptance criterion in #928.
+///
+/// `proactive` and `none` modes are not channel-targeted and are not
+/// validated here.
+fn validate_delivery(config: &Config, delivery: &DeliveryConfig) -> Result<(), String> {
+    let mode = delivery.mode.trim().to_ascii_lowercase();
+    if mode != "announce" {
+        return Ok(());
+    }
+
+    let channel = delivery
+        .channel
+        .as_deref()
+        .map(str::trim)
+        .filter(|s| !s.is_empty())
+        .ok_or_else(|| "delivery.channel is required for announce mode".to_string())?;
+    let to = delivery
+        .to
+        .as_deref()
+        .map(str::trim)
+        .filter(|s| !s.is_empty())
+        .ok_or_else(|| "delivery.to is required for announce mode".to_string())?;
+
+    // "web" announce is a degenerate case (web has no allowed_users
+    // gate). Other unknown channels (e.g. "email") fall through to the
+    // generic reject.
+    if channel.eq_ignore_ascii_case("web") {
+        return Ok(());
+    }
+
+    match allowed_users_for_channel(config, channel) {
+        Some(list) if list.is_empty() => Ok(()),
+        Some(list) => {
+            if list.iter().any(|u| u == to) {
+                Ok(())
+            } else {
+                Err(format!(
+                    "delivery target '{to}' on channel '{channel}' is not in allowed_users \
+                     for that channel; refusing to schedule cross-tenant delivery"
+                ))
+            }
+        }
+        None => Err(format!(
+            "delivery channel '{channel}' is not configured; cannot validate target"
+        )),
+    }
+}
+
 pub struct CronAddTool {
     config: Arc<Config>,
     security: Arc<SecurityPolicy>,
@@ -24,7 +98,15 @@ impl Tool for CronAddTool {
     }
 
     fn description(&self) -> &str {
-        "Create a scheduled cron job (shell or agent) with cron/at/every schedules. Standardizes on device-local timezone unless 'tz' is set. Note: the scheduler polls on an interval (default 15s, minimum 5s) and does not 'catch up' missed runs."
+        "Create a scheduled cron job (shell or agent) with cron/at/every schedules. \
+         Standardizes on device-local timezone unless 'tz' is set. The scheduler polls on an \
+         interval (default 15s, minimum 5s) and does not 'catch up' missed runs.\n\
+         Delivery: agent jobs default to `mode: \"proactive\"` which lands in the in-app/web \
+         stream. When the current turn includes a `[Channel context]` block (e.g. Telegram, \
+         Discord, Slack), set `delivery` to `{ \"mode\": \"announce\", \"channel\": <channel>, \
+         \"to\": <reply target from the context block> }` so the reminder is delivered back to \
+         the same chat instead of the desktop. Only use the default proactive mode when the \
+         user explicitly asks for an in-app notification or when no channel context is present."
     }
 
     fn parameters_schema(&self) -> serde_json::Value {
@@ -225,6 +307,12 @@ impl Tool for CronAddTool {
                     }),
                 };
 
+                if let Some(ref cfg) = delivery {
+                    if let Err(msg) = validate_delivery(&self.config, cfg) {
+                        return Ok(ToolResult::error(msg));
+                    }
+                }
+
                 cron::add_agent_job(
                     &self.config,
                     name,
@@ -442,4 +530,194 @@ mod tests {
         assert!(result.is_error);
         assert!(result.output().contains("Missing 'prompt'"));
     }
+
+    // ── #928: announce-mode delivery validation ───────────────────
+
+    use crate::openhuman::config::TelegramConfig;
+
+    fn cfg_with_telegram(tmp: &TempDir, allowed: Vec<String>) -> Arc<Config> {
+        let mut config = Config {
+            workspace_dir: tmp.path().join("workspace"),
+            config_path: tmp.path().join("config.toml"),
+            ..Config::default()
+        };
+        config.channels_config.telegram = Some(TelegramConfig {
+            bot_token: "test-token".into(),
+            allowed_users: allowed,
+            stream_mode: Default::default(),
+            draft_update_interval_ms: 1000,
+            silent_streaming: true,
+            mention_only: false,
+        });
+        std::fs::create_dir_all(&config.workspace_dir).unwrap();
+        Arc::new(config)
+    }
+
+    #[tokio::test]
+    async fn agent_job_announce_telegram_authorized_chat_succeeds() {
+        let tmp = TempDir::new().unwrap();
+        let cfg = cfg_with_telegram(&tmp, vec!["123456".into()]);
+        let tool = CronAddTool::new(cfg.clone(), test_security(&cfg));
+        let result = tool
+            .execute(json!({
+                "schedule": { "kind": "every", "every_ms": 300000 },
+                "job_type": "agent",
+                "prompt": "remind me to drink water",
+                "delivery": {
+                    "mode": "announce",
+                    "channel": "telegram",
+                    "to": "123456"
+                }
+            }))
+            .await
+            .unwrap();
+
+        assert!(!result.is_error, "{:?}", result.output());
+        let jobs = cron::list_jobs(&cfg).unwrap();
+        assert_eq!(jobs.len(), 1);
+        assert_eq!(jobs[0].delivery.mode, "announce");
+        assert_eq!(jobs[0].delivery.channel.as_deref(), Some("telegram"));
+        assert_eq!(jobs[0].delivery.to.as_deref(), Some("123456"));
+    }
+
+    #[tokio::test]
+    async fn agent_job_announce_telegram_open_bot_allows_any_chat() {
+        // Empty allowed_users == "any sender ok". Mirrors the existing
+        // channel runtime behavior: an open bot accepts cron targets too.
+        let tmp = TempDir::new().unwrap();
+        let cfg = cfg_with_telegram(&tmp, vec![]);
+        let tool = CronAddTool::new(cfg.clone(), test_security(&cfg));
+        let result = tool
+            .execute(json!({
+                "schedule": { "kind": "every", "every_ms": 300000 },
+                "job_type": "agent",
+                "prompt": "ping",
+                "delivery": {
+                    "mode": "announce",
+                    "channel": "telegram",
+                    "to": "999"
+                }
+            }))
+            .await
+            .unwrap();
+
+        assert!(!result.is_error, "{:?}", result.output());
+    }
+
+    #[tokio::test]
+    async fn agent_job_announce_telegram_unauthorized_chat_rejected() {
+        let tmp = TempDir::new().unwrap();
+        let cfg = cfg_with_telegram(&tmp, vec!["alice".into()]);
+        let tool = CronAddTool::new(cfg.clone(), test_security(&cfg));
+        let result = tool
+            .execute(json!({
+                "schedule": { "kind": "every", "every_ms": 300000 },
+                "job_type": "agent",
+                "prompt": "ping",
+                "delivery": {
+                    "mode": "announce",
+                    "channel": "telegram",
+                    "to": "mallory"
+                }
+            }))
+            .await
+            .unwrap();
+
+        assert!(result.is_error);
+        assert!(result.output().contains("not in allowed_users"));
+        // Job must not be persisted on rejection.
+        assert!(cron::list_jobs(&cfg).unwrap().is_empty());
+    }
+
+    #[tokio::test]
+    async fn agent_job_announce_unconfigured_channel_rejected() {
+        let tmp = TempDir::new().unwrap();
+        let cfg = test_config(&tmp).await; // no telegram block
+        let tool = CronAddTool::new(cfg.clone(), test_security(&cfg));
+        let result = tool
+            .execute(json!({
+                "schedule": { "kind": "every", "every_ms": 300000 },
+                "job_type": "agent",
+                "prompt": "ping",
+                "delivery": {
+                    "mode": "announce",
+                    "channel": "telegram",
+                    "to": "123"
+                }
+            }))
+            .await
+            .unwrap();
+
+        assert!(result.is_error);
+        assert!(result.output().contains("not configured"));
+    }
+
+    #[tokio::test]
+    async fn agent_job_announce_missing_target_rejected() {
+        let tmp = TempDir::new().unwrap();
+        let cfg = cfg_with_telegram(&tmp, vec![]);
+        let tool = CronAddTool::new(cfg.clone(), test_security(&cfg));
+        let result = tool
+            .execute(json!({
+                "schedule": { "kind": "every", "every_ms": 300000 },
+                "job_type": "agent",
+                "prompt": "ping",
+                "delivery": {
+                    "mode": "announce",
+                    "channel": "telegram"
+                }
+            }))
+            .await
+            .unwrap();
+
+        assert!(result.is_error);
+        assert!(result.output().contains("delivery.to is required"));
+    }
+
+    #[test]
+    fn validate_delivery_skips_proactive_and_none_modes() {
+        let tmp = TempDir::new().unwrap();
+        let cfg = cfg_with_telegram(&tmp, vec!["alice".into()]);
+
+        let proactive = DeliveryConfig {
+            mode: "proactive".into(),
+            channel: None,
+            to: None,
+            best_effort: true,
+        };
+        assert!(validate_delivery(&cfg, &proactive).is_ok());
+
+        let none = DeliveryConfig {
+            mode: "none".into(),
+            channel: None,
+            to: None,
+            best_effort: true,
+        };
+        assert!(validate_delivery(&cfg, &none).is_ok());
+    }
+
+    #[test]
+    fn validate_delivery_announce_web_is_a_no_op() {
+        // "web" doesn't have an allowed_users gate; announce to web is
+        // a degenerate but valid configuration (in-app explicit).
+        let tmp = TempDir::new().unwrap();
+        let cfg = test_config_sync(&tmp);
+        let cfg_unused = DeliveryConfig {
+            mode: "announce".into(),
+            channel: Some("web".into()),
+            to: Some("any".into()),
+            best_effort: true,
+        };
+        assert!(validate_delivery(&cfg, &cfg_unused).is_ok());
+    }
+
+    fn test_config_sync(tmp: &TempDir) -> Arc<Config> {
+        let config = Config {
+            workspace_dir: tmp.path().join("workspace"),
+            config_path: tmp.path().join("config.toml"),
+            ..Config::default()
+        };
+        std::fs::create_dir_all(&config.workspace_dir).unwrap();
+        Arc::new(config)
+    }
 }