diff --git a/src/channels/channel.rs b/src/channels/channel.rs index e126ca1f..60cdfe7a 100644 --- a/src/channels/channel.rs +++ b/src/channels/channel.rs @@ -344,6 +344,24 @@ pub trait Channel: Send + Sync { } } +/// Trait for channels that support hot-secret-swapping during SIGHUP reload. +/// +/// This allows channels to update authentication credentials without restarting, +/// enabling zero-downtime configuration reloads. Channels that don't support +/// secret updates can simply not implement this trait. +#[async_trait] +pub trait ChannelSecretUpdater: Send + Sync { + /// Update the secret for this channel. + /// + /// Called during SIGHUP configuration reload. Implementation should: + /// - Apply the new secret atomically + /// - Not fail the entire reload if secret update fails + /// - Log appropriate errors/info messages + /// + /// The secret is optional (may be None if secret is no longer configured). + async fn update_secret(&self, new_secret: Option); +} + #[cfg(test)] mod tests { use super::*; diff --git a/src/channels/http.rs b/src/channels/http.rs index 6851b337..af0fafcf 100644 --- a/src/channels/http.rs +++ b/src/channels/http.rs @@ -18,7 +18,8 @@ use tokio_stream::wrappers::ReceiverStream; use uuid::Uuid; use crate::channels::{ - AttachmentKind, Channel, IncomingAttachment, IncomingMessage, MessageStream, OutgoingResponse, + AttachmentKind, Channel, ChannelSecretUpdater, IncomingAttachment, IncomingMessage, + MessageStream, OutgoingResponse, }; use crate::config::HttpConfig; use crate::error::ChannelError; @@ -35,9 +36,10 @@ pub struct HttpChannelState { /// Pending responses keyed by message ID. pending_responses: RwLock>>, /// Expected webhook secret for authentication (if configured). - /// Wrapped in RwLock for hot-swapping on SIGHUP. + /// Stored in a separate Arc> to avoid contending with other state operations. + /// Rarely changes (only on SIGHUP), so isolated from hot-path state accesses. /// Uses SecretString to prevent accidental logging and memory dump exposure. - webhook_secret: RwLock>, + webhook_secret: Arc>>, /// Fixed user ID for this HTTP channel. user_id: String, /// Rate limiting state. @@ -85,7 +87,7 @@ impl HttpChannel { state: Arc::new(HttpChannelState { tx: RwLock::new(None), pending_responses: RwLock::new(std::collections::HashMap::new()), - webhook_secret: RwLock::new(webhook_secret), + webhook_secret: Arc::new(RwLock::new(webhook_secret)), user_id, rate_limit: tokio::sync::Mutex::new(RateLimitState { window_start: std::time::Instant::now(), @@ -496,6 +498,16 @@ impl Channel for HttpChannel { } } +/// Implement secret update for HTTP channel state. +/// This allows SIGHUP handler to update secrets generically via the trait. +#[async_trait] +impl ChannelSecretUpdater for HttpChannelState { + async fn update_secret(&self, new_secret: Option) { + *self.webhook_secret.write().await = new_secret; + tracing::info!("HTTP webhook secret updated"); + } +} + #[cfg(test)] mod tests { use axum::body::Body; diff --git a/src/channels/mod.rs b/src/channels/mod.rs index a6bc2956..038b432f 100644 --- a/src/channels/mod.rs +++ b/src/channels/mod.rs @@ -37,8 +37,8 @@ pub mod web; mod webhook_server; pub use channel::{ - AttachmentKind, Channel, IncomingAttachment, IncomingMessage, MessageStream, OutgoingResponse, - StatusUpdate, + AttachmentKind, Channel, ChannelSecretUpdater, IncomingAttachment, IncomingMessage, + MessageStream, OutgoingResponse, StatusUpdate, }; pub use http::{HttpChannel, HttpChannelState}; pub use manager::ChannelManager; diff --git a/src/llm/anthropic_oauth.rs b/src/llm/anthropic_oauth.rs index 0badda93..12ca223c 100644 --- a/src/llm/anthropic_oauth.rs +++ b/src/llm/anthropic_oauth.rs @@ -19,7 +19,8 @@ use crate::llm::costs; use crate::llm::error::LlmError; use crate::llm::provider::{ ChatMessage, CompletionRequest, CompletionResponse, FinishReason, LlmProvider, Role, ToolCall, - ToolCompletionRequest, ToolCompletionResponse, + ToolCompletionRequest, ToolCompletionResponse, strip_unsupported_completion_params, + strip_unsupported_tool_params, }; const ANTHROPIC_API_URL: &str = "https://api.anthropic.com/v1/messages"; @@ -80,28 +81,12 @@ impl AnthropicOAuthProvider { /// Strip unsupported fields from a `CompletionRequest` in place. fn strip_unsupported_completion_params(&self, req: &mut CompletionRequest) { - if self.unsupported_params.is_empty() { - return; - } - if self.unsupported_params.contains("temperature") { - req.temperature = None; - } - if self.unsupported_params.contains("max_tokens") { - req.max_tokens = None; - } + strip_unsupported_completion_params(&self.unsupported_params, req); } /// Strip unsupported fields from a `ToolCompletionRequest` in place. fn strip_unsupported_tool_params(&self, req: &mut ToolCompletionRequest) { - if self.unsupported_params.is_empty() { - return; - } - if self.unsupported_params.contains("temperature") { - req.temperature = None; - } - if self.unsupported_params.contains("max_tokens") { - req.max_tokens = None; - } + strip_unsupported_tool_params(&self.unsupported_params, req); } fn api_url(&self) -> String { diff --git a/src/llm/provider.rs b/src/llm/provider.rs index 40ab8100..787bbff1 100644 --- a/src/llm/provider.rs +++ b/src/llm/provider.rs @@ -455,6 +455,73 @@ pub fn sanitize_tool_messages(messages: &mut [ChatMessage]) { } } +/// Represents a request parameter that may not be supported by all LLM providers. +/// +/// This typed enum replaces stringly-typed parameter names across the codebase, +/// providing type safety and single-point-of-maintenance for parameter handling. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum UnsupportedParam { + Temperature, + MaxTokens, + StopSequences, +} + +impl UnsupportedParam { + /// Get the string name of this parameter for config/error messages. + pub fn name(&self) -> &'static str { + match self { + UnsupportedParam::Temperature => "temperature", + UnsupportedParam::MaxTokens => "max_tokens", + UnsupportedParam::StopSequences => "stop_sequences", + } + } +} + +/// Strip unsupported parameters from a `CompletionRequest` in place. +/// +/// This is the single helper function used by all providers to remove +/// parameters they don't support, replacing duplicate stringly-typed logic. +pub fn strip_unsupported_completion_params( + unsupported: &std::collections::HashSet, + req: &mut CompletionRequest, +) { + if unsupported.is_empty() { + return; + } + if unsupported.contains(UnsupportedParam::Temperature.name()) { + req.temperature = None; + } + if unsupported.contains(UnsupportedParam::MaxTokens.name()) { + req.max_tokens = None; + } + if unsupported.contains(UnsupportedParam::StopSequences.name()) { + req.stop_sequences = None; + } +} + +/// Strip unsupported parameters from a `ToolCompletionRequest` in place. +/// +/// This is the single helper function used by all providers to remove +/// parameters they don't support from tool calls, replacing duplicate stringly-typed logic. +/// +/// Note: Only `Temperature` and `MaxTokens` are supported in `ToolCompletionRequest`. +/// `StopSequences` is only available in `CompletionRequest` and is not applicable to tool calls. +pub fn strip_unsupported_tool_params( + unsupported: &std::collections::HashSet, + req: &mut ToolCompletionRequest, +) { + if unsupported.is_empty() { + return; + } + if unsupported.contains(UnsupportedParam::Temperature.name()) { + req.temperature = None; + } + if unsupported.contains(UnsupportedParam::MaxTokens.name()) { + req.max_tokens = None; + } + // Note: StopSequences is not a field in ToolCompletionRequest, so no action needed +} + #[cfg(test)] mod tests { use super::*; diff --git a/src/llm/registry.rs b/src/llm/registry.rs index 36cb7001..434c698a 100644 --- a/src/llm/registry.rs +++ b/src/llm/registry.rs @@ -113,6 +113,33 @@ impl SetupHint { } } +/// Validates unsupported_params during deserialization. +/// +/// Only allows: "temperature", "max_tokens", "stop_sequences". +/// Invalid parameter names cause a deserialization error. +mod unsupported_params_de { + use serde::{Deserialize, Deserializer}; + + const VALID_PARAMS: &[&str] = &["temperature", "max_tokens", "stop_sequences"]; + + pub fn deserialize<'de, D>(deserializer: D) -> Result, D::Error> + where + D: Deserializer<'de>, + { + let params: Vec = Deserialize::deserialize(deserializer)?; + for param in ¶ms { + if !VALID_PARAMS.contains(¶m.as_str()) { + return Err(serde::de::Error::custom(format!( + "unsupported parameter name '{}': must be one of: {}", + param, + VALID_PARAMS.join(", ") + ))); + } + } + Ok(params) + } +} + /// Declarative definition of an LLM provider. /// /// One JSON object in `providers.json` maps to one `ProviderDefinition`. @@ -155,7 +182,8 @@ pub struct ProviderDefinition { /// Parameter names that this provider does not support (e.g., `["temperature"]`). /// Supported keys: `"temperature"`, `"max_tokens"`, `"stop_sequences"`. /// Listed parameters are stripped from requests before sending to avoid 400 errors. - #[serde(default)] + /// Invalid parameter names cause a deserialization error. + #[serde(default, deserialize_with = "unsupported_params_de::deserialize")] pub unsupported_params: Vec, } @@ -752,7 +780,8 @@ mod tests { "groq should have empty unsupported_params (field absent in JSON)" ); - // Every non-empty entry should contain valid param names + // All entries should only contain valid param names + // (Invalid names should be rejected at deserialization time) for def in &providers { for param in &def.unsupported_params { assert!( @@ -760,10 +789,42 @@ mod tests { "{}: unsupported_params contains empty string", def.id ); + assert!( + matches!( + param.as_str(), + "temperature" | "max_tokens" | "stop_sequences" + ), + "{}: unsupported_params contains invalid parameter '{}'", + def.id, + param + ); } } } + #[test] + fn test_unsupported_params_validation_rejects_invalid() { + // Invalid parameter names should cause deserialization error + let invalid_json = r#"[{ + "id": "test", + "protocol": "open_ai_completions", + "model_env": "TEST_MODEL", + "default_model": "test-model", + "description": "Test provider", + "unsupported_params": ["temperrature"] + }]"#; + + let result: Result, _> = serde_json::from_str(invalid_json); + assert!( + result.is_err(), + "should reject invalid parameter name 'temperrature'" + ); + assert!( + result.err().unwrap().to_string().contains("temperrature"), + "error message should mention the invalid parameter" + ); + } + #[test] fn test_all_builtin_api_key_providers_have_api_key_env() { // Every built-in provider with SetupHint::ApiKey must have api_key_env diff --git a/src/llm/rig_adapter.rs b/src/llm/rig_adapter.rs index 5b835536..41724c31 100644 --- a/src/llm/rig_adapter.rs +++ b/src/llm/rig_adapter.rs @@ -28,7 +28,8 @@ use crate::llm::error::LlmError; use crate::llm::provider::{ ChatMessage, CompletionRequest, CompletionResponse, FinishReason, LlmProvider, ToolCall as IronToolCall, ToolCompletionRequest, ToolCompletionResponse, - ToolDefinition as IronToolDefinition, + ToolDefinition as IronToolDefinition, strip_unsupported_completion_params, + strip_unsupported_tool_params, }; /// Adapter that wraps a rig-core `CompletionModel` and implements `LlmProvider`. @@ -100,31 +101,12 @@ impl RigAdapter { /// Strip unsupported fields from a `CompletionRequest` in place. fn strip_unsupported_completion_params(&self, req: &mut CompletionRequest) { - if self.unsupported_params.is_empty() { - return; - } - if self.unsupported_params.contains("temperature") { - req.temperature = None; - } - if self.unsupported_params.contains("max_tokens") { - req.max_tokens = None; - } - if self.unsupported_params.contains("stop_sequences") { - req.stop_sequences = None; - } + strip_unsupported_completion_params(&self.unsupported_params, req); } /// Strip unsupported fields from a `ToolCompletionRequest` in place. fn strip_unsupported_tool_params(&self, req: &mut ToolCompletionRequest) { - if self.unsupported_params.is_empty() { - return; - } - if self.unsupported_params.contains("temperature") { - req.temperature = None; - } - if self.unsupported_params.contains("max_tokens") { - req.max_tokens = None; - } + strip_unsupported_tool_params(&self.unsupported_params, req); } } diff --git a/src/main.rs b/src/main.rs index 8c771eed..58f7769e 100644 --- a/src/main.rs +++ b/src/main.rs @@ -9,8 +9,8 @@ use ironclaw::{ agent::{Agent, AgentDeps}, app::{AppBuilder, AppBuilderFlags}, channels::{ - ChannelManager, GatewayChannel, HttpChannel, ReplChannel, SignalChannel, WebhookServer, - WebhookServerConfig, + ChannelManager, ChannelSecretUpdater, GatewayChannel, HttpChannel, ReplChannel, + SignalChannel, WebhookServer, WebhookServerConfig, wasm::{WasmChannelRouter, WasmChannelRuntime}, web::log_layer::LogBroadcaster, }, @@ -677,12 +677,21 @@ async fn async_main() -> anyhow::Result<()> { } // Prepare SIGHUP handler for hot-reloading HTTP webhook config + // Broadcast channel for clean shutdown of background tasks + let (shutdown_tx, _) = tokio::sync::broadcast::channel::<()>(1); + #[cfg(unix)] { + // Collect all channels that support secret updates + let mut secret_updaters: Vec> = Vec::new(); + if let Some(ref state) = http_channel_state { + secret_updaters.push(Arc::clone(state) as Arc); + } + let sighup_webhook_server = webhook_server.clone(); - let sighup_http_state = http_channel_state.clone(); let sighup_settings_store_clone = sighup_settings_store.clone(); let sighup_secrets_store = components.secrets_store.clone(); + let mut shutdown_rx = shutdown_tx.subscribe(); tokio::spawn(async move { use tokio::signal::unix::{SignalKind, signal}; @@ -695,10 +704,19 @@ async fn async_main() -> anyhow::Result<()> { }; loop { - sighup.recv().await; + // Exit loop on shutdown signal or when SIGHUP is received + tokio::select! { + _ = shutdown_rx.recv() => { + tracing::debug!("SIGHUP handler shutting down"); + break; + } + _ = sighup.recv() => { + // Handle SIGHUP signal + } + } tracing::info!("SIGHUP received — reloading HTTP webhook config"); - // Inject channel secrets from database into environment variables + // Inject channel secrets from database into thread-safe overlay // (similar to inject_llm_keys_from_secrets for LLM providers) if let Some(ref secrets_store) = sighup_secrets_store { // Inject HTTP webhook secret from encrypted store @@ -706,11 +724,12 @@ async fn async_main() -> anyhow::Result<()> { .get_decrypted("default", "http_webhook_secret") .await { - // Safe: Environment variable modification during runtime SIGHUP reload. - // All threads are synchronized via config reload, not reading env vars directly. - unsafe { - std::env::set_var("HTTP_WEBHOOK_SECRET", webhook_secret.expose()); - } + // Thread-safe: Uses INJECTED_VARS mutex instead of unsafe std::env::set_var + // Config::from_env() will read from the overlay via optional_env() + ironclaw::config::inject_single_var( + "HTTP_WEBHOOK_SECRET", + webhook_secret.expose(), + ); tracing::debug!("Injected HTTP_WEBHOOK_SECRET from secrets store"); } } @@ -750,34 +769,49 @@ async fn async_main() -> anyhow::Result<()> { }; // Restart listener if addr changed + let mut restart_failed = false; if let Some(ref ws_arc) = sighup_webhook_server { - let mut ws = ws_arc.lock().await; - let old_addr = ws.current_addr(); + // Read old address while holding lock, then drop immediately + let old_addr = { + let ws = ws_arc.lock().await; + ws.current_addr() + }; // Lock released here + if old_addr != new_addr { tracing::info!( "SIGHUP: HTTP addr {} -> {}, restarting listener", old_addr, new_addr ); - if let Err(e) = ws.restart_with_addr(new_addr).await { - tracing::error!("SIGHUP: listener restart failed: {}", e); - } else { - tracing::info!("SIGHUP: webhook server restarted on {}", new_addr); + // Wait for restart to complete before proceeding with secret update. + // This ensures atomicity: if restart fails, secret is not updated (partial state corruption). + let mut ws = ws_arc.lock().await; + match ws.restart_with_addr(new_addr).await { + Ok(()) => { + tracing::info!("SIGHUP: webhook server restarted on {}", new_addr); + } + Err(e) => { + tracing::error!("SIGHUP: listener restart failed: {}", e); + restart_failed = true; + } } } else { tracing::debug!("SIGHUP: addr unchanged ({})", old_addr); } } - // Always update secret in-place (zero-downtime) - if let Some(ref state) = sighup_http_state { + // Update secrets in all configured channels (if restart succeeded or wasn't needed) + if !restart_failed { use secrecy::{ExposeSecret, SecretString}; let new_secret = new_http .webhook_secret .as_ref() .map(|s| SecretString::from(s.expose_secret().to_string())); - state.update_secret(new_secret).await; - tracing::info!("SIGHUP: webhook secret updated"); + + // Update all channels that support secret swapping + for updater in &secret_updaters { + updater.update_secret(new_secret.clone()).await; + } } } }); @@ -787,6 +821,9 @@ async fn async_main() -> anyhow::Result<()> { // ── Shutdown ──────────────────────────────────────────────────────── + // Signal background tasks (SIGHUP handler, etc.) to gracefully shut down + let _ = shutdown_tx.send(()); + // Shut down all stdio MCP server child processes. components.mcp_process_manager.shutdown_all().await;