orch/channels/
mod.rs

1//! Channel abstraction — the core trait for all I/O surfaces.
2//!
3//! Every external interface (GitHub, Telegram, Discord, tmux) implements
4//! the `Channel` trait. This gives the engine a uniform way to:
5//! - Receive commands/messages
6//! - Send task updates and agent output
7//!
8//! Channels are bidirectional and async. The engine doesn't care whether
9//! a message came from a Telegram DM, a GitHub issue comment, or a
10//! Discord thread — it processes the same `IncomingMessage` type.
11
12pub mod capture;
13pub mod discord;
14pub mod discord_ws;
15pub mod github;
16pub mod notification;
17pub mod routing;
18pub mod slack;
19pub mod stream;
20pub mod telegram;
21pub mod tmux;
22pub mod transport;
23
24use std::fmt;
25use std::fmt::Debug;
26use std::fmt::Formatter;
27
28use async_trait::async_trait;
29use serde::{Deserialize, Serialize};
30use tokio::sync::mpsc::Receiver;
31
32/// Escape `&`, `<` and `>` for Telegram's HTML parse_mode, which rejects bare ones
33pub(crate) fn html_escape(s: &str) -> String {
34    s.replace('&', "&amp;")
35        .replace('<', "&lt;")
36        .replace('>', "&gt;")
37}
38
39/// Split content into chunks of at most `max_bytes`, never inside a UTF-8 character
40pub(crate) fn split_utf8(content: &str, max_bytes: usize) -> Vec<String> {
41    if content.len() <= max_bytes {
42        return vec![content.to_string()];
43    }
44
45    let mut chunks = Vec::new();
46    let mut start = 0;
47    let total = content.len();
48
49    while start < total {
50        let mut end = (start + max_bytes).min(total);
51        while end > start && !content.is_char_boundary(end) {
52            end -= 1;
53        }
54        if end == start {
55            // A single char over the limit still goes out whole
56            if let Some(ch) = content[start..].chars().next() {
57                end = start + ch.len_utf8();
58            } else {
59                break;
60            }
61        }
62        chunks.push(content[start..end].to_string());
63        start = end;
64    }
65
66    chunks
67}
68
69/// A message received from any channel.
70#[derive(Debug, Clone, Serialize, Deserialize)]
71pub struct IncomingMessage {
72    /// Channel that produced this message (e.g. "github", "telegram", "discord")
73    pub channel: String,
74    /// Unique ID within the channel (comment ID, message ID, etc.)
75    pub id: String,
76    /// Thread/conversation context (issue number, chat ID, channel ID)
77    pub thread_id: String,
78    /// Author identifier
79    pub author: String,
80    /// Raw text content
81    pub body: String,
82    /// When the message was created
83    pub timestamp: chrono::DateTime<chrono::Utc>,
84    /// Optional metadata (labels, attachments, etc.)
85    #[serde(default)]
86    pub metadata: serde_json::Value,
87    /// Optional forum topic / thread ID (e.g. Telegram forum topic)
88    #[serde(default)]
89    pub topic_id: Option<String>,
90}
91
92/// A message to send to a channel.
93#[derive(Debug, Clone, Serialize, Deserialize)]
94pub struct OutgoingMessage {
95    /// Target thread/conversation
96    pub thread_id: String,
97    /// Message body (markdown)
98    pub body: String,
99    /// Optional: reply to a specific message ID
100    pub reply_to: Option<String>,
101    /// Optional metadata for channel-specific features
102    #[serde(default)]
103    pub metadata: serde_json::Value,
104    /// Optional forum topic / thread ID (e.g. Telegram forum topic)
105    #[serde(default)]
106    pub topic_id: Option<String>,
107}
108
109/// An agent output chunk for streaming.
110#[derive(Debug, Clone)]
111pub struct OutputChunk {
112    pub content: String,
113    pub is_final: bool,
114}
115
116/// The core channel trait. All external interfaces implement this.
117#[async_trait]
118pub trait Channel: Send + Sync {
119    /// Human-readable channel name.
120    fn name(&self) -> &str;
121
122    /// Start listening for incoming messages.
123    /// Returns a receiver that the engine polls.
124    async fn start(&self) -> anyhow::Result<Receiver<IncomingMessage>>;
125
126    /// Send a message to a thread/conversation.
127    async fn send(&self, msg: &OutgoingMessage) -> anyhow::Result<()>;
128
129    /// Acknowledge an interactive callback to dismiss the loading indicator.
130    ///
131    /// Telegram requires `answerCallbackQuery` to dismiss the spinner after a button press.
132    /// Discord buttons are already acknowledged in the websocket handler — no-op here.
133    /// Default implementation is a no-op so non-interactive channels need not implement it.
134    async fn ack_interaction(&self, _callback_query_id: &str) -> anyhow::Result<()> {
135        Ok(())
136    }
137
138    /// Check if this channel is healthy/connected.
139    async fn health_check(&self) -> anyhow::Result<()>;
140}
141
142/// Registry of active channels.
143pub struct ChannelRegistry {
144    channels: Vec<Box<dyn Channel>>,
145}
146
147impl Debug for ChannelRegistry {
148    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
149        f.debug_list()
150            .entries(self.channels.iter().map(|c| c.name()))
151            .finish()
152    }
153}
154
155impl ChannelRegistry {
156    pub fn new() -> Self {
157        Self {
158            channels: Vec::new(),
159        }
160    }
161
162    pub fn register(&mut self, channel: Box<dyn Channel>) {
163        tracing::info!(channel = channel.name(), "registered channel");
164        self.channels.push(channel);
165    }
166
167    pub fn iter(&self) -> impl Iterator<Item = &dyn Channel> {
168        self.channels.iter().map(AsRef::as_ref)
169    }
170}
orch/channels/mod.rs

orch/channels/
mod.rs
