diff --git a/src/agent/agent_loop.rs b/src/agent/agent_loop.rs index c31145d5..dbc9d38b 100644 --- a/src/agent/agent_loop.rs +++ b/src/agent/agent_loop.rs @@ -120,6 +120,17 @@ async fn resolve_routine_notification_target( .await } +pub(crate) fn chat_tool_execution_metadata(message: &IncomingMessage) -> serde_json::Value { + serde_json::json!({ + "notify_channel": message.channel, + "notify_user": message + .routing_target() + .unwrap_or_else(|| message.user_id.clone()), + "notify_thread_id": message.thread_id, + "notify_metadata": message.metadata, + }) +} + fn should_fallback_routine_notification(error: &ChannelError) -> bool { !matches!(error, ChannelError::MissingRoutingTarget { .. }) } @@ -1177,9 +1188,10 @@ impl Agent { #[cfg(test)] mod tests { use super::{ - resolve_routine_notification_user, should_fallback_routine_notification, - truncate_for_preview, + chat_tool_execution_metadata, resolve_routine_notification_user, + should_fallback_routine_notification, truncate_for_preview, }; + use crate::channels::IncomingMessage; use crate::error::ChannelError; #[test] @@ -1275,6 +1287,50 @@ mod tests { assert_eq!(resolve_routine_notification_user(&metadata), None); // safety: test-only assertion } + #[test] + fn chat_tool_execution_metadata_prefers_message_routing_target() { + let message = IncomingMessage::new("telegram", "owner-scope", "hello") + .with_sender_id("telegram-user") + .with_thread("thread-7") + .with_metadata(serde_json::json!({ + "chat_id": 424242, + "chat_type": "private", + })); + + let metadata = chat_tool_execution_metadata(&message); + assert_eq!( + metadata.get("notify_channel").and_then(|v| v.as_str()), + Some("telegram") + ); // safety: test-only assertion + assert_eq!( + metadata.get("notify_user").and_then(|v| v.as_str()), + Some("424242") + ); // safety: test-only assertion + assert_eq!( + metadata.get("notify_thread_id").and_then(|v| v.as_str()), + Some("thread-7") + ); // safety: test-only assertion + } + + #[test] + fn chat_tool_execution_metadata_falls_back_to_user_scope_without_route() { + let message = IncomingMessage::new("gateway", "owner-scope", "hello").with_sender_id(""); + + let metadata = chat_tool_execution_metadata(&message); + assert_eq!( + metadata.get("notify_channel").and_then(|v| v.as_str()), + Some("gateway") + ); // safety: test-only assertion + assert_eq!( + metadata.get("notify_user").and_then(|v| v.as_str()), + Some("owner-scope") + ); // safety: test-only assertion + assert_eq!( + metadata.get("notify_thread_id"), + Some(&serde_json::Value::Null) + ); // safety: test-only assertion + } + #[test] fn targeted_routine_notifications_do_not_fallback_without_owner_route() { let error = ChannelError::MissingRoutingTarget { diff --git a/src/agent/dispatcher.rs b/src/agent/dispatcher.rs index 0b47c928..fc3da61b 100644 --- a/src/agent/dispatcher.rs +++ b/src/agent/dispatcher.rs @@ -144,12 +144,7 @@ impl Agent { .with_requester_id(&message.sender_id); job_ctx.http_interceptor = self.deps.http_interceptor.clone(); job_ctx.user_timezone = user_tz.name().to_string(); - job_ctx.metadata = serde_json::json!({ - "notify_channel": message.channel, - "notify_user": message.user_id, - "notify_thread_id": message.thread_id, - "notify_metadata": message.metadata, - }); + job_ctx.metadata = crate::agent::agent_loop::chat_tool_execution_metadata(message); // Build system prompts once for this turn. Two variants: with tools // (normal iterations) and without (force_text final iteration). diff --git a/src/agent/thread_ops.rs b/src/agent/thread_ops.rs index e8b8d09a..0fb968f1 100644 --- a/src/agent/thread_ops.rs +++ b/src/agent/thread_ops.rs @@ -939,6 +939,7 @@ impl Agent { JobContext::with_user(&message.user_id, "chat", "Interactive chat session") .with_requester_id(&message.sender_id); job_ctx.http_interceptor = self.deps.http_interceptor.clone(); + job_ctx.metadata = crate::agent::agent_loop::chat_tool_execution_metadata(message); // Prefer a valid timezone from the approval message, fall back to the // resolved timezone stored when the approval was originally requested. let tz_candidate = message diff --git a/src/tools/builtin/message.rs b/src/tools/builtin/message.rs index 1d2ed059..83041b80 100644 --- a/src/tools/builtin/message.rs +++ b/src/tools/builtin/message.rs @@ -67,6 +67,95 @@ impl MessageTool { } } +fn metadata_string(metadata: &serde_json::Value, key: &str) -> Option { + metadata + .get(key) + .and_then(|value| value.as_str()) + .map(str::trim) + .filter(|value| !value.is_empty()) + .map(ToOwned::to_owned) +} + +fn metadata_notify_user(metadata: &serde_json::Value) -> Option { + metadata_string(metadata, "notify_user").filter(|value| value != "default") +} + +fn channel_matches_source(resolved_channel: Option<&str>, source_channel: Option<&str>) -> bool { + match (resolved_channel, source_channel) { + (None, _) => true, + (Some(resolved), Some(source)) if resolved == source => true, + _ => false, + } +} + +async fn resolve_channel_fallback_target( + extension_manager: Option<&Arc>, + channel: Option<&str>, + ctx_user_id: &str, +) -> Option { + let channel_name = channel?; + + if let Some(extension_manager) = extension_manager + && let Some(target) = extension_manager + .notification_target_for_channel(channel_name) + .await + { + return Some(target); + } + + Some(ctx_user_id.to_string()) +} + +struct MessageTargetResolution<'a> { + extension_manager: Option<&'a Arc>, + explicit_target: Option, + metadata_target: Option, + default_target: Option, + channel: Option<&'a str>, + metadata_channel: Option<&'a str>, + default_channel: Option<&'a str>, + has_execution_routing_metadata: bool, + ctx_user_id: &'a str, +} + +async fn resolve_message_target(inputs: MessageTargetResolution<'_>) -> Option { + if let Some(target) = inputs.explicit_target { + return Some(target); + } + + if inputs.has_execution_routing_metadata { + if channel_matches_source(inputs.channel, inputs.metadata_channel) + && let Some(target) = inputs.metadata_target + { + return Some(target); + } + + return resolve_channel_fallback_target( + inputs.extension_manager, + inputs.channel, + inputs.ctx_user_id, + ) + .await; + } + + if channel_matches_source(inputs.channel, inputs.default_channel) + && let Some(target) = inputs.default_target + { + return Some(target); + } + + if inputs.channel.is_some() { + return resolve_channel_fallback_target( + inputs.extension_manager, + inputs.channel, + inputs.ctx_user_id, + ) + .await; + } + + None +} + #[async_trait] impl Tool for MessageTool { fn name(&self) -> &str { @@ -123,68 +212,52 @@ impl Tool for MessageTool { .get("channel") .and_then(|v| v.as_str()) .map(|value| value.to_string()); + let metadata_channel = metadata_string(&ctx.metadata, "notify_channel"); let default_channel = self .default_channel .read() .unwrap_or_else(|e| e.into_inner()) .clone(); - let metadata_channel = ctx - .metadata - .get("notify_channel") + let default_target = self + .default_target + .read() + .unwrap_or_else(|e| e.into_inner()) + .clone(); + let metadata_target = metadata_notify_user(&ctx.metadata); + let has_execution_routing_metadata = + metadata_channel.is_some() || metadata_target.is_some(); + + // Job metadata is authoritative for autonomous executions. The shared + // conversation defaults are only a legacy fallback when no execution-local + // routing metadata is available. + let channel: Option = explicit_channel + .clone() + .or_else(|| metadata_channel.clone()) + .or_else(|| { + (!has_execution_routing_metadata) + .then(|| default_channel.clone()) + .flatten() + }); + + let explicit_target = params + .get("target") .and_then(|v| v.as_str()) .map(|value| value.to_string()); - // Get channel: use param → conversation default → job metadata → None (broadcast all) - let channel: Option = explicit_channel - .clone() - .or_else(|| default_channel.clone()) - .or_else(|| metadata_channel.clone()); - - let can_use_default_target = match (explicit_channel.as_deref(), default_channel.as_deref()) - { - (None, _) => true, - (Some(explicit), Some(current)) if explicit == current => true, - _ => false, - }; - let can_use_metadata_target = match (channel.as_deref(), metadata_channel.as_deref()) { - (None, _) => true, - (Some(resolved), Some(current)) if resolved == current => true, - _ => false, - }; - - // Get target: use param → conversation default → job metadata → owner scope - // fallback when a specific channel is known. - let target = if let Some(t) = params.get("target").and_then(|v| v.as_str()) { - Some(t.to_string()) - } else if can_use_default_target - && let Some(t) = self - .default_target - .read() - .unwrap_or_else(|e| e.into_inner()) - .clone() - { - Some(t) - } else if can_use_metadata_target - && let Some(t) = ctx.metadata.get("notify_user").and_then(|v| v.as_str()) - { - Some(t.to_string()) - } else if channel.is_some() { - if let Some(channel_name) = channel.as_deref() { - if let Some(extension_manager) = self.extension_manager.as_ref() - && let Some(target) = extension_manager - .notification_target_for_channel(channel_name) - .await - { - Some(target) - } else { - Some(ctx.user_id.clone()) - } - } else { - Some(ctx.user_id.clone()) - } - } else { - None - }; + // Prefer explicit params, then execution-local routing metadata. Shared + // conversation defaults are only consulted when no job metadata exists. + let target = resolve_message_target(MessageTargetResolution { + extension_manager: self.extension_manager.as_ref(), + explicit_target, + metadata_target, + default_target, + channel: channel.as_deref(), + metadata_channel: metadata_channel.as_deref(), + default_channel: default_channel.as_deref(), + has_execution_routing_metadata, + ctx_user_id: &ctx.user_id, + }) + .await; let Some(target) = target else { return Err(ToolError::ExecutionFailed( @@ -230,6 +303,12 @@ impl Tool for MessageTool { if !attachments.is_empty() { response = response.with_attachments(attachments); } + if channel.as_deref() == Some("gateway") + && response.thread_id.is_none() + && let Some(thread_id) = metadata_string(&ctx.metadata, "notify_thread_id") + { + response = response.in_thread(thread_id); + } if let Some(ref channel) = channel { // Send to a specific channel @@ -326,6 +405,92 @@ impl Tool for MessageTool { #[cfg(test)] mod tests { use super::*; + use async_trait::async_trait; + use tokio::sync::{Mutex, mpsc}; + + use crate::channels::{ + Channel, IncomingMessage, MessageStream, OutgoingResponse, StatusUpdate, + }; + use crate::error::ChannelError; + + type BroadcastCapture = Arc>>; + + struct RecordingChannel { + name: &'static str, + captures: BroadcastCapture, + } + + impl RecordingChannel { + fn new(name: &'static str) -> (Self, BroadcastCapture) { + let captures = Arc::new(Mutex::new(Vec::new())); + ( + Self { + name, + captures: Arc::clone(&captures), + }, + captures, + ) + } + } + + #[async_trait] + impl Channel for RecordingChannel { + fn name(&self) -> &str { + self.name + } + + async fn start(&self) -> Result { + let (_tx, rx) = mpsc::channel::(1); + Ok(Box::pin(tokio_stream::wrappers::ReceiverStream::new(rx))) + } + + async fn respond( + &self, + _msg: &IncomingMessage, + _response: OutgoingResponse, + ) -> Result<(), ChannelError> { + Ok(()) + } + + async fn send_status( + &self, + _status: StatusUpdate, + _metadata: &serde_json::Value, + ) -> Result<(), ChannelError> { + Ok(()) + } + + async fn broadcast( + &self, + user_id: &str, + response: OutgoingResponse, + ) -> Result<(), ChannelError> { + self.captures + .lock() + .await + .push((user_id.to_string(), response)); + Ok(()) + } + + async fn health_check(&self) -> Result<(), ChannelError> { + Ok(()) + } + } + + async fn message_tool_with_recording_channels() + -> (MessageTool, BroadcastCapture, BroadcastCapture) { + let channel_manager = ChannelManager::new(); + let (gateway, gateway_captures) = RecordingChannel::new("gateway"); + let (telegram, telegram_captures) = RecordingChannel::new("telegram"); + channel_manager.add(Box::new(gateway)).await; + channel_manager.add(Box::new(telegram)).await; + + ( + MessageTool::new(Arc::new(channel_manager)), + gateway_captures, + telegram_captures, + ) + } #[test] fn message_tool_name() { @@ -782,31 +947,94 @@ mod tests { } #[tokio::test] - async fn message_tool_does_not_apply_metadata_target_to_different_default_channel() { - let tool = MessageTool::new(Arc::new(ChannelManager::new())); - tool.set_context(Some("telegram".to_string()), None).await; + async fn message_tool_prefers_metadata_over_stale_default_context() { + let (tool, gateway_captures, telegram_captures) = + message_tool_with_recording_channels().await; + tool.set_context( + Some("gateway".to_string()), + Some("stale-gateway-target".to_string()), + ) + .await; let mut ctx = crate::context::JobContext::with_user("owner-scope", "test", "test"); ctx.metadata = serde_json::json!({ - "notify_channel": "signal", - "notify_user": "metadata-user", + "notify_channel": "telegram", + "notify_user": "424242", }); let result = tool .execute(serde_json::json!({"content": "hello"}), &ctx) - .await; + .await + .expect("message tool should use telegram metadata routing"); + assert_eq!( + result.result.as_str(), + Some("Sent message to telegram:424242") + ); - assert!(result.is_err()); - let err = result.unwrap_err().to_string(); + assert!(gateway_captures.lock().await.is_empty()); + let telegram = telegram_captures.lock().await.clone(); + assert_eq!(telegram.len(), 1); + assert_eq!(telegram[0].0, "424242"); + assert_eq!(telegram[0].1.content, "hello"); + } + + #[tokio::test] + async fn message_tool_notify_user_only_metadata_does_not_reuse_stale_default_channel() { + let (tool, gateway_captures, telegram_captures) = + message_tool_with_recording_channels().await; + tool.set_context( + Some("gateway".to_string()), + Some("stale-gateway-target".to_string()), + ) + .await; + + let mut ctx = crate::context::JobContext::with_user("owner-scope", "test", "test"); + ctx.metadata = serde_json::json!({ + "notify_user": "424242", + }); + + let result = tool + .execute(serde_json::json!({"content": "hello"}), &ctx) + .await + .expect("message tool should broadcast when only notify_user is provided"); assert!( - !err.contains("metadata-user"), - "metadata target should not be applied to a different default channel: {}", - err - ); - assert!( - err.contains("owner-scope"), - "expected owner-scope fallback target when metadata channel differs: {}", - err + result + .result + .as_str() + .is_some_and(|message| message.contains("Broadcast message to")) ); + + let gateway = gateway_captures.lock().await.clone(); + assert_eq!(gateway.len(), 1); + assert_eq!(gateway[0].0, "424242"); + assert_eq!(gateway[0].1.content, "hello"); + + let telegram = telegram_captures.lock().await.clone(); + assert_eq!(telegram.len(), 1); + assert_eq!(telegram[0].0, "424242"); + assert_eq!(telegram[0].1.content, "hello"); + } + + #[tokio::test] + async fn message_tool_applies_notify_thread_id_for_gateway_delivery() { + let (tool, gateway_captures, telegram_captures) = + message_tool_with_recording_channels().await; + + let mut ctx = crate::context::JobContext::with_user("owner-scope", "test", "test"); + ctx.metadata = serde_json::json!({ + "notify_channel": "gateway", + "notify_user": "owner-scope", + "notify_thread_id": "thread-123", + }); + + tool.execute(serde_json::json!({"content": "hello"}), &ctx) + .await + .expect("gateway routing with thread id should succeed"); + + assert!(telegram_captures.lock().await.is_empty()); + let gateway = gateway_captures.lock().await.clone(); + assert_eq!(gateway.len(), 1); + assert_eq!(gateway[0].0, "owner-scope"); + assert_eq!(gateway[0].1.thread_id.as_deref(), Some("thread-123")); } }