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

orch/channels/
mod.rs
