fix: prefer execution-local message routing metadata (#1449)

* fix: prefer execution-local message routing metadata

* test: cover message routing fallback metadata

* refactor: simplify message target resolution

* fix: ignore stale channel defaults for notify user metadata

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 {

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).

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

src/tools/builtin/message.rs

@@ -67,6 +67,95 @@ impl MessageTool {
     }
 }
 
+fn metadata_string(metadata: &serde_json::Value, key: &str) -> Option<String> {
+    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<String> {
+    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<ExtensionManager>>,
+    channel: Option<&str>,
+    ctx_user_id: &str,
+) -> Option<String> {
+    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<ExtensionManager>>,
+    explicit_target: Option<String>,
+    metadata_target: Option<String>,
+    default_target: Option<String>,
+    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<String> {
+    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<String> = 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<String> = 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<Mutex<Vec<(String, OutgoingResponse)>>>;
+
+    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<MessageStream, ChannelError> {
+            let (_tx, rx) = mpsc::channel::<IncomingMessage>(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"));
     }
 }