fix(security): prevent metadata spoofing of internal job monitor flag (#1195)

The `__internal_job_monitor` metadata key that bypassed the entire
agent pipeline (hooks, safety checks, LLM processing) was spoofable
by external channels — WASM channel plugins could inject arbitrary
metadata including this key, causing attacker-controlled content to be
forwarded directly as assistant responses.

Replace the metadata-based check with a dedicated `is_internal` field
on `IncomingMessage` that can only be set via `into_internal()` by
trusted in-process code. Both the field and setter are `pub(crate)` to
prevent external crates from spoofing the flag. Also remove
`notify_metadata` forwarding (the monitor only needs channel/user/thread
routing) and the unused `__job_monitor_job_id` metadata key.

Co-authored-by: Claude Opus 4.6 (1M context) <[email protected]>

src/agent/agent_loop.rs

@@ -750,6 +750,20 @@ impl Agent {
             "Message details"
         );
 
+        // Internal messages (e.g. job-monitor notifications) are already
+        // rendered text and should be forwarded directly to the user without
+        // entering the normal user-input pipeline (LLM/tool loop).
+        // The `is_internal` field and `into_internal()` setter are pub(crate),
+        // so external channels cannot spoof this flag.
+        if message.is_internal {
+            tracing::debug!(
+                message_id = %message.id,
+                channel = %message.channel,
+                "Forwarding internal message"
+            );
+            return Ok(Some(message.content.clone()));
+        }
+
         // Set message tool context for this turn (current channel and target)
         // For Signal, use signal_target from metadata (group:ID or phone number),
         // otherwise fall back to user_id

src/agent/dispatcher.rs

@@ -143,6 +143,11 @@ impl Agent {
             JobContext::with_user(&message.user_id, "chat", "Interactive chat session");
         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,
+        });
 
         // Build system prompts once for this turn. Two variants: with tools
         // (normal iterations) and without (force_text final iteration).

src/agent/job_monitor.rs

@@ -21,6 +21,14 @@ use uuid::Uuid;
 use crate::channels::IncomingMessage;
 use crate::channels::web::types::SseEvent;
 
+/// Route context for forwarding job monitor events back to the user's channel.
+#[derive(Debug, Clone)]
+pub struct JobMonitorRoute {
+    pub channel: String,
+    pub user_id: String,
+    pub thread_id: Option<String>,
+}
+
 /// Spawn a background task that watches for events from a specific job and
 /// injects assistant messages into the agent loop.
 ///
@@ -35,6 +43,7 @@ pub fn spawn_job_monitor(
     job_id: Uuid,
     mut event_rx: broadcast::Receiver<(Uuid, SseEvent)>,
     inject_tx: mpsc::Sender<IncomingMessage>,
+    route: JobMonitorRoute,
 ) -> JoinHandle<()> {
     let short_id = job_id.to_string()[..8].to_string();
 
@@ -50,11 +59,15 @@ pub fn spawn_job_monitor(
 
                     match event {
                         SseEvent::JobMessage { role, content, .. } if role == "assistant" => {
-                            let msg = IncomingMessage::new(
-                                "job_monitor",
-                                "system",
+                            let mut msg = IncomingMessage::new(
+                                route.channel.clone(),
+                                route.user_id.clone(),
                                 format!("[Job {}] Claude Code: {}", short_id, content),
-                            );
+                            )
+                            .into_internal();
+                            if let Some(ref thread_id) = route.thread_id {
+                                msg = msg.with_thread(thread_id.clone());
+                            }
                             if inject_tx.send(msg).await.is_err() {
                                 tracing::debug!(
                                     job_id = %short_id,
@@ -64,14 +77,18 @@ pub fn spawn_job_monitor(
                             }
                         }
                         SseEvent::JobResult { status, .. } => {
-                            let msg = IncomingMessage::new(
-                                "job_monitor",
-                                "system",
+                            let mut msg = IncomingMessage::new(
+                                route.channel.clone(),
+                                route.user_id.clone(),
                                 format!(
                                     "[Job {}] Container finished (status: {})",
                                     short_id, status
                                 ),
-                            );
+                            )
+                            .into_internal();
+                            if let Some(ref thread_id) = route.thread_id {
+                                msg = msg.with_thread(thread_id.clone());
+                            }
                             let _ = inject_tx.send(msg).await;
                             tracing::debug!(
                                 job_id = %short_id,
@@ -108,13 +125,21 @@ pub fn spawn_job_monitor(
 mod tests {
     use super::*;
 
+    fn test_route() -> JobMonitorRoute {
+        JobMonitorRoute {
+            channel: "cli".to_string(),
+            user_id: "user-1".to_string(),
+            thread_id: Some("thread-1".to_string()),
+        }
+    }
+
     #[tokio::test]
     async fn test_monitor_forwards_assistant_messages() {
         let (event_tx, _) = broadcast::channel::<(Uuid, SseEvent)>(16);
         let (inject_tx, mut inject_rx) = mpsc::channel::<IncomingMessage>(16);
 
         let job_id = Uuid::new_v4();
-        let _handle = spawn_job_monitor(job_id, event_tx.subscribe(), inject_tx);
+        let _handle = spawn_job_monitor(job_id, event_tx.subscribe(), inject_tx, test_route());
 
         // Send an assistant message
         event_tx
@@ -133,9 +158,11 @@ mod tests {
             .unwrap()
             .unwrap();
 
-        assert_eq!(msg.channel, "job_monitor");
-        assert_eq!(msg.user_id, "system");
+        assert_eq!(msg.channel, "cli");
+        assert_eq!(msg.user_id, "user-1");
+        assert_eq!(msg.thread_id, Some("thread-1".to_string()));
         assert!(msg.content.contains("I found a bug"));
+        assert!(msg.is_internal, "monitor messages must be marked internal");
     }
 
     #[tokio::test]
@@ -145,7 +172,7 @@ mod tests {
 
         let job_id = Uuid::new_v4();
         let other_job_id = Uuid::new_v4();
-        let _handle = spawn_job_monitor(job_id, event_tx.subscribe(), inject_tx);
+        let _handle = spawn_job_monitor(job_id, event_tx.subscribe(), inject_tx, test_route());
 
         // Send a message for a different job
         event_tx
@@ -174,7 +201,7 @@ mod tests {
         let (inject_tx, mut inject_rx) = mpsc::channel::<IncomingMessage>(16);
 
         let job_id = Uuid::new_v4();
-        let handle = spawn_job_monitor(job_id, event_tx.subscribe(), inject_tx);
+        let handle = spawn_job_monitor(job_id, event_tx.subscribe(), inject_tx, test_route());
 
         // Send a completion event
         event_tx
@@ -208,7 +235,7 @@ mod tests {
         let (inject_tx, mut inject_rx) = mpsc::channel::<IncomingMessage>(16);
 
         let job_id = Uuid::new_v4();
-        let _handle = spawn_job_monitor(job_id, event_tx.subscribe(), inject_tx);
+        let _handle = spawn_job_monitor(job_id, event_tx.subscribe(), inject_tx, test_route());
 
         // Send tool use event (should be skipped)
         event_tx
@@ -242,4 +269,28 @@ mod tests {
             "should have timed out, no message expected"
         );
     }
+
+    /// Regression test: external channels must not be able to spoof the
+    /// `is_internal` flag via metadata keys. A message created through
+    /// the normal `IncomingMessage::new` + `with_metadata` path must
+    /// always have `is_internal == false`, regardless of metadata content.
+    #[test]
+    fn test_external_metadata_cannot_spoof_internal_flag() {
+        let msg = IncomingMessage::new("wasm_channel", "attacker", "pwned").with_metadata(
+            serde_json::json!({
+                "__internal_job_monitor": true,
+                "is_internal": true,
+            }),
+        );
+        assert!(
+            !msg.is_internal,
+            "with_metadata must not set is_internal — only into_internal() can"
+        );
+    }
+
+    #[test]
+    fn test_into_internal_sets_flag() {
+        let msg = IncomingMessage::new("monitor", "system", "test").into_internal();
+        assert!(msg.is_internal);
+    }
 }

src/channels/channel.rs

@@ -83,6 +83,11 @@ pub struct IncomingMessage {
     pub timezone: Option<String>,
     /// File or media attachments on this message.
     pub attachments: Vec<IncomingAttachment>,
+    /// Internal-only flag: message was generated inside the process (e.g. job
+    /// monitor) and must bypass the normal user-input pipeline.  This field is
+    /// **not** settable via `with_metadata()` — only trusted code paths inside
+    /// the binary can set it, preventing external channels from spoofing it.
+    pub(crate) is_internal: bool,
 }
 
 impl IncomingMessage {
@@ -103,6 +108,7 @@ impl IncomingMessage {
             metadata: serde_json::Value::Null,
             timezone: None,
             attachments: Vec::new(),
+            is_internal: false,
         }
     }
 
@@ -135,6 +141,12 @@ impl IncomingMessage {
         self.attachments = attachments;
         self
     }
+
+    /// Mark this message as internal (bypasses user-input pipeline).
+    pub(crate) fn into_internal(mut self) -> Self {
+        self.is_internal = true;
+        self
+    }
 }
 
 /// Stream of incoming messages.

src/tools/builtin/job.rs

@@ -415,7 +415,19 @@ impl CreateJobTool {
             // loop stops consuming from inject_tx the send will fail and the
             // monitor terminates. No JoinHandle is retained.
             if let (Some(etx), Some(itx)) = (&self.event_tx, &self.inject_tx) {
-                crate::agent::job_monitor::spawn_job_monitor(job_id, etx.subscribe(), itx.clone());
+                if let Some(route) = monitor_route_from_ctx(ctx) {
+                    crate::agent::job_monitor::spawn_job_monitor(
+                        job_id,
+                        etx.subscribe(),
+                        itx.clone(),
+                        route,
+                    );
+                } else {
+                    tracing::debug!(
+                        job_id = %job_id,
+                        "Skipping job monitor injection due to missing route metadata"
+                    );
+                }
             }
 
             let result = serde_json::json!({
@@ -680,6 +692,36 @@ fn resolve_project_dir(
     Ok((canonical_dir, browse_id))
 }
 
+fn monitor_route_from_ctx(ctx: &JobContext) -> Option<crate::agent::job_monitor::JobMonitorRoute> {
+    // 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.
+    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.
+    let user_id = ctx
+        .metadata
+        .get("notify_user")
+        .and_then(|v| v.as_str())
+        .unwrap_or(&ctx.user_id)
+        .to_string();
+    let thread_id = ctx
+        .metadata
+        .get("notify_thread_id")
+        .and_then(|v| v.as_str())
+        .map(|s| s.to_string());
+
+    Some(crate::agent::job_monitor::JobMonitorRoute {
+        channel,
+        user_id,
+        thread_id,
+    })
+}
+
 #[async_trait]
 impl Tool for CreateJobTool {
     fn name(&self) -> &str {

tests/e2e_routine_heartbeat.rs

@@ -218,18 +218,7 @@ mod tests {
         engine.refresh_event_cache().await;
 
         // Positive match: message containing "deploy to production".
-        let matching_msg = IncomingMessage {
-            id: Uuid::new_v4(),
-            channel: "test".to_string(),
-            user_id: "default".to_string(),
-            user_name: None,
-            content: "deploy to production now".to_string(),
-            thread_id: None,
-            received_at: Utc::now(),
-            metadata: serde_json::json!({}),
-            timezone: None,
-            attachments: Vec::new(),
-        };
+        let matching_msg = IncomingMessage::new("test", "default", "deploy to production now");
         let fired = engine.check_event_triggers(&matching_msg).await;
         assert!(
             fired >= 1,
@@ -240,18 +229,8 @@ mod tests {
         tokio::time::sleep(Duration::from_millis(500)).await;
 
         // Negative match: message that doesn't match.
-        let non_matching_msg = IncomingMessage {
-            id: Uuid::new_v4(),
-            channel: "test".to_string(),
-            user_id: "default".to_string(),
-            user_name: None,
-            content: "check the staging environment".to_string(),
-            thread_id: None,
-            received_at: Utc::now(),
-            metadata: serde_json::json!({}),
-            timezone: None,
-            attachments: Vec::new(),
-        };
+        let non_matching_msg =
+            IncomingMessage::new("test", "default", "check the staging environment");
         let fired_neg = engine.check_event_triggers(&non_matching_msg).await;
         assert_eq!(fired_neg, 0, "Expected 0 routines fired on non-match");
     }
@@ -455,18 +434,7 @@ mod tests {
         engine.refresh_event_cache().await;
 
         // First fire should work.
-        let msg = IncomingMessage {
-            id: Uuid::new_v4(),
-            channel: "test".to_string(),
-            user_id: "default".to_string(),
-            user_name: None,
-            content: "test-cooldown trigger".to_string(),
-            thread_id: None,
-            received_at: Utc::now(),
-            metadata: serde_json::json!({}),
-            timezone: None,
-            attachments: Vec::new(),
-        };
+        let msg = IncomingMessage::new("test", "default", "test-cooldown trigger");
         let fired1 = engine.check_event_triggers(&msg).await;
         assert!(fired1 >= 1, "First fire should work");