diff --git a/src/openhuman/channels/providers/telegram/channel_tests.rs b/src/openhuman/channels/providers/telegram/channel_tests.rs index 76ca42977..9066e5b75 100644 --- a/src/openhuman/channels/providers/telegram/channel_tests.rs +++ b/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>>, + } + + #[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, + ) -> anyhow::Result<()> { + Ok(()) + } + async fn send_draft(&self, _: &SendMessage) -> anyhow::Result> { + 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::(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")); + } +} diff --git a/src/openhuman/channels/runtime/dispatch.rs b/src/openhuman/channels/runtime/dispatch.rs index 07397169a..18edab68f 100644 --- a/src/openhuman/channels/runtime/dispatch.rs +++ b/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::(64); + let (progress_tx, progress_rx) = if use_streaming { + let (tx, rx) = tokio::sync::mpsc::channel::(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,