diff --git a/.github/workflows/e2e.yml b/.github/workflows/e2e.yml index ee16c0f8..5b20345e 100644 --- a/.github/workflows/e2e.yml +++ b/.github/workflows/e2e.yml @@ -5,6 +5,8 @@ on: - cron: "0 6 * * 1" # Weekly Monday 6 AM UTC workflow_dispatch: pull_request: + branches: + - main paths: - "src/channels/web/**" - "tests/e2e/**" @@ -50,9 +52,11 @@ jobs: - group: core files: "tests/e2e/scenarios/test_connection.py tests/e2e/scenarios/test_chat.py tests/e2e/scenarios/test_sse_reconnect.py tests/e2e/scenarios/test_html_injection.py tests/e2e/scenarios/test_csp.py" - group: features - files: "tests/e2e/scenarios/test_skills.py tests/e2e/scenarios/test_tool_approval.py" + files: "tests/e2e/scenarios/test_skills.py tests/e2e/scenarios/test_tool_approval.py tests/e2e/scenarios/test_webhook.py" - group: extensions - files: "tests/e2e/scenarios/test_extensions.py tests/e2e/scenarios/test_extension_oauth.py tests/e2e/scenarios/test_telegram_token_validation.py tests/e2e/scenarios/test_wasm_lifecycle.py tests/e2e/scenarios/test_tool_execution.py tests/e2e/scenarios/test_pairing.py tests/e2e/scenarios/test_oauth_credential_fallback.py tests/e2e/scenarios/test_routine_oauth_credential_injection.py" + files: "tests/e2e/scenarios/test_extensions.py tests/e2e/scenarios/test_extension_oauth.py tests/e2e/scenarios/test_telegram_token_validation.py tests/e2e/scenarios/test_telegram_hot_activation.py tests/e2e/scenarios/test_wasm_lifecycle.py tests/e2e/scenarios/test_tool_execution.py tests/e2e/scenarios/test_pairing.py tests/e2e/scenarios/test_mcp_auth_flow.py tests/e2e/scenarios/test_oauth_credential_fallback.py tests/e2e/scenarios/test_routine_oauth_credential_injection.py" + - group: routines + files: "tests/e2e/scenarios/test_owner_scope.py tests/e2e/scenarios/test_routine_event_batch.py" steps: - uses: actions/checkout@v6 diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index c3ceb8b6..7946c353 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -17,7 +17,7 @@ jobs: matrix: include: - name: all-features - flags: "--features postgres,libsql,html-to-markdown" + flags: "--all-features" - name: default flags: "" - name: libsql-only diff --git a/registry/channels/discord.json b/registry/channels/discord.json index 6f5cd4e7..50ef85ee 100644 --- a/registry/channels/discord.json +++ b/registry/channels/discord.json @@ -2,7 +2,7 @@ "name": "discord", "display_name": "Discord Channel", "kind": "channel", - "version": "0.2.0", + "version": "0.2.1", "wit_version": "0.3.0", "description": "Talk to your agent in Discord", "keywords": [ diff --git a/registry/tools/github.json b/registry/tools/github.json index e84f756d..e775ac82 100644 --- a/registry/tools/github.json +++ b/registry/tools/github.json @@ -2,7 +2,7 @@ "name": "github", "display_name": "GitHub", "kind": "tool", - "version": "0.2.0", + "version": "0.2.1", "wit_version": "0.3.0", "description": "GitHub integration for issues, PRs, repos, and code search", "keywords": [ diff --git a/registry/tools/web-search.json b/registry/tools/web-search.json index 4da5744b..1722c391 100644 --- a/registry/tools/web-search.json +++ b/registry/tools/web-search.json @@ -2,7 +2,7 @@ "name": "web-search", "display_name": "Web Search", "kind": "tool", - "version": "0.2.0", + "version": "0.2.1", "wit_version": "0.3.0", "description": "Search the web using Brave Search API", "keywords": [ diff --git a/src/agent/dispatcher.rs b/src/agent/dispatcher.rs index 4301c09a..9be0d654 100644 --- a/src/agent/dispatcher.rs +++ b/src/agent/dispatcher.rs @@ -148,6 +148,7 @@ impl Agent { "notify_channel": message.channel, "notify_user": message.user_id, "notify_thread_id": message.thread_id, + "notify_metadata": message.metadata, }); // Build system prompts once for this turn. Two variants: with tools diff --git a/src/config/llm.rs b/src/config/llm.rs index 4ad24399..64bf4ab8 100644 --- a/src/config/llm.rs +++ b/src/config/llm.rs @@ -38,6 +38,8 @@ impl LlmConfig { provider: None, bedrock: None, request_timeout_secs: 120, + cheap_model: None, + smart_routing_cascade: false, } } @@ -168,6 +170,14 @@ impl LlmConfig { let request_timeout_secs = parse_optional_env("LLM_REQUEST_TIMEOUT_SECS", 120)?; + // Generic cheap model (works with any backend). + // Falls back to NearAI-specific cheap_model in provider chain logic. + let cheap_model = optional_env("LLM_CHEAP_MODEL")?; + + // Generic smart routing cascade flag. + // Defaults to true. Overrides NearAI-specific smart_routing_cascade. + let smart_routing_cascade = parse_optional_env("SMART_ROUTING_CASCADE", true)?; + Ok(Self { backend: if is_nearai { "nearai".to_string() @@ -183,6 +193,8 @@ impl LlmConfig { provider, bedrock, request_timeout_secs, + cheap_model, + smart_routing_cascade, }) } diff --git a/src/config/transcription.rs b/src/config/transcription.rs index b0f76066..da2bac25 100644 --- a/src/config/transcription.rs +++ b/src/config/transcription.rs @@ -9,11 +9,15 @@ use crate::settings::Settings; pub struct TranscriptionConfig { /// Whether audio transcription is enabled. pub enabled: bool, - /// Provider: "openai" (default). + /// Provider: "openai" (default) or "chat_completions". pub provider: String, /// OpenAI API key (reuses OPENAI_API_KEY). pub openai_api_key: Option, - /// Model to use (default: "whisper-1"). + /// Explicit transcription API key (overrides provider-specific keys). + pub api_key: Option, + /// LLM API key (reuses LLM_API_KEY, used as fallback for chat_completions). + pub llm_api_key: Option, + /// Model to use (default depends on provider). pub model: String, /// Base URL override for the transcription API. pub base_url: Option, @@ -25,6 +29,8 @@ impl Default for TranscriptionConfig { enabled: false, provider: "openai".to_string(), openai_api_key: None, + api_key: None, + llm_api_key: None, model: "whisper-1".to_string(), base_url: None, } @@ -42,8 +48,15 @@ impl TranscriptionConfig { optional_env("TRANSCRIPTION_PROVIDER")?.unwrap_or_else(|| "openai".to_string()); let openai_api_key = optional_env("OPENAI_API_KEY")?.map(SecretString::from); + let api_key = optional_env("TRANSCRIPTION_API_KEY")?.map(SecretString::from); + let llm_api_key = optional_env("LLM_API_KEY")?.map(SecretString::from); - let model = optional_env("TRANSCRIPTION_MODEL")?.unwrap_or_else(|| "whisper-1".to_string()); + let default_model = match provider.as_str() { + "chat_completions" => "google/gemini-2.0-flash-001", + _ => "whisper-1", + }; + let model = + optional_env("TRANSCRIPTION_MODEL")?.unwrap_or_else(|| default_model.to_string()); let base_url = optional_env("TRANSCRIPTION_BASE_URL")?; @@ -51,29 +64,67 @@ impl TranscriptionConfig { enabled, provider, openai_api_key, + api_key, + llm_api_key, model, base_url, }) } + /// Resolve the API key for the configured provider. + /// + /// Priority: `TRANSCRIPTION_API_KEY` > provider-specific key. + fn resolve_api_key(&self) -> Option<&SecretString> { + self.api_key + .as_ref() + .or_else(|| match self.provider.as_str() { + "chat_completions" => self.llm_api_key.as_ref().or(self.openai_api_key.as_ref()), + _ => self.openai_api_key.as_ref(), + }) + } + /// Create the transcription provider if enabled and configured. pub fn create_provider(&self) -> Option> { if !self.enabled { return None; } - // Currently only OpenAI Whisper is supported; more providers can be - // added here with a match on self.provider. - let api_key = self.openai_api_key.as_ref()?; - tracing::info!(model = %self.model, "Audio transcription enabled via OpenAI Whisper"); + let api_key = self.resolve_api_key()?; - let mut provider = crate::transcription::OpenAiWhisperProvider::new(api_key.clone()) - .with_model(&self.model); + match self.provider.as_str() { + "chat_completions" => { + tracing::info!( + model = %self.model, + "Audio transcription enabled via Chat Completions API" + ); - if let Some(ref base_url) = self.base_url { - provider = provider.with_base_url(base_url); + let mut provider = crate::transcription::ChatCompletionsTranscriptionProvider::new( + api_key.clone(), + ) + .with_model(&self.model); + + if let Some(ref base_url) = self.base_url { + provider = provider.with_base_url(base_url); + } + + Some(Box::new(provider)) + } + _ => { + tracing::info!( + model = %self.model, + "Audio transcription enabled via OpenAI Whisper" + ); + + let mut provider = + crate::transcription::OpenAiWhisperProvider::new(api_key.clone()) + .with_model(&self.model); + + if let Some(ref base_url) = self.base_url { + provider = provider.with_base_url(base_url); + } + + Some(Box::new(provider)) + } } - - Some(Box::new(provider)) } } diff --git a/src/db/tls.rs b/src/db/tls.rs index e612704f..bbcb6c6f 100644 --- a/src/db/tls.rs +++ b/src/db/tls.rs @@ -5,13 +5,22 @@ //! certificates — the same TLS stack that `reqwest` already uses for HTTP. use deadpool_postgres::{Pool, Runtime}; +use thiserror::Error; use tokio_postgres::NoTls; use tokio_postgres_rustls::MakeRustlsConnect; use crate::config::SslMode; +#[derive(Debug, Error)] +pub enum CreatePoolError { + #[error("{0}")] + Pool(#[from] deadpool_postgres::CreatePoolError), + #[error("postgres TLS configuration failed: {0}")] + TlsConfig(#[from] rustls::Error), +} + /// Build a rustls-based TLS connector using the platform's root certificate store. -fn make_rustls_connector() -> MakeRustlsConnect { +fn make_rustls_connector() -> Result { let mut root_store = rustls::RootCertStore::empty(); let native = rustls_native_certs::load_native_certs(); for e in &native.errors { @@ -25,10 +34,15 @@ fn make_rustls_connector() -> MakeRustlsConnect { if root_store.is_empty() { tracing::error!("no system root certificates found -- TLS connections will fail"); } - let config = rustls::ClientConfig::builder() - .with_root_certificates(root_store) - .with_no_client_auth(); - MakeRustlsConnect::new(config) + // `--all-features` brings in both aws-lc-rs and ring-backed rustls providers. + // Pick the same ring provider reqwest already uses so postgres TLS setup stays deterministic. + let config = rustls::ClientConfig::builder_with_provider( + rustls::crypto::ring::default_provider().into(), + ) + .with_safe_default_protocol_versions()? + .with_root_certificates(root_store) + .with_no_client_auth(); + Ok(MakeRustlsConnect::new(config)) } /// Create a [`deadpool_postgres::Pool`] with the appropriate TLS connector. @@ -45,12 +59,16 @@ fn make_rustls_connector() -> MakeRustlsConnect { pub fn create_pool( config: &deadpool_postgres::Config, ssl_mode: SslMode, -) -> Result { +) -> Result { match ssl_mode { - SslMode::Disable => config.create_pool(Some(Runtime::Tokio1), NoTls), + SslMode::Disable => config + .create_pool(Some(Runtime::Tokio1), NoTls) + .map_err(CreatePoolError::from), SslMode::Prefer | SslMode::Require => { - let tls = make_rustls_connector(); - config.create_pool(Some(Runtime::Tokio1), tls) + let tls = make_rustls_connector()?; + config + .create_pool(Some(Runtime::Tokio1), tls) + .map_err(CreatePoolError::from) } } } diff --git a/src/llm/config.rs b/src/llm/config.rs index 8b7d41c3..413f80e2 100644 --- a/src/llm/config.rs +++ b/src/llm/config.rs @@ -138,6 +138,30 @@ pub struct LlmConfig { /// Default: 120. Increase for local LLMs (Ollama, vLLM, LM Studio) that /// need more time for prompt evaluation on consumer hardware. pub request_timeout_secs: u64, + /// Generic cheap/fast model for lightweight tasks (heartbeat, routing, evaluation). + /// Works with any backend. Set via `LLM_CHEAP_MODEL` env var. + /// When set, takes priority over the NearAI-specific `NEARAI_CHEAP_MODEL`. + pub cheap_model: Option, + /// Enable cascade mode for smart routing (retry with primary if cheap model + /// response seems uncertain). Default: true. Set via `SMART_ROUTING_CASCADE`. + pub smart_routing_cascade: bool, +} + +impl LlmConfig { + /// Resolve the effective cheap model name. + /// + /// Resolution order: + /// 1. `LLM_CHEAP_MODEL` (generic, works with any backend) + /// 2. `NEARAI_CHEAP_MODEL` (NearAI-only, backward compatibility) + pub fn cheap_model_name(&self) -> Option<&str> { + self.cheap_model.as_deref().or_else(|| { + if self.backend == "nearai" { + self.nearai.cheap_model.as_deref() + } else { + None + } + }) + } } /// NEAR AI configuration. diff --git a/src/llm/mod.rs b/src/llm/mod.rs index 51309bf3..3b6b01c4 100644 --- a/src/llm/mod.rs +++ b/src/llm/mod.rs @@ -376,32 +376,61 @@ fn create_ollama_from_registry( /// Create a cheap/fast LLM provider for lightweight tasks (heartbeat, routing, evaluation). /// -/// Uses `NEARAI_CHEAP_MODEL` if set, otherwise falls back to the main provider. -/// Currently only supports NEAR AI backend. +/// Resolution order: +/// 1. `LLM_CHEAP_MODEL` (generic, works with any backend) +/// 2. `NEARAI_CHEAP_MODEL` (NearAI-only, backward compatibility) +/// +/// Returns `None` if no cheap model is configured. pub fn create_cheap_llm_provider( config: &LlmConfig, session: Arc, ) -> Result>, LlmError> { - let Some(ref cheap_model) = config.nearai.cheap_model else { + let Some(cheap_model) = config.cheap_model_name() else { return Ok(None); }; - if config.backend != "nearai" { - tracing::warn!( - "NEARAI_CHEAP_MODEL is set but LLM_BACKEND is '{}', not nearai. \ - Cheap model setting will be ignored.", - config.backend - ); - return Ok(None); + create_cheap_provider_for_backend(config, session, cheap_model) +} + +/// Create a cheap provider for a specific backend. +/// +/// Handles backend-specific provider construction: +/// - `nearai` — clones NearAiConfig, swaps model, uses `create_llm_provider_with_config` +/// - `bedrock` — returns error (smart routing not yet supported) +/// - All others — clones `RegistryProviderConfig`, swaps model, uses `create_registry_provider` +fn create_cheap_provider_for_backend( + config: &LlmConfig, + session: Arc, + cheap_model: &str, +) -> Result>, LlmError> { + if config.backend == "nearai" { + let mut cheap_config = config.nearai.clone(); + cheap_config.model = cheap_model.to_string(); + let provider = + create_llm_provider_with_config(&cheap_config, session, config.request_timeout_secs)?; + return Ok(Some(provider)); } - let mut cheap_config = config.nearai.clone(); - cheap_config.model = cheap_model.clone(); + if config.backend == "bedrock" { + return Err(LlmError::RequestFailed { + provider: "bedrock".to_string(), + reason: "Smart routing with cheap model is not supported for Bedrock yet".to_string(), + }); + } - Ok(Some(Arc::new(NearAiChatProvider::new( - cheap_config, - session, - )?))) + // Registry-based provider: clone config and swap model + let reg_config = config.provider.as_ref().ok_or_else(|| LlmError::RequestFailed { + provider: config.backend.clone(), + reason: format!( + "Cannot create cheap provider for backend '{}': no registry provider config available", + config.backend + ), + })?; + + let mut cheap_reg_config = reg_config.clone(); + cheap_reg_config.model = cheap_model.to_string(); + let provider = create_registry_provider(&cheap_reg_config, config.request_timeout_secs)?; + Ok(Some(provider)) } /// Build the full LLM provider chain with all configured wrappers. @@ -449,14 +478,15 @@ pub async fn build_provider_chain( }; // 2. Smart routing (cheap/primary split) - let llm: Arc = if let Some(ref cheap_model) = config.nearai.cheap_model { - let mut cheap_config = config.nearai.clone(); - cheap_config.model = cheap_model.clone(); - let cheap = create_llm_provider_with_config( - &cheap_config, - session.clone(), - config.request_timeout_secs, - )?; + let llm: Arc = if let Some(cheap_model) = config.cheap_model_name() { + let cheap = create_cheap_provider_for_backend(config, session.clone(), cheap_model)? + .ok_or_else(|| LlmError::RequestFailed { + provider: config.backend.clone(), + reason: format!( + "Failed to create cheap provider for model '{cheap_model}' on backend '{}'", + config.backend + ), + })?; let cheap: Arc = if retry_config.max_retries > 0 { Arc::new(RetryProvider::new(cheap, retry_config.clone())) } else { @@ -471,7 +501,7 @@ pub async fn build_provider_chain( llm, cheap, SmartRoutingConfig { - cascade_enabled: config.nearai.smart_routing_cascade, + cascade_enabled: config.smart_routing_cascade, ..SmartRoutingConfig::default() }, )) @@ -600,6 +630,8 @@ mod tests { provider: None, bedrock: None, request_timeout_secs: 120, + cheap_model: None, + smart_routing_cascade: true, } } @@ -614,7 +646,7 @@ mod tests { } #[test] - fn test_create_cheap_llm_provider_creates_provider_when_configured() { + fn test_create_cheap_llm_provider_creates_provider_with_nearai_cheap_model() { let mut config = test_llm_config(); config.nearai.cheap_model = Some("cheap-test-model".to_string()); @@ -628,7 +660,26 @@ mod tests { } #[test] - fn test_create_cheap_llm_provider_ignored_for_non_nearai_backend() { + fn test_create_cheap_llm_provider_generic_overrides_nearai() { + let mut config = test_llm_config(); + config.nearai.cheap_model = Some("nearai-cheap".to_string()); + config.cheap_model = Some("generic-cheap".to_string()); + + let session = Arc::new(SessionManager::new(SessionConfig::default())); + let result = create_cheap_llm_provider(&config, session); + + assert!(result.is_ok()); + let provider = result.unwrap(); + assert!(provider.is_some()); + assert_eq!( + provider.unwrap().model_name(), + "generic-cheap", + "LLM_CHEAP_MODEL should take priority over NEARAI_CHEAP_MODEL" + ); + } + + #[test] + fn test_create_cheap_llm_provider_nearai_cheap_ignored_for_non_nearai_backend() { let mut config = test_llm_config(); config.backend = "openai".to_string(); config.nearai.cheap_model = Some("cheap-test-model".to_string()); @@ -637,6 +688,48 @@ mod tests { let result = create_cheap_llm_provider(&config, session); assert!(result.is_ok()); - assert!(result.unwrap().is_none()); + assert!( + result.unwrap().is_none(), + "NEARAI_CHEAP_MODEL should be ignored when backend is not nearai" + ); + } + + #[test] + fn test_create_cheap_llm_provider_bedrock_returns_error() { + let mut config = test_llm_config(); + config.backend = "bedrock".to_string(); + config.cheap_model = Some("cheap-model".to_string()); + + let session = Arc::new(SessionManager::new(SessionConfig::default())); + let result = create_cheap_llm_provider(&config, session); + + assert!( + result.is_err(), + "Bedrock should return an error for cheap model" + ); + } + + #[test] + fn test_cheap_model_name_resolution() { + // Generic takes priority + let mut config = test_llm_config(); + config.cheap_model = Some("generic".to_string()); + config.nearai.cheap_model = Some("nearai".to_string()); + assert_eq!(config.cheap_model_name(), Some("generic")); + + // NearAI fallback when backend is nearai + let mut config = test_llm_config(); + config.nearai.cheap_model = Some("nearai".to_string()); + assert_eq!(config.cheap_model_name(), Some("nearai")); + + // NearAI ignored for non-nearai backend + let mut config = test_llm_config(); + config.backend = "openai".to_string(); + config.nearai.cheap_model = Some("nearai".to_string()); + assert_eq!(config.cheap_model_name(), None); + + // None when nothing configured + let config = test_llm_config(); + assert_eq!(config.cheap_model_name(), None); } } diff --git a/src/llm/models.rs b/src/llm/models.rs index 7022d3cf..daec9df3 100644 --- a/src/llm/models.rs +++ b/src/llm/models.rs @@ -345,5 +345,7 @@ pub(crate) fn build_nearai_model_fetch_config() -> crate::config::LlmConfig { provider: None, bedrock: None, request_timeout_secs: 120, + cheap_model: None, + smart_routing_cascade: false, } } diff --git a/src/orchestrator/mod.rs b/src/orchestrator/mod.rs index 5e750ddf..b72f90ee 100644 --- a/src/orchestrator/mod.rs +++ b/src/orchestrator/mod.rs @@ -51,6 +51,15 @@ use crate::db::Database; use crate::llm::LlmProvider; use crate::secrets::SecretsStore; +/// Resolve the orchestrator port from the `ORCHESTRATOR_PORT` environment +/// variable, falling back to 50051. +fn resolve_orchestrator_port() -> u16 { + std::env::var("ORCHESTRATOR_PORT") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(50051) +} + /// Result of orchestrator setup, containing all handles needed by the agent. pub struct OrchestratorSetup { pub container_job_manager: Option>, @@ -101,11 +110,12 @@ pub async fn setup_orchestrator( let job_event_tx = Some(tx); let token_store = TokenStore::new(); + let orchestrator_port = resolve_orchestrator_port(); let job_config = ContainerJobConfig { image: config.sandbox.image.clone(), memory_limit_mb: config.sandbox.memory_limit_mb, cpu_shares: config.sandbox.cpu_shares, - orchestrator_port: 50051, + orchestrator_port, claude_code_api_key: std::env::var("ANTHROPIC_API_KEY").ok(), claude_code_oauth_token: crate::config::ClaudeCodeConfig::extract_oauth_token(), claude_code_model: config.claude_code.model.clone(), @@ -127,7 +137,7 @@ pub async fn setup_orchestrator( }; tokio::spawn(async move { - if let Err(e) = OrchestratorApi::start(orchestrator_state, 50051).await { + if let Err(e) = OrchestratorApi::start(orchestrator_state, orchestrator_port).await { tracing::error!("Orchestrator API failed: {}", e); } }); @@ -151,3 +161,40 @@ pub async fn setup_orchestrator( docker_status, } } + +#[cfg(test)] +mod tests { + use std::sync::Mutex; + + use super::*; + + /// Serialize access to `ORCHESTRATOR_PORT` env var across test threads. + static ENV_LOCK: Mutex<()> = Mutex::new(()); + + #[test] + fn resolve_orchestrator_port_from_env() { + let _guard = ENV_LOCK.lock().unwrap(); + + // Safety: env-var mutation requires unsafe in edition 2024; + // ENV_LOCK serializes concurrent access from other test threads. + + // Absent env var → default 50051 + unsafe { std::env::remove_var("ORCHESTRATOR_PORT") }; + assert_eq!(resolve_orchestrator_port(), 50051); + + // Valid custom port + unsafe { std::env::set_var("ORCHESTRATOR_PORT", "50052") }; + assert_eq!(resolve_orchestrator_port(), 50052); + + // Non-numeric value → fallback to default + unsafe { std::env::set_var("ORCHESTRATOR_PORT", "not_a_port") }; + assert_eq!(resolve_orchestrator_port(), 50051); + + // Out of u16 range → fallback to default + unsafe { std::env::set_var("ORCHESTRATOR_PORT", "99999") }; + assert_eq!(resolve_orchestrator_port(), 50051); + + // Cleanup + unsafe { std::env::remove_var("ORCHESTRATOR_PORT") }; + } +} diff --git a/src/transcription/chat_completions.rs b/src/transcription/chat_completions.rs new file mode 100644 index 00000000..e23818aa --- /dev/null +++ b/src/transcription/chat_completions.rs @@ -0,0 +1,179 @@ +//! Chat Completions-based transcription provider. +//! +//! Uses the `/v1/chat/completions` endpoint with `input_audio` content type +//! to transcribe audio. Compatible with OpenRouter, OpenAI GPT-4o-audio, and +//! any provider that supports audio input via the Chat Completions API. + +use async_trait::async_trait; +use base64::Engine; +use secrecy::{ExposeSecret, SecretString}; + +use super::{AudioFormat, TranscriptionError, TranscriptionProvider}; + +/// Transcription provider that sends audio via the Chat Completions API. +/// +/// Unlike the Whisper provider (which uses `/v1/audio/transcriptions` with +/// multipart upload), this provider sends base64-encoded audio as an +/// `input_audio` content part in a chat message, enabling use with +/// OpenRouter and other providers that only expose audio through the +/// Chat Completions API. +pub struct ChatCompletionsTranscriptionProvider { + client: reqwest::Client, + api_key: SecretString, + model: String, + base_url: String, +} + +impl ChatCompletionsTranscriptionProvider { + /// Create a new provider with the given API key. + pub fn new(api_key: SecretString) -> Self { + Self { + client: match reqwest::Client::builder() + .timeout(std::time::Duration::from_secs(120)) + .build() + { + Ok(c) => c, + Err(e) => { + tracing::error!( + "Failed to build HTTP client with timeout, falling back to default: {e}" + ); + reqwest::Client::default() + } + }, + api_key, + model: "google/gemini-2.0-flash-001".to_string(), + base_url: "https://openrouter.ai/api".to_string(), + } + } + + /// Override the base URL. + pub fn with_base_url(mut self, base_url: impl Into) -> Self { + self.base_url = base_url.into().trim_end_matches('/').to_string(); + self + } + + /// Override the model name. + pub fn with_model(mut self, model: impl Into) -> Self { + self.model = model.into(); + self + } +} + +/// Map [`AudioFormat`] to the format string expected by the Chat Completions API. +fn audio_format_str(format: AudioFormat) -> &'static str { + match format { + AudioFormat::Ogg => "ogg", + AudioFormat::Mp3 => "mp3", + AudioFormat::Mp4 => "mp4", + AudioFormat::Wav => "wav", + AudioFormat::Webm => "webm", + AudioFormat::Flac => "flac", + AudioFormat::M4a => "m4a", + } +} + +#[async_trait] +impl TranscriptionProvider for ChatCompletionsTranscriptionProvider { + async fn transcribe( + &self, + audio_data: &[u8], + format: AudioFormat, + ) -> Result { + if audio_data.is_empty() { + return Err(TranscriptionError::EmptyAudio); + } + + let b64 = base64::engine::general_purpose::STANDARD.encode(audio_data); + + let body = serde_json::json!({ + "model": self.model, + "messages": [{ + "role": "user", + "content": [ + { + "type": "text", + "text": "Transcribe this audio. Return only the transcript text, nothing else." + }, + { + "type": "input_audio", + "input_audio": { + "data": b64, + "format": audio_format_str(format) + } + } + ] + }] + }); + + let url = format!("{}/v1/chat/completions", self.base_url); + + let response = self + .client + .post(&url) + .header( + "Authorization", + format!("Bearer {}", self.api_key.expose_secret()), + ) + .json(&body) + .send() + .await + .map_err(|e| TranscriptionError::RequestFailed(e.to_string()))?; + + let status = response.status(); + if !status.is_success() { + let body = response + .text() + .await + .unwrap_or_else(|_| "unknown error".to_string()); + return Err(TranscriptionError::RequestFailed(format!( + "HTTP {}: {}", + status, body + ))); + } + + let json: serde_json::Value = response + .json() + .await + .map_err(|e| TranscriptionError::RequestFailed(e.to_string()))?; + + // Extract text from the standard Chat Completions response format: + // { "choices": [{ "message": { "content": "..." } }] } + let text = json + .get("choices") + .and_then(|c| c.get(0)) + .and_then(|c| c.get("message")) + .and_then(|m| m.get("content")) + .and_then(|c| c.as_str()) + .ok_or_else(|| { + TranscriptionError::RequestFailed( + "unexpected response format: missing choices[0].message.content".to_string(), + ) + })?; + + Ok(text.trim().to_string()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn audio_format_str_maps_all_variants() { + assert_eq!(audio_format_str(AudioFormat::Ogg), "ogg"); + assert_eq!(audio_format_str(AudioFormat::Mp3), "mp3"); + assert_eq!(audio_format_str(AudioFormat::Mp4), "mp4"); + assert_eq!(audio_format_str(AudioFormat::Wav), "wav"); + assert_eq!(audio_format_str(AudioFormat::Webm), "webm"); + assert_eq!(audio_format_str(AudioFormat::Flac), "flac"); + assert_eq!(audio_format_str(AudioFormat::M4a), "m4a"); + } + + #[tokio::test] + async fn rejects_empty_audio() { + let provider = + ChatCompletionsTranscriptionProvider::new(SecretString::from("test-key".to_string())); + let result = provider.transcribe(&[], AudioFormat::Ogg).await; + assert!(matches!(result, Err(TranscriptionError::EmptyAudio))); + } +} diff --git a/src/transcription/mod.rs b/src/transcription/mod.rs index d0a7d31c..ab2e43f9 100644 --- a/src/transcription/mod.rs +++ b/src/transcription/mod.rs @@ -4,8 +4,10 @@ //! backends and a [`TranscriptionMiddleware`] that detects audio attachments //! on incoming messages and replaces them with transcribed text. +mod chat_completions; mod openai; +pub use self::chat_completions::ChatCompletionsTranscriptionProvider; pub use self::openai::OpenAiWhisperProvider; use async_trait::async_trait; diff --git a/tests/e2e/conftest.py b/tests/e2e/conftest.py index 56a478c9..06c7da03 100644 --- a/tests/e2e/conftest.py +++ b/tests/e2e/conftest.py @@ -319,15 +319,14 @@ async def http_channel_server(ironclaw_server, server_ports): @pytest.fixture(scope="session") -async def ironclaw_server_with_webhook_secret(ironclaw_binary, mock_llm_server, wasm_tools_dir): - """Start ironclaw with HTTP_WEBHOOK_SECRET configured for webhook tests. - - Yields a dict with: - - 'url': base URL of the gateway - - 'secret': the webhook secret value - """ +async def http_channel_server_without_secret( + ironclaw_binary, + mock_llm_server, + wasm_tools_dir, +): + """Start the HTTP webhook channel without a configured secret.""" gateway_port = _find_free_port() - webhook_secret = "test-webhook-secret-e2e-12345" + http_port = _find_free_port() env = { # Minimal env: PATH for process spawning, HOME for Rust/cargo defaults "PATH": os.environ.get("PATH", "/usr/bin:/bin"), @@ -339,13 +338,14 @@ async def ironclaw_server_with_webhook_secret(ironclaw_binary, mock_llm_server, "GATEWAY_PORT": str(gateway_port), "GATEWAY_AUTH_TOKEN": AUTH_TOKEN, "GATEWAY_USER_ID": "e2e-tester", - "HTTP_WEBHOOK_SECRET": webhook_secret, + "HTTP_HOST": "127.0.0.1", + "HTTP_PORT": str(http_port), "CLI_ENABLED": "false", "LLM_BACKEND": "openai_compatible", "LLM_BASE_URL": mock_llm_server, "LLM_MODEL": "mock-model", "DATABASE_BACKEND": "libsql", - "LIBSQL_PATH": os.path.join(_DB_TMPDIR.name, "e2e-webhook.db"), + "LIBSQL_PATH": os.path.join(_DB_TMPDIR.name, "e2e-webhook-no-secret.db"), "SANDBOX_ENABLED": "false", "SKILLS_ENABLED": "true", "ROUTINES_ENABLED": "false", @@ -375,13 +375,12 @@ async def ironclaw_server_with_webhook_secret(ironclaw_binary, mock_llm_server, stderr=asyncio.subprocess.PIPE, env=env, ) - base_url = f"http://127.0.0.1:{gateway_port}" + gateway_url = f"http://127.0.0.1:{gateway_port}" + http_base_url = f"http://127.0.0.1:{http_port}" try: - await wait_for_ready(f"{base_url}/api/health", timeout=60) - yield { - "url": base_url, - "secret": webhook_secret, - } + await wait_for_ready(f"{gateway_url}/api/health", timeout=60) + await wait_for_ready(f"{http_base_url}/health", timeout=30) + yield http_base_url except TimeoutError: # Dump stderr so CI logs show why the server failed to start returncode = proc.returncode @@ -394,7 +393,8 @@ async def ironclaw_server_with_webhook_secret(ironclaw_binary, mock_llm_server, stderr_text = stderr_bytes.decode("utf-8", errors="replace") proc.kill() pytest.fail( - f"ironclaw server with webhook secret failed to start on port {gateway_port} " + f"ironclaw server without webhook secret failed to start on ports " + f"gateway={gateway_port}, http={http_port} " f"(returncode={returncode}).\nstderr:\n{stderr_text}" ) finally: diff --git a/tests/e2e/ironclaw_e2e.egg-info/SOURCES.txt b/tests/e2e/ironclaw_e2e.egg-info/SOURCES.txt index 7f011382..c2784f64 100644 --- a/tests/e2e/ironclaw_e2e.egg-info/SOURCES.txt +++ b/tests/e2e/ironclaw_e2e.egg-info/SOURCES.txt @@ -12,11 +12,17 @@ scenarios/test_csp.py scenarios/test_extension_oauth.py scenarios/test_extensions.py scenarios/test_html_injection.py +scenarios/test_mcp_auth_flow.py scenarios/test_oauth_credential_fallback.py +scenarios/test_owner_scope.py scenarios/test_pairing.py +scenarios/test_routine_event_batch.py scenarios/test_routine_oauth_credential_injection.py scenarios/test_skills.py scenarios/test_sse_reconnect.py +scenarios/test_telegram_hot_activation.py +scenarios/test_telegram_token_validation.py scenarios/test_tool_approval.py scenarios/test_tool_execution.py -scenarios/test_wasm_lifecycle.py \ No newline at end of file +scenarios/test_wasm_lifecycle.py +scenarios/test_webhook.py \ No newline at end of file diff --git a/tests/e2e/mock_llm.py b/tests/e2e/mock_llm.py index c53da894..b091fc17 100644 --- a/tests/e2e/mock_llm.py +++ b/tests/e2e/mock_llm.py @@ -55,6 +55,25 @@ TOOL_CALL_PATTERNS = [ "action_type": "full_job", }, ), + ( + re.compile( + r"create event routine (?P[a-z0-9][a-z0-9_-]*) " + r"channel (?P[a-z0-9_-]+) pattern (?P[a-z0-9_|-]+)", + re.IGNORECASE, + ), + "routine_create", + lambda m: { + "name": m.group("name"), + "description": f"Event routine {m.group('name')}", + "trigger_type": "event", + "event_channel": None if m.group("channel").lower() == "any" else m.group("channel"), + "event_pattern": m.group("pattern"), + "prompt": f"Acknowledge that {m.group('name')} fired.", + "action_type": "lightweight", + "use_tools": False, + "cooldown_secs": 0, + }, + ), ( re.compile(r"list owner routines", re.IGNORECASE), "routine_list", diff --git a/tests/e2e/scenarios/test_routine_event_batch.py b/tests/e2e/scenarios/test_routine_event_batch.py index d8c59e6d..7da78a15 100644 --- a/tests/e2e/scenarios/test_routine_event_batch.py +++ b/tests/e2e/scenarios/test_routine_event_batch.py @@ -1,534 +1,317 @@ -""" -E2E tests for event-triggered routines with batch loading. - -These tests verify that the N+1 query fix correctly: -1. Fires event-triggered routines on matching messages -2. Enforces concurrent limits via batch-loaded counts -3. Maintains performance with multiple simultaneous triggers -4. Works correctly through the full UI and agent loop - -Playwright-based UI tests + SSE verification. -""" +"""E2E tests for event-triggered routines over the HTTP channel.""" import asyncio import json +import uuid + +import httpx import pytest -from datetime import datetime, timedelta -from typing import List, Dict, Any -from playwright.async_api import async_playwright, Page, Browser, BrowserContext +from helpers import AUTH_TOKEN, SEL, signed_http_webhook_headers -@pytest.fixture -async def browser_and_context(): - """Create a Playwright browser and context for testing.""" - async with async_playwright() as p: - browser = await p.chromium.launch(headless=True) - context = await browser.new_context() - yield browser, context - await context.close() - await browser.close() +async def _send_chat_message(page, message: str) -> None: + """Send a chat message and wait for the assistant turn to appear.""" + chat_input = page.locator(SEL["chat_input"]) + await chat_input.wait_for(state="visible", timeout=5000) + assistant_messages = page.locator(SEL["message_assistant"]) + before_count = await assistant_messages.count() + + await chat_input.fill(message) + await chat_input.press("Enter") + + await page.wait_for_function( + """({ selector, expectedCount }) => { + return document.querySelectorAll(selector).length >= expectedCount; + }""", + arg={ + "selector": SEL["message_assistant"], + "expectedCount": before_count + 1, + }, + timeout=30000, + ) -class EventTriggerHelper: - """Helper methods for event trigger testing.""" +async def _create_event_routine( + page, + base_url: str, + *, + name: str, + pattern: str, + channel: str = "http", +) -> dict: + """Create an event routine through chat and return its API record.""" + await _send_chat_message( + page, + f"create event routine {name} channel {channel} pattern {pattern}", + ) + return await _wait_for_routine(base_url, name) - def __init__(self, page: Page): - self.page = page - async def navigate_to_routines(self): - """Navigate to the routines page.""" - await self.page.goto("http://localhost:8000/routines") - await self.page.wait_for_load_state("networkidle") +async def _post_http_message( + http_channel_server: str, + *, + content: str, + sender_id: str | None = None, + thread_id: str | None = None, +) -> dict: + """Send a signed HTTP-channel message and return the JSON body.""" + payload = { + "user_id": sender_id or f"sender-{uuid.uuid4().hex[:8]}", + "thread_id": thread_id or f"thread-{uuid.uuid4().hex[:8]}", + "content": content, + "wait_for_response": True, + } + body = json.dumps(payload).encode("utf-8") - async def create_event_routine( - self, - name: str, - trigger_regex: str, - channel: str = "slack", - max_concurrent: int = 1, - ) -> str: - """ - Create an event-triggered routine via UI. - Returns the routine ID. - """ - await self.navigate_to_routines() - - # Click "New Routine" button - await self.page.click('button:has-text("New Routine")') - await self.page.wait_for_selector('input[name="routine_name"]') - - # Fill routine details - await self.page.fill('input[name="routine_name"]', name) - await self.page.fill( - 'textarea[name="routine_description"]', - f"Test routine: {name}", + async with httpx.AsyncClient() as client: + response = await client.post( + f"{http_channel_server}/webhook", + content=body, + headers=signed_http_webhook_headers(body), + timeout=90, ) - # Select "Event Trigger" type - await self.page.click('label:has-text("Event Trigger")') - await self.page.wait_for_selector('input[name="trigger_regex"]') - - # Fill trigger details - await self.page.fill('input[name="trigger_regex"]', trigger_regex) - await self.page.select_option('select[name="trigger_channel"]', channel) - - # Set guardrails - await self.page.fill('input[name="max_concurrent"]', str(max_concurrent)) - - # Select lightweight action - await self.page.click('label:has-text("Lightweight")') - await self.page.fill( - 'textarea[name="lightweight_prompt"]', - "Acknowledge the message and confirm trigger worked.", - ) - - # Save routine - await self.page.click('button:has-text("Save Routine")') - await self.page.wait_for_selector('text=Routine created successfully') - - # Extract routine ID from success message or URL - routine_id = await self.page.locator('data-testid=routine-id').text_content() - return routine_id.strip() if routine_id else None - - async def create_multiple_routines( - self, base_name: str, count: int, trigger_regex: str = None - ) -> List[str]: - """Create multiple event-triggered routines.""" - routine_ids = [] - for i in range(count): - name = f"{base_name}_{i}" - regex = trigger_regex or f"({i}|{base_name})" - routine_id = await self.create_event_routine(name, regex) - routine_ids.append(routine_id) - await asyncio.sleep(0.1) # Small delay between creations - return routine_ids - - async def send_chat_message(self, message: str) -> List[str]: - """ - Send a chat message and return SSE events received. - Captures all routine firing events. - """ - await self.page.goto("http://localhost:8000/chat") - await self.page.wait_for_selector('input[placeholder*="message"]', timeout=5000) - - # Collect SSE events - sse_events = [] - - async def capture_sse(response): - """Intercept SSE events.""" - if "event-stream" in response.headers.get("content-type", ""): - text = await response.text() - for line in text.split("\n"): - if line.startswith("data:"): - try: - event = json.loads(line[5:]) - sse_events.append(event) - except json.JSONDecodeError: - pass - - self.page.on("response", capture_sse) - - # Send message - await self.page.fill('input[placeholder*="message"]', message) - await self.page.press('input[placeholder*="message"]', "Enter") - - # Wait for response - await self.page.wait_for_selector('text=Message processed', timeout=10000) - await asyncio.sleep(0.5) # Allow time for SSE events - - self.page.remove_listener("response", capture_sse) - return sse_events - - async def get_routine_execution_log(self, routine_id: str) -> List[Dict]: - """Get execution log entries for a routine.""" - await self.page.goto(f"http://localhost:8000/routines/{routine_id}/executions") - await self.page.wait_for_load_state("networkidle") - - # Extract log entries from table - rows = await self.page.locator("tbody tr").all() - executions = [] - - for row in rows: - cells = await row.locator("td").all() - if len(cells) >= 3: - execution = { - "timestamp": await cells[0].text_content(), - "status": await cells[1].text_content(), - "details": await cells[2].text_content(), - } - executions.append(execution) - - return executions - - async def check_database_queries_in_logs( - self, max_queries_expected: int = 1 - ) -> int: - """Check debug logs for database query count.""" - await self.page.goto("http://localhost:8000/debug/logs?filter=database") - await self.page.wait_for_load_state("networkidle") - - # Count batch queries - log_lines = await self.page.locator("tr:has-text('batch')").all() - batch_count = len(log_lines) - - # Count individual COUNT queries (should be 0 after fix) - count_queries = await self.page.locator("tr:has-text('COUNT')").all() - count_query_count = len(count_queries) - - return batch_count, count_query_count - - -# ============================================================================= -# Tests -# ============================================================================= - - -@pytest.mark.asyncio -async def test_create_event_trigger_routine(browser_and_context): - """Test creating an event-triggered routine via UI.""" - browser, context = browser_and_context - page = await context.new_page() - helper = EventTriggerHelper(page) - - try: - routine_id = await helper.create_event_routine( - name="Test Trigger", - trigger_regex="test|demo", - channel="slack", - max_concurrent=1, - ) - - assert routine_id is not None, "Routine ID should be returned" - assert len(routine_id) > 0, "Routine ID should not be empty" - - finally: - await page.close() - - -@pytest.mark.asyncio -async def test_event_trigger_fires_on_matching_message(browser_and_context): - """Test that event-triggered routine fires when message matches.""" - browser, context = browser_and_context - page = await context.new_page() - helper = EventTriggerHelper(page) - - try: - # Create routine - routine_id = await helper.create_event_routine( - name="Alert Handler", - trigger_regex="urgent|critical|alert", - channel="slack", - ) - - # Send matching message - sse_events = await helper.send_chat_message("URGENT: Server down!") - - # Verify routine fired (look for event in SSE stream) - routine_fired = any( - event.get("type") == "routine_fired" and event.get("routine_id") == routine_id - for event in sse_events - ) - assert routine_fired, "Routine should fire on matching message" - - # Check execution log - executions = await helper.get_routine_execution_log(routine_id) - assert len(executions) > 0, "Execution should be logged" - assert "success" in executions[0]["status"].lower() - - finally: - await page.close() - - -@pytest.mark.asyncio -async def test_event_trigger_skips_non_matching_message(browser_and_context): - """Test that event-triggered routine skips when message doesn't match.""" - browser, context = browser_and_context - page = await context.new_page() - helper = EventTriggerHelper(page) - - try: - # Create routine - routine_id = await helper.create_event_routine( - name="Alert Handler", - trigger_regex="urgent|critical|alert", - channel="slack", - ) - - # Send non-matching message - sse_events = await helper.send_chat_message("Hello, how are you?") - - # Verify routine did NOT fire - routine_fired = any( - event.get("type") == "routine_fired" and event.get("routine_id") == routine_id - for event in sse_events - ) - assert not routine_fired, "Routine should not fire on non-matching message" - - finally: - await page.close() - - -@pytest.mark.asyncio -async def test_multiple_routines_fire_on_matching_message(browser_and_context): - """Test that multiple event-triggered routines fire on same message.""" - browser, context = browser_and_context - page = await context.new_page() - helper = EventTriggerHelper(page) - - try: - # Create 3 overlapping routines - routine_ids = await helper.create_multiple_routines( - base_name="Handler", count=3, trigger_regex="alert|warning|error" - ) - - # Send matching message - sse_events = await helper.send_chat_message("ERROR: Database connection failed") - - # Verify all 3 routines fired - fired_count = sum( - 1 - for event in sse_events - if event.get("type") == "routine_fired" and event.get("routine_id") in routine_ids - ) - - assert ( - fired_count >= 3 - ), f"Expected all 3 routines to fire, got {fired_count}" - - finally: - await page.close() - - -@pytest.mark.asyncio -async def test_concurrent_limit_prevents_additional_fires(browser_and_context): - """Test that concurrent limit is enforced via batch counts.""" - browser, context = browser_and_context - page = await context.new_page() - helper = EventTriggerHelper(page) - - try: - # Create routine with max_concurrent=1 - routine_id = await helper.create_event_routine( - name="Limited Handler", - trigger_regex="process|task", - max_concurrent=1, - ) - - # Trigger first message - await helper.send_chat_message("Process message 1") - await asyncio.sleep(1) - - # Check first execution logged - executions_1 = await helper.get_routine_execution_log(routine_id) - assert len(executions_1) >= 1 - - # Trigger second message while first is still running - sse_events = await helper.send_chat_message("Process message 2") - - # Second routine should be skipped (concurrent limit) - routine_skipped = any( - event.get("type") == "routine_skipped" - and event.get("reason") == "max_concurrent_reached" - and event.get("routine_id") == routine_id - for event in sse_events - ) - assert routine_skipped, "Routine should be skipped when concurrent limit reached" - - finally: - await page.close() - - -@pytest.mark.asyncio -async def test_rapid_messages_with_multiple_triggers_efficiency(browser_and_context): - """Test efficiency of batch loading with multiple rapid messages.""" - browser, context = browser_and_context - page = await context.new_page() - helper = EventTriggerHelper(page) - - try: - # Create 5 overlapping routines - routine_ids = await helper.create_multiple_routines( - base_name="Rapid", count=5, trigger_regex="test|demo|check" - ) - - # Send 10 matching messages rapidly - for i in range(10): - message = f"test message {i}" - await helper.send_chat_message(message) - await asyncio.sleep(0.1) - - # Check database logs for query efficiency - batch_count, count_query_count = await helper.check_database_queries_in_logs() - - # After fix: should have ~10 batch queries (1 per message) - # Before fix: would have ~50 individual COUNT queries (5 routines × 10 messages) - assert ( - count_query_count == 0 - ), f"Should have 0 individual COUNT queries after fix, got {count_query_count}" - assert ( - batch_count <= 15 - ), f"Should have <=15 batch queries for 10 messages, got {batch_count}" - - finally: - await page.close() - - -@pytest.mark.asyncio -async def test_channel_filter_applied_correctly(browser_and_context): - """Test that channel filter prevents non-matching messages.""" - browser, context = browser_and_context - page = await context.new_page() - helper = EventTriggerHelper(page) - - try: - # Create routine for Slack channel - slack_routine_id = await helper.create_event_routine( - name="Slack Handler", - trigger_regex="alert", - channel="slack", - ) - - # Simulate message from Telegram channel - # (Note: In real UI, would need to change channel context) - page.goto( - "http://localhost:8000/chat?channel=telegram" - ) # Switch channel - await helper.send_chat_message("alert: something urgent") - - # Routine should not fire (different channel) - executions = await helper.get_routine_execution_log(slack_routine_id) - - # Check if any recent execution (last 5 min) exists - recent = [ - e - for e in executions - if (datetime.now() - datetime.fromisoformat(e["timestamp"])).total_seconds() - < 300 - ] - assert ( - len(recent) == 0 - ), "Routine should not fire for different channel" - - finally: - await page.close() - - -@pytest.mark.asyncio -async def test_batch_query_failure_handling(browser_and_context): - """Test graceful handling of batch query failures.""" - browser, context = browser_and_context - page = await context.new_page() - helper = EventTriggerHelper(page) - - try: - # Create routine - routine_id = await helper.create_event_routine( - name="Error Handler", - trigger_regex="test", - ) - - # Simulate database error in logs (if possible with test hooks) - # For now, just verify error handling doesn't crash UI - await helper.send_chat_message("test message") - - # Check that UI remains responsive - assert await page.locator("text=Message processed").is_visible() - - finally: - await page.close() - - -@pytest.mark.asyncio -async def test_routine_execution_history_display(browser_and_context): - """Test that execution history correctly displays routine firings.""" - browser, context = browser_and_context - page = await context.new_page() - helper = EventTriggerHelper(page) - - try: - # Create routine - routine_id = await helper.create_event_routine( - name="History Test", - trigger_regex="test", - ) - - # Trigger routine 3 times - for i in range(3): - await helper.send_chat_message(f"test message {i}") - await asyncio.sleep(0.2) - - # Check execution log - executions = await helper.get_routine_execution_log(routine_id) - assert len(executions) >= 3, "Should have at least 3 executions logged" - - # Verify all are recent (within last 5 minutes) - for execution in executions[:3]: - timestamp = datetime.fromisoformat(execution["timestamp"]) - age = datetime.now() - timestamp - assert age < timedelta(minutes=5), "Execution should be recent" - - finally: - await page.close() - - -@pytest.mark.asyncio -async def test_concurrent_batch_loads_independent(browser_and_context): - """Test that concurrent messages each get independent batch queries.""" - browser, context = browser_and_context - page = await context.new_page() - helper = EventTriggerHelper(page) - - try: - # Create 5 routines matching different patterns - r1_id = await helper.create_event_routine( - name="Pattern A", trigger_regex="alpha|alpha_only" - ) - r2_id = await helper.create_event_routine( - name="Pattern B", trigger_regex="beta|beta_only" - ) - r3_id = await helper.create_event_routine( - name="Pattern AB", trigger_regex="alpha|beta|common" - ) - - # Send overlapping messages - # Message 1: matches r1, r3 - sse1 = await helper.send_chat_message("alpha common") - await asyncio.sleep(0.1) - - # Message 2: matches r2, r3 - sse2 = await helper.send_chat_message("beta common") - await asyncio.sleep(0.1) - - # Verify correct routines fired - r1_fired_msg1 = any( - e.get("routine_id") == r1_id for e in sse1 if e.get("type") == "routine_fired" - ) - r2_fired_msg2 = any( - e.get("routine_id") == r2_id for e in sse2 if e.get("type") == "routine_fired" - ) - r3_fired_both = ( - any( - e.get("routine_id") == r3_id for e in sse1 if e.get("type") == "routine_fired" + assert response.status_code == 200, ( + f"HTTP webhook failed: {response.status_code} {response.text[:400]}" + ) + return response.json() + + +async def _wait_for_routine(base_url: str, name: str, timeout: float = 20.0) -> dict: + """Poll the routines API until the named routine exists.""" + async with httpx.AsyncClient() as client: + for _ in range(int(timeout * 2)): + response = await client.get( + f"{base_url}/api/routines", + headers={"Authorization": f"Bearer {AUTH_TOKEN}"}, + timeout=10, ) - and any( - e.get("routine_id") == r3_id for e in sse2 if e.get("type") == "routine_fired" + response.raise_for_status() + for routine in response.json()["routines"]: + if routine["name"] == name: + return routine + await asyncio.sleep(0.5) + raise AssertionError(f"Routine '{name}' was not created within {timeout}s") + + +async def _get_routine_runs(base_url: str, routine_id: str) -> list[dict]: + """Fetch recent routine runs from the web API.""" + async with httpx.AsyncClient() as client: + response = await client.get( + f"{base_url}/api/routines/{routine_id}/runs", + headers={"Authorization": f"Bearer {AUTH_TOKEN}"}, + timeout=10, + ) + response.raise_for_status() + return response.json()["runs"] + + +async def _wait_for_run_count( + base_url: str, + routine_id: str, + *, + expected_at_least: int, + timeout: float = 20.0, +) -> list[dict]: + """Poll until the routine has at least the expected run count.""" + for _ in range(int(timeout * 2)): + runs = await _get_routine_runs(base_url, routine_id) + if len(runs) >= expected_at_least: + return runs + await asyncio.sleep(0.5) + raise AssertionError( + f"Routine '{routine_id}' did not reach {expected_at_least} runs within {timeout}s" + ) + + +async def _wait_for_completed_run( + base_url: str, + routine_id: str, + *, + timeout: float = 30.0, +) -> dict: + """Poll until the newest run is no longer marked running.""" + for _ in range(int(timeout * 2)): + runs = await _get_routine_runs(base_url, routine_id) + if runs and runs[0]["status"].lower() != "running": + return runs[0] + await asyncio.sleep(0.5) + raise AssertionError(f"Routine '{routine_id}' did not complete within {timeout}s") + + +@pytest.mark.asyncio +async def test_create_event_trigger_routine(page, ironclaw_server): + """Event routines can be created through the supported chat flow.""" + name = f"evt-{uuid.uuid4().hex[:8]}" + routine = await _create_event_routine( + page, + ironclaw_server, + name=name, + pattern="test|demo", + ) + + assert routine["id"] + assert routine["trigger_type"] == "event" + assert "test|demo" in routine["trigger_summary"] + + +@pytest.mark.asyncio +async def test_event_trigger_fires_on_matching_message( + page, + ironclaw_server, + http_channel_server, +): + """Matching HTTP-channel messages create routine runs.""" + name = f"evt-{uuid.uuid4().hex[:8]}" + routine = await _create_event_routine( + page, + ironclaw_server, + name=name, + pattern="urgent|critical|alert", + ) + + response = await _post_http_message( + http_channel_server, + content="urgent: server down", + ) + assert response["status"] == "accepted" + + await _wait_for_run_count( + ironclaw_server, + routine["id"], + expected_at_least=1, + ) + completed_run = await _wait_for_completed_run(ironclaw_server, routine["id"]) + + assert completed_run["status"].lower() == "attention" + assert completed_run["trigger_type"] == "event" + + +@pytest.mark.asyncio +async def test_event_trigger_skips_non_matching_message( + page, + ironclaw_server, + http_channel_server, +): + """Non-matching messages do not create routine runs.""" + name = f"evt-{uuid.uuid4().hex[:8]}" + routine = await _create_event_routine( + page, + ironclaw_server, + name=name, + pattern="urgent|critical|alert", + ) + + await _post_http_message( + http_channel_server, + content="hello there", + ) + await asyncio.sleep(2) + + assert await _get_routine_runs(ironclaw_server, routine["id"]) == [] + + +@pytest.mark.asyncio +async def test_multiple_routines_fire_on_matching_message( + page, + ironclaw_server, + http_channel_server, +): + """A single matching message can fire multiple event routines.""" + routines = [] + for _ in range(3): + name = f"evt-{uuid.uuid4().hex[:8]}" + routines.append( + await _create_event_routine( + page, + ironclaw_server, + name=name, + pattern="error|warning|alert", ) ) - assert r1_fired_msg1, "Routine 1 should fire on message 1" - assert r2_fired_msg2, "Routine 2 should fire on message 2" - assert r3_fired_both, "Routine 3 should fire on both messages" + await _post_http_message( + http_channel_server, + content="error: database connection failed", + ) - finally: - await page.close() + for routine in routines: + await _wait_for_run_count( + ironclaw_server, + routine["id"], + expected_at_least=1, + ) + completed_run = await _wait_for_completed_run(ironclaw_server, routine["id"]) + assert completed_run["status"].lower() == "attention" -# ============================================================================= -# Integration with existing test patterns -# ============================================================================= +@pytest.mark.asyncio +async def test_channel_filter_applied_correctly( + page, + ironclaw_server, + http_channel_server, +): + """Channel filters prevent HTTP messages from firing non-HTTP routines.""" + http_routine = await _create_event_routine( + page, + ironclaw_server, + name=f"evt-{uuid.uuid4().hex[:8]}", + pattern="alert", + channel="http", + ) + telegram_routine = await _create_event_routine( + page, + ironclaw_server, + name=f"evt-{uuid.uuid4().hex[:8]}", + pattern="alert", + channel="telegram", + ) + + await _post_http_message( + http_channel_server, + content="alert from webhook", + ) + + await _wait_for_run_count( + ironclaw_server, + http_routine["id"], + expected_at_least=1, + ) + http_run = await _wait_for_completed_run(ironclaw_server, http_routine["id"]) + await asyncio.sleep(2) + telegram_runs = await _get_routine_runs(ironclaw_server, telegram_routine["id"]) + + assert http_run["status"].lower() == "attention" + assert telegram_runs == [] -if __name__ == "__main__": - # Run tests with: pytest tests/e2e/scenarios/test_routine_event_batch.py -v - pytest.main([__file__, "-v", "-s"]) +@pytest.mark.asyncio +async def test_routine_execution_history_is_available( + page, + ironclaw_server, + http_channel_server, +): + """Routine run history is exposed by the routines runs API.""" + routine = await _create_event_routine( + page, + ironclaw_server, + name=f"evt-{uuid.uuid4().hex[:8]}", + pattern="history", + ) + + await _post_http_message( + http_channel_server, + content="history event", + ) + + await _wait_for_run_count( + ironclaw_server, + routine["id"], + expected_at_least=1, + ) + completed_run = await _wait_for_completed_run(ironclaw_server, routine["id"]) + + assert completed_run["id"] + assert completed_run["started_at"] + assert completed_run["status"].lower() == "attention" diff --git a/tests/e2e/scenarios/test_webhook.py b/tests/e2e/scenarios/test_webhook.py index c0227c97..e6f9b26e 100644 --- a/tests/e2e/scenarios/test_webhook.py +++ b/tests/e2e/scenarios/test_webhook.py @@ -7,7 +7,7 @@ import json import httpx import pytest -from helpers import AUTH_TOKEN +from helpers import HTTP_WEBHOOK_SECRET def compute_signature(secret: str, body: bytes) -> str: @@ -16,325 +16,188 @@ def compute_signature(secret: str, body: bytes) -> str: return f"sha256={mac.hexdigest()}" -@pytest.mark.asyncio -async def test_webhook_requires_http_webhook_secret_configured(ironclaw_server): - """ - Webhook endpoint rejects requests when HTTP_WEBHOOK_SECRET is not configured. - This tests the fail-closed security posture. - """ - headers = {"Authorization": f"Bearer {AUTH_TOKEN}"} +async def _post_webhook( + base_url: str, + body_data: dict, + *, + signature: str | None = None, + content_type: str = "application/json", +) -> httpx.Response: + """Send a raw webhook request with optional signature.""" + body_bytes = json.dumps(body_data).encode() + headers = {"Content-Type": content_type} + if signature is not None: + headers["X-Hub-Signature-256"] = signature + async with httpx.AsyncClient() as client: - # When no webhook secret is configured on the server, all requests fail - r = await client.post( - f"{ironclaw_server}/webhook", - json={"content": "test message"}, + return await client.post( + f"{base_url}/webhook", + content=body_bytes, headers=headers, ) - # Server should reject with 503 Service Unavailable (fail closed) - assert r.status_code in (401, 503) @pytest.mark.asyncio -async def test_webhook_hmac_signature_valid(ironclaw_server_with_webhook_secret): +async def test_webhook_requires_http_webhook_secret_configured( + http_channel_server_without_secret, +): + """Webhook fails closed when no secret is configured.""" + response = await _post_webhook( + http_channel_server_without_secret, + {"content": "test message"}, + ) + + assert response.status_code == 503 + data = response.json() + assert data["status"] == "error" + assert "Webhook authentication not configured" in data.get("response", "") + + +@pytest.mark.asyncio +async def test_webhook_hmac_signature_valid(http_channel_server): """Valid X-Hub-Signature-256 HMAC signature is accepted.""" - secret = ironclaw_server_with_webhook_secret["secret"] - base_url = ironclaw_server_with_webhook_secret["url"] + body = {"content": "hello from webhook"} + signature = compute_signature(HTTP_WEBHOOK_SECRET, json.dumps(body).encode()) - headers = {"Authorization": f"Bearer {AUTH_TOKEN}"} - body_data = {"content": "hello from webhook"} - body_bytes = json.dumps(body_data).encode() - signature = compute_signature(secret, body_bytes) + response = await _post_webhook(http_channel_server, body, signature=signature) - async with httpx.AsyncClient() as client: - r = await client.post( - f"{base_url}/webhook", - content=body_bytes, - headers={ - **headers, - "Content-Type": "application/json", - "X-Hub-Signature-256": signature, - }, - ) - assert r.status_code == 200, f"Expected 200, got {r.status_code}: {r.text}" - resp = r.json() - assert resp["status"] == "ok" + assert response.status_code == 200, ( + f"Expected 200, got {response.status_code}: {response.text}" + ) + data = response.json() + assert data["status"] == "accepted" @pytest.mark.asyncio -async def test_webhook_invalid_hmac_signature_rejected( - ironclaw_server_with_webhook_secret, -): +async def test_webhook_invalid_hmac_signature_rejected(http_channel_server): """Invalid X-Hub-Signature-256 signature is rejected with 401.""" - base_url = ironclaw_server_with_webhook_secret["url"] + response = await _post_webhook( + http_channel_server, + {"content": "hello"}, + signature="sha256=0000000000000000000000000000000000000000000000000000000000000000", + ) - headers = {"Authorization": f"Bearer {AUTH_TOKEN}"} - body_data = {"content": "hello"} - body_bytes = json.dumps(body_data).encode() - invalid_signature = "sha256=0000000000000000000000000000000000000000000000000000000000000000" - - async with httpx.AsyncClient() as client: - r = await client.post( - f"{base_url}/webhook", - content=body_bytes, - headers={ - **headers, - "Content-Type": "application/json", - "X-Hub-Signature-256": invalid_signature, - }, - ) - assert r.status_code == 401, f"Expected 401, got {r.status_code}" - resp = r.json() - assert resp["status"] == "error" - assert "Invalid webhook signature" in resp.get("response", "") + assert response.status_code == 401 + data = response.json() + assert data["status"] == "error" + assert "Invalid webhook signature" in data.get("response", "") @pytest.mark.asyncio -async def test_webhook_wrong_secret_rejected(ironclaw_server_with_webhook_secret): +async def test_webhook_wrong_secret_rejected(http_channel_server): """Signature computed with wrong secret is rejected.""" - base_url = ironclaw_server_with_webhook_secret["url"] + body = {"content": "hello"} + signature = compute_signature("wrong-secret", json.dumps(body).encode()) - headers = {"Authorization": f"Bearer {AUTH_TOKEN}"} - body_data = {"content": "hello"} - body_bytes = json.dumps(body_data).encode() - # Compute signature with wrong secret - wrong_signature = compute_signature("wrong-secret", body_bytes) + response = await _post_webhook(http_channel_server, body, signature=signature) - async with httpx.AsyncClient() as client: - r = await client.post( - f"{base_url}/webhook", - content=body_bytes, - headers={ - **headers, - "Content-Type": "application/json", - "X-Hub-Signature-256": wrong_signature, - }, - ) - assert r.status_code == 401 - resp = r.json() - assert resp["status"] == "error" + assert response.status_code == 401 + assert response.json()["status"] == "error" @pytest.mark.asyncio -async def test_webhook_malformed_signature_rejected( - ironclaw_server_with_webhook_secret, -): - """Malformed X-Hub-Signature-256 header is rejected.""" - base_url = ironclaw_server_with_webhook_secret["url"] +async def test_webhook_missing_signature_header_rejected(http_channel_server): + """Missing X-Hub-Signature-256 header is rejected when no body secret is provided.""" + response = await _post_webhook(http_channel_server, {"content": "hello"}) - headers = {"Authorization": f"Bearer {AUTH_TOKEN}"} - body_data = {"content": "hello"} - body_bytes = json.dumps(body_data).encode() - - async with httpx.AsyncClient() as client: - # Missing sha256= prefix - r = await client.post( - f"{base_url}/webhook", - content=body_bytes, - headers={ - **headers, - "Content-Type": "application/json", - "X-Hub-Signature-256": "deadbeef", - }, - ) - assert r.status_code == 401 + assert response.status_code == 401 + data = response.json() + assert "Webhook authentication required" in data.get("response", "") + assert "X-Hub-Signature-256" in data.get("response", "") @pytest.mark.asyncio -async def test_webhook_missing_signature_header_rejected( - ironclaw_server_with_webhook_secret, -): - """Missing X-Hub-Signature-256 header is rejected when no body secret provided.""" - base_url = ironclaw_server_with_webhook_secret["url"] +async def test_webhook_deprecated_body_secret_still_works(http_channel_server): + """Deprecated body secret support still accepts old clients.""" + response = await _post_webhook( + http_channel_server, + {"content": "hello", "secret": HTTP_WEBHOOK_SECRET}, + ) - headers = {"Authorization": f"Bearer {AUTH_TOKEN}"} - body_data = {"content": "hello"} - body_bytes = json.dumps(body_data).encode() - - async with httpx.AsyncClient() as client: - # No X-Hub-Signature-256 header and no body secret - r = await client.post( - f"{base_url}/webhook", - content=body_bytes, - headers={ - **headers, - "Content-Type": "application/json", - }, - ) - assert r.status_code == 401 - resp = r.json() - assert "Webhook authentication required" in resp.get("response", "") - assert "X-Hub-Signature-256" in resp.get("response", "") + assert response.status_code == 200, ( + f"Expected 200, got {response.status_code}: {response.text}" + ) + assert response.json()["status"] == "accepted" @pytest.mark.asyncio -async def test_webhook_deprecated_body_secret_still_works( - ironclaw_server_with_webhook_secret, -): - """ - Deprecated: body 'secret' field still works for backward compatibility. - This test ensures we don't break existing clients during the migration period. - """ - secret = ironclaw_server_with_webhook_secret["secret"] - base_url = ironclaw_server_with_webhook_secret["url"] +async def test_webhook_header_takes_precedence_over_body_secret(http_channel_server): + """Header signature wins when both header and body secret are provided.""" + body = {"content": "hello", "secret": "wrong-secret-in-body"} + signature = compute_signature(HTTP_WEBHOOK_SECRET, json.dumps(body).encode()) - headers = {"Authorization": f"Bearer {AUTH_TOKEN}"} - # Old-style request with secret in body - body_data = {"content": "hello", "secret": secret} - body_bytes = json.dumps(body_data).encode() + response = await _post_webhook(http_channel_server, body, signature=signature) - async with httpx.AsyncClient() as client: - r = await client.post( - f"{base_url}/webhook", - content=body_bytes, - headers={ - **headers, - "Content-Type": "application/json", - }, - ) - # Should succeed (backward compatibility) - assert r.status_code == 200, f"Expected 200, got {r.status_code}: {r.text}" - resp = r.json() - assert resp["status"] == "ok" + assert response.status_code == 200 + assert response.json()["status"] == "accepted" @pytest.mark.asyncio -async def test_webhook_header_takes_precedence_over_body_secret( - ironclaw_server_with_webhook_secret, -): - """ - When both X-Hub-Signature-256 header and body secret are provided, - header takes precedence. - """ - secret = ironclaw_server_with_webhook_secret["secret"] - base_url = ironclaw_server_with_webhook_secret["url"] - - headers = {"Authorization": f"Bearer {AUTH_TOKEN}"} - body_data = {"content": "hello", "secret": "wrong-secret-in-body"} - body_bytes = json.dumps(body_data).encode() - # Compute signature with correct secret - signature = compute_signature(secret, body_bytes) +async def test_webhook_case_insensitive_header_lookup(http_channel_server): + """HTTP headers are treated case-insensitively.""" + body = {"content": "hello"} + body_bytes = json.dumps(body).encode() + signature = compute_signature(HTTP_WEBHOOK_SECRET, body_bytes) async with httpx.AsyncClient() as client: - r = await client.post( - f"{base_url}/webhook", + response = await client.post( + f"{http_channel_server}/webhook", content=body_bytes, headers={ - **headers, - "Content-Type": "application/json", - "X-Hub-Signature-256": signature, - }, - ) - # Should succeed because header signature is valid (takes precedence) - assert r.status_code == 200 - resp = r.json() - assert resp["status"] == "ok" - - -@pytest.mark.asyncio -async def test_webhook_case_insensitive_header_lookup( - ironclaw_server_with_webhook_secret, -): - """HTTP headers are case-insensitive. Test with different cases.""" - secret = ironclaw_server_with_webhook_secret["secret"] - base_url = ironclaw_server_with_webhook_secret["url"] - - headers = {"Authorization": f"Bearer {AUTH_TOKEN}"} - body_data = {"content": "hello"} - body_bytes = json.dumps(body_data).encode() - signature = compute_signature(secret, body_bytes) - - async with httpx.AsyncClient() as client: - # Try with lowercase - r = await client.post( - f"{base_url}/webhook", - content=body_bytes, - headers={ - **headers, "Content-Type": "application/json", "x-hub-signature-256": signature, }, ) - assert r.status_code == 200 + + assert response.status_code == 200 @pytest.mark.asyncio -async def test_webhook_wrong_content_type_rejected( - ironclaw_server_with_webhook_secret, -): +async def test_webhook_wrong_content_type_rejected(http_channel_server): """Webhook only accepts application/json Content-Type.""" - secret = ironclaw_server_with_webhook_secret["secret"] - base_url = ironclaw_server_with_webhook_secret["url"] + body = {"content": "hello"} + signature = compute_signature(HTTP_WEBHOOK_SECRET, json.dumps(body).encode()) - headers = {"Authorization": f"Bearer {AUTH_TOKEN}"} - body_data = {"content": "hello"} - body_bytes = json.dumps(body_data).encode() - signature = compute_signature(secret, body_bytes) + response = await _post_webhook( + http_channel_server, + body, + signature=signature, + content_type="text/plain", + ) - async with httpx.AsyncClient() as client: - r = await client.post( - f"{base_url}/webhook", - content=body_bytes, - headers={ - **headers, - "Content-Type": "text/plain", - "X-Hub-Signature-256": signature, - }, - ) - assert r.status_code == 415 # Unsupported Media Type - resp = r.json() - assert "application/json" in resp.get("response", "") + assert response.status_code == 415 + assert "application/json" in response.json().get("response", "") @pytest.mark.asyncio -async def test_webhook_invalid_json_rejected(ironclaw_server_with_webhook_secret): +async def test_webhook_invalid_json_rejected(http_channel_server): """Invalid JSON in body is rejected.""" - secret = ironclaw_server_with_webhook_secret["secret"] - base_url = ironclaw_server_with_webhook_secret["url"] - - headers = {"Authorization": f"Bearer {AUTH_TOKEN}"} body_bytes = b"not valid json" - signature = compute_signature(secret, body_bytes) + signature = compute_signature(HTTP_WEBHOOK_SECRET, body_bytes) async with httpx.AsyncClient() as client: - r = await client.post( - f"{base_url}/webhook", + response = await client.post( + f"{http_channel_server}/webhook", content=body_bytes, headers={ - **headers, "Content-Type": "application/json", "X-Hub-Signature-256": signature, }, ) - assert r.status_code == 401 or r.status_code == 400 + + assert response.status_code in (400, 401) @pytest.mark.asyncio -async def test_webhook_message_queued_for_processing( - ironclaw_server_with_webhook_secret, -): - """Message via webhook is queued and can be retrieved.""" - secret = ironclaw_server_with_webhook_secret["secret"] - base_url = ironclaw_server_with_webhook_secret["url"] +async def test_webhook_message_queued_for_processing(http_channel_server): + """Accepted webhook requests return a real message id.""" + body = {"content": "webhook test message 12345"} + signature = compute_signature(HTTP_WEBHOOK_SECRET, json.dumps(body).encode()) - headers = {"Authorization": f"Bearer {AUTH_TOKEN}"} - test_message = "webhook test message 12345" - body_data = {"content": test_message} - body_bytes = json.dumps(body_data).encode() - signature = compute_signature(secret, body_bytes) + response = await _post_webhook(http_channel_server, body, signature=signature) - async with httpx.AsyncClient() as client: - r = await client.post( - f"{base_url}/webhook", - content=body_bytes, - headers={ - **headers, - "Content-Type": "application/json", - "X-Hub-Signature-256": signature, - }, - ) - assert r.status_code == 200 - resp = r.json() - assert resp["status"] == "ok" - # Message ID should be present - assert "message_id" in resp - assert resp["message_id"] != "00000000-0000-0000-0000-000000000000" + assert response.status_code == 200 + data = response.json() + assert data["status"] == "accepted" + assert "message_id" in data + assert data["message_id"] != "00000000-0000-0000-0000-000000000000"