orch/engine/subscribers/
notify.rs

1//! Reacts to status transitions — pushes notifications to channels for meaningful states.
2//!
3//! Only review-active and terminal transitions are forwarded by default
4//! (`level: all`): `done`, `in_review`, `blocked`, `failed`. `needs_review`
5//! is intentionally suppressed: it's the agent-handoff signal that always
6//! pairs with a follow-up `in_review`/`blocked`/`failed` event within
7//! seconds — notifying on it produces a redundant ping. Other intermediate
8//! transitions (`new`, `routed`, `in_progress`) are suppressed unless
9//! `notifications.level: verbose` is set in config.
10
11use crate::channels::notification::{NotificationLevel, TaskNotification};
12use crate::channels::transport::Transport;
13use crate::engine::events::TaskEvent;
14use std::sync::Arc;
15use tokio::sync::broadcast;
16
17/// Spawn a task that listens for events and pushes to transport.
18///
19/// This subscriber is spawned ONCE (not per-project) since the transport
20/// handles all repos. It filters events using `NotificationLevel::should_notify()`
21/// so only meaningful status transitions reach the channels.
22pub fn spawn(mut rx: broadcast::Receiver<TaskEvent>, transport: Arc<Transport>) {
23    let level = NotificationLevel::from_config();
24    tokio::spawn(async move {
25        loop {
26            match rx.recv().await {
27                Ok(event) => {
28                    if !level.should_notify(&event.new_status) {
29                        continue;
30                    }
31                    let notification = TaskNotification {
32                        task_id: event.task_id.clone(),
33                        title: event.title.unwrap_or_default(),
34                        status: event.new_status.clone(),
35                        agent: event.agent.unwrap_or_default(),
36                        duration_seconds: event.duration_seconds.unwrap_or(0.0),
37                        summary: event.summary.unwrap_or_default(),
38                        repo: Some(event.repo.clone()),
39                        notify_target: None,
40                        pr_number: event.pr_number.clone(),
41                    };
42                    transport.push_notification(notification);
43                }
44                Err(broadcast::error::RecvError::Lagged(n)) => {
45                    tracing::debug!(n, "notify subscriber lagged");
46                }
47                Err(_) => break, // Channel closed
48            }
49        }
50    });
51}
orch/engine/subscribers/notify.rs

orch/engine/subscribers/
notify.rs
