diff --git a/Cargo.lock b/Cargo.lock index 79ce265b886..7e7f749983f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3329,16 +3329,6 @@ dependencies = [ "markup5ever 0.36.1", ] -[[package]] -name = "html5ever" -version = "0.38.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1054432bae2f14e0061e33d23402fbaa67a921d319d56adc6bcf887ddad1cbc2" -dependencies = [ - "log", - "markup5ever 0.38.0", -] - [[package]] name = "html5ever" version = "0.39.0" @@ -4641,17 +4631,17 @@ dependencies = [ [[package]] name = "kuchikikiki" -version = "0.9.2" +version = "0.9.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b73885c6a3cefdf7a1db0327cefbe4b9b72cac94cae4b19ede4fa492d8af02a0" +checksum = "14683223e533503d404478bfd32826a4cba1b4906034ff74135404372390e87b" dependencies = [ "bitflags 2.11.0", "crc", "cssparser", - "html5ever 0.38.0", + "html5ever 0.36.1", "indexmap 2.14.0", "precomputed-hash", - "selectors 0.35.0", + "selectors", ] [[package]] @@ -5031,17 +5021,6 @@ dependencies = [ "web_atoms", ] -[[package]] -name = "markup5ever" -version = "0.38.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8983d30f2915feeaaab2d6babdd6bc7e9ed1a00b66b5e6d74df19aa9c0e91862" -dependencies = [ - "log", - "tendril 0.5.0", - "web_atoms", -] - [[package]] name = "markup5ever" version = "0.39.0" @@ -7164,7 +7143,7 @@ dependencies = [ "getopts", "html5ever 0.36.1", "precomputed-hash", - "selectors 0.33.0", + "selectors", "tendril 0.4.3", ] @@ -7258,25 +7237,6 @@ dependencies = [ "smallvec", ] -[[package]] -name = "selectors" -version = "0.35.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "93fdfed56cd634f04fe8b9ddf947ae3dc493483e819593d2ba17df9ad05db8b2" -dependencies = [ - "bitflags 2.11.0", - "cssparser", - "derive_more", - "log", - "new_debug_unreachable", - "phf 0.13.1", - "phf_codegen 0.13.1", - "precomputed-hash", - "rustc-hash 2.1.2", - "servo_arc", - "smallvec", -] - [[package]] name = "semver" version = "1.0.27" diff --git a/src/agent/agent_loop.rs b/src/agent/agent_loop.rs index 7312213230a..3459ac5367c 100644 --- a/src/agent/agent_loop.rs +++ b/src/agent/agent_loop.rs @@ -388,6 +388,9 @@ pub struct AgentDeps { pub llm_backend: String, /// Per-tenant rate limiting registry (lazily creates rate state per user). pub tenant_rates: Arc, + /// Per-channel tool routing config. Wrapped in RwLock for hot-reload support. + pub channel_routing: + Arc>>, } /// The main agent that coordinates all components. @@ -466,6 +469,7 @@ impl Agent { if let Some(ref interceptor) = deps.http_interceptor { scheduler.set_http_interceptor(Arc::clone(interceptor)); } + scheduler.set_channel_routing(Arc::clone(&deps.channel_routing)); let scheduler = Arc::new(scheduler); Self { @@ -2473,6 +2477,7 @@ mod tests { builder: None, llm_backend: "nearai".to_string(), tenant_rates: Arc::new(crate::tenant::TenantRateRegistry::new(4, 3)), + channel_routing: Arc::new(tokio::sync::RwLock::new(None)), }; Agent::new( diff --git a/src/agent/channel_routing.rs b/src/agent/channel_routing.rs new file mode 100644 index 00000000000..191b540aca6 --- /dev/null +++ b/src/agent/channel_routing.rs @@ -0,0 +1,1225 @@ +//! Per-channel tool routing. +//! +//! Filters which tools (MCP and built-in) the LLM can see based on the +//! originating channel. Configuration is stored in the database-backed +//! [`crate::db::SettingsStore`] under the key `channel_routing` and is +//! loaded at startup via [`ChannelRoutingConfig::load_from_store`]. +//! A file-based loader ([`ChannelRoutingConfig::load`]) is available as +//! a migration utility. + +use std::collections::{HashMap, HashSet}; +use std::path::Path; + +use serde::{Deserialize, Deserializer, Serialize}; + +use crate::db::SettingsStore; +use ironclaw_llm::ToolDefinition; + +/// Channel-to-tool-group routing configuration. +/// +/// Stored in the database-backed settings system (key: `channel_routing`). +/// When present, each incoming message's channel name is mapped to a group, +/// and only tools belonging to that group's allowed MCP servers (plus any +/// whitelisted built-in tools) are shown to the LLM. Use +/// [`Self::load_from_store`] / [`Self::save_to_store`] for persistence. +#[derive(Debug, Clone, Serialize)] +pub struct ChannelRoutingConfig { + /// MCP server allowlist per group. Key = group name, value = server names. + /// A tool named `ServerName_tool` belongs to server `ServerName`. + pub groups: HashMap>, + + /// Built-in tool allowlist per group. If a group is absent from this map, + /// all built-in tools are available. If present, only listed tools are kept. + #[serde(default)] + pub builtin_whitelist: HashMap>, + + /// Channel name → group name mapping. + pub channels: HashMap, + + /// Fallback group for channels not listed in `channels`. + pub default_group: String, + + /// Pre-computed MCP server prefixes sorted by length descending. + /// Populated by `precompute_prefixes()` after deserialization. + #[serde(skip)] + sorted_prefixes: Vec, + + /// Per-group allowed-server sets for O(1) membership tests. + /// Mirrors `groups` values; populated by `precompute_prefixes()`. + #[serde(skip)] + allowed_servers_sets: HashMap>, + + /// Per-group built-in whitelist sets for O(1) membership tests. + /// Mirrors `builtin_whitelist` values; populated by `precompute_prefixes()`. + #[serde(skip)] + builtin_whitelist_sets: HashMap>, +} + +#[derive(Debug, Clone, Deserialize)] +struct ChannelRoutingConfigSerde { + groups: HashMap>, + #[serde(default)] + builtin_whitelist: HashMap>, + channels: HashMap, + default_group: String, +} + +impl<'de> Deserialize<'de> for ChannelRoutingConfig { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + let raw = ChannelRoutingConfigSerde::deserialize(deserializer)?; + let mut config = Self { + groups: raw.groups, + builtin_whitelist: raw.builtin_whitelist, + channels: raw.channels, + default_group: raw.default_group, + sorted_prefixes: Vec::new(), + allowed_servers_sets: HashMap::new(), + builtin_whitelist_sets: HashMap::new(), + }; + config.precompute_prefixes(); + Ok(config) + } +} + +impl PartialEq for ChannelRoutingConfig { + fn eq(&self, other: &Self) -> bool { + self.groups == other.groups + && self.builtin_whitelist == other.builtin_whitelist + && self.channels == other.channels + && self.default_group == other.default_group + } +} + +impl Eq for ChannelRoutingConfig {} + +/// Exact channel names that identify direct messages (bypass routing entirely). +const DM_EXACT: &[&str] = &["gateway", "cli", "repl", "tui", "http"]; + +/// Channel-owned metadata bit for "this message is a DM". +/// +/// Trusted channel adapters set this after validating or normalizing the +/// inbound event. Routing should not rely on raw relay webhook fields alone. +pub const TRUSTED_DM_METADATA_KEY: &str = "channel_routing_dm"; + +impl ChannelRoutingConfig { + /// Return an `Arc>>` initialised to `None`. + /// + /// Convenience constructor for `AgentDeps.channel_routing` at startup + /// (before any config has been loaded) and for test helpers. + pub fn none_arc() -> std::sync::Arc>> { + std::sync::Arc::new(tokio::sync::RwLock::new(None)) + } + + /// Load config from `store` and atomically replace the value in `routing`. + /// + /// Returns `true` if the stored config differed from the current value + /// (using the semantic `PartialEq` impl — presence/absence *and* content + /// changes both count as changed). Used at startup and by the SIGHUP + /// handler for hot-reload. + pub async fn reload_from_store( + store: &(dyn SettingsStore + Send + Sync), + user_id: &str, + routing: &std::sync::Arc>>, + ) -> bool { + let new_routing = Self::load_from_store(store, user_id).await; + let mut guard = routing.write().await; + // Retain last-known-good: if the new value failed to load/parse (None) + // but we currently have a valid config (Some), keep the old one rather + // than silently disabling all routing. load_from_store already logged a + // warning, so the operator knows the config is stale. + if new_routing.is_none() && guard.is_some() { + return false; + } + let changed = new_routing != *guard; + *guard = new_routing; + changed + } + + /// Load from `/channel-routing.json`. + /// + /// **File-based utility** — not used in the normal startup path. Useful for + /// one-off migration of an existing `channel-routing.json` into the database + /// via [`Self::save_to_store`]. Returns `None` if the file doesn't exist or + /// can't be parsed (logged as warning). + pub fn load(base_dir: &Path) -> Option { + let path = base_dir.join("channel-routing.json"); + let content = match std::fs::read_to_string(&path) { + Ok(c) => c, + Err(e) if e.kind() == std::io::ErrorKind::NotFound => return None, + Err(e) => { + tracing::warn!("Failed to read {}: {}", path.display(), e); + return None; + } + }; + match serde_json::from_str::(&content) { + Ok(config) => { + if let Err(e) = config.validate() { + tracing::warn!("Channel routing config validation failed: {}", e); + return None; + } + tracing::debug!( + groups = ?config.groups.keys().collect::>(), + channels = config.channels.len(), + "Loaded channel routing config" + ); + Some(config) + } + Err(e) => { + tracing::warn!("Failed to parse {}: {}", path.display(), e); + None + } + } + } + + /// Load from database-backed SettingsStore. + /// + /// This provides hot-reload support — the settings system handles cache + /// invalidation and the web UI can modify the config without restarts. + pub async fn load_from_store( + store: &(dyn SettingsStore + Send + Sync), + user_id: &str, + ) -> Option { + match store.get_setting(user_id, "channel_routing").await { + Ok(Some(value)) => Self::parse_stored_value(value, "database"), + Ok(None) => { + tracing::debug!("No channel routing config in database"); + None + } + Err(e) => { + tracing::warn!("Failed to read channel routing from DB: {}", e); + None + } + } + } + + /// Load from a system-scoped database handle for a specific user. + /// + /// This is used by autonomous workers, which hold a [`SystemScope`] + /// rather than a raw [`SettingsStore`] but still need to apply the same + /// per-user routing rules as interactive dispatcher turns. + pub async fn load_from_system_scope( + store: &crate::tenant::SystemScope, + user_id: &str, + ) -> Option { + match store.get_channel_routing(user_id).await { + Ok(Some(value)) => Self::parse_stored_value(value, "database"), + Ok(None) => { + tracing::debug!("No channel routing config in database"); + None + } + Err(e) => { + tracing::warn!("Failed to read channel routing from DB: {}", e); + None + } + } + } + + /// Persist to database-backed SettingsStore. + /// + /// Write counterpart to [`Self::load_from_store`]. Call this after + /// modifying the config via the web UI or settings API to persist the + /// updated routing rules to the database. + pub async fn save_to_store( + &self, + store: &(dyn SettingsStore + Send + Sync), + user_id: &str, + ) -> Result<(), crate::error::DatabaseError> { + let value = serde_json::to_value(self) + .map_err(|e| crate::error::DatabaseError::Serialization(e.to_string()))?; + store.set_setting(user_id, "channel_routing", &value).await + } + + /// Validate configuration at load time. + fn validate(&self) -> Result<(), String> { + // default_group must exist in groups + if !self.groups.contains_key(&self.default_group) { + return Err(format!( + "default_group '{}' not found in groups", + self.default_group + )); + } + // All channel mappings must reference existing groups + for (channel, group) in &self.channels { + if !self.groups.contains_key(group) { + return Err(format!( + "channel '{}' maps to group '{}' which does not exist", + channel, group + )); + } + } + // All builtin_whitelist keys must reference existing groups + for group in self.builtin_whitelist.keys() { + if !self.groups.contains_key(group) { + return Err(format!( + "builtin_whitelist references group '{}' which does not exist", + group + )); + } + } + Ok(()) + } + + /// Pre-compute sorted MCP server prefixes (longest first) to avoid + /// allocations on the hot path. + fn precompute_prefixes(&mut self) { + // Sorted prefixes for longest-prefix-first MCP server extraction. + let mut all_servers: Vec = self + .groups + .values() + .flatten() + .cloned() + .collect::>() + .into_iter() + .collect(); + all_servers.sort_by_key(|s| std::cmp::Reverse(s.len())); + self.sorted_prefixes = all_servers; + + // Per-group HashSets for O(1) membership tests in filter_tool_defs. + self.allowed_servers_sets = self + .groups + .iter() + .map(|(group, servers)| (group.clone(), servers.iter().cloned().collect())) + .collect(); + self.builtin_whitelist_sets = self + .builtin_whitelist + .iter() + .map(|(group, tools)| (group.clone(), tools.iter().cloned().collect())) + .collect(); + } + + fn parse_stored_value(value: serde_json::Value, source: &str) -> Option { + match serde_json::from_value::(value) { + Ok(config) => { + if let Err(e) = config.validate() { + tracing::warn!("Channel routing config from {} invalid: {}", source, e); + return None; + } + tracing::debug!( + source, + groups = ?config.groups.keys().collect::>(), + channels = config.channels.len(), + "Loaded channel routing config" + ); + Some(config) + } + Err(e) => { + tracing::warn!("Failed to parse channel routing from {}: {}", source, e); + None + } + } + } + + /// Resolve which group a channel belongs to. + pub fn resolve_group(&self, channel: &str) -> &str { + self.channels + .get(channel) + .map(|s| s.as_str()) + .unwrap_or(&self.default_group) + } + + /// Whether this channel name represents a direct message (no filtering). + pub fn is_dm(channel: &str, metadata: &serde_json::Value) -> bool { + // Exact matches for web/CLI channels + if DM_EXACT.contains(&channel) { + return true; + } + // Any trusted relay adapter can stamp this flag to bypass routing. + // Set server-side after webhook/auth validation — not spoofable by clients. + if metadata + .get(TRUSTED_DM_METADATA_KEY) + .and_then(|v| v.as_bool()) + .unwrap_or(false) + { + return true; + } + // Slack DMs: channel name starts with 'D' (Slack convention) + if channel == "slack" { + if metadata + .get("channel") + .and_then(|v| v.as_str()) + .is_some_and(|ch| ch.starts_with('D')) + { + return true; + } + if metadata + .get("event_type") + .and_then(|v| v.as_str()) + .is_some_and(|et| et == "direct_message") + { + return true; + } + } + // Telegram DMs: chat_type == "private" + if channel == "telegram" + && metadata + .get("chat_type") + .and_then(|v| v.as_str()) + .is_some_and(|ct| ct == "private") + { + return true; + } + false + } + + /// Core allow/deny decision for a single tool name within an + /// already-resolved `group`. + /// + /// Single source of truth for the allowed-set / built-in-whitelist / + /// fail-closed branching, shared by [`Self::filter_tool_defs`] (the + /// presentation filter that hides tools from the LLM) and + /// [`Self::is_tool_permitted`] (the execution-time gate). Keeping both + /// callers on this one implementation is what guarantees a tool the LLM + /// can see is exactly a tool it can execute, and vice versa. + /// + /// Does **not** handle DM bypass — callers resolve that (via + /// [`Self::routing_group_for`]) before passing a group in. An unknown + /// `group` (absent from `allowed_servers_sets`) is treated fail-closed: MCP + /// tools blocked, built-ins gated by the *default* group's whitelist. + /// + /// **Classification order**: registered built-in names are checked *before* MCP + /// prefix extraction. An MCP server named, e.g. `memory` would otherwise cause + /// `memory_search` (a built-in) to be authorised as an MCP tool, bypassing the + /// `builtin_whitelist` check. Built-in registration is authoritative. + /// + /// **Built-in default**: a group with no `builtin_whitelist` entry denies all + /// built-ins. Operators must explicitly enumerate allowed built-ins (or use `"*"` + /// when that shorthand is added). This is the safe default — `shell`, + /// `http_request`, and `create_job` should not be silently granted to restricted + /// groups that omit the whitelist key. + fn permit_in_group( + &self, + group: &str, + tool_name: &str, + builtin_names: &std::collections::HashSet, + ) -> bool { + match self.allowed_servers_sets.get(group) { + Some(allowed_set) => { + // Check registered built-ins first — built-in registration is + // authoritative; checking MCP prefix first would let a server named + // "memory" authorise built-in "memory_search" via the MCP allowlist, + // bypassing the builtin_whitelist check. + if builtin_names.contains(tool_name) { + match self.builtin_whitelist_sets.get(group) { + Some(set) => set.contains(tool_name), + // No whitelist entry ⇒ deny. Operators must enumerate allowed + // built-ins explicitly; omitting the key is not a wildcard grant. + None => false, + } + } else if let Some(server) = self.extract_mcp_server(tool_name) { + allowed_set.contains(server) + } else { + // Neither a known MCP server prefix nor a registered + // built-in — fail closed so unregistered-server tools + // can't leak through as apparent built-ins. + false + } + } + None => { + // Unknown group — fail-closed: built-ins gated by default group's + // whitelist; anything else (MCP or unregistered) blocked. + if builtin_names.contains(tool_name) { + self.builtin_whitelist_sets + .get(&self.default_group) + .is_some_and(|set| set.contains(tool_name)) + } else { + false + } + } + } + } + + /// Resolve the effective routing group for a message/job given its origin + /// `channel` and `metadata`. + /// + /// Returns `None` when the message is a DM (routing bypassed entirely). + /// `Some(group)` otherwise. When `channel` is `None` (e.g. a routine- or + /// heartbeat-spawned job with no origin channel) this falls back to the + /// default group — fail-closed, never a DM bypass. + /// + /// Shared by [`crate::tools::builtin::job`]'s `derive_routed_mcp_servers` + /// so the group-resolution + DM-bypass logic isn't reimplemented there. + pub fn routing_group_for( + &self, + channel: Option<&str>, + metadata: &serde_json::Value, + ) -> Option<&str> { + match channel { + Some(ch) if Self::is_dm(ch, metadata) => None, + Some(ch) => Some(self.resolve_group(ch)), + None => Some(&self.default_group), + } + } + + /// Filter tool definitions based on channel routing rules. + /// + /// MCP tools are identified by having an underscore-separated server prefix + /// (e.g. `Notion_post_search` → server `Notion`). Tools not matching any + /// known MCP server prefix are checked against `builtin_names`; tools that + /// are neither a known MCP server NOR a registered built-in are blocked + /// (fail-closed) to prevent tools from unregistered MCP servers leaking + /// through as apparent built-ins. The per-tool decision is delegated to + /// [`Self::permit_in_group`] — the same predicate the execution-time gate + /// [`Self::is_tool_permitted`] uses. + pub fn filter_tool_defs( + &self, + channel: &str, + metadata: &serde_json::Value, + tools: Vec, + builtin_names: &std::collections::HashSet, + ) -> Vec { + // DM bypass + group resolution share one implementation with the + // execution gate and `derive_routed_mcp_servers`. + let group = match self.routing_group_for(Some(channel), metadata) { + None => return tools, // DM — routing does not apply. + Some(group) => group, + }; + + // Warn once (not per tool) when the resolved group is unknown — a + // misconfiguration. `permit_in_group` still fails closed below. + if !self.allowed_servers_sets.contains_key(group) { + tracing::warn!( + group, + default_group = %self.default_group, + "Channel routing group not found, blocking MCP tools and applying default built-in policy" + ); + } + + tools + .into_iter() + .filter(|tool| { + let permitted = self.permit_in_group(group, &tool.name, builtin_names); + if !permitted { + tracing::debug!( + tool_name = %tool.name, + channel, + group, + "Channel routing: blocking tool not permitted in group" + ); + } + permitted + }) + .collect() + } + + /// Returns `true` if the named tool is permitted to execute on `channel`. + /// + /// Execution-layer counterpart to [`Self::filter_tool_defs`] for a single + /// tool name (no full `ToolDefinition` required). Uses the same allow/deny + /// core ([`Self::permit_in_group`]) and the same DM handling + /// ([`Self::is_dm`]) as the presentation filter, so the set of executable + /// tools matches the set shown to the LLM. + /// + /// Called from two gates: + /// - [`crate::tools::dispatch::ToolDispatcher::dispatch`] (gateway/CLI/ + /// routine path) passes `Value::Null` for `metadata` — those callers + /// carry no per-message metadata, so only `DM_EXACT` channels bypass. + /// - the worker agent loop in [`crate::worker::job`] passes the job's + /// `notify_metadata`, so metadata-based DM bypass (Slack `D…`, Telegram + /// `private`, the trusted flag) matches what the LLM was shown. + pub fn is_tool_permitted( + &self, + channel: &str, + metadata: &serde_json::Value, + tool_name: &str, + builtin_names: &std::collections::HashSet, + ) -> bool { + match self.routing_group_for(Some(channel), metadata) { + None => true, // DM — routing bypassed, same as the presentation filter. + Some(group) => self.permit_in_group(group, tool_name, builtin_names), + } + } + + /// Try to extract the MCP server name from a tool name. + /// + /// Uses pre-computed prefixes sorted by length descending to avoid + /// `Kit` matching `KitchenAI_recipe_search`. + fn extract_mcp_server<'a>(&self, tool_name: &'a str) -> Option<&'a str> { + for server in &self.sorted_prefixes { + // `server.len()` is a *byte* length (servers may be multi-byte, + // e.g. "Café" — see `test_extract_mcp_server_handles_multibyte_prefix`). + // Indexing `as_bytes()` at that offset is bounds-checked by the + // preceding length guard, and is always valid on a byte slice. + if tool_name.len() > server.len() + && tool_name.as_bytes()[server.len()] == b'_' + && tool_name.starts_with(server.as_str()) + { + // `starts_with` guarantees the prefix matches byte-for-byte, so + // `server.len()` falls on a char boundary of `tool_name` and the + // UTF-8-safe slice below always returns `Some`. + debug_assert!( + tool_name.is_char_boundary(server.len()), + "matched MCP prefix must end on a char boundary" + ); + return tool_name.get(..server.len()); + } + } + None + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn sample_config() -> ChannelRoutingConfig { + let json = r#"{ + "groups": { + "minimal": ["Archon"], + "dev": ["Archon", "Kiro", "Notion"] + }, + "builtin_whitelist": { + "minimal": ["memory_search", "create_job"] + }, + "channels": { + "agentiffai-dev-issues": "dev" + }, + "default_group": "minimal" + }"#; + serde_json::from_str(json).unwrap() + } + + fn no_metadata() -> serde_json::Value { + serde_json::json!({}) + } + + fn make_tool_def(name: &str) -> ToolDefinition { + ToolDefinition { + name: name.to_string(), + description: String::new(), + parameters: serde_json::json!({}), + } + } + + /// Built-in tool names used across filter tests. MCP-prefixed tools + /// (Archon_*, Notion_*, etc.) must NOT appear here. + fn test_builtin_names() -> std::collections::HashSet { + [ + "memory_search", + "memory_write", + "create_job", + "shell", + "http_request", + "web_search", + ] + .iter() + .map(|s| s.to_string()) + .collect() + } + + #[test] + fn test_deserialize_config() { + let config = sample_config(); + assert_eq!(config.groups.len(), 2); + assert_eq!(config.default_group, "minimal"); + assert_eq!(config.channels["agentiffai-dev-issues"], "dev"); + } + + #[test] + fn test_resolve_group_mapped_channel() { + let config = sample_config(); + assert_eq!(config.resolve_group("agentiffai-dev-issues"), "dev"); + } + + #[test] + fn test_resolve_group_unmapped_falls_to_default() { + let config = sample_config(); + assert_eq!(config.resolve_group("random-channel"), "minimal"); + } + + #[test] + fn test_is_dm_exact_matches() { + let md = no_metadata(); + assert!(ChannelRoutingConfig::is_dm("gateway", &md)); + assert!(ChannelRoutingConfig::is_dm("cli", &md)); + assert!(ChannelRoutingConfig::is_dm("repl", &md)); + assert!(ChannelRoutingConfig::is_dm("tui", &md)); + assert!(ChannelRoutingConfig::is_dm("http", &md)); + // "web" alone should NOT match (use "gateway" for web chat) + assert!(!ChannelRoutingConfig::is_dm("web", &md)); + // "web-team-standup" must not bypass routing + assert!(!ChannelRoutingConfig::is_dm("web-team-standup", &md)); + assert!(!ChannelRoutingConfig::is_dm("agentiffai-dev-issues", &md)); + } + + #[test] + fn test_is_dm_slack_metadata() { + // Slack DM via channel ID starting with 'D' + let dm_meta = serde_json::json!({"channel": "D12345"}); + assert!(ChannelRoutingConfig::is_dm("slack", &dm_meta)); + assert!(!ChannelRoutingConfig::is_dm("slack-relay", &dm_meta)); + + // Slack channel message (not DM) + let chan_meta = serde_json::json!({"channel": "C12345"}); + assert!(!ChannelRoutingConfig::is_dm("slack", &chan_meta)); + + // Slack DM via event_type + let event_meta = serde_json::json!({"event_type": "direct_message"}); + assert!(ChannelRoutingConfig::is_dm("slack", &event_meta)); + } + + #[test] + fn test_is_dm_trusted_flag_works_for_any_channel() { + // Without the flag, relay channels are subject to routing. + let untrusted_meta = serde_json::json!({"event_type": "direct_message"}); + assert!(!ChannelRoutingConfig::is_dm("slack-relay", &untrusted_meta)); + assert!(!ChannelRoutingConfig::is_dm( + "telegram-relay", + &untrusted_meta + )); + + // With the trusted server-side flag, any relay channel bypasses routing. + let trusted_meta = serde_json::json!({ + "event_type": "direct_message", + TRUSTED_DM_METADATA_KEY: true, + }); + assert!(ChannelRoutingConfig::is_dm("slack-relay", &trusted_meta)); + assert!(ChannelRoutingConfig::is_dm("telegram-relay", &trusted_meta)); + assert!(ChannelRoutingConfig::is_dm( + "some-future-relay", + &trusted_meta + )); + } + + #[test] + fn test_is_dm_telegram_metadata() { + let private_meta = serde_json::json!({"chat_type": "private"}); + assert!(ChannelRoutingConfig::is_dm("telegram", &private_meta)); + + let group_meta = serde_json::json!({"chat_type": "group"}); + assert!(!ChannelRoutingConfig::is_dm("telegram", &group_meta)); + } + + #[test] + fn test_filter_dm_returns_all_tools_for_tui_and_http() { + let config = sample_config(); + let tools = vec![ + make_tool_def("Archon_list_tasks"), + make_tool_def("Smartlead_send"), + make_tool_def("shell"), + ]; + let md = no_metadata(); + let bn = test_builtin_names(); + assert_eq!( + config + .filter_tool_defs("tui", &md, tools.clone(), &bn) + .len(), + 3 + ); + assert_eq!(config.filter_tool_defs("http", &md, tools, &bn).len(), 3); + } + + #[test] + fn test_filter_keeps_allowed_mcp_tools() { + let json = r#"{ + "groups": { + "minimal": ["Archon"], + "dev": ["Archon", "Kiro", "Notion"], + "leads": ["Archon", "Smartlead"] + }, + "builtin_whitelist": { + "minimal": ["memory_search", "create_job"] + }, + "channels": { + "agentiffai-dev-issues": "dev" + }, + "default_group": "minimal" + }"#; + let config: ChannelRoutingConfig = serde_json::from_str(json).unwrap(); + let tools = vec![ + make_tool_def("Archon_list_tasks"), + make_tool_def("Kiro_run_task"), + make_tool_def("Notion_post_search"), + make_tool_def("Smartlead_send"), + ]; + let md = no_metadata(); + let filtered = + config.filter_tool_defs("agentiffai-dev-issues", &md, tools, &test_builtin_names()); + let names: Vec<&str> = filtered.iter().map(|t| t.name.as_str()).collect(); + assert!(names.contains(&"Archon_list_tasks")); + assert!(names.contains(&"Kiro_run_task")); + assert!(names.contains(&"Notion_post_search")); + assert!(!names.contains(&"Smartlead_send")); + } + + #[test] + fn test_filter_restricts_builtins_when_whitelisted() { + let config = sample_config(); + let tools = vec![ + make_tool_def("Archon_list_tasks"), + make_tool_def("memory_search"), + make_tool_def("create_job"), + make_tool_def("shell"), + make_tool_def("http_request"), + ]; + let md = no_metadata(); + let filtered = + config.filter_tool_defs("unmapped-channel", &md, tools, &test_builtin_names()); + let names: Vec<&str> = filtered.iter().map(|t| t.name.as_str()).collect(); + assert!(names.contains(&"Archon_list_tasks")); + assert!(names.contains(&"memory_search")); + assert!(names.contains(&"create_job")); + assert!(!names.contains(&"shell")); + assert!(!names.contains(&"http_request")); + } + + #[test] + fn test_filter_denies_builtins_when_no_whitelist() { + // A group without a `builtin_whitelist` entry denies all built-ins + // (deny-by-default). Operators must explicitly enumerate allowed built-ins. + // The "dev" group in sample_config has no whitelist, so shell and + // memory_search are blocked even though the group is otherwise permissive + // for MCP tools. + let config = sample_config(); + let tools = vec![ + make_tool_def("Archon_list_tasks"), + make_tool_def("shell"), + make_tool_def("memory_search"), + ]; + let md = no_metadata(); + let filtered = + config.filter_tool_defs("agentiffai-dev-issues", &md, tools, &test_builtin_names()); + // Only the MCP tool passes; built-ins are denied (no whitelist entry for "dev"). + assert_eq!(filtered.len(), 1); + assert_eq!(filtered[0].name, "Archon_list_tasks"); + } + + #[test] + fn test_filter_dm_returns_all_tools() { + let config = sample_config(); + let tools = vec![ + make_tool_def("Archon_list_tasks"), + make_tool_def("Smartlead_send"), + make_tool_def("shell"), + ]; + let md = no_metadata(); + let filtered = config.filter_tool_defs("gateway", &md, tools, &test_builtin_names()); + assert_eq!(filtered.len(), 3); + } + + #[test] + fn test_load_returns_none_for_missing_file() { + let dir = tempfile::tempdir().unwrap(); + let config = ChannelRoutingConfig::load(dir.path()); + assert!(config.is_none()); + } + + #[test] + fn test_load_parses_valid_file() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("channel-routing.json"); + std::fs::write( + &path, + r#"{ + "groups": {"minimal": ["Archon"]}, + "channels": {}, + "default_group": "minimal" + }"#, + ) + .unwrap(); + let config = ChannelRoutingConfig::load(dir.path()); + assert!(config.is_some()); + assert_eq!(config.unwrap().default_group, "minimal"); + } + + #[test] + fn test_load_returns_none_for_invalid_json() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("channel-routing.json"); + std::fs::write(&path, "not json").unwrap(); + let config = ChannelRoutingConfig::load(dir.path()); + assert!(config.is_none()); + } + + #[test] + fn test_validate_rejects_bad_default_group() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("channel-routing.json"); + std::fs::write( + &path, + r#"{ + "groups": {"minimal": ["Archon"]}, + "channels": {}, + "default_group": "typo" + }"#, + ) + .unwrap(); + let config = ChannelRoutingConfig::load(dir.path()); + assert!(config.is_none()); + } + + #[test] + fn test_validate_rejects_bad_channel_mapping() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("channel-routing.json"); + std::fs::write( + &path, + r#"{ + "groups": {"minimal": ["Archon"]}, + "channels": {"some-channel": "nonexistent"}, + "default_group": "minimal" + }"#, + ) + .unwrap(); + let config = ChannelRoutingConfig::load(dir.path()); + assert!(config.is_none()); + } + + #[test] + fn test_validate_rejects_bad_builtin_whitelist_key() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("channel-routing.json"); + std::fs::write( + &path, + r#"{ + "groups": {"minimal": ["Archon"]}, + "builtin_whitelist": {"minmal": ["shell"]}, + "channels": {}, + "default_group": "minimal" + }"#, + ) + .unwrap(); + let config = ChannelRoutingConfig::load(dir.path()); + assert!(config.is_none()); + } + + #[test] + fn test_prefix_matching_longest_wins() { + // Kit vs KitchenAI — KitchenAI must match first + let json = r#"{ + "groups": { + "all": ["Kit", "KitchenAI"] + }, + "channels": {}, + "default_group": "all" + }"#; + let config: ChannelRoutingConfig = serde_json::from_str(json).unwrap(); + + assert_eq!( + config.extract_mcp_server("KitchenAI_recipe_search"), + Some("KitchenAI") + ); + assert_eq!( + config.extract_mcp_server("Kit_list_subscribers"), + Some("Kit") + ); + } + + #[test] + fn test_deserialize_precomputes_prefixes() { + let json = r#"{ + "groups": { + "all": ["Kit", "KitchenAI"] + }, + "channels": {}, + "default_group": "all" + }"#; + let config: ChannelRoutingConfig = serde_json::from_str(json).unwrap(); + assert_eq!( + config.extract_mcp_server("KitchenAI_recipe_search"), + Some("KitchenAI") + ); + } + + #[test] + fn test_unknown_group_blocks_mcp_and_applies_default_builtin_policy() { + let mut config = sample_config(); + // Force a channel to map to a nonexistent group (bypassing validation for test) + config + .channels + .insert("hacked-channel".to_string(), "nonexistent".to_string()); + let tools = vec![ + make_tool_def("Archon_list_tasks"), + make_tool_def("memory_search"), + make_tool_def("shell"), + ]; + let md = no_metadata(); + let filtered = config.filter_tool_defs("hacked-channel", &md, tools, &test_builtin_names()); + let names: Vec<&str> = filtered.iter().map(|t| t.name.as_str()).collect(); + assert!(!names.contains(&"Archon_list_tasks")); + assert!(names.contains(&"memory_search")); + assert!(!names.contains(&"shell")); + } + + #[test] + fn test_partial_eq_detects_content_changes() { + let config_a = sample_config(); + let mut config_b = sample_config(); + assert_eq!(config_a, config_b); + + config_b.default_group = "dev".to_string(); + assert_ne!(config_a, config_b); + } + + #[test] + fn test_full_routing_scenario() { + let json = r#"{ + "groups": { + "minimal": ["Archon"], + "content": ["Archon", "Notion", "Kit"], + "dev": ["Archon", "Kiro", "Notion"] + }, + "builtin_whitelist": { + "content": ["memory_search", "memory_write", "create_job"] + }, + "channels": { + "agentiffai-marketing": "content", + "agentiffai-dev-issues": "dev" + }, + "default_group": "minimal" + }"#; + let config: ChannelRoutingConfig = serde_json::from_str(json).unwrap(); + + let md = no_metadata(); + + // Content channel: only Archon+Notion+Kit MCP tools + whitelisted builtins + let all_tools = vec![ + make_tool_def("Archon_list_tasks"), + make_tool_def("Notion_post_search"), + make_tool_def("Kit_list_subscribers"), + make_tool_def("Kiro_run_task"), + make_tool_def("memory_search"), + make_tool_def("shell"), + make_tool_def("create_job"), + ]; + + let bn = test_builtin_names(); + let content_tools = + config.filter_tool_defs("agentiffai-marketing", &md, all_tools.clone(), &bn); + let content_names: Vec<&str> = content_tools.iter().map(|t| t.name.as_str()).collect(); + assert_eq!( + content_names, + vec![ + "Archon_list_tasks", + "Notion_post_search", + "Kit_list_subscribers", + "memory_search", + "create_job", + ] + ); + + // Dev channel: Archon+Kiro+Notion MCP tools only; no builtin_whitelist entry → built-ins denied + let dev_tools = + config.filter_tool_defs("agentiffai-dev-issues", &md, all_tools.clone(), &bn); + let dev_names: Vec<&str> = dev_tools.iter().map(|t| t.name.as_str()).collect(); + assert_eq!( + dev_names, + vec!["Archon_list_tasks", "Notion_post_search", "Kiro_run_task",] + ); + + // DM (gateway): everything + let dm_tools = config.filter_tool_defs("gateway", &md, all_tools.clone(), &bn); + assert_eq!(dm_tools.len(), 7); + + // Slack DM via metadata: everything + let slack_dm_meta = serde_json::json!({"channel": "D12345"}); + let slack_dm_tools = config.filter_tool_defs("slack", &slack_dm_meta, all_tools, &bn); + assert_eq!(slack_dm_tools.len(), 7); + } + + /// Tests for `reload_from_store` — exercises the SIGHUP hot-reload path + /// end-to-end: persist → load → mutate → reload, asserting `changed` is + /// correct at each step and the arc reflects the latest config. + /// + /// Requires the `libsql` feature because it uses an in-memory SQLite + /// database. Run with: `cargo test --features libsql channel_routing` + #[cfg(feature = "libsql")] + mod reload_tests { + use super::*; + use crate::db::{Database, libsql::LibSqlBackend}; + + /// Create a temporary local SQLite DB with migrations applied. + /// + /// In-memory databases do not share state between libSQL connections + /// (each `connect()` call sees an empty DB), so tests that perform + /// multiple operations must use a local file with a temp dir. + async fn test_db() -> (LibSqlBackend, tempfile::TempDir) { + let dir = tempfile::tempdir().unwrap(); + let db = LibSqlBackend::new_local(&dir.path().join("test.db")) + .await + .unwrap(); + db.run_migrations().await.unwrap(); + (db, dir) // keep `dir` alive so the temp file isn't deleted + } + + #[tokio::test] + async fn test_reload_from_store_detects_changes() { + let (db, _dir) = test_db().await; + let arc = ChannelRoutingConfig::none_arc(); + + // Nothing in DB yet — arc stays None, reported as unchanged + let changed = ChannelRoutingConfig::reload_from_store(&db, "test-user", &arc).await; + assert!(!changed, "empty DB should be unchanged"); + assert!(arc.read().await.is_none()); + + // Store a config; reload should transition None → Some (changed) + let config = sample_config(); + config.save_to_store(&db, "test-user").await.unwrap(); + let changed = ChannelRoutingConfig::reload_from_store(&db, "test-user", &arc).await; + assert!(changed, "None → Some must be reported as changed"); + assert!(arc.read().await.is_some()); + + // Same config again — content identical, not changed + let changed = ChannelRoutingConfig::reload_from_store(&db, "test-user", &arc).await; + assert!(!changed, "identical reload must be unchanged"); + + // Mutate default_group and reload — content changed + let mut updated = sample_config(); + updated.default_group = "dev".to_string(); + updated.save_to_store(&db, "test-user").await.unwrap(); + let changed = ChannelRoutingConfig::reload_from_store(&db, "test-user", &arc).await; + assert!(changed, "content change must be reported as changed"); + assert_eq!(arc.read().await.as_ref().unwrap().default_group, "dev"); + } + + #[tokio::test] + async fn test_none_arc_initialises_to_none() { + let arc = ChannelRoutingConfig::none_arc(); + assert!(arc.read().await.is_none()); + } + + #[tokio::test] + async fn test_save_and_load_roundtrip() { + let (db, _dir) = test_db().await; + + let config = sample_config(); + config.save_to_store(&db, "user1").await.unwrap(); + + let loaded = ChannelRoutingConfig::load_from_store(&db, "user1") + .await + .expect("should load saved config"); + assert_eq!(loaded, config); + } + } + + #[test] + fn test_extract_mcp_server_handles_multibyte_prefix() { + // Regression: tool_name.as_bytes()[server.len()] byte-indexes into the + // string. If `server` contains multi-byte chars (e.g. "Café"), its + // `.len()` is the *byte* length (5), not char length (4). The check + // `tool_name.as_bytes()[server.len()] == b'_'` correctly accesses byte 5, + // and `tool_name.get(..server.len())` is the UTF-8-safe slice that + // returns None rather than panicking on a non-char-boundary index. + // Both operations must not panic on multi-byte server names. + let json = r#"{ + "groups": { + "all": ["Café"] + }, + "channels": {}, + "default_group": "all" + }"#; + let config: ChannelRoutingConfig = serde_json::from_str(json).unwrap(); + // "Café" is 5 bytes (UTF-8: C, a, f, 0xc3, 0xa9). + // "Café_query" — byte 5 is '_'; server.len() == 5 → safe. + assert_eq!(config.extract_mcp_server("Café_query"), Some("Café")); + // Ensure non-matching multi-byte prefix doesn't panic either. + assert_eq!(config.extract_mcp_server("Cafe_query"), None); + assert_eq!(config.extract_mcp_server("Caf"), None); + } + + #[test] + fn test_is_tool_permitted_enforces_routing() { + let json = r#"{ + "groups": { + "minimal": ["Archon"], + "dev": ["Archon", "Kiro"] + }, + "builtin_whitelist": { + "minimal": ["create_job"] + }, + "channels": { + "agentiffai-dev": "dev" + }, + "default_group": "minimal" + }"#; + let config: ChannelRoutingConfig = serde_json::from_str(json).unwrap(); + + let builtin_names: std::collections::HashSet = + ["shell", "create_job", "memory_search", "web_search"] + .iter() + .map(|s| s.to_string()) + .collect(); + + let md = no_metadata(); + + // DM_EXACT channels bypass routing entirely. + assert!(config.is_tool_permitted("gateway", &md, "Smartlead_send", &builtin_names)); + assert!(config.is_tool_permitted("cli", &md, "shell", &builtin_names)); + + // Dev channel: Archon + Kiro allowed; no builtin_whitelist entry → built-ins denied. + assert!(config.is_tool_permitted("agentiffai-dev", &md, "Archon_list", &builtin_names)); + assert!(config.is_tool_permitted("agentiffai-dev", &md, "Kiro_run", &builtin_names)); + assert!(!config.is_tool_permitted("agentiffai-dev", &md, "Smartlead_send", &builtin_names)); + assert!(!config.is_tool_permitted("agentiffai-dev", &md, "shell", &builtin_names)); // no whitelist → builtins denied + + // Minimal (default) channel: Archon only, create_job built-in only. + assert!(config.is_tool_permitted("other-channel", &md, "Archon_list", &builtin_names)); + assert!(!config.is_tool_permitted("other-channel", &md, "Kiro_run", &builtin_names)); + assert!(config.is_tool_permitted("other-channel", &md, "create_job", &builtin_names)); + assert!(!config.is_tool_permitted("other-channel", &md, "shell", &builtin_names)); + + // Unknown MCP server (not in any group, not a builtin) — fail closed. + assert!(!config.is_tool_permitted( + "other-channel", + &md, + "Serpstat_keywords", + &builtin_names + )); + assert!(!config.is_tool_permitted( + "agentiffai-dev", + &md, + "Serpstat_keywords", + &builtin_names + )); + } + + #[test] + fn test_is_tool_permitted_metadata_dm_bypass() { + // Metadata-based DM (Slack D-channel) must bypass routing at the + // execution gate too — otherwise the worker would hide a tool from the + // LLM's view via filter_tool_defs (DM → all tools) but then refuse to + // execute it. The two gates must agree. + let config = sample_config(); + let builtin_names = test_builtin_names(); + + let slack_dm = serde_json::json!({"channel": "D12345"}); + // "slack" is not in DM_EXACT, but the metadata marks it a DM. + assert!(config.is_tool_permitted("slack", &slack_dm, "Smartlead_send", &builtin_names)); + + // Same channel as a group message (C-channel) → routing applies, and + // Smartlead is not in the default "minimal" group → blocked. + let slack_channel = serde_json::json!({"channel": "C12345"}); + assert!(!config.is_tool_permitted( + "slack", + &slack_channel, + "Smartlead_send", + &builtin_names + )); + } + + #[test] + fn test_routing_group_for_dm_and_fallback() { + let config = sample_config(); + let md = no_metadata(); + + // Mapped channel → its group. + assert_eq!( + config.routing_group_for(Some("agentiffai-dev-issues"), &md), + Some("dev") + ); + // Unmapped channel → default group. + assert_eq!( + config.routing_group_for(Some("random"), &md), + Some("minimal") + ); + // DM_EXACT channel → None (routing bypassed). + assert_eq!(config.routing_group_for(Some("gateway"), &md), None); + // Metadata DM → None. + let slack_dm = serde_json::json!({"channel": "D1"}); + assert_eq!(config.routing_group_for(Some("slack"), &slack_dm), None); + // No channel (routine/heartbeat) → default group, never DM bypass. + assert_eq!(config.routing_group_for(None, &md), Some("minimal")); + } +} diff --git a/src/agent/commands.rs b/src/agent/commands.rs index 8d3fd23ed42..f2df2ff55eb 100644 --- a/src/agent/commands.rs +++ b/src/agent/commands.rs @@ -1249,6 +1249,7 @@ mod tests { builder: None, llm_backend: "nearai".to_string(), tenant_rates: Arc::new(crate::tenant::TenantRateRegistry::new(4, 3)), + channel_routing: crate::agent::channel_routing::ChannelRoutingConfig::none_arc(), }; Agent::new( diff --git a/src/agent/dispatcher.rs b/src/agent/dispatcher.rs index c1a266cb782..2146b67b679 100644 --- a/src/agent/dispatcher.rs +++ b/src/agent/dispatcher.rs @@ -109,6 +109,33 @@ impl TurnUsageSummary { } impl Agent { + /// Apply per-channel tool filtering if routing config is loaded. + async fn apply_channel_routing( + &self, + channel: &str, + metadata: &serde_json::Value, + tools: Vec, + ) -> Vec { + let routing = { self.deps.channel_routing.read().await.clone() }; + if let Some(routing) = routing { + let builtin_names = self.deps.tools.builtin_tool_names().await; + let before = tools.len(); + let filtered = routing.filter_tool_defs(channel, metadata, tools, &builtin_names); + if filtered.len() < before { + tracing::debug!( + channel, + group = routing.resolve_group(channel), + before, + after = filtered.len(), + "Channel routing filtered tools" + ); + } + filtered + } else { + tools + } + } + /// Run the agentic loop: call LLM, execute tools, repeat until text response. /// /// Returns `AgenticLoopResult::Response` on completion, or @@ -262,6 +289,9 @@ impl Agent { // Build system prompts once for this turn. Two variants: with tools // (normal iterations) and without (force_text final iteration). let initial_tool_defs = self.tools().tool_definitions().await; + let initial_tool_defs = self + .apply_channel_routing(&message.channel, &message.metadata, initial_tool_defs) + .await; let initial_tool_defs = if !active_skills.is_empty() { crate::skills::attenuate_tools(&initial_tool_defs, &active_skills).tools } else { @@ -458,6 +488,10 @@ impl<'a> LoopDelegate for ChatDelegate<'a> { // Refresh tool definitions each iteration so newly built tools become visible let tool_defs = self.agent.tools().tool_definitions().await; + let tool_defs = self + .agent + .apply_channel_routing(&self.message.channel, &self.message.metadata, tool_defs) + .await; // Apply trust-based tool attenuation if skills are active. let tool_defs = if !self.active_skills.is_empty() { @@ -879,6 +913,15 @@ impl<'a> LoopDelegate for ChatDelegate<'a> { bool, // allow_always )> = None; + // Snapshot routing state once for the entire preflight pass — avoids + // repeated lock acquisitions and keeps the gate consistent per turn. + let routing_snapshot = self.agent.deps.channel_routing.read().await.clone(); + let routing_builtin_names = if routing_snapshot.is_some() { + Some(self.agent.deps.tools.builtin_tool_names().await) + } else { + None + }; + for (idx, original_tc) in tool_calls.iter().enumerate() { let mut tc = original_tc.clone(); @@ -941,6 +984,26 @@ impl<'a> LoopDelegate for ChatDelegate<'a> { _ => {} } + // Channel routing execution-time gate — mirrors worker/job.rs::execute_tool_inner. + // apply_channel_routing filtered the LLM's tool list (presentation layer); + // this gate enforces the same policy at execution time so a jailbroken model + // or replayed tool call cannot run a tool that was hidden from the LLM. + if let (Some(config), Some(builtin_names)) = (&routing_snapshot, &routing_builtin_names) + && !config.is_tool_permitted( + &self.message.channel, + &self.message.metadata, + &tc.name, + builtin_names, + ) + { + let reject_msg = format!( + "Tool '{}' is not permitted on channel '{}' by channel routing config", + tc.name, self.message.channel + ); + preflight.push((tc, PreflightOutcome::Rejected(reject_msg))); + continue; + } + // Check if tool requires approval if !self.agent.config.auto_approve_tools && let Some(tool) = tool_opt @@ -968,7 +1031,7 @@ impl<'a> LoopDelegate for ChatDelegate<'a> { .and_then(|v| v.as_str()) == Some("direct_message"); if is_relay && !is_dm { - tracing::info!( + tracing::debug!( tool = %tc.name, channel = %self.message.channel, "Auto-denying approval-requiring tool in non-DM relay channel" @@ -2091,6 +2154,7 @@ mod tests { builder: None, llm_backend: "nearai".to_string(), tenant_rates: Arc::new(crate::tenant::TenantRateRegistry::new(4, 3)), + channel_routing: crate::agent::channel_routing::ChannelRoutingConfig::none_arc(), }; Agent::new( @@ -2400,6 +2464,7 @@ mod tests { builder: None, llm_backend: "nearai".to_string(), tenant_rates: Arc::new(crate::tenant::TenantRateRegistry::new(4, 3)), + channel_routing: crate::agent::channel_routing::ChannelRoutingConfig::none_arc(), }; let agent = Agent::new( @@ -3409,6 +3474,7 @@ mod tests { builder: None, llm_backend: "nearai".to_string(), tenant_rates: Arc::new(crate::tenant::TenantRateRegistry::new(4, 3)), + channel_routing: crate::agent::channel_routing::ChannelRoutingConfig::none_arc(), }; Agent::new( @@ -3558,6 +3624,7 @@ mod tests { builder: None, llm_backend: "nearai".to_string(), tenant_rates: Arc::new(crate::tenant::TenantRateRegistry::new(4, 3)), + channel_routing: crate::agent::channel_routing::ChannelRoutingConfig::none_arc(), }; let agent = Agent::new( @@ -3693,6 +3760,7 @@ mod tests { builder: None, llm_backend: "nearai".to_string(), tenant_rates: Arc::new(crate::tenant::TenantRateRegistry::new(4, 3)), + channel_routing: crate::agent::channel_routing::ChannelRoutingConfig::none_arc(), }; Agent::new( diff --git a/src/agent/mod.rs b/src/agent/mod.rs index c6377f69356..ff88b21b720 100644 --- a/src/agent/mod.rs +++ b/src/agent/mod.rs @@ -13,6 +13,7 @@ mod agent_loop; pub mod agentic_loop; mod attachments; +pub mod channel_routing; mod commands; pub mod compaction; pub mod context_monitor; diff --git a/src/agent/scheduler.rs b/src/agent/scheduler.rs index edfd4a3741a..32a4161f506 100644 --- a/src/agent/scheduler.rs +++ b/src/agent/scheduler.rs @@ -70,6 +70,9 @@ pub struct Scheduler { sse_tx: Option>, /// HTTP interceptor for trace recording/replay (propagated to workers). http_interceptor: Option>, + /// Cached channel routing config — shared with the SIGHUP handler for hot-reload. + channel_routing: + Arc>>, /// Running jobs (main LLM-driven jobs). jobs: Arc>>, /// Running sub-tasks (tool executions, background tasks). @@ -96,6 +99,7 @@ impl Scheduler { hooks: deps.hooks, sse_tx: None, http_interceptor: None, + channel_routing: crate::agent::channel_routing::ChannelRoutingConfig::none_arc(), jobs: Arc::new(RwLock::new(HashMap::new())), subtasks: Arc::new(RwLock::new(HashMap::new())), } @@ -114,6 +118,16 @@ impl Scheduler { self.http_interceptor = Some(interceptor); } + /// Wire the shared channel routing Arc so workers inherit the live SIGHUP-reloaded config. + pub fn set_channel_routing( + &mut self, + routing: Arc< + tokio::sync::RwLock>, + >, + ) { + self.channel_routing = routing; + } + /// Create, persist, and schedule a job in one shot. /// /// This is the preferred entry point for dispatching new jobs. It: @@ -314,6 +328,7 @@ impl Scheduler { approval_context, http_interceptor: self.http_interceptor.clone(), multi_tenant: self.config.multi_tenant, + channel_routing: Arc::clone(&self.channel_routing), }; let worker = Worker::new(job_id, deps); diff --git a/src/agent/thread_ops.rs b/src/agent/thread_ops.rs index d2d7a8e32ff..61fbf830363 100644 --- a/src/agent/thread_ops.rs +++ b/src/agent/thread_ops.rs @@ -1644,6 +1644,28 @@ impl Agent { ) .await; + // Re-verify channel routing — a hot-reload between approval-queue and + // approval-resume can narrow policy, and a directly-submitted ExecApproval + // JSON bypasses the preflight gate in execute_tool_calls entirely. + { + let routing = self.deps.channel_routing.read().await.clone(); + if let Some(config) = routing { + let builtin_names = self.tools().builtin_tool_names().await; + if !config.is_tool_permitted( + &message.channel, + &message.metadata, + &pending.tool_name, + &builtin_names, + ) { + return Ok(SubmissionResult::ok_with_message(format!( + "Tool '{}' is no longer permitted on channel '{}' \ + (channel routing policy changed since approval was queued).", + pending.tool_name, message.channel + ))); + } + } + } + let started_at = std::time::Instant::now(); let tool_result = self .execute_chat_tool(&pending.tool_name, &pending.parameters, &job_ctx) @@ -2930,6 +2952,7 @@ mod tests { builder: None, llm_backend: "nearai".to_string(), tenant_rates: Arc::new(crate::tenant::TenantRateRegistry::new(4, 3)), + channel_routing: crate::agent::channel_routing::ChannelRoutingConfig::none_arc(), }; let agent = Agent::new( @@ -3727,6 +3750,7 @@ mod tests { builder: None, llm_backend: "nearai".to_string(), tenant_rates: Arc::new(crate::tenant::TenantRateRegistry::new(4, 3)), + channel_routing: crate::agent::channel_routing::ChannelRoutingConfig::none_arc(), }; let agent = Agent::new( diff --git a/src/bridge/router.rs b/src/bridge/router.rs index 78d47128862..85920d9d68f 100644 --- a/src/bridge/router.rs +++ b/src/bridge/router.rs @@ -8574,6 +8574,7 @@ mod tests { builder: None, llm_backend: "nearai".to_string(), tenant_rates: Arc::new(crate::tenant::TenantRateRegistry::new(4, 3)), + channel_routing: Arc::new(tokio::sync::RwLock::new(None)), }; let channels = Arc::new(crate::channels::ChannelManager::new()); @@ -10354,6 +10355,7 @@ mod tests { builder: None, llm_backend: "nearai".to_string(), tenant_rates: Arc::new(crate::tenant::TenantRateRegistry::new(4, 3)), + channel_routing: Arc::new(tokio::sync::RwLock::new(None)), }; let agent = Agent::new( diff --git a/src/channels/relay/channel.rs b/src/channels/relay/channel.rs index 25827cb4b87..2d7ac90fc43 100644 --- a/src/channels/relay/channel.rs +++ b/src/channels/relay/channel.rs @@ -337,6 +337,8 @@ impl Channel for RelayChannel { "event_type": event.event_type, "thread_id": event.thread_id.as_deref().unwrap_or(&event.id), "provider": event.provider, + crate::agent::channel_routing::TRUSTED_DM_METADATA_KEY: event.event_type + == crate::channels::relay::client::event_types::DIRECT_MESSAGE, })); // Use the original thread_id if present (already in a thread), diff --git a/src/main.rs b/src/main.rs index 2a2f9983202..fa6be5e4fd5 100644 --- a/src/main.rs +++ b/src/main.rs @@ -834,6 +834,11 @@ async fn async_main() -> anyhow::Result<()> { let scheduler_slot: ironclaw::tools::builtin::SchedulerSlot = Arc::new(tokio::sync::RwLock::new(None)); + // Create channel routing Arc early so CreateJobTool can share it. + // Config is loaded from the DB later; the Arc is None until then but the + // tool reads it lazily at job creation time. + let channel_routing_arc = ironclaw::agent::channel_routing::ChannelRoutingConfig::none_arc(); + // Register job tools even under --cli-only so scheduler-backed jobs remain available. // Sandbox-only dependencies are injected only when the container manager is running. components.tools.register_job_tools( @@ -849,6 +854,7 @@ async fn async_main() -> anyhow::Result<()> { None }, components.secrets_store.clone(), + Some(Arc::clone(&channel_routing_arc)), ); // ── Gateway channel ──────────────────────────────────────────────── @@ -881,12 +887,13 @@ async fn async_main() -> anyhow::Result<()> { gw = gw.with_log_level_handle(Arc::clone(&log_level_handle)); gw = gw.with_tool_registry(Arc::clone(&components.tools)); if let Some(ref db) = components.db { - let dispatcher = Arc::new(ironclaw::tools::dispatch::ToolDispatcher::new( + let dispatcher = ironclaw::tools::dispatch::ToolDispatcher::new( Arc::clone(&components.tools), Arc::clone(&components.safety), Arc::clone(db), - )); - gw = gw.with_tool_dispatcher(dispatcher); + ) + .with_channel_routing(Arc::clone(&channel_routing_arc)); + gw = gw.with_tool_dispatcher(Arc::new(dispatcher)); } if let Some(ref ext_mgr) = components.extension_manager { // Enable gateway mode so MCP OAuth returns auth URLs to the frontend @@ -1266,6 +1273,17 @@ async fn async_main() -> anyhow::Result<()> { // Clone context_manager for the reaper before it's moved into Agent::new() let reaper_context_manager = Arc::clone(&components.context_manager); + // Load channel routing config from database before components.db is moved. + // Arc shared with register_job_tools (created earlier) and the SIGHUP handler. + if let Some(ref db) = components.db { + ironclaw::agent::channel_routing::ChannelRoutingConfig::reload_from_store( + db.as_ref(), + &config.owner_id, + &channel_routing_arc, + ) + .await; + } + // Capture settings store for SIGHUP handler before AppComponents is consumed. // Prefer the workspace-backed adapter (so SIGHUP-driven config reloads pick // up settings written through the workspace) and fall back to the raw db @@ -1333,6 +1351,7 @@ async fn async_main() -> anyhow::Result<()> { config.agent.max_llm_concurrent_per_user.unwrap_or(4), config.agent.max_jobs_concurrent_per_user.unwrap_or(3), )), + channel_routing: Arc::clone(&channel_routing_arc), }; let channels_for_warnings = Arc::clone(&channels); @@ -1386,6 +1405,7 @@ async fn async_main() -> anyhow::Result<()> { let sighup_settings_store_clone = sighup_settings_store.clone(); let sighup_secrets_store = components.secrets_store.clone(); let sighup_owner_id = config.owner_id.clone(); + let sighup_channel_routing = Arc::clone(&channel_routing_arc); let mut shutdown_rx = shutdown_tx.subscribe(); tokio::spawn(async move { @@ -1409,7 +1429,7 @@ async fn async_main() -> anyhow::Result<()> { // Handle SIGHUP signal } } - tracing::info!("SIGHUP received — reloading HTTP webhook config"); + tracing::debug!("SIGHUP received — reloading HTTP webhook config"); // Flush settings cache so direct DB edits are picked up. if let Some(ref cache) = sighup_settings_cache { @@ -1544,6 +1564,22 @@ async fn async_main() -> anyhow::Result<()> { updater.update_secret(new_secret.clone()).await; } } + + // Hot-reload channel routing config from SettingsStore + if let Some(ref store) = sighup_settings_store_clone { + let changed = + ironclaw::agent::channel_routing::ChannelRoutingConfig::reload_from_store( + store.as_ref(), + &sighup_owner_id, + &sighup_channel_routing, + ) + .await; + if changed { + tracing::debug!("SIGHUP: channel routing config reloaded"); + } else { + tracing::debug!("SIGHUP: channel routing config unchanged"); + } + } } }); } diff --git a/src/tenant.rs b/src/tenant.rs index b7ccde001c5..7e59a656309 100644 --- a/src/tenant.rs +++ b/src/tenant.rs @@ -564,6 +564,19 @@ impl SystemScope { Self { inner: db } } + /// Read the channel routing config for a specific user. + /// + /// Typed wrapper over the settings store so callers don't need to know the + /// raw settings key. Used by `ChannelRoutingConfig::load_from_system_scope` + /// for autonomous workers that hold a `SystemScope` rather than a raw + /// `SettingsStore`. + pub async fn get_channel_routing( + &self, + user_id: &str, + ) -> Result, DatabaseError> { + self.inner.get_setting(user_id, "channel_routing").await + } + /// Construct a per-user workspace for system-process operations. /// /// Used by the heartbeat and routine engine to get a workspace scoped to diff --git a/src/testing/mod.rs b/src/testing/mod.rs index ba3b2867e91..b56204d8dd3 100644 --- a/src/testing/mod.rs +++ b/src/testing/mod.rs @@ -394,6 +394,7 @@ impl TestHarnessBuilder { builder: None, llm_backend: "nearai".to_string(), tenant_rates: std::sync::Arc::new(crate::tenant::TenantRateRegistry::new(4, 3)), + channel_routing: crate::agent::channel_routing::ChannelRoutingConfig::none_arc(), }; TestHarness { diff --git a/src/tools/builtin/job.rs b/src/tools/builtin/job.rs index cb1255d5f22..b5151c67e0e 100644 --- a/src/tools/builtin/job.rs +++ b/src/tools/builtin/job.rs @@ -15,6 +15,7 @@ use chrono::Utc; use tokio::sync::RwLock; use uuid::Uuid; +use crate::agent::channel_routing::ChannelRoutingConfig; use crate::bootstrap::ironclaw_base_dir; use crate::channels::IncomingMessage; use crate::context::{ContextManager, JobContext, JobState}; @@ -93,6 +94,10 @@ pub struct CreateJobTool { inject_tx: Option>, /// Encrypted secrets store for validating credential grants. secrets_store: Option>, + /// Cached channel routing config (shared with dispatcher and SIGHUP handler). + /// When present, avoids a per-job DB read and guarantees the same version + /// seen by the dispatcher is used for MCP server scoping. + channel_routing: Option>>>, } impl CreateJobTool { @@ -105,6 +110,7 @@ impl CreateJobTool { event_tx: None, inject_tx: None, secrets_store: None, + channel_routing: None, } } @@ -143,6 +149,20 @@ impl CreateJobTool { self } + /// Inject the shared channel routing Arc (from AgentDeps). + /// + /// When present, `derive_routed_mcp_servers` reads from this cached value + /// instead of issuing a fresh DB query per job creation, avoiding extra + /// round-trips and ensuring the job inherits the same routing version as + /// the dispatcher turn that created it. + pub fn with_channel_routing( + mut self, + routing: Arc>>, + ) -> Self { + self.channel_routing = Some(routing); + self + } + pub fn sandbox_enabled(&self) -> bool { self.job_manager.is_some() } @@ -267,6 +287,81 @@ impl CreateJobTool { crate::tools::mcp::config::load_master_mcp_config_value(store.as_ref(), user_id).await } + /// Derive an MCP server allowlist for a job based on its origin channel. + /// + /// Security model: + /// - `None` return means **no constraint** (DM bypass or no routing config) — + /// the job inherits the full unfiltered MCP server set. + /// - `Some(servers)` means the job is constrained to those server prefixes. + /// - An absent channel (routine/heartbeat) falls back to the default group + /// (fail-closed), never `None`. + /// - Caller-supplied `explicit_mcp_servers` are **intersected** with the + /// channel's allowlist — a child job cannot request servers its parent + /// channel forbids (privilege escalation by delegation). + /// + /// Uses the cached routing Arc when available (no extra DB round-trip). + async fn derive_routed_mcp_servers( + &self, + ctx: &JobContext, + explicit_mcp_servers: Option>, + ) -> Option> { + // Read from cached Arc first; fall back to a DB query when not wired. + let routing = if let Some(ref arc) = self.channel_routing { + arc.read().await.clone() + } else { + let store = self.store.as_ref()?; + ChannelRoutingConfig::load_from_store(store.as_ref(), &ctx.user_id).await + }; + let Some(routing) = routing else { + // No routing config — no constraint; return explicit as-is. + return explicit_mcp_servers; + }; + + let null_metadata = serde_json::Value::Null; + let routing_metadata = ctx + .metadata + .get("notify_metadata") + .filter(|value| value.is_object()) + .unwrap_or(&null_metadata); + // Route by adapter channel name so is_dm() sees "slack-relay" / + // "telegram", not the platform-specific ID in notify_metadata.channel. + let routing_channel = ctx.metadata.get("notify_channel").and_then(|v| v.as_str()); + + // Shared group resolution + DM bypass (same logic as the tool filters). + match routing.routing_group_for(routing_channel, routing_metadata) { + // DM — no MCP constraint; the job inherits the full unfiltered set. + // Explicit servers pass through unchanged. + None => explicit_mcp_servers, + // Constrained to the resolved group's servers. Explicit `mcp_servers` + // from the tool call are **intersected** with the channel allowlist — + // a child job cannot request servers its parent channel forbids. + Some(group) => { + let group_allowed = routing + .groups + .get(group) + .or_else(|| routing.groups.get(&routing.default_group)) + .cloned(); + match (explicit_mcp_servers, group_allowed) { + // No explicit request, group not in map — no constraint. + (explicit, None) => explicit, + // No explicit request — use routing-derived set. + (None, Some(allowed)) => Some(allowed), + // Explicit request — intersect; cannot widen past allowlist. + (Some(explicit), Some(allowed)) => { + let allowed_set: std::collections::HashSet<&str> = + allowed.iter().map(|s| s.as_str()).collect(); + Some( + explicit + .into_iter() + .filter(|s| allowed_set.contains(s.as_str())) + .collect(), + ) + } + } + } + } + } + /// Persist a sandbox job record (fire-and-forget). fn persist_job(&self, record: SandboxJobRecord) { if let Some(store) = self.store.clone() { @@ -361,8 +456,25 @@ impl CreateJobTool { if let Some(ref slot) = self.scheduler_slot && let Some(ref scheduler) = *slot.read().await { + // Propagate origin channel context so the spawned worker can apply + // channel routing. Without this, jobs dispatched from restricted + // channels lose their origin and run with the full tool set. + let channel_metadata = { + let mut m = serde_json::Map::new(); + if let Some(ch) = ctx.metadata.get("notify_channel") { + m.insert("notify_channel".to_string(), ch.clone()); + } + if let Some(meta) = ctx.metadata.get("notify_metadata") { + m.insert("notify_metadata".to_string(), meta.clone()); + } + if m.is_empty() { + None + } else { + Some(serde_json::Value::Object(m)) + } + }; return match scheduler - .dispatch_job(&ctx.user_id, title, description, None) + .dispatch_job(&ctx.user_id, title, description, channel_metadata) .await { Ok(job_id) => { @@ -828,16 +940,13 @@ fn resolve_project_dir( } fn monitor_route_from_ctx(ctx: &JobContext) -> Option { - // notify_channel is required — without it we don't know which channel to - // route the monitor output to, so return None to skip monitoring entirely. + // Without notify_channel we cannot route monitor output, so skip monitoring. let channel = ctx .metadata .get("notify_channel") .and_then(|v| v.as_str())? .to_string(); - // notify_user is optional — fall back to the job's own user_id, which is - // always present. The channel is the routing decision; the user is just - // for attribution and can default safely. + // notify_user is optional; fall back to the job user for attribution. let user_id = ctx .metadata .get("notify_user") @@ -1043,7 +1152,7 @@ impl Tool for CreateJobTool { // Parse optional MCP server filter and iteration cap. // Validate types: warn if present but wrong type so callers know why it was ignored. - let mcp_servers: Option> = match params.get("mcp_servers") { + let explicit_mcp_servers: Option> = match params.get("mcp_servers") { Some(v) if v.is_array() => v.as_array().map(|arr| { arr.iter() .filter_map(|v| v.as_str().map(String::from)) @@ -1055,6 +1164,9 @@ impl Tool for CreateJobTool { } None => None, }; + let mcp_servers = self + .derive_routed_mcp_servers(ctx, explicit_mcp_servers) + .await; let max_iterations: Option = match params.get("max_iterations") { Some(v) if v.is_u64() || v.is_i64() => v.as_u64().map(|n| n.clamp(1, 500) as u32), Some(_) => { @@ -2614,4 +2726,45 @@ mod tests { "description should not mention claude_code when mode is disabled, got: {desc}" ); } + + /// Regression: `derive_routed_mcp_servers` must intersect caller-supplied + /// `mcp_servers` with the channel's routing allowlist — a caller cannot + /// widen MCP access past what the channel group permits. + #[tokio::test] + async fn test_derive_routed_mcp_servers_intersects_with_channel_allowlist() { + let json = r#"{ + "groups": { "research": ["Archon", "Serpstat"] }, + "channels": { "research-channel": "research" }, + "default_group": "research" + }"#; + let config: ChannelRoutingConfig = serde_json::from_str(json).unwrap(); + let arc = Arc::new(RwLock::new(Some(config))); + + let manager = Arc::new(ContextManager::new(5)); + let tool = CreateJobTool::new(manager).with_channel_routing(arc); + + let ctx = JobContext { + metadata: serde_json::json!({ "notify_channel": "research-channel" }), + ..Default::default() + }; + + // Caller supplies Serpstat (allowed) + Kiro (not in research group). + // Only Serpstat should survive the intersection. + let result = tool + .derive_routed_mcp_servers(&ctx, Some(vec!["Serpstat".to_string(), "Kiro".to_string()])) + .await; + assert_eq!(result, Some(vec!["Serpstat".to_string()])); + + // Caller supplies only out-of-group server → empty list, not the full group. + let result_empty = tool + .derive_routed_mcp_servers(&ctx, Some(vec!["Kiro".to_string()])) + .await; + assert_eq!(result_empty, Some(vec![])); + + // No explicit servers → falls back to routing-derived set (both group members). + let result_derived = tool.derive_routed_mcp_servers(&ctx, None).await; + let mut derived = result_derived.unwrap_or_default(); + derived.sort(); + assert_eq!(derived, vec!["Archon".to_string(), "Serpstat".to_string()]); + } } diff --git a/src/tools/dispatch.rs b/src/tools/dispatch.rs index 663cc45953f..c667d341927 100644 --- a/src/tools/dispatch.rs +++ b/src/tools/dispatch.rs @@ -20,6 +20,7 @@ use std::time::Instant; use tracing::debug; use uuid::Uuid; +use crate::agent::channel_routing::ChannelRoutingConfig; use crate::context::{ActionRecord, JobContext}; use crate::db::Database; use crate::tools::registry::ToolRegistry; @@ -62,10 +63,29 @@ impl std::fmt::Display for DispatchSource { /// same safety pipeline as the agent worker (param normalization, schema /// validation, sensitive-param redaction, per-tool timeout, output /// sanitization) plus `ActionRecord` persistence. +/// +/// When a `channel_routing` config is wired in, `dispatch()` enforces +/// channel routing at execution time for `DispatchSource::Channel` callers +/// — this is the security boundary that prevents a jailbroken LLM or +/// misbehaving gateway handler from executing a filtered tool by naming it +/// directly. The presentation-layer filter (`apply_channel_routing`) hides +/// tools from the LLM; dispatch-time enforcement is the authoritative gate. +/// +/// **Scope of this gate.** `dispatch()` covers only non-agent callers routed +/// through it (gateway handlers, CLI, routine engine). The agent's own +/// LLM-driven worker loop does **not** go through `dispatch()` — it calls +/// `Tool::execute` via `crate::worker::job::JobWorker::execute_tool_inner`, +/// which carries the *parallel* execution-time gate (same +/// [`ChannelRoutingConfig::is_tool_permitted`] check, fed the job's +/// `notify_channel` / `notify_metadata`). Both paths must keep their gate in +/// sync with the presentation filter so a tool the LLM can see is exactly a +/// tool it can execute. pub struct ToolDispatcher { registry: Arc, safety: Arc, store: Arc, + /// Optional channel routing config for execution-time enforcement. + channel_routing: Option>>>, } impl ToolDispatcher { @@ -79,9 +99,31 @@ impl ToolDispatcher { registry, safety, store, + channel_routing: None, } } + /// Wire in the shared channel routing Arc for execution-time enforcement. + /// + /// **Scope (v1 agent loop).** This gate covers `ToolDispatcher::dispatch` + /// (gateway/CLI non-agent callers), the chat-path preflight + /// (`ChatDelegate::execute_tool_calls`), the approval-resume path + /// (`Agent::process_approval`), and the worker loop + /// (`JobDelegate::execute_tool_inner`). + /// + /// **Intentionally out of scope:** + /// - Engine v2 (`EffectBridgeAdapter::execute_action`) — has its own + /// event-sourced audit trail; channel routing integration is a v2 follow-up. + /// - `Scheduler::execute_tool_task` (subtask/`ToolExec`) — operator-initiated + /// subtasks run outside the user's channel context. + pub fn with_channel_routing( + mut self, + routing: Arc>>, + ) -> Self { + self.channel_routing = Some(routing); + self + } + /// Execute a tool by name with the given parameters. /// /// Pipeline (mirrors `Worker::execute_tool`): @@ -125,6 +167,35 @@ impl ToolDispatcher { ToolError::ExecutionFailed(format!("tool not found: {tool_name}")) })?; + // 0. Channel routing enforcement — execution-time gate. + // Complements the presentation-layer filter in `apply_channel_routing` + // (which hides tools from the LLM). This check enforces the boundary + // for any caller that names a tool string directly, regardless of + // what the LLM saw. Only applied for Channel sources — Routine and + // System callers are internal and bypass routing. + if let DispatchSource::Channel(ref channel) = source + && let Some(ref routing_arc) = self.channel_routing + { + let routing = routing_arc.read().await; + if let Some(ref config) = *routing { + let builtin_names = self.registry.builtin_tool_names().await; + // `DispatchSource::Channel` carries no per-message metadata, so + // pass `Null` — only `DM_EXACT` channels bypass here. The worker + // agent loop has its own gate (see `crate::worker::job`) which + // passes real `notify_metadata`. + if !config.is_tool_permitted( + channel, + &serde_json::Value::Null, + &resolved_name, + &builtin_names, + ) { + return Err(ToolError::ExecutionFailed(format!( + "tool '{resolved_name}' is not available on channel '{channel}'" + ))); + } + } + } + // 1. Normalize parameters (coerce types, fill defaults). let normalized_params = prepare_tool_params(tool.as_ref(), ¶ms); @@ -814,4 +885,126 @@ mod integration_tests { "unprotected target must succeed; got: {result:?}" ); } + + // ── Channel routing enforcement (dispatch-time) ────────── + // + // Drives `ToolDispatcher::dispatch()` with a channel routing config that + // restricts a channel to a specific MCP server prefix. Asserts that a + // tool outside the allowlist is rejected at execution time, not just + // hidden from the LLM at the presentation layer. Per + // `.claude/rules/testing.md` ("Test Through the Caller") — the unit test + // for `is_tool_permitted` alone doesn't cover the dispatcher wiring. + + struct RestrictedMcpTool; + + #[async_trait] + impl Tool for RestrictedMcpTool { + fn name(&self) -> &str { + "Smartlead_send_email" + } + fn description(&self) -> &str { + "Stub representing a restricted MCP tool." + } + fn parameters_schema(&self) -> serde_json::Value { + serde_json::json!({ "type": "object" }) + } + async fn execute( + &self, + _params: serde_json::Value, + _ctx: &JobContext, + ) -> Result { + Ok(ToolOutput::success( + serde_json::json!({ "sent": true }), + Duration::from_millis(1), + )) + } + } + + #[tokio::test] + async fn dispatch_rejects_filtered_tool_at_execution_time() { + // Build a routing config that restricts "research" channel to Archon + // only — Smartlead_send_email is outside the allowlist. + let config_json = serde_json::json!({ + "groups": { + "research": ["Archon"] + }, + "builtin_whitelist": {}, + "channels": { + "research-channel": "research" + }, + "default_group": "research" + }); + let routing_config: ChannelRoutingConfig = + serde_json::from_value(config_json).expect("valid routing config"); + let routing_arc = ChannelRoutingConfig::none_arc(); + *routing_arc.write().await = Some(routing_config); + + let (dispatcher, _backend, _db, registry, _dir) = test_dispatcher().await; + // Use async register (not register_sync) to simulate an MCP tool. + // register_sync adds to builtin_tool_names; builtins default-allow when no + // whitelist is configured for the group, which would make the assertion fail. + registry.register(Arc::new(RestrictedMcpTool)).await; + + let dispatcher = ToolDispatcher::new( + Arc::clone(dispatcher.registry()), + Arc::new(SafetyLayer::new(&SafetyConfig { + max_output_length: 65_536, + injection_check_enabled: false, + })), + _db, + ) + .with_channel_routing(routing_arc); + + // Attempt to dispatch the filtered tool from the restricted channel. + let result = dispatcher + .dispatch( + "Smartlead_send_email", + serde_json::json!({}), + "tester", + DispatchSource::Channel("research-channel".into()), + ) + .await; + + assert!( + matches!(result, Err(ToolError::ExecutionFailed(ref msg)) if msg.contains("not available on channel")), + "filtered tool must be rejected at dispatch time; got {result:?}" + ); + + // Same tool on a DM channel (gateway = DM_EXACT) must pass through. + let (dispatcher2, _backend2, _db2, registry2, _dir2) = test_dispatcher().await; + let routing_arc2 = ChannelRoutingConfig::none_arc(); + *routing_arc2.write().await = Some( + serde_json::from_value(serde_json::json!({ + "groups": { "research": ["Archon"] }, + "builtin_whitelist": {}, + "channels": {}, + "default_group": "research" + })) + .unwrap(), + ); + registry2.register(Arc::new(RestrictedMcpTool)).await; + let dispatcher2 = ToolDispatcher::new( + Arc::clone(dispatcher2.registry()), + Arc::new(SafetyLayer::new(&SafetyConfig { + max_output_length: 65_536, + injection_check_enabled: false, + })), + _db2, + ) + .with_channel_routing(routing_arc2); + + // "gateway" is in DM_EXACT — routing is bypassed entirely. + let dm_result = dispatcher2 + .dispatch( + "Smartlead_send_email", + serde_json::json!({}), + "tester", + DispatchSource::Channel("gateway".into()), + ) + .await; + assert!( + dm_result.is_ok(), + "DM_EXACT channel must bypass dispatch-time routing; got {dm_result:?}" + ); + } } diff --git a/src/tools/registry.rs b/src/tools/registry.rs index a3db7bd36b0..0a13da0dad0 100644 --- a/src/tools/registry.rs +++ b/src/tools/registry.rs @@ -640,6 +640,9 @@ impl ToolRegistry { inject_tx: Option>, prompt_queue: Option, secrets_store: Option>, + channel_routing: Option< + Arc>>, + >, ) { let mut create_tool = CreateJobTool::new(Arc::clone(&context_manager)); if let Some(slot) = scheduler_slot { @@ -657,6 +660,9 @@ impl ToolRegistry { if let Some(secrets) = secrets_store { create_tool = create_tool.with_secrets(secrets); } + if let Some(routing) = channel_routing { + create_tool = create_tool.with_channel_routing(routing); + } self.register_sync(Arc::new(create_tool)); self.register_sync(Arc::new(ListJobsTool::new(Arc::clone(&context_manager)))); self.register_sync(Arc::new(JobStatusTool::new(Arc::clone(&context_manager)))); diff --git a/src/worker/job.rs b/src/worker/job.rs index f99335d51ed..20fc073c2b1 100644 --- a/src/worker/job.rs +++ b/src/worker/job.rs @@ -37,6 +37,8 @@ use ironclaw_llm::{ }; use ironclaw_safety::SafetyLayer; +use crate::agent::channel_routing::ChannelRoutingConfig; + /// Shared dependencies for worker execution. /// /// This bundles the dependencies that are shared across all workers, @@ -61,6 +63,9 @@ pub struct WorkerDeps { pub http_interceptor: Option>, /// Whether the deployment is multi-tenant (used for admin tool policy filtering). pub multi_tenant: bool, + /// Cached channel routing config — same Arc shared with the SIGHUP handler so + /// workers see live config without per-iteration DB queries. + pub channel_routing: Arc>>, } /// Worker that executes a single job. @@ -110,6 +115,60 @@ impl Worker { self.deps.use_planning } + async fn tool_definitions_for_current_context(&self) -> Vec { + let tool_defs = self.tools().tool_definitions().await; + + let Ok(job_ctx) = self.context_manager().get_context(self.job_id).await else { + return tool_defs; + }; + + // Extract origin channel — may be absent for jobs spawned by routines/heartbeats. + let channel = job_ctx + .metadata + .get("notify_channel") + .and_then(|v| v.as_str()); + + let null_metadata = serde_json::Value::Null; + let routing_metadata = job_ctx + .metadata + .get("notify_metadata") + .filter(|value| value.is_object()) + .unwrap_or(&null_metadata); + + // Use the cached Arc — same version as the SIGHUP-aware dispatcher, + // no extra DB round-trip per iteration. + let routing = self.deps.channel_routing.read().await.clone(); + let Some(routing) = routing else { + return tool_defs; + }; + + // Resolve routing channel: use origin channel if present; fall back to + // empty string so resolve_group() maps to the default group (fail-closed). + // The empty-string sentinel is intentional — `is_dm("")` returns false, so + // routine/heartbeat-spawned jobs always get the default group, never DM bypass. + let effective_channel = channel.unwrap_or(""); + + let builtin_names = self.tools().builtin_tool_names().await; + let before = tool_defs.len(); + let filtered = routing.filter_tool_defs( + effective_channel, + routing_metadata, + tool_defs, + &builtin_names, + ); + if filtered.len() < before { + tracing::debug!( + job_id = %self.job_id, + channel = effective_channel, + group = routing.resolve_group(effective_channel), + before, + after = filtered.len(), + "Job channel routing filtered tools" + ); + } + filtered + } + /// Fire-and-forget persistence of job status. fn persist_status(&self, status: JobState, reason: Option) { if let Some(store) = self.store() { @@ -339,8 +398,9 @@ Report when the job is complete or if you encounter issues you cannot resolve."# .unwrap_or(50) as usize; let max_iterations = max_iterations.min(ironclaw_common::MAX_WORKER_ITERATIONS as usize); - // Initial tool definitions for planning (will be refreshed in loop) - reason_ctx.available_tools = self.tools().tool_definitions().await; + // Initial tool definitions for planning — use the same filtered set + // that execute_tool_calls sees, so the plan only references permitted tools. + reason_ctx.available_tools = self.tool_definitions_for_current_context().await; // Generate plan if planning is enabled let plan = if self.use_planning() { @@ -527,6 +587,47 @@ Report when the job is complete or if you encounter issues you cannot resolve."# // Fetch job context early for approval checking and other needs let mut job_ctx = deps.context_manager.get_context(job_id).await?; + // Channel routing enforcement — execution-time gate (defense in depth). + // + // The agent's LLM-driven worker loop does NOT go through + // `ToolDispatcher::dispatch`, so the dispatcher's gate doesn't cover it. + // `tool_definitions_for_current_context` already filters the tool list + // the LLM sees, but that's a presentation-layer hint: a jailbroken or + // confused model can still name a hidden tool directly. Block it here + // before execution. Channel + metadata are read from the same job-context + // fields the presentation filter uses, so the two stay consistent + // (including DM bypass). builtin names are fetched before taking the + // routing read lock so the lock is only held across the sync check. + let builtin_names = deps.tools.builtin_tool_names().await; + if let Some(config) = deps.channel_routing.read().await.as_ref() { + let routing_channel = job_ctx + .metadata + .get("notify_channel") + .and_then(|v| v.as_str()) + .unwrap_or(""); + let null_metadata = serde_json::Value::Null; + let routing_metadata = job_ctx + .metadata + .get("notify_metadata") + .filter(|value| value.is_object()) + .unwrap_or(&null_metadata); + if !config.is_tool_permitted( + routing_channel, + routing_metadata, + tool_name, + &builtin_names, + ) { + return Err(crate::error::ToolError::AutonomousUnavailable { + name: tool_name.to_string(), + reason: format!( + "Tool '{}' is not permitted on channel '{}' by channel routing", + tool_name, routing_channel + ), + } + .into()); + } + } + // Check approval: additive semantics - BOTH job-level AND worker-level must approve let requirement = tool.requires_approval(&normalized_params); @@ -1437,7 +1538,7 @@ impl<'a> LoopDelegate for JobDelegate<'a> { reason_ctx.available_tools.clear(); } else { // Refresh tool definitions so newly built tools become visible - let tool_defs = self.worker.tools().tool_definitions().await; + let tool_defs = self.worker.tool_definitions_for_current_context().await; // Apply admin tool policy filtering (multi-tenant only). let (user_id, is_admin) = self.resolve_user_info().await; @@ -1839,9 +1940,11 @@ impl From for Result { #[cfg(test)] mod tests { + use std::collections::HashSet; use std::sync::Arc; use crate::channels::ChannelManager; + use crate::db::Database; use ironclaw_llm::ToolSelection; use super::*; @@ -1941,6 +2044,7 @@ mod tests { approval_context: None, http_interceptor: None, multi_tenant: false, + channel_routing: ChannelRoutingConfig::none_arc(), }; Worker::new(job_id, deps) @@ -2161,6 +2265,7 @@ mod tests { approval_context, http_interceptor: None, multi_tenant: false, + channel_routing: ChannelRoutingConfig::none_arc(), }; Worker::new(job_id, deps) @@ -2766,6 +2871,226 @@ mod tests { assert_eq!(telegram[0].1.content, "hello from routine"); } + // TODO(postgres-parity): no symmetric test for the postgres backend per + // .claude/rules/database.md — dual-backend features need testcontainers + // or #[cfg(feature = "integration")] coverage. Tracked as tech debt; + // the libsql test below covers the shared filtering logic in + // tool_definitions_for_current_context, which is backend-agnostic. + #[cfg(feature = "libsql")] + #[tokio::test] + async fn routed_jobs_filter_tools_by_originating_channel() { + let dir = tempfile::tempdir().unwrap(); + let backend = crate::db::libsql::LibSqlBackend::new_local(&dir.path().join("test.db")) + .await + .unwrap(); + backend.run_migrations().await.unwrap(); + + let routing: crate::agent::channel_routing::ChannelRoutingConfig = + serde_json::from_value(serde_json::json!({ + "groups": { + "content": ["Notion"] + }, + "builtin_whitelist": { + "content": ["memory_search"] + }, + "channels": { + "telegram": "content" + }, + "default_group": "content" + })) + .unwrap(); + routing.save_to_store(&backend, "user-1").await.unwrap(); + + // Populate the routing Arc from the store so the filter actually + // activates inside tool_definitions_for_current_context. + // none_arc() starts as None → the early return would skip filtering. + let channel_routing = ChannelRoutingConfig::none_arc(); + ChannelRoutingConfig::reload_from_store(&backend, "user-1", &channel_routing).await; + + let registry = ToolRegistry::new(); + // MCP tools registered dynamically (async — not in builtin_tool_names). + registry + .register(Arc::new(SlowTool { + tool_name: "Notion_post_search".to_string(), + delay: Duration::ZERO, + })) + .await; + registry + .register(Arc::new(SlowTool { + tool_name: "Archon_search".to_string(), + delay: Duration::ZERO, + })) + .await; + // Built-in tool registered at startup (register_sync — in builtin_tool_names). + // Matches production: memory_search is a startup built-in, not a dynamic MCP tool. + registry.register_sync(Arc::new(SlowTool { + tool_name: "memory_search".to_string(), + delay: Duration::ZERO, + })); + + let cm = Arc::new(crate::context::ContextManager::new(5)); + let job_id = cm.create_job("test", "test routed job").await.unwrap(); + cm.update_context(job_id, |ctx| { + ctx.user_id = "user-1".to_string(); + ctx.metadata = serde_json::json!({ + "notify_channel": "telegram", + "notify_metadata": { + "chat_type": "group" + } + }); + Ok::<(), String>(()) + }) + .await + .unwrap() + .unwrap(); + + let db: Arc = Arc::new(backend); + let worker = Worker::new( + job_id, + WorkerDeps { + context_manager: cm, + llm: Arc::new(StubLlm), + safety: Arc::new(SafetyLayer::new(&SafetyConfig { + max_output_length: 100_000, + injection_check_enabled: false, + })), + tools: Arc::new(registry), + store: Some(crate::tenant::SystemScope::new(db)), + hooks: Arc::new(crate::hooks::HookRegistry::new()), + timeout: Duration::from_secs(30), + use_planning: false, + sse_tx: None, + approval_context: None, + http_interceptor: None, + multi_tenant: false, + channel_routing, + }, + ); + + let names: HashSet<_> = worker + .tool_definitions_for_current_context() + .await + .into_iter() + .map(|tool| tool.name) + .collect(); + + assert!(names.contains("Notion_post_search")); + assert!(names.contains("memory_search")); + assert!(!names.contains("Archon_search")); + } + + /// Regression test for the worker execution-time routing gate. + /// + /// The presentation filter (`tool_definitions_for_current_context`, tested + /// above) only hides tools from the LLM. This drives the actual execution + /// path (`execute_tool` → `execute_tool_inner`) to prove a tool routing + /// blocked is *rejected at execution time* even when named directly — the + /// case a jailbroken LLM exploits. Tools allowed by routing still run. + #[cfg(feature = "libsql")] + #[tokio::test] + async fn routed_jobs_block_filtered_tool_at_execution_time() { + let dir = tempfile::tempdir().unwrap(); + let backend = crate::db::libsql::LibSqlBackend::new_local(&dir.path().join("test.db")) + .await + .unwrap(); + backend.run_migrations().await.unwrap(); + + // content group: only Notion MCP + memory_search built-in. + let routing: crate::agent::channel_routing::ChannelRoutingConfig = + serde_json::from_value(serde_json::json!({ + "groups": { "content": ["Notion"] }, + "builtin_whitelist": { "content": ["memory_search"] }, + "channels": { "telegram": "content" }, + "default_group": "content" + })) + .unwrap(); + routing.save_to_store(&backend, "user-1").await.unwrap(); + let channel_routing = ChannelRoutingConfig::none_arc(); + ChannelRoutingConfig::reload_from_store(&backend, "user-1", &channel_routing).await; + + let registry = ToolRegistry::new(); + registry + .register(Arc::new(SlowTool { + tool_name: "Notion_post_search".to_string(), + delay: Duration::ZERO, + })) + .await; + registry + .register(Arc::new(SlowTool { + tool_name: "Archon_search".to_string(), + delay: Duration::ZERO, + })) + .await; + registry.register_sync(Arc::new(SlowTool { + tool_name: "memory_search".to_string(), + delay: Duration::ZERO, + })); + + let cm = Arc::new(crate::context::ContextManager::new(5)); + let job_id = cm + .create_job("test", "test routed exec gate") + .await + .unwrap(); + cm.update_context(job_id, |ctx| { + ctx.user_id = "user-1".to_string(); + // Group message (not a DM) so routing applies. + ctx.metadata = serde_json::json!({ + "notify_channel": "telegram", + "notify_metadata": { "chat_type": "group" } + }); + Ok::<(), String>(()) + }) + .await + .unwrap() + .unwrap(); + + let db: Arc = Arc::new(backend); + let worker = Worker::new( + job_id, + WorkerDeps { + context_manager: cm, + llm: Arc::new(StubLlm), + safety: Arc::new(SafetyLayer::new(&SafetyConfig { + max_output_length: 100_000, + injection_check_enabled: false, + })), + tools: Arc::new(registry), + store: Some(crate::tenant::SystemScope::new(db)), + hooks: Arc::new(crate::hooks::HookRegistry::new()), + timeout: Duration::from_secs(30), + use_planning: false, + sse_tx: None, + approval_context: None, + http_interceptor: None, + multi_tenant: false, + channel_routing, + }, + ); + + let params = serde_json::json!({}); + + // Allowed by routing → executes. + assert!( + worker + .execute_tool("Notion_post_search", ¶ms) + .await + .is_ok(), + "Notion is in the content group and must execute" + ); + assert!( + worker.execute_tool("memory_search", ¶ms).await.is_ok(), + "memory_search is whitelisted and must execute" + ); + + // Hidden from the LLM AND must be rejected when named directly. + let blocked = worker.execute_tool("Archon_search", ¶ms).await; + let err = blocked.expect_err("Archon is not in the content group; must be blocked"); + assert!( + err.to_string().contains("channel routing"), + "expected a channel-routing rejection, got: {err}" + ); + } + /// Regression test: only `EmptyResponse` errors are eligible for /// completion-swallowing. Infrastructure errors must always propagate. #[test] diff --git a/tests/e2e_builtin_tool_coverage.rs b/tests/e2e_builtin_tool_coverage.rs index 808223217de..7a728a845d3 100644 --- a/tests/e2e_builtin_tool_coverage.rs +++ b/tests/e2e_builtin_tool_coverage.rs @@ -275,6 +275,7 @@ mod tests { let rig = TestRigBuilder::new() .with_trace(trace.clone()) .with_auto_approve_tools(true) + .with_local_jobs_only() .build() .await; @@ -316,6 +317,7 @@ mod tests { let rig = TestRigBuilder::new() .with_trace(trace.clone()) .with_auto_approve_tools(true) + .with_local_jobs_only() .build() .await; @@ -353,6 +355,7 @@ mod tests { let rig = TestRigBuilder::new() .with_trace(trace.clone()) .with_auto_approve_tools(true) + .with_local_jobs_only() .build() .await; @@ -394,6 +397,7 @@ mod tests { let rig = TestRigBuilder::new() .with_trace(trace.clone()) .with_auto_approve_tools(true) + .with_local_jobs_only() .build() .await; @@ -435,6 +439,7 @@ mod tests { let rig = TestRigBuilder::new() .with_trace(trace.clone()) .with_auto_approve_tools(true) + .with_local_jobs_only() .build() .await; @@ -472,6 +477,7 @@ mod tests { let rig = TestRigBuilder::new() .with_trace(trace.clone()) .with_auto_approve_tools(true) + .with_local_jobs_only() .build() .await; @@ -952,6 +958,7 @@ mod tests { let rig = TestRigBuilder::new() .with_trace(trace.clone()) .with_auto_approve_tools(true) + .with_local_jobs_only() .build() .await; diff --git a/tests/e2e_telegram_message_routing.rs b/tests/e2e_telegram_message_routing.rs index 5e9ce7b98fe..bafd0abb68b 100644 --- a/tests/e2e_telegram_message_routing.rs +++ b/tests/e2e_telegram_message_routing.rs @@ -204,6 +204,7 @@ mod tests { builder: None, llm_backend: "nearai".to_string(), tenant_rates: std::sync::Arc::new(ironclaw::tenant::TenantRateRegistry::new(4, 3)), + channel_routing: std::sync::Arc::new(tokio::sync::RwLock::new(None)), }; let gateway = Arc::new(TestChannel::new()); diff --git a/tests/support/gateway_workflow_harness.rs b/tests/support/gateway_workflow_harness.rs index 71e35f6f5ed..a0617f96db0 100644 --- a/tests/support/gateway_workflow_harness.rs +++ b/tests/support/gateway_workflow_harness.rs @@ -186,6 +186,7 @@ impl GatewayWorkflowHarness { None, None, None, + None, ); // Agent::run() creates its own RoutineEngine and populates this slot. @@ -289,6 +290,7 @@ impl GatewayWorkflowHarness { builder: None, llm_backend: "nearai".to_string(), tenant_rates: std::sync::Arc::new(ironclaw::tenant::TenantRateRegistry::new(4, 3)), + channel_routing: std::sync::Arc::new(tokio::sync::RwLock::new(None)), }, channels, None, diff --git a/tests/support/test_rig.rs b/tests/support/test_rig.rs index bf9887c8366..d7a54bbf473 100644 --- a/tests/support/test_rig.rs +++ b/tests/support/test_rig.rs @@ -766,6 +766,7 @@ pub struct TestRigBuilder { wasm_tools: Vec, keep_bootstrap: bool, engine_v2: bool, + local_jobs_only: bool, channel_name_override: Option, seeded_secrets: Option, /// Pre-seed the SecretsStore with `(name, value)` pairs before the @@ -795,6 +796,7 @@ impl TestRigBuilder { wasm_tools: Vec::new(), keep_bootstrap: false, engine_v2: false, + local_jobs_only: false, channel_name_override: None, seeded_secrets: None, pre_seed_secrets: Vec::new(), @@ -979,6 +981,16 @@ impl TestRigBuilder { self } + /// Force built-in job tools to use the local ContextManager-only path. + /// + /// This avoids spawning background worker loops that would otherwise share + /// the same replay LLM and consume trace steps intended for the foreground + /// chat turn. + pub fn with_local_jobs_only(mut self) -> Self { + self.local_jobs_only = true; + self + } + /// Add pre-recorded HTTP exchanges for the `ReplayingHttpInterceptor`. /// /// When set, all `http` tool calls will return these responses in order @@ -1028,6 +1040,7 @@ impl TestRigBuilder { wasm_tools, keep_bootstrap, engine_v2, + local_jobs_only, channel_name_override, seeded_secrets, pre_seed_secrets, @@ -1235,13 +1248,18 @@ impl TestRigBuilder { components.tools.register_job_tools( Arc::clone(&components.context_manager), - Some(scheduler_slot.clone()), + if local_jobs_only { + None + } else { + Some(scheduler_slot.clone()) + }, None, components.db.clone(), None, None, None, None, + None, ); // Routine tools: create a RoutineEngine with the LLM and workspace. @@ -1443,6 +1461,7 @@ impl TestRigBuilder { builder: None, llm_backend: "nearai".to_string(), tenant_rates: std::sync::Arc::new(ironclaw::tenant::TenantRateRegistry::new(4, 3)), + channel_routing: std::sync::Arc::new(tokio::sync::RwLock::new(None)), }; // 7. Create TestChannel and ChannelManager.