bamboo_server/connect/
mod.rs

1//! bamboo-connect: drive bamboo sessions from IM platforms (Telegram first).
2//!
3//! Issue #452 (MVP, phase 1 of epic #447). Sibling module of
4//! `schedule_app`/`notify_sinks` — the closest precedents: `notify_sinks` is
5//! one-way outbound, `schedule_app::manager` is the canonical
6//! background-execution pattern this module's [`bridge::ConnectBridge`]
7//! reuses verbatim (`spawn_session_execution`, event-forwarder, runner
8//! reservation).
9//!
10//! ```text
11//! connect/
12//!   platform.rs      — the Platform trait + Capabilities + message types
13//!   bridge.rs         — chat ⇄ bamboo-session routing, busy lock, queueing
14//!   render.rs         — AgentEvent stream → platform messages
15//!   platforms/telegram.rs — long-poll adapter
16//! ```
17//!
18//! [`ConnectManager`] is constructed once at server startup (mirrors
19//! `ScheduleManager` / the notification relay — see
20//! `app_state::init::build_connect_manager`) and is FULLY INERT when
21//! `config.connect.platforms` is empty: no `[connect]` section in
22//! `config.json` means zero background tasks are spawned.
23
24pub mod bridge;
25pub mod platform;
26pub mod platforms;
27pub mod render;
28
29pub use bridge::{ConnectBridge, ConnectContext, SessionKey};
30pub use platform::{
31    Capabilities, InboundMessage, MessageRef, OutboundMessage, Platform, PlatformError,
32    PlatformResult, ReplyCtx,
33};
34
35use std::path::PathBuf;
36use std::sync::Arc;
37
38use tokio::sync::mpsc;
39
40use bamboo_llm::Config;
41
42/// Owns every configured platform's long-poll/dispatch background task plus
43/// the shared [`ConnectBridge`]. All tasks are aborted when the manager
44/// drops (mirrors `app_state::builder::EmbeddedBroker`/`HealthMonitor`'s
45/// Drop-based stop — the closest server-lifecycle precedent for a
46/// fire-and-forget background subsystem).
47pub struct ConnectManager {
48    tasks: Vec<tokio::task::JoinHandle<()>>,
49}
50
51impl ConnectManager {
52    /// Builds the bridge, loads its persisted session map, and starts one
53    /// long-poll + one dispatch task per recognized platform entry in
54    /// `config_snapshot.connect.platforms`. An empty (or absent) `[connect]`
55    /// config starts zero tasks — fully inert by default, per #452.
56    pub async fn start(
57        ctx: ConnectContext,
58        config_snapshot: &Config,
59        data_dir: Option<PathBuf>,
60    ) -> Self {
61        let map_path = data_dir.map(|dir| dir.join("connect_sessions.json"));
62        let bridge = Arc::new(ConnectBridge::new(ctx, map_path));
63        bridge.load_session_map().await;
64
65        let mut tasks = Vec::new();
66        for platform_cfg in &config_snapshot.connect.platforms {
67            match platform_cfg.platform_type.as_str() {
68                "telegram" => {
69                    let token = platform_cfg.token.clone().unwrap_or_default();
70                    if token.trim().is_empty() {
71                        tracing::warn!(
72                            "connect: telegram platform configured without a token; skipping"
73                        );
74                        continue;
75                    }
76                    if platform_cfg.allow_from.is_empty() {
77                        tracing::warn!(
78                            "connect: telegram platform has an EMPTY allow_from list — every \
79                             inbound message will be denied until you add allowed user ids to \
80                             connect.platforms[].allow_from"
81                        );
82                    }
83
84                    let platform: Arc<dyn Platform> =
85                        Arc::new(platforms::telegram::TelegramPlatform::new(token));
86                    let allow_from = platform_cfg.allow_from.clone();
87                    let (tx, rx) = mpsc::channel(64);
88
89                    let platform_for_start = platform.clone();
90                    tasks.push(tokio::spawn(async move {
91                        if let Err(error) = platform_for_start.start(tx).await {
92                            tracing::warn!("connect: telegram platform loop exited: {error}");
93                        }
94                    }));
95
96                    let bridge_for_dispatch = bridge.clone();
97                    let platform_for_dispatch = platform.clone();
98                    tasks.push(tokio::spawn(dispatch_loop(
99                        bridge_for_dispatch,
100                        platform_for_dispatch,
101                        allow_from,
102                        rx,
103                    )));
104
105                    tracing::info!("connect: started telegram platform");
106                }
107                other => {
108                    tracing::warn!("connect: unknown platform type '{other}'; skipping");
109                }
110            }
111        }
112
113        Self { tasks }
114    }
115}
116
117impl Drop for ConnectManager {
118    fn drop(&mut self) {
119        for task in &self.tasks {
120            task.abort();
121        }
122    }
123}
124
125/// Pulls inbound messages off a single platform's channel and hands each to
126/// the bridge. Kept as its OWN task per platform (not merged into the
127/// platform's `start()` loop) so a slow/queued chat can never stall the next
128/// `getUpdates` poll — `ConnectBridge::handle_inbound` itself returns quickly
129/// (it spawns the actual run), so this loop only ever blocks briefly.
130async fn dispatch_loop(
131    bridge: Arc<ConnectBridge>,
132    platform: Arc<dyn Platform>,
133    allow_from: Vec<String>,
134    mut rx: mpsc::Receiver<InboundMessage>,
135) {
136    while let Some(msg) = rx.recv().await {
137        ConnectBridge::handle_inbound(bridge.clone(), platform.clone(), allow_from.clone(), msg)
138            .await;
139    }
140}
bamboo_server/connect/mod.rs

bamboo_server/connect/
mod.rs
