[UX] Telegram: Calm reasoning placeholders and suppressed thought streaming (#949)

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

@@ -1475,3 +1475,135 @@ fn parse_update_message_forum_topic_encodes_thread_in_reply_target_and_thread_ts
         "thread_ts is the inbound message_id (not the quoted parent)"
     );
 }
+
+#[tokio::test]
+async fn test_thinking_placeholder_logic() {
+    use crate::openhuman::agent::progress::AgentProgress;
+    use crate::openhuman::channels::traits::Channel;
+    use crate::openhuman::channels::SendMessage;
+    use parking_lot::Mutex;
+    use std::sync::Arc;
+
+    // Mock channel that records updates
+    struct MockTelegramChannel {
+        updates: Arc<Mutex<Vec<String>>>,
+    }
+
+    #[async_trait::async_trait]
+    impl Channel for MockTelegramChannel {
+        fn name(&self) -> &str {
+            "telegram"
+        }
+        fn supports_draft_updates(&self) -> bool {
+            true
+        }
+        async fn send(&self, _: &SendMessage) -> anyhow::Result<()> {
+            Ok(())
+        }
+        async fn listen(
+            &self,
+            _: tokio::sync::mpsc::Sender<crate::openhuman::channels::traits::ChannelMessage>,
+        ) -> anyhow::Result<()> {
+            Ok(())
+        }
+        async fn send_draft(&self, _: &SendMessage) -> anyhow::Result<Option<String>> {
+            Ok(Some("123".to_string()))
+        }
+        async fn update_draft(&self, _: &str, _: &str, text: &str) -> anyhow::Result<()> {
+            self.updates.lock().push(text.to_string());
+            Ok(())
+        }
+        async fn finalize_draft(
+            &self,
+            _: &str,
+            _: &str,
+            text: &str,
+            _: Option<&str>,
+        ) -> anyhow::Result<()> {
+            self.updates.lock().push(format!("FINAL: {}", text));
+            Ok(())
+        }
+    }
+
+    let updates = Arc::new(Mutex::new(Vec::new()));
+    let channel = Arc::new(MockTelegramChannel {
+        updates: Arc::clone(&updates),
+    });
+    let (tx, mut rx) = tokio::sync::mpsc::channel::<AgentProgress>(10);
+
+    let channel_clone = Arc::clone(&channel);
+    let handle = tokio::spawn(async move {
+        let mut accumulated = String::new();
+        let reply_target = "test_chat";
+        let draft_id = "123";
+        while let Some(progress) = rx.recv().await {
+            match progress {
+                AgentProgress::TextDelta { delta, .. } => {
+                    accumulated.push_str(&delta);
+                    let _ = channel_clone
+                        .update_draft(reply_target, draft_id, &accumulated)
+                        .await;
+                }
+                AgentProgress::ThinkingDelta { .. } => {
+                    if accumulated.is_empty() {
+                        let _ = channel_clone
+                            .update_draft(reply_target, draft_id, "Thinking...")
+                            .await;
+                    }
+                }
+                AgentProgress::ToolCallStarted { tool_name, .. } => {
+                    if accumulated.is_empty() {
+                        let _ = channel_clone
+                            .update_draft(
+                                reply_target,
+                                draft_id,
+                                &format!("Working ({})...", tool_name),
+                            )
+                            .await;
+                    }
+                }
+                _ => {}
+            }
+        }
+    });
+
+    // Simulate thinking then text
+    tx.send(AgentProgress::ThinkingDelta {
+        delta: "thought 1".to_string(),
+        iteration: 1,
+    })
+    .await
+    .unwrap();
+    tx.send(AgentProgress::ThinkingDelta {
+        delta: "thought 2".to_string(),
+        iteration: 1,
+    })
+    .await
+    .unwrap();
+    tx.send(AgentProgress::ToolCallStarted {
+        call_id: "c1".into(),
+        tool_name: "shell".into(),
+        arguments: serde_json::json!({}),
+        iteration: 1,
+    })
+    .await
+    .unwrap();
+    tx.send(AgentProgress::TextDelta {
+        delta: "Hello".to_string(),
+        iteration: 2,
+    })
+    .await
+    .unwrap();
+    drop(tx);
+    handle.await.unwrap();
+
+    let history = updates.lock();
+    assert!(history.contains(&"Thinking...".to_string()));
+    assert!(history.contains(&"Working (shell)...".to_string()));
+    assert!(history.contains(&"Hello".to_string()));
+    // Ensure actual thought text was NOT sent
+    for update in history.iter() {
+        assert!(!update.contains("thought 1"));
+        assert!(!update.contains("thought 2"));
+    }
+}

src/openhuman/channels/runtime/dispatch.rs

@@ -7,6 +7,7 @@ use crate::openhuman::agent::bus::{AgentTurnRequest, AgentTurnResponse, AGENT_RU
 use crate::openhuman::agent::harness::definition::{
     AgentDefinition, AgentDefinitionRegistry, ToolScope,
 };
+use crate::openhuman::agent::progress::AgentProgress;
 use crate::openhuman::channels::context::{
     build_memory_context, compact_sender_history, conversation_history_key,
     conversation_memory_key, is_context_window_overflow_error, ChannelRuntimeContext,
@@ -791,8 +792,8 @@ pub(crate) async fn process_channel_message(
         .is_some_and(|ch| ch.supports_draft_updates());
 
     // Set up streaming channel if supported
-    let (delta_tx, delta_rx) = if use_streaming {
-        let (tx, rx) = tokio::sync::mpsc::channel::<String>(64);
+    let (progress_tx, progress_rx) = if use_streaming {
+        let (tx, rx) = tokio::sync::mpsc::channel::<AgentProgress>(64);
         (Some(tx), Some(rx))
     } else {
         (None, None)
@@ -820,9 +821,9 @@ pub(crate) async fn process_channel_message(
         None
     };
 
-    // Spawn a task to forward streaming deltas to draft updates
+    // Spawn a task to forward streaming progress to draft updates
     let draft_updater = if let (Some(mut rx), Some(draft_id_ref), Some(channel_ref)) = (
-        delta_rx,
+        progress_rx,
         draft_message_id.as_deref(),
         target_channel.as_ref(),
     ) {
@@ -831,13 +832,56 @@ pub(crate) async fn process_channel_message(
         let draft_id = draft_id_ref.to_string();
         Some(tokio::spawn(async move {
             let mut accumulated = String::new();
-            while let Some(delta) = rx.recv().await {
-                accumulated.push_str(&delta);
-                if let Err(e) = channel
-                    .update_draft(&reply_target, &draft_id, &accumulated)
-                    .await
-                {
-                    tracing::debug!("Draft update failed: {e}");
+            let mut last_thinking_update = None;
+            const THINKING_UPDATE_INTERVAL_MS: u128 = 2000;
+
+            while let Some(progress) = rx.recv().await {
+                match progress {
+                    AgentProgress::TextDelta { delta, .. } => {
+                        accumulated.push_str(&delta);
+                        if let Err(e) = channel
+                            .update_draft(&reply_target, &draft_id, &accumulated)
+                            .await
+                        {
+                            tracing::debug!("Draft update failed: {e}");
+                        }
+                    }
+                    AgentProgress::ThinkingDelta { .. } => {
+                        // Suppress thinking text to Telegram; only show a placeholder if we haven't
+                        // started receiving the final answer yet.
+                        if accumulated.is_empty() {
+                            let now = std::time::Instant::now();
+                            let should_update = match last_thinking_update {
+                                None => true,
+                                Some(last) => {
+                                    now.duration_since(last).as_millis()
+                                        > THINKING_UPDATE_INTERVAL_MS
+                                }
+                            };
+
+                            if should_update {
+                                if let Err(e) = channel
+                                    .update_draft(&reply_target, &draft_id, "Thinking...")
+                                    .await
+                                {
+                                    tracing::debug!("Thinking update failed: {e}");
+                                }
+                                last_thinking_update = Some(now);
+                            }
+                        }
+                    }
+                    AgentProgress::ToolCallStarted { tool_name, .. } => {
+                        if accumulated.is_empty() {
+                            let _ = channel
+                                .update_draft(
+                                    &reply_target,
+                                    &draft_id,
+                                    &format!("Working ({})...", tool_name),
+                                )
+                                .await;
+                        }
+                    }
+                    _ => {}
                 }
             }
         }))
@@ -903,11 +947,11 @@ pub(crate) async fn process_channel_message(
         channel_name: msg.channel.clone(),
         multimodal: ctx.multimodal.clone(),
         max_tool_iterations: ctx.max_tool_iterations,
-        on_delta: delta_tx,
+        on_delta: None, // on_progress handles text deltas now
         target_agent_id: scoping.target_agent_id,
         visible_tool_names: scoping.visible_tool_names,
         extra_tools: scoping.extra_tools,
-        on_progress: None,
+        on_progress: progress_tx,
     };
     tracing::debug!(
         channel = %msg.channel,