From fbca4be5be3e71ec54689f37b205059f317b7d92 Mon Sep 17 00:00:00 2001 From: shanu Date: Thu, 2 Jul 2026 16:51:03 +0530 Subject: [PATCH 1/7] fix(channels): pair Telegram self-bot-token on /start and keep channel tool-calling alive MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two channel defects that left the messaging bots doing plain inference with no tool calling and, on Telegram's own-bot-token path, no reply at all. Telegram self-bot-token approval (openhuman#4381) - A blank "Allowed Users" list arms first-run pairing, but the one-time bind code was only printed to core stdout — invisible to a desktop operator — so every message stayed stuck on the approval prompt and never reached the agent. Treat the operator's first `/start` (while pairing is pending AND the allowlist is still empty) as their explicit setup signal: allowlist that sender (runtime + persisted) and let their messages through, matching the "first sender after /start" behaviour the issue sanctions. The guard is tight — an explicitly configured allowlist never auto-approves a stranger's /start, and once the operator is bound the window closes. - Replace the misleading "approve the pairing in the web UI" prompt (no such action exists for this path) with accurate /start / `/bind ` guidance. - Factor the shared approve+persist+ack path so /start and /bind stay in lock-step. Channel tool-calling delegation (both Discord and Telegram) - resolve_target_agent laundered a transient Composio failure or a 3s timeout into an empty connected-integration set, which dropped delegate_to_integrations_agent from the turn and disabled tool calling for that message. Use the status-returning fetch and fall back to the cached integration snapshot on Unavailable/timeout (the same defence the first-party turn path uses); only an authoritative empty set collapses the delegation surface. Tests: /start + bindable-identity + allowlist-empty guards; connected-integration fallback matrix (authoritative/unavailable/timeout/no-cache). --- .../providers/telegram/channel_core.rs | 17 +- .../providers/telegram/channel_recv.rs | 189 ++++++++++++------ .../providers/telegram/channel_tests.rs | 64 ++++++ .../channels/providers/telegram/text.rs | 1 + .../channels/runtime/dispatch/routing.rs | 141 +++++++++++-- 5 files changed, 340 insertions(+), 72 deletions(-) diff --git a/src/openhuman/channels/providers/telegram/channel_core.rs b/src/openhuman/channels/providers/telegram/channel_core.rs index 61073ca9ce..d4f080c941 100644 --- a/src/openhuman/channels/providers/telegram/channel_core.rs +++ b/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; @@ -140,6 +140,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) { diff --git a/src/openhuman/channels/providers/telegram/channel_recv.rs b/src/openhuman/channels/providers/telegram/channel_recv.rs index db058ba421..ad7715571f 100644 --- a/src/openhuman/channels/providers/telegram/channel_recv.rs +++ b/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,71 @@ 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. + if self.pairing.is_some() + && self.allowlist_is_empty() + && 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; + } + 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 +431,82 @@ 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; - + // Copy depends on whether first-run pairing is armed. In pairing mode the + // operator unlocks the bot by sending `/start` (or `/bind ` 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). if self.pairing_code_active() { 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 `.", + "🔐 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 `.".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 { + 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 ` 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; + } } } diff --git a/src/openhuman/channels/providers/telegram/channel_tests.rs b/src/openhuman/channels/providers/telegram/channel_tests.rs index 6e3f305d66..4571dfb424 100644 --- a/src/openhuman/channels/providers/telegram/channel_tests.rs +++ b/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]"; diff --git a/src/openhuman/channels/providers/telegram/text.rs b/src/openhuman/channels/providers/telegram/text.rs index e2451505ad..b0c35e5be3 100644 --- a/src/openhuman/channels/providers/telegram/text.rs +++ b/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 { if message.chars().count() <= TELEGRAM_MAX_MESSAGE_LENGTH { diff --git a/src/openhuman/channels/runtime/dispatch/routing.rs b/src/openhuman/channels/runtime/dispatch/routing.rs index 8c8d1ae102..dafd9a0399 100644 --- a/src/openhuman/channels/runtime/dispatch/routing.rs +++ b/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, 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,37 @@ 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)" + ); + } + let connected = connected_with_fallback(fetched, cached_active_integrations(&config)); tracing::debug!( channel = %channel, target_agent = target_id, @@ -159,6 +175,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, + cached: Option>, +) -> Vec { + 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 +231,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 { + 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()); + } +} From 9b6fdd5c7684ef844175e7f7520f621ca94b2c14 Mon Sep 17 00:00:00 2001 From: shanu Date: Fri, 3 Jul 2026 00:22:05 +0530 Subject: [PATCH 2/7] fix(channels): suppress progressive-UI placeholders on Discord MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The channel streaming engine posts ephemeral placeholders during a turn — an evolving "_working…_" draft, rotating "💭 Still working on it…" fillers, and a "thinking" bubble — then edits them in place and deletes them once the final reply lands. This assumes the channel backend supports both message edit and delete. Discord's adapter supports neither: edits 404 (so each tick posts a fresh bubble instead of updating one) and deleteMessage is a hard "Delete not supported" stub (so nothing gets cleaned up). The result is a wall of un-deletable "💭" messages on every Discord turn, especially long ones. Add channel_supports_progressive_ui() (false for Discord) and gate the draft/filler/thinking sends on it. Discord now shows only the typing indicator during the turn and a single clean final reply; Telegram keeps its evolving-bubble UX unchanged. --- src/openhuman/channels/bus.rs | 56 ++++++++++++++++++++++++++++++----- 1 file changed, 49 insertions(+), 7 deletions(-) diff --git a/src/openhuman/channels/bus.rs b/src/openhuman/channels/bus.rs index 62fb87a996..8d13398731 100644 --- a/src/openhuman/channels/bus.rs +++ b/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,25 @@ 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. For such channels we suppress the +/// placeholders entirely and rely on the typing indicator plus the final +/// atomic reply, so the user sees a clean single answer. +fn channel_supports_progressive_ui(channel: &str) -> bool { + // Inbound channels arrive provider-prefixed from the socket layer + // (e.g. `discord:`, `tg:`), so compare the provider prefix, + // not the whole id — mirroring `channel_is_telegram`. + let provider = channel.split(':').next().unwrap_or(channel); + provider != "discord" +} + /// Maximum consecutive filler-send failures before we stop trying. /// Same rationale as the thinking/typing latches. const MAX_FILLER_FAILURES: u32 = 2; @@ -1042,7 +1068,23 @@ 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 discord_suppresses_progressive_ui_other_channels_keep_it() { + // Discord's adapter supports neither edit nor delete, so evolving + // drafts / "💭" fillers / thinking bubbles would become permanent + // un-cleanable spam — suppress them there. Channels with edit+delete + // (Telegram) keep the progressive UI. + assert!(!channel_supports_progressive_ui("discord")); + // Inbound channels arrive provider-prefixed — the prefix must still match. + assert!(!channel_supports_progressive_ui("discord:guild-1")); + assert!(channel_supports_progressive_ui("telegram")); + assert!(channel_supports_progressive_ui("tg")); + assert!(channel_supports_progressive_ui("tg:12345")); + } #[test] fn socket_inbound_client_id_keys_per_sender() { From 07c56655ac932c80336ee60509fe5922c675f0b3 Mon Sep 17 00:00:00 2001 From: shanu Date: Fri, 3 Jul 2026 13:27:48 +0530 Subject: [PATCH 3/7] =?UTF-8?q?fix(channels):=20address=20review=20?= =?UTF-8?q?=E2=80=94=20consume=20pairing=20code=20on=20/start=20+=20stale-?= =?UTF-8?q?cache=20fallback?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Codex review comments on #4414: - channel_recv.rs / pairing.rs (C5): the `/start` self-token onboarding path allowlisted the operator but never consumed the one-time pairing code (only `try_pair` did), leaving the `/bind ` window open for a later sender who obtains the stdout code. Add `PairingGuard::invalidate_code()` and call it once the operator is bound via `/start`, finishing the one-time flow. Test asserts the old code no longer pairs after invalidation. - routing.rs / connected_integrations.rs (C6): the transient-failure fallback read the cache via `cached_active_integrations`, which returns None past the 60s TTL — so a backend blip landing just after expiry still collapsed the delegation surface to empty and dropped tool-calling. Add `cached_active_integrations_including_expired()` (TTL-tolerant) and use it only for the fallback; `cached_active_integrations` keeps its freshness semantics for other callers. Test seeds an expired entry: the fresh read drops it, the fallback read serves it. --- .../providers/telegram/channel_recv.rs | 7 +++++ .../channels/runtime/dispatch/routing.rs | 10 ++++-- .../composio/connected_integrations.rs | 29 ++++++++++++++++- src/openhuman/composio/mod.rs | 4 +-- src/openhuman/composio/ops/mod.rs | 4 +-- src/openhuman/composio/ops_tests.rs | 31 +++++++++++++++++++ src/openhuman/security/pairing.rs | 18 +++++++++++ src/openhuman/security/pairing_tests.rs | 20 ++++++++++++ 8 files changed, 116 insertions(+), 7 deletions(-) diff --git a/src/openhuman/channels/providers/telegram/channel_recv.rs b/src/openhuman/channels/providers/telegram/channel_recv.rs index ad7715571f..193ec1ef98 100644 --- a/src/openhuman/channels/providers/telegram/channel_recv.rs +++ b/src/openhuman/channels/providers/telegram/channel_recv.rs @@ -340,6 +340,13 @@ impl TelegramChannel { "[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 , 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 diff --git a/src/openhuman/channels/runtime/dispatch/routing.rs b/src/openhuman/channels/runtime/dispatch/routing.rs index dafd9a0399..a4b9031876 100644 --- a/src/openhuman/channels/runtime/dispatch/routing.rs +++ b/src/openhuman/channels/runtime/dispatch/routing.rs @@ -10,7 +10,7 @@ use crate::openhuman::agent::harness::definition::{ AgentDefinition, AgentDefinitionRegistry, ToolScope, }; use crate::openhuman::composio::{ - cached_active_integrations, fetch_connected_integrations_status, + cached_active_integrations_including_expired, fetch_connected_integrations_status, FetchConnectedIntegrationsStatus, }; use crate::openhuman::config::Config; @@ -142,7 +142,13 @@ pub(super) async fn resolve_target_agent(channel: &str) -> AgentScoping { "[dispatch::routing] Composio unavailable/timed out — using cached integration snapshot instead of an empty set (keeps delegate_to_integrations_agent live)" ); } - let connected = connected_with_fallback(fetched, cached_active_integrations(&config)); + // 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, diff --git a/src/openhuman/composio/connected_integrations.rs b/src/openhuman/composio/connected_integrations.rs index 695fd4fc7f..915593388c 100644 --- a/src/openhuman/composio/connected_integrations.rs +++ b/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> { + 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> { + 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> { 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 CACHE_TTL { + if !allow_expired && age > CACHE_TTL { tracing::trace!( key = %key, age_ms = age.as_millis() as u64, @@ -142,6 +168,7 @@ pub fn cached_active_integrations(config: &Config) -> Option 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] diff --git a/src/openhuman/security/pairing.rs b/src/openhuman/security/pairing.rs index e86c041aa0..e3121e64c5 100644 --- a/src/openhuman/security/pairing.rs +++ b/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 `, 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 diff --git a/src/openhuman/security/pairing_tests.rs b/src/openhuman/security/pairing_tests.rs index d402788bf8..bb7ae91409 100644 --- a/src/openhuman/security/pairing_tests.rs +++ b/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()); +} From 37bd9f76224763b436ecc0da460f144cf76324e2 Mon Sep 17 00:00:00 2001 From: shanu Date: Fri, 3 Jul 2026 15:22:19 +0530 Subject: [PATCH 4/7] =?UTF-8?q?fix(channels):=20address=20sanil-23=20revie?= =?UTF-8?q?w=20=E2=80=94=20private-only=20/start=20onboarding=20+=20harden?= =?UTF-8?q?ing?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Security review on #4414: - channel_recv.rs: gate the /start onboarding AND the "send /start" hint on !is_group_message. Previously, if the un-paired bot was added to a group during setup, the prompt told every member "send /start to finish connecting" and the first to do so was allowlisted as operator. Operator setup is a private/DM action; a group /start now falls through to the normal approval prompt. Documented the residual first-sender-wins TOFU risk (no secret, no rate-limit). - channel_tests.rs (+ #[cfg(test)] api_base seam in channel_core.rs): test the onboarding decision — private /start onboards + consumes the code; group /start does not onboard (send() aimed at a dead port, no network). - bus.rs: channel_supports_progressive_ui is now an allowlist (telegram|tg) so a new/unknown adapter fails safe instead of re-introducing the placeholder spam. - connected_integrations.rs: warn! when the transient-failure fallback serves a snapshot older than 5x CACHE_TTL (observability). --- src/openhuman/channels/bus.rs | 29 ++++++---- .../providers/telegram/channel_core.rs | 8 +++ .../providers/telegram/channel_recv.rs | 20 ++++++- .../providers/telegram/channel_tests.rs | 55 +++++++++++++++++++ .../composio/connected_integrations.rs | 10 ++++ 5 files changed, 109 insertions(+), 13 deletions(-) diff --git a/src/openhuman/channels/bus.rs b/src/openhuman/channels/bus.rs index 8d13398731..12b1407181 100644 --- a/src/openhuman/channels/bus.rs +++ b/src/openhuman/channels/bus.rs @@ -312,15 +312,18 @@ const FILLER_INTERVAL: tokio::time::Duration = tokio::time::Duration::from_secs( /// 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. For such channels we suppress the -/// placeholders entirely and rely on the typing indicator plus the final -/// atomic reply, so the user sees a clean single answer. +/// "💭 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:`, `tg:`), so compare the provider prefix, // not the whole id — mirroring `channel_is_telegram`. let provider = channel.split(':').next().unwrap_or(channel); - provider != "discord" + matches!(provider, "telegram" | "tg") } /// Maximum consecutive filler-send failures before we stop trying. @@ -1073,17 +1076,19 @@ mod inbound_thread_id_tests { }; #[test] - fn discord_suppresses_progressive_ui_other_channels_keep_it() { - // Discord's adapter supports neither edit nor delete, so evolving - // drafts / "💭" fillers / thinking bubbles would become permanent - // un-cleanable spam — suppress them there. Channels with edit+delete - // (Telegram) keep the progressive UI. - assert!(!channel_supports_progressive_ui("discord")); - // Inbound channels arrive provider-prefixed — the prefix must still match. - assert!(!channel_supports_progressive_ui("discord:guild-1")); + 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] diff --git a/src/openhuman/channels/providers/telegram/channel_core.rs b/src/openhuman/channels/providers/telegram/channel_core.rs index d4f080c941..c1219867f1 100644 --- a/src/openhuman/channels/providers/telegram/channel_core.rs +++ b/src/openhuman/channels/providers/telegram/channel_core.rs @@ -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) { + self.api_base = base.into(); + } + pub(crate) fn pairing_code_active(&self) -> bool { self.pairing .as_ref() diff --git a/src/openhuman/channels/providers/telegram/channel_recv.rs b/src/openhuman/channels/providers/telegram/channel_recv.rs index 193ec1ef98..5da5ae04df 100644 --- a/src/openhuman/channels/providers/telegram/channel_recv.rs +++ b/src/openhuman/channels/providers/telegram/channel_recv.rs @@ -328,8 +328,22 @@ impl TelegramChannel { // 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 ` — 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()) { @@ -442,7 +456,11 @@ impl TelegramChannel { // operator unlocks the bot by sending `/start` (or `/bind ` 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). - if self.pairing_code_active() { + // + // 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, diff --git a/src/openhuman/channels/providers/telegram/channel_tests.rs b/src/openhuman/channels/providers/telegram/channel_tests.rs index 4571dfb424..4edbda2169 100644 --- a/src/openhuman/channels/providers/telegram/channel_tests.rs +++ b/src/openhuman/channels/providers/telegram/channel_tests.rs @@ -2231,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" + ); +} diff --git a/src/openhuman/composio/connected_integrations.rs b/src/openhuman/composio/connected_integrations.rs index 915593388c..eca07cddf6 100644 --- a/src/openhuman/composio/connected_integrations.rs +++ b/src/openhuman/composio/connected_integrations.rs @@ -164,6 +164,16 @@ fn read_cached_integrations( ); 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(), From 653e783550c707ae1a958d84e7da17d688aec16d Mon Sep 17 00:00:00 2001 From: shanu Date: Fri, 3 Jul 2026 16:55:38 +0530 Subject: [PATCH 5/7] fix(app_state): key runtime-snapshot cache by config MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The process-global runtime-snapshot cache (`RUNTIME_SNAPSHOT_CACHE`) held a single entry with no config identity, so a snapshot built for one config was served to any caller within the 10s TTL. In the app_state e2e suite this made `app_state_snapshot_degrades_runtime_service_status_failures` fail deterministically: ~15 prior `app_state_snapshot` tests populate the cache, and the target test's 2.1s age-out sleep is far short of the 10s TTL, so it read a stale foreign runtime instead of its injected `OPENHUMAN_SERVICE_MOCK` failure state. Latent on main; surfaced once the build was repaired so Rust E2E runs. Key the cache entry by `config.workspace_dir` and treat a key mismatch as a miss (rebuild against the requesting config). Production behaviour is unchanged — one config per process means a steady cache hit within the TTL — while a second config or an E2E harness with a unique workspace now gets a fresh build. Drops the obsolete 2.1s sleep from the e2e test and adds a unit test for the key-mismatch miss. --- src/openhuman/app_state/ops.rs | 17 ++++++-- src/openhuman/app_state/ops_tests.rs | 42 +++++++++++++++++-- .../config_auth_app_state_connectivity_e2e.rs | 9 ++-- 3 files changed, 57 insertions(+), 11 deletions(-) diff --git a/src/openhuman/app_state/ops.rs b/src/openhuman/app_state/ops.rs index f4e89fafc1..1e10fb700f 100644 --- a/src/openhuman/app_state/ops.rs +++ b/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 Option { +fn fresh_cached_runtime_snapshot(config: &Config, req_id: u64) -> Option { 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 { 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 diff --git a/src/openhuman/app_state/ops_tests.rs b/src/openhuman/app_state/ops_tests.rs index aadbd11c69..20eb6b784c 100644 --- a/src/openhuman/app_state/ops_tests.rs +++ b/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] diff --git a/tests/config_auth_app_state_connectivity_e2e.rs b/tests/config_auth_app_state_connectivity_e2e.rs index 2c4d1ec035..266d46584c 100644 --- a/tests/config_auth_app_state_connectivity_e2e.rs +++ b/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, From 7ac85276adb2bae918e0fcadd63efd98f83d15cd Mon Sep 17 00:00:00 2001 From: Shanu Date: Fri, 3 Jul 2026 18:41:09 +0530 Subject: [PATCH 6/7] fix(mcp): add frontend_agent and reasoning_agent to MCP resource catalog The orchestration wake-graph agents frontend_agent and reasoning_agent were registered in BUILTINS but missing from RESOURCE_CATALOG, so the catalog_mirrors_builtins parity test failed and cascaded into the PR CI Gate. Add a PromptResource for each, serving its bundled prompt.md, which restores catalog<->BUILTINS parity (39 == 39) and lets MCP clients read their prompt templates. --- src/openhuman/mcp_server/resources.rs | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/src/openhuman/mcp_server/resources.rs b/src/openhuman/mcp_server/resources.rs index 2a08f744e9..3ec6b85718 100644 --- a/src/openhuman/mcp_server/resources.rs +++ b/src/openhuman/mcp_server/resources.rs @@ -271,6 +271,18 @@ const RESOURCE_CATALOG: &[PromptResource] = &[ description: "Sandboxed worker that runs installed skill packages.", content: include_str!("../skill_runtime/agent/skill_executor/prompt.md"), }, + PromptResource { + uri: "openhuman://prompts/agents/frontend_agent", + name: "frontend_agent", + description: "Quick-tier front end of the tiny.place orchestration wake graph that triages master DMs into an immediate channel reply or macro-instructions for the reasoning core.", + content: include_str!("../orchestration/frontend_agent/prompt.md"), + }, + PromptResource { + uri: "openhuman://prompts/agents/reasoning_agent", + name: "reasoning_agent", + description: "Deep-thinking reasoning core of the tiny.place orchestration wake graph that applies the subconscious steering directive and spawns execution sub-agents.", + content: include_str!("../orchestration/reasoning_agent/prompt.md"), + }, ]; /// Returns the `resources/list` result payload listing every catalog entry. From e64b97c41d42ef88e4416f44b5b5bb2304304c11 Mon Sep 17 00:00:00 2001 From: M3gA-Mind Date: Fri, 3 Jul 2026 21:03:43 +0530 Subject: [PATCH 7/7] fix(mcp): drop duplicate frontend/reasoning_agent catalog entries after merge The merge with main silently duplicated the frontend_agent and reasoning_agent PromptResource entries in RESOURCE_CATALOG: both this branch and main independently added them with identical uri/name/content but slightly different descriptions, so git kept both insertions instead of collapsing them. That broke two parity tests: - all_catalog_uris_are_unique (duplicate agent URIs) - catalog_mirrors_builtins (41 catalog agent entries vs 39 BUILTINS) Keep main's canonical entries (split-brain wording) and drop the redundant tiny.place-wording copies. Catalog agent entries back to 39 == BUILTINS.len(); URIs unique again. --- src/openhuman/mcp_server/resources.rs | 12 ------------ 1 file changed, 12 deletions(-) diff --git a/src/openhuman/mcp_server/resources.rs b/src/openhuman/mcp_server/resources.rs index 94c7a8c4fd..a289ad2960 100644 --- a/src/openhuman/mcp_server/resources.rs +++ b/src/openhuman/mcp_server/resources.rs @@ -283,18 +283,6 @@ const RESOURCE_CATALOG: &[PromptResource] = &[ description: "Sandboxed worker that runs installed skill packages.", content: include_str!("../skill_runtime/agent/skill_executor/prompt.md"), }, - PromptResource { - uri: "openhuman://prompts/agents/frontend_agent", - name: "frontend_agent", - description: "Quick-tier front end of the tiny.place orchestration wake graph that triages master DMs into an immediate channel reply or macro-instructions for the reasoning core.", - content: include_str!("../orchestration/frontend_agent/prompt.md"), - }, - PromptResource { - uri: "openhuman://prompts/agents/reasoning_agent", - name: "reasoning_agent", - description: "Deep-thinking reasoning core of the tiny.place orchestration wake graph that applies the subconscious steering directive and spawns execution sub-agents.", - content: include_str!("../orchestration/reasoning_agent/prompt.md"), - }, ]; /// Returns the `resources/list` result payload listing every catalog entry.