1use std::path::PathBuf;
4use std::sync::Arc;
5
6use axum::{extract::State, routing::post, Json, Router};
7
8use crate::agent::AgentRunner;
9use crate::channels::telegram::TelegramChannel;
10use crate::channels::telegram::TelegramIngressFilter;
11use crate::channels::Channel;
12use crate::channels::IncomingMessage;
13use crate::config::ChannelConfig;
14use crate::memory::MemoryBackend;
15#[cfg(feature = "channel-telegram")]
16use crate::tools::message::MessageTool;
17
18pub struct TelegramChatRun<'a> {
19 pub runner: Arc<AgentRunner>,
20 pub memory: Arc<dyn MemoryBackend>,
21 pub token: String,
22 pub chat_id: i64,
23 pub model: String,
24 pub skills_count: usize,
25 pub workspace: PathBuf,
26 pub channel_cfg: &'a ChannelConfig,
27 pub cron_runtime: Option<(
28 tokio::sync::mpsc::Receiver<crate::cron_scheduler::DueJob>,
29 Arc<tokio::sync::Notify>,
30 Arc<crate::cron_scheduler::CronScheduler>,
31 )>,
32}
33
34pub async fn run_telegram_chat(run: TelegramChatRun<'_>) -> anyhow::Result<()> {
35 let TelegramChatRun {
36 runner,
37 memory,
38 token,
39 chat_id,
40 model,
41 skills_count,
42 workspace,
43 channel_cfg,
44 cron_runtime,
45 } = run;
46 let ingress = TelegramIngressFilter {
47 allowed_chat_ids: channel_cfg.allowed_chat_ids.clone(),
48 allowed_sender_ids: channel_cfg.allowed_sender_ids.clone(),
49 };
50
51 let tg = TelegramChannel::new(token.clone(), chat_id)
52 .with_memory(memory.clone())
53 .with_ingress_filter(ingress.clone());
54 let tg_arc = Arc::new(tg.clone());
55
56 runner.add_tool(Arc::new(MessageTool::new(tg_arc))).await;
57
58 println!("apollo — {} via Telegram", model);
59 println!(" Workspace: {}", workspace.display());
60 println!(" Chat ID: {}", chat_id);
61 println!(" Tools: {}", runner.list_tools().await.join(", "));
62 println!(" API: http://127.0.0.1:31337/message");
63 println!(" Listening for messages...");
64
65 let mut ch = TelegramChannel::new(token, chat_id)
66 .with_memory(memory.clone())
67 .with_ingress_filter(ingress);
68 let mut rx = ch.start().await?;
69
70 let cron_shutdown = cron_runtime
71 .as_ref()
72 .map(|(_, shutdown, _)| Arc::clone(shutdown));
73 let _cron_handle = cron_runtime.map(|(mut cron_rx, _, scheduler)| {
74 let runner = Arc::clone(&runner);
75 let channel = ch.clone();
76 tokio::spawn(async move {
77 while let Some(due) = cron_rx.recv().await {
78 let job = due.job;
79 let job_id = job.id.clone().unwrap_or_default();
80 let run_token = job.run_token.clone().unwrap_or_default();
81 if job.channel != channel.name() {
82 let _ = scheduler.release_run(&job_id, &run_token).await;
83 continue;
84 }
85 let msg = IncomingMessage {
86 id: format!("cron-{job_id}"),
87 sender_id: "scheduler".to_string(),
88 sender_name: Some("Scheduler".to_string()),
89 chat_id: job.chat_id.clone(),
90 text: job.task.clone(),
91 is_group: false,
92 reply_to: None,
93 timestamp: chrono::Utc::now(),
94 };
95 match runner
96 .handle_message_with_model(
97 &msg,
98 &channel,
99 (!job.model.is_empty()).then_some(job.model.as_str()),
100 )
101 .await
102 {
103 Ok(response) => {
104 let delivery = channel
105 .send(crate::channels::OutgoingMessage {
106 chat_id: job.chat_id,
107 text: response,
108 reply_to: None,
109 })
110 .await;
111 if let Err(error) = delivery {
112 let _ = scheduler
113 .fail_run(&job_id, &run_token, &error.to_string())
114 .await;
115 } else {
116 let _ = scheduler.mark_run(&job_id, &run_token, &job.schedule).await;
117 }
118 }
119 Err(error) => {
120 let _ = scheduler
121 .fail_run(&job_id, &run_token, &error.to_string())
122 .await;
123 }
124 }
125 }
126 })
127 });
128
129 let (cli_tx, mut cli_rx) = tokio::sync::mpsc::channel::<IncomingMessage>(32);
130 spawn_local_message_bridge(cli_tx, chat_id);
131
132 let processing = Arc::new(std::sync::atomic::AtomicBool::new(false));
133
134 loop {
135 let msg = tokio::select! {
136 Some(msg) = rx.recv() => msg,
137 Some(msg) = cli_rx.recv() => msg,
138 else => break,
139 };
140 if msg.chat_id != chat_id.to_string() {
141 tracing::warn!("Ignoring Telegram message for unbound chat {}", msg.chat_id);
142 continue;
143 }
144 let text = msg.text.trim();
145
146 if processing.load(std::sync::atomic::Ordering::SeqCst) && !text.starts_with('/') {
147 runner.steer(text.to_string());
148 let _ = tg.send_message("📌 Noted — steering current task.").await;
149 continue;
150 }
151
152 if text.starts_with('/')
153 && handle_command(&runner, &memory, &tg, &msg, text, skills_count).await?
154 {
155 continue;
156 }
157
158 processing.store(true, std::sync::atomic::Ordering::SeqCst);
159
160 let _ = tg.send_typing(&msg.chat_id).await;
162
163 match runner.handle_message(&msg, &tg).await {
164 Ok(response) => {
165 if response.trim().is_empty() {
166 continue;
167 }
168 let _ = tg.send_message(&response).await;
169 }
170 Err(error) => {
171 let _ = tg.send_message(&format!("❌ {}", error)).await;
172 }
173 }
174
175 processing.store(false, std::sync::atomic::Ordering::SeqCst);
176 }
177
178 if let Some(shutdown) = cron_shutdown {
179 shutdown.notify_waiters();
180 }
181 Ok(())
182}
183
184#[derive(Clone)]
185struct BridgeState {
186 tx: tokio::sync::mpsc::Sender<IncomingMessage>,
187 chat_id: String,
188}
189
190async fn handle_bridge_message(
191 State(state): State<BridgeState>,
192 Json(body): Json<serde_json::Value>,
193) -> (axum::http::StatusCode, &'static str) {
194 let text = body["message"].as_str().unwrap_or("").to_string();
195 if text.is_empty() {
196 return (
197 axum::http::StatusCode::BAD_REQUEST,
198 "missing 'message' field",
199 );
200 }
201 let msg = IncomingMessage {
202 id: format!("cli-{}", chrono::Utc::now().timestamp()),
203 chat_id: state.chat_id,
204 sender_id: "cli".to_string(),
205 sender_name: Some("CLI".to_string()),
206 text,
207 timestamp: chrono::Utc::now(),
208 is_group: false,
209 reply_to: None,
210 };
211 let _ = state.tx.send(msg).await;
212 (axum::http::StatusCode::OK, "queued")
213}
214
215fn spawn_local_message_bridge(cli_tx: tokio::sync::mpsc::Sender<IncomingMessage>, chat_id: i64) {
216 let state = BridgeState {
217 tx: cli_tx,
218 chat_id: chat_id.to_string(),
219 };
220 tokio::spawn(async move {
221 let app = Router::new()
222 .route("/message", post(handle_bridge_message))
223 .with_state(state);
224
225 let listener = tokio::net::TcpListener::bind("127.0.0.1:31337")
226 .await
227 .unwrap();
228 axum::serve(listener, app).await.unwrap();
229 });
230}
231
232async fn handle_command(
233 runner: &Arc<AgentRunner>,
234 memory: &Arc<dyn MemoryBackend>,
235 tg: &TelegramChannel,
236 msg: &IncomingMessage,
237 text: &str,
238 discovered_skills_len: usize,
239) -> anyhow::Result<bool> {
240 let parts: Vec<&str> = text.splitn(2, ' ').collect();
241 let cmd = parts[0].to_lowercase();
242 let arg = parts.get(1).map(|s| s.trim()).unwrap_or("");
243
244 match cmd.as_str() {
245 "/stop" | "/cancel" => {
246 let _ = tg.send_message("⛔ Stopped.").await;
247 Ok(true)
248 }
249 "/help" => {
250 let _ = tg
251 .send_message(
252 "🐾 *apollo commands:*\n\n\
253 /stop — Stop current operation (saves tokens!)\n\
254 /help — Show this message\n\
255 /model — Show current model\n\
256 /model <name> — Switch model\n\
257 /models — List available models\n\
258 /tools — List available tools\n\
259 /status — Bot status\n\
260 /cost — API usage & spending\n\
261 /reset — Clear conversation history\n\n\
262 Everything else is sent to the AI.",
263 )
264 .await;
265 Ok(true)
266 }
267 "/model" | "/model@apollo_bot" => {
268 if arg.is_empty() {
269 let _ = tg
270 .send_message(&format!(
271 "Current model: `{}`\n\nUse `/model <name>` to switch.\nUse `/models` for available options.",
272 runner.get_model()
273 ))
274 .await;
275 } else {
276 runner.set_model(arg);
277 let _ = tg
278 .send_message(&format!("✅ Model switched to: `{}`", arg))
279 .await;
280 tracing::info!("Model switched to: {}", arg);
281 }
282 Ok(true)
283 }
284 "/models" => {
285 let _ = tg
286 .send_message(
287 "📋 *Available models:*\n\n\
288 `claude-sonnet-4-5` — Fast, smart (default)\n\
289 `claude-opus-4` — Most capable\n\
290 `claude-haiku-3-5` — Fastest, cheapest\n\n\
291 Switch with: `/model claude-opus-4`",
292 )
293 .await;
294 Ok(true)
295 }
296 "/tools" => {
297 let tool_list = runner.list_tools().await;
298 let formatted = tool_list
299 .iter()
300 .map(|t| format!("• `{}`", t))
301 .collect::<Vec<_>>()
302 .join("\n");
303 let _ = tg
304 .send_message(&format!(
305 "🔧 *Available tools ({}):\n\n{}*",
306 tool_list.len(),
307 formatted
308 ))
309 .await;
310 Ok(true)
311 }
312 "/status" => {
313 let _ = tg
314 .send_message(&format!(
315 "🐾 *apollo status:*\n\n\
316 Model: `{}`\n\
317 Tools: {}\n\
318 Skills: {}\n\
319 Channel: Telegram\n\
320 PID: {}",
321 runner.get_model(),
322 runner.list_tools().await.len(),
323 discovered_skills_len,
324 std::process::id(),
325 ))
326 .await;
327 Ok(true)
328 }
329 "/reset" => {
330 let _ = memory
331 .forget("chat", &format!("conv_{}", msg.chat_id))
332 .await;
333 let _ = tg.send_message("🗑 Conversation history cleared.").await;
334 Ok(true)
335 }
336 "/cost" => {
337 let summary = runner.get_cost_summary().await;
338 let mut by_model: Vec<_> = summary.by_model.iter().collect();
339 by_model.sort_by(|a, b| b.1.partial_cmp(a.1).unwrap());
340
341 let model_breakdown = if by_model.is_empty() {
342 "No usage yet.".to_string()
343 } else {
344 by_model
345 .iter()
346 .map(|(model, cost)| format!(" • {}: ${:.4}", model, cost))
347 .collect::<Vec<_>>()
348 .join("\n")
349 };
350
351 let _ = tg
352 .send_message(&format!(
353 "💰 *Cost Summary:*\n\n\
354 Total: ${:.4}\n\
355 Tokens: {}\n\
356 Calls: {}\n\n\
357 By model:\n{}",
358 summary.total_cost, summary.total_tokens, summary.call_count, model_breakdown,
359 ))
360 .await;
361 Ok(true)
362 }
363 "/start" => {
364 let _ = tg
365 .send_message(
366 "🐾 *apollo* — AI assistant\n\n\
367 Just type a message to chat.\n\
368 Use /help for commands.\n\
369 Use /tools to see what I can do.",
370 )
371 .await;
372 Ok(true)
373 }
374 _ => Ok(false),
375 }
376}