talon_core/
channel.rs

1//! Messaging channel abstraction.
2//!
3//! A channel (HTTP/CoPilot, Telegram, webhook, Emacs, ...) owns a set of JIDs,
4//! delivers inbound messages to the host through [`ChannelHost`], and sends
5//! outbound text. Channels are constructed by the host with an
6//! `Arc<dyn ChannelHost>` and must keep any background tasks they spawn
7//! (servers, pollers) tied to their own lifetime (`disconnect` stops them).
8
9use std::collections::HashMap;
10
11use async_trait::async_trait;
12
13use crate::Result;
14use crate::types::{ChatMetadata, NewMessage, RegisteredGroup};
15
16/// Text dialect a channel renders. Drives Markdown -> native conversion in
17/// `talon-router`.
18#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
19pub enum ChannelType {
20    /// Passthrough Markdown (HTTP API, Discord, Emacs).
21    Markdown,
22    Telegram,
23    WhatsApp,
24    Slack,
25    Signal,
26    /// Plain text, all markup stripped (webhook payloads).
27    Plain,
28}
29
30#[async_trait]
31pub trait Channel: Send + Sync {
32    /// Stable short name (`"http"`, `"telegram"`, ...).
33    fn name(&self) -> &str;
34
35    /// Dialect for outbound formatting.
36    fn channel_type(&self) -> ChannelType {
37        ChannelType::Markdown
38    }
39
40    /// Start the channel (bind servers, start polling). Must return once the
41    /// channel is ready; long-running work goes into spawned tasks.
42    async fn connect(&self) -> Result<()>;
43
44    /// Deliver `text` to `jid`. Text is already formatted for `channel_type`.
45    async fn send_message(&self, jid: &str, text: &str) -> Result<()>;
46
47    fn is_connected(&self) -> bool;
48
49    /// Whether this channel is responsible for `jid`.
50    fn owns_jid(&self, jid: &str) -> bool;
51
52    /// Stop background tasks and release sockets.
53    async fn disconnect(&self) -> Result<()>;
54
55    /// Optional typing indicator.
56    async fn set_typing(&self, _jid: &str, _is_typing: bool) -> Result<()> {
57        Ok(())
58    }
59
60    /// Called when the agent finishes a turn successfully, before the
61    /// container goes idle (lets streaming channels close their streams).
62    async fn on_turn_complete(&self, _jid: &str) -> Result<()> {
63        Ok(())
64    }
65
66    /// Optional: refresh chat names from the platform.
67    async fn sync_groups(&self, _force: bool) -> Result<()> {
68        Ok(())
69    }
70}
71
72/// Callbacks a channel uses to talk back to the host orchestrator.
73/// Implemented by the `talahon` binary.
74#[async_trait]
75pub trait ChannelHost: Send + Sync {
76    /// Deliver an inbound message. The host applies the sender allowlist,
77    /// persists it, and wakes the message loop. Never blocks on the agent.
78    async fn on_message(&self, chat_jid: &str, message: NewMessage);
79
80    /// Report chat metadata discovered by the channel.
81    async fn on_chat_metadata(&self, metadata: ChatMetadata);
82
83    /// Snapshot of all registered groups keyed by JID.
84    fn registered_groups(&self) -> HashMap<String, RegisteredGroup>;
85
86    /// Register (or update) a lane. In-memory lanes (per-user HTTP lanes)
87    /// pass `persist = false`.
88    async fn register_group(&self, jid: &str, group: RegisteredGroup, persist: bool);
89
90    /// Drop an in-memory lane registration. Never removes a persisted group.
91    async fn unregister_group(&self, jid: &str) -> bool;
92
93    /// Drop a lane's stored session so its next run starts fresh.
94    /// Returns false for an unknown JID.
95    async fn clear_session(&self, jid: &str) -> bool;
96
97    /// Scheduled tasks were created/changed out of band (e.g. by the HTTP
98    /// API); wake the scheduler.
99    fn notify_tasks_changed(&self);
100
101    /// Live queue / container status for observability endpoints.
102    fn queue_status(&self) -> QueueStatus {
103        QueueStatus::default()
104    }
105}
106
107/// Snapshot of the group queue for `/status`-style endpoints.
108#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
109pub struct QueueStatus {
110    pub active_containers: usize,
111    pub max_containers: usize,
112    pub waiting_groups: usize,
113    /// JIDs with a running container.
114    pub active_jids: Vec<String>,
115}
talon_core/channel.rs

talon_core/
channel.rs
