refactor(telegram): extract remote-control helpers

Cargo.toml

@@ -1,6 +1,6 @@
 [workspace]
-members = ["crates/tinychannels-bus", "crates/tinychannels-module"]
-default-members = [".", "crates/tinychannels-bus", "crates/tinychannels-module"]
+members = ["crates/tinychannels-bus", "crates/tinychannels-runtime", "crates/tinychannels-module"]
+default-members = [".", "crates/tinychannels-bus", "crates/tinychannels-runtime", "crates/tinychannels-module"]
 # The bus is a submodule with its own workspace; it is a path dependency of the
 # module crate, not a member here.
 exclude = ["vendor/tinybus"]
@@ -69,6 +69,7 @@ whatsapp-web = [
 # provider stack below is the implementation of that contract; a host that
 # only names channel types depends on the bus crate alone.
 tinychannels-bus = { version = "0.1.2", path = "crates/tinychannels-bus" }
+tinychannels-runtime = { version = "0.1.2", path = "crates/tinychannels-runtime" }
 anyhow = "1"
 async-trait = "0.1"
 base64 = "0.22"

crates/tinychannels-bus/src/channel/mod.rs

@@ -38,7 +38,7 @@ pub use receipt::{
 };
 pub use session::{
     LegacySessionKeys, SessionKeyPolicy, build_session_key, build_session_key_for_inbound_envelope,
-    conversation_history_key_candidates,
+    conversation_history_key_candidates, derive_inbound_client_id, derive_inbound_thread_id,
 };
 pub use types::{
     ChannelDescriptor, ChannelRef, ConversationKind, ConversationRef, SecretRef, SenderRef,

crates/tinychannels-bus/src/channel/session.rs

@@ -118,9 +118,50 @@ pub fn conversation_history_key_candidates(msg: &ChannelMessage) -> LegacySessio
     }
 }
 
+/// Derive a stable host-local thread key from inbound channel facts.
+pub fn derive_inbound_thread_id(
+    channel: &str,
+    sender: Option<&str>,
+    reply_target: Option<&str>,
+    thread_ts: Option<&str>,
+) -> String {
+    let mut key = format!("channel:{channel}");
+    if let Some(sender) = sender.and_then(nonempty) {
+        key.push('/');
+        key.push_str(sender);
+    }
+    if let Some(reply_target) = reply_target.and_then(nonempty) {
+        key.push('/');
+        key.push_str(reply_target);
+    }
+    let provider = channel.split(':').next().unwrap_or("");
+    if !matches!(provider, "telegram" | "tg") {
+        if let Some(thread_ts) = thread_ts.and_then(nonempty) {
+            key.push_str("#thread:");
+            key.push_str(thread_ts);
+        }
+    }
+    key
+}
+
+/// Derive a stable client identifier for an inbound channel sender.
+pub fn derive_inbound_client_id(channel: &str, sender: Option<&str>) -> String {
+    let channel = channel.trim();
+    match sender.map(str::trim).filter(|sender| !sender.is_empty()) {
+        Some(sender) if !channel.is_empty() => format!("inbound:{channel}:{sender}"),
+        Some(sender) => format!("inbound:{sender}"),
+        None => "inbound".to_string(),
+    }
+}
+
 fn normalize_namespace(namespace: &str) -> &str {
     match namespace.trim() {
         "" | "default" => "main",
         value => value,
     }
 }
+
+fn nonempty(value: &str) -> Option<&str> {
+    let value = value.trim();
+    (!value.is_empty()).then_some(value)
+}

crates/tinychannels-bus/src/lib.rs

@@ -62,9 +62,10 @@ pub mod version;
 
 pub use channel::{
     ChannelInboundEnvelope, ChannelOutboundIntent, DeliveryDurability, OutboundPayload,
-    build_session_key_for_inbound_envelope, inbound_envelope_from_legacy_message,
-    legacy_message_from_inbound_envelope, legacy_message_value_from_outbound_intent,
-    outbound_intent_from_legacy_message, outbound_intent_from_send_message,
+    build_session_key_for_inbound_envelope, derive_inbound_client_id, derive_inbound_thread_id,
+    inbound_envelope_from_legacy_message, legacy_message_from_inbound_envelope,
+    legacy_message_value_from_outbound_intent, outbound_intent_from_legacy_message,
+    outbound_intent_from_send_message,
 };
 pub use config::ChannelsConfig;
 pub use controllers::{ChannelAuthMode, ChannelDefinition};

crates/tinychannels-runtime/Cargo.toml

@@ -0,0 +1,20 @@
+[package]
+name = "tinychannels-runtime"
+version.workspace = true
+edition = "2024"
+license = "GPL-3.0-only"
+description = "Reusable listener supervision and runtime helpers for TinyChannels."
+repository = "https://github.com/tinyhumansai/tinychannels"
+publish = false
+
+[dependencies]
+anyhow = "1"
+rand = "0.10"
+tinychannels-bus = { version = "0.1.2", path = "../tinychannels-bus" }
+tokio = { version = "1", default-features = false, features = ["rt", "sync", "time"] }
+tokio-util = { version = "0.7", default-features = false, features = ["rt"] }
+tracing = "0.1"
+
+[dev-dependencies]
+async-trait = "0.1"
+tokio = { version = "1", features = ["macros", "rt", "sync", "test-util", "time"] }

crates/tinychannels-runtime/src/lib.rs

@@ -0,0 +1,202 @@
+//! Runtime mechanics shared by TinyChannels hosts.
+//!
+//! This crate deliberately owns no provider, persistence, event bus, or host
+//! policy. Hosts observe listener lifecycle through [`ListenerObserver`].
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use rand::RngExt as _;
+use tinychannels_bus::{Channel, ChannelMessage};
+use tokio_util::sync::CancellationToken;
+
+/// Maximum reconnect jitter added to a listener retry.
+pub const MAX_JITTER_MS: u64 = 1_000;
+
+/// Host callback for listener lifecycle facts.
+pub trait ListenerObserver: Send + Sync {
+    /// A listener is about to enter its receive loop.
+    fn connected(&self, _channel: &str) {}
+    /// A listener exited and will be retried.
+    fn disconnected(&self, _channel: &str, _reason: &str, _failed: bool) {}
+    /// A retry has been scheduled after a listener exit.
+    fn restarted(&self, _channel: &str) {}
+}
+
+/// A listener observer with no host side effects.
+#[derive(Debug, Default)]
+pub struct NoopListenerObserver;
+impl ListenerObserver for NoopListenerObserver {}
+
+/// Compute the bounded listener queue capacity for a provider count.
+pub fn compute_max_in_flight_messages(channel_count: usize) -> usize {
+    channel_count.saturating_mul(4).clamp(8, 64)
+}
+
+/// Deterministically choose a broadly-supported acknowledgement reaction.
+pub fn select_acknowledgment_reaction(content: &str) -> &'static str {
+    let lower = content.to_lowercase();
+    let variant = content
+        .len()
+        .wrapping_add(content.chars().next().map_or(0, |ch| ch as usize))
+        & 1;
+    let contains = |words: &[&str]| words.iter().any(|word| lower.contains(word));
+    let starts = |words: &[&str]| words.iter().any(|word| lower.starts_with(word));
+    let options: &[&str] = if contains(&["thank", "thx", "appreciate", "grateful", "cheers"]) {
+        &["❤️", "🙏"]
+    } else if contains(&[
+        "amazing",
+        "awesome",
+        "incredible",
+        "love it",
+        "congrat",
+        "!!",
+    ]) {
+        &["🔥", "🎉"]
+    } else if contains(&[
+        "price", "btc", "eth", "crypto", "trade", "pump", "dump", "market", "token", "wallet",
+        "defi", "nft", "sol", "bnb",
+    ]) {
+        &["💯", "⚡"]
+    } else if contains(&[
+        "code",
+        "function",
+        "api",
+        "deploy",
+        "build",
+        "debug",
+        "script",
+        "git",
+        "rust",
+        "python",
+        "js",
+        "typescript",
+    ]) {
+        &["👨‍💻", "🤓"]
+    } else if starts(&[
+        "hi",
+        "hello",
+        "hey",
+        "sup",
+        "good morning",
+        "good evening",
+        "good afternoon",
+    ]) || lower == "yo"
+        || lower.starts_with("yo ")
+    {
+        &["🤗", "😁"]
+    } else if lower.contains('?')
+        || starts(&[
+            "how",
+            "what",
+            "why",
+            "when",
+            "where",
+            "who",
+            "can you",
+            "could you",
+            "would you",
+            "is ",
+            "are ",
+            "do you",
+            "does",
+        ])
+    {
+        &["🤔", "✍️"]
+    } else {
+        &["👀", "✍️"]
+    };
+    options[variant % options.len()]
+}
+
+/// Spawn a reconnecting listener. Host-specific observability is delivered to
+/// `observer`; the retry policy remains identical for every host.
+pub fn spawn_supervised_listener(
+    channel: Arc<dyn Channel>,
+    tx: tokio::sync::mpsc::Sender<ChannelMessage>,
+    initial_backoff_secs: u64,
+    max_backoff_secs: u64,
+    observer: Arc<dyn ListenerObserver>,
+) -> tokio::task::JoinHandle<()> {
+    tokio::spawn(async move {
+        let name = channel.name().to_owned();
+        let mut backoff = initial_backoff_secs.max(1);
+        let max_backoff = max_backoff_secs.max(backoff);
+        loop {
+            observer.connected(&name);
+            let result = channel.listen(tx.clone()).await;
+            if tx.is_closed() {
+                break;
+            }
+            match result {
+                Ok(()) => observer.disconnected(&name, "exited unexpectedly", false),
+                Err(error) => observer.disconnected(&name, &error.to_string(), true),
+            }
+            observer.restarted(&name);
+            tokio::time::sleep(
+                Duration::from_secs(backoff) + Duration::from_millis(jitter_millis(backoff)),
+            )
+            .await;
+            backoff = backoff.saturating_mul(2).min(max_backoff);
+        }
+    })
+}
+
+/// Sample full reconnect jitter, bounded to avoid dwarfing the base retry.
+pub fn jitter_millis(backoff_secs: u64) -> u64 {
+    let window = backoff_secs.saturating_mul(1_000).min(MAX_JITTER_MS);
+    (window != 0)
+        .then(|| rand::rng().random_range(0..window))
+        .unwrap_or(0)
+}
+
+/// Log a failed worker join without imposing host-specific error reporting.
+pub fn log_worker_join_result(result: Result<(), tokio::task::JoinError>) {
+    if let Err(error) = result {
+        tracing::error!("Channel message worker crashed: {error}");
+    }
+}
+
+/// Maintain a typing indicator until `cancellation_token` is cancelled.
+pub fn spawn_scoped_typing_task(
+    channel: Arc<dyn Channel>,
+    recipient: String,
+    cancellation_token: CancellationToken,
+    refresh_interval: Duration,
+) -> tokio::task::JoinHandle<()> {
+    tokio::spawn(async move {
+        loop {
+            tokio::select! {
+                () = cancellation_token.cancelled() => break,
+                _ = tokio::time::sleep(refresh_interval) => {
+                    if let Err(error) = channel.start_typing(&recipient).await {
+                        tracing::debug!(channel = channel.name(), "typing start failed: {error}");
+                    }
+                }
+            }
+        }
+        if let Err(error) = channel.stop_typing(&recipient).await {
+            tracing::debug!(channel = channel.name(), "typing stop failed: {error}");
+        }
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn acknowledgement_selection_is_stable_and_contextual() {
+        assert!(matches!(
+            select_acknowledgment_reaction("thanks"),
+            "❤️" | "🙏"
+        ));
+        assert_eq!(
+            select_acknowledgment_reaction("thanks"),
+            select_acknowledgment_reaction("thanks")
+        );
+        assert!(jitter_millis(1) < MAX_JITTER_MS);
+        assert_eq!(jitter_millis(0), 0);
+        assert_eq!(compute_max_in_flight_messages(100), 64);
+    }
+}

src/providers/telegram/approval.rs

@@ -0,0 +1,24 @@
+//! Telegram approval-prompt vocabulary shared by hosts.
+
+/// Identifier used for Telegram-originated approval contexts.
+pub const TELEGRAM_APPROVAL_CLIENT_ID: &str = "telegram";
+
+/// Render an approval request as a Telegram message body.
+pub fn format_approval_prompt(tool_name: &str, action_summary: &str) -> String {
+    format!(
+        "🔐 Approval needed\nTool: `{tool_name}`\nAction: {action_summary}\n\nReply `yes` to approve or `no` to deny."
+    )
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn approval_prompt_includes_action_and_reply_instructions() {
+        let body = format_approval_prompt("git_operations", "git commit -m fix");
+        assert!(body.contains("git_operations"));
+        assert!(body.contains("git commit"));
+        assert!(body.contains("yes") && body.contains("no"));
+    }
+}