fix: channel reply stream cold start (#366)

## Summary
- Warm channel conversations before subscribing to the agent stream so
cold-start external replies can relay back to Telegram, Lark, DingTalk,
and Weixin.
- Add regression coverage for the runtime channel cold-task path.
- Keep Telegram credential checks from starting long-running polling
during plugin tests.

## Verification
- `just push -u origin aio-9-channel-reply-stream`
- `cargo nextest run --workspace` via `just push`: 5706 passed, 18
skipped

Co-authored-by: zynx <>

Author: piorpua

crates/aionui-channel/Cargo.toml

@@ -36,3 +36,6 @@ rustls = { workspace = true, optional = true }
 rustls-native-certs = { workspace = true, optional = true }
 base64 = { workspace = true, optional = true }
 uuid = { workspace = true, optional = true }
+
+[dev-dependencies]
+aionui-ai-agent = { workspace = true, features = ["test-support"] }

crates/aionui-channel/src/message_service.rs

@@ -46,8 +46,9 @@ impl ChannelMessageService {
     /// Sends a text message from a channel user to the AI agent.
     ///
     /// 1. Ensures the session has a backing conversation (creates one if needed)
-    /// 2. Sends the message via ConversationService
-    /// 3. Returns the conversation_id for stream subscription
+    /// 2. Warms up the backing agent task so stream subscription is available
+    /// 3. Sends the message via ConversationService
+    /// 4. Returns the conversation_id and stream receiver for relay
     ///
     /// The caller is responsible for subscribing to stream events and
     /// relaying them to the IM platform.
@@ -75,30 +76,40 @@ impl ChannelMessageService {
         };
 
         let user_id = &self.owner_user_id;
+        // Channel relays need a stream subscription before the agent starts
+        // emitting. `ConversationService::send_message` returns immediately
+        // and builds cold agents in the background, so warm the conversation
+        // explicitly for channel traffic.
         self.conversation_svc
-            .send_message(user_id, &conversation_id, req, &self.task_manager)
+            .warmup(user_id, &conversation_id, &self.task_manager)
             .await
             .map_err(|e| ChannelError::MessageSendFailed(e.to_string()))?;
 
-        // Subscribe to the agent's broadcast channel for the ChannelStreamRelay.
-        // The agent task exists because send_message just called get_or_build_task.
-        // ConversationService spawns agent.send_message in a background task,
-        // so the first events have not been emitted yet — no race condition.
         let stream_rx = self
             .task_manager
             .get_task(&conversation_id)
-            .map(|handle| handle.subscribe());
+            .map(|handle| handle.subscribe())
+            .ok_or_else(|| {
+                ChannelError::MessageSendFailed(format!(
+                    "Agent task missing after warmup for conversation {conversation_id}"
+                ))
+            })?;
+
+        self.conversation_svc
+            .send_message(user_id, &conversation_id, req, &self.task_manager)
+            .await
+            .map_err(|e| ChannelError::MessageSendFailed(e.to_string()))?;
 
         info!(
             conversation_id = %conversation_id,
             session_id = %session.id,
-            has_stream = stream_rx.is_some(),
+            has_stream = true,
             "message sent to agent"
         );
 
         Ok(SendResult {
             conversation_id,
-            stream_rx,
+            stream_rx: Some(stream_rx),
         })
     }
 

crates/aionui-channel/src/plugin.rs

@@ -9,6 +9,7 @@ use crate::types::{BotInfo, PluginConfig, PluginStatus, PluginType, UnifiedIncom
 ///
 /// This addresses M-63 — the API Spec `BasePlugin.onMessage/onConfirm`
 /// callbacks are mapped to channel-based injection.
+#[derive(Clone)]
 pub struct PluginCallbacks {
     /// Sender for incoming messages from the platform.
     pub message_tx: tokio::sync::mpsc::Sender<UnifiedIncomingMessage>,

crates/aionui-channel/src/plugins/telegram/plugin.rs

@@ -32,6 +32,7 @@ pub struct TelegramPlugin {
     bot_info: Option<BotInfo>,
     last_error: Option<String>,
     api: Option<Arc<TelegramApi>>,
+    callbacks: Option<PluginCallbacks>,
     poll_handle: Option<JoinHandle<()>>,
     shutdown_tx: Option<watch::Sender<bool>>,
 }
@@ -43,6 +44,7 @@ impl Default for TelegramPlugin {
             bot_info: None,
             last_error: None,
             api: None,
+            callbacks: None,
             poll_handle: None,
             shutdown_tx: None,
         }
@@ -102,27 +104,38 @@ impl ChannelPlugin for TelegramPlugin {
         );
 
         self.api = Some(api);
-        // Store callbacks in a shared container for the polling task
+        self.callbacks = Some(callbacks);
+        self.status = PluginStatus::Ready;
+        Ok(())
+    }
+
+    async fn start(&mut self) -> Result<(), ChannelError> {
+        self.status = PluginStatus::Starting;
+
+        if self.poll_handle.is_some() {
+            self.status = PluginStatus::Running;
+            return Ok(());
+        }
+
+        let api = self
+            .api
+            .as_ref()
+            .cloned()
+            .ok_or_else(|| ChannelError::PlatformApi("Telegram plugin not initialized".into()))?;
+        let callbacks = self
+            .callbacks
+            .clone()
+            .ok_or_else(|| ChannelError::PlatformApi("Telegram callbacks not initialized".into()))?;
+
         let (shutdown_tx, shutdown_rx) = watch::channel(false);
         self.shutdown_tx = Some(shutdown_tx);
-
-        // Spawn the long-polling task
-        let api_clone = Arc::clone(self.api.as_ref().expect("api just set"));
         self.poll_handle = Some(tokio::spawn(poll_loop(
-            api_clone,
+            api,
             callbacks.message_tx,
             callbacks.confirm_tx,
             shutdown_rx,
         )));
 
-        self.status = PluginStatus::Ready;
-        Ok(())
-    }
-
-    async fn start(&mut self) -> Result<(), ChannelError> {
-        self.status = PluginStatus::Starting;
-        // The polling task was already spawned in initialize;
-        // `start` just transitions the status.
         self.status = PluginStatus::Running;
         info!("Telegram plugin started");
         Ok(())
@@ -143,6 +156,7 @@ impl ChannelPlugin for TelegramPlugin {
         }
 
         self.api = None;
+        self.callbacks = None;
         self.status = PluginStatus::Stopped;
         info!("Telegram plugin stopped");
         Ok(())

crates/aionui-channel/tests/manager_integration.rs

@@ -5,6 +5,7 @@
 //! TP-1..TP-5, CS-1..CS-2, WS-2.
 
 use std::collections::HashMap;
+use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::{Arc, Mutex};
 
 use aionui_api_types::WebSocketMessage;
@@ -51,6 +52,7 @@ struct MockPlugin {
     bot_info: Option<BotInfo>,
     last_error: Option<String>,
     should_fail_init: bool,
+    start_calls: Arc<AtomicUsize>,
 }
 
 impl MockPlugin {
@@ -61,6 +63,7 @@ impl MockPlugin {
             bot_info: None,
             last_error: None,
             should_fail_init: false,
+            start_calls: Arc::new(AtomicUsize::new(0)),
         }
     }
 
@@ -91,6 +94,7 @@ impl ChannelPlugin for MockPlugin {
     }
 
     async fn start(&mut self) -> Result<(), ChannelError> {
+        self.start_calls.fetch_add(1, Ordering::SeqCst);
         self.status = PluginStatus::Starting;
         self.status = PluginStatus::Running;
         Ok(())
@@ -164,6 +168,17 @@ fn make_no_impl_factory() -> PluginFactory {
     Box::new(|_pt| None)
 }
 
+fn make_counting_factory() -> (PluginFactory, Arc<AtomicUsize>) {
+    let start_calls = Arc::new(AtomicUsize::new(0));
+    let captured = Arc::clone(&start_calls);
+    let factory = Box::new(move |pt| {
+        let mut plugin = MockPlugin::new(pt);
+        plugin.start_calls = Arc::clone(&captured);
+        Some(Box::new(plugin) as Box<dyn ChannelPlugin>)
+    });
+    (factory, start_calls)
+}
+
 fn make_telegram_config() -> serde_json::Value {
     serde_json::json!({
         "credentials": { "token": "bot:valid123" },
@@ -372,6 +387,24 @@ async fn tp1_test_valid_credentials() {
     assert_eq!(result.as_deref(), Some("mock_bot_user"));
 }
 
+#[tokio::test]
+async fn test_plugin_initializes_without_starting_runtime() {
+    let (mgr, _repo, _bc) = setup().await;
+    let (factory, start_calls) = make_counting_factory();
+
+    let username = mgr
+        .test_plugin("telegram", make_plugin_config(), &factory)
+        .await
+        .unwrap();
+
+    assert_eq!(username.as_deref(), Some("mock_bot_user"));
+    assert_eq!(
+        start_calls.load(Ordering::SeqCst),
+        0,
+        "credential tests must not start long-running plugin runtime"
+    );
+}
+
 // ── TP-2: Test invalid credentials propagates error ───────────────
 
 #[tokio::test]