diff --git a/src/openhuman/channels/runtime/dispatch.rs b/src/openhuman/channels/runtime/dispatch.rs index 18edab68f..9929cba4c 100644 --- a/src/openhuman/channels/runtime/dispatch.rs +++ b/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")); + } + } } diff --git a/src/openhuman/cron/README.md b/src/openhuman/cron/README.md index 4880dba77..d7dad76eb 100644 --- a/src/openhuman/cron/README.md +++ b/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). diff --git a/src/openhuman/tools/impl/cron/add.rs b/src/openhuman/tools/impl/cron/add.rs index 4a8d45d5e..30174a54f 100644 --- a/src/openhuman/tools/impl/cron/add.rs +++ b/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, security: Arc, @@ -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\": , \ + \"to\": }` 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) -> Arc { + 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 { + 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) + } }