Webhook and polling integrations for channels

src/channels/wasm/loader.rs

@@ -40,7 +40,7 @@ impl WasmChannelLoader {
         name: &str,
         wasm_path: &Path,
         capabilities_path: Option<&Path>,
-    ) -> Result<WasmChannel, WasmChannelError> {
+    ) -> Result<LoadedChannel, WasmChannelError> {
         // Validate name
         if name.is_empty() || name.contains('/') || name.contains('\\') || name.contains("..") {
             return Err(WasmChannelError::InvalidName(name.to_string()));
@@ -53,56 +53,59 @@ impl WasmChannelLoader {
         let wasm_bytes = fs::read(wasm_path).await?;
 
         // Read capabilities file
-        let (capabilities, config_json, description) = if let Some(cap_path) = capabilities_path {
-            if cap_path.exists() {
-                let cap_bytes = fs::read(cap_path).await?;
-                let cap_file = ChannelCapabilitiesFile::from_bytes(&cap_bytes)
-                    .map_err(|e| WasmChannelError::InvalidCapabilities(e.to_string()))?;
+        let (capabilities, config_json, description, cap_file) =
+            if let Some(cap_path) = capabilities_path {
+                if cap_path.exists() {
+                    let cap_bytes = fs::read(cap_path).await?;
+                    let cap_file = ChannelCapabilitiesFile::from_bytes(&cap_bytes)
+                        .map_err(|e| WasmChannelError::InvalidCapabilities(e.to_string()))?;
 
-                // Debug: log raw capabilities
-                tracing::debug!(
-                    channel = name,
-                    raw_capabilities = ?cap_file.capabilities,
-                    "Parsed capabilities file"
-                );
+                    // Debug: log raw capabilities
+                    tracing::debug!(
+                        channel = name,
+                        raw_capabilities = ?cap_file.capabilities,
+                        "Parsed capabilities file"
+                    );
 
-                let caps = cap_file.to_capabilities();
+                    let caps = cap_file.to_capabilities();
 
-                // Debug: log resulting capabilities
-                tracing::info!(
-                    channel = name,
-                    http_allowed = caps.tool_capabilities.http.is_some(),
-                    http_allowlist_count = caps
-                        .tool_capabilities
-                        .http
-                        .as_ref()
-                        .map(|h| h.allowlist.len())
-                        .unwrap_or(0),
-                    "Channel capabilities loaded"
-                );
+                    // Debug: log resulting capabilities
+                    tracing::info!(
+                        channel = name,
+                        http_allowed = caps.tool_capabilities.http.is_some(),
+                        http_allowlist_count = caps
+                            .tool_capabilities
+                            .http
+                            .as_ref()
+                            .map(|h| h.allowlist.len())
+                            .unwrap_or(0),
+                        "Channel capabilities loaded"
+                    );
 
-                let config = cap_file.config_json();
-                let desc = cap_file.description.clone();
+                    let config = cap_file.config_json();
+                    let desc = cap_file.description.clone();
 
-                (caps, config, desc)
+                    (caps, config, desc, Some(cap_file))
+                } else {
+                    tracing::warn!(
+                        path = %cap_path.display(),
+                        "Capabilities file not found, using defaults"
+                    );
+                    (
+                        ChannelCapabilities::for_channel(name),
+                        "{}".to_string(),
+                        None,
+                        None,
+                    )
+                }
             } else {
-                tracing::warn!(
-                    path = %cap_path.display(),
-                    "Capabilities file not found, using defaults"
-                );
                 (
                     ChannelCapabilities::for_channel(name),
                     "{}".to_string(),
                     None,
+                    None,
                 )
-            }
-        } else {
-            (
-                ChannelCapabilities::for_channel(name),
-                "{}".to_string(),
-                None,
-            )
-        };
+            };
 
         // Prepare the module
         let prepared = self
@@ -119,7 +122,10 @@ impl WasmChannelLoader {
             "Loaded WASM channel from file"
         );
 
-        Ok(channel)
+        Ok(LoadedChannel {
+            channel,
+            capabilities_file: cap_file,
+        })
     }
 
     /// Load all WASM channels from a directory.
@@ -176,8 +182,8 @@ impl WasmChannelLoader {
             };
 
             match self.load_from_files(&name, &path, cap_path_option).await {
-                Ok(channel) => {
-                    results.loaded.push(channel);
+                Ok(loaded) => {
+                    results.loaded.push(loaded);
                 }
                 Err(e) => {
                     tracing::error!(
@@ -194,7 +200,7 @@ impl WasmChannelLoader {
         if !results.loaded.is_empty() {
             tracing::info!(
                 count = results.loaded.len(),
-                channels = ?results.loaded.iter().map(|c| c.channel_name()).collect::<Vec<_>>(),
+                channels = ?results.loaded.iter().map(|c| c.name()).collect::<Vec<_>>(),
                 "Loaded WASM channels from directory"
             );
         }
@@ -203,11 +209,42 @@ impl WasmChannelLoader {
     }
 }
 
+/// A loaded WASM channel with its capabilities file.
+pub struct LoadedChannel {
+    /// The loaded channel.
+    pub channel: WasmChannel,
+
+    /// The parsed capabilities file (if present).
+    pub capabilities_file: Option<ChannelCapabilitiesFile>,
+}
+
+impl LoadedChannel {
+    /// Get the channel name.
+    pub fn name(&self) -> &str {
+        self.channel.channel_name()
+    }
+
+    /// Get the webhook secret header name from capabilities.
+    pub fn webhook_secret_header(&self) -> Option<&str> {
+        self.capabilities_file
+            .as_ref()
+            .and_then(|f| f.webhook_secret_header())
+    }
+
+    /// Get the webhook secret name from capabilities.
+    pub fn webhook_secret_name(&self) -> String {
+        self.capabilities_file
+            .as_ref()
+            .map(|f| f.webhook_secret_name())
+            .unwrap_or_else(|| format!("{}_webhook_secret", self.channel.channel_name()))
+    }
+}
+
 /// Results from loading multiple channels.
 #[derive(Default)]
 pub struct LoadResults {
-    /// Successfully loaded channels.
-    pub loaded: Vec<WasmChannel>,
+    /// Successfully loaded channels with their capabilities.
+    pub loaded: Vec<LoadedChannel>,
 
     /// Errors encountered (path, error).
     pub errors: Vec<(PathBuf, WasmChannelError)>,
@@ -229,9 +266,9 @@ impl LoadResults {
         self.errors.len()
     }
 
-    /// Take ownership of loaded channels.
+    /// Take ownership of loaded channels (extracts just the WasmChannel).
     pub fn take_channels(self) -> Vec<WasmChannel> {
-        self.loaded
+        self.loaded.into_iter().map(|l| l.channel).collect()
     }
 }
 

src/channels/wasm/mod.rs

@@ -92,11 +92,14 @@ pub use capabilities::{ChannelCapabilities, EmitRateLimitConfig, HttpEndpointCon
 pub use error::WasmChannelError;
 pub use host::{ChannelEmitRateLimiter, ChannelHostState, EmittedMessage};
 pub use loader::{
-    DiscoveredChannel, LoadResults, WasmChannelLoader, default_channels_dir, discover_channels,
+    DiscoveredChannel, LoadResults, LoadedChannel, WasmChannelLoader, default_channels_dir,
+    discover_channels,
 };
 pub use router::{
     RegisteredEndpoint, WasmChannelRouter, WasmChannelServer, create_wasm_channel_router,
 };
 pub use runtime::{PreparedChannelModule, WasmChannelRuntime, WasmChannelRuntimeConfig};
-pub use schema::{ChannelCapabilitiesFile, ChannelConfig};
+pub use schema::{
+    ChannelCapabilitiesFile, ChannelConfig, SecretSetupSchema, SetupSchema, WebhookSchema,
+};
 pub use wrapper::{HttpResponse, SharedWasmChannel, WasmChannel};

src/channels/wasm/router.rs

@@ -41,6 +41,8 @@ pub struct WasmChannelRouter {
     path_to_channel: RwLock<HashMap<String, String>>,
     /// Expected webhook secrets by channel name.
     secrets: RwLock<HashMap<String, String>>,
+    /// Webhook secret header names by channel name (e.g., "X-Telegram-Bot-Api-Secret-Token").
+    secret_headers: RwLock<HashMap<String, String>>,
 }
 
 impl WasmChannelRouter {
@@ -50,15 +52,24 @@ impl WasmChannelRouter {
             channels: RwLock::new(HashMap::new()),
             path_to_channel: RwLock::new(HashMap::new()),
             secrets: RwLock::new(HashMap::new()),
+            secret_headers: RwLock::new(HashMap::new()),
         }
     }
 
     /// Register a channel with its endpoints.
+    ///
+    /// # Arguments
+    /// * `channel` - The WASM channel to register
+    /// * `endpoints` - HTTP endpoints to register for this channel
+    /// * `secret` - Optional webhook secret for validation
+    /// * `secret_header` - Optional HTTP header name for secret validation
+    ///   (e.g., "X-Telegram-Bot-Api-Secret-Token"). Defaults to "X-Webhook-Secret".
     pub async fn register(
         &self,
         channel: Arc<WasmChannel>,
         endpoints: Vec<RegisteredEndpoint>,
         secret: Option<String>,
+        secret_header: Option<String>,
     ) {
         let name = channel.channel_name().to_string();
 
@@ -79,14 +90,32 @@ impl WasmChannelRouter {
 
         // Store secret if provided
         if let Some(s) = secret {
-            self.secrets.write().await.insert(name, s);
+            self.secrets.write().await.insert(name.clone(), s);
         }
+
+        // Store secret header if provided
+        if let Some(h) = secret_header {
+            self.secret_headers.write().await.insert(name, h);
+        }
+    }
+
+    /// Get the secret header name for a channel.
+    ///
+    /// Returns the configured header or "X-Webhook-Secret" as default.
+    pub async fn get_secret_header(&self, channel_name: &str) -> String {
+        self.secret_headers
+            .read()
+            .await
+            .get(channel_name)
+            .cloned()
+            .unwrap_or_else(|| "X-Webhook-Secret".to_string())
     }
 
     /// Unregister a channel and its endpoints.
     pub async fn unregister(&self, channel_name: &str) {
         self.channels.write().await.remove(channel_name);
         self.secrets.write().await.remove(channel_name);
+        self.secret_headers.write().await.remove(channel_name);
 
         // Remove all paths for this channel
         self.path_to_channel
@@ -224,23 +253,29 @@ async fn webhook_handler(
 
     // Check if secret is required
     if state.router.requires_secret(channel_name).await {
-        // Try to get secret from query param or header
-        // Telegram uses X-Telegram-Bot-Api-Secret-Token header
+        // Get the secret header name for this channel (from capabilities or default)
+        let secret_header_name = state.router.get_secret_header(channel_name).await;
+
+        // Try to get secret from query param or the channel's configured header
         let provided_secret = query
             .get("secret")
             .cloned()
             .or_else(|| {
                 headers
-                    .get("X-Telegram-Bot-Api-Secret-Token")
+                    .get(&secret_header_name)
                     .and_then(|v| v.to_str().ok())
                     .map(|s| s.to_string())
             })
             .or_else(|| {
-                // Fallback to generic header
-                headers
-                    .get("X-Webhook-Secret")
-                    .and_then(|v| v.to_str().ok())
-                    .map(|s| s.to_string())
+                // Fallback to generic header if different from configured
+                if secret_header_name != "X-Webhook-Secret" {
+                    headers
+                        .get("X-Webhook-Secret")
+                        .and_then(|v| v.to_str().ok())
+                        .map(|s| s.to_string())
+                } else {
+                    None
+                }
             });
 
         tracing::debug!(
@@ -447,7 +482,7 @@ mod tests {
         }];
 
         router
-            .register(channel, endpoints, Some("secret123".to_string()))
+            .register(channel, endpoints, Some("secret123".to_string()), None)
             .await;
 
         // Should find channel by path
@@ -466,7 +501,7 @@ mod tests {
         let channel = create_test_channel("slack");
 
         router
-            .register(channel, vec![], Some("secret123".to_string()))
+            .register(channel, vec![], Some("secret123".to_string()), None)
             .await;
 
         // Correct secret
@@ -477,7 +512,7 @@ mod tests {
 
         // Channel without secret always validates
         let channel2 = create_test_channel("telegram");
-        router.register(channel2, vec![], None).await;
+        router.register(channel2, vec![], None, None).await;
         assert!(router.validate_secret("telegram", "anything").await);
     }
 
@@ -493,7 +528,7 @@ mod tests {
             require_secret: false,
         }];
 
-        router.register(channel, endpoints, None).await;
+        router.register(channel, endpoints, None, None).await;
 
         // Should exist
         assert!(
@@ -522,12 +557,41 @@ mod tests {
         let channel1 = create_test_channel("slack");
         let channel2 = create_test_channel("telegram");
 
-        router.register(channel1, vec![], None).await;
-        router.register(channel2, vec![], None).await;
+        router.register(channel1, vec![], None, None).await;
+        router.register(channel2, vec![], None, None).await;
 
         let channels = router.list_channels().await;
         assert_eq!(channels.len(), 2);
         assert!(channels.contains(&"slack".to_string()));
         assert!(channels.contains(&"telegram".to_string()));
     }
+
+    #[tokio::test]
+    async fn test_router_secret_header() {
+        let router = WasmChannelRouter::new();
+        let channel = create_test_channel("telegram");
+
+        // Register with custom secret header
+        router
+            .register(
+                channel,
+                vec![],
+                Some("secret123".to_string()),
+                Some("X-Telegram-Bot-Api-Secret-Token".to_string()),
+            )
+            .await;
+
+        // Should return the custom header
+        assert_eq!(
+            router.get_secret_header("telegram").await,
+            "X-Telegram-Bot-Api-Secret-Token"
+        );
+
+        // Channel without custom header should use default
+        let channel2 = create_test_channel("slack");
+        router
+            .register(channel2, vec![], Some("secret456".to_string()), None)
+            .await;
+        assert_eq!(router.get_secret_header("slack").await, "X-Webhook-Secret");
+    }
 }

src/channels/wasm/schema.rs

@@ -62,6 +62,10 @@ pub struct ChannelCapabilitiesFile {
     #[serde(default)]
     pub description: Option<String>,
 
+    /// Setup configuration for the wizard.
+    #[serde(default)]
+    pub setup: SetupSchema,
+
     /// Capabilities (tool + channel specific).
     #[serde(default)]
     pub capabilities: ChannelCapabilitiesSchema,
@@ -95,6 +99,29 @@ impl ChannelCapabilitiesFile {
     pub fn config_json(&self) -> String {
         serde_json::to_string(&self.config).unwrap_or_else(|_| "{}".to_string())
     }
+
+    /// Get the webhook secret header name for this channel.
+    ///
+    /// Returns the configured header name from capabilities, or a sensible default.
+    pub fn webhook_secret_header(&self) -> Option<&str> {
+        self.capabilities
+            .channel
+            .as_ref()
+            .and_then(|c| c.webhook.as_ref())
+            .and_then(|w| w.secret_header.as_deref())
+    }
+
+    /// Get the webhook secret name for this channel.
+    ///
+    /// Returns the configured secret name or defaults to "{channel_name}_webhook_secret".
+    pub fn webhook_secret_name(&self) -> String {
+        self.capabilities
+            .channel
+            .as_ref()
+            .and_then(|c| c.webhook.as_ref())
+            .and_then(|w| w.secret_name.clone())
+            .unwrap_or_else(|| format!("{}_webhook_secret", self.name))
+    }
 }
 
 /// Schema for channel capabilities.
@@ -178,6 +205,80 @@ pub struct ChannelSpecificCapabilitiesSchema {
     /// Callback timeout in seconds.
     #[serde(default)]
     pub callback_timeout_secs: Option<u64>,
+
+    /// Webhook configuration (secret header, etc.).
+    #[serde(default)]
+    pub webhook: Option<WebhookSchema>,
+}
+
+/// Webhook configuration schema.
+///
+/// Allows channels to specify their webhook validation requirements.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct WebhookSchema {
+    /// HTTP header name for secret validation.
+    ///
+    /// Examples:
+    /// - Telegram: "X-Telegram-Bot-Api-Secret-Token"
+    /// - Slack: "X-Slack-Signature"
+    /// - GitHub: "X-Hub-Signature-256"
+    /// - Generic: "X-Webhook-Secret"
+    #[serde(default)]
+    pub secret_header: Option<String>,
+
+    /// Secret name in secrets store for webhook validation.
+    /// Default: "{channel_name}_webhook_secret"
+    #[serde(default)]
+    pub secret_name: Option<String>,
+}
+
+/// Setup configuration schema.
+///
+/// Allows channels to declare their setup requirements for the wizard.
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
+pub struct SetupSchema {
+    /// Required secrets that must be configured during setup.
+    #[serde(default)]
+    pub required_secrets: Vec<SecretSetupSchema>,
+
+    /// Optional validation endpoint to verify configuration.
+    /// Placeholders like {secret_name} are replaced with actual values.
+    #[serde(default)]
+    pub validation_endpoint: Option<String>,
+}
+
+/// Configuration for a secret required during setup.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct SecretSetupSchema {
+    /// Secret name in the secrets store (e.g., "telegram_bot_token").
+    pub name: String,
+
+    /// Prompt to show the user during setup.
+    pub prompt: String,
+
+    /// Optional regex for validation.
+    #[serde(default)]
+    pub validation: Option<String>,
+
+    /// Whether this secret is optional.
+    #[serde(default)]
+    pub optional: bool,
+
+    /// Auto-generate configuration if the user doesn't provide a value.
+    #[serde(default)]
+    pub auto_generate: Option<AutoGenerateSchema>,
+}
+
+/// Configuration for auto-generating a secret value.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct AutoGenerateSchema {
+    /// Length of the generated value in bytes (will be hex-encoded).
+    #[serde(default = "default_auto_generate_length")]
+    pub length: usize,
+}
+
+fn default_auto_generate_length() -> usize {
+    32
 }
 
 /// Schema for emit rate limiting.
@@ -412,4 +513,76 @@ mod tests {
         assert_eq!(caps.emit_rate_limit.messages_per_minute, 50);
         assert_eq!(caps.emit_rate_limit.messages_per_hour, 1000);
     }
+
+    #[test]
+    fn test_webhook_schema() {
+        let json = r#"{
+            "name": "telegram",
+            "capabilities": {
+                "channel": {
+                    "allowed_paths": ["/webhook/telegram"],
+                    "webhook": {
+                        "secret_header": "X-Telegram-Bot-Api-Secret-Token",
+                        "secret_name": "telegram_webhook_secret"
+                    }
+                }
+            }
+        }"#;
+
+        let file = ChannelCapabilitiesFile::from_json(json).unwrap();
+        assert_eq!(
+            file.webhook_secret_header(),
+            Some("X-Telegram-Bot-Api-Secret-Token")
+        );
+        assert_eq!(file.webhook_secret_name(), "telegram_webhook_secret");
+    }
+
+    #[test]
+    fn test_webhook_secret_name_default() {
+        let json = r#"{
+            "name": "mybot",
+            "capabilities": {}
+        }"#;
+
+        let file = ChannelCapabilitiesFile::from_json(json).unwrap();
+        assert_eq!(file.webhook_secret_header(), None);
+        assert_eq!(file.webhook_secret_name(), "mybot_webhook_secret");
+    }
+
+    #[test]
+    fn test_setup_schema() {
+        let json = r#"{
+            "name": "telegram",
+            "setup": {
+                "required_secrets": [
+                    {
+                        "name": "telegram_bot_token",
+                        "prompt": "Enter your Telegram Bot Token",
+                        "validation": "^[0-9]+:[A-Za-z0-9_-]+$"
+                    },
+                    {
+                        "name": "telegram_webhook_secret",
+                        "prompt": "Webhook secret (leave empty to auto-generate)",
+                        "optional": true,
+                        "auto_generate": { "length": 64 }
+                    }
+                ],
+                "validation_endpoint": "https://api.telegram.org/bot{telegram_bot_token}/getMe"
+            }
+        }"#;
+
+        let file = ChannelCapabilitiesFile::from_json(json).unwrap();
+        assert_eq!(file.setup.required_secrets.len(), 2);
+        assert_eq!(file.setup.required_secrets[0].name, "telegram_bot_token");
+        assert!(!file.setup.required_secrets[0].optional);
+        assert!(file.setup.required_secrets[1].optional);
+        assert_eq!(
+            file.setup.required_secrets[1]
+                .auto_generate
+                .as_ref()
+                .unwrap()
+                .length,
+            64
+        );
+    }
 }

src/channels/wasm/wrapper.rs

@@ -356,23 +356,29 @@ pub struct WasmChannel {
     channel_config: RwLock<Option<ChannelConfig>>,
 
     /// Message sender (for emitting messages to the stream).
-    message_tx: RwLock<Option<mpsc::Sender<IncomingMessage>>>,
+    /// Wrapped in Arc for sharing with the polling task.
+    message_tx: Arc<RwLock<Option<mpsc::Sender<IncomingMessage>>>>,
 
     /// Pending responses (for synchronous response handling).
     pending_responses: RwLock<HashMap<Uuid, oneshot::Sender<String>>>,
 
     /// Rate limiter for message emission.
-    rate_limiter: RwLock<ChannelEmitRateLimiter>,
+    /// Wrapped in Arc for sharing with the polling task.
+    rate_limiter: Arc<RwLock<ChannelEmitRateLimiter>>,
 
     /// Shutdown signal sender.
     shutdown_tx: RwLock<Option<oneshot::Sender<()>>>,
 
+    /// Polling shutdown signal sender (keeps polling alive while held).
+    poll_shutdown_tx: RwLock<Option<oneshot::Sender<()>>>,
+
     /// Registered HTTP endpoints.
     endpoints: RwLock<Vec<RegisteredEndpoint>>,
 
     /// Injected credentials for HTTP requests (e.g., bot tokens).
     /// Keys are placeholder names like "TELEGRAM_BOT_TOKEN".
-    credentials: RwLock<HashMap<String, String>>,
+    /// Wrapped in Arc for sharing with the polling task.
+    credentials: Arc<RwLock<HashMap<String, String>>>,
 }
 
 impl WasmChannel {
@@ -393,12 +399,13 @@ impl WasmChannel {
             capabilities,
             config_json,
             channel_config: RwLock::new(None),
-            message_tx: RwLock::new(None),
+            message_tx: Arc::new(RwLock::new(None)),
             pending_responses: RwLock::new(HashMap::new()),
-            rate_limiter: RwLock::new(rate_limiter),
+            rate_limiter: Arc::new(RwLock::new(rate_limiter)),
             shutdown_tx: RwLock::new(None),
+            poll_shutdown_tx: RwLock::new(None),
             endpoints: RwLock::new(Vec::new()),
-            credentials: RwLock::new(HashMap::new()),
+            credentials: Arc::new(RwLock::new(HashMap::new())),
         }
     }
 
@@ -430,142 +437,6 @@ impl WasmChannel {
         self.endpoints.read().await.clone()
     }
 
-    /// Register a webhook URL with Telegram.
-    ///
-    /// Called during channel startup if tunnel_url is configured.
-    /// This enables instant message delivery instead of polling.
-    pub async fn register_telegram_webhook(
-        &self,
-        tunnel_url: &str,
-        bot_token: &str,
-        secret_token: Option<&str>,
-    ) -> Result<(), WasmChannelError> {
-        let webhook_url = format!("{}/webhook/telegram", tunnel_url);
-
-        tracing::info!(
-            channel = %self.name,
-            webhook_url = %webhook_url,
-            "Registering Telegram webhook"
-        );
-
-        // Build form parameters
-        let mut form_params = vec![
-            ("url", webhook_url.as_str()),
-            ("allowed_updates", r#"["message","edited_message"]"#),
-        ];
-
-        let secret_owned: String;
-        if let Some(secret) = secret_token {
-            secret_owned = secret.to_string();
-            form_params.push(("secret_token", &secret_owned));
-        }
-
-        let client = reqwest::Client::builder()
-            .timeout(std::time::Duration::from_secs(30))
-            .build()
-            .map_err(|e| WasmChannelError::HttpRequest(e.to_string()))?;
-
-        let response = client
-            .post(format!(
-                "https://api.telegram.org/bot{}/setWebhook",
-                bot_token
-            ))
-            .form(&form_params)
-            .send()
-            .await
-            .map_err(|e| WasmChannelError::HttpRequest(e.to_string()))?;
-
-        if !response.status().is_success() {
-            let status = response.status();
-            let body = response.text().await.unwrap_or_default();
-            return Err(WasmChannelError::WebhookRegistration {
-                name: self.name.clone(),
-                reason: format!("HTTP {}: {}", status, body),
-            });
-        }
-
-        // Parse Telegram API response
-        let result: serde_json::Value = response
-            .json()
-            .await
-            .map_err(|e| WasmChannelError::HttpRequest(e.to_string()))?;
-
-        if result["ok"].as_bool() != Some(true) {
-            let description = result["description"]
-                .as_str()
-                .unwrap_or("unknown error")
-                .to_string();
-            return Err(WasmChannelError::WebhookRegistration {
-                name: self.name.clone(),
-                reason: description,
-            });
-        }
-
-        tracing::info!(
-            channel = %self.name,
-            webhook_url = %webhook_url,
-            "Telegram webhook registered successfully"
-        );
-
-        Ok(())
-    }
-
-    /// Delete the webhook and switch back to polling mode.
-    ///
-    /// Called during shutdown if webhook was registered.
-    pub async fn delete_telegram_webhook(&self, bot_token: &str) -> Result<(), WasmChannelError> {
-        tracing::info!(
-            channel = %self.name,
-            "Deleting Telegram webhook"
-        );
-
-        let client = reqwest::Client::builder()
-            .timeout(std::time::Duration::from_secs(30))
-            .build()
-            .map_err(|e| WasmChannelError::HttpRequest(e.to_string()))?;
-
-        let response = client
-            .post(format!(
-                "https://api.telegram.org/bot{}/deleteWebhook",
-                bot_token
-            ))
-            .send()
-            .await
-            .map_err(|e| WasmChannelError::HttpRequest(e.to_string()))?;
-
-        if !response.status().is_success() {
-            let status = response.status();
-            let body = response.text().await.unwrap_or_default();
-            return Err(WasmChannelError::WebhookRegistration {
-                name: self.name.clone(),
-                reason: format!("HTTP {} (delete): {}", status, body),
-            });
-        }
-
-        let result: serde_json::Value = response
-            .json()
-            .await
-            .map_err(|e| WasmChannelError::HttpRequest(e.to_string()))?;
-
-        if result["ok"].as_bool() != Some(true) {
-            let description = result["description"]
-                .as_str()
-                .unwrap_or("unknown error")
-                .to_string();
-            return Err(WasmChannelError::WebhookRegistration {
-                name: self.name.clone(),
-                reason: format!("delete failed: {}", description),
-            });
-        }
-
-        tracing::info!(
-            channel = %self.name,
-            "Telegram webhook deleted"
-        );
-
-        Ok(())
-    }
-
     /// Add channel host functions to the linker using generated bindings.
     ///
     /// Uses the wasmtime::component::bindgen! generated `add_to_linker` function
@@ -1154,11 +1025,20 @@ impl WasmChannel {
     }
 
     /// Start the polling loop if configured.
+    ///
+    /// Since we can't hold `Arc<Self>` from `&self`, we pass all the components
+    /// needed for polling to a spawned task. Each poll tick creates a fresh WASM
+    /// instance (matching our "fresh instance per callback" pattern).
     fn start_polling(&self, interval: Duration, shutdown_rx: oneshot::Receiver<()>) {
         let channel_name = self.name.clone();
+        let runtime = Arc::clone(&self.runtime);
+        let prepared = Arc::clone(&self.prepared);
+        let capabilities = self.capabilities.clone();
+        let message_tx = self.message_tx.clone();
+        let rate_limiter = self.rate_limiter.clone();
+        let credentials = self.credentials.clone();
+        let callback_timeout = self.runtime.config().callback_timeout;
 
-        // Clone self reference for the async block
-        // In a real implementation, we'd hold an Arc<Self>
         tokio::spawn(async move {
             let mut interval_timer = tokio::time::interval(interval);
             let mut shutdown = std::pin::pin!(shutdown_rx);
@@ -1168,9 +1048,45 @@ impl WasmChannel {
                     _ = interval_timer.tick() => {
                         tracing::debug!(
                             channel = %channel_name,
-                            "Polling tick (stub - would call on_poll)"
+                            "Polling tick - calling on_poll"
                         );
-                        // In real implementation: self.call_on_poll().await
+
+                        // Execute on_poll with fresh WASM instance
+                        let result = Self::execute_poll(
+                            &channel_name,
+                            &runtime,
+                            &prepared,
+                            &capabilities,
+                            &credentials,
+                            callback_timeout,
+                        ).await;
+
+                        match result {
+                            Ok(emitted_messages) => {
+                                // Process any emitted messages
+                                if !emitted_messages.is_empty() {
+                                    if let Err(e) = Self::dispatch_emitted_messages(
+                                        &channel_name,
+                                        emitted_messages,
+                                        &message_tx,
+                                        &rate_limiter,
+                                    ).await {
+                                        tracing::warn!(
+                                            channel = %channel_name,
+                                            error = %e,
+                                            "Failed to dispatch emitted messages from poll"
+                                        );
+                                    }
+                                }
+                            }
+                            Err(e) => {
+                                tracing::warn!(
+                                    channel = %channel_name,
+                                    error = %e,
+                                    "Polling callback failed"
+                                );
+                            }
+                        }
                     }
                     _ = &mut shutdown => {
                         tracing::info!(
@@ -1183,6 +1099,156 @@ impl WasmChannel {
             }
         });
     }
+
+    /// Execute a single poll callback with a fresh WASM instance.
+    ///
+    /// Returns any emitted messages from the callback.
+    async fn execute_poll(
+        channel_name: &str,
+        runtime: &Arc<WasmChannelRuntime>,
+        prepared: &Arc<PreparedChannelModule>,
+        capabilities: &ChannelCapabilities,
+        credentials: &RwLock<HashMap<String, String>>,
+        timeout: Duration,
+    ) -> Result<Vec<EmittedMessage>, WasmChannelError> {
+        // Skip if no WASM bytes (testing mode)
+        if prepared.component_bytes.is_empty() {
+            tracing::debug!(
+                channel = %channel_name,
+                "WASM channel on_poll called (no WASM module)"
+            );
+            return Ok(Vec::new());
+        }
+
+        let runtime = Arc::clone(runtime);
+        let prepared = Arc::clone(prepared);
+        let capabilities = capabilities.clone();
+        let credentials_snapshot = credentials.read().await.clone();
+        let channel_name_owned = channel_name.to_string();
+
+        // Execute in blocking task with timeout
+        let result = tokio::time::timeout(timeout, async move {
+            tokio::task::spawn_blocking(move || {
+                let mut store =
+                    Self::create_store(&runtime, &prepared, &capabilities, credentials_snapshot)?;
+                let instance = Self::instantiate_component(&runtime, &prepared, &mut store)?;
+
+                // Call on_poll using the generated typed interface
+                let channel_iface = instance.near_agent_channel();
+                channel_iface
+                    .call_on_poll(&mut store)
+                    .map_err(|e| Self::map_wasm_error(e, &prepared.name, prepared.limits.fuel))?;
+
+                let host_state =
+                    Self::extract_host_state(&mut store, &prepared.name, &capabilities);
+                Ok(host_state)
+            })
+            .await
+            .map_err(|e| WasmChannelError::ExecutionPanicked {
+                name: channel_name_owned.clone(),
+                reason: e.to_string(),
+            })?
+        })
+        .await;
+
+        match result {
+            Ok(Ok(mut host_state)) => {
+                let emitted = host_state.take_emitted_messages();
+                tracing::debug!(
+                    channel = %channel_name,
+                    emitted_count = emitted.len(),
+                    "WASM channel on_poll completed"
+                );
+                Ok(emitted)
+            }
+            Ok(Err(e)) => Err(e),
+            Err(_) => Err(WasmChannelError::Timeout {
+                name: channel_name.to_string(),
+                callback: "on_poll".to_string(),
+            }),
+        }
+    }
+
+    /// Dispatch emitted messages to the message channel.
+    ///
+    /// This is a static helper used by the polling loop since it doesn't have
+    /// access to `&self`.
+    async fn dispatch_emitted_messages(
+        channel_name: &str,
+        messages: Vec<EmittedMessage>,
+        message_tx: &RwLock<Option<mpsc::Sender<IncomingMessage>>>,
+        rate_limiter: &RwLock<ChannelEmitRateLimiter>,
+    ) -> Result<(), WasmChannelError> {
+        tracing::info!(
+            channel = %channel_name,
+            message_count = messages.len(),
+            "Processing emitted messages from polling callback"
+        );
+
+        let tx_guard = message_tx.read().await;
+        let Some(tx) = tx_guard.as_ref() else {
+            tracing::error!(
+                channel = %channel_name,
+                count = messages.len(),
+                "Messages emitted but no sender available - channel may not be started!"
+            );
+            return Ok(());
+        };
+
+        let mut limiter = rate_limiter.write().await;
+
+        for emitted in messages {
+            // Check rate limit
+            if !limiter.check_and_record() {
+                tracing::warn!(
+                    channel = %channel_name,
+                    "Message emission rate limited"
+                );
+                return Err(WasmChannelError::EmitRateLimited {
+                    name: channel_name.to_string(),
+                });
+            }
+
+            // Convert to IncomingMessage
+            let mut msg = IncomingMessage::new(channel_name, &emitted.user_id, &emitted.content);
+
+            if let Some(name) = emitted.user_name {
+                msg = msg.with_user_name(name);
+            }
+
+            if let Some(thread_id) = emitted.thread_id {
+                msg = msg.with_thread(thread_id);
+            }
+
+            // Parse metadata JSON
+            if let Ok(metadata) = serde_json::from_str(&emitted.metadata_json) {
+                msg = msg.with_metadata(metadata);
+            }
+
+            // Send to stream
+            tracing::info!(
+                channel = %channel_name,
+                user_id = %emitted.user_id,
+                content_len = emitted.content.len(),
+                "Sending polled message to agent"
+            );
+
+            if tx.send(msg).await.is_err() {
+                tracing::error!(
+                    channel = %channel_name,
+                    "Failed to send polled message, channel closed"
+                );
+                break;
+            }
+
+            tracing::info!(
+                channel = %channel_name,
+                "Message successfully sent to agent queue"
+            );
+        }
+
+        Ok(())
+    }
 }
 
 #[async_trait]
@@ -1245,8 +1311,9 @@ impl Channel for WasmChannel {
                         reason: e,
                     })?;
 
-                // Create a new shutdown receiver for polling
-                let (_poll_shutdown_tx, poll_shutdown_rx) = oneshot::channel();
+                // Create shutdown channel for polling and store the sender to keep it alive
+                let (poll_shutdown_tx, poll_shutdown_rx) = oneshot::channel();
+                *self.poll_shutdown_tx.write().await = Some(poll_shutdown_tx);
 
                 self.start_polling(Duration::from_millis(interval as u64), poll_shutdown_rx);
             }
@@ -1316,6 +1383,9 @@ impl Channel for WasmChannel {
             let _ = tx.send(());
         }
 
+        // Stop polling by dropping the sender (receiver will complete)
+        let _ = self.poll_shutdown_tx.write().await.take();
+
         // Clear the message sender
         *self.message_tx.write().await = None;
 
@@ -1558,4 +1628,151 @@ mod tests {
         // Health check should fail after shutdown
         assert!(channel.health_check().await.is_err());
     }
+
+    #[tokio::test]
+    async fn test_execute_poll_no_wasm_returns_empty() {
+        // When there's no WASM module (empty component_bytes), execute_poll
+        // should return an empty vector of messages
+        let config = WasmChannelRuntimeConfig::for_testing();
+        let runtime = Arc::new(WasmChannelRuntime::new(config).unwrap());
+
+        let prepared = Arc::new(PreparedChannelModule {
+            name: "poll-test".to_string(),
+            description: "Test channel".to_string(),
+            component_bytes: Vec::new(), // No WASM bytes
+            limits: ResourceLimits::default(),
+        });
+
+        let capabilities = ChannelCapabilities::for_channel("poll-test").with_polling(1000);
+        let credentials = Arc::new(tokio::sync::RwLock::new(std::collections::HashMap::new()));
+        let timeout = std::time::Duration::from_secs(5);
+
+        let result = WasmChannel::execute_poll(
+            "poll-test",
+            &runtime,
+            &prepared,
+            &capabilities,
+            &credentials,
+            timeout,
+        )
+        .await;
+
+        assert!(result.is_ok());
+        assert!(result.unwrap().is_empty());
+    }
+
+    #[tokio::test]
+    async fn test_dispatch_emitted_messages_sends_to_channel() {
+        use crate::channels::wasm::host::EmittedMessage;
+
+        let (tx, mut rx) = tokio::sync::mpsc::channel(10);
+        let message_tx = Arc::new(tokio::sync::RwLock::new(Some(tx)));
+
+        let rate_limiter = Arc::new(tokio::sync::RwLock::new(
+            crate::channels::wasm::host::ChannelEmitRateLimiter::new(
+                crate::channels::wasm::capabilities::EmitRateLimitConfig::default(),
+            ),
+        ));
+
+        let messages = vec![
+            EmittedMessage::new("user1", "Hello from polling!"),
+            EmittedMessage::new("user2", "Another message"),
+        ];
+
+        let result = WasmChannel::dispatch_emitted_messages(
+            "test-channel",
+            messages,
+            &message_tx,
+            &rate_limiter,
+        )
+        .await;
+
+        assert!(result.is_ok());
+
+        // Verify messages were sent
+        let msg1 = rx.try_recv().expect("Should receive first message");
+        assert_eq!(msg1.user_id, "user1");
+        assert_eq!(msg1.content, "Hello from polling!");
+
+        let msg2 = rx.try_recv().expect("Should receive second message");
+        assert_eq!(msg2.user_id, "user2");
+        assert_eq!(msg2.content, "Another message");
+
+        // No more messages
+        assert!(rx.try_recv().is_err());
+    }
+
+    #[tokio::test]
+    async fn test_dispatch_emitted_messages_no_sender_returns_ok() {
+        use crate::channels::wasm::host::EmittedMessage;
+
+        // No sender available (channel not started)
+        let message_tx = Arc::new(tokio::sync::RwLock::new(None));
+        let rate_limiter = Arc::new(tokio::sync::RwLock::new(
+            crate::channels::wasm::host::ChannelEmitRateLimiter::new(
+                crate::channels::wasm::capabilities::EmitRateLimitConfig::default(),
+            ),
+        ));
+
+        let messages = vec![EmittedMessage::new("user1", "Hello!")];
+
+        // Should return Ok even without a sender (logs warning but doesn't fail)
+        let result = WasmChannel::dispatch_emitted_messages(
+            "test-channel",
+            messages,
+            &message_tx,
+            &rate_limiter,
+        )
+        .await;
+
+        assert!(result.is_ok());
+    }
+
+    #[tokio::test]
+    async fn test_channel_with_polling_stores_shutdown_sender() {
+        // Create a channel with polling capabilities
+        let config = WasmChannelRuntimeConfig::for_testing();
+        let runtime = Arc::new(WasmChannelRuntime::new(config).unwrap());
+
+        let prepared = Arc::new(PreparedChannelModule {
+            name: "poll-channel".to_string(),
+            description: "Polling test channel".to_string(),
+            component_bytes: Vec::new(),
+            limits: ResourceLimits::default(),
+        });
+
+        // Enable polling with a 1 second minimum interval
+        let capabilities = ChannelCapabilities::for_channel("poll-channel")
+            .with_path("/webhook/poll")
+            .with_polling(1000);
+
+        let channel = WasmChannel::new(runtime, prepared, capabilities, "{}".to_string());
+
+        // Start the channel
+        let _stream = channel.start().await.expect("Channel should start");
+
+        // Verify poll_shutdown_tx is set (polling was started)
+        // Note: For testing channels without WASM, on_start returns no poll config,
+        // so polling won't actually be started. This verifies the basic lifecycle.
+        assert!(channel.health_check().await.is_ok());
+
+        // Shutdown should clean up properly
+        channel.shutdown().await.expect("Shutdown should succeed");
+        assert!(channel.health_check().await.is_err());
+    }
+
+    #[tokio::test]
+    async fn test_call_on_poll_no_wasm_succeeds() {
+        // Verify call_on_poll returns Ok when there's no WASM module
+        let channel = create_test_channel();
+
+        // Start the channel first to set up message_tx
+        let _stream = channel.start().await.expect("Channel should start");
+
+        // call_on_poll should succeed (no-op for no WASM)
+        let result = channel.call_on_poll().await;
+        assert!(result.is_ok());
+
+        channel.shutdown().await.expect("Shutdown should succeed");
+    }
 }