fix(channels): pair Telegram self-bot-token on /start and keep channel tool-calling alive (#4414)

Co-authored-by: M3gA-Mind <megamind@mahadao.com>

src/openhuman/app_state/ops.rs

@@ -67,6 +67,11 @@ static SNAPSHOT_REQ_COUNTER: AtomicU64 = AtomicU64::new(0);
 struct CachedRuntimeSnapshot {
     snapshot: RuntimeSnapshot,
     fetched_at: Instant,
+    /// Config identity (`workspace_dir`) the snapshot was built for. The cache
+    /// holds one entry process-wide, so a snapshot built for one config must
+    /// never be served to another — otherwise a different user/workspace (or an
+    /// E2E test with an injected service mock) reads a stale, foreign runtime.
+    config_key: PathBuf,
 }
 
 #[derive(Debug, Clone)]
@@ -756,9 +761,14 @@ pub fn peek_cached_current_user_identity() -> Option<crate::openhuman::agent::pr
 /// Return the cached runtime snapshot when it is still within
 /// `RUNTIME_SNAPSHOT_TTL`, else `None`. Kept as a small helper so both the
 /// fast-path read and the post-lock double-check share identical freshness logic.
-fn fresh_cached_runtime_snapshot(req_id: u64) -> Option<RuntimeSnapshot> {
+fn fresh_cached_runtime_snapshot(config: &Config, req_id: u64) -> Option<RuntimeSnapshot> {
     let cache = RUNTIME_SNAPSHOT_CACHE.lock();
     let entry = cache.as_ref()?;
+    // A snapshot built for a different config identity is a miss: rebuild against
+    // this config rather than serve another workspace's runtime.
+    if entry.config_key != config.workspace_dir {
+        return None;
+    }
     let age = entry.fetched_at.elapsed();
     if age < RUNTIME_SNAPSHOT_TTL {
         debug!(
@@ -774,7 +784,7 @@ fn fresh_cached_runtime_snapshot(req_id: u64) -> Option<RuntimeSnapshot> {
 async fn build_runtime_snapshot(config: &Config, req_id: u64) -> RuntimeSnapshot {
     // Fast path: a fresh cached snapshot serves every poller without touching the
     // sub-op fan-out.
-    if let Some(snapshot) = fresh_cached_runtime_snapshot(req_id) {
+    if let Some(snapshot) = fresh_cached_runtime_snapshot(config, req_id) {
         return snapshot;
     }
 
@@ -783,7 +793,7 @@ async fn build_runtime_snapshot(config: &Config, req_id: u64) -> RuntimeSnapshot
     // double-check) and return it instead of launching a duplicate build —
     // collapsing an N-way stampede into one build per TTL window.
     let _rebuild_guard = RUNTIME_SNAPSHOT_REBUILD.lock().await;
-    if let Some(snapshot) = fresh_cached_runtime_snapshot(req_id) {
+    if let Some(snapshot) = fresh_cached_runtime_snapshot(config, req_id) {
         debug!(
             "{LOG_PREFIX} build_runtime_snapshot: coalesced onto concurrent rebuild req_id={req_id}"
         );
@@ -912,6 +922,7 @@ async fn build_runtime_snapshot(config: &Config, req_id: u64) -> RuntimeSnapshot
     *RUNTIME_SNAPSHOT_CACHE.lock() = Some(CachedRuntimeSnapshot {
         snapshot: snapshot.clone(),
         fetched_at: Instant::now(),
+        config_key: config.workspace_dir.clone(),
     });
 
     snapshot

src/openhuman/app_state/ops_tests.rs

@@ -210,6 +210,7 @@ fn runtime_snapshot_cache_hit_within_ttl() {
     *RUNTIME_SNAPSHOT_CACHE.lock() = Some(CachedRuntimeSnapshot {
         snapshot: dummy.clone(),
         fetched_at: Instant::now(),
+        config_key: std::path::PathBuf::new(),
     });
 
     let cache = RUNTIME_SNAPSHOT_CACHE.lock();
@@ -229,6 +230,7 @@ fn runtime_snapshot_cache_miss_after_ttl() {
     *RUNTIME_SNAPSHOT_CACHE.lock() = Some(CachedRuntimeSnapshot {
         snapshot: build_dummy_runtime_snapshot(),
         fetched_at: Instant::now() - (RUNTIME_SNAPSHOT_TTL + Duration::from_millis(100)),
+        config_key: std::path::PathBuf::new(),
     });
 
     let cache = RUNTIME_SNAPSHOT_CACHE.lock();
@@ -245,12 +247,14 @@ fn fresh_cached_runtime_snapshot_returns_entry_within_ttl() {
     let _reset = SnapshotCacheResetGuard;
 
     let dummy = build_dummy_runtime_snapshot();
+    let cfg = Config::default();
     *RUNTIME_SNAPSHOT_CACHE.lock() = Some(CachedRuntimeSnapshot {
         snapshot: dummy.clone(),
         fetched_at: Instant::now(),
+        config_key: cfg.workspace_dir.clone(),
     });
 
-    let served = fresh_cached_runtime_snapshot(1).expect("fresh entry should be served");
+    let served = fresh_cached_runtime_snapshot(&cfg, 1).expect("fresh entry should be served");
     assert_eq!(served.autocomplete.phase, dummy.autocomplete.phase);
 }
 
@@ -259,16 +263,48 @@ fn fresh_cached_runtime_snapshot_misses_when_stale_or_empty() {
     let _cache_lock = APP_STATE_CACHE_TEST_LOCK.lock();
     let _reset = SnapshotCacheResetGuard;
 
+    let cfg = Config::default();
+
     // Empty cache → miss (forces the single-flight rebuild path).
     *RUNTIME_SNAPSHOT_CACHE.lock() = None;
-    assert!(fresh_cached_runtime_snapshot(2).is_none());
+    assert!(fresh_cached_runtime_snapshot(&cfg, 2).is_none());
 
     // Stale cache → miss, so the TTL bump can't silently keep serving old data.
     *RUNTIME_SNAPSHOT_CACHE.lock() = Some(CachedRuntimeSnapshot {
         snapshot: build_dummy_runtime_snapshot(),
         fetched_at: Instant::now() - (RUNTIME_SNAPSHOT_TTL + Duration::from_millis(100)),
+        config_key: cfg.workspace_dir.clone(),
     });
-    assert!(fresh_cached_runtime_snapshot(3).is_none());
+    assert!(fresh_cached_runtime_snapshot(&cfg, 3).is_none());
+}
+
+#[test]
+fn fresh_cached_runtime_snapshot_misses_on_config_key_mismatch() {
+    let _cache_lock = APP_STATE_CACHE_TEST_LOCK.lock();
+    let _reset = SnapshotCacheResetGuard;
+
+    // A fresh entry cached for one workspace must never be served to another
+    // config — a second user, or an E2E harness with an injected service mock,
+    // has to rebuild against its own runtime instead of reading a foreign one.
+    let mut owner = Config::default();
+    owner.workspace_dir = std::path::PathBuf::from("/tmp/ws-owner");
+    let mut other = Config::default();
+    other.workspace_dir = std::path::PathBuf::from("/tmp/ws-other");
+
+    *RUNTIME_SNAPSHOT_CACHE.lock() = Some(CachedRuntimeSnapshot {
+        snapshot: build_dummy_runtime_snapshot(),
+        fetched_at: Instant::now(),
+        config_key: owner.workspace_dir.clone(),
+    });
+
+    assert!(
+        fresh_cached_runtime_snapshot(&owner, 4).is_some(),
+        "a config reads back its own fresh snapshot"
+    );
+    assert!(
+        fresh_cached_runtime_snapshot(&other, 5).is_none(),
+        "a foreign config misses instead of serving the wrong runtime"
+    );
 }
 
 #[test]

src/openhuman/channels/bus.rs

@@ -245,11 +245,16 @@ impl EventHandler for ChannelInboundSubscriber {
                     }
                 }
                 _ = edit_timer.tick() => {
-                    if streaming_state.thinking_dirty && !streaming_state.thinking_edit_disabled {
-                        flush_thinking_message(channel, &mut streaming_state).await;
-                    }
-                    if streaming_state.dirty && !streaming_state.edit_disabled {
-                        flush_streaming_edit(channel, &mut streaming_state).await;
+                    // Progressive draft/thinking bubbles require edit+delete
+                    // support; skip them on channels that lack it (Discord) so
+                    // they don't leave un-cleanable placeholder messages.
+                    if channel_supports_progressive_ui(channel) {
+                        if streaming_state.thinking_dirty && !streaming_state.thinking_edit_disabled {
+                            flush_thinking_message(channel, &mut streaming_state).await;
+                        }
+                        if streaming_state.dirty && !streaming_state.edit_disabled {
+                            flush_streaming_edit(channel, &mut streaming_state).await;
+                        }
                     }
                 }
                 _ = typing_timer.tick() => {
@@ -258,7 +263,9 @@ impl EventHandler for ChannelInboundSubscriber {
                     }
                 }
                 _ = filler_timer.tick() => {
-                    if !streaming_state.filler_disabled {
+                    // Fillers ("💭 Still working on it…") are ephemeral and
+                    // deleted on finalize — only post them where cleanup works.
+                    if channel_supports_progressive_ui(channel) && !streaming_state.filler_disabled {
                         send_filler_message(channel, &mut streaming_state).await;
                     }
                 }
@@ -297,6 +304,28 @@ const MAX_TYPING_FAILURES: u32 = 2;
 /// on finalization alongside the ephemeral thinking bubble.
 const FILLER_INTERVAL: tokio::time::Duration = tokio::time::Duration::from_secs(13);
 
+/// Whether a channel supports the progressive-UI placeholders — the
+/// evolving draft bubble, the rotating "💭" fillers, and the ephemeral
+/// "thinking" bubble. All three rely on the backend supporting **both**
+/// message *edit* and *delete*: edit keeps a single bubble evolving in
+/// place, delete removes it once the final reply lands. Telegram supports
+/// both. Discord's adapter supports **neither** (edits 404, delete is a
+/// hard `Delete not supported` stub), so every placeholder becomes a
+/// permanent, un-editable, un-deletable message — the channel fills with
+/// "💭 Still working on it…" bubbles.
+///
+/// This is an **allowlist**, not a denylist: only channels confirmed to
+/// support edit+delete opt in. A new/unknown adapter therefore fails *safe*
+/// (placeholders suppressed) rather than silently re-introducing the spam bug
+/// this gate was added to fix.
+fn channel_supports_progressive_ui(channel: &str) -> bool {
+    // Inbound channels arrive provider-prefixed from the socket layer
+    // (e.g. `discord:<guild>`, `tg:<chat>`), so compare the provider prefix,
+    // not the whole id — mirroring `channel_is_telegram`.
+    let provider = channel.split(':').next().unwrap_or(channel);
+    matches!(provider, "telegram" | "tg")
+}
+
 /// Maximum consecutive filler-send failures before we stop trying.
 /// Same rationale as the thinking/typing latches.
 const MAX_FILLER_FAILURES: u32 = 2;
@@ -1042,7 +1071,25 @@ fn channel_is_telegram(channel: &str) -> bool {
 
 #[cfg(test)]
 mod inbound_thread_id_tests {
-    use super::{derive_inbound_client_id, derive_inbound_thread_id};
+    use super::{
+        channel_supports_progressive_ui, derive_inbound_client_id, derive_inbound_thread_id,
+    };
+
+    #[test]
+    fn progressive_ui_is_an_allowlist_failing_safe_for_unknown_channels() {
+        // Only edit+delete-capable providers opt in. Telegram supports both;
+        // everything else (Discord's stub delete / 404 edits, and any new or
+        // unknown adapter) is suppressed so the "💭" spam can't reappear.
+        assert!(channel_supports_progressive_ui("telegram"));
+        assert!(channel_supports_progressive_ui("tg"));
+        // Inbound channels arrive provider-prefixed — the prefix must still match.
+        assert!(channel_supports_progressive_ui("tg:12345"));
+        assert!(!channel_supports_progressive_ui("discord"));
+        assert!(!channel_supports_progressive_ui("discord:guild-1"));
+        // Unknown/new adapters fail safe (allowlist, not denylist).
+        assert!(!channel_supports_progressive_ui("slack"));
+        assert!(!channel_supports_progressive_ui("whatsapp:123"));
+    }
 
     #[test]
     fn socket_inbound_client_id_keys_per_sender() {

src/openhuman/channels/providers/telegram/channel_core.rs

@@ -3,7 +3,7 @@
 use super::channel_types::{
     TelegramChannel, TelegramUpdateWindow, TELEGRAM_RECENT_UPDATE_CACHE_SIZE,
 };
-use super::text::TELEGRAM_BIND_COMMAND;
+use super::text::{TELEGRAM_BIND_COMMAND, TELEGRAM_START_COMMAND};
 use crate::openhuman::config::{Config, StreamMode};
 use crate::openhuman::security::pairing::PairingGuard;
 use anyhow::Context;
@@ -123,6 +123,14 @@ impl TelegramChannel {
         format!("{}/bot{}/{method}", self.api_base, self.bot_token)
     }
 
+    /// Point outbound Telegram API calls at `base` (test-only seam). Used to
+    /// aim `send()` at a dead local port so onboarding tests exercise the
+    /// decision logic without reaching api.telegram.org.
+    #[cfg(test)]
+    pub(crate) fn set_api_base_for_tests(&mut self, base: impl Into<String>) {
+        self.api_base = base.into();
+    }
+
     pub(crate) fn pairing_code_active(&self) -> bool {
         self.pairing
             .as_ref()
@@ -140,6 +148,21 @@ impl TelegramChannel {
         parts.next().map(str::trim).filter(|code| !code.is_empty())
     }
 
+    /// Whether `text` is the standard Telegram `/start` bot-onboarding command
+    /// (optionally addressed as `/start@botname`, with or without a payload).
+    ///
+    /// On the self-bot-token path this is the operator's explicit "I'm setting up
+    /// my bot" signal: the first `/start` while pairing is still pending pairs the
+    /// sender (see `handle_unauthorized_message`), matching the "first sender after
+    /// /start" behaviour sanctioned by openhuman#4381.
+    pub(crate) fn is_start_command(text: &str) -> bool {
+        let Some(command) = text.split_whitespace().next() else {
+            return false;
+        };
+        let base_command = command.split('@').next().unwrap_or(command);
+        base_command == TELEGRAM_START_COMMAND
+    }
+
     pub(crate) fn track_update_id(&self, update_id: i64) -> bool {
         let mut window = self.recent_updates.lock();
         if window.recent_lookup.contains(&update_id) {

src/openhuman/channels/providers/telegram/channel_recv.rs

@@ -204,12 +204,16 @@ impl TelegramChannel {
     /// Legitimate first-run pairing (`allowed_users=[]` at construction) always sets
     /// `pairing = Some(...)` so it is never suppressed here.
     pub(crate) fn is_race_condition_instance(&self) -> bool {
-        let runtime_empty = self
-            .allowed_users
+        self.allowlist_is_empty() && self.pairing.is_none()
+    }
+
+    /// Whether the runtime allowlist currently has no entries. A poisoned lock is
+    /// treated as non-empty (fail-closed) so we never widen access on a lock error.
+    pub(crate) fn allowlist_is_empty(&self) -> bool {
+        self.allowed_users
             .read()
             .map(|users| users.is_empty())
-            .unwrap_or(false);
-        runtime_empty && self.pairing.is_none()
+            .unwrap_or(false)
     }
 
     /// Build the de-bounce key for approval prompts: `"{chat_id}:{sender}"`.
@@ -308,55 +312,92 @@ impl TelegramChannel {
             return;
         }
 
+        // ── First-run onboarding: `/start` pairs the operator ────────────────────
+        // On the self-bot-token path a blank allowlist arms `pairing = Some(..)` (a
+        // fresh bot is world-reachable by @username, so we must not allow-all like
+        // Discord). The one-time bind code, however, is only printed to core stdout
+        // and is invisible to a desktop operator — leaving the gate un-openable and
+        // every message stuck on the approval prompt (openhuman#4381).
+        //
+        // The operator's first `/start` is their explicit "I'm setting up my bot"
+        // signal. While pairing is still pending we treat that sender as the owner,
+        // add them to the allowlist, and let their subsequent messages reach the
+        // agent — matching the "first sender after /start" behaviour the issue
+        // sanctions. The guard is tight: `pairing.is_some()` excludes an
+        // explicitly-configured allowlist, and `allowlist_is_empty()` restricts
+        // onboarding to the genuine first sender — once the operator is bound the
+        // list is non-empty, so a later stranger's `/start` falls through to the
+        // normal approval prompt instead of being auto-approved.
+        // SECURITY (first-sender-wins TOFU): unlike `/bind <code>` — which
+        // requires the stdout secret and goes through `try_pair`'s lockout —
+        // `/start` onboarding trusts the first sender with no secret and no
+        // rate-limit. Anyone who learns the bot's `@username` before the operator
+        // sends the first message could claim ownership. The private-chat guard
+        // below removes the group attack surface (the common hijack); the residual
+        // window is a stale, world-reachable un-paired bot. Bounding onboarding to
+        // a startup time-window is a reasonable future hardening (see openhuman#4381).
+        if self.pairing.is_some()
+            && self.allowlist_is_empty()
+            // Private chats only: operator setup for a self-bot-token is a DM
+            // action. Onboarding the first `/start` sender in a *group* would let
+            // any member claim operator ownership (the un-paired bot may be added
+            // to a group mid-setup), so a group `/start` falls through to the
+            // normal approval prompt instead.
+            && !Self::is_group_message(message)
+            && text.map(Self::is_start_command).unwrap_or(false)
+        {
+            match Self::bindable_identity(&normalized_username, normalized_sender_id.as_deref()) {
+                Some(identity) => {
+                    tracing::info!(
+                        chat_id,
+                        identity,
+                        "[telegram][approval] /start onboarding: pairing first sender as operator"
+                    );
+                    self.approve_and_persist_sender(&identity, &chat_id).await;
+                    // Finish the one-time pairing flow: the operator is bound
+                    // via /start rather than /bind <code>, so consume the code
+                    // here too — otherwise the stdout code stays live and a
+                    // later sender who obtains it could still /bind themselves.
+                    if let Some(pairing) = self.pairing.as_ref() {
+                        pairing.invalidate_code();
+                    }
+                }
+                None => {
+                    let _ = self
+                        .send(&SendMessage::new(
+                            "❌ Could not identify your Telegram account from /start. Ensure your account has a username or stable user ID, then try again.",
+                            &chat_id,
+                        ))
+                        .await;
+                }
+            }
+            return;
+        }
+
         if let Some(code) = text.and_then(Self::extract_bind_code) {
             if let Some(pairing) = self.pairing.as_ref() {
                 match pairing.try_pair(code).await {
                     Ok(Some(_token)) => {
-                        let bind_identity = normalized_sender_id.clone().or_else(|| {
-                            if normalized_username.is_empty() || normalized_username == "unknown" {
-                                None
-                            } else {
-                                Some(normalized_username.clone())
+                        match Self::bindable_identity(
+                            &normalized_username,
+                            normalized_sender_id.as_deref(),
+                        ) {
+                            Some(identity) => {
+                                tracing::info!(
+                                    chat_id,
+                                    identity,
+                                    "[telegram][approval] paired via bind code and allowlisted identity"
+                                );
+                                self.approve_and_persist_sender(&identity, &chat_id).await;
                             }
-                        });
-
-                        if let Some(identity) = bind_identity {
-                            self.add_allowed_identity_runtime(&identity);
-                            match self.persist_allowed_identity(&identity).await {
-                                Ok(()) => {
-                                    let _ = self
-                                        .send(&SendMessage::new(
-                                            "✅ Telegram account bound successfully. You can talk to OpenHuman now.",
-                                            &chat_id,
-                                        ))
-                                        .await;
-                                    tracing::info!(
-                                        chat_id,
-                                        identity,
-                                        "[telegram][approval] paired and allowlisted identity"
-                                    );
-                                }
-                                Err(e) => {
-                                    tracing::error!(
-                                        chat_id,
-                                        error = %e,
-                                        "[telegram][approval] failed to persist allowlist after bind"
-                                    );
-                                    let _ = self
-                                        .send(&SendMessage::new(
-                                            "⚠️ Bound for this runtime, but failed to persist config. Access may be lost after restart; check config file permissions.",
-                                            &chat_id,
-                                        ))
-                                        .await;
-                                }
+                            None => {
+                                let _ = self
+                                    .send(&SendMessage::new(
+                                        "❌ Could not identify your Telegram account. Ensure your account has a username or stable user ID, then retry.",
+                                        &chat_id,
+                                    ))
+                                    .await;
                             }
-                        } else {
-                            let _ = self
-                                .send(&SendMessage::new(
-                                    "❌ Could not identify your Telegram account. Ensure your account has a username or stable user ID, then retry.",
-                                    &chat_id,
-                                ))
-                                .await;
                         }
                     }
                     Ok(None) => {
@@ -411,25 +452,86 @@ impl TelegramChannel {
              Allowlist Telegram username (without '@') or numeric user ID."
         );
 
-        let _ = self
-            .send(&SendMessage::new(
-                "🔐 This bot requires operator approval.\n\nAsk the operator to approve the pairing in the web UI, then send your message again.".to_string(),
-                &chat_id,
-            ))
-            .await;
-
-        if self.pairing_code_active() {
+        // Copy depends on whether first-run pairing is armed. In pairing mode the
+        // operator unlocks the bot by sending `/start` (or `/bind <code>` if they
+        // have the code from the app); there is no "approve in the web UI" action for
+        // the self-bot-token path, so we must not point the user at one (openhuman#4381).
+        //
+        // Only advertise the `/start` onboarding hint in a private chat — in a group
+        // it would invite any member to claim operator ownership, matching the
+        // private-only onboarding gate above.
+        if self.pairing_code_active() && !Self::is_group_message(message) {
             tracing::debug!(
                 chat_id,
                 sender = sender_key,
-                "[telegram][approval] pairing code active — sending /bind hint"
+                "[telegram][approval] pairing pending — sending /start onboarding prompt"
             );
             let _ = self
                 .send(&SendMessage::new(
-                    "ℹ️ If operator provides a one-time pairing code, you can also run `/bind <code>`.",
+                    "🔐 This bot isn't set up yet.\n\nIf you're the operator, send /start to finish connecting your bot. \
+                     Otherwise ask the operator to add your Telegram username (without '@') or numeric user ID to the bot's Allowed Users, then message again.\n\n\
+                     If the operator gave you a one-time pairing code, run `/bind <code>`.".to_string(),
                     &chat_id,
                 ))
                 .await;
+        } else {
+            let _ = self
+                .send(&SendMessage::new(
+                    "🔐 This bot requires operator approval.\n\nAsk the operator to add your Telegram username (without '@') or numeric user ID to the bot's Allowed Users, then send your message again.".to_string(),
+                    &chat_id,
+                ))
+                .await;
+        }
+    }
+
+    /// Resolve a stable identity to allowlist for a sender: prefer the numeric user
+    /// ID (immutable), fall back to a real username. Returns `None` when the sender
+    /// has neither (`normalized_username` empty or the `"unknown"` sentinel and no id).
+    pub(crate) fn bindable_identity(
+        normalized_username: &str,
+        normalized_sender_id: Option<&str>,
+    ) -> Option<String> {
+        if let Some(id) = normalized_sender_id.filter(|id| !id.is_empty()) {
+            return Some(id.to_string());
+        }
+        if normalized_username.is_empty() || normalized_username == "unknown" {
+            return None;
+        }
+        Some(normalized_username.to_string())
+    }
+
+    /// Add `identity` to the allowlist (runtime + persisted config) and acknowledge
+    /// to the chat. Shared by the `/start` onboarding and `/bind <code>` paths so
+    /// both stay in lock-step on persistence and messaging.
+    pub(crate) async fn approve_and_persist_sender(&self, identity: &str, chat_id: &str) {
+        self.add_allowed_identity_runtime(identity);
+        match self.persist_allowed_identity(identity).await {
+            Ok(()) => {
+                let _ = self
+                    .send(&SendMessage::new(
+                        "✅ You're all set — OpenHuman is connected. Send me a message and I'll take it from here.",
+                        chat_id,
+                    ))
+                    .await;
+                tracing::info!(
+                    chat_id,
+                    identity,
+                    "[telegram][approval] allowlisted identity (runtime + persisted)"
+                );
+            }
+            Err(e) => {
+                tracing::error!(
+                    chat_id,
+                    error = %e,
+                    "[telegram][approval] failed to persist allowlist after approval"
+                );
+                let _ = self
+                    .send(&SendMessage::new(
+                        "⚠️ Connected for now, but I couldn't save it — access may be lost after a restart. Check the config file permissions.",
+                        chat_id,
+                    ))
+                    .await;
+            }
         }
     }
 

src/openhuman/channels/providers/telegram/channel_tests.rs

@@ -313,6 +313,70 @@ fn telegram_extract_bind_code_rejects_invalid_forms() {
     assert_eq!(TelegramChannel::extract_bind_code("/start"), None);
 }
 
+#[test]
+fn telegram_is_start_command_accepts_valid_forms() {
+    assert!(TelegramChannel::is_start_command("/start"));
+    // Addressed to a specific bot in a group.
+    assert!(TelegramChannel::is_start_command("/start@openhuman_bot"));
+    // Deep-link / payload after the command (still a /start).
+    assert!(TelegramChannel::is_start_command("/start deadbeef"));
+    // Leading whitespace is tolerated (split_whitespace skips it).
+    assert!(TelegramChannel::is_start_command("  /start"));
+}
+
+#[test]
+fn telegram_is_start_command_rejects_non_start() {
+    assert!(!TelegramChannel::is_start_command("/bind 123"));
+    assert!(!TelegramChannel::is_start_command("start"));
+    assert!(!TelegramChannel::is_start_command("hello"));
+    assert!(!TelegramChannel::is_start_command(""));
+    // Must be the whole command token, not a prefix.
+    assert!(!TelegramChannel::is_start_command("/started"));
+}
+
+#[test]
+fn telegram_bindable_identity_prefers_numeric_id() {
+    // Numeric id is immutable, so it wins over a mutable username.
+    assert_eq!(
+        TelegramChannel::bindable_identity("alice", Some("123456789")),
+        Some("123456789".to_string())
+    );
+}
+
+#[test]
+fn telegram_bindable_identity_falls_back_to_username() {
+    assert_eq!(
+        TelegramChannel::bindable_identity("alice", None),
+        Some("alice".to_string())
+    );
+    // An empty id string is ignored, not used as the identity.
+    assert_eq!(
+        TelegramChannel::bindable_identity("alice", Some("")),
+        Some("alice".to_string())
+    );
+}
+
+#[test]
+fn telegram_bindable_identity_none_when_unidentified() {
+    assert_eq!(TelegramChannel::bindable_identity("unknown", None), None);
+    assert_eq!(TelegramChannel::bindable_identity("", None), None);
+}
+
+#[test]
+fn telegram_allowlist_is_empty_tracks_runtime_state() {
+    // Fresh pairing-mode channel starts empty ...
+    let ch = TelegramChannel::new("t".into(), vec![], false);
+    assert!(ch.allowlist_is_empty());
+    // ... and flips to non-empty once the first sender is approved at runtime,
+    // which is what closes the `/start` first-run onboarding window.
+    ch.add_allowed_identity_runtime("123456789");
+    assert!(!ch.allowlist_is_empty());
+
+    // A channel constructed with an explicit allowlist is never "empty".
+    let configured = TelegramChannel::new("t".into(), vec!["alice".into()], false);
+    assert!(!configured.allowlist_is_empty());
+}
+
 #[test]
 fn parse_attachment_markers_extracts_multiple_types() {
     let message = "Here are files [IMAGE:/tmp/a.png] and [DOCUMENT:https://example.com/a.pdf]";
@@ -2167,3 +2231,58 @@ fn approval_debounce_evicts_entries_past_window() {
         "[telegram][approval] the new entry must be inserted"
     );
 }
+
+#[tokio::test]
+async fn start_onboarding_is_private_only_and_consumes_the_code() {
+    // Aim outbound sends at a dead local port so the approve/hint `self.send()`
+    // fast-fails (connection refused) instead of reaching api.telegram.org — the
+    // onboarding *decision* (runtime allowlist + one-time code) is asserted
+    // regardless of the send outcome.
+    let mut ch = TelegramChannel::new("fake-token".into(), vec![], false);
+    ch.set_api_base_for_tests("http://127.0.0.1:1");
+    assert!(ch.allowlist_is_empty());
+    assert!(
+        ch.pairing_code_active(),
+        "fresh pairing-mode channel arms a code"
+    );
+
+    // A PRIVATE `/start` onboards the first sender AND consumes the code, so it
+    // can't later be replayed via `/bind`.
+    let private_start = serde_json::json!({
+        "message": {
+            "chat": { "id": 111, "type": "private" },
+            "from": { "id": 222, "username": "operator" },
+            "text": "/start"
+        }
+    });
+    ch.handle_unauthorized_message(&private_start).await;
+    assert!(
+        !ch.allowlist_is_empty(),
+        "private /start onboards the operator"
+    );
+    assert!(
+        !ch.pairing_code_active(),
+        "the one-time code is consumed on /start onboarding"
+    );
+
+    // A GROUP `/start` must NOT onboard — otherwise any member could claim
+    // operator ownership. It falls through to the normal approval prompt.
+    let mut group_ch = TelegramChannel::new("fake-token".into(), vec![], false);
+    group_ch.set_api_base_for_tests("http://127.0.0.1:1");
+    let group_start = serde_json::json!({
+        "message": {
+            "chat": { "id": -100, "type": "supergroup" },
+            "from": { "id": 333, "username": "member" },
+            "text": "/start"
+        }
+    });
+    group_ch.handle_unauthorized_message(&group_start).await;
+    assert!(
+        group_ch.allowlist_is_empty(),
+        "a group /start must not onboard anyone"
+    );
+    assert!(
+        group_ch.pairing_code_active(),
+        "a group /start leaves the one-time code intact"
+    );
+}

src/openhuman/channels/providers/telegram/text.rs

@@ -3,6 +3,7 @@
 /// Telegram's maximum message length for text messages
 pub(crate) const TELEGRAM_MAX_MESSAGE_LENGTH: usize = 4096;
 pub(crate) const TELEGRAM_BIND_COMMAND: &str = "/bind";
+pub(crate) const TELEGRAM_START_COMMAND: &str = "/start";
 
 pub(crate) fn split_message_for_telegram(message: &str) -> Vec<String> {
     if message.chars().count() <= TELEGRAM_MAX_MESSAGE_LENGTH {

src/openhuman/channels/runtime/dispatch/routing.rs

@@ -9,8 +9,12 @@
 use crate::openhuman::agent::harness::definition::{
     AgentDefinition, AgentDefinitionRegistry, ToolScope,
 };
-use crate::openhuman::composio::fetch_connected_integrations;
+use crate::openhuman::composio::{
+    cached_active_integrations_including_expired, fetch_connected_integrations_status,
+    FetchConnectedIntegrationsStatus,
+};
 use crate::openhuman::config::Config;
+use crate::openhuman::context::prompt::ConnectedIntegration;
 use crate::openhuman::tools::{orchestrator_tools, Tool};
 use std::collections::HashSet;
 use std::time::Duration;
@@ -108,25 +112,43 @@ pub(super) async fn resolve_target_agent(channel: &str) -> AgentScoping {
     //
     // Wrap the Composio fetch in a 3-second timeout so a slow/unresponsive
     // Composio API can never block turn dispatch indefinitely.
+    //
+    // Crucially, a transient failure (backend 5xx / no client for a beat) or a
+    // timeout must NOT be laundered into "zero connected integrations": that
+    // would drop `delegate_to_integrations_agent` from the turn's tool surface
+    // and leave the channel agent unable to reach Gmail/Slack/etc. — the exact
+    // "just normal inference, no tool calling" symptom. So we take the
+    // status-returning fetch and, on `Unavailable`/timeout, fall back to the
+    // last cached snapshot (same defence the first-party turn path uses) rather
+    // than an empty set. Only an `Authoritative` result — the backend explicitly
+    // confirming an empty set — legitimately collapses the delegation surface.
     const COMPOSIO_FETCH_TIMEOUT_SECS: u64 = 3;
     let extra_tools = if !definition.subagents.is_empty() {
-        let connected = match tokio::time::timeout(
+        // `Ok(status)` on success, `None` when the 3s timeout elapsed.
+        let fetched = tokio::time::timeout(
             Duration::from_secs(COMPOSIO_FETCH_TIMEOUT_SECS),
-            fetch_connected_integrations(&config),
+            fetch_connected_integrations_status(&config),
         )
         .await
-        {
-            Ok(list) => list,
-            Err(_) => {
-                tracing::warn!(
-                    channel = %channel,
-                    target_agent = target_id,
-                    "[dispatch::routing] Composio fetch timed out after {}s — proceeding without connected integrations",
-                    COMPOSIO_FETCH_TIMEOUT_SECS
-                );
-                Vec::new()
-            }
-        };
+        .ok();
+        if matches!(
+            fetched,
+            None | Some(FetchConnectedIntegrationsStatus::Unavailable)
+        ) {
+            tracing::warn!(
+                channel = %channel,
+                target_agent = target_id,
+                timed_out = fetched.is_none(),
+                "[dispatch::routing] Composio unavailable/timed out — using cached integration snapshot instead of an empty set (keeps delegate_to_integrations_agent live)"
+            );
+        }
+        // Use the expiry-tolerant read for the fallback: a transient blip that
+        // lands just after the 60s cache TTL must still preserve the last-known
+        // integrations rather than collapse tool-calling to an empty set.
+        let connected = connected_with_fallback(
+            fetched,
+            cached_active_integrations_including_expired(&config),
+        );
         tracing::debug!(
             channel = %channel,
             target_agent = target_id,
@@ -159,6 +181,31 @@ pub(super) async fn resolve_target_agent(channel: &str) -> AgentScoping {
     }
 }
 
+/// Decide the connected-integration list to expand delegation tools from,
+/// preferring authoritative truth but never letting a transient failure erase
+/// the surface.
+///
+/// * `fetched` — `Some(status)` from the Composio fetch, or `None` when the
+///   dispatch timeout elapsed before it returned.
+/// * `cached` — the last cached snapshot (`cached_active_integrations`).
+///
+/// Only an `Authoritative` result (the backend explicitly reporting the current
+/// set, even if empty) is taken at face value. `Unavailable` or a timeout falls
+/// back to `cached`, so a one-off 5xx/slow call can't drop
+/// `delegate_to_integrations_agent` and silently disable tool calling for the
+/// turn (the "just normal inference" bug). With no cache to fall back on the
+/// result is empty — the same conservative default as before, but reached only
+/// when we genuinely have no better truth.
+pub(super) fn connected_with_fallback(
+    fetched: Option<FetchConnectedIntegrationsStatus>,
+    cached: Option<Vec<ConnectedIntegration>>,
+) -> Vec<ConnectedIntegration> {
+    match fetched {
+        Some(FetchConnectedIntegrationsStatus::Authoritative(list)) => list,
+        Some(FetchConnectedIntegrationsStatus::Unavailable) | None => cached.unwrap_or_default(),
+    }
+}
+
 /// Build the visible-tool whitelist for an agent.
 ///
 /// The set is the union of:
@@ -190,3 +237,73 @@ pub(super) fn build_visible_tool_set(
         }
     }
 }
+
+#[cfg(test)]
+mod connected_fallback_tests {
+    use super::*;
+
+    fn integration(toolkit: &str) -> ConnectedIntegration {
+        ConnectedIntegration {
+            toolkit: toolkit.into(),
+            description: String::new(),
+            tools: vec![],
+            gated_tools: vec![],
+            connected: true,
+            connections: Vec::new(),
+            non_active_status: None,
+        }
+    }
+
+    fn toolkits(list: &[ConnectedIntegration]) -> Vec<String> {
+        list.iter().map(|i| i.toolkit.clone()).collect()
+    }
+
+    #[test]
+    fn authoritative_result_is_taken_verbatim_even_when_empty() {
+        // The backend confirming "zero connections" is truth — do NOT paper over
+        // it with a stale cache, or the agent would advertise integrations the
+        // user actually disconnected.
+        let out = connected_with_fallback(
+            Some(FetchConnectedIntegrationsStatus::Authoritative(vec![])),
+            Some(vec![integration("gmail")]),
+        );
+        assert!(out.is_empty());
+
+        let out = connected_with_fallback(
+            Some(FetchConnectedIntegrationsStatus::Authoritative(vec![
+                integration("gmail"),
+                integration("slack"),
+            ])),
+            None,
+        );
+        assert_eq!(toolkits(&out), vec!["gmail", "slack"]);
+    }
+
+    #[test]
+    fn unavailable_falls_back_to_cached_snapshot() {
+        // A transient backend failure must not drop the delegation surface.
+        let out = connected_with_fallback(
+            Some(FetchConnectedIntegrationsStatus::Unavailable),
+            Some(vec![integration("gmail")]),
+        );
+        assert_eq!(toolkits(&out), vec!["gmail"]);
+    }
+
+    #[test]
+    fn timeout_falls_back_to_cached_snapshot() {
+        // `None` models the dispatch timeout elapsing before the fetch returned.
+        let out = connected_with_fallback(None, Some(vec![integration("notion")]));
+        assert_eq!(toolkits(&out), vec!["notion"]);
+    }
+
+    #[test]
+    fn unavailable_without_cache_is_empty() {
+        // No authoritative truth and no cache → conservative empty set (same
+        // default as before, but only when we genuinely have nothing better).
+        assert!(
+            connected_with_fallback(Some(FetchConnectedIntegrationsStatus::Unavailable), None)
+                .is_empty()
+        );
+        assert!(connected_with_fallback(None, None).is_empty());
+    }
+}

src/openhuman/composio/connected_integrations.rs

@@ -110,6 +110,32 @@ pub fn invalidate_connected_integrations_cache() {
 /// entry would otherwise pin the agent to a frozen view if every
 /// invalidation path silently failed).
 pub fn cached_active_integrations(config: &Config) -> Option<Vec<ConnectedIntegration>> {
+    read_cached_integrations(config, false)
+}
+
+/// Like [`cached_active_integrations`] but returns the last cached snapshot
+/// even when it has aged past [`CACHE_TTL`].
+///
+/// Intended ONLY as a transient-failure fallback: when a live fetch reports
+/// `Unavailable`/times out, preserving the last-known integrations is strictly
+/// better than collapsing the delegation surface to an empty set (which drops
+/// `delegate_to_integrations_agent` and silently disables channel tool-calling).
+/// Without this, a backend blip that lands just after the 60 s TTL expiry still
+/// wipes tool-calling despite having a perfectly good previous snapshot. A fresh
+/// fetch repopulates the cache the moment the backend recovers, so the stale
+/// window is bounded by the outage, not by this call.
+pub fn cached_active_integrations_including_expired(
+    config: &Config,
+) -> Option<Vec<ConnectedIntegration>> {
+    read_cached_integrations(config, true)
+}
+
+/// Shared reader for the integrations cache. `allow_expired` bypasses the
+/// [`CACHE_TTL`] freshness check for the transient-failure fallback path.
+fn read_cached_integrations(
+    config: &Config,
+    allow_expired: bool,
+) -> Option<Vec<ConnectedIntegration>> {
     let key = cache_key(config);
     let guard = match INTEGRATIONS_CACHE.try_read() {
         Ok(g) => g,
@@ -129,7 +155,7 @@ pub fn cached_active_integrations(config: &Config) -> Option<Vec<ConnectedIntegr
         return None;
     };
     let age = cached.cached_at.elapsed();
-    if age > CACHE_TTL {
+    if !allow_expired && age > CACHE_TTL {
         tracing::trace!(
             key = %key,
             age_ms = age.as_millis() as u64,
@@ -138,10 +164,21 @@ pub fn cached_active_integrations(config: &Config) -> Option<Vec<ConnectedIntegr
         );
         return None;
     }
+    // Surface a *very* stale fallback so an unusually long backend outage is
+    // observable rather than silently pinning the agent to an ancient snapshot.
+    if allow_expired && age > 5 * CACHE_TTL {
+        tracing::warn!(
+            key = %key,
+            age_ms = age.as_millis() as u64,
+            ttl_ms = CACHE_TTL.as_millis() as u64,
+            "[composio][integrations_cache] serving a heavily-stale integrations snapshot on transient-failure fallback (backend outage?)"
+        );
+    }
     tracing::trace!(
         key = %key,
         entries = cached.entries.len(),
         age_ms = age.as_millis() as u64,
+        allow_expired,
         "[composio][integrations_cache] cached_active_integrations:hit"
     );
     Some(cached.entries.clone())

src/openhuman/composio/mod.rs

@@ -73,8 +73,8 @@ pub use action_tool::ComposioActionTool;
 pub use client::ComposioClient;
 pub use identity::connection_identity;
 pub use ops::{
-    cached_active_integrations, connected_set_hash, fetch_connected_integrations,
-    fetch_connected_integrations_status, fetch_toolkit_actions,
+    cached_active_integrations, cached_active_integrations_including_expired, connected_set_hash,
+    fetch_connected_integrations, fetch_connected_integrations_status, fetch_toolkit_actions,
     invalidate_connected_integrations_cache, FetchConnectedIntegrationsStatus,
 };
 pub use schemas::{

src/openhuman/composio/ops/mod.rs

@@ -56,8 +56,8 @@ pub use triggers::{
 // (originally at the bottom of ops.rs)
 
 pub use super::connected_integrations::{
-    cached_active_integrations, connected_set_hash, fetch_connected_integrations,
-    fetch_connected_integrations_status, fetch_toolkit_actions,
+    cached_active_integrations, cached_active_integrations_including_expired, connected_set_hash,
+    fetch_connected_integrations, fetch_connected_integrations_status, fetch_toolkit_actions,
     invalidate_connected_integrations_cache, FetchConnectedIntegrationsStatus,
 };
 

src/openhuman/composio/ops_tests.rs

@@ -1637,6 +1637,37 @@ fn cache_entries_expire_after_ttl() {
     );
 }
 
+#[test]
+fn including_expired_serves_stale_snapshot_for_transient_fallback() {
+    let _guard = cache_guard();
+    let tmp = tempfile::TempDir::new().unwrap();
+    let config = test_config(&tmp);
+    let key = crate::openhuman::composio::connected_integrations::cache_key(&config);
+    clear_cache_key(&key);
+    seed_cache(&key, vec![integration("gmail", true)]);
+
+    // Age the entry past the TTL (simulates a session idle > 60s).
+    {
+        let mut guard = INTEGRATIONS_CACHE.write().unwrap();
+        guard.get_mut(&key).unwrap().cached_at =
+            Instant::now() - (CACHE_TTL + Duration::from_secs(1));
+    }
+
+    // The TTL-enforcing read treats the expired entry as missing…
+    assert!(
+        cached_active_integrations(&config).is_none(),
+        "expired entry must not be served by the freshness-checked read"
+    );
+    // …but the transient-failure fallback read preserves the last-known set,
+    // so a backend blip just after TTL expiry doesn't drop tool-calling.
+    let stale = cached_active_integrations_including_expired(&config)
+        .expect("expired entry should still be returned by the fallback read");
+    assert_eq!(stale.len(), 1);
+    assert_eq!(stale[0].toolkit, "gmail");
+
+    clear_cache_key(&key);
+}
+
 // ── Trigger management ops (PR #671) ────────────────────────────────
 
 #[tokio::test]

src/openhuman/orchestration/graph/state.rs

@@ -3,7 +3,7 @@
 //! `OrchestrationState` is the spec's single `StateGraph` state object: one
 //! value flows through the whole wake path (normalize → frontend → execute →
 //! frontend → send_dm → context_guard → END) and is checkpointed at every
-//! super-step boundary by [`SqlRunLedgerCheckpointer`](crate::openhuman::tinyagents::SqlRunLedgerCheckpointer)
+//! super-step boundary by [`SqliteCheckpointer`](tinyagents::graph::SqliteCheckpointer)
 //! under the thread id `orchestration:<session_id>`.
 //!
 //! Every field is serde-serializable so a mid-cycle crash can resume from the

src/openhuman/security/pairing.rs

@@ -86,6 +86,24 @@ impl PairingGuard {
         self.pairing_code.lock().clone()
     }
 
+    /// Invalidate the one-time pairing code without pairing a token.
+    ///
+    /// `try_pair` consumes the code as part of a successful `/bind <code>`, but
+    /// some onboarding flows bind the operator through a different signal (e.g.
+    /// the Telegram self-token `/start` path allowlists the first sender
+    /// directly). Those paths must still finish the one-time flow so the code
+    /// can't later be replayed by another sender who obtains it — call this once
+    /// the operator is bound. Idempotent; a no-op if the code is already gone.
+    pub fn invalidate_code(&self) {
+        let mut code = self.pairing_code.lock();
+        if code.is_some() {
+            *code = None;
+            log::info!(
+                "[openhuman:pairing] one-time pairing code invalidated (operator onboarded)"
+            );
+        }
+    }
+
     /// Whether pairing is required at all.
     pub fn require_pairing(&self) -> bool {
         self.require_pairing

src/openhuman/security/pairing_tests.rs

@@ -316,3 +316,23 @@ async fn lockout_returns_remaining_seconds() {
         "Remaining lockout should be ~{PAIR_LOCKOUT_SECS}s, got {err}s"
     );
 }
+
+#[test]
+async fn invalidate_code_closes_the_bind_window() {
+    let (guard, _) = PairingGuard::new(true, &[]);
+    let code = guard.pairing_code().unwrap().to_string();
+
+    // A non-/bind onboarding path (e.g. Telegram `/start`) finishes the
+    // one-time flow by invalidating the code directly.
+    guard.invalidate_code();
+
+    assert!(guard.pairing_code().is_none(), "code must be cleared");
+    // The old code can no longer pair — the window is closed.
+    assert!(
+        guard.try_pair(&code).await.unwrap().is_none(),
+        "invalidated code must not pair a later sender"
+    );
+    // Idempotent.
+    guard.invalidate_code();
+    assert!(guard.pairing_code().is_none());
+}

tests/config_auth_app_state_connectivity_e2e.rs

@@ -4871,11 +4871,10 @@ async fn app_state_snapshot_degrades_runtime_service_status_failures() {
     let _service_state =
         EnvVarGuard::set_to_path("OPENHUMAN_SERVICE_MOCK_STATE_FILE", &service_state_path);
 
-    // The runtime snapshot cache is process-global and not keyed by config.
-    // Let prior app_state_snapshot tests age out so this call exercises the
-    // service-status fallback instead of returning a cached runtime.
-    tokio::time::sleep(Duration::from_millis(2_100)).await;
-
+    // The runtime snapshot cache is keyed by config identity (workspace_dir), so
+    // this harness's unique workspace guarantees a cache miss regardless of prior
+    // app_state_snapshot tests — the call exercises the service-status fallback
+    // against our injected mock rather than returning a foreign cached runtime.
     let snapshot = rpc(
         &harness.rpc_base,
         30_051,