diff --git a/.env.example b/.env.example index 765ea3f6..55c3adb5 100644 --- a/.env.example +++ b/.env.example @@ -18,6 +18,11 @@ DATABASE_POOL_SIZE=10 # === OpenAI Direct === # OPENAI_API_KEY=sk-... +# Reuse Codex CLI auth.json instead of setting OPENAI_API_KEY manually. +# Works with both OpenAI API-key mode and Codex ChatGPT OAuth mode. +# In ChatGPT mode this uses the private `chatgpt.com/backend-api/codex` endpoint. +# LLM_USE_CODEX_AUTH=true +# CODEX_AUTH_PATH=~/.codex/auth.json # === NEAR AI (Chat Completions API) === # Two auth modes: diff --git a/Cargo.lock b/Cargo.lock index dab77b8d..854d103a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3461,6 +3461,7 @@ dependencies = [ "dirs 6.0.0", "dotenvy", "ed25519-dalek", + "eventsource-stream", "flate2", "fs4", "futures", @@ -4364,9 +4365,9 @@ dependencies = [ [[package]] name = "openssl" -version = "0.10.75" +version = "0.10.76" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "08838db121398ad17ab8531ce9de97b244589089e290a384c900cb9ff7434328" +checksum = "951c002c75e16ea2c65b8c7e4d3d51d5530d8dfa7d060b4776828c88cfb18ecf" dependencies = [ "bitflags 2.11.0", "cfg-if", @@ -4402,9 +4403,9 @@ checksum = "7c87def4c32ab89d880effc9e097653c8da5d6ef28e6b539d313baaacfbafcbe" [[package]] name = "openssl-sys" -version = "0.9.111" +version = "0.9.112" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "82cab2d520aa75e3c58898289429321eb788c3106963d0dc886ec7a5f4adc321" +checksum = "57d55af3b3e226502be1526dfdba67ab0e9c96fc293004e79576b2b9edb0dbdb" dependencies = [ "cc", "libc", diff --git a/Cargo.toml b/Cargo.toml index 122c90ec..aef4e687 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -40,6 +40,7 @@ eula = false tokio = { version = "1", features = ["full"] } tokio-stream = { version = "0.1", features = ["sync"] } futures = "0.3" +eventsource-stream = "0.2" # HTTP client reqwest = { version = "0.12", default-features = false, features = ["json", "multipart", "rustls-tls-native-roots", "stream"] } diff --git a/channels-src/telegram/src/lib.rs b/channels-src/telegram/src/lib.rs index d8718ebb..936197bc 100644 --- a/channels-src/telegram/src/lib.rs +++ b/channels-src/telegram/src/lib.rs @@ -100,6 +100,15 @@ struct TelegramMessage { /// Sticker. sticker: Option, + + /// Forum topic ID. Present when the message is sent inside a forum topic. + /// https://core.telegram.org/bots/api#message + #[serde(default)] + message_thread_id: Option, + + /// True when this message is sent inside a forum topic. + #[serde(default)] + is_topic_message: Option, } /// Telegram PhotoSize object. @@ -198,6 +207,10 @@ struct TelegramChat { /// Title for groups/channels. title: Option, + /// True when the supergroup has topics (forum mode) enabled. + #[serde(default)] + is_forum: Option, + /// Username for private chats. username: Option, } @@ -290,6 +303,10 @@ struct TelegramMessageMetadata { /// Whether this is a private (DM) chat. is_private: bool, + + /// Forum topic thread ID (for routing replies back to the correct topic). + #[serde(default, skip_serializing_if = "Option::is_none")] + message_thread_id: Option, } /// Channel configuration injected by host. @@ -680,7 +697,7 @@ impl Guest for TelegramChannel { let metadata: TelegramMessageMetadata = serde_json::from_str(&response.metadata_json) .map_err(|e| format!("Failed to parse metadata: {}", e))?; - send_response(metadata.chat_id, &response, Some(metadata.message_id)) + send_response(metadata.chat_id, &response, Some(metadata.message_id), metadata.message_thread_id) } fn on_broadcast(user_id: String, response: AgentResponse) -> Result<(), String> { @@ -688,7 +705,7 @@ impl Guest for TelegramChannel { .parse() .map_err(|e| format!("Invalid chat_id '{}': {}", user_id, e))?; - send_response(chat_id, &response, None) + send_response(chat_id, &response, None, None) } fn on_status(update: StatusUpdate) { @@ -712,11 +729,17 @@ impl Guest for TelegramChannel { match action { TelegramStatusAction::Typing => { // POST /sendChatAction with action "typing" - let payload = serde_json::json!({ + let mut payload = serde_json::json!({ "chat_id": metadata.chat_id, "action": "typing" }); + // sendChatAction requires message_thread_id even for the General + // topic (id=1), unlike sendMessage which rejects it. + if let Some(thread_id) = metadata.message_thread_id { + payload["message_thread_id"] = serde_json::Value::Number(thread_id.into()); + } + let payload_bytes = match serde_json::to_vec(&payload) { Ok(b) => b, Err(_) => return, @@ -744,7 +767,7 @@ impl Guest for TelegramChannel { TelegramStatusAction::Notify(prompt) => { // Send user-visible status updates for actionable events. if let Err(first_err) = - send_message(metadata.chat_id, &prompt, Some(metadata.message_id), None) + send_message(metadata.chat_id, &prompt, Some(metadata.message_id), None, metadata.message_thread_id) { channel_host::log( channel_host::LogLevel::Warn, @@ -754,7 +777,7 @@ impl Guest for TelegramChannel { ), ); - if let Err(retry_err) = send_message(metadata.chat_id, &prompt, None, None) { + if let Err(retry_err) = send_message(metadata.chat_id, &prompt, None, None, metadata.message_thread_id) { channel_host::log( channel_host::LogLevel::Debug, &format!( @@ -797,6 +820,15 @@ impl std::fmt::Display for SendError { } } +/// Normalize `message_thread_id` for outbound API calls. +/// +/// Telegram rejects `sendMessage` (and other send methods) when +/// `message_thread_id = 1` (the "General" topic). Return `None` in that +/// case so the field is omitted from the payload. +fn normalize_thread_id(thread_id: Option) -> Option { + thread_id.filter(|&id| id != 1) +} + /// Send a message via the Telegram Bot API. /// /// Returns the sent message_id on success. When `parse_mode` is set and @@ -807,7 +839,10 @@ fn send_message( text: &str, reply_to_message_id: Option, parse_mode: Option<&str>, + message_thread_id: Option, ) -> Result { + let message_thread_id = normalize_thread_id(message_thread_id); + let mut payload = serde_json::json!({ "chat_id": chat_id, "text": text, @@ -821,6 +856,10 @@ fn send_message( payload["parse_mode"] = serde_json::Value::String(mode.to_string()); } + if let Some(thread_id) = message_thread_id { + payload["message_thread_id"] = serde_json::Value::Number(thread_id.into()); + } + let payload_bytes = serde_json::to_vec(&payload) .map_err(|e| SendError::Other(format!("Failed to serialize payload: {}", e)))?; @@ -1036,7 +1075,10 @@ fn send_photo( mime_type: &str, data: &[u8], reply_to_message_id: Option, + message_thread_id: Option, ) -> Result<(), String> { + let message_thread_id = normalize_thread_id(message_thread_id); + if data.len() > MAX_PHOTO_SIZE { channel_host::log( channel_host::LogLevel::Info, @@ -1046,7 +1088,7 @@ fn send_photo( data.len() ), ); - return send_document(chat_id, filename, mime_type, data, reply_to_message_id); + return send_document(chat_id, filename, mime_type, data, reply_to_message_id, message_thread_id); } let boundary = format!("ironclaw-{}", channel_host::now_millis()); @@ -1056,6 +1098,9 @@ fn send_photo( if let Some(msg_id) = reply_to_message_id { write_multipart_field(&mut body, &boundary, "reply_to_message_id", &msg_id.to_string()); } + if let Some(thread_id) = message_thread_id { + write_multipart_field(&mut body, &boundary, "message_thread_id", &thread_id.to_string()); + } write_multipart_file(&mut body, &boundary, "photo", filename, mime_type, data); body.extend_from_slice(format!("--{}--\r\n", boundary).as_bytes()); @@ -1097,7 +1142,10 @@ fn send_document( mime_type: &str, data: &[u8], reply_to_message_id: Option, + message_thread_id: Option, ) -> Result<(), String> { + let message_thread_id = normalize_thread_id(message_thread_id); + let boundary = format!("ironclaw-{}", channel_host::now_millis()); let mut body = Vec::new(); @@ -1105,6 +1153,9 @@ fn send_document( if let Some(msg_id) = reply_to_message_id { write_multipart_field(&mut body, &boundary, "reply_to_message_id", &msg_id.to_string()); } + if let Some(thread_id) = message_thread_id { + write_multipart_field(&mut body, &boundary, "message_thread_id", &thread_id.to_string()); + } write_multipart_file(&mut body, &boundary, "document", filename, mime_type, data); body.extend_from_slice(format!("--{}--\r\n", boundary).as_bytes()); @@ -1154,10 +1205,11 @@ fn send_response( chat_id: i64, response: &AgentResponse, reply_to_message_id: Option, + message_thread_id: Option, ) -> Result<(), String> { // Send attachments first (photos/documents) for attachment in &response.attachments { - send_attachment(chat_id, attachment, reply_to_message_id)?; + send_attachment(chat_id, attachment, reply_to_message_id, message_thread_id)?; } // Skip text if empty and we already sent attachments @@ -1166,10 +1218,10 @@ fn send_response( } // Try Markdown, fall back to plain text on parse errors - match send_message(chat_id, &response.content, reply_to_message_id, Some("Markdown")) { + match send_message(chat_id, &response.content, reply_to_message_id, Some("Markdown"), message_thread_id) { Ok(_) => Ok(()), Err(SendError::ParseEntities(_)) => { - send_message(chat_id, &response.content, reply_to_message_id, None) + send_message(chat_id, &response.content, reply_to_message_id, None, message_thread_id) .map(|_| ()) .map_err(|e| format!("Plain-text retry also failed: {}", e)) } @@ -1182,6 +1234,7 @@ fn send_attachment( chat_id: i64, attachment: &Attachment, reply_to_message_id: Option, + message_thread_id: Option, ) -> Result<(), String> { if PHOTO_MIME_TYPES.contains(&attachment.mime_type.as_str()) { send_photo( @@ -1190,6 +1243,7 @@ fn send_attachment( &attachment.mime_type, &attachment.data, reply_to_message_id, + message_thread_id, ) } else { send_document( @@ -1198,6 +1252,7 @@ fn send_attachment( &attachment.mime_type, &attachment.data, reply_to_message_id, + message_thread_id, ) } } @@ -1357,6 +1412,7 @@ fn send_pairing_reply(chat_id: i64, code: &str) -> Result<(), String> { ), None, Some("Markdown"), + None, // Pairing happens in DMs, not forum topics ) .map(|_| ()) .map_err(|e| e.to_string()) @@ -1774,6 +1830,8 @@ fn handle_message(message: TelegramMessage) { } } + let bot_username = channel_host::workspace_read(BOT_USERNAME_PATH).unwrap_or_default(); + // For group chats, only respond if bot was mentioned or respond_to_all is enabled if !is_private { let respond_to_all = channel_host::workspace_read(RESPOND_TO_ALL_GROUP_PATH) @@ -1783,7 +1841,6 @@ fn handle_message(message: TelegramMessage) { if !respond_to_all { let has_command = content.starts_with('/'); - let bot_username = channel_host::workspace_read(BOT_USERNAME_PATH).unwrap_or_default(); let has_bot_mention = if bot_username.is_empty() { content.contains('@') } else { @@ -1814,11 +1871,23 @@ fn handle_message(message: TelegramMessage) { message_id: message.message_id, user_id: from.id, is_private, + message_thread_id: message.message_thread_id, }; let metadata_json = serde_json::to_string(&metadata).unwrap_or_else(|_| "{}".to_string()); - let bot_username = channel_host::workspace_read(BOT_USERNAME_PATH).unwrap_or_default(); + // Compute thread_id for forum topics: "chat_id:topic_id" to prevent + // collisions across different groups (topic IDs are only unique per chat). + // Only use message_thread_id when the chat is a forum — non-forum groups + // also carry message_thread_id for reply threads, which are not topics. + let thread_id = if message.chat.is_forum == Some(true) { + message.message_thread_id.map(|topic_id| { + format!("{}:{}", message.chat.id, topic_id) + }) + } else { + None + }; + let content_to_emit = match content_to_emit_for_agent( &content, if bot_username.is_empty() { @@ -1838,7 +1907,7 @@ fn handle_message(message: TelegramMessage) { user_id: from.id.to_string(), user_name: Some(user_name), content: content_to_emit, - thread_id: None, // Telegram doesn't have threads in the same way + thread_id, metadata_json, attachments, }); @@ -2657,4 +2726,100 @@ mod tests { // Verify the constant is 20 MB, matching the Slack channel limit assert_eq!(MAX_DOWNLOAD_SIZE_BYTES, 20 * 1024 * 1024); } + + // === Forum Topics (thread_id) tests === + + #[test] + fn test_parse_forum_message_with_thread_id() { + let json = r#"{ + "message_id": 100, + "message_thread_id": 42, + "is_topic_message": true, + "from": {"id": 1, "is_bot": false, "first_name": "A"}, + "chat": {"id": -1001234567890, "type": "supergroup", "is_forum": true}, + "text": "Hello from a topic" + }"#; + let msg: TelegramMessage = serde_json::from_str(json).unwrap(); + assert_eq!(msg.message_thread_id, Some(42)); + assert_eq!(msg.is_topic_message, Some(true)); + assert_eq!(msg.chat.is_forum, Some(true)); + } + + #[test] + fn test_parse_non_forum_message_backward_compat() { + let json = r#"{ + "message_id": 1, + "from": {"id": 1, "is_bot": false, "first_name": "A"}, + "chat": {"id": 1, "type": "private"}, + "text": "Hello" + }"#; + let msg: TelegramMessage = serde_json::from_str(json).unwrap(); + assert_eq!(msg.message_thread_id, None); + assert_eq!(msg.is_topic_message, None); + assert_eq!(msg.chat.is_forum, None); + } + + #[test] + fn test_metadata_with_message_thread_id() { + let metadata = TelegramMessageMetadata { + chat_id: -1001234567890, + message_id: 100, + user_id: 42, + is_private: false, + message_thread_id: Some(7), + }; + let json = serde_json::to_string(&metadata).unwrap(); + let parsed: TelegramMessageMetadata = serde_json::from_str(&json).unwrap(); + assert_eq!(parsed.message_thread_id, Some(7)); + } + + #[test] + fn test_metadata_backward_compat_no_thread_id() { + // Old metadata JSON without message_thread_id should deserialize with None + let json = r#"{"chat_id":123,"message_id":1,"user_id":42,"is_private":true}"#; + let metadata: TelegramMessageMetadata = serde_json::from_str(json).unwrap(); + assert_eq!(metadata.message_thread_id, None); + } + + #[test] + fn test_metadata_thread_id_not_serialized_when_none() { + let metadata = TelegramMessageMetadata { + chat_id: 123, + message_id: 1, + user_id: 42, + is_private: true, + message_thread_id: None, + }; + let json = serde_json::to_string(&metadata).unwrap(); + assert!(!json.contains("message_thread_id")); + } + + #[test] + fn test_thread_id_composition() { + // Verify "chat_id:topic_id" format for forum topics + let chat_id: i64 = -1001234567890; + let topic_id: i64 = 42; + let thread_id = format!("{}:{}", chat_id, topic_id); + assert_eq!(thread_id, "-1001234567890:42"); + } + + #[test] + fn test_normalize_thread_id_general_topic() { + // General topic (id=1) must be omitted — Telegram rejects sendMessage + // with message_thread_id=1. + assert_eq!(normalize_thread_id(Some(1)), None); + } + + #[test] + fn test_normalize_thread_id_regular_topic() { + // Non-General topics pass through unchanged + assert_eq!(normalize_thread_id(Some(42)), Some(42)); + assert_eq!(normalize_thread_id(Some(123)), Some(123)); + } + + #[test] + fn test_normalize_thread_id_none() { + // None stays None + assert_eq!(normalize_thread_id(None), None); + } } diff --git a/src/agent/heartbeat.rs b/src/agent/heartbeat.rs index 15c51b61..77bdeadb 100644 --- a/src/agent/heartbeat.rs +++ b/src/agent/heartbeat.rs @@ -26,6 +26,8 @@ use std::sync::Arc; use std::time::Duration; +use chrono::TimeZone as _; +use chrono_tz::Tz; use tokio::sync::mpsc; use crate::channels::OutgoingResponse; @@ -37,7 +39,7 @@ use crate::workspace::hygiene::HygieneConfig; /// Configuration for the heartbeat runner. #[derive(Debug, Clone)] pub struct HeartbeatConfig { - /// Interval between heartbeat checks. + /// Interval between heartbeat checks (used when fire_at is not set). pub interval: Duration, /// Whether heartbeat is enabled. pub enabled: bool, @@ -47,11 +49,13 @@ pub struct HeartbeatConfig { pub notify_user_id: Option, /// Channel to notify on heartbeat findings. pub notify_channel: Option, + /// Fixed time-of-day to fire (24h). When set, interval is ignored. + pub fire_at: Option, /// Hour (0-23) when quiet hours start. pub quiet_hours_start: Option, /// Hour (0-23) when quiet hours end. pub quiet_hours_end: Option, - /// Timezone for quiet hours evaluation (IANA name). + /// Timezone for fire_at and quiet hours evaluation (IANA name). pub timezone: Option, } @@ -63,6 +67,7 @@ impl Default for HeartbeatConfig { max_failures: 3, notify_user_id: None, notify_channel: None, + fire_at: None, quiet_hours_start: None, quiet_hours_end: None, timezone: None, @@ -109,6 +114,21 @@ impl HeartbeatConfig { self.notify_channel = Some(channel.into()); self } + + /// Set a fixed time-of-day to fire (overrides interval). + pub fn with_fire_at(mut self, time: chrono::NaiveTime, tz: Option) -> Self { + self.fire_at = Some(time); + self.timezone = tz; + self + } + + /// Resolve timezone string to chrono_tz::Tz (defaults to UTC). + fn resolved_tz(&self) -> Tz { + self.timezone + .as_deref() + .and_then(crate::timezone::parse_timezone) + .unwrap_or(chrono_tz::UTC) + } } /// Result of a heartbeat check. @@ -124,6 +144,33 @@ pub enum HeartbeatResult { Failed(String), } +/// Compute how long to sleep until the next occurrence of `fire_at` in `tz`. +/// +/// If the target time today is still in the future, sleep until then. +/// Otherwise sleep until the same time tomorrow. +fn duration_until_next_fire(fire_at: chrono::NaiveTime, tz: Tz) -> Duration { + let now = chrono::Utc::now().with_timezone(&tz); + let today = now.date_naive(); + + // Try to build today's target datetime in the given timezone. + // `.earliest()` picks the first occurrence if DST creates ambiguity. + let candidate = tz.from_local_datetime(&today.and_time(fire_at)).earliest(); + + let target = match candidate { + Some(t) if t > now => t, + _ => { + // Already past (or ambiguous) — schedule for tomorrow + let tomorrow = today + chrono::Duration::days(1); + tz.from_local_datetime(&tomorrow.and_time(fire_at)) + .earliest() + .unwrap_or_else(|| now + chrono::Duration::days(1)) + } + }; + + let secs = (target - now).num_seconds().max(1) as u64; + Duration::from_secs(secs) +} + /// Heartbeat runner for proactive periodic execution. pub struct HeartbeatRunner { config: HeartbeatConfig, @@ -175,17 +222,39 @@ impl HeartbeatRunner { return; } - tracing::info!( - "Starting heartbeat loop with interval {:?}", - self.config.interval - ); + // Two scheduling modes: + // fire_at → sleep until the next occurrence (recalculated each iteration) + // interval → tokio::time::interval (drift-free, accounts for loop body time) + let mut tick_interval = if self.config.fire_at.is_none() { + let mut iv = tokio::time::interval(self.config.interval); + // Don't fire immediately on startup. + iv.tick().await; + Some(iv) + } else { + None + }; - let mut interval = tokio::time::interval(self.config.interval); - // Don't run immediately on startup - interval.tick().await; + if let Some(fire_at) = self.config.fire_at { + tracing::info!( + "Starting heartbeat loop: fire daily at {:?} {:?}", + fire_at, + self.config.timezone + ); + } else { + tracing::info!( + "Starting heartbeat loop with interval {:?}", + self.config.interval + ); + } loop { - interval.tick().await; + if let Some(fire_at) = self.config.fire_at { + let sleep_dur = duration_until_next_fire(fire_at, self.config.resolved_tz()); + tracing::info!("Next heartbeat in {:.1}h", sleep_dur.as_secs_f64() / 3600.0); + tokio::time::sleep(sleep_dur).await; + } else if let Some(ref mut iv) = tick_interval { + iv.tick().await; + } // Skip during quiet hours if self.config.is_quiet_hours() { @@ -656,4 +725,63 @@ mod tests { ) -> tokio::task::JoinHandle<()> = spawn_heartbeat; let _ = _fn_ptr; } + + // ==================== fire_at scheduling ==================== + + #[test] + fn test_default_config_has_no_fire_at() { + let config = HeartbeatConfig::default(); + assert!(config.fire_at.is_none()); + // Interval-based scheduling should be the default + assert_eq!(config.interval, Duration::from_secs(30 * 60)); + } + + #[test] + fn test_with_fire_at_builder() { + let time = chrono::NaiveTime::from_hms_opt(9, 0, 0).unwrap(); + let config = + HeartbeatConfig::default().with_fire_at(time, Some("Pacific/Auckland".to_string())); + assert_eq!(config.fire_at, Some(time)); + assert_eq!(config.timezone, Some("Pacific/Auckland".to_string())); + } + + #[test] + fn test_duration_until_next_fire_is_bounded() { + // Result must always be between 1 second and ~24 hours + let time = chrono::NaiveTime::from_hms_opt(14, 0, 0).unwrap(); + let dur = duration_until_next_fire(time, chrono_tz::UTC); + assert!(dur.as_secs() >= 1, "duration must be at least 1 second"); + assert!( + dur.as_secs() <= 86_401, + "duration must be at most ~24 hours, got {}s", + dur.as_secs() + ); + } + + #[test] + fn test_duration_until_next_fire_dst_timezone_no_panic() { + // Use a timezone with DST (US Eastern) — should never panic + let tz: Tz = "America/New_York".parse().unwrap(); + // Test a range of times including midnight boundaries + for hour in [0, 2, 3, 12, 23] { + let time = chrono::NaiveTime::from_hms_opt(hour, 30, 0).unwrap(); + let dur = duration_until_next_fire(time, tz); + assert!(dur.as_secs() >= 1); + assert!(dur.as_secs() <= 86_401); + } + } + + #[test] + fn test_resolved_tz_defaults_to_utc() { + let config = HeartbeatConfig::default(); + assert_eq!(config.resolved_tz(), chrono_tz::UTC); + } + + #[test] + fn test_resolved_tz_parses_iana() { + let time = chrono::NaiveTime::from_hms_opt(9, 0, 0).unwrap(); + let config = + HeartbeatConfig::default().with_fire_at(time, Some("Europe/London".to_string())); + assert_eq!(config.resolved_tz(), chrono_tz::Europe::London); + } } diff --git a/src/config/builder.rs b/src/config/builder.rs index 90bbb185..088db90c 100644 --- a/src/config/builder.rs +++ b/src/config/builder.rs @@ -32,13 +32,16 @@ impl Default for BuilderModeConfig { } impl BuilderModeConfig { - pub(crate) fn resolve() -> Result { + pub(crate) fn resolve(settings: &crate::settings::Settings) -> Result { + let bs = &settings.builder; Ok(Self { - enabled: parse_bool_env("BUILDER_ENABLED", true)?, - build_dir: optional_env("BUILDER_DIR")?.map(PathBuf::from), - max_iterations: parse_optional_env("BUILDER_MAX_ITERATIONS", 20)?, - timeout_secs: parse_optional_env("BUILDER_TIMEOUT_SECS", 600)?, - auto_register: parse_bool_env("BUILDER_AUTO_REGISTER", true)?, + enabled: parse_bool_env("BUILDER_ENABLED", bs.enabled)?, + build_dir: optional_env("BUILDER_DIR")? + .map(PathBuf::from) + .or_else(|| bs.build_dir.clone()), + max_iterations: parse_optional_env("BUILDER_MAX_ITERATIONS", bs.max_iterations)?, + timeout_secs: parse_optional_env("BUILDER_TIMEOUT_SECS", bs.timeout_secs)?, + auto_register: parse_bool_env("BUILDER_AUTO_REGISTER", bs.auto_register)?, }) } @@ -56,3 +59,36 @@ impl BuilderModeConfig { } } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::config::helpers::ENV_MUTEX; + use crate::settings::Settings; + + #[test] + fn resolve_falls_back_to_settings() { + let _guard = ENV_MUTEX.lock().expect("env mutex poisoned"); + let mut settings = Settings::default(); + settings.builder.max_iterations = 99; + settings.builder.auto_register = false; + + let cfg = BuilderModeConfig::resolve(&settings).expect("resolve"); + assert_eq!(cfg.max_iterations, 99); + assert!(!cfg.auto_register); + } + + #[test] + fn env_overrides_settings() { + let _guard = ENV_MUTEX.lock().expect("env mutex poisoned"); + let mut settings = Settings::default(); + settings.builder.timeout_secs = 123; + + // SAFETY: Under ENV_MUTEX, no concurrent env access. + unsafe { std::env::set_var("BUILDER_TIMEOUT_SECS", "3") }; + let cfg = BuilderModeConfig::resolve(&settings).expect("resolve"); + unsafe { std::env::remove_var("BUILDER_TIMEOUT_SECS") }; + + assert_eq!(cfg.timeout_secs, 3); + } +} diff --git a/src/config/heartbeat.rs b/src/config/heartbeat.rs index 3de1da66..1dd456d7 100644 --- a/src/config/heartbeat.rs +++ b/src/config/heartbeat.rs @@ -7,17 +7,19 @@ use crate::settings::Settings; pub struct HeartbeatConfig { /// Whether heartbeat is enabled. pub enabled: bool, - /// Interval between heartbeat checks in seconds. + /// Interval between heartbeat checks in seconds (used when fire_at is not set). pub interval_secs: u64, /// Channel to notify on heartbeat findings. pub notify_channel: Option, /// User ID to notify on heartbeat findings. pub notify_user: Option, + /// Fixed time-of-day to fire (HH:MM, 24h). When set, interval_secs is ignored. + pub fire_at: Option, /// Hour (0-23) when quiet hours start. pub quiet_hours_start: Option, /// Hour (0-23) when quiet hours end. pub quiet_hours_end: Option, - /// Timezone for quiet hours evaluation (IANA name). + /// Timezone for fire_at and quiet hours evaluation (IANA name). pub timezone: Option, } @@ -28,6 +30,7 @@ impl Default for HeartbeatConfig { interval_secs: 1800, // 30 minutes notify_channel: None, notify_user: None, + fire_at: None, quiet_hours_start: None, quiet_hours_end: None, timezone: None, @@ -37,6 +40,19 @@ impl Default for HeartbeatConfig { impl HeartbeatConfig { pub(crate) fn resolve(settings: &Settings) -> Result { + let fire_at_str = + optional_env("HEARTBEAT_FIRE_AT")?.or_else(|| settings.heartbeat.fire_at.clone()); + let fire_at = fire_at_str + .map(|s| { + chrono::NaiveTime::parse_from_str(&s, "%H:%M").map_err(|e| { + ConfigError::InvalidValue { + key: "HEARTBEAT_FIRE_AT".to_string(), + message: format!("must be HH:MM (24h), e.g. '14:00': {e}"), + } + }) + }) + .transpose()?; + Ok(Self { enabled: parse_bool_env("HEARTBEAT_ENABLED", settings.heartbeat.enabled)?, interval_secs: parse_optional_env( @@ -47,6 +63,7 @@ impl HeartbeatConfig { .or_else(|| settings.heartbeat.notify_channel.clone()), notify_user: optional_env("HEARTBEAT_NOTIFY_USER")? .or_else(|| settings.heartbeat.notify_user.clone()), + fire_at, quiet_hours_start: parse_option_env::("HEARTBEAT_QUIET_START")? .or(settings.heartbeat.quiet_hours_start) .map(|h| { diff --git a/src/config/llm.rs b/src/config/llm.rs index 31b8ff4c..4ad24399 100644 --- a/src/config/llm.rs +++ b/src/config/llm.rs @@ -9,7 +9,6 @@ use crate::llm::config::*; use crate::llm::registry::{ProviderProtocol, ProviderRegistry}; use crate::llm::session::SessionConfig; use crate::settings::Settings; - impl LlmConfig { /// Create a test-friendly config without reading env vars. #[cfg(feature = "libsql")] @@ -241,8 +240,30 @@ impl LlmConfig { ) }; - // Resolve API key from env - let api_key = if let Some(env_var) = api_key_env { + // Codex auth.json override: when LLM_USE_CODEX_AUTH=true, + // credentials from the Codex CLI's auth.json take highest priority + // (over env vars AND secrets store). In ChatGPT mode, the base URL + // is also overridden to the private ChatGPT backend endpoint. + let mut codex_base_url_override: Option = None; + let codex_creds = if parse_optional_env("LLM_USE_CODEX_AUTH", false)? { + let path = optional_env("CODEX_AUTH_PATH")? + .map(std::path::PathBuf::from) + .unwrap_or_else(crate::llm::codex_auth::default_codex_auth_path); + crate::llm::codex_auth::load_codex_credentials(&path) + } else { + None + }; + + let codex_refresh_token = codex_creds.as_ref().and_then(|c| c.refresh_token.clone()); + let codex_auth_path = codex_creds.as_ref().and_then(|c| c.auth_path.clone()); + + let api_key = if let Some(creds) = codex_creds { + if creds.is_chatgpt_mode { + codex_base_url_override = Some(creds.base_url().to_string()); + } + Some(creds.token) + } else if let Some(env_var) = api_key_env { + // Resolve API key from env (including secrets store overlay) optional_env(env_var)?.map(SecretString::from) } else { None @@ -259,22 +280,28 @@ impl LlmConfig { } } - // Resolve base URL: env var > settings (backward compat) > registry default - let base_url = if let Some(env_var) = base_url_env { - optional_env(env_var)? - } else { - None - } - .or_else(|| { - // Backward compat: check legacy settings fields - match backend { - "ollama" => settings.ollama_base_url.clone(), - "openai_compatible" | "openrouter" => settings.openai_compatible_base_url.clone(), - _ => None, - } - }) - .or_else(|| default_base_url.map(String::from)) - .unwrap_or_default(); + // Resolve base URL: codex override > env var > settings (backward compat) > registry default + let is_codex_chatgpt = codex_base_url_override.is_some(); + let base_url = codex_base_url_override + .or_else(|| { + if let Some(env_var) = base_url_env { + optional_env(env_var).ok().flatten() + } else { + None + } + }) + .or_else(|| { + // Backward compat: check legacy settings fields + match backend { + "ollama" => settings.ollama_base_url.clone(), + "openai_compatible" | "openrouter" => { + settings.openai_compatible_base_url.clone() + } + _ => None, + } + }) + .or_else(|| default_base_url.map(String::from)) + .unwrap_or_default(); if base_url_required && base_url.is_empty() @@ -340,6 +367,9 @@ impl LlmConfig { model, extra_headers, oauth_token, + is_codex_chatgpt, + refresh_token: codex_refresh_token, + auth_path: codex_auth_path, cache_retention, unsupported_params, }) diff --git a/src/config/mod.rs b/src/config/mod.rs index 52997963..1c81329e 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -317,15 +317,15 @@ impl Config { channels: ChannelsConfig::resolve(settings, tunnel.is_enabled())?, tunnel, agent: AgentConfig::resolve(settings)?, - safety: resolve_safety_config()?, - wasm: WasmConfig::resolve()?, + safety: resolve_safety_config(settings)?, + wasm: WasmConfig::resolve(settings)?, secrets: SecretsConfig::resolve().await?, - builder: BuilderModeConfig::resolve()?, + builder: BuilderModeConfig::resolve(settings)?, heartbeat: HeartbeatConfig::resolve(settings)?, hygiene: HygieneConfig::resolve()?, routines: RoutineConfig::resolve()?, - sandbox: SandboxModeConfig::resolve()?, - claude_code: ClaudeCodeConfig::resolve()?, + sandbox: SandboxModeConfig::resolve(settings)?, + claude_code: ClaudeCodeConfig::resolve(settings)?, skills: SkillsConfig::resolve()?, transcription: TranscriptionConfig::resolve(settings)?, search: WorkspaceSearchConfig::resolve()?, diff --git a/src/config/safety.rs b/src/config/safety.rs index f804d6ad..ff9e900a 100644 --- a/src/config/safety.rs +++ b/src/config/safety.rs @@ -3,9 +3,48 @@ use crate::error::ConfigError; pub use ironclaw_safety::SafetyConfig; -pub(crate) fn resolve_safety_config() -> Result { +pub(crate) fn resolve_safety_config( + settings: &crate::settings::Settings, +) -> Result { + let ss = &settings.safety; Ok(SafetyConfig { - max_output_length: parse_optional_env("SAFETY_MAX_OUTPUT_LENGTH", 100_000)?, - injection_check_enabled: parse_bool_env("SAFETY_INJECTION_CHECK_ENABLED", true)?, + max_output_length: parse_optional_env("SAFETY_MAX_OUTPUT_LENGTH", ss.max_output_length)?, + injection_check_enabled: parse_bool_env( + "SAFETY_INJECTION_CHECK_ENABLED", + ss.injection_check_enabled, + )?, }) } + +#[cfg(test)] +mod tests { + use super::*; + use crate::config::helpers::ENV_MUTEX; + use crate::settings::Settings; + + #[test] + fn resolve_falls_back_to_settings() { + let _guard = ENV_MUTEX.lock().expect("env mutex poisoned"); + let mut settings = Settings::default(); + settings.safety.max_output_length = 42; + settings.safety.injection_check_enabled = false; + + let cfg = resolve_safety_config(&settings).expect("resolve"); + assert_eq!(cfg.max_output_length, 42); + assert!(!cfg.injection_check_enabled); + } + + #[test] + fn env_overrides_settings() { + let _guard = ENV_MUTEX.lock().expect("env mutex poisoned"); + let mut settings = Settings::default(); + settings.safety.max_output_length = 42; + + // SAFETY: Under ENV_MUTEX, no concurrent env access. + unsafe { std::env::set_var("SAFETY_MAX_OUTPUT_LENGTH", "7") }; + let cfg = resolve_safety_config(&settings).expect("resolve"); + unsafe { std::env::remove_var("SAFETY_MAX_OUTPUT_LENGTH") }; + + assert_eq!(cfg.max_output_length, 7); + } +} diff --git a/src/config/sandbox.rs b/src/config/sandbox.rs index e9b7ca76..8c0eb689 100644 --- a/src/config/sandbox.rs +++ b/src/config/sandbox.rs @@ -52,11 +52,20 @@ impl Default for SandboxModeConfig { } impl SandboxModeConfig { - pub(crate) fn resolve() -> Result { + pub(crate) fn resolve(settings: &crate::settings::Settings) -> Result { + let ss = &settings.sandbox; + let extra_domains = optional_env("SANDBOX_EXTRA_DOMAINS")? .map(|s| s.split(',').map(|d| d.trim().to_string()).collect()) - .unwrap_or_default(); + .unwrap_or_else(|| { + if ss.extra_allowed_domains.is_empty() { + Vec::new() + } else { + ss.extra_allowed_domains.clone() + } + }); + // reaper/orphan fields have no Settings counterpart — env > default only. let reaper_interval_secs: u64 = parse_optional_env("SANDBOX_REAPER_INTERVAL_SECS", 300)?; let orphan_threshold_secs: u64 = parse_optional_env("SANDBOX_ORPHAN_THRESHOLD_SECS", 600)?; @@ -76,14 +85,15 @@ impl SandboxModeConfig { } Ok(Self { - enabled: parse_bool_env("SANDBOX_ENABLED", true)?, - policy: parse_string_env("SANDBOX_POLICY", "readonly")?, + enabled: parse_bool_env("SANDBOX_ENABLED", ss.enabled)?, + policy: parse_string_env("SANDBOX_POLICY", ss.policy.clone())?, + // allow_full_access has no Settings counterpart — env > default only. allow_full_access: parse_bool_env("SANDBOX_ALLOW_FULL_ACCESS", false)?, - timeout_secs: parse_optional_env("SANDBOX_TIMEOUT_SECS", 120)?, - memory_limit_mb: parse_optional_env("SANDBOX_MEMORY_LIMIT_MB", 2048)?, - cpu_shares: parse_optional_env("SANDBOX_CPU_SHARES", 1024)?, - image: parse_string_env("SANDBOX_IMAGE", "ironclaw-worker:latest")?, - auto_pull_image: parse_bool_env("SANDBOX_AUTO_PULL", true)?, + timeout_secs: parse_optional_env("SANDBOX_TIMEOUT_SECS", ss.timeout_secs)?, + memory_limit_mb: parse_optional_env("SANDBOX_MEMORY_LIMIT_MB", ss.memory_limit_mb)?, + cpu_shares: parse_optional_env("SANDBOX_CPU_SHARES", ss.cpu_shares)?, + image: parse_string_env("SANDBOX_IMAGE", ss.image.clone())?, + auto_pull_image: parse_bool_env("SANDBOX_AUTO_PULL", ss.auto_pull_image)?, extra_allowed_domains: extra_domains, reaper_interval_secs, orphan_threshold_secs, @@ -200,7 +210,7 @@ impl ClaudeCodeConfig { /// Load from environment variables only (used inside containers where /// there is no database or full config). pub fn from_env() -> Self { - match Self::resolve() { + match Self::resolve_env_only() { Ok(c) => c, Err(e) => { tracing::warn!("Failed to resolve ClaudeCodeConfig: {e}, using defaults"); @@ -253,7 +263,33 @@ impl ClaudeCodeConfig { None } - pub(crate) fn resolve() -> Result { + pub(crate) fn resolve(settings: &crate::settings::Settings) -> Result { + let defaults = Self::default(); + Ok(Self { + // Use settings.sandbox.claude_code_enabled as fallback (written by setup wizard). + enabled: parse_bool_env("CLAUDE_CODE_ENABLED", settings.sandbox.claude_code_enabled)?, + config_dir: optional_env("CLAUDE_CONFIG_DIR")? + .map(std::path::PathBuf::from) + .unwrap_or(defaults.config_dir), + model: parse_string_env("CLAUDE_CODE_MODEL", defaults.model)?, + max_turns: parse_optional_env("CLAUDE_CODE_MAX_TURNS", defaults.max_turns)?, + memory_limit_mb: parse_optional_env( + "CLAUDE_CODE_MEMORY_LIMIT_MB", + defaults.memory_limit_mb, + )?, + allowed_tools: optional_env("CLAUDE_CODE_ALLOWED_TOOLS")? + .map(|s| { + s.split(',') + .map(|t| t.trim().to_string()) + .filter(|t| !t.is_empty()) + .collect() + }) + .unwrap_or(defaults.allowed_tools), + }) + } + + /// Resolve from env vars only, no Settings. Used inside containers. + fn resolve_env_only() -> Result { let defaults = Self::default(); Ok(Self { enabled: parse_bool_env("CLAUDE_CODE_ENABLED", defaults.enabled)?, @@ -554,6 +590,80 @@ mod tests { ); } + // ── Settings fallback tests ────────────────────────────────────── + + #[test] + fn sandbox_resolve_falls_back_to_settings() { + let _guard = crate::config::helpers::ENV_MUTEX + .lock() + .expect("env mutex poisoned"); + let mut settings = crate::settings::Settings::default(); + settings.sandbox.cpu_shares = 99; + settings.sandbox.auto_pull_image = false; + settings.sandbox.enabled = false; + + let cfg = SandboxModeConfig::resolve(&settings).expect("resolve"); + assert!(!cfg.enabled); + assert_eq!(cfg.cpu_shares, 99); + assert!(!cfg.auto_pull_image); + } + + #[test] + fn sandbox_env_overrides_settings() { + let _guard = crate::config::helpers::ENV_MUTEX + .lock() + .expect("env mutex poisoned"); + let mut settings = crate::settings::Settings::default(); + settings.sandbox.timeout_secs = 999; + + // SAFETY: Under ENV_MUTEX, no concurrent env access. + unsafe { std::env::set_var("SANDBOX_TIMEOUT_SECS", "5") }; + let cfg = SandboxModeConfig::resolve(&settings).expect("resolve"); + unsafe { std::env::remove_var("SANDBOX_TIMEOUT_SECS") }; + + assert_eq!(cfg.timeout_secs, 5); + } + + // ── ClaudeCodeConfig settings fallback tests ──────────────────── + + #[test] + fn claude_code_resolve_uses_settings_enabled() { + let _guard = crate::config::helpers::ENV_MUTEX + .lock() + .expect("env mutex poisoned"); + let mut settings = crate::settings::Settings::default(); + settings.sandbox.claude_code_enabled = true; + + let cfg = ClaudeCodeConfig::resolve(&settings).expect("resolve"); + assert!(cfg.enabled); + } + + #[test] + fn claude_code_resolve_defaults_disabled() { + let _guard = crate::config::helpers::ENV_MUTEX + .lock() + .expect("env mutex poisoned"); + let settings = crate::settings::Settings::default(); + let cfg = ClaudeCodeConfig::resolve(&settings).expect("resolve"); + assert!(!cfg.enabled); + } + + #[test] + fn claude_code_env_overrides_settings() { + let _guard = crate::config::helpers::ENV_MUTEX + .lock() + .expect("env mutex poisoned"); + let mut settings = crate::settings::Settings::default(); + settings.sandbox.claude_code_enabled = true; + + // SAFETY: Under ENV_MUTEX, no concurrent env access. + unsafe { std::env::set_var("CLAUDE_CODE_ENABLED", "false") }; + let cfg = ClaudeCodeConfig::resolve(&settings).expect("resolve"); + unsafe { std::env::remove_var("CLAUDE_CODE_ENABLED") }; + + assert!(!cfg.enabled); + } + #[test] fn test_readonly_policy_unaffected() { let config = SandboxModeConfig { diff --git a/src/config/wasm.rs b/src/config/wasm.rs index 224f2e95..a9bfbd35 100644 --- a/src/config/wasm.rs +++ b/src/config/wasm.rs @@ -44,20 +44,30 @@ fn default_tools_dir() -> PathBuf { } impl WasmConfig { - pub(crate) fn resolve() -> Result { + pub(crate) fn resolve(settings: &crate::settings::Settings) -> Result { + let ws = &settings.wasm; Ok(Self { - enabled: parse_bool_env("WASM_ENABLED", true)?, + enabled: parse_bool_env("WASM_ENABLED", ws.enabled)?, tools_dir: optional_env("WASM_TOOLS_DIR")? .map(PathBuf::from) + .or_else(|| ws.tools_dir.clone()) .unwrap_or_else(default_tools_dir), default_memory_limit: parse_optional_env( "WASM_DEFAULT_MEMORY_LIMIT", - 10 * 1024 * 1024, + ws.default_memory_limit, )?, - default_timeout_secs: parse_optional_env("WASM_DEFAULT_TIMEOUT_SECS", 60)?, - default_fuel_limit: parse_optional_env("WASM_DEFAULT_FUEL_LIMIT", 10_000_000)?, - cache_compiled: parse_bool_env("WASM_CACHE_COMPILED", true)?, - cache_dir: optional_env("WASM_CACHE_DIR")?.map(PathBuf::from), + default_timeout_secs: parse_optional_env( + "WASM_DEFAULT_TIMEOUT_SECS", + ws.default_timeout_secs, + )?, + default_fuel_limit: parse_optional_env( + "WASM_DEFAULT_FUEL_LIMIT", + ws.default_fuel_limit, + )?, + cache_compiled: parse_bool_env("WASM_CACHE_COMPILED", ws.cache_compiled)?, + cache_dir: optional_env("WASM_CACHE_DIR")? + .map(PathBuf::from) + .or_else(|| ws.cache_dir.clone()), }) } @@ -81,3 +91,36 @@ impl WasmConfig { } } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::config::helpers::ENV_MUTEX; + use crate::settings::Settings; + + #[test] + fn resolve_falls_back_to_settings() { + let _guard = ENV_MUTEX.lock().expect("env mutex poisoned"); + let mut settings = Settings::default(); + settings.wasm.default_memory_limit = 42; + settings.wasm.cache_compiled = false; + + let cfg = WasmConfig::resolve(&settings).expect("resolve"); + assert_eq!(cfg.default_memory_limit, 42); + assert!(!cfg.cache_compiled); + } + + #[test] + fn env_overrides_settings() { + let _guard = ENV_MUTEX.lock().expect("env mutex poisoned"); + let mut settings = Settings::default(); + settings.wasm.default_fuel_limit = 42; + + // SAFETY: Under ENV_MUTEX, no concurrent env access. + unsafe { std::env::set_var("WASM_DEFAULT_FUEL_LIMIT", "7") }; + let cfg = WasmConfig::resolve(&settings).expect("resolve"); + unsafe { std::env::remove_var("WASM_DEFAULT_FUEL_LIMIT") }; + + assert_eq!(cfg.default_fuel_limit, 7); + } +} diff --git a/src/context/state.rs b/src/context/state.rs index 22aca311..768e4da6 100644 --- a/src/context/state.rs +++ b/src/context/state.rs @@ -48,6 +48,14 @@ impl JobState { pub fn can_transition_to(&self, target: JobState) -> bool { use JobState::*; + // Allow idempotent Completed -> Completed transition. + // Both the execution loop and the worker wrapper may race to mark a + // job complete; the second call should be a harmless no-op rather + // than an error that masks the successful completion. + if matches!((self, target), (Completed, Completed)) { + return true; + } + matches!( (self, target), // From Pending @@ -238,6 +246,18 @@ impl JobContext { )); } + // Idempotent: already in the target state, skip recording a duplicate + // transition. This handles the Completed -> Completed race between + // execution_loop and the worker wrapper. + if self.state == new_state { + tracing::debug!( + job_id = %self.job_id, + state = %self.state, + "idempotent state transition (already in target state), skipping" + ); + return Ok(()); + } + let transition = StateTransition { from: self.state, to: new_state, @@ -340,6 +360,45 @@ mod tests { assert!(!JobState::Accepted.can_transition_to(JobState::InProgress)); } + #[test] + fn test_completed_to_completed_is_idempotent() { + // Regression test for the race condition where both execution_loop + // and the worker wrapper call mark_completed(). The second call + // must succeed without error and must not record a duplicate + // transition. + let mut ctx = JobContext::new("Test", "Idempotent completion test"); + ctx.transition_to(JobState::InProgress, None).unwrap(); + ctx.transition_to(JobState::Completed, Some("first".into())) + .unwrap(); + assert_eq!(ctx.state, JobState::Completed); + let transitions_before = ctx.transitions.len(); + + // Second Completed -> Completed must be a no-op + let result = ctx.transition_to(JobState::Completed, Some("duplicate".into())); + assert!( + result.is_ok(), + "Completed -> Completed should be idempotent" + ); + assert_eq!(ctx.state, JobState::Completed); + assert_eq!( + ctx.transitions.len(), + transitions_before, + "idempotent transition should not record a new history entry" + ); + } + + #[test] + fn test_other_self_transitions_still_rejected() { + // Ensure we only allow Completed -> Completed, not arbitrary X -> X. + assert!(!JobState::Pending.can_transition_to(JobState::Pending)); + assert!(!JobState::InProgress.can_transition_to(JobState::InProgress)); + assert!(!JobState::Failed.can_transition_to(JobState::Failed)); + assert!(!JobState::Stuck.can_transition_to(JobState::Stuck)); + assert!(!JobState::Submitted.can_transition_to(JobState::Submitted)); + assert!(!JobState::Accepted.can_transition_to(JobState::Accepted)); + assert!(!JobState::Cancelled.can_transition_to(JobState::Cancelled)); + } + #[test] fn test_terminal_states() { assert!(JobState::Accepted.is_terminal()); diff --git a/src/extensions/manager.rs b/src/extensions/manager.rs index 13067b77..163e3a5e 100644 --- a/src/extensions/manager.rs +++ b/src/extensions/manager.rs @@ -57,6 +57,17 @@ struct ChannelRuntimeState { } /// Central manager for extension lifecycle operations. +/// +/// # Initialization Order +/// +/// Relay-channel restoration depends on a channel manager being injected first. +/// Call one of the following before `restore_relay_channels()`: +/// +/// 1. [`ExtensionManager::set_channel_runtime`] (also sets relay manager), or +/// 2. [`ExtensionManager::set_relay_channel_manager`]. +/// +/// If `restore_relay_channels()` runs first, each restore attempt fails with +/// "Channel manager not initialized" and channels remain inactive. pub struct ExtensionManager { registry: ExtensionRegistry, discovery: OnlineDiscovery, @@ -312,6 +323,9 @@ impl ExtensionManager { /// /// Call this when WASM channel runtime is not available but relay channels /// still need to be hot-added. + /// + /// This must be called before [`ExtensionManager::restore_relay_channels`] + /// unless [`ExtensionManager::set_channel_runtime`] was already called. pub async fn set_relay_channel_manager(&self, channel_manager: Arc) { *self.relay_channel_manager.write().await = Some(channel_manager); } @@ -356,7 +370,10 @@ impl ExtensionManager { /// /// Loads the persisted active channel list, filters to relay types (those with /// a stored stream token), and activates each via `activate_stored_relay()`. - /// Skips channels that are already active. Call this after `set_relay_channel_manager()`. + /// Skips channels that are already active. + /// + /// Call this only after `set_relay_channel_manager()` or `set_channel_runtime()`. + /// Otherwise, each activation attempt fails with "Channel manager not initialized". pub async fn restore_relay_channels(&self) { let persisted = self.load_persisted_active_channels().await; let already_active = self.active_channel_names.read().await.clone(); diff --git a/src/llm/CLAUDE.md b/src/llm/CLAUDE.md index d1b9eea2..38d69010 100644 --- a/src/llm/CLAUDE.md +++ b/src/llm/CLAUDE.md @@ -7,8 +7,12 @@ Multi-provider LLM integration with circuit breaker, retry, failover, and respon | File | Role | |------|------| | `mod.rs` | Provider factory (`create_llm_provider`, `build_provider_chain`); `LlmBackend` enum | +| `config.rs` | LLM config types (`LlmConfig`, `RegistryProviderConfig`, `NearAiConfig`, `BedrockConfig`) | +| `error.rs` | `LlmError` enum used by all providers | | `provider.rs` | `LlmProvider` trait, `ChatMessage`, `ToolCall`, `CompletionRequest`, `sanitize_tool_messages` | | `nearai_chat.rs` | NEAR AI Chat Completions provider (dual auth: session token or API key) | +| `codex_auth.rs` | Reads Codex CLI `auth.json`, extracts tokens, refreshes ChatGPT OAuth access tokens | +| `codex_chatgpt.rs` | Custom Responses API provider for Codex ChatGPT backend (`/backend-api/codex`) | | `reasoning.rs` | `Reasoning` struct, `ReasoningContext`, `RespondResult`, `ActionPlan`, `ToolSelection`; thinking-tag stripping; `SILENT_REPLY_TOKEN` | | `session.rs` | NEAR AI session token management with disk + DB persistence, OAuth login flow | | `circuit_breaker.rs` | Circuit breaker: Closed → Open → HalfOpen state machine | @@ -35,6 +39,12 @@ Set via `LLM_BACKEND` env var: | `tinfoil` | Tinfoil TEE inference | `TINFOIL_API_KEY`, `TINFOIL_MODEL` | | `bedrock` | AWS Bedrock (requires `--features bedrock`) | `BEDROCK_REGION`, `BEDROCK_MODEL`, `AWS_PROFILE` | +Codex auth reuse: +- Set `LLM_USE_CODEX_AUTH=true` to load credentials from `~/.codex/auth.json` (override with `CODEX_AUTH_PATH`). +- If Codex is logged in with API-key mode, IronClaw uses the standard OpenAI endpoint. +- If Codex is logged in with ChatGPT OAuth mode, IronClaw routes to the private `chatgpt.com/backend-api/codex` Responses API via `codex_chatgpt.rs`. +- ChatGPT mode supports one automatic 401 refresh using the refresh token persisted in `auth.json`. + ## AWS Bedrock Provider Uses the native Converse API via `aws-sdk-bedrockruntime` (`bedrock.rs`). Requires `--features bedrock` at build time — not in default features due to heavy AWS SDK dependencies. diff --git a/src/llm/anthropic_oauth.rs b/src/llm/anthropic_oauth.rs index 12ca223c..12c527f1 100644 --- a/src/llm/anthropic_oauth.rs +++ b/src/llm/anthropic_oauth.rs @@ -34,7 +34,9 @@ const DEFAULT_MAX_TOKENS: u32 = 8192; /// Anthropic provider using OAuth Bearer authentication. pub struct AnthropicOAuthProvider { client: Client, - token: SecretString, + /// OAuth token, wrapped in RwLock so it can be updated after a successful + /// Keychain refresh (fixes #1136: stale token reuse after expiry). + token: std::sync::RwLock, model: String, base_url: Option, active_model: std::sync::RwLock, @@ -71,7 +73,7 @@ impl AnthropicOAuthProvider { Ok(Self { client, - token, + token: std::sync::RwLock::new(token), model: config.model.clone(), base_url, active_model, @@ -98,6 +100,22 @@ impl AnthropicOAuthProvider { } } + /// Read the current token from the RwLock. + fn current_token(&self) -> String { + match self.token.read() { + Ok(guard) => guard.expose_secret().to_string(), + Err(poisoned) => poisoned.into_inner().expose_secret().to_string(), + } + } + + /// Update the stored token after a successful Keychain refresh. + fn update_token(&self, new_token: SecretString) { + match self.token.write() { + Ok(mut guard) => *guard = new_token, + Err(poisoned) => *poisoned.into_inner() = new_token, + } + } + async fn send_request Deserialize<'de>>( &self, body: &AnthropicRequest, @@ -109,7 +127,7 @@ impl AnthropicOAuthProvider { let response = self .client .post(&url) - .bearer_auth(self.token.expose_secret()) + .bearer_auth(self.current_token()) .header("anthropic-version", ANTHROPIC_API_VERSION) .header("anthropic-beta", ANTHROPIC_OAUTH_BETA) .header("Content-Type", "application/json") @@ -141,6 +159,11 @@ impl AnthropicOAuthProvider { // OAuth tokens from `claude login` expire in ~8-12h. Attempt // to re-extract a fresh token from the OS credential store // (macOS Keychain / Linux credentials file) before giving up. + // + // Brief delay to give Claude Code time to complete its async + // Keychain refresh write (fixes race in #1136). + tokio::time::sleep(std::time::Duration::from_millis(500)).await; + if let Some(fresh) = crate::config::ClaudeCodeConfig::extract_oauth_token() { let fresh_token = SecretString::from(fresh); // Retry once with the refreshed token @@ -159,6 +182,11 @@ impl AnthropicOAuthProvider { reason: e.to_string(), })?; if retry.status().is_success() { + // Persist the refreshed token so subsequent requests + // don't hit 401 again (fixes #1136). + self.update_token(fresh_token); + tracing::info!("Anthropic OAuth token refreshed from credential store"); + let text = retry.text().await.map_err(|e| LlmError::RequestFailed { provider: "anthropic_oauth".to_string(), reason: format!("Failed to read response body: {}", e), @@ -659,4 +687,22 @@ mod tests { assert_eq!(tool_calls.len(), 1); assert_eq!(tool_calls[0].name, "search"); } + + /// Regression test for #1136: token field must be mutable via RwLock + /// so that a refreshed token persists across subsequent requests. + #[test] + fn test_token_update_persists() { + let original = SecretString::from("old_token".to_string()); + let token = std::sync::RwLock::new(original); + + // Read the original + assert_eq!(token.read().unwrap().expose_secret(), "old_token"); + + // Simulate a successful refresh + let refreshed = SecretString::from("new_token".to_string()); + *token.write().unwrap() = refreshed; + + // Subsequent reads see the updated token + assert_eq!(token.read().unwrap().expose_secret(), "new_token"); + } } diff --git a/src/llm/codex_auth.rs b/src/llm/codex_auth.rs new file mode 100644 index 00000000..6f302436 --- /dev/null +++ b/src/llm/codex_auth.rs @@ -0,0 +1,377 @@ +//! Read Codex CLI credentials for LLM authentication. +//! +//! When `LLM_USE_CODEX_AUTH=true`, IronClaw reads the Codex CLI's +//! `auth.json` file (default: `~/.codex/auth.json`) and extracts +//! credentials. This lets IronClaw piggyback on a Codex login without +//! implementing its own OAuth flow. +//! +//! Codex supports two auth modes: +//! - **API key** (`auth_mode: "apiKey"`) → uses `OPENAI_API_KEY` field +//! against `api.openai.com/v1`. +//! - **ChatGPT** (`auth_mode: "chatgpt"`) → uses `tokens.access_token` +//! (OAuth JWT) against `chatgpt.com/backend-api/codex`. +//! +//! When in ChatGPT mode, the provider supports automatic token refresh +//! on 401 responses using the `refresh_token` from `auth.json`. + +use std::path::{Path, PathBuf}; + +use secrecy::{ExposeSecret, SecretString}; +use serde::{Deserialize, Serialize}; + +/// ChatGPT backend API endpoint used by Codex in ChatGPT auth mode. +const CHATGPT_BACKEND_URL: &str = "https://chatgpt.com/backend-api/codex"; + +/// Standard OpenAI API endpoint used by Codex in API key mode. +const OPENAI_API_URL: &str = "https://api.openai.com/v1"; + +/// OAuth token refresh endpoint (same as Codex CLI). +const REFRESH_TOKEN_URL: &str = "https://auth.openai.com/oauth/token"; + +/// OAuth client ID used for token refresh (same as Codex CLI). +const CLIENT_ID: &str = "app_EMoamEEZ73f0CkXaXp7hrann"; + +/// Credentials extracted from Codex's `auth.json`. +#[derive(Debug, Clone)] +pub struct CodexCredentials { + /// The bearer token (API key or ChatGPT access_token). + pub token: SecretString, + /// Whether this is a ChatGPT OAuth token (vs. an OpenAI API key). + pub is_chatgpt_mode: bool, + /// OAuth refresh token (only present in ChatGPT mode). + pub refresh_token: Option, + /// Path to the auth.json file (for persisting refreshed tokens). + pub auth_path: Option, +} + +impl CodexCredentials { + /// Returns the correct base URL for the auth mode. + /// + /// - ChatGPT mode → `https://chatgpt.com/backend-api/codex` + /// - API key mode → `https://api.openai.com/v1` + pub fn base_url(&self) -> &'static str { + if self.is_chatgpt_mode { + CHATGPT_BACKEND_URL + } else { + OPENAI_API_URL + } + } +} + +/// Partial representation of Codex's `$CODEX_HOME/auth.json`. +#[derive(Debug, Deserialize)] +struct CodexAuthJson { + auth_mode: Option, + #[serde(rename = "OPENAI_API_KEY")] + openai_api_key: Option, + tokens: Option, +} + +#[derive(Debug, Deserialize)] +struct CodexTokens { + access_token: SecretString, + refresh_token: Option, +} + +/// Request body for OAuth token refresh. +#[derive(Serialize)] +struct RefreshRequest<'a> { + client_id: &'a str, + grant_type: &'a str, + refresh_token: &'a str, +} + +/// Response from the OAuth token refresh endpoint. +#[derive(Debug, Deserialize)] +struct RefreshResponse { + access_token: SecretString, + refresh_token: Option, +} + +/// Default path used by Codex CLI: `~/.codex/auth.json`. +pub fn default_codex_auth_path() -> PathBuf { + let home_dir = dirs::home_dir().unwrap_or_else(|| { + tracing::warn!( + "Could not determine home directory; falling back to current working directory for Codex auth.json path" + ); + PathBuf::from(".") + }); + + home_dir.join(".codex").join("auth.json") +} + +/// Load credentials from a Codex `auth.json` file. +/// +/// Returns `None` if the file is missing, unreadable, or contains +/// no usable credentials. +pub fn load_codex_credentials(path: &Path) -> Option { + let content = match std::fs::read_to_string(path) { + Ok(c) => c, + Err(e) => { + tracing::debug!("Could not read Codex auth file {}: {}", path.display(), e); + return None; + } + }; + + let auth: CodexAuthJson = match serde_json::from_str(&content) { + Ok(a) => a, + Err(e) => { + tracing::warn!("Failed to parse Codex auth file {}: {}", path.display(), e); + return None; + } + }; + + let is_chatgpt = auth + .auth_mode + .as_deref() + .map(|m| m == "chatgpt" || m == "chatgptAuthTokens") + .unwrap_or(false); + + // API key mode: use OPENAI_API_KEY field. + if !is_chatgpt { + if let Some(key) = auth.openai_api_key.filter(|k| !k.is_empty()) { + tracing::info!("Loaded API key from Codex auth.json (API key mode)"); + return Some(CodexCredentials { + token: SecretString::from(key), + is_chatgpt_mode: false, + refresh_token: None, + auth_path: None, + }); + } + // If auth_mode was explicitly `apiKey`, do not fall back to checking for a token. + if auth.auth_mode.is_some() { + return None; + } + } + + // ChatGPT mode: use access_token as bearer token. + if let Some(tokens) = auth.tokens + && !tokens.access_token.expose_secret().is_empty() + { + tracing::info!( + "Loaded access token from Codex auth.json (ChatGPT mode, base_url={})", + CHATGPT_BACKEND_URL + ); + return Some(CodexCredentials { + token: tokens.access_token, + is_chatgpt_mode: true, + refresh_token: tokens.refresh_token, + auth_path: Some(path.to_path_buf()), + }); + } + + tracing::debug!( + "Codex auth.json at {} contains no usable credentials", + path.display() + ); + None +} + +/// Attempt to refresh an expired access token using the refresh token. +/// +/// On success, returns the new `access_token` and persists the refreshed +/// tokens back to `auth.json`. This follows the same OAuth protocol as +/// Codex CLI (`POST https://auth.openai.com/oauth/token`). +/// +/// Returns `None` if the refresh token is missing, the request fails, +/// or the response is malformed. +pub async fn refresh_access_token( + client: &reqwest::Client, + refresh_token: &SecretString, + auth_path: Option<&Path>, +) -> Option { + let req = RefreshRequest { + client_id: CLIENT_ID, + grant_type: "refresh_token", + refresh_token: refresh_token.expose_secret(), + }; + + tracing::info!("Attempting to refresh Codex OAuth access token"); + + let resp = match client + .post(REFRESH_TOKEN_URL) + .header("Content-Type", "application/json") + .json(&req) + .timeout(std::time::Duration::from_secs(10)) + .send() + .await + { + Ok(r) => r, + Err(e) => { + tracing::warn!("Token refresh request failed: {e}"); + return None; + } + }; + + if !resp.status().is_success() { + let status = resp.status(); + let body = resp.text().await.unwrap_or_default(); + tracing::warn!("Token refresh failed: HTTP {status}: {body}"); + if status.as_u16() == 401 { + tracing::warn!( + "Refresh token may be expired or revoked. \ + Please re-authenticate with: codex --login" + ); + } + return None; + } + + let refresh_resp: RefreshResponse = match resp.json().await { + Ok(r) => r, + Err(e) => { + tracing::warn!("Failed to parse token refresh response: {e}"); + return None; + } + }; + + let new_access_token = refresh_resp.access_token.clone(); + + // Persist refreshed tokens back to auth.json + if let Some(path) = auth_path { + if let Err(e) = persist_refreshed_tokens( + path, + refresh_resp.access_token.expose_secret(), + refresh_resp + .refresh_token + .as_ref() + .map(ExposeSecret::expose_secret), + ) { + tracing::warn!( + "Failed to persist refreshed tokens to {}: {e}", + path.display() + ); + } else { + tracing::info!("Refreshed tokens persisted to {}", path.display()); + } + } + + Some(new_access_token) +} + +/// Update `auth.json` with refreshed tokens, preserving other fields. +fn persist_refreshed_tokens( + path: &Path, + new_access_token: &str, + new_refresh_token: Option<&str>, +) -> Result<(), Box> { + let content = std::fs::read_to_string(path)?; + let mut json: serde_json::Value = serde_json::from_str(&content)?; + + if let Some(tokens) = json.get_mut("tokens") { + tokens["access_token"] = serde_json::Value::String(new_access_token.to_string()); + if let Some(rt) = new_refresh_token { + tokens["refresh_token"] = serde_json::Value::String(rt.to_string()); + } + } + + let updated = serde_json::to_string_pretty(&json)?; + let tmp_path = path.with_extension("json.tmp"); + std::fs::write(&tmp_path, updated)?; + if let Err(e) = std::fs::rename(&tmp_path, path) { + let _ = std::fs::remove_file(&tmp_path); + return Err(Box::new(e)); + } + set_auth_file_permissions(path)?; + Ok(()) +} + +#[cfg(unix)] +fn set_auth_file_permissions(path: &Path) -> Result<(), Box> { + use std::os::unix::fs::PermissionsExt; + + std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600))?; + Ok(()) +} + +#[cfg(not(unix))] +fn set_auth_file_permissions(_path: &Path) -> Result<(), Box> { + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::io::Write; + use tempfile::NamedTempFile; + + #[test] + fn loads_api_key_mode() { + let mut f = NamedTempFile::new().unwrap(); + writeln!( + f, + r#"{{"auth_mode":"apiKey","OPENAI_API_KEY":"sk-test-123"}}"# + ) + .unwrap(); + let creds = load_codex_credentials(f.path()).expect("should load"); + assert_eq!(creds.token.expose_secret(), "sk-test-123"); + assert!(!creds.is_chatgpt_mode); + assert_eq!(creds.base_url(), OPENAI_API_URL); + } + + #[test] + fn loads_chatgpt_mode() { + let mut f = NamedTempFile::new().unwrap(); + writeln!( + f, + r#"{{"auth_mode":"chatgpt","tokens":{{"id_token":{{}},"access_token":"eyJ-test","refresh_token":"rt-x"}}}}"# + ) + .unwrap(); + let creds = load_codex_credentials(f.path()).expect("should load"); + assert_eq!(creds.token.expose_secret(), "eyJ-test"); + assert!(creds.is_chatgpt_mode); + assert_eq!( + creds + .refresh_token + .as_ref() + .expect("refresh token should be present") + .expose_secret(), + "rt-x" + ); + assert_eq!(creds.base_url(), CHATGPT_BACKEND_URL); + } + + #[test] + fn api_key_mode_ignores_tokens() { + let mut f = NamedTempFile::new().unwrap(); + writeln!( + f, + r#"{{"auth_mode":"apiKey","OPENAI_API_KEY":"sk-priority","tokens":{{"id_token":{{}},"access_token":"eyJ-fallback","refresh_token":"rt-x"}}}}"# + ) + .unwrap(); + let creds = load_codex_credentials(f.path()).expect("should load"); + assert_eq!(creds.token.expose_secret(), "sk-priority"); + assert!(!creds.is_chatgpt_mode); + } + + #[test] + fn returns_none_for_missing_file() { + assert!(load_codex_credentials(Path::new("/tmp/nonexistent_codex_auth.json")).is_none()); + } + + #[test] + fn returns_none_for_empty_json() { + let mut f = NamedTempFile::new().unwrap(); + writeln!(f, "{{}}").unwrap(); + assert!(load_codex_credentials(f.path()).is_none()); + } + + #[test] + fn returns_none_for_empty_key() { + let mut f = NamedTempFile::new().unwrap(); + writeln!(f, r#"{{"auth_mode":"apiKey","OPENAI_API_KEY":""}}"#).unwrap(); + assert!(load_codex_credentials(f.path()).is_none()); + } + + #[test] + fn api_key_mode_missing_key_does_not_fallback_to_chatgpt() { + // Bug: if auth_mode is "apiKey" but key is missing, the old code would + // fall through to check for a ChatGPT token, returning is_chatgpt_mode: true. + let mut f = NamedTempFile::new().unwrap(); + writeln!( + f, + r#"{{"auth_mode":"apiKey","OPENAI_API_KEY":"","tokens":{{"id_token":{{}},"access_token":"eyJ-bad","refresh_token":"rt-x"}}}}"# + ) + .unwrap(); + assert!(load_codex_credentials(f.path()).is_none()); + } +} diff --git a/src/llm/codex_chatgpt.rs b/src/llm/codex_chatgpt.rs new file mode 100644 index 00000000..56cb3378 --- /dev/null +++ b/src/llm/codex_chatgpt.rs @@ -0,0 +1,932 @@ +//! Codex ChatGPT Responses API provider. +//! +//! Implements `LlmProvider` by speaking the OpenAI Responses API protocol +//! (`POST /responses`) used by the ChatGPT backend at +//! `chatgpt.com/backend-api/codex`. This bypasses `rig-core`'s Chat +//! Completions path, which is incompatible with this endpoint. +//! +//! # Warning +//! +//! The ChatGPT backend endpoint (`chatgpt.com/backend-api/codex`) is a +//! **private, undocumented API**. Using subscriber OAuth tokens from a +//! third-party application may violate the token's intended scope or +//! OpenAI's Terms of Service. This feature is provided as-is for +//! convenience and may break without notice. + +use async_trait::async_trait; +use eventsource_stream::Eventsource; +use futures::{Stream, StreamExt}; +use reqwest::Client; +use rust_decimal::Decimal; +use secrecy::{ExposeSecret, SecretString}; +use serde_json::{Value, json}; +use std::path::PathBuf; +use std::time::Duration; +use tokio::sync::{Mutex, RwLock}; + +use super::codex_auth; +use crate::error::LlmError; + +use super::provider::{ + ChatMessage, CompletionRequest, CompletionResponse, ContentPart, FinishReason, LlmProvider, + Role, ToolCall, ToolCompletionRequest, ToolCompletionResponse, ToolDefinition, +}; + +/// Provider that speaks the Responses API protocol against the ChatGPT backend. +pub struct CodexChatGptProvider { + client: Client, + base_url: String, + api_key: RwLock, + /// User-configured model name (or empty/"default" for auto-detect). + configured_model: String, + /// Lazily resolved model name (populated on first LLM call). + resolved_model: tokio::sync::OnceCell, + /// OAuth refresh token for automatic 401 retry. + refresh_token: Option, + /// Path to auth.json for persisting refreshed tokens. + auth_path: Option, + /// Timeout for actual `/responses` requests. + request_timeout: Duration, + /// Prevent concurrent 401 handlers from racing the same refresh token. + refresh_lock: Mutex<()>, +} + +impl CodexChatGptProvider { + #[cfg(test)] + fn new(base_url: &str, api_key: &str, model: &str) -> Self { + Self { + client: Client::new(), + base_url: base_url.trim_end_matches('/').to_string(), + api_key: RwLock::new(SecretString::from(api_key.to_string())), + configured_model: model.to_string(), + resolved_model: tokio::sync::OnceCell::const_new(), + refresh_token: None, + auth_path: None, + request_timeout: Duration::from_secs(120), + refresh_lock: Mutex::new(()), + } + } + + /// Create a provider with lazy model detection. + /// + /// The model is **not** resolved during construction. Instead, it is + /// resolved on the first LLM call via [`resolve_model`], avoiding the + /// need for `block_in_place` / `block_on` during provider setup. + /// + /// **Model selection priority** (applied at resolution time): + /// 1. If `configured_model` is non-empty, validate it against the + /// `/models` endpoint. If it isn't in the supported list, log a + /// warning with available models and fall back to the top model. + /// 2. If `configured_model` is empty (or a generic placeholder like + /// "default"), auto-detect the highest-priority model from the API. + pub fn with_lazy_model( + base_url: &str, + api_key: SecretString, + configured_model: &str, + refresh_token: Option, + auth_path: Option, + request_timeout_secs: u64, + ) -> Self { + tracing::warn!( + "Codex ChatGPT provider uses a private, undocumented API \ + (chatgpt.com/backend-api/codex). This may violate OpenAI's \ + Terms of Service and could break without notice." + ); + + Self { + client: Client::new(), + base_url: base_url.trim_end_matches('/').to_string(), + api_key: RwLock::new(api_key), + configured_model: configured_model.to_string(), + resolved_model: tokio::sync::OnceCell::const_new(), + refresh_token, + auth_path, + request_timeout: Duration::from_secs(request_timeout_secs), + refresh_lock: Mutex::new(()), + } + } + + /// Resolve the model to use, lazily on first call. + /// + /// Uses `OnceCell` so the `/models` fetch happens at most once. + async fn resolve_model(&self) -> &str { + self.resolved_model + .get_or_init(|| async { + let api_key = self.api_key.read().await.clone(); + let available = Self::fetch_available_models(&self.client, &self.base_url, &api_key) + .await; + + let configured = &self.configured_model; + if !configured.is_empty() && configured != "default" { + // User explicitly configured a model — validate it + if available.is_empty() { + tracing::warn!( + "Could not fetch model list; using configured model '{configured}'" + ); + return configured.clone(); + } + if available.iter().any(|m| m == configured) { + tracing::info!(model = %configured, "Codex ChatGPT: using configured model"); + return configured.clone(); + } + tracing::warn!( + configured = %configured, + available = ?available, + "Configured model not found in supported list, falling back to top model" + ); + available + .into_iter() + .next() + .unwrap_or_else(|| configured.clone()) + } else { + // No user preference — auto-detect + if let Some(top) = available.into_iter().next() { + tracing::info!(model = %top, "Codex ChatGPT: auto-detected model"); + top + } else { + tracing::warn!( + "Could not auto-detect model, using fallback '{configured}'" + ); + configured.clone() + } + } + }) + .await + } + + /// Query `/models?client_version=0.111.0` and return the list of available + /// model slugs, ordered by priority (highest first). + async fn fetch_available_models( + client: &Client, + base_url: &str, + api_key: &SecretString, + ) -> Vec { + let url = format!("{base_url}/models?client_version=0.111.0"); + let resp = match client + .get(&url) + .bearer_auth(api_key.expose_secret()) + .timeout(Duration::from_secs(10)) + .send() + .await + { + Ok(r) => r, + Err(e) => { + tracing::warn!("Failed to fetch Codex models: {e}"); + return Vec::new(); + } + }; + if !resp.status().is_success() { + tracing::warn!(status = %resp.status(), "Failed to fetch Codex models"); + return Vec::new(); + } + let body: Value = match resp.json().await { + Ok(v) => v, + Err(_) => return Vec::new(), + }; + // The response has { "models": [ { "slug": "...", ... }, ... ] } + body.get("models") + .and_then(|m| m.as_array()) + .map(|models| { + models + .iter() + .filter_map(|m| { + m.get("slug") + .and_then(|s| s.as_str()) + .map(|s| s.to_string()) + }) + .collect() + }) + .unwrap_or_default() + } + + /// Convert IronClaw messages to Responses API request JSON. + fn build_request_body( + &self, + model: &str, + messages: &[ChatMessage], + tools: &[ToolDefinition], + tool_choice: Option<&str>, + ) -> Value { + // Extract system instructions + let instructions: String = messages + .iter() + .filter(|m| m.role == Role::System) + .map(|m| m.content.as_str()) + .collect::>() + .join("\n\n"); + + // Convert non-system messages to Responses API input items + let input: Vec = messages + .iter() + .filter(|m| m.role != Role::System) + .flat_map(Self::message_to_input_items) + .collect(); + + // Convert tool definitions + let api_tools: Vec = tools + .iter() + .map(|t| { + json!({ + "type": "function", + "name": t.name, + "description": t.description, + "parameters": t.parameters, + }) + }) + .collect(); + + let mut body = json!({ + "model": model, + "instructions": instructions, + "input": input, + "stream": true, + "store": false, + }); + + if !api_tools.is_empty() { + body["tools"] = json!(api_tools); + body["tool_choice"] = json!(tool_choice.unwrap_or("auto")); + } + + body + } + + /// Convert a single ChatMessage to one or more Responses API input items. + fn message_to_input_items(msg: &ChatMessage) -> Vec { + let mut items = Vec::new(); + + match msg.role { + Role::User => { + // Build content array: if content_parts is populated, use it + // to include multimodal content (images). Otherwise fall back + // to the plain text content field. + let content = if !msg.content_parts.is_empty() { + msg.content_parts + .iter() + .map(|part| match part { + ContentPart::Text { text } => json!({ + "type": "input_text", + "text": text, + }), + ContentPart::ImageUrl { image_url } => json!({ + "type": "input_image", + "image_url": image_url.url, + }), + }) + .collect::>() + } else { + vec![json!({ + "type": "input_text", + "text": msg.content, + })] + }; + + items.push(json!({ + "type": "message", + "role": "user", + "content": content, + })); + } + Role::Assistant => { + // If the assistant message has tool calls, emit function_call items + if let Some(ref tool_calls) = msg.tool_calls { + // Emit the assistant text as a message if non-empty + if !msg.content.is_empty() { + items.push(json!({ + "type": "message", + "role": "assistant", + "content": [{ + "type": "output_text", + "text": msg.content, + }], + })); + } + for tc in tool_calls { + let args = if tc.arguments.is_string() { + tc.arguments.as_str().unwrap_or("{}").to_string() + } else { + serde_json::to_string(&tc.arguments).unwrap_or_default() + }; + items.push(json!({ + "type": "function_call", + "name": tc.name, + "arguments": args, + "call_id": tc.id, + })); + } + } else { + items.push(json!({ + "type": "message", + "role": "assistant", + "content": [{ + "type": "output_text", + "text": msg.content, + }], + })); + } + } + Role::Tool => { + items.push(json!({ + "type": "function_call_output", + "call_id": msg.tool_call_id.as_deref().unwrap_or(""), + "output": msg.content, + })); + } + Role::System => { + // System messages are handled via `instructions` field + } + } + + items + } + + /// Send a request and parse the SSE response. + /// + /// On HTTP 401, if a refresh token is available, attempts to refresh + /// the access token and retry the request once. + async fn send_request(&self, body: Value) -> Result { + let url = format!("{}/responses", self.base_url); + + tracing::debug!( + url = %url, + model = %body.get("model").and_then(|m| m.as_str()).unwrap_or("?"), + "Codex ChatGPT: sending request" + ); + + let api_key = self.api_key.read().await.clone(); + let resp = + Self::send_http_request(&self.client, &url, &api_key, &body, self.request_timeout) + .await?; + + let status = resp.status(); + if status.as_u16() == 401 { + // Attempt token refresh if we have a refresh token + if let Some(ref rt) = self.refresh_token { + let _refresh_guard = self.refresh_lock.lock().await; + let current_token = self.api_key.read().await.clone(); + + if current_token.expose_secret() != api_key.expose_secret() { + tracing::info!("Received 401, but another request already refreshed the token"); + let retry_resp = Self::send_http_request( + &self.client, + &url, + ¤t_token, + &body, + self.request_timeout, + ) + .await?; + let retry_status = retry_resp.status(); + if !retry_status.is_success() { + let body_text = + tokio::time::timeout(Duration::from_secs(5), retry_resp.text()) + .await + .unwrap_or(Ok(String::new())) + .unwrap_or_default(); + return Err(LlmError::RequestFailed { + provider: "codex_chatgpt".to_string(), + reason: format!( + "HTTP {retry_status} from {url} (after concurrent token refresh): {body_text}" + ), + }); + } + return Self::parse_sse_response_stream(retry_resp, self.request_timeout).await; + } + + tracing::info!("Received 401, attempting token refresh"); + if let Some(new_token) = + codex_auth::refresh_access_token(&self.client, rt, self.auth_path.as_deref()) + .await + { + // Update stored api_key + *self.api_key.write().await = new_token.clone(); + tracing::info!("Token refreshed, retrying request"); + + // Retry the request with the new token + let retry_resp = Self::send_http_request( + &self.client, + &url, + &new_token, + &body, + self.request_timeout, + ) + .await?; + + let retry_status = retry_resp.status(); + if !retry_status.is_success() { + let body_text = + tokio::time::timeout(Duration::from_secs(5), retry_resp.text()) + .await + .unwrap_or(Ok(String::new())) + .unwrap_or_default(); + return Err(LlmError::RequestFailed { + provider: "codex_chatgpt".to_string(), + reason: format!( + "HTTP {retry_status} from {url} (after token refresh): {body_text}" + ), + }); + } + + return Self::parse_sse_response_stream(retry_resp, self.request_timeout).await; + } else { + tracing::warn!( + "Token refresh failed. Please re-authenticate with: codex --login" + ); + } + } + + // No refresh token or refresh failed — return the 401 error + // Drain the response body to release the connection + let _ = resp.text().await; + return Err(LlmError::AuthFailed { + provider: "codex_chatgpt".to_string(), + }); + } + + if !status.is_success() { + // Read the error body with a timeout to avoid hanging + let body_text = tokio::time::timeout(Duration::from_secs(5), resp.text()) + .await + .unwrap_or(Ok(String::new())) + .unwrap_or_default(); + return Err(LlmError::RequestFailed { + provider: "codex_chatgpt".to_string(), + reason: format!("HTTP {status} from {url}: {body_text}",), + }); + } + + Self::parse_sse_response_stream(resp, self.request_timeout).await + } + + /// Low-level HTTP POST to the /responses endpoint. + async fn send_http_request( + client: &Client, + url: &str, + api_key: &SecretString, + body: &Value, + timeout: Duration, + ) -> Result { + client + .post(url) + .bearer_auth(api_key.expose_secret()) + .header("Content-Type", "application/json") + .header("Accept", "text/event-stream") + .json(body) + .timeout(timeout) + .send() + .await + .map_err(|e| LlmError::RequestFailed { + provider: "codex_chatgpt".to_string(), + reason: format!("HTTP request failed: {e}"), + }) + } + + async fn parse_sse_response_stream( + resp: reqwest::Response, + idle_timeout: Duration, + ) -> Result { + let stream = resp + .bytes_stream() + .map(|chunk| chunk.map_err(|e| e.to_string())); + Self::parse_sse_stream(stream, idle_timeout).await + } + + async fn parse_sse_stream( + stream: S, + idle_timeout: Duration, + ) -> Result + where + S: Stream> + Unpin, + { + let mut result = ResponsesResult::default(); + let mut stream = stream.eventsource(); + + loop { + match tokio::time::timeout(idle_timeout, stream.next()).await { + Ok(Some(Ok(event))) => { + let data = event.data.trim(); + if data.is_empty() { + continue; + } + + let parsed: Value = match serde_json::from_str(data) { + Ok(v) => v, + Err(_) => continue, + }; + + if Self::handle_sse_event(&mut result, event.event.as_str(), &parsed) { + return Ok(result); + } + } + Ok(Some(Err(e))) => { + return Err(LlmError::RequestFailed { + provider: "codex_chatgpt".to_string(), + reason: format!("Failed to read SSE stream: {e}"), + }); + } + Ok(None) => return Ok(result), + Err(_) => { + return Err(LlmError::RequestFailed { + provider: "codex_chatgpt".to_string(), + reason: format!( + "Timed out waiting for SSE event after {}s", + idle_timeout.as_secs() + ), + }); + } + } + } + } + + /// Parse SSE events from the response text. + #[cfg(test)] + fn parse_sse_response(sse_text: &str) -> Result { + let mut result = ResponsesResult::default(); + let mut current_event_type = String::new(); + + for line in sse_text.lines() { + if let Some(event) = line.strip_prefix("event: ") { + current_event_type = event.trim().to_string(); + continue; + } + + if let Some(data) = line.strip_prefix("data: ") { + let data = data.trim(); + if data.is_empty() { + continue; + } + + let parsed: Value = match serde_json::from_str(data) { + Ok(v) => v, + Err(_) => continue, + }; + + if Self::handle_sse_event(&mut result, current_event_type.as_str(), &parsed) { + return Ok(result); + } + } + } + + Ok(result) + } + + fn handle_sse_event(result: &mut ResponsesResult, event_type: &str, parsed: &Value) -> bool { + match event_type { + "response.output_text.delta" => { + if let Some(delta) = parsed.get("delta").and_then(|d| d.as_str()) { + result.text.push_str(delta); + } + } + "response.output_item.added" => { + // Capture function call metadata when the item is first added. + // The item has: id (item_id), call_id, name, type. + let item = parsed.get("item").unwrap_or(parsed); + if item.get("type").and_then(|t| t.as_str()) == Some("function_call") { + let item_id = item + .get("id") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(); + let call_id = item + .get("call_id") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(); + let name = item + .get("name") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(); + + result + .pending_tool_calls + .entry(item_id) + .or_insert_with(|| PendingToolCall { + call_id, + name, + arguments: String::new(), + }); + } + } + "response.function_call_arguments.delta" => { + // Delta events use `item_id` (not `call_id`) + if let Some(item_id) = parsed.get("item_id").and_then(|v| v.as_str()) + && let Some(entry) = result.pending_tool_calls.get_mut(item_id) + && let Some(delta) = parsed.get("delta").and_then(|d| d.as_str()) + { + entry.arguments.push_str(delta); + } + } + "response.completed" => { + if let Some(response) = parsed.get("response") + && let Some(usage) = response.get("usage") + { + result.input_tokens = usage + .get("input_tokens") + .and_then(|v| v.as_u64()) + .unwrap_or(0) as u32; + result.output_tokens = usage + .get("output_tokens") + .and_then(|v| v.as_u64()) + .unwrap_or(0) as u32; + } + return true; + } + _ => {} + } + + false + } + + /// Remove keys with empty-string values from a JSON object. + /// + /// gpt-5.2-codex fills optional tool parameters with `""` (e.g. + /// `"timestamp": ""`). IronClaw's tool validation treats these as + /// invalid "non-empty input expected". Stripping them makes the + /// tool see only the actually-provided values. + fn strip_empty_string_values(value: Value) -> Value { + match value { + Value::Object(map) => { + let cleaned: serde_json::Map = map + .into_iter() + .filter(|(_, v)| !matches!(v, Value::String(s) if s.is_empty())) + .map(|(k, v)| (k, Self::strip_empty_string_values(v))) + .collect(); + Value::Object(cleaned) + } + other => other, + } + } +} + +#[derive(Debug, Default)] +struct ResponsesResult { + text: String, + /// Keyed by item_id (the SSE item identifier, e.g. "fc_..."). + pending_tool_calls: std::collections::HashMap, + input_tokens: u32, + output_tokens: u32, +} + +#[derive(Debug)] +struct PendingToolCall { + /// The call_id from the API (e.g. "call_..."), used to match results. + call_id: String, + name: String, + arguments: String, +} + +#[async_trait] +impl LlmProvider for CodexChatGptProvider { + fn model_name(&self) -> &str { + // Return resolved model if available, otherwise the configured name. + self.resolved_model + .get() + .map(|s| s.as_str()) + .unwrap_or(&self.configured_model) + } + + fn cost_per_token(&self) -> (Decimal, Decimal) { + // ChatGPT backend doesn't expose per-token pricing + (Decimal::ZERO, Decimal::ZERO) + } + + async fn complete(&self, request: CompletionRequest) -> Result { + let model = self.resolve_model().await; + let body = self.build_request_body(model, &request.messages, &[], None); + let result = self.send_request(body).await?; + + Ok(CompletionResponse { + content: result.text, + input_tokens: result.input_tokens, + output_tokens: result.output_tokens, + finish_reason: FinishReason::Stop, + cache_read_input_tokens: 0, + cache_creation_input_tokens: 0, + }) + } + + async fn complete_with_tools( + &self, + request: ToolCompletionRequest, + ) -> Result { + let model = self.resolve_model().await; + let body = self.build_request_body( + model, + &request.messages, + &request.tools, + request.tool_choice.as_deref(), + ); + let result = self.send_request(body).await?; + + let tool_calls: Vec = result + .pending_tool_calls + .into_values() + .map(|tc| { + let args: Value = + serde_json::from_str(&tc.arguments).unwrap_or_else(|_| json!(tc.arguments)); + // gpt-5.2-codex fills optional parameters with empty strings (e.g. + // `"timestamp": ""`), which IronClaw's tool validation rejects. + // Strip them so only actually-provided values reach the tool. + let args = Self::strip_empty_string_values(args); + ToolCall { + id: tc.call_id, + name: tc.name, + arguments: args, + } + }) + .collect(); + + let finish_reason = if tool_calls.is_empty() { + FinishReason::Stop + } else { + FinishReason::ToolUse + }; + + Ok(ToolCompletionResponse { + content: if result.text.is_empty() { + None + } else { + Some(result.text) + }, + tool_calls, + input_tokens: result.input_tokens, + output_tokens: result.output_tokens, + finish_reason, + cache_read_input_tokens: 0, + cache_creation_input_tokens: 0, + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use bytes::Bytes; + use futures::stream; + + #[test] + fn test_message_conversion_user() { + let items = CodexChatGptProvider::message_to_input_items(&ChatMessage::user("hello")); + assert_eq!(items.len(), 1); + assert_eq!(items[0]["type"], "message"); + assert_eq!(items[0]["role"], "user"); + assert_eq!(items[0]["content"][0]["type"], "input_text"); + assert_eq!(items[0]["content"][0]["text"], "hello"); + } + + #[test] + fn test_message_conversion_user_with_image() { + use super::super::provider::ImageUrl; + let parts = vec![ + ContentPart::Text { + text: "What's in this image?".to_string(), + }, + ContentPart::ImageUrl { + image_url: ImageUrl { + url: "data:image/png;base64,iVBOR...".to_string(), + detail: None, + }, + }, + ]; + let msg = ChatMessage::user_with_parts("", parts); + let items = CodexChatGptProvider::message_to_input_items(&msg); + assert_eq!(items.len(), 1); + assert_eq!(items[0]["type"], "message"); + assert_eq!(items[0]["role"], "user"); + let content = items[0]["content"].as_array().unwrap(); + assert_eq!(content.len(), 2); + assert_eq!(content[0]["type"], "input_text"); + assert_eq!(content[0]["text"], "What's in this image?"); + assert_eq!(content[1]["type"], "input_image"); + assert_eq!(content[1]["image_url"], "data:image/png;base64,iVBOR..."); + } + #[test] + fn test_message_conversion_assistant() { + let items = CodexChatGptProvider::message_to_input_items(&ChatMessage::assistant("hi")); + assert_eq!(items.len(), 1); + assert_eq!(items[0]["type"], "message"); + assert_eq!(items[0]["role"], "assistant"); + assert_eq!(items[0]["content"][0]["type"], "output_text"); + } + + #[test] + fn test_message_conversion_tool_result() { + let msg = ChatMessage::tool_result("call_1", "search", "result text"); + let items = CodexChatGptProvider::message_to_input_items(&msg); + assert_eq!(items.len(), 1); + assert_eq!(items[0]["type"], "function_call_output"); + assert_eq!(items[0]["call_id"], "call_1"); + assert_eq!(items[0]["output"], "result text"); + } + + #[test] + fn test_message_conversion_assistant_with_tool_calls() { + let tc = ToolCall { + id: "call_1".to_string(), + name: "search".to_string(), + arguments: json!({"query": "rust"}), + }; + let msg = ChatMessage::assistant_with_tool_calls(Some("thinking...".into()), vec![tc]); + let items = CodexChatGptProvider::message_to_input_items(&msg); + // Should produce: 1 text message + 1 function_call + assert_eq!(items.len(), 2); + assert_eq!(items[0]["type"], "message"); + assert_eq!(items[1]["type"], "function_call"); + assert_eq!(items[1]["name"], "search"); + assert_eq!(items[1]["call_id"], "call_1"); + } + + #[test] + fn test_build_request_extracts_system_as_instructions() { + let provider = CodexChatGptProvider::new("https://example.com", "key", "gpt-4o"); + let messages = vec![ + ChatMessage::system("You are helpful."), + ChatMessage::user("hello"), + ]; + let body = provider.build_request_body("gpt-4o", &messages, &[], None); + assert_eq!(body["instructions"], "You are helpful."); + // input should only contain the user message, not the system message + assert_eq!(body["input"].as_array().unwrap().len(), 1); + // store must be false for ChatGPT backend + assert_eq!(body["store"], false); + } + + #[test] + fn test_parse_sse_text_response() { + let sse = r#"event: response.output_text.delta +data: {"delta":"Hello"} + +event: response.output_text.delta +data: {"delta":" world!"} + +event: response.completed +data: {"response":{"usage":{"input_tokens":10,"output_tokens":5}}} + +"#; + let result = CodexChatGptProvider::parse_sse_response(sse).unwrap(); + assert_eq!(result.text, "Hello world!"); + assert_eq!(result.input_tokens, 10); + assert_eq!(result.output_tokens, 5); + assert!(result.pending_tool_calls.is_empty()); + } + + #[test] + fn test_parse_sse_tool_call() { + // Real API format: output_item.added has item.id (item_id) + item.call_id, + // delta events use item_id (not call_id) + let sse = r#"event: response.output_item.added +data: {"item":{"id":"fc_1","type":"function_call","call_id":"call_1","name":"search"}} + +event: response.function_call_arguments.delta +data: {"item_id":"fc_1","delta":"{\"query\":"} + +event: response.function_call_arguments.delta +data: {"item_id":"fc_1","delta":"\"rust\"}"} + +event: response.completed +data: {"response":{"usage":{"input_tokens":20,"output_tokens":15}}} + +"#; + let result = CodexChatGptProvider::parse_sse_response(sse).unwrap(); + assert!(result.text.is_empty()); + assert_eq!(result.pending_tool_calls.len(), 1); + let tc = result.pending_tool_calls.get("fc_1").unwrap(); + assert_eq!(tc.call_id, "call_1"); + assert_eq!(tc.name, "search"); + assert_eq!(tc.arguments, "{\"query\":\"rust\"}"); + } + + #[tokio::test] + async fn test_parse_sse_stream_response() { + let stream = stream::iter(vec![ + Ok(Bytes::from_static( + b"event: response.output_text.delta\ndata: {\"delta\":\"Hello\"}\n\n", + )), + Ok(Bytes::from_static( + b"event: response.output_text.delta\ndata: {\"delta\":\" world\"}\n\n", + )), + Ok(Bytes::from_static( + b"event: response.completed\ndata: {\"response\":{\"usage\":{\"input_tokens\":3,\"output_tokens\":2}}}\n\n", + )), + ]); + + let result = CodexChatGptProvider::parse_sse_stream(stream, Duration::from_secs(1)) + .await + .unwrap(); + assert_eq!(result.text, "Hello world"); + assert_eq!(result.input_tokens, 3); + assert_eq!(result.output_tokens, 2); + } + + #[test] + fn test_strip_empty_string_values() { + let input = json!({ + "format": "%Y-%m-%d", + "operation": "now", + "timestamp": "", + "timestamp2": "", + }); + let cleaned = CodexChatGptProvider::strip_empty_string_values(input); + assert_eq!(cleaned, json!({"format": "%Y-%m-%d", "operation": "now"})); + } +} diff --git a/src/llm/config.rs b/src/llm/config.rs index a3e76ef7..8b7d41c3 100644 --- a/src/llm/config.rs +++ b/src/llm/config.rs @@ -5,6 +5,8 @@ //! extracted into a standalone crate. Resolution logic (reading env vars, //! settings) lives in `crate::config::llm`. +use std::path::PathBuf; + use secrecy::SecretString; use crate::llm::registry::ProviderProtocol; @@ -85,6 +87,13 @@ pub struct RegistryProviderConfig { /// OAuth token for providers that support Bearer auth (e.g. Anthropic via `claude login`). /// When set, the provider factory routes to the OAuth-specific provider implementation. pub oauth_token: Option, + /// When true, route OpenAI-compatible traffic to the Codex ChatGPT + /// Responses API provider instead of rig-core's Chat Completions path. + pub is_codex_chatgpt: bool, + /// OAuth refresh token for Codex ChatGPT token refresh. + pub refresh_token: Option, + /// Path to Codex auth.json for persisting refreshed tokens. + pub auth_path: Option, /// Prompt cache retention (Anthropic-specific). pub cache_retention: CacheRetention, /// Parameter names that this provider does not support (e.g., `["temperature"]`). diff --git a/src/llm/mod.rs b/src/llm/mod.rs index 3dccf221..ef2009c4 100644 --- a/src/llm/mod.rs +++ b/src/llm/mod.rs @@ -12,6 +12,8 @@ mod anthropic_oauth; #[cfg(feature = "bedrock")] mod bedrock; pub mod circuit_breaker; +pub(crate) mod codex_auth; +mod codex_chatgpt; pub mod config; pub mod costs; pub mod error; @@ -104,7 +106,7 @@ pub async fn create_llm_provider( provider: config.backend.clone(), })?; - create_registry_provider(reg_config) + create_registry_provider(reg_config, timeout) } /// Create an LLM provider from a `NearAiConfig` directly. @@ -142,7 +144,13 @@ pub fn create_llm_provider_with_config( /// `create_*_provider` functions. fn create_registry_provider( config: &RegistryProviderConfig, + request_timeout_secs: u64, ) -> Result, LlmError> { + // Codex ChatGPT mode: use the Responses API provider + if config.is_codex_chatgpt { + return create_codex_chatgpt_from_registry(config, request_timeout_secs); + } + match config.protocol { ProviderProtocol::OpenAiCompletions => create_openai_compat_from_registry(config), ProviderProtocol::Anthropic => create_anthropic_from_registry(config), @@ -150,6 +158,36 @@ fn create_registry_provider( } } +fn create_codex_chatgpt_from_registry( + config: &RegistryProviderConfig, + request_timeout_secs: u64, +) -> Result, LlmError> { + let api_key = config + .api_key + .as_ref() + .cloned() + .ok_or_else(|| LlmError::AuthFailed { + provider: "codex_chatgpt".to_string(), + })?; + + tracing::info!( + configured_model = %config.model, + base_url = %config.base_url, + "Using Codex ChatGPT provider (Responses API) — model detection deferred to first call" + ); + + let provider = codex_chatgpt::CodexChatGptProvider::with_lazy_model( + &config.base_url, + api_key, + &config.model, + config.refresh_token.clone(), + config.auth_path.clone(), + request_timeout_secs, + ); + + Ok(Arc::new(provider)) +} + #[cfg(feature = "bedrock")] async fn create_bedrock_provider(config: &LlmConfig) -> Result, LlmError> { let br = config @@ -165,6 +203,7 @@ async fn create_bedrock_provider(config: &LlmConfig) -> Result (Option, Vec { - // Tool result message: wrap as User { ToolResult } + // Tool result message: wrap as User { ToolResult }. + // Merge consecutive tool results into a single User message + // so the API sees one multi-result message instead of + // multiple consecutive User messages (which Anthropic rejects). let tool_id = normalized_tool_call_id(msg.tool_call_id.as_deref(), history.len()); - history.push(RigMessage::User { - content: OneOrMany::one(UserContent::ToolResult(RigToolResult { - id: tool_id.clone(), - call_id: Some(tool_id), - content: OneOrMany::one(ToolResultContent::text(&msg.content)), - })), + let tool_result = UserContent::ToolResult(RigToolResult { + id: tool_id.clone(), + call_id: Some(tool_id), + content: OneOrMany::one(ToolResultContent::text(&msg.content)), }); + + let should_merge = matches!( + history.last(), + Some(RigMessage::User { content }) if content.iter().all(|c| matches!(c, UserContent::ToolResult(_))) + ); + + if should_merge { + if let Some(RigMessage::User { content }) = history.last_mut() { + content.push(tool_result); + } + } else { + history.push(RigMessage::User { + content: OneOrMany::one(tool_result), + }); + } } } } @@ -1280,4 +1296,68 @@ mod tests { assert!(adapter.unsupported_params.is_empty()); } + + /// Regression test: consecutive tool_result messages from parallel tool + /// execution must be merged into a single User message with multiple + /// ToolResult content items. Without merging, APIs like Anthropic reject + /// the request due to consecutive User messages. + #[test] + fn test_consecutive_tool_results_merged_into_single_user_message() { + let tc1 = IronToolCall { + id: "call_a".to_string(), + name: "search".to_string(), + arguments: serde_json::json!({"q": "rust"}), + }; + let tc2 = IronToolCall { + id: "call_b".to_string(), + name: "fetch".to_string(), + arguments: serde_json::json!({"url": "https://example.com"}), + }; + let assistant = ChatMessage::assistant_with_tool_calls(None, vec![tc1, tc2]); + let result_a = ChatMessage::tool_result("call_a", "search", "search results"); + let result_b = ChatMessage::tool_result("call_b", "fetch", "fetch results"); + + let messages = vec![assistant, result_a, result_b]; + let (_preamble, history) = convert_messages(&messages); + + // Should be: 1 assistant + 1 merged user (not 1 assistant + 2 users) + assert_eq!( + history.len(), + 2, + "Expected 2 messages (assistant + merged user), got {}", + history.len() + ); + + // The second message should contain both tool results + match &history[1] { + RigMessage::User { content } => { + assert_eq!( + content.len(), + 2, + "Expected 2 tool results in merged user message, got {}", + content.len() + ); + for item in content.iter() { + assert!( + matches!(item, UserContent::ToolResult(_)), + "Expected ToolResult content" + ); + } + } + other => panic!("Expected User message, got: {:?}", other), + } + } + + /// Verify that a tool_result after a non-tool User message is NOT merged. + #[test] + fn test_tool_result_after_user_text_not_merged() { + let user_msg = ChatMessage::user("hello"); + let tool_msg = ChatMessage::tool_result("call_1", "search", "results"); + + let messages = vec![user_msg, tool_msg]; + let (_preamble, history) = convert_messages(&messages); + + // Should be 2 separate User messages (text user + tool result user) + assert_eq!(history.len(), 2); + } } diff --git a/src/main.rs b/src/main.rs index 0b469530..57461677 100644 --- a/src/main.rs +++ b/src/main.rs @@ -27,6 +27,8 @@ use ironclaw::{ webhooks::{self, ToolWebhookState}, }; +#[cfg(unix)] +use ironclaw::channels::ChannelSecretUpdater; #[cfg(any(feature = "postgres", feature = "libsql"))] use ironclaw::setup::{SetupConfig, SetupWizard}; @@ -521,6 +523,30 @@ async fn async_main() -> anyhow::Result<()> { } } + // Persist auto-generated auth token so it survives restarts. + // Write to the "default" settings namespace, which is the namespace + // Config::from_db() reads from — NOT the gateway channel's user_id. + if gw_config.auth_token.is_none() { + let token_to_persist = gw.auth_token().to_string(); + if let Some(ref db) = components.db { + let db = db.clone(); + tokio::spawn(async move { + if let Err(e) = db + .set_setting( + "default", + "channels.gateway_auth_token", + &serde_json::Value::String(token_to_persist), + ) + .await + { + tracing::warn!("Failed to persist auto-generated gateway auth token: {e}"); + } else { + tracing::debug!("Persisted auto-generated gateway auth token to settings"); + } + }); + } + } + gateway_url = Some(format!( "http://{}:{}/?token={}", gw_config.host, @@ -740,7 +766,6 @@ async fn async_main() -> anyhow::Result<()> { #[cfg(unix)] { - use ironclaw::channels::ChannelSecretUpdater; // Collect all channels that support secret updates let mut secret_updaters: Vec> = Vec::new(); if let Some(ref state) = http_channel_state { diff --git a/src/sandbox/manager.rs b/src/sandbox/manager.rs index ce709f50..1c0decc8 100644 --- a/src/sandbox/manager.rs +++ b/src/sandbox/manager.rs @@ -236,14 +236,59 @@ impl SandboxManager { self.initialize().await?; } - // Get proxy port if running + // Retry transient container failures (Docker daemon glitches, container + // creation races) up to MAX_SANDBOX_RETRIES times with exponential backoff. + const MAX_SANDBOX_RETRIES: u32 = 2; + let mut last_err: Option = None; + + for attempt in 0..=MAX_SANDBOX_RETRIES { + if attempt > 0 { + let delay = std::time::Duration::from_secs(1 << attempt); // 2s, 4s + tracing::warn!( + attempt = attempt + 1, + max_attempts = MAX_SANDBOX_RETRIES + 1, + delay_secs = delay.as_secs(), + "Retrying sandbox execution after transient failure" + ); + tokio::time::sleep(delay).await; + } + + match self + .try_execute_in_container(command, cwd, policy, env.clone()) + .await + { + Ok(output) => return Ok(output), + Err(e) if is_transient_sandbox_error(&e) => { + tracing::warn!( + attempt = attempt + 1, + error = %e, + "Transient sandbox error, will retry" + ); + last_err = Some(e); + } + Err(e) => return Err(e), + } + } + + Err(last_err.unwrap_or_else(|| SandboxError::ExecutionFailed { + reason: "all retry attempts exhausted".to_string(), + })) + } + + /// Single attempt at container execution (no retry logic). + async fn try_execute_in_container( + &self, + command: &str, + cwd: &Path, + policy: SandboxPolicy, + env: HashMap, + ) -> Result { let proxy_port = if let Some(proxy) = self.proxy.read().await.as_ref() { proxy.addr().await.map(|a| a.port()).unwrap_or(0) } else { 0 }; - // Reuse the stored Docker connection, create a runner with the current proxy port let docker = self.docker .read() @@ -262,7 +307,6 @@ impl SandboxManager { }; let container_output = runner.execute(command, cwd, policy, &limits, env).await?; - Ok(container_output.into()) } @@ -373,6 +417,20 @@ impl Drop for SandboxManager { } } +/// Check whether a sandbox error is transient and worth retrying. +/// +/// Transient errors are those caused by Docker daemon glitches, container +/// creation race conditions, or container start failures — not by command +/// execution failures, timeouts, or policy violations. +fn is_transient_sandbox_error(err: &SandboxError) -> bool { + matches!( + err, + SandboxError::DockerNotAvailable { .. } + | SandboxError::ContainerCreationFailed { .. } + | SandboxError::ContainerStartFailed { .. } + ) +} + /// Builder for creating a sandbox manager. pub struct SandboxManagerBuilder { config: SandboxConfig, @@ -597,4 +655,43 @@ mod tests { assert!(output.truncated); assert!(output.stdout.len() <= 32 * 1024); } + + #[test] + fn transient_errors_are_retryable() { + assert!(super::is_transient_sandbox_error( + &SandboxError::DockerNotAvailable { + reason: "daemon restarting".to_string() + } + )); + assert!(super::is_transient_sandbox_error( + &SandboxError::ContainerCreationFailed { + reason: "image pull glitch".to_string() + } + )); + assert!(super::is_transient_sandbox_error( + &SandboxError::ContainerStartFailed { + reason: "cgroup race".to_string() + } + )); + } + + #[test] + fn non_transient_errors_are_not_retryable() { + assert!(!super::is_transient_sandbox_error(&SandboxError::Timeout( + std::time::Duration::from_secs(30) + ))); + assert!(!super::is_transient_sandbox_error( + &SandboxError::ExecutionFailed { + reason: "exit code 1".to_string() + } + )); + assert!(!super::is_transient_sandbox_error( + &SandboxError::NetworkBlocked { + reason: "policy violation".to_string() + } + )); + assert!(!super::is_transient_sandbox_error(&SandboxError::Config { + reason: "bad config".to_string() + })); + } } diff --git a/src/settings.rs b/src/settings.rs index 1c0b737e..2a5b6bbd 100644 --- a/src/settings.rs +++ b/src/settings.rs @@ -360,6 +360,10 @@ pub struct HeartbeatSettings { #[serde(default)] pub notify_user: Option, + /// Fixed time-of-day to fire (HH:MM, 24h). When set, interval_secs is ignored. + #[serde(default)] + pub fire_at: Option, + /// Hour (0-23) when quiet hours start (heartbeat skipped). #[serde(default)] pub quiet_hours_start: Option, @@ -368,7 +372,7 @@ pub struct HeartbeatSettings { #[serde(default)] pub quiet_hours_end: Option, - /// Timezone for quiet hours evaluation (IANA name, e.g. "America/New_York"). + /// Timezone for fire_at and quiet hours (IANA name, e.g. "Pacific/Auckland"). #[serde(default)] pub timezone: Option, } @@ -384,6 +388,7 @@ impl Default for HeartbeatSettings { interval_secs: default_heartbeat_interval(), notify_channel: None, notify_user: None, + fire_at: None, quiet_hours_start: None, quiet_hours_end: None, timezone: None, diff --git a/src/worker/job.rs b/src/worker/job.rs index 1247a552..0f0e969e 100644 --- a/src/worker/job.rs +++ b/src/worker/job.rs @@ -1170,11 +1170,16 @@ impl<'a> LoopDelegate for JobDelegate<'a> { // Reset counter after a successful LLM call self.consecutive_rate_limits .store(0, std::sync::atomic::Ordering::Relaxed); + // Preserve the LLM's reasoning text so it appears in the + // assistant_with_tool_calls message pushed by execute_tool_calls. + let reasoning_text = s + .iter() + .find_map(|sel| (!sel.reasoning.is_empty()).then_some(sel.reasoning.clone())); let tool_calls: Vec = selections_to_tool_calls(&s); return Ok(crate::llm::RespondOutput { result: RespondResult::ToolCalls { tool_calls, - content: None, + content: reasoning_text, }, usage: crate::llm::TokenUsage::default(), }); @@ -1586,7 +1591,7 @@ mod tests { } #[tokio::test] - async fn test_mark_completed_twice_returns_error() { + async fn test_mark_completed_twice_is_idempotent() { let worker = make_worker(vec![]).await; worker @@ -1607,11 +1612,22 @@ mod tests { .unwrap(); assert_eq!(ctx.state, JobState::Completed); + // Second mark_completed should succeed (idempotent) rather than + // erroring, matching the fix for the execution_loop / worker wrapper + // race condition. let result = worker.mark_completed().await; assert!( - result.is_err(), - "Completed → Completed transition should be rejected by state machine" + result.is_ok(), + "Completed -> Completed transition should be idempotent" ); + + // State should still be Completed + let ctx = worker + .context_manager() + .get_context(worker.job_id) + .await + .unwrap(); + assert_eq!(ctx.state, JobState::Completed); } /// Build a Worker with the given approval context. @@ -1849,4 +1865,128 @@ mod tests { "Iteration cap should transition to Failed, not Stuck" ); } + + /// Regression test: selections_to_tool_calls must preserve tool_call_id + /// so that tool_result messages match the assistant_with_tool_calls message + /// and are not treated as orphaned by sanitize_tool_messages. + #[test] + fn test_selections_to_tool_calls_preserves_ids() { + let selections = vec![ + ToolSelection { + tool_name: "search".into(), + parameters: serde_json::json!({"q": "test"}), + reasoning: "Need to search".into(), + alternatives: vec![], + tool_call_id: "call_abc".into(), + }, + ToolSelection { + tool_name: "fetch".into(), + parameters: serde_json::json!({"url": "https://example.com"}), + reasoning: "Need to fetch".into(), + alternatives: vec![], + tool_call_id: "call_def".into(), + }, + ]; + + let tool_calls = selections_to_tool_calls(&selections); + + assert_eq!(tool_calls.len(), 2); + assert_eq!(tool_calls[0].id, "call_abc"); + assert_eq!(tool_calls[0].name, "search"); + assert_eq!(tool_calls[1].id, "call_def"); + assert_eq!(tool_calls[1].name, "fetch"); + } + + /// Regression test: when select_tools returns selections with reasoning, + /// the reasoning text should be preserved as content in the RespondResult + /// so it appears in the assistant_with_tool_calls message. Without this, + /// the LLM's reasoning context is lost and subsequent turns lack context. + #[test] + fn test_reasoning_text_extraction_from_selections() { + // Simulate what call_llm does: extract first non-empty reasoning + let selections = [ + ToolSelection { + tool_name: "search".into(), + parameters: serde_json::json!({}), + reasoning: "I need to search for relevant information".into(), + alternatives: vec![], + tool_call_id: "call_1".into(), + }, + ToolSelection { + tool_name: "fetch".into(), + parameters: serde_json::json!({}), + reasoning: "I need to search for relevant information".into(), + alternatives: vec![], + tool_call_id: "call_2".into(), + }, + ]; + + let reasoning_text = selections + .iter() + .find_map(|sel| (!sel.reasoning.is_empty()).then_some(sel.reasoning.clone())); + + assert_eq!( + reasoning_text.as_deref(), + Some("I need to search for relevant information"), + "Reasoning text should be extracted from first non-empty selection" + ); + + // Empty reasoning should result in None + let empty_selections = [ToolSelection { + tool_name: "echo".into(), + parameters: serde_json::json!({}), + reasoning: String::new(), + alternatives: vec![], + tool_call_id: "call_3".into(), + }]; + + let empty_reasoning = empty_selections + .iter() + .find_map(|sel| (!sel.reasoning.is_empty()).then_some(sel.reasoning.clone())); + + assert!( + empty_reasoning.is_none(), + "Empty reasoning should not be included as content" + ); + } + + /// When the first selection has empty reasoning but a subsequent one has + /// non-empty reasoning, find_map should skip the empty one and return the + /// first non-empty reasoning. + #[test] + fn test_reasoning_text_skips_empty_first_selection() { + let selections = [ + ToolSelection { + tool_name: "echo".into(), + parameters: serde_json::json!({}), + reasoning: String::new(), + alternatives: vec![], + tool_call_id: "call_1".into(), + }, + ToolSelection { + tool_name: "search".into(), + parameters: serde_json::json!({}), + reasoning: "Found the answer in the second selection".into(), + alternatives: vec![], + tool_call_id: "call_2".into(), + }, + ToolSelection { + tool_name: "fetch".into(), + parameters: serde_json::json!({}), + reasoning: "Third selection reasoning".into(), + alternatives: vec![], + tool_call_id: "call_3".into(), + }, + ]; + + let reasoning_text = selections + .iter() + .find_map(|sel| (!sel.reasoning.is_empty()).then_some(sel.reasoning.clone())); + + assert_eq!( + reasoning_text.as_deref(), + Some("Found the answer in the second selection"), + "Should skip empty first reasoning and return the first non-empty one" + ); + } } diff --git a/tests/support/gateway_workflow_harness.rs b/tests/support/gateway_workflow_harness.rs index dd9e8643..c539dad5 100644 --- a/tests/support/gateway_workflow_harness.rs +++ b/tests/support/gateway_workflow_harness.rs @@ -143,6 +143,9 @@ impl GatewayWorkflowHarness { model: model.to_string(), extra_headers: Vec::new(), oauth_token: None, + is_codex_chatgpt: false, + refresh_token: None, + auth_path: None, cache_retention: Default::default(), unsupported_params: Vec::new(), });