diff --git a/src/agent/agent_loop.rs b/src/agent/agent_loop.rs index 5ca094e4..4b7ed538 100644 --- a/src/agent/agent_loop.rs +++ b/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 diff --git a/src/agent/dispatcher.rs b/src/agent/dispatcher.rs index a91f59a6..9e6747f2 100644 --- a/src/agent/dispatcher.rs +++ b/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). diff --git a/src/agent/job_monitor.rs b/src/agent/job_monitor.rs index b2db8852..714caeac 100644 --- a/src/agent/job_monitor.rs +++ b/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, +} + /// 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, + 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::(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::(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::(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); + } } diff --git a/src/channels/channel.rs b/src/channels/channel.rs index 1fc76fd7..ed8c28ff 100644 --- a/src/channels/channel.rs +++ b/src/channels/channel.rs @@ -83,6 +83,11 @@ pub struct IncomingMessage { pub timezone: Option, /// File or media attachments on this message. pub attachments: Vec, + /// 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. diff --git a/src/tools/builtin/job.rs b/src/tools/builtin/job.rs index 8744f75b..9346d14a 100644 --- a/src/tools/builtin/job.rs +++ b/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 { + // 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 { diff --git a/tests/e2e_routine_heartbeat.rs b/tests/e2e_routine_heartbeat.rs index f5a28c25..6d6deb8b 100644 --- a/tests/e2e_routine_heartbeat.rs +++ b/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");