From a7f17c3981cc7bb1b8d1e4ddc57fb0fd3dfe61ef Mon Sep 17 00:00:00 2001 From: Wylabb <77673282+Wylabb@users.noreply.github.com> Date: Sun, 5 Apr 2026 09:59:29 +0200 Subject: [PATCH] Notify Telegram about background agent completion --- rust/crates/claw-profile-worker/src/server.rs | 15 ++ rust/crates/claw-telegram/src/gateway.rs | 133 ++++++++++++++++++ .../crates/claw-telegram/src/worker_client.rs | 8 ++ rust/crates/runtime/src/runtime_task_store.rs | 53 +++++++ 4 files changed, 209 insertions(+) diff --git a/rust/crates/claw-profile-worker/src/server.rs b/rust/crates/claw-profile-worker/src/server.rs index cd7a770..243495b 100644 --- a/rust/crates/claw-profile-worker/src/server.rs +++ b/rust/crates/claw-profile-worker/src/server.rs @@ -55,6 +55,7 @@ fn app_router(config: WorkerConfig, runtime: Arc) -> Router { .route("/v1/team", get(get_team)) .route("/v1/agents", get(list_agents)) .route("/v1/agents/:agent_id", get(get_agent)) + .route("/v1/agents/:agent_id/notified", post(mark_agent_notified)) .route("/v1/mailbox", get(get_mailbox)) .route("/v1/turns", post(post_turn)) .route("/v1/turns/:turn_id/events", get(stream_events)) @@ -338,6 +339,20 @@ async fn get_agent( Ok(Json(task)) } +async fn mark_agent_notified( + State(state): State>, + headers: HeaderMap, + AxumPath(agent_id): AxumPath, +) -> Result { + authorize(&headers, &state.config.auth_token)?; + RuntimeTaskStore::new() + .mark_notified(&agent_id) + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)? + .filter(|task| task.kind == RuntimeTaskKind::Agent) + .ok_or(StatusCode::NOT_FOUND)?; + Ok(StatusCode::ACCEPTED) +} + async fn get_mailbox( State(state): State>, headers: HeaderMap, diff --git a/rust/crates/claw-telegram/src/gateway.rs b/rust/crates/claw-telegram/src/gateway.rs index 2b3220b..48d4fb2 100644 --- a/rust/crates/claw-telegram/src/gateway.rs +++ b/rust/crates/claw-telegram/src/gateway.rs @@ -26,6 +26,7 @@ use crate::worker_client::{WorkerClient, WorkerClientError}; const BUSY_MESSAGE: &str = "An agent turn is already running for this profile. Use /cancel to stop it, or wait for it to finish."; +const BACKGROUND_AGENT_NOTIFY_INTERVAL_SECS: u64 = 5; pub struct TelegramGateway { config: GatewayConfig, @@ -72,6 +73,8 @@ impl TelegramGateway { me.username.as_deref().unwrap_or("(unknown)") ); + tokio::spawn(self.spawn_background_agent_notifier()); + let mut offset = self.load_offset()?; loop { match self @@ -420,6 +423,24 @@ impl TelegramGateway { Ok(()) } + fn spawn_background_agent_notifier( + &self, + ) -> impl std::future::Future + Send + 'static { + let api = self.api.clone(); + let config = self.config.clone(); + async move { + loop { + if let Err(error) = notify_background_agents_once(&api, &config).await { + eprintln!("background agent notifier error: {error}"); + } + tokio::time::sleep(Duration::from_secs( + BACKGROUND_AGENT_NOTIFY_INTERVAL_SECS, + )) + .await; + } + } + } + fn spawn_delivery_task( &self, profile_id: String, @@ -1051,6 +1072,64 @@ async fn clear_active_turn( active_turns.lock().await.remove(profile_id); } +async fn notify_background_agents_once( + api: &TelegramApi, + config: &GatewayConfig, +) -> Result<(), GatewayError> { + let manifest = load_manifest(config)?; + for profile in &manifest.profiles { + let chat_ids = profile + .channels + .iter() + .filter_map(|channel| channel.telegram_user_id()) + .collect::>(); + if chat_ids.is_empty() { + continue; + } + let base_url = format!( + "http://{}:{}", + profile.worker.container_name, manifest.worker_defaults.bind_port + ); + let client = match WorkerClient::new(&base_url, &config.worker_auth_token) { + Ok(client) => client, + Err(error) => { + eprintln!( + "background notifier skipped profile {}: {error}", + profile.profile_id + ); + continue; + } + }; + let agents = match client.agents().await { + Ok(agents) => agents, + Err(error) => { + eprintln!( + "background notifier could not list agents for profile {}: {error}", + profile.profile_id + ); + continue; + } + }; + for agent in agents + .agents + .into_iter() + .filter(|agent| agent.status.is_terminal() && !agent.notified) + { + let text = render_background_agent_terminal_notice(&agent); + let mut delivered = false; + for chat_id in &chat_ids { + if api.send_message(*chat_id, &text, None).await.is_ok() { + delivered = true; + } + } + if delivered { + let _ = client.mark_agent_notified(&agent.task_id).await; + } + } + } + Ok(()) +} + fn turn_source(user: &User, message: &Message) -> TurnSource { TurnSource { channel: "telegram".to_string(), @@ -1325,6 +1404,60 @@ fn render_agent_snapshot(agent: &RuntimeTaskRecord) -> String { ) } +fn render_background_agent_terminal_notice(agent: &RuntimeTaskRecord) -> String { + let title = match agent.status { + runtime::RuntimeTaskStatus::Completed => "Background agent completed", + runtime::RuntimeTaskStatus::Failed => "Background agent failed", + runtime::RuntimeTaskStatus::Stopped => "Background agent stopped", + runtime::RuntimeTaskStatus::Running => "Background agent update", + }; + let mut lines = vec![ + title.to_string(), + format!( + "{} [{}] {}", + agent + .agent_name + .as_deref() + .or(agent.agent_id.as_deref()) + .unwrap_or(agent.task_id.as_str()), + agent.status, + agent.description + ), + format!("Team: {}", agent.team_name.as_deref().unwrap_or("(none)")), + format!("CWD: {}", agent.cwd.as_deref().unwrap_or("(none)")), + ]; + if let Some(worktree_path) = agent.worktree_path.as_deref() { + lines.push(format!("Worktree: {worktree_path}")); + lines.push(format!( + "Worktree branch: {}", + agent.worktree_branch.as_deref().unwrap_or("(none)") + )); + } + if let Some(result) = agent.final_result.as_deref().filter(|value| !value.trim().is_empty()) { + lines.push(String::from("Result:")); + lines.push(truncate_notice_text(result, 1200)); + } + if let Some(error) = agent.error.as_deref().filter(|value| !value.trim().is_empty()) { + lines.push(String::from("Error:")); + lines.push(truncate_notice_text(error, 1200)); + } + if let Some(path) = agent.output_file.as_deref() { + lines.push(format!("Output file: {path}")); + } + lines.join("\n") +} + +fn truncate_notice_text(value: &str, max_chars: usize) -> String { + let mut truncated = String::new(); + for ch in value.chars().take(max_chars) { + truncated.push(ch); + } + if value.chars().count() > max_chars { + truncated.push_str("\n…"); + } + truncated +} + fn render_runtime_task(label: &str, task: &RuntimeTaskRecord) -> String { let worktree = task.worktree_path.as_deref().map_or_else(String::new, |path| { format!( diff --git a/rust/crates/claw-telegram/src/worker_client.rs b/rust/crates/claw-telegram/src/worker_client.rs index e02854e..154c1bd 100644 --- a/rust/crates/claw-telegram/src/worker_client.rs +++ b/rust/crates/claw-telegram/src/worker_client.rs @@ -86,6 +86,14 @@ impl WorkerClient { self.get_json(&format!("/v1/agents/{agent_id}")).await } + pub async fn mark_agent_notified(&self, agent_id: &str) -> Result<(), WorkerClientError> { + self.post_no_content( + &format!("/v1/agents/{agent_id}/notified"), + &serde_json::json!({}), + ) + .await + } + pub async fn mailbox(&self) -> Result { self.get_json("/v1/mailbox").await } diff --git a/rust/crates/runtime/src/runtime_task_store.rs b/rust/crates/runtime/src/runtime_task_store.rs index 254442c..c6514bb 100644 --- a/rust/crates/runtime/src/runtime_task_store.rs +++ b/rust/crates/runtime/src/runtime_task_store.rs @@ -273,6 +273,21 @@ impl RuntimeTaskStore { Ok(Some(record)) } + pub fn mark_notified(&self, task_id: &str) -> io::Result> { + let _lock = store_lock() + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let Some(mut record) = self.get(task_id)? else { + return Ok(None); + }; + if record.notified { + return Ok(Some(record)); + } + record.notified = true; + self.write_locked(&record)?; + Ok(Some(record)) + } + pub fn output( &self, task_id: &str, @@ -479,4 +494,42 @@ mod tests { let _ = fs::remove_dir_all(&root); std::env::remove_var("CLAW_WORKER_STATE_ROOT"); } + + #[test] + fn runtime_tasks_can_be_marked_notified() { + let _lock = test_env_lock(); + let root = std::env::temp_dir().join("runtime-task-store-notified"); + let _ = fs::remove_dir_all(&root); + std::env::set_var("CLAW_WORKER_STATE_ROOT", &root); + + let store = RuntimeTaskStore::new(); + let record = store + .create_agent_task( + "agent-notify-1".to_string(), + "agent-notify".to_string(), + "Notify me".to_string(), + "Finish the work".to_string(), + root.join("agent.md").display().to_string(), + None, + Some("/tmp/runtime-agent".to_string()), + None, + None, + ) + .expect("create agent task"); + + let notified = store + .mark_notified(&record.task_id) + .expect("mark notified should succeed") + .expect("task should exist"); + assert!(notified.notified); + + let persisted = store + .get(&record.task_id) + .expect("reload task") + .expect("task exists"); + assert!(persisted.notified); + + let _ = fs::remove_dir_all(&root); + std::env::remove_var("CLAW_WORKER_STATE_ROOT"); + } }