Notify Telegram about background agent completion

rust/crates/claw-profile-worker/src/server.rs

@@ -55,6 +55,7 @@ fn app_router(config: WorkerConfig, runtime: Arc<dyn WorkerRuntime>) -> 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<Arc<AppState>>,
+    headers: HeaderMap,
+    AxumPath(agent_id): AxumPath<String>,
+) -> Result<StatusCode, StatusCode> {
+    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<Arc<AppState>>,
     headers: HeaderMap,

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<Output = ()> + 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::<Vec<_>>();
+        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!(

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<WorkerMailboxSummaryResponse, WorkerClientError> {
         self.get_json("/v1/mailbox").await
     }

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<Option<RuntimeTaskRecord>> {
+        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");
+    }
 }