diff --git a/Cargo.lock b/Cargo.lock index c6b3e6f1..cf0c7092 100644 --- a/Cargo.lock +++ b/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", diff --git a/src/agent/agent_loop.rs b/src/agent/agent_loop.rs index 8fda4143..e54b66ca 100644 --- a/src/agent/agent_loop.rs +++ b/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 diff --git a/src/agent/dispatcher.rs b/src/agent/dispatcher.rs index d0ae98ad..85a281e0 100644 --- a/src/agent/dispatcher.rs +++ b/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). diff --git a/src/agent/job_monitor.rs b/src/agent/job_monitor.rs index b2db8852..181bc853 100644 --- a/src/agent/job_monitor.rs +++ b/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, + 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, + 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::(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::(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::(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 diff --git a/src/tools/builtin/job.rs b/src/tools/builtin/job.rs index 880f8622..c4c3fb6b 100644 --- a/src/tools/builtin/job.rs +++ b/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 { + 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 {