diff --git a/app/src/components/channels/TelegramConfig.tsx b/app/src/components/channels/TelegramConfig.tsx index e28d36205..b87be36cf 100644 --- a/app/src/components/channels/TelegramConfig.tsx +++ b/app/src/components/channels/TelegramConfig.tsx @@ -323,6 +323,16 @@ const TelegramConfig = ({ definition }: TelegramConfigProps) => { return (
+
+

+ Remote control (Telegram) +

+

+ From an allowed Telegram chat, send /status, /sessions, /new, or /help. Model routing + still uses /model and /models. +

+
+ {error && (
{error} diff --git a/app/src/components/channels/__tests__/TelegramConfig.test.tsx b/app/src/components/channels/__tests__/TelegramConfig.test.tsx index e55b9f1cd..7bc841e98 100644 --- a/app/src/components/channels/__tests__/TelegramConfig.test.tsx +++ b/app/src/components/channels/__tests__/TelegramConfig.test.tsx @@ -38,6 +38,13 @@ describe('TelegramConfig', () => { expect(screen.getByText('Login with OpenHuman')).toBeInTheDocument(); }); + it('documents Telegram remote-control commands', () => { + renderWithProviders(); + expect(screen.getByText('Remote control (Telegram)')).toBeInTheDocument(); + expect(screen.getByText(/send \/status, \/sessions, \/new, or \/help/i)).toBeInTheDocument(); + expect(screen.getByText(/Model routing still uses \/model and \/models/i)).toBeInTheDocument(); + }); + it('shows credential fields for bot_token mode', () => { renderWithProviders(); expect(screen.getByPlaceholderText(/ABC-DEF1234/)).toBeInTheDocument(); diff --git a/src/openhuman/about_app/catalog.rs b/src/openhuman/about_app/catalog.rs index b15b1bf08..dd478dd00 100644 --- a/src/openhuman/about_app/catalog.rs +++ b/src/openhuman/about_app/catalog.rs @@ -917,6 +917,17 @@ const CAPABILITIES: &[Capability] = &[ status: CapabilityStatus::Beta, privacy: None, }, + Capability { + id: "channels.telegram_remote_control", + name: "Telegram Remote Control", + domain: "channels", + category: CapabilityCategory::Channels, + description: + "Operate OpenHuman from Telegram with slash commands: /status, /sessions, /new, and /help.", + how_to: "Settings > Messaging Channels > Telegram (connect), then message the bot", + status: CapabilityStatus::Beta, + privacy: None, + }, Capability { id: "channels.disconnect_platform", name: "Disconnect Messaging Platforms", diff --git a/src/openhuman/channels/providers/telegram/bus.rs b/src/openhuman/channels/providers/telegram/bus.rs new file mode 100644 index 000000000..78b4ad82d --- /dev/null +++ b/src/openhuman/channels/providers/telegram/bus.rs @@ -0,0 +1,79 @@ +//! Event-bus subscriber for Telegram remote-control lifecycle signals. + +use crate::core::event_bus::{DomainEvent, EventHandler}; +use crate::openhuman::channels::providers::telegram::session_store::with_store; +use async_trait::async_trait; +use std::path::PathBuf; + +const LOG_PREFIX: &str = "[telegram-remote]"; + +/// Tracks Telegram turn lifecycle via channel domain events and exposes busy +/// state for `/status`. +pub struct TelegramRemoteSubscriber { + workspace_dir: PathBuf, +} + +impl TelegramRemoteSubscriber { + pub fn new(workspace_dir: PathBuf) -> Self { + Self { workspace_dir } + } + + async fn set_busy(&self, reply_target: &str, busy: bool) { + let workspace_dir = self.workspace_dir.clone(); + let reply_target_owned = reply_target.to_string(); + let join_result = tokio::task::spawn_blocking(move || { + with_store(&workspace_dir, |store| { + store.set_busy(&reply_target_owned, busy); + Ok(()) + }) + }) + .await; + + match join_result { + Ok(Ok(())) => {} + Ok(Err(error)) => tracing::warn!( + "{LOG_PREFIX} failed to persist busy={busy} reply_target={reply_target}: {error}" + ), + Err(error) => tracing::warn!( + "{LOG_PREFIX} join error persisting busy={busy} reply_target={reply_target}: {error}" + ), + } + } +} + +#[async_trait] +impl EventHandler for TelegramRemoteSubscriber { + fn name(&self) -> &str { + "telegram::remote_control" + } + + fn domains(&self) -> Option<&[&str]> { + Some(&["channel"]) + } + + async fn handle(&self, event: &DomainEvent) { + match event { + DomainEvent::ChannelMessageReceived { + channel, + reply_target, + .. + } if channel == "telegram" => { + tracing::debug!("{LOG_PREFIX} turn started reply_target={reply_target}"); + self.set_busy(reply_target, true).await; + } + DomainEvent::ChannelMessageProcessed { + channel, + reply_target, + success, + elapsed_ms, + .. + } if channel == "telegram" => { + tracing::debug!( + "{LOG_PREFIX} turn finished reply_target={reply_target} success={success} elapsed_ms={elapsed_ms}" + ); + self.set_busy(reply_target, false).await; + } + _ => {} + } + } +} diff --git a/src/openhuman/channels/providers/telegram/bus_tests.rs b/src/openhuman/channels/providers/telegram/bus_tests.rs new file mode 100644 index 000000000..65ef6e6a9 --- /dev/null +++ b/src/openhuman/channels/providers/telegram/bus_tests.rs @@ -0,0 +1,65 @@ +use super::bus::TelegramRemoteSubscriber; +use crate::core::event_bus::{DomainEvent, EventHandler}; +use tempfile::tempdir; + +#[tokio::test] +async fn subscriber_marks_busy_on_received_and_clears_on_processed() { + let dir = tempdir().expect("tempdir"); + let subscriber = TelegramRemoteSubscriber::new(dir.path().to_path_buf()); + assert_eq!(subscriber.name(), "telegram::remote_control"); + assert_eq!(subscriber.domains(), Some(&["channel"][..])); + + subscriber + .handle(&DomainEvent::ChannelMessageReceived { + channel: "telegram".into(), + message_id: "m1".into(), + sender: "alice".into(), + reply_target: "chat-99".into(), + content: "hi".into(), + thread_ts: Some("1".into()), + }) + .await; + + let busy = super::session_store::with_store(dir.path(), |store| Ok(store.is_busy("chat-99"))) + .expect("store"); + assert!(busy); + + subscriber + .handle(&DomainEvent::ChannelMessageProcessed { + channel: "telegram".into(), + message_id: "m1".into(), + sender: "alice".into(), + reply_target: "chat-99".into(), + content: "hi".into(), + thread_ts: Some("1".into()), + response: "ok".into(), + elapsed_ms: 10, + success: true, + }) + .await; + + let busy = super::session_store::with_store(dir.path(), |store| Ok(store.is_busy("chat-99"))) + .expect("store"); + assert!(!busy); +} + +#[tokio::test] +async fn subscriber_ignores_non_telegram_channel_events() { + let dir = tempdir().expect("tempdir"); + let subscriber = TelegramRemoteSubscriber::new(dir.path().to_path_buf()); + + subscriber + .handle(&DomainEvent::ChannelMessageReceived { + channel: "discord".into(), + message_id: "m1".into(), + sender: "alice".into(), + reply_target: "chat-99".into(), + content: "hi".into(), + thread_ts: None, + }) + .await; + + let busy = super::session_store::with_store(dir.path(), |store| Ok(store.is_busy("chat-99"))) + .expect("store"); + assert!(!busy); +} diff --git a/src/openhuman/channels/providers/telegram/mod.rs b/src/openhuman/channels/providers/telegram/mod.rs index 9c14c9bd3..03c1a5637 100644 --- a/src/openhuman/channels/providers/telegram/mod.rs +++ b/src/openhuman/channels/providers/telegram/mod.rs @@ -1,12 +1,21 @@ //! Telegram channel — long-polls the Bot API for updates. mod attachments; +mod bus; mod channel; mod channel_core; mod channel_ops; mod channel_recv; mod channel_send; mod channel_types; +pub mod remote_control; +mod session_store; mod text; +pub use bus::TelegramRemoteSubscriber; pub use channel_types::TelegramChannel; +pub use remote_control::TelegramRemoteCommand; + +#[cfg(test)] +#[path = "bus_tests.rs"] +mod bus_tests; diff --git a/src/openhuman/channels/providers/telegram/remote_control.rs b/src/openhuman/channels/providers/telegram/remote_control.rs new file mode 100644 index 000000000..d1e2b0278 --- /dev/null +++ b/src/openhuman/channels/providers/telegram/remote_control.rs @@ -0,0 +1,286 @@ +//! Telegram remote-control slash commands (phase 1: `/status`, `/sessions`, `/new`). + +use super::session_store::{with_store, with_store_read, TelegramChatBinding}; +use crate::openhuman::channels::context::{ + clear_sender_history, conversation_history_key, ChannelRouteSelection, ChannelRuntimeContext, +}; +use crate::openhuman::channels::traits::ChannelMessage; +use crate::openhuman::memory::conversations::{self, ConversationThread, CreateConversationThread}; + +const LOG_PREFIX: &str = "[telegram-remote]"; + +pub(crate) const TELEGRAM_CMD_STATUS: &str = "/status"; +pub(crate) const TELEGRAM_CMD_SESSIONS: &str = "/sessions"; +pub(crate) const TELEGRAM_CMD_NEW: &str = "/new"; +pub(crate) const TELEGRAM_CMD_HELP: &str = "/help"; + +const SESSIONS_LIST_LIMIT: usize = 8; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum TelegramRemoteCommand { + Status, + Sessions, + New, + Help, +} + +pub(crate) fn parse_telegram_remote_command(content: &str) -> Option { + let trimmed = content.trim(); + if !trimmed.starts_with('/') { + return None; + } + let command_token = trimmed.split_whitespace().next()?; + let base = command_token + .split('@') + .next() + .unwrap_or(command_token) + .to_ascii_lowercase(); + match base.as_str() { + TELEGRAM_CMD_STATUS => Some(TelegramRemoteCommand::Status), + TELEGRAM_CMD_SESSIONS => Some(TelegramRemoteCommand::Sessions), + TELEGRAM_CMD_NEW => Some(TelegramRemoteCommand::New), + TELEGRAM_CMD_HELP => Some(TelegramRemoteCommand::Help), + _ => None, + } +} + +pub(crate) async fn build_remote_command_response( + ctx: &ChannelRuntimeContext, + msg: &ChannelMessage, + command: TelegramRemoteCommand, +) -> String { + tracing::debug!( + "{LOG_PREFIX} command={command:?} reply_target={} sender={}", + msg.reply_target, + msg.sender + ); + match command { + TelegramRemoteCommand::Status => build_status_response(ctx, msg).await, + TelegramRemoteCommand::Sessions => build_sessions_response(ctx, msg).await, + TelegramRemoteCommand::New => build_new_session_response(ctx, msg).await, + TelegramRemoteCommand::Help => build_help_response(), + } +} + +fn build_help_response() -> String { + [ + "OpenHuman Telegram remote control (phase 1):", + "", + &format!("• `{TELEGRAM_CMD_STATUS}` — active thread, model, and turn state"), + &format!("• `{TELEGRAM_CMD_SESSIONS}` — recent conversation threads"), + &format!("• `{TELEGRAM_CMD_NEW}` — start a fresh thread for this chat"), + &format!("• `{TELEGRAM_CMD_HELP}` — this message"), + "", + "Model routing: `/model`, `/models` (same as before).", + ] + .join("\n") +} + +fn route_for_sender(ctx: &ChannelRuntimeContext, sender_key: &str) -> ChannelRouteSelection { + ctx.route_overrides + .lock() + .unwrap_or_else(|e| e.into_inner()) + .get(sender_key) + .cloned() + .unwrap_or_else(|| ChannelRouteSelection { + provider: ctx.default_provider.as_str().to_string(), + model: ctx.model.as_str().to_string(), + }) +} + +async fn build_status_response(ctx: &ChannelRuntimeContext, msg: &ChannelMessage) -> String { + let sender_key = conversation_history_key(msg); + let route = route_for_sender(ctx, &sender_key); + let history_len = ctx + .conversation_histories + .lock() + .unwrap_or_else(|e| e.into_inner()) + .get(&sender_key) + .map(|h| h.len()) + .unwrap_or(0); + + let workspace = ctx.workspace_dir.clone(); + let reply_target = msg.reply_target.clone(); + // Use with_store_read (no disk write) and spawn_blocking to keep the async + // executor thread unblocked during mutex acquisition + file I/O. + let (binding, busy) = tokio::task::spawn_blocking(move || { + with_store_read(&workspace, |store| { + Ok(( + store.binding(&reply_target).cloned(), + store.is_busy(&reply_target), + )) + }) + }) + .await + .unwrap_or_else(|join_err| { + tracing::warn!("{LOG_PREFIX} status: join error reading session store: {join_err}"); + Ok((None, false)) + }) + .unwrap_or_else(|store_err| { + tracing::warn!("{LOG_PREFIX} status: session store error: {store_err}"); + (None, false) + }); + + let thread_line = match binding { + Some(TelegramChatBinding { + ref thread_id, + ref title, + .. + }) => { + // Use the stored title (captured at /new time) — O(1), no disk read. + let display_title = title + .as_deref() + .filter(|t| !t.trim().is_empty()) + .unwrap_or(thread_id); + format!("Thread: `{display_title}` (`{thread_id}`)") + } + None => "Thread: _(none — send `/new` to bind a thread)_".to_string(), + }; + + let turn_state = if busy { "in progress ⏳" } else { "idle" }; + + format!( + "**Status**\n\ + {thread_line}\n\ + Provider: `{provider}`\n\ + Model: `{model}`\n\ + In-memory turns: {history_len}\n\ + Turn: {turn_state}", + provider = route.provider, + model = route.model, + ) +} + +async fn build_sessions_response(ctx: &ChannelRuntimeContext, msg: &ChannelMessage) -> String { + let workspace = ctx.workspace_dir.clone(); + let reply_target = msg.reply_target.clone(); + // Read-only lookup — use with_store_read (no save) wrapped in spawn_blocking. + let active_thread_id = tokio::task::spawn_blocking(move || { + with_store_read(&workspace, |store| { + Ok(store.binding(&reply_target).map(|b| b.thread_id.clone())) + }) + }) + .await + .ok() + .and_then(|res| res.ok()) + .flatten(); + let workspace = ctx.workspace_dir.as_path(); + + let threads = match conversations::list_threads(workspace.to_path_buf()) { + Ok(list) => list, + Err(error) => { + tracing::warn!("{LOG_PREFIX} sessions: list_threads failed: {error}"); + return format!("Could not list sessions: {error}"); + } + }; + + if threads.is_empty() { + return "No conversation threads yet. Send `/new` to create one.".to_string(); + } + + let mut sorted = threads; + sorted.sort_by(|a, b| b.last_message_at.cmp(&a.last_message_at)); + + let mut lines = vec![ + "**Recent sessions**".to_string(), + format!("Showing up to {SESSIONS_LIST_LIMIT} threads:"), + String::new(), + ]; + + for thread in sorted.into_iter().take(SESSIONS_LIST_LIMIT) { + lines.push(format_session_line(&thread, active_thread_id.as_deref())); + } + + lines.join("\n") +} + +fn format_session_line(thread: &ConversationThread, active_id: Option<&str>) -> String { + let marker = if active_id == Some(thread.id.as_str()) { + "→ " + } else { + " " + }; + let title = if thread.title.trim().is_empty() { + thread.id.as_str() + } else { + thread.title.as_str() + }; + format!( + "{marker}`{title}` — {count} msgs (id: `{id}`)", + count = thread.message_count, + id = thread.id, + ) +} + +async fn build_new_session_response(ctx: &ChannelRuntimeContext, msg: &ChannelMessage) -> String { + let workspace = ctx.workspace_dir.as_path(); + let sender_key = conversation_history_key(msg); + let thread_id = format!("thread-{}", uuid::Uuid::new_v4()); + let now = chrono::Utc::now(); + let title = format!( + "Telegram {} {}", + now.format("%b %-d"), + now.format("%-I:%M %p") + ); + let created_at = now.to_rfc3339(); + + if let Err(error) = conversations::ensure_thread( + workspace.to_path_buf(), + CreateConversationThread { + id: thread_id.clone(), + title: title.clone(), + created_at, + parent_thread_id: None, + labels: Some(vec!["telegram".to_string(), "remote".to_string()]), + }, + ) { + tracing::warn!("{LOG_PREFIX} new: ensure_thread failed: {error}"); + return format!("Failed to create session: {error}"); + } + + clear_sender_history(ctx, &sender_key); + + let workspace_dir = ctx.workspace_dir.clone(); + let reply_target_owned = msg.reply_target.clone(); + let thread_id_owned = thread_id.clone(); + let sender_key_owned = sender_key.clone(); + let title_owned = title.clone(); + // Write-back path — use with_store (saves) wrapped in spawn_blocking. + let bind_result = tokio::task::spawn_blocking(move || { + with_store(&workspace_dir, |store| { + store.set_binding( + &reply_target_owned, + thread_id_owned, + sender_key_owned, + Some(title_owned), + ); + Ok(()) + }) + }) + .await + .unwrap_or_else(|e| Err(anyhow::anyhow!("join error: {e}"))); + + if let Err(error) = bind_result { + tracing::warn!("{LOG_PREFIX} new: persist binding failed: {error}"); + return format!( + "Created thread `{thread_id}` but failed to persist Telegram binding: {error}" + ); + } + + crate::openhuman::channels::providers::web::invalidate_thread_sessions(&thread_id).await; + + tracing::info!( + "{LOG_PREFIX} new session thread_id={thread_id} reply_target={} sender_key={sender_key}", + msg.reply_target + ); + + format!( + "Started new session **{title}**.\n\ + Thread id: `{thread_id}`\n\ + In-memory channel history cleared for this chat." + ) +} + +#[cfg(test)] +#[path = "remote_control_tests.rs"] +mod tests; diff --git a/src/openhuman/channels/providers/telegram/remote_control_tests.rs b/src/openhuman/channels/providers/telegram/remote_control_tests.rs new file mode 100644 index 000000000..08035bb1e --- /dev/null +++ b/src/openhuman/channels/providers/telegram/remote_control_tests.rs @@ -0,0 +1,41 @@ +use super::*; + +#[test] +fn parse_remote_commands() { + assert_eq!( + parse_telegram_remote_command("/status"), + Some(TelegramRemoteCommand::Status) + ); + assert_eq!( + parse_telegram_remote_command("/status@MyBot"), + Some(TelegramRemoteCommand::Status) + ); + assert_eq!( + parse_telegram_remote_command(" /sessions "), + Some(TelegramRemoteCommand::Sessions) + ); + assert_eq!( + parse_telegram_remote_command("/new"), + Some(TelegramRemoteCommand::New) + ); + assert_eq!( + parse_telegram_remote_command("/help"), + Some(TelegramRemoteCommand::Help) + ); + assert_eq!( + parse_telegram_remote_command(" /STATUS@OpenHumanBot now "), + Some(TelegramRemoteCommand::Status) + ); + // Case insensitivity for other variants + assert_eq!( + parse_telegram_remote_command("/Sessions"), + Some(TelegramRemoteCommand::Sessions) + ); + assert_eq!( + parse_telegram_remote_command("/NEW@Bot"), + Some(TelegramRemoteCommand::New) + ); + assert!(parse_telegram_remote_command("hello").is_none()); + assert!(parse_telegram_remote_command("/model").is_none()); + assert!(parse_telegram_remote_command("/unknown@OpenHumanBot").is_none()); +} diff --git a/src/openhuman/channels/providers/telegram/session_store.rs b/src/openhuman/channels/providers/telegram/session_store.rs new file mode 100644 index 000000000..a55866abc --- /dev/null +++ b/src/openhuman/channels/providers/telegram/session_store.rs @@ -0,0 +1,236 @@ +//! Workspace-backed Telegram chat → thread bindings for remote control. + +use serde::{Deserialize, Serialize}; +use std::collections::HashMap; +use std::path::{Path, PathBuf}; + +const STORE_FILE: &str = "state/telegram_remote_sessions.json"; +const LOG_PREFIX: &str = "[telegram-remote]"; + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub(crate) struct TelegramChatBinding { + pub(crate) thread_id: String, + pub(crate) sender_key: String, + pub(crate) updated_at: String, + /// Human-readable title captured at `/new` time so `/status` can display it + /// without listing all threads (O(1) instead of O(n) disk reads). + #[serde(default)] + pub(crate) title: Option, +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +struct TelegramSessionStoreFile { + bindings: HashMap, + #[serde(default)] + busy_reply_targets: HashMap, +} + +pub(crate) struct TelegramSessionStore { + file: TelegramSessionStoreFile, + path: PathBuf, +} + +impl TelegramSessionStore { + pub(crate) fn load(workspace_dir: &Path) -> anyhow::Result { + let path = workspace_dir.join(STORE_FILE); + let file = if path.exists() { + let raw = std::fs::read_to_string(&path)?; + serde_json::from_str(&raw).unwrap_or_else(|error| { + tracing::warn!( + "{LOG_PREFIX} corrupt session store at {}: {error}; resetting", + path.display() + ); + TelegramSessionStoreFile::default() + }) + } else { + TelegramSessionStoreFile::default() + }; + tracing::debug!( + "{LOG_PREFIX} loaded session store bindings={} busy={}", + file.bindings.len(), + file.busy_reply_targets.len() + ); + Ok(Self { file, path }) + } + + pub(crate) fn save(&self) -> anyhow::Result<()> { + if let Some(parent) = self.path.parent() { + std::fs::create_dir_all(parent)?; + } + let raw = serde_json::to_string_pretty(&self.file)?; + std::fs::write(&self.path, raw)?; + Ok(()) + } + + pub(crate) fn binding(&self, reply_target: &str) -> Option<&TelegramChatBinding> { + self.file.bindings.get(reply_target) + } + + pub(crate) fn set_binding( + &mut self, + reply_target: &str, + thread_id: String, + sender_key: String, + title: Option, + ) { + let updated_at = chrono::Utc::now().to_rfc3339(); + self.file.bindings.insert( + reply_target.to_string(), + TelegramChatBinding { + thread_id, + sender_key, + updated_at, + title, + }, + ); + } + + pub(crate) fn set_busy(&mut self, reply_target: &str, busy: bool) { + if busy { + self.file + .busy_reply_targets + .insert(reply_target.to_string(), true); + } else { + self.file.busy_reply_targets.remove(reply_target); + } + } + + pub(crate) fn is_busy(&self, reply_target: &str) -> bool { + self.file + .busy_reply_targets + .get(reply_target) + .copied() + .unwrap_or(false) + } +} + +static STORE: std::sync::OnceLock>> = + std::sync::OnceLock::new(); + +/// Read-write accessor: runs `f` against the cached store, then flushes to disk. +/// Use for operations that mutate state (e.g. `set_binding`, `set_busy`). +pub(crate) fn with_store(workspace_dir: &Path, f: F) -> anyhow::Result +where + F: FnOnce(&mut TelegramSessionStore) -> anyhow::Result, +{ + let lock = STORE.get_or_init(|| std::sync::Mutex::new(None)); + let mut guard = lock.lock().expect("telegram session store mutex poisoned"); + let expected_path = workspace_dir.join(STORE_FILE); + let needs_load = guard + .as_ref() + .map(|store| store.path != expected_path) + .unwrap_or(true); + if needs_load { + *guard = Some(TelegramSessionStore::load(workspace_dir)?); + } + let store = guard.as_mut().expect("store initialized"); + let result = f(store)?; + store.save()?; + Ok(result) +} + +/// Read-only accessor: runs `f` against the cached store but does **not** flush +/// to disk. Use for operations that only read state (e.g. `binding`, `is_busy`) +/// to avoid unnecessary serialization and disk I/O on every query. +pub(crate) fn with_store_read(workspace_dir: &Path, f: F) -> anyhow::Result +where + F: FnOnce(&TelegramSessionStore) -> anyhow::Result, +{ + let lock = STORE.get_or_init(|| std::sync::Mutex::new(None)); + let mut guard = lock.lock().expect("telegram session store mutex poisoned"); + let expected_path = workspace_dir.join(STORE_FILE); + let needs_load = guard + .as_ref() + .map(|store| store.path != expected_path) + .unwrap_or(true); + if needs_load { + *guard = Some(TelegramSessionStore::load(workspace_dir)?); + } + let store = guard.as_ref().expect("store initialized"); + f(store) +} + +#[cfg(test)] +mod tests { + use super::*; + use tempfile::tempdir; + + #[test] + fn round_trip_binding_and_busy_flag() { + let dir = tempdir().expect("tempdir"); + let mut store = TelegramSessionStore::load(dir.path()).expect("load"); + store.set_binding( + "12345", + "thread-abc".into(), + "telegram_alice_12345".into(), + Some("My Session".into()), + ); + store.set_busy("12345", true); + store.save().expect("save"); + + let reloaded = TelegramSessionStore::load(dir.path()).expect("reload"); + let binding = reloaded.binding("12345").expect("binding"); + assert_eq!(binding.thread_id, "thread-abc"); + assert_eq!(binding.title.as_deref(), Some("My Session")); + assert!(reloaded.is_busy("12345")); + } + + #[test] + fn corrupt_store_resets_and_clearing_busy_removes_flag() { + let dir = tempdir().expect("tempdir"); + let path = dir.path().join(STORE_FILE); + std::fs::create_dir_all(path.parent().expect("state dir")).expect("state dir"); + std::fs::write(&path, "{ not valid json").expect("write corrupt store"); + + let mut store = TelegramSessionStore::load(dir.path()).expect("load corrupt store"); + assert!(store.binding("12345").is_none()); + + store.set_busy("12345", true); + assert!(store.is_busy("12345")); + store.set_busy("12345", false); + assert!(!store.is_busy("12345")); + store.save().expect("save reset store"); + + let raw = std::fs::read_to_string(path).expect("read saved store"); + assert!(raw.contains("\"bindings\"")); + assert!(!raw.contains("12345")); + } + + /// Tests `with_store` workspace-change detection by using `TelegramSessionStore` + /// directly for cross-workspace assertions — avoids races with the process-global + /// `STORE` singleton when tests run in parallel. + #[test] + fn store_isolates_bindings_across_workspaces() { + let first = tempdir().expect("first tempdir"); + let second = tempdir().expect("second tempdir"); + + // Write binding into first workspace directly (no global singleton). + let mut store_a = TelegramSessionStore::load(first.path()).expect("load first"); + store_a.set_binding("chat-a", "thread-a".into(), "telegram_a".into(), None); + store_a.save().expect("save first"); + + // Write binding into second workspace directly. + let mut store_b = TelegramSessionStore::load(second.path()).expect("load second"); + assert!( + store_b.binding("chat-a").is_none(), + "second workspace must not see first workspace's binding" + ); + store_b.set_binding("chat-b", "thread-b".into(), "telegram_b".into(), None); + store_b.save().expect("save second"); + + let first_store = TelegramSessionStore::load(first.path()).expect("reload first"); + let second_store = TelegramSessionStore::load(second.path()).expect("reload second"); + assert_eq!( + first_store + .binding("chat-a") + .map(|binding| binding.thread_id.as_str()), + Some("thread-a") + ); + assert_eq!( + second_store + .binding("chat-b") + .map(|binding| binding.thread_id.as_str()), + Some("thread-b") + ); + } +} diff --git a/src/openhuman/channels/routes.rs b/src/openhuman/channels/routes.rs index 3daa0650b..e1d42eb2a 100644 --- a/src/openhuman/channels/routes.rs +++ b/src/openhuman/channels/routes.rs @@ -20,6 +20,7 @@ enum ChannelRuntimeCommand { SetProvider(String), ShowModel, SetModel(String), + TelegramRemote(super::providers::telegram::TelegramRemoteCommand), } #[derive(Debug, Clone, Default, Deserialize)] @@ -37,13 +38,25 @@ fn supports_runtime_model_switch(channel_name: &str) -> bool { matches!(channel_name, "telegram" | "discord") } +fn supports_telegram_remote_control(channel_name: &str) -> bool { + channel_name == "telegram" +} + fn parse_runtime_command(channel_name: &str, content: &str) -> Option { - if !supports_runtime_model_switch(channel_name) { + let trimmed = content.trim(); + if !trimmed.starts_with('/') { return None; } - let trimmed = content.trim(); - if !trimmed.starts_with('/') { + if supports_telegram_remote_control(channel_name) { + if let Some(remote) = + super::providers::telegram::remote_control::parse_telegram_remote_command(content) + { + return Some(ChannelRuntimeCommand::TelegramRemote(remote)); + } + } + + if !supports_runtime_model_switch(channel_name) { return None; } @@ -269,6 +282,12 @@ pub(crate) async fn handle_runtime_command_if_needed( let mut current = get_route_selection(ctx, &sender_key); let response = match command { + ChannelRuntimeCommand::TelegramRemote(remote) => { + super::providers::telegram::remote_control::build_remote_command_response( + ctx, msg, remote, + ) + .await + } ChannelRuntimeCommand::ShowProviders => build_providers_help_response(¤t), ChannelRuntimeCommand::SetProvider(raw_provider) => { match resolve_provider_alias(&raw_provider) { diff --git a/src/openhuman/channels/routes_tests.rs b/src/openhuman/channels/routes_tests.rs index 06124b51b..78630a764 100644 --- a/src/openhuman/channels/routes_tests.rs +++ b/src/openhuman/channels/routes_tests.rs @@ -1,7 +1,9 @@ use super::*; +use crate::core::event_bus::{DomainEvent, EventHandler}; use crate::openhuman::channels::context::{ ChannelRuntimeContext, ProviderCacheMap, RouteSelectionMap, }; +use crate::openhuman::channels::telegram::{TelegramRemoteCommand, TelegramRemoteSubscriber}; use crate::openhuman::channels::traits::ChannelMessage; use crate::openhuman::inference::provider::{ChatMessage, Provider}; use crate::openhuman::memory::{Memory, MemoryCategory, MemoryEntry}; @@ -177,7 +179,20 @@ fn runtime_command_parsing_and_provider_support_are_channel_scoped() { parse_runtime_command("telegram", "/model"), Some(ChannelRuntimeCommand::ShowModel) ); + assert_eq!( + parse_runtime_command("telegram", "/status@OpenHumanBot"), + Some(ChannelRuntimeCommand::TelegramRemote( + TelegramRemoteCommand::Status + )) + ); + assert_eq!( + parse_runtime_command("telegram", "/help"), + Some(ChannelRuntimeCommand::TelegramRemote( + TelegramRemoteCommand::Help + )) + ); assert_eq!(parse_runtime_command("slack", "/models"), None); + assert_eq!(parse_runtime_command("discord", "/status"), None); assert_eq!(parse_runtime_command("telegram", "hello"), None); } @@ -358,3 +373,200 @@ async fn handle_runtime_command_set_model_clears_sender_history_and_persists_rou assert_eq!(sent.len(), 1); assert!(sent[0].content.contains("Model switched to `gpt-5-mini`")); } + +#[tokio::test] +async fn handle_runtime_command_telegram_status_replies_without_agent() { + let ctx = runtime_context(PathBuf::from("/tmp")); + let channel_impl = Arc::new(RecordingChannel::default()); + let channel: Arc = channel_impl.clone(); + let msg = ChannelMessage { + id: "1".into(), + sender: "alice".into(), + reply_target: "chat-remote".into(), + content: "/status".into(), + channel: "telegram".into(), + timestamp: 0, + thread_ts: Some("42".into()), + }; + + let handled = handle_runtime_command_if_needed(&ctx, &msg, Some(&channel)).await; + assert!(handled); + + let sent = channel_impl.sent.lock().unwrap(); + assert_eq!(sent.len(), 1); + assert!(sent[0].content.contains("**Status**")); + assert!(sent[0].content.contains("Provider:")); +} + +#[tokio::test] +async fn handle_runtime_command_without_target_channel_still_consumes_command() { + let ctx = runtime_context(PathBuf::from("/tmp")); + let msg = ChannelMessage { + id: "1".into(), + sender: "alice".into(), + reply_target: "chat-remote".into(), + content: "/help".into(), + channel: "telegram".into(), + timestamp: 0, + thread_ts: None, + }; + + let handled = handle_runtime_command_if_needed(&ctx, &msg, None).await; + assert!(handled); +} + +#[tokio::test] +async fn handle_runtime_command_telegram_help_replies_with_remote_command_list() { + let ctx = runtime_context(PathBuf::from("/tmp")); + let channel_impl = Arc::new(RecordingChannel::default()); + let channel: Arc = channel_impl.clone(); + let msg = ChannelMessage { + id: "1".into(), + sender: "alice".into(), + reply_target: "chat-remote".into(), + content: "/help".into(), + channel: "telegram".into(), + timestamp: 0, + thread_ts: Some("42".into()), + }; + + let handled = handle_runtime_command_if_needed(&ctx, &msg, Some(&channel)).await; + assert!(handled); + + let sent = channel_impl.sent.lock().unwrap(); + assert_eq!(sent.len(), 1); + assert!(sent[0] + .content + .contains("OpenHuman Telegram remote control (phase 1):")); + assert!(sent[0].content.contains("`/status`")); + assert!(sent[0].content.contains("`/sessions`")); + assert!(sent[0].content.contains("`/new`")); + assert!(sent[0] + .content + .contains("Model routing: `/model`, `/models`")); +} + +#[tokio::test] +async fn handle_runtime_command_telegram_sessions_reports_empty_store() { + let tempdir = tempfile::tempdir().unwrap(); + let ctx = runtime_context(tempdir.path().to_path_buf()); + let channel_impl = Arc::new(RecordingChannel::default()); + let channel: Arc = channel_impl.clone(); + let msg = ChannelMessage { + id: "1".into(), + sender: "alice".into(), + reply_target: "chat-remote".into(), + content: "/sessions".into(), + channel: "telegram".into(), + timestamp: 0, + thread_ts: Some("42".into()), + }; + + let handled = handle_runtime_command_if_needed(&ctx, &msg, Some(&channel)).await; + assert!(handled); + + let sent = channel_impl.sent.lock().unwrap(); + assert_eq!(sent.len(), 1); + assert!(sent[0] + .content + .contains("No conversation threads yet. Send `/new` to create one.")); + assert_eq!(sent[0].thread_ts.as_deref(), Some("42")); +} + +#[tokio::test] +async fn handle_runtime_command_telegram_new_status_and_sessions_round_trip() { + let tempdir = tempfile::tempdir().unwrap(); + let ctx = runtime_context(tempdir.path().to_path_buf()); + let channel_impl = Arc::new(RecordingChannel::default()); + let channel: Arc = channel_impl.clone(); + let sender_key = "telegram_alice_chat-remote"; + + ctx.conversation_histories.lock().unwrap().insert( + sender_key.to_string(), + vec![ChatMessage::user("old history")], + ); + + let new_msg = ChannelMessage { + id: "1".into(), + sender: "alice".into(), + reply_target: "chat-remote".into(), + content: "/new".into(), + channel: "telegram".into(), + timestamp: 0, + thread_ts: Some("42".into()), + }; + assert!(handle_runtime_command_if_needed(&ctx, &new_msg, Some(&channel)).await); + assert!(ctx + .conversation_histories + .lock() + .unwrap() + .get(sender_key) + .is_none()); + + ctx.conversation_histories + .lock() + .unwrap() + .insert(sender_key.to_string(), vec![ChatMessage::user("after new")]); + set_route_selection( + &ctx, + sender_key, + ChannelRouteSelection { + provider: "anthropic".into(), + model: "claude-3".into(), + }, + ); + + let subscriber = TelegramRemoteSubscriber::new(tempdir.path().to_path_buf()); + subscriber + .handle(&DomainEvent::ChannelMessageReceived { + channel: "telegram".into(), + message_id: "2".into(), + sender: "alice".into(), + reply_target: "chat-remote".into(), + content: "work".into(), + thread_ts: Some("42".into()), + }) + .await; + + let status_msg = ChannelMessage { + id: "3".into(), + sender: "alice".into(), + reply_target: "chat-remote".into(), + content: "/status".into(), + channel: "telegram".into(), + timestamp: 0, + thread_ts: Some("42".into()), + }; + assert!(handle_runtime_command_if_needed(&ctx, &status_msg, Some(&channel)).await); + + let sessions_msg = ChannelMessage { + id: "4".into(), + sender: "alice".into(), + reply_target: "chat-remote".into(), + content: "/sessions".into(), + channel: "telegram".into(), + timestamp: 0, + thread_ts: Some("42".into()), + }; + assert!(handle_runtime_command_if_needed(&ctx, &sessions_msg, Some(&channel)).await); + + let sent = channel_impl.sent.lock().unwrap(); + assert_eq!(sent.len(), 3); + assert!(sent[0].content.contains("Started new session")); + assert!(sent[0] + .content + .contains("In-memory channel history cleared for this chat.")); + + assert!(sent[1].content.contains("**Status**")); + assert!(sent[1].content.contains("Thread: `Telegram")); + assert!(sent[1].content.contains("Provider: `anthropic`")); + assert!(sent[1].content.contains("Model: `claude-3`")); + assert!(sent[1].content.contains("In-memory turns: 1")); + assert!(sent[1].content.contains("Turn: in progress")); + + assert!(sent[2].content.contains("**Recent sessions**")); + assert!(sent[2].content.contains("→ `Telegram")); + assert!(sent + .iter() + .all(|message| message.thread_ts.as_deref() == Some("42"))); +} diff --git a/src/openhuman/channels/runtime/startup.rs b/src/openhuman/channels/runtime/startup.rs index 975dfc173..deb05f537 100644 --- a/src/openhuman/channels/runtime/startup.rs +++ b/src/openhuman/channels/runtime/startup.rs @@ -574,6 +574,17 @@ pub async fn start_channels(config: Config) -> Result<()> { config.channels_config.active_channel.clone(), ), )); + let _telegram_remote_handle = if channels_by_name.contains_key("telegram") { + let handle = bus.subscribe(Arc::new( + crate::openhuman::channels::providers::telegram::TelegramRemoteSubscriber::new( + config.workspace_dir.clone(), + ), + )); + tracing::debug!("[telegram-remote] registered TelegramRemoteSubscriber"); + Some(handle) + } else { + None + }; // Register the tree summarizer event subscriber for observability logging. let _tree_summarizer_handle = bus.subscribe(Arc::new( crate::openhuman::tree_summarizer::bus::TreeSummarizerEventSubscriber::new(),