Fix subagent monitor events being treated as user input (#1173)

* Fix subagent monitor routing to avoid LLM re-entry

* Update yanked uds_windows dependency in lockfile

Cargo.lock

@@ -7134,9 +7134,9 @@ checksum = "2896d95c02a80c6d6a5d6e953d479f5ddf2dfdb6a244441010e373ac0fb88971"
 
 [[package]]
 name = "uds_windows"
-version = "1.2.0"
+version = "1.2.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "51b70b87d15e91f553711b40df3048faf27a7a04e01e0ddc0cf9309f0af7c2ca"
+checksum = "f2f6fb2847f6742cd76af783a2a2c49e9375d0a111c7bef6f71cd9e738c72d6e"
 dependencies = [
  "memoffset",
  "tempfile",

src/agent/agent_loop.rs

@@ -750,6 +750,23 @@ impl Agent {
             "Message details"
         );
 
+        // Internal job-monitor notifications are already rendered text and
+        // should be forwarded directly to the user without entering the
+        // normal user-input pipeline (which would run the LLM/tool loop).
+        if message
+            .metadata
+            .get("__internal_job_monitor")
+            .and_then(|v| v.as_bool())
+            == Some(true)
+        {
+            tracing::debug!(
+                message_id = %message.id,
+                channel = %message.channel,
+                "Forwarding internal job monitor notification"
+            );
+            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,12 @@ 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,
+            "notify_metadata": message.metadata,
+        });
 
         // 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,33 @@ 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>,
+    pub metadata: serde_json::Value,
+}
+
+fn build_internal_metadata(route: &JobMonitorRoute, job_id: Uuid) -> serde_json::Value {
+    let mut metadata = route.metadata.clone();
+    if !metadata.is_object() {
+        metadata = serde_json::json!({});
+    }
+    if let Some(obj) = metadata.as_object_mut() {
+        obj.insert(
+            "__internal_job_monitor".to_string(),
+            serde_json::Value::Bool(true),
+        );
+        obj.insert(
+            "__job_monitor_job_id".to_string(),
+            serde_json::Value::String(job_id.to_string()),
+        );
+    }
+    metadata
+}
+
 /// Spawn a background task that watches for events from a specific job and
 /// injects assistant messages into the agent loop.
 ///
@@ -35,6 +62,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 +78,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),
-                            );
+                            )
+                            .with_metadata(build_internal_metadata(&route, job_id));
+                            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 +96,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
                                 ),
-                            );
+                            )
+                            .with_metadata(build_internal_metadata(&route, job_id));
+                            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 +144,24 @@ 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()),
+            metadata: serde_json::json!({
+                "source": "test",
+            }),
+        }
+    }
+
     #[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 +180,16 @@ 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_eq!(
+            msg.metadata
+                .get("__internal_job_monitor")
+                .and_then(|v| v.as_bool()),
+            Some(true)
+        );
     }
 
     #[tokio::test]
@@ -145,7 +199,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 +228,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 +262,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

src/tools/builtin/job.rs

@@ -411,7 +411,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!({
@@ -676,6 +688,37 @@ fn resolve_project_dir(
     Ok((canonical_dir, browse_id))
 }
 
+fn monitor_route_from_ctx(ctx: &JobContext) -> Option<crate::agent::job_monitor::JobMonitorRoute> {
+    let channel = ctx
+        .metadata
+        .get("notify_channel")
+        .and_then(|v| v.as_str())?
+        .to_string();
+    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());
+    let metadata = ctx
+        .metadata
+        .get("notify_metadata")
+        .cloned()
+        .unwrap_or_else(|| serde_json::json!({}));
+
+    Some(crate::agent::job_monitor::JobMonitorRoute {
+        channel,
+        user_id,
+        thread_id,
+        metadata,
+    })
+}
+
 #[async_trait]
 impl Tool for CreateJobTool {
     fn name(&self) -> &str {