orch-cli 0.91.2

Orch, the agent orchestration engine: runs coding agents on tasks and cron jobs
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

//! Reacts to status transitions — pushes notifications to channels for meaningful states.
//!
//! Only review-active and terminal transitions are forwarded by default
//! (`level: all`): `done`, `in_review`, `blocked`, `failed`. `needs_review`
//! is intentionally suppressed: it's the agent-handoff signal that always
//! pairs with a follow-up `in_review`/`blocked`/`failed` event within
//! seconds — notifying on it produces a redundant ping. Other intermediate
//! transitions (`new`, `routed`, `in_progress`) are suppressed unless
//! `notifications.level: verbose` is set in config.

use crate::channels::notification::{NotificationLevel, TaskNotification};
use crate::channels::transport::Transport;
use crate::engine::events::TaskEvent;
use std::sync::Arc;
use tokio::sync::broadcast;

/// Spawn a task that listens for events and pushes to transport.
///
/// This subscriber is spawned ONCE (not per-project) since the transport
/// handles all repos. It filters events using `NotificationLevel::should_notify()`
/// so only meaningful status transitions reach the channels.
pub fn spawn(mut rx: broadcast::Receiver<TaskEvent>, transport: Arc<Transport>) {
    let level = NotificationLevel::from_config();
    tokio::spawn(async move {
        loop {
            match rx.recv().await {
                Ok(event) => {
                    if !level.should_notify(&event.new_status) {
                        continue;
                    }
                    let notification = TaskNotification {
                        task_id: event.task_id.clone(),
                        title: event.title.unwrap_or_default(),
                        status: event.new_status.clone(),
                        agent: event.agent.unwrap_or_default(),
                        duration_seconds: event.duration_seconds.unwrap_or(0.0),
                        summary: event.summary.unwrap_or_default(),
                        repo: Some(event.repo.clone()),
                        notify_target: None,
                        pr_number: event.pr_number.clone(),
                    };
                    transport.push_notification(notification);
                }
                Err(broadcast::error::RecvError::Lagged(n)) => {
                    tracing::debug!(n, "notify subscriber lagged");
                }
                Err(_) => break, // Channel closed
            }
        }
    });
}