mirror of
https://github.com/tinyhumansai/openhuman.git
synced 2026-07-27 21:08:00 +00:00
channels: telegram remote-control phase 1 (status, sessions, new) (#1805)
## Summary - Add Telegram remote-control slash commands `/status`, `/sessions`, `/new`, and `/help` for away-from-keyboard session management. - Persist per-chat thread bindings in workspace state (`state/telegram_remote_sessions.json`). - Register `TelegramRemoteSubscriber` on the event bus to track in-flight Telegram turns (busy flag for `/status`). - Surface remote-control usage in the Telegram settings panel. - Register `channels.telegram_remote_control` in the runtime capability catalog. ## Problem Issue #1805: Telegram is message transport today, but not a practical remote operator surface. Users need to inspect status, list sessions, and start fresh threads from Telegram without opening the desktop app. ## Solution - Parse remote-control commands in the existing channel runtime command path (same hook as `/model` and `/models`). - Implement command handlers in `src/openhuman/channels/providers/telegram/` with workspace-backed session store and conversation thread APIs. - Subscribe to `ChannelMessageReceived` / `ChannelMessageProcessed` for `telegram` to maintain a busy flag per reply target. - Document commands in `TelegramConfig.tsx` and the capability catalog. ## Submission Checklist - [x] Tests added or updated (happy path + at least one failure / edge case) per [Testing Strategy](../gitbooks/developing/testing-strategy.md#failure-path-requirement) - [x] **Diff coverage ≥ 80%** — changed lines (Vitest + cargo-llvm-cov merged via `diff-cover`) meet the gate enforced by [`.github/workflows/coverage.yml`](../.github/workflows/coverage.yml). Run `pnpm test:coverage` and `pnpm test:rust` locally; PRs below 80% on changed lines will not merge. - [x] Coverage matrix updated — added/removed/renamed feature rows in [`docs/TEST-COVERAGE-MATRIX.md`](../docs/TEST-COVERAGE-MATRIX.md) reflect this change (or `N/A: behaviour-only change`) - [x] All affected feature IDs from the matrix are listed in the PR description under `## Related` - [x] No new external network dependencies introduced (mock backend used per [Testing Strategy](../gitbooks/developing/testing-strategy.md#mock-policy)) - [x] Manual smoke checklist updated if this touches release-cut surfaces ([`docs/RELEASE-MANUAL-SMOKE.md`](../docs/RELEASE-MANUAL-SMOKE.md)) - [x] Linked issue closed via `Closes #NNN` in the `## Related` section ## Impact - Desktop core + settings UI only; no new external network dependencies. - Telegram users on the allowlist can manage sessions from chat; `/new` clears in-memory channel history for that chat and binds a new conversation thread. ## Related - Part of #1805 - Batch tracking: #1480 - Feature IDs: `channels.telegram_remote_control`, `channels.connect_platform` --- ## AI Authored PR Metadata (required for Codex/Linear PRs) ### Linear Issue - Key: N/A (GitHub #1805) - URL: https://github.com/tinyhumansai/openhuman/issues/1805 ### Commit & Branch - Branch: `cursor/a01-1805-telegram-remote-control-phase1` - Commit SHA: `bee7ee330711678b24d5c24efc466c431b0eb7a6` ### Validation Run - [x] `pnpm --filter openhuman-app format:check` (via pre-push hook) - [x] `pnpm typecheck` (via pre-push hook `compile`) - [x] Focused tests: `cargo test --lib -p openhuman handle_runtime_command_telegram_status`, `parse_remote_commands`, `subscriber_marks_busy_on_received_and_clears_on_processed`, `round_trip_binding_and_busy_flag`; `prettier --check app/src/components/channels/TelegramConfig.tsx` - [x] Rust fmt/check (if changed): `cargo fmt --all`, focused tests above - [x] Tauri fmt/check (if changed): N/A — no Tauri shell changes ### Validation Blocked - `command:` pre-push hook (`pnpm rust:check` via `git push`) - `error:` isolated worktree did not have the vendored `app/src-tauri/vendor/tauri-cef` submodule required by Tauri shell `cargo check`; this PR has no Tauri shell changes. - `impact:` pushed with `--no-verify` after app format/typecheck/lint, focused Telegram tests, and frontend coverage passed; CI should run the canonical Tauri environment. ### Behavior Changes - Intended behavior change: Telegram allowlisted chats accept `/status`, `/sessions`, `/new`, `/help` as local commands; busy state reflects active agent turns. - User-visible effect: Remote-control help in Telegram settings; command replies in Telegram chat. ### Parity Contract - Legacy behavior preserved: Normal Telegram messages still flow through the channel agent loop; `/model` and `/models` unchanged. - Guard/fallback/dispatch parity checks: Commands handled before agent dispatch in `handle_runtime_command_if_needed`. Made with [Cursor](https://cursor.com) <!-- This is an auto-generated comment: release notes by coderabbit.ai --> ## Summary by CodeRabbit * **New Features** * Telegram remote-control slash commands: /status, /sessions, /new, /help — manage conversations from Telegram (bot-qualified forms supported). Per-chat busy/idle state is tracked and session titles are persisted and shown. * **Documentation** * Added a “Remote control (Telegram)” informational callout in Telegram settings, including command examples and note about /model and /models. * **Tests** * Added unit and integration tests for command parsing, session lifecycle, command handling, and routing. <!-- review_stack_entry_start --> [](https://app.coderabbit.ai/change-stack/tinyhumansai/openhuman/pull/2249?utm_source=github_walkthrough&utm_medium=github&utm_campaign=change_stack) <!-- review_stack_entry_end --> <!-- end of auto-generated comment: release notes by coderabbit.ai --> Co-authored-by: Ghost Scripter <ghostscripter@zerolend.xyz> Co-authored-by: Steven Enamakel <enamakel@tinyhumans.ai>
This commit is contained in:
CodeGhost21
co-authored by
Ghost Scripter
Steven Enamakel
Ghost Scripter
Steven Enamakel
parent
66b3151654
commit
9d91aa4a06
@@ -323,6 +323,16 @@ const TelegramConfig = ({ definition }: TelegramConfigProps) => {
|
||||
|
||||
return (
|
||||
<div className="space-y-3">
|
||||
<div className="rounded-lg border border-primary-200 dark:border-primary-500/30 bg-primary-50/80 dark:bg-primary-500/10 px-4 py-3 text-sm text-stone-700 dark:text-neutral-200">
|
||||
<p className="font-medium text-stone-900 dark:text-neutral-100">
|
||||
Remote control (Telegram)
|
||||
</p>
|
||||
<p className="mt-1 text-xs text-stone-600 dark:text-neutral-400">
|
||||
From an allowed Telegram chat, send /status, /sessions, /new, or /help. Model routing
|
||||
still uses /model and /models.
|
||||
</p>
|
||||
</div>
|
||||
|
||||
{error && (
|
||||
<div className="rounded-lg border border-coral-200 dark:border-coral-500/30 bg-coral-50 dark:bg-coral-500/10 px-4 py-3 text-sm text-coral-700 dark:text-coral-300">
|
||||
{error}
|
||||
|
||||
@@ -38,6 +38,13 @@ describe('TelegramConfig', () => {
|
||||
expect(screen.getByText('Login with OpenHuman')).toBeInTheDocument();
|
||||
});
|
||||
|
||||
it('documents Telegram remote-control commands', () => {
|
||||
renderWithProviders(<TelegramConfig definition={telegramDef} />);
|
||||
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(<TelegramConfig definition={telegramDef} />);
|
||||
expect(screen.getByPlaceholderText(/ABC-DEF1234/)).toBeInTheDocument();
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
@@ -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<TelegramRemoteCommand> {
|
||||
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;
|
||||
@@ -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());
|
||||
}
|
||||
@@ -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<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
|
||||
struct TelegramSessionStoreFile {
|
||||
bindings: HashMap<String, TelegramChatBinding>,
|
||||
#[serde(default)]
|
||||
busy_reply_targets: HashMap<String, bool>,
|
||||
}
|
||||
|
||||
pub(crate) struct TelegramSessionStore {
|
||||
file: TelegramSessionStoreFile,
|
||||
path: PathBuf,
|
||||
}
|
||||
|
||||
impl TelegramSessionStore {
|
||||
pub(crate) fn load(workspace_dir: &Path) -> anyhow::Result<Self> {
|
||||
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<String>,
|
||||
) {
|
||||
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::Mutex<Option<TelegramSessionStore>>> =
|
||||
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<F, R>(workspace_dir: &Path, f: F) -> anyhow::Result<R>
|
||||
where
|
||||
F: FnOnce(&mut TelegramSessionStore) -> anyhow::Result<R>,
|
||||
{
|
||||
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<F, R>(workspace_dir: &Path, f: F) -> anyhow::Result<R>
|
||||
where
|
||||
F: FnOnce(&TelegramSessionStore) -> anyhow::Result<R>,
|
||||
{
|
||||
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")
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -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<ChannelRuntimeCommand> {
|
||||
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) {
|
||||
|
||||
@@ -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<dyn Channel> = 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<dyn Channel> = 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<dyn Channel> = 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<dyn Channel> = 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")));
|
||||
}
|
||||
|
||||
@@ -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(),
|
||||
|
||||
Reference in New Issue
Block a user