diff --git a/src/agent/agent_loop.rs b/src/agent/agent_loop.rs index 8fda4143..891dc36f 100644 --- a/src/agent/agent_loop.rs +++ b/src/agent/agent_loop.rs @@ -74,7 +74,7 @@ pub struct AgentDeps { /// Cost enforcement guardrails (daily budget, hourly rate limits). pub cost_guard: Arc, /// SSE broadcast sender for live job event streaming to the web gateway. - pub sse_tx: Option>, + pub sse_tx: Option>, /// HTTP interceptor for trace recording/replay. pub http_interceptor: Option>, /// Audio transcription middleware for voice messages. diff --git a/src/agent/job_monitor.rs b/src/agent/job_monitor.rs index b2db8852..608b75bd 100644 --- a/src/agent/job_monitor.rs +++ b/src/agent/job_monitor.rs @@ -19,7 +19,7 @@ use tokio::task::JoinHandle; use uuid::Uuid; use crate::channels::IncomingMessage; -use crate::channels::web::types::SseEvent; +use crate::events::DomainEvent as SseEvent; /// Spawn a background task that watches for events from a specific job and /// injects assistant messages into the agent loop. diff --git a/src/agent/routine.rs b/src/agent/routine.rs index 2dee6333..2cbf11ef 100644 --- a/src/agent/routine.rs +++ b/src/agent/routine.rs @@ -1,811 +1,7 @@ -//! Core types for the routines system. +//! Re-exports routine types from `crate::models::routine`. //! -//! A routine is a named, persistent, user-owned task with a trigger and an action. -//! Each routine fires independently when its trigger condition is met, with only -//! that routine's prompt and context sent to the LLM. -//! -//! ```text -//! ┌──────────┐ ┌─────────┐ ┌──────────────────┐ -//! │ Trigger │────▶│ Engine │────▶│ Execution Mode │ -//! │ cron/event│ │guardrail│ │lightweight│full_job│ -//! │ system │ │ check │ └──────────────────┘ -//! │ manual │ └─────────┘ │ -//! └──────────┘ ▼ -//! ┌──────────────┐ -//! │ Notify user │ -//! │ if needed │ -//! └──────────────┘ -//! ``` +//! The canonical definitions now live in `src/models/routine.rs` to break the +//! circular dependency between `db` and `agent`. This module re-exports +//! everything for backward compatibility within the agent module. -use std::collections::hash_map::DefaultHasher; -use std::hash::{Hash, Hasher}; -use std::str::FromStr; -use std::time::Duration; - -use chrono::{DateTime, Utc}; -use serde::{Deserialize, Serialize}; -use uuid::Uuid; - -use crate::error::RoutineError; - -/// A routine is a named, persistent, user-owned task with a trigger and an action. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct Routine { - pub id: Uuid, - pub name: String, - pub description: String, - pub user_id: String, - pub enabled: bool, - pub trigger: Trigger, - pub action: RoutineAction, - pub guardrails: RoutineGuardrails, - pub notify: NotifyConfig, - - // Runtime state (DB-managed) - pub last_run_at: Option>, - pub next_fire_at: Option>, - pub run_count: u64, - pub consecutive_failures: u32, - pub state: serde_json::Value, - - pub created_at: DateTime, - pub updated_at: DateTime, -} - -/// When a routine should fire. -#[derive(Debug, Clone, Serialize, Deserialize)] -#[serde(tag = "type", rename_all = "snake_case")] -pub enum Trigger { - /// Fire on a cron schedule (e.g. "0 9 * * MON-FRI" or "every 2h"). - Cron { - schedule: String, - #[serde(default)] - timezone: Option, - }, - /// Fire when a channel message matches a pattern. - Event { - /// Optional channel filter (e.g. "telegram", "slack"). - channel: Option, - /// Regex pattern to match against message content. - pattern: String, - }, - /// Fire when a structured system event is emitted. - SystemEvent { - /// Event source namespace (e.g. "github", "workflow", "tool"). - source: String, - /// Event type within the source (e.g. "issue.opened"). - event_type: String, - /// Optional exact-match filters against payload top-level fields. - #[serde(default)] - filters: std::collections::HashMap, - }, - /// Only fires via tool call or CLI. - Manual, -} - -impl Trigger { - /// The string tag stored in the DB trigger_type column. - pub fn type_tag(&self) -> &'static str { - match self { - Trigger::Cron { .. } => "cron", - Trigger::Event { .. } => "event", - Trigger::SystemEvent { .. } => "system_event", - Trigger::Manual => "manual", - } - } - - /// Parse a trigger from its DB representation. - pub fn from_db(trigger_type: &str, config: serde_json::Value) -> Result { - match trigger_type { - "cron" => { - let schedule = config - .get("schedule") - .and_then(|v| v.as_str()) - .ok_or_else(|| RoutineError::MissingField { - context: "cron trigger".into(), - field: "schedule".into(), - })? - .to_string(); - let timezone = config - .get("timezone") - .and_then(|v| v.as_str()) - .and_then(|tz| { - if crate::timezone::parse_timezone(tz).is_some() { - Some(tz.to_string()) - } else { - tracing::warn!( - "Ignoring invalid timezone '{}' from DB for cron trigger", - tz - ); - None - } - }); - Ok(Trigger::Cron { schedule, timezone }) - } - "event" => { - let pattern = config - .get("pattern") - .and_then(|v| v.as_str()) - .ok_or_else(|| RoutineError::MissingField { - context: "event trigger".into(), - field: "pattern".into(), - })? - .to_string(); - let channel = config - .get("channel") - .and_then(|v| v.as_str()) - .map(String::from); - Ok(Trigger::Event { channel, pattern }) - } - "system_event" => { - let source = config - .get("source") - .and_then(|v| v.as_str()) - .ok_or_else(|| RoutineError::MissingField { - context: "system_event trigger".into(), - field: "source".into(), - })? - .to_string(); - let event_type = config - .get("event_type") - .and_then(|v| v.as_str()) - .ok_or_else(|| RoutineError::MissingField { - context: "system_event trigger".into(), - field: "event_type".into(), - })? - .to_string(); - let filters = config - .get("filters") - .and_then(|v| v.as_object()) - .map(|m| { - m.iter() - .filter_map(|(k, v)| { - json_value_as_filter_string(v).map(|s| (k.clone(), s)) - }) - .collect() - }) - .unwrap_or_default(); - Ok(Trigger::SystemEvent { - source, - event_type, - filters, - }) - } - "manual" => Ok(Trigger::Manual), - other => Err(RoutineError::UnknownTriggerType { - trigger_type: other.to_string(), - }), - } - } - - /// Serialize trigger-specific config to JSON for DB storage. - pub fn to_config_json(&self) -> serde_json::Value { - match self { - Trigger::Cron { schedule, timezone } => serde_json::json!({ - "schedule": schedule, - "timezone": timezone, - }), - Trigger::Event { channel, pattern } => serde_json::json!({ - "pattern": pattern, - "channel": channel, - }), - Trigger::SystemEvent { - source, - event_type, - filters, - } => serde_json::json!({ - "source": source, - "event_type": event_type, - "filters": filters, - }), - Trigger::Manual => serde_json::json!({}), - } - } -} - -/// What happens when a routine fires. -#[derive(Debug, Clone, Serialize, Deserialize)] -#[serde(tag = "type", rename_all = "snake_case")] -pub enum RoutineAction { - /// Single LLM call (optionally with tools). Cheap and fast. - Lightweight { - /// The prompt sent to the LLM. - prompt: String, - /// Workspace paths to load as context (e.g. ["context/priorities.md"]). - #[serde(default)] - context_paths: Vec, - /// Max output tokens (default: 4096). - #[serde(default = "default_max_tokens")] - max_tokens: u32, - /// Enable tool access (default: false for backward compatibility). - /// When true, the LLM can call tools during execution. - /// Tools requiring approval are automatically filtered out. - #[serde(default)] - use_tools: bool, - /// Max tool call rounds (default: 3). Only used when use_tools is true. - #[serde(default = "default_max_tool_rounds")] - max_tool_rounds: u32, - }, - /// Full multi-turn worker job with tool access. - FullJob { - /// Job title for the scheduler. - title: String, - /// Job description / initial prompt. - description: String, - /// Max reasoning iterations (default: 10). - #[serde(default = "default_max_iterations")] - max_iterations: u32, - /// Tool names pre-authorized for `Always`-approval tools (e.g. destructive - /// shell commands, cross-channel messaging). `UnlessAutoApproved` tools are - /// automatically permitted in routine jobs without listing them here. - #[serde(default)] - tool_permissions: Vec, - }, -} - -fn default_max_tokens() -> u32 { - 4096 -} - -fn default_max_iterations() -> u32 { - 10 -} - -fn default_max_tool_rounds() -> u32 { - 3 -} - -/// Hard upper bound for max_tool_rounds to prevent runaway loops and cost explosion. -pub(crate) const MAX_TOOL_ROUNDS_LIMIT: u32 = 20; - -/// Clamp max_tool_rounds to [1, MAX_TOOL_ROUNDS_LIMIT]. -/// Accepts u64 to avoid truncation before clamping. -fn clamp_max_tool_rounds(value: u64) -> u32 { - value.clamp(1, MAX_TOOL_ROUNDS_LIMIT as u64) as u32 -} - -/// Parse a `tool_permissions` JSON array into a `Vec`. -pub fn parse_tool_permissions(value: &serde_json::Value) -> Vec { - value - .get("tool_permissions") - .and_then(|v| v.as_array()) - .map(|arr| { - arr.iter() - .filter_map(|v| v.as_str().map(String::from)) - .collect() - }) - .unwrap_or_default() -} - -impl RoutineAction { - /// The string tag stored in the DB action_type column. - pub fn type_tag(&self) -> &'static str { - match self { - RoutineAction::Lightweight { .. } => "lightweight", - RoutineAction::FullJob { .. } => "full_job", - } - } - - /// Parse an action from its DB representation. - pub fn from_db(action_type: &str, config: serde_json::Value) -> Result { - match action_type { - "lightweight" => { - let prompt = config - .get("prompt") - .and_then(|v| v.as_str()) - .ok_or_else(|| RoutineError::MissingField { - context: "lightweight action".into(), - field: "prompt".into(), - })? - .to_string(); - let context_paths = config - .get("context_paths") - .and_then(|v| v.as_array()) - .map(|arr| { - arr.iter() - .filter_map(|v| v.as_str().map(String::from)) - .collect() - }) - .unwrap_or_default(); - let max_tokens = config - .get("max_tokens") - .and_then(|v| v.as_u64()) - .unwrap_or(default_max_tokens() as u64) as u32; - let use_tools = config - .get("use_tools") - .and_then(|v| v.as_bool()) - .unwrap_or(false); - let max_tool_rounds = clamp_max_tool_rounds( - config - .get("max_tool_rounds") - .and_then(|v| v.as_u64()) - .unwrap_or(default_max_tool_rounds() as u64), - ); - Ok(RoutineAction::Lightweight { - prompt, - context_paths, - max_tokens, - use_tools, - max_tool_rounds, - }) - } - "full_job" => { - let title = config - .get("title") - .and_then(|v| v.as_str()) - .ok_or_else(|| RoutineError::MissingField { - context: "full_job action".into(), - field: "title".into(), - })? - .to_string(); - let description = config - .get("description") - .and_then(|v| v.as_str()) - .ok_or_else(|| RoutineError::MissingField { - context: "full_job action".into(), - field: "description".into(), - })? - .to_string(); - let max_iterations = config - .get("max_iterations") - .and_then(|v| v.as_u64()) - .unwrap_or(default_max_iterations() as u64) - as u32; - let tool_permissions = parse_tool_permissions(&config); - Ok(RoutineAction::FullJob { - title, - description, - max_iterations, - tool_permissions, - }) - } - other => Err(RoutineError::UnknownActionType { - action_type: other.to_string(), - }), - } - } - - /// Serialize action config to JSON for DB storage. - pub fn to_config_json(&self) -> serde_json::Value { - match self { - RoutineAction::Lightweight { - prompt, - context_paths, - max_tokens, - use_tools, - max_tool_rounds, - } => serde_json::json!({ - "prompt": prompt, - "context_paths": context_paths, - "max_tokens": max_tokens, - "use_tools": use_tools, - "max_tool_rounds": max_tool_rounds, - }), - RoutineAction::FullJob { - title, - description, - max_iterations, - tool_permissions, - } => serde_json::json!({ - "title": title, - "description": description, - "max_iterations": max_iterations, - "tool_permissions": tool_permissions, - }), - } - } -} - -/// Guardrails to prevent runaway execution. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct RoutineGuardrails { - /// Minimum time between fires. - pub cooldown: Duration, - /// Max simultaneous runs of this routine. - pub max_concurrent: u32, - /// Window for content-hash dedup (event triggers). None = no dedup. - pub dedup_window: Option, -} - -impl Default for RoutineGuardrails { - fn default() -> Self { - Self { - cooldown: Duration::from_secs(300), - max_concurrent: 1, - dedup_window: None, - } - } -} - -/// Notification preferences for a routine. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct NotifyConfig { - /// Channel to notify on (None = default/broadcast all). - pub channel: Option, - /// User to notify. - pub user: String, - /// Notify when routine produces actionable output. - pub on_attention: bool, - /// Notify when routine errors. - pub on_failure: bool, - /// Notify when routine runs with no findings. - pub on_success: bool, -} - -impl Default for NotifyConfig { - fn default() -> Self { - Self { - channel: None, - user: "default".to_string(), - on_attention: true, - on_failure: true, - on_success: false, - } - } -} - -/// Status of a routine run. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] -#[serde(rename_all = "snake_case")] -pub enum RunStatus { - Running, - Ok, - Attention, - Failed, -} - -impl std::fmt::Display for RunStatus { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - match self { - RunStatus::Running => write!(f, "running"), - RunStatus::Ok => write!(f, "ok"), - RunStatus::Attention => write!(f, "attention"), - RunStatus::Failed => write!(f, "failed"), - } - } -} - -impl FromStr for RunStatus { - type Err = RoutineError; - fn from_str(s: &str) -> Result { - match s { - "running" => Ok(RunStatus::Running), - "ok" => Ok(RunStatus::Ok), - "attention" => Ok(RunStatus::Attention), - "failed" => Ok(RunStatus::Failed), - other => Err(RoutineError::UnknownRunStatus { - status: other.to_string(), - }), - } - } -} - -/// A single execution of a routine. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct RoutineRun { - pub id: Uuid, - pub routine_id: Uuid, - pub trigger_type: String, - pub trigger_detail: Option, - pub started_at: DateTime, - pub completed_at: Option>, - pub status: RunStatus, - pub result_summary: Option, - pub tokens_used: Option, - pub job_id: Option, - pub created_at: DateTime, -} - -/// Convert a JSON value to a string for filter storage. -/// -/// Handles strings, numbers, and booleans — consistent with the matching -/// logic in `routine_engine::json_value_as_string`. -pub fn json_value_as_filter_string(v: &serde_json::Value) -> Option { - match v { - serde_json::Value::String(s) => Some(s.clone()), - serde_json::Value::Number(n) => Some(n.to_string()), - serde_json::Value::Bool(b) => Some(b.to_string()), - _ => None, - } -} - -/// Compute a content hash for event dedup. -pub fn content_hash(content: &str) -> u64 { - let mut hasher = DefaultHasher::new(); - content.hash(&mut hasher); - hasher.finish() -} - -/// Parse a cron expression and compute the next fire time from now. -/// -/// When `timezone` is provided and valid, the schedule is evaluated in that -/// timezone and the result is converted back to UTC. Otherwise UTC is used. -pub fn next_cron_fire( - schedule: &str, - timezone: Option<&str>, -) -> Result>, RoutineError> { - let cron_schedule = - cron::Schedule::from_str(schedule).map_err(|e| RoutineError::InvalidCron { - reason: e.to_string(), - })?; - if let Some(tz) = timezone.and_then(crate::timezone::parse_timezone) { - Ok(cron_schedule - .upcoming(tz) - .next() - .map(|dt| dt.with_timezone(&Utc))) - } else { - Ok(cron_schedule.upcoming(Utc).next()) - } -} - -#[cfg(test)] -mod tests { - use crate::agent::routine::{ - MAX_TOOL_ROUNDS_LIMIT, RoutineAction, RoutineGuardrails, RunStatus, Trigger, content_hash, - next_cron_fire, - }; - - #[test] - fn test_trigger_roundtrip() { - let trigger = Trigger::Cron { - schedule: "0 9 * * MON-FRI".to_string(), - timezone: None, - }; - let json = trigger.to_config_json(); - let parsed = Trigger::from_db("cron", json).expect("parse cron"); - assert!(matches!(parsed, Trigger::Cron { schedule, .. } if schedule == "0 9 * * MON-FRI")); - } - - #[test] - fn test_event_trigger_roundtrip() { - let trigger = Trigger::Event { - channel: Some("telegram".to_string()), - pattern: r"deploy\s+\w+".to_string(), - }; - let json = trigger.to_config_json(); - let parsed = Trigger::from_db("event", json).expect("parse event"); - assert!(matches!(parsed, Trigger::Event { channel, pattern } - if channel == Some("telegram".to_string()) && pattern == r"deploy\s+\w+")); - } - - #[test] - fn test_system_event_trigger_roundtrip() { - let mut filters = std::collections::HashMap::new(); - filters.insert("repo".to_string(), "nearai/ironclaw".to_string()); - filters.insert("action".to_string(), "opened".to_string()); - let trigger = Trigger::SystemEvent { - source: "github".to_string(), - event_type: "issue".to_string(), - filters: filters.clone(), - }; - let json = trigger.to_config_json(); - let parsed = Trigger::from_db("system_event", json).expect("parse system_event"); - assert!( - matches!(parsed, Trigger::SystemEvent { source, event_type, filters: f } - if source == "github" && event_type == "issue" && f == filters) - ); - } - - #[test] - fn test_action_lightweight_roundtrip() { - let action = RoutineAction::Lightweight { - prompt: "Check PRs".to_string(), - context_paths: vec!["context/priorities.md".to_string()], - max_tokens: 2048, - use_tools: false, - max_tool_rounds: 3, - }; - let json = action.to_config_json(); - let parsed = RoutineAction::from_db("lightweight", json).expect("parse lightweight"); - assert!( - matches!(parsed, RoutineAction::Lightweight { prompt, context_paths, max_tokens, .. } - if prompt == "Check PRs" && context_paths.len() == 1 && max_tokens == 2048) - ); - } - - #[test] - fn test_action_full_job_roundtrip() { - let action = RoutineAction::FullJob { - title: "Deploy review".to_string(), - description: "Review and deploy pending changes".to_string(), - max_iterations: 5, - tool_permissions: vec!["shell".to_string()], - }; - let json = action.to_config_json(); - let parsed = RoutineAction::from_db("full_job", json).expect("parse full_job"); - assert!( - matches!(parsed, RoutineAction::FullJob { title, max_iterations, tool_permissions, .. } - if title == "Deploy review" && max_iterations == 5 && tool_permissions == vec!["shell".to_string()]) - ); - } - - #[test] - fn test_run_status_display_parse() { - for status in [ - RunStatus::Running, - RunStatus::Ok, - RunStatus::Attention, - RunStatus::Failed, - ] { - let s = status.to_string(); - let parsed: RunStatus = s.parse().expect("parse status"); - assert_eq!(parsed, status); - } - } - - #[test] - fn test_content_hash_deterministic() { - let h1 = content_hash("deploy production"); - let h2 = content_hash("deploy production"); - assert_eq!(h1, h2); - - let h3 = content_hash("deploy staging"); - assert_ne!(h1, h3); - } - - #[test] - fn test_next_cron_fire_valid() { - // Every minute should always have a next fire - let next = next_cron_fire("* * * * * *", None).expect("valid cron"); - assert!(next.is_some()); - } - - #[test] - fn test_next_cron_fire_invalid() { - let result = next_cron_fire("not a cron", None); - assert!(result.is_err()); - } - - #[test] - fn test_trigger_cron_timezone_roundtrip() { - let trigger = Trigger::Cron { - schedule: "0 9 * * MON-FRI".to_string(), - timezone: Some("America/New_York".to_string()), - }; - let json = trigger.to_config_json(); - let parsed = Trigger::from_db("cron", json).expect("parse cron"); - assert!(matches!(parsed, Trigger::Cron { schedule, timezone } - if schedule == "0 9 * * MON-FRI" - && timezone.as_deref() == Some("America/New_York"))); - } - - #[test] - fn test_trigger_cron_no_timezone_backward_compat() { - let json = serde_json::json!({"schedule": "0 9 * * *"}); - let parsed = Trigger::from_db("cron", json).expect("parse cron"); - assert!(matches!(parsed, Trigger::Cron { timezone, .. } if timezone.is_none())); - } - - #[test] - fn test_trigger_cron_invalid_timezone_coerced_to_none() { - let json = serde_json::json!({"schedule": "0 9 * * *", "timezone": "Fake/Zone"}); - let parsed = Trigger::from_db("cron", json).expect("parse cron"); - assert!( - matches!(parsed, Trigger::Cron { timezone, .. } if timezone.is_none()), - "invalid timezone should be coerced to None" - ); - } - - #[test] - fn test_next_cron_fire_with_timezone() { - let next_utc = next_cron_fire("0 0 9 * * * *", None) - .expect("valid cron") - .expect("has next"); - let next_est = next_cron_fire("0 0 9 * * * *", Some("America/New_York")) - .expect("valid cron") - .expect("has next"); - // EST is UTC-5 (or EDT UTC-4), so the UTC result should differ - assert_ne!(next_utc, next_est, "timezone should shift the fire time"); - } - - #[test] - fn test_guardrails_default() { - let g = RoutineGuardrails::default(); - assert_eq!(g.cooldown.as_secs(), 300); - assert_eq!(g.max_concurrent, 1); - assert!(g.dedup_window.is_none()); - } - - #[test] - fn test_trigger_type_tag() { - assert_eq!( - Trigger::Cron { - schedule: String::new(), - timezone: None, - } - .type_tag(), - "cron" - ); - assert_eq!( - Trigger::Event { - channel: None, - pattern: String::new() - } - .type_tag(), - "event" - ); - assert_eq!( - Trigger::SystemEvent { - source: String::new(), - event_type: String::new(), - filters: std::collections::HashMap::new(), - } - .type_tag(), - "system_event" - ); - assert_eq!(Trigger::Manual.type_tag(), "manual"); - } - - #[test] - fn test_action_lightweight_backward_compat_no_use_tools() { - // Simulate old DB record without use_tools field - let json = serde_json::json!({ - "prompt": "old routine", - "context_paths": [], - "max_tokens": 4096 - }); - let parsed = RoutineAction::from_db("lightweight", json).expect("parse lightweight"); - assert!( - matches!(parsed, RoutineAction::Lightweight { use_tools, max_tool_rounds, .. } - if !use_tools && max_tool_rounds == 3), - "missing use_tools should default to false, max_tool_rounds to 3" - ); - } - - #[test] - fn test_max_tool_rounds_clamped_to_upper_bound() { - let json = serde_json::json!({ - "prompt": "test", - "use_tools": true, - "max_tool_rounds": 9999 - }); - let parsed = RoutineAction::from_db("lightweight", json).expect("parse"); - match parsed { - RoutineAction::Lightweight { - max_tool_rounds, .. - } => { - assert_eq!( - max_tool_rounds, MAX_TOOL_ROUNDS_LIMIT, - "should clamp to MAX_TOOL_ROUNDS_LIMIT" - ); - } - _ => panic!("expected Lightweight"), - } - } - - #[test] - fn test_max_tool_rounds_clamped_to_lower_bound() { - let json = serde_json::json!({ - "prompt": "test", - "use_tools": true, - "max_tool_rounds": 0 - }); - let parsed = RoutineAction::from_db("lightweight", json).expect("parse"); - match parsed { - RoutineAction::Lightweight { - max_tool_rounds, .. - } => { - assert_eq!(max_tool_rounds, 1, "should clamp 0 to 1"); - } - _ => panic!("expected Lightweight"), - } - } - - #[test] - fn test_max_tool_rounds_normal_value_passes_through() { - let json = serde_json::json!({ - "prompt": "test", - "use_tools": true, - "max_tool_rounds": 10 - }); - let parsed = RoutineAction::from_db("lightweight", json).expect("parse"); - match parsed { - RoutineAction::Lightweight { - max_tool_rounds, .. - } => { - assert_eq!(max_tool_rounds, 10, "normal value should pass through"); - } - _ => panic!("expected Lightweight"), - } - } -} +pub use crate::models::routine::*; diff --git a/src/agent/scheduler.rs b/src/agent/scheduler.rs index 5e4bf01a..6fa99b0e 100644 --- a/src/agent/scheduler.rs +++ b/src/agent/scheduler.rs @@ -9,11 +9,11 @@ use tokio::task::JoinHandle; use uuid::Uuid; use crate::agent::task::{Task, TaskContext, TaskOutput}; -use crate::channels::web::types::SseEvent; use crate::config::AgentConfig; use crate::context::{ContextManager, JobContext, JobState}; use crate::db::Database; use crate::error::{Error, JobError}; +use crate::events::DomainEvent as SseEvent; use crate::hooks::HookRegistry; use crate::llm::LlmProvider; use crate::safety::SafetyLayer; diff --git a/src/agent/self_repair.rs b/src/agent/self_repair.rs index a67fe23e..92d17e4a 100644 --- a/src/agent/self_repair.rs +++ b/src/agent/self_repair.rs @@ -22,17 +22,11 @@ pub struct StuckJob { pub repair_attempts: u32, } -/// A tool that has been detected as broken. -#[derive(Debug, Clone)] -pub struct BrokenTool { - pub name: String, - pub failure_count: u32, - pub last_error: Option, - pub first_failure: DateTime, - pub last_failure: DateTime, - pub last_build_result: Option, - pub repair_attempts: u32, -} +/// Backward-compatible alias for `ToolFailureRecord`. +/// +/// The canonical type now lives in `crate::models::tool_failure` to break +/// the circular dependency between `db` and `agent`. +pub type BrokenTool = crate::models::tool_failure::ToolFailureRecord; /// Result of a repair attempt. #[derive(Debug)] diff --git a/src/agent/session.rs b/src/agent/session.rs index 193e0309..4e0784bf 100644 --- a/src/agent/session.rs +++ b/src/agent/session.rs @@ -16,8 +16,8 @@ use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; use uuid::Uuid; -use crate::channels::web::util::truncate_preview; use crate::llm::{ChatMessage, ToolCall}; +use crate::util::truncate_preview; /// A session containing one or more threads. #[derive(Debug, Clone, Serialize, Deserialize)] diff --git a/src/agent/thread_ops.rs b/src/agent/thread_ops.rs index 3438d1cd..3943584f 100644 --- a/src/agent/thread_ops.rs +++ b/src/agent/thread_ops.rs @@ -16,12 +16,12 @@ use crate::agent::dispatcher::{ }; use crate::agent::session::{PendingApproval, Session, ThreadState}; use crate::agent::submission::SubmissionResult; -use crate::channels::web::util::truncate_preview; use crate::channels::{IncomingMessage, StatusUpdate}; use crate::context::JobContext; use crate::error::Error; use crate::llm::{ChatMessage, ToolCall}; use crate::tools::redact_params; +use crate::util::truncate_preview; const FORGED_THREAD_ID_ERROR: &str = "Invalid or unauthorized thread ID."; diff --git a/src/channels/web/types.rs b/src/channels/web/types.rs index b8690b78..ab30e736 100644 --- a/src/channels/web/types.rs +++ b/src/channels/web/types.rs @@ -116,149 +116,9 @@ pub struct ApprovalRequest { // --- SSE Event Types --- -#[derive(Debug, Clone, Serialize)] -#[serde(tag = "type")] -pub enum SseEvent { - #[serde(rename = "response")] - Response { content: String, thread_id: String }, - #[serde(rename = "thinking")] - Thinking { - message: String, - #[serde(skip_serializing_if = "Option::is_none")] - thread_id: Option, - }, - #[serde(rename = "tool_started")] - ToolStarted { - name: String, - #[serde(skip_serializing_if = "Option::is_none")] - thread_id: Option, - }, - #[serde(rename = "tool_completed")] - ToolCompleted { - name: String, - success: bool, - #[serde(skip_serializing_if = "Option::is_none")] - error: Option, - #[serde(skip_serializing_if = "Option::is_none")] - parameters: Option, - #[serde(skip_serializing_if = "Option::is_none")] - thread_id: Option, - }, - #[serde(rename = "tool_result")] - ToolResult { - name: String, - preview: String, - #[serde(skip_serializing_if = "Option::is_none")] - thread_id: Option, - }, - #[serde(rename = "stream_chunk")] - StreamChunk { - content: String, - #[serde(skip_serializing_if = "Option::is_none")] - thread_id: Option, - }, - #[serde(rename = "status")] - Status { - message: String, - #[serde(skip_serializing_if = "Option::is_none")] - thread_id: Option, - }, - #[serde(rename = "job_started")] - JobStarted { - job_id: String, - title: String, - browse_url: String, - }, - #[serde(rename = "approval_needed")] - ApprovalNeeded { - request_id: String, - tool_name: String, - description: String, - parameters: String, - #[serde(skip_serializing_if = "Option::is_none")] - thread_id: Option, - }, - #[serde(rename = "auth_required")] - AuthRequired { - extension_name: String, - #[serde(skip_serializing_if = "Option::is_none")] - instructions: Option, - #[serde(skip_serializing_if = "Option::is_none")] - auth_url: Option, - #[serde(skip_serializing_if = "Option::is_none")] - setup_url: Option, - }, - #[serde(rename = "auth_completed")] - AuthCompleted { - extension_name: String, - success: bool, - message: String, - }, - #[serde(rename = "error")] - Error { - message: String, - #[serde(skip_serializing_if = "Option::is_none")] - thread_id: Option, - }, - #[serde(rename = "heartbeat")] - Heartbeat, - - // Sandbox job streaming events (worker + Claude Code bridge) - #[serde(rename = "job_message")] - JobMessage { - job_id: String, - role: String, - content: String, - }, - #[serde(rename = "job_tool_use")] - JobToolUse { - job_id: String, - tool_name: String, - input: serde_json::Value, - }, - #[serde(rename = "job_tool_result")] - JobToolResult { - job_id: String, - tool_name: String, - output: String, - }, - #[serde(rename = "job_status")] - JobStatus { job_id: String, message: String }, - #[serde(rename = "job_result")] - JobResult { - job_id: String, - status: String, - #[serde(skip_serializing_if = "Option::is_none")] - session_id: Option, - }, - - /// An image was generated by a tool. - #[serde(rename = "image_generated")] - ImageGenerated { - data_url: String, - #[serde(skip_serializing_if = "Option::is_none")] - path: Option, - #[serde(skip_serializing_if = "Option::is_none")] - thread_id: Option, - }, - - /// Suggested follow-up messages for the user. - #[serde(rename = "suggestions")] - Suggestions { - suggestions: Vec, - #[serde(skip_serializing_if = "Option::is_none")] - thread_id: Option, - }, - - /// Extension activation status change (WASM channels). - #[serde(rename = "extension_status")] - ExtensionStatus { - extension_name: String, - status: String, - #[serde(skip_serializing_if = "Option::is_none")] - message: Option, - }, -} +/// Re-export from `crate::events::DomainEvent` — the canonical event enum now +/// lives in a channel-neutral location so agent code doesn't depend on `channels::web`. +pub use crate::events::DomainEvent as SseEvent; // --- Memory --- diff --git a/src/channels/web/util.rs b/src/channels/web/util.rs index 060afeab..03571cf4 100644 --- a/src/channels/web/util.rs +++ b/src/channels/web/util.rs @@ -2,28 +2,10 @@ use crate::channels::web::types::{ToolCallInfo, TurnInfo}; -/// Truncate a string to at most `max_bytes` bytes at a char boundary, appending "...". -/// -/// If the input is wrapped in `…` and truncation -/// removes the closing tag, the tag is re-appended so downstream XML parsers -/// never see an unclosed element. +/// Delegates to [`crate::util::truncate_preview`] — the canonical implementation +/// now lives in the shared utility module so non-web code can use it too. pub fn truncate_preview(s: &str, max_bytes: usize) -> String { - if s.len() <= max_bytes { - return s.to_string(); - } - // Walk backwards from max_bytes to find a valid char boundary - let mut end = max_bytes; - while end > 0 && !s.is_char_boundary(end) { - end -= 1; - } - let mut result = format!("{}...", &s[..end]); - - // Re-close if truncation cut through the closing tag. - if s.starts_with("") { - result.push_str("\n"); - } - - result + crate::util::truncate_preview(s, max_bytes) } /// Build TurnInfo pairs from flat DB messages (user/tool_calls/assistant triples). diff --git a/src/context/state.rs b/src/context/state.rs index 22aca311..3d811d30 100644 --- a/src/context/state.rs +++ b/src/context/state.rs @@ -9,7 +9,7 @@ use rust_decimal::Decimal; use serde::{Deserialize, Serialize}; use uuid::Uuid; -use crate::llm::recording::HttpInterceptor; +use crate::observability::HttpInterceptor; /// Error returned when a job exceeds its token budget. #[derive(Debug, thiserror::Error)] diff --git a/src/db/mod.rs b/src/db/mod.rs index 4afd1db8..bf13140c 100644 --- a/src/db/mod.rs +++ b/src/db/mod.rs @@ -29,8 +29,6 @@ use chrono::{DateTime, Utc}; use rust_decimal::Decimal; use uuid::Uuid; -use crate::agent::BrokenTool; -use crate::agent::routine::{Routine, RoutineRun, RunStatus}; use crate::context::{ActionRecord, JobContext, JobState}; use crate::error::DatabaseError; use crate::error::WorkspaceError; @@ -38,6 +36,8 @@ use crate::history::{ AgentJobRecord, AgentJobSummary, ConversationMessage, ConversationSummary, JobEventRecord, LlmCallRecord, SandboxJobRecord, SandboxJobSummary, SettingRow, }; +use crate::models::routine::{Routine, RoutineRun, RunStatus}; +use crate::models::tool_failure::ToolFailureRecord; use crate::workspace::{MemoryChunk, MemoryDocument, WorkspaceEntry}; use crate::workspace::{SearchConfig, SearchResult}; @@ -401,7 +401,10 @@ pub trait ToolFailureStore: Send + Sync { tool_name: &str, error_message: &str, ) -> Result<(), DatabaseError>; - async fn get_broken_tools(&self, threshold: i32) -> Result, DatabaseError>; + async fn get_broken_tools( + &self, + threshold: i32, + ) -> Result, DatabaseError>; async fn mark_tool_repaired(&self, tool_name: &str) -> Result<(), DatabaseError>; async fn increment_repair_attempts(&self, tool_name: &str) -> Result<(), DatabaseError>; } diff --git a/src/events.rs b/src/events.rs new file mode 100644 index 00000000..7e289280 --- /dev/null +++ b/src/events.rs @@ -0,0 +1,159 @@ +//! Domain events for cross-module communication. +//! +//! `DomainEvent` is the canonical event type published by the agent, scheduler, +//! and other core modules. Channel-specific code (web gateway, CLI, etc.) +//! subscribes and maps these to its wire format. +//! +//! By living in `src/events.rs` rather than `channels::web::types`, these events +//! can be used by any module without creating a dependency on a specific channel. + +use serde::Serialize; + +/// Domain events emitted by the agent and related subsystems. +/// +/// The `#[serde(tag = "type")]` attribute ensures each variant serializes with +/// a `"type"` discriminator field, matching the SSE wire format expected by +/// the web gateway. +#[derive(Debug, Clone, Serialize)] +#[serde(tag = "type")] +pub enum DomainEvent { + #[serde(rename = "response")] + Response { content: String, thread_id: String }, + #[serde(rename = "thinking")] + Thinking { + message: String, + #[serde(skip_serializing_if = "Option::is_none")] + thread_id: Option, + }, + #[serde(rename = "tool_started")] + ToolStarted { + name: String, + #[serde(skip_serializing_if = "Option::is_none")] + thread_id: Option, + }, + #[serde(rename = "tool_completed")] + ToolCompleted { + name: String, + success: bool, + #[serde(skip_serializing_if = "Option::is_none")] + error: Option, + #[serde(skip_serializing_if = "Option::is_none")] + parameters: Option, + #[serde(skip_serializing_if = "Option::is_none")] + thread_id: Option, + }, + #[serde(rename = "tool_result")] + ToolResult { + name: String, + preview: String, + #[serde(skip_serializing_if = "Option::is_none")] + thread_id: Option, + }, + #[serde(rename = "stream_chunk")] + StreamChunk { + content: String, + #[serde(skip_serializing_if = "Option::is_none")] + thread_id: Option, + }, + #[serde(rename = "status")] + Status { + message: String, + #[serde(skip_serializing_if = "Option::is_none")] + thread_id: Option, + }, + #[serde(rename = "job_started")] + JobStarted { + job_id: String, + title: String, + browse_url: String, + }, + #[serde(rename = "approval_needed")] + ApprovalNeeded { + request_id: String, + tool_name: String, + description: String, + parameters: String, + #[serde(skip_serializing_if = "Option::is_none")] + thread_id: Option, + }, + #[serde(rename = "auth_required")] + AuthRequired { + extension_name: String, + #[serde(skip_serializing_if = "Option::is_none")] + instructions: Option, + #[serde(skip_serializing_if = "Option::is_none")] + auth_url: Option, + #[serde(skip_serializing_if = "Option::is_none")] + setup_url: Option, + }, + #[serde(rename = "auth_completed")] + AuthCompleted { + extension_name: String, + success: bool, + message: String, + }, + #[serde(rename = "error")] + Error { + message: String, + #[serde(skip_serializing_if = "Option::is_none")] + thread_id: Option, + }, + #[serde(rename = "heartbeat")] + Heartbeat, + + // Sandbox job streaming events (worker + Claude Code bridge) + #[serde(rename = "job_message")] + JobMessage { + job_id: String, + role: String, + content: String, + }, + #[serde(rename = "job_tool_use")] + JobToolUse { + job_id: String, + tool_name: String, + input: serde_json::Value, + }, + #[serde(rename = "job_tool_result")] + JobToolResult { + job_id: String, + tool_name: String, + output: String, + }, + #[serde(rename = "job_status")] + JobStatus { job_id: String, message: String }, + #[serde(rename = "job_result")] + JobResult { + job_id: String, + status: String, + #[serde(skip_serializing_if = "Option::is_none")] + session_id: Option, + }, + + /// An image was generated by a tool. + #[serde(rename = "image_generated")] + ImageGenerated { + data_url: String, + #[serde(skip_serializing_if = "Option::is_none")] + path: Option, + #[serde(skip_serializing_if = "Option::is_none")] + thread_id: Option, + }, + + /// Suggested follow-up messages for the user. + #[serde(rename = "suggestions")] + Suggestions { + suggestions: Vec, + #[serde(skip_serializing_if = "Option::is_none")] + thread_id: Option, + }, + + /// Extension activation status change (WASM channels). + #[serde(rename = "extension_status")] + ExtensionStatus { + extension_name: String, + status: String, + #[serde(skip_serializing_if = "Option::is_none")] + message: Option, + }, +} diff --git a/src/lib.rs b/src/lib.rs index 51e54909..689d9fed 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -51,16 +51,19 @@ pub mod document_extraction; pub mod error; pub mod estimation; pub mod evaluation; +pub mod events; pub mod extensions; pub mod history; pub mod hooks; #[cfg(feature = "import")] pub mod import; pub mod llm; +pub mod models; pub mod observability; pub mod orchestrator; pub mod pairing; pub mod registry; +pub mod resilience; pub mod safety; pub mod sandbox; pub mod secrets; @@ -68,6 +71,7 @@ pub mod service; pub mod settings; pub mod setup; pub mod skills; +pub mod state_bus; pub mod timezone; pub mod tools; pub mod tracing_fmt; diff --git a/src/llm/recording.rs b/src/llm/recording.rs index 77f7b257..e135dd6f 100644 --- a/src/llm/recording.rs +++ b/src/llm/recording.rs @@ -51,32 +51,10 @@ pub struct MemorySnapshotEntry { pub content: String, } -/// A recorded HTTP request/response pair. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct HttpExchange { - pub request: HttpExchangeRequest, - pub response: HttpExchangeResponse, -} - -/// The request side of an HTTP exchange. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct HttpExchangeRequest { - pub method: String, - pub url: String, - #[serde(default, skip_serializing_if = "Vec::is_empty")] - pub headers: Vec<(String, String)>, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub body: Option, -} - -/// The response side of an HTTP exchange. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct HttpExchangeResponse { - pub status: u16, - #[serde(default, skip_serializing_if = "Vec::is_empty")] - pub headers: Vec<(String, String)>, - pub body: String, -} +// Re-export HTTP exchange types from their canonical location in observability. +pub use crate::observability::http_interceptor::{ + HttpExchange, HttpExchangeRequest, HttpExchangeResponse, HttpInterceptor, +}; /// A single step in the trace. #[derive(Debug, Clone, Serialize, Deserialize)] @@ -144,23 +122,7 @@ pub struct ExpectedToolResult { pub content: String, } -// ── HTTP interceptor ─────────────────────────────────────────────── - -/// Trait for intercepting HTTP requests from tools. -/// -/// During recording, the interceptor captures exchanges after the real -/// request completes. During replay, it short-circuits with a recorded response. -#[async_trait] -pub trait HttpInterceptor: Send + Sync + std::fmt::Debug { - /// Called before making an HTTP request. - /// - /// Return `Some(response)` to short-circuit (replay mode). - /// Return `None` to let the real request proceed (recording mode). - async fn before_request(&self, request: &HttpExchangeRequest) -> Option; - - /// Called after a real HTTP request completes (recording mode only). - async fn after_response(&self, request: &HttpExchangeRequest, response: &HttpExchangeResponse); -} +// ── HTTP interceptor impls ───────────────────────────────────────── /// Records HTTP exchanges during a live session. #[derive(Debug)] diff --git a/src/models/mod.rs b/src/models/mod.rs new file mode 100644 index 00000000..2303e9da --- /dev/null +++ b/src/models/mod.rs @@ -0,0 +1,13 @@ +//! Shared domain types used across module boundaries. +//! +//! Types in this module are imported by both the persistence layer (`db`) and +//! the domain logic (`agent`), breaking the circular dependency that existed +//! when these types lived inside `agent/`. + +pub mod routine; +pub mod tool_failure; + +pub use routine::{ + NotifyConfig, Routine, RoutineAction, RoutineGuardrails, RoutineRun, RunStatus, Trigger, +}; +pub use tool_failure::ToolFailureRecord; diff --git a/src/models/routine.rs b/src/models/routine.rs new file mode 100644 index 00000000..a27b2c49 --- /dev/null +++ b/src/models/routine.rs @@ -0,0 +1,823 @@ +//! Core types for the routines system. +//! +//! A routine is a named, persistent, user-owned task with a trigger and an action. +//! Each routine fires independently when its trigger condition is met, with only +//! that routine's prompt and context sent to the LLM. +//! +//! These types live in `models` (rather than `agent`) so that both the `db` and +//! `agent` modules can import them without circular dependencies. + +use std::collections::hash_map::DefaultHasher; +use std::hash::{Hash, Hasher}; +use std::str::FromStr; +use std::time::Duration; + +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; +use uuid::Uuid; + +use crate::error::RoutineError; + +/// A routine is a named, persistent, user-owned task with a trigger and an action. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct Routine { + pub id: Uuid, + pub name: String, + pub description: String, + pub user_id: String, + pub enabled: bool, + pub trigger: Trigger, + pub action: RoutineAction, + pub guardrails: RoutineGuardrails, + pub notify: NotifyConfig, + + // Runtime state (DB-managed) + pub last_run_at: Option>, + pub next_fire_at: Option>, + pub run_count: u64, + pub consecutive_failures: u32, + pub state: serde_json::Value, + + pub created_at: DateTime, + pub updated_at: DateTime, +} + +/// When a routine should fire. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(tag = "type", rename_all = "snake_case")] +pub enum Trigger { + /// Fire on a cron schedule (e.g. "0 9 * * MON-FRI" or "every 2h"). + Cron { + schedule: String, + #[serde(default)] + timezone: Option, + }, + /// Fire when a channel message matches a pattern. + Event { + /// Optional channel filter (e.g. "telegram", "slack"). + channel: Option, + /// Regex pattern to match against message content. + pattern: String, + }, + /// Fire when a structured system event is emitted. + SystemEvent { + /// Event source namespace (e.g. "github", "workflow", "tool"). + source: String, + /// Event type within the source (e.g. "issue.opened"). + event_type: String, + /// Optional exact-match filters against payload top-level fields. + #[serde(default)] + filters: std::collections::HashMap, + }, + /// Only fires via tool call or CLI. + Manual, +} + +impl Trigger { + /// The string tag stored in the DB trigger_type column. + pub fn type_tag(&self) -> &'static str { + match self { + Trigger::Cron { .. } => "cron", + Trigger::Event { .. } => "event", + Trigger::SystemEvent { .. } => "system_event", + Trigger::Manual => "manual", + } + } + + /// Parse a trigger from its DB representation. + pub fn from_db(trigger_type: &str, config: serde_json::Value) -> Result { + match trigger_type { + "cron" => { + let schedule = config + .get("schedule") + .and_then(|v| v.as_str()) + .ok_or_else(|| RoutineError::MissingField { + context: "cron trigger".into(), + field: "schedule".into(), + })? + .to_string(); + let timezone = config + .get("timezone") + .and_then(|v| v.as_str()) + .and_then(|tz| { + if crate::timezone::parse_timezone(tz).is_some() { + Some(tz.to_string()) + } else { + tracing::warn!( + "Ignoring invalid timezone '{}' from DB for cron trigger", + tz + ); + None + } + }); + Ok(Trigger::Cron { schedule, timezone }) + } + "event" => { + let pattern = config + .get("pattern") + .and_then(|v| v.as_str()) + .ok_or_else(|| RoutineError::MissingField { + context: "event trigger".into(), + field: "pattern".into(), + })? + .to_string(); + let channel = config + .get("channel") + .and_then(|v| v.as_str()) + .map(String::from); + Ok(Trigger::Event { channel, pattern }) + } + "system_event" => { + let source = config + .get("source") + .and_then(|v| v.as_str()) + .ok_or_else(|| RoutineError::MissingField { + context: "system_event trigger".into(), + field: "source".into(), + })? + .to_string(); + let event_type = config + .get("event_type") + .and_then(|v| v.as_str()) + .ok_or_else(|| RoutineError::MissingField { + context: "system_event trigger".into(), + field: "event_type".into(), + })? + .to_string(); + let filters = config + .get("filters") + .and_then(|v| v.as_object()) + .map(|m| { + m.iter() + .filter_map(|(k, v)| { + json_value_as_filter_string(v).map(|s| (k.clone(), s)) + }) + .collect() + }) + .unwrap_or_default(); + Ok(Trigger::SystemEvent { + source, + event_type, + filters, + }) + } + "manual" => Ok(Trigger::Manual), + other => Err(RoutineError::UnknownTriggerType { + trigger_type: other.to_string(), + }), + } + } + + /// Serialize trigger-specific config to JSON for DB storage. + pub fn to_config_json(&self) -> serde_json::Value { + match self { + Trigger::Cron { schedule, timezone } => serde_json::json!({ + "schedule": schedule, + "timezone": timezone, + }), + Trigger::Event { channel, pattern } => serde_json::json!({ + "pattern": pattern, + "channel": channel, + }), + Trigger::SystemEvent { + source, + event_type, + filters, + } => serde_json::json!({ + "source": source, + "event_type": event_type, + "filters": filters, + }), + Trigger::Manual => serde_json::json!({}), + } + } +} + +/// What happens when a routine fires. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(tag = "type", rename_all = "snake_case")] +pub enum RoutineAction { + /// Single LLM call (optionally with tools). Cheap and fast. + Lightweight { + /// The prompt sent to the LLM. + prompt: String, + /// Workspace paths to load as context (e.g. ["context/priorities.md"]). + #[serde(default)] + context_paths: Vec, + /// Max output tokens (default: 4096). + #[serde(default = "default_max_tokens")] + max_tokens: u32, + /// Enable tool access (default: false for backward compatibility). + /// When true, the LLM can call tools during execution. + /// Tools requiring approval are automatically filtered out. + #[serde(default)] + use_tools: bool, + /// Max tool call rounds (default: 3). Only used when use_tools is true. + #[serde(default = "default_max_tool_rounds")] + max_tool_rounds: u32, + }, + /// Full multi-turn worker job with tool access. + FullJob { + /// Job title for the scheduler. + title: String, + /// Job description / initial prompt. + description: String, + /// Max reasoning iterations (default: 10). + #[serde(default = "default_max_iterations")] + max_iterations: u32, + /// Tool names pre-authorized for `Always`-approval tools (e.g. destructive + /// shell commands, cross-channel messaging). `UnlessAutoApproved` tools are + /// automatically permitted in routine jobs without listing them here. + #[serde(default)] + tool_permissions: Vec, + }, +} + +fn default_max_tokens() -> u32 { + 4096 +} + +fn default_max_iterations() -> u32 { + 10 +} + +fn default_max_tool_rounds() -> u32 { + 3 +} + +/// Hard upper bound for max_tool_rounds to prevent runaway loops and cost explosion. +pub(crate) const MAX_TOOL_ROUNDS_LIMIT: u32 = 20; + +/// Clamp max_tool_rounds to [1, MAX_TOOL_ROUNDS_LIMIT]. +/// Accepts u64 to avoid truncation before clamping. +fn clamp_max_tool_rounds(value: u64) -> u32 { + value.clamp(1, MAX_TOOL_ROUNDS_LIMIT as u64) as u32 +} + +/// Parse a `tool_permissions` JSON array into a `Vec`. +pub fn parse_tool_permissions(value: &serde_json::Value) -> Vec { + value + .get("tool_permissions") + .and_then(|v| v.as_array()) + .map(|arr| { + arr.iter() + .filter_map(|v| v.as_str().map(String::from)) + .collect() + }) + .unwrap_or_default() +} + +impl RoutineAction { + /// The string tag stored in the DB action_type column. + pub fn type_tag(&self) -> &'static str { + match self { + RoutineAction::Lightweight { .. } => "lightweight", + RoutineAction::FullJob { .. } => "full_job", + } + } + + /// Parse an action from its DB representation. + pub fn from_db(action_type: &str, config: serde_json::Value) -> Result { + match action_type { + "lightweight" => { + let prompt = config + .get("prompt") + .and_then(|v| v.as_str()) + .ok_or_else(|| RoutineError::MissingField { + context: "lightweight action".into(), + field: "prompt".into(), + })? + .to_string(); + let context_paths = config + .get("context_paths") + .and_then(|v| v.as_array()) + .map(|arr| { + arr.iter() + .filter_map(|v| v.as_str().map(String::from)) + .collect() + }) + .unwrap_or_default(); + let max_tokens = config + .get("max_tokens") + .and_then(|v| v.as_u64()) + .unwrap_or(default_max_tokens() as u64) as u32; + let use_tools = config + .get("use_tools") + .and_then(|v| v.as_bool()) + .unwrap_or(false); + let max_tool_rounds = clamp_max_tool_rounds( + config + .get("max_tool_rounds") + .and_then(|v| v.as_u64()) + .unwrap_or(default_max_tool_rounds() as u64), + ); + Ok(RoutineAction::Lightweight { + prompt, + context_paths, + max_tokens, + use_tools, + max_tool_rounds, + }) + } + "full_job" => { + let title = config + .get("title") + .and_then(|v| v.as_str()) + .ok_or_else(|| RoutineError::MissingField { + context: "full_job action".into(), + field: "title".into(), + })? + .to_string(); + let description = config + .get("description") + .and_then(|v| v.as_str()) + .ok_or_else(|| RoutineError::MissingField { + context: "full_job action".into(), + field: "description".into(), + })? + .to_string(); + let max_iterations = config + .get("max_iterations") + .and_then(|v| v.as_u64()) + .unwrap_or(default_max_iterations() as u64) + as u32; + let tool_permissions = parse_tool_permissions(&config); + Ok(RoutineAction::FullJob { + title, + description, + max_iterations, + tool_permissions, + }) + } + other => Err(RoutineError::UnknownActionType { + action_type: other.to_string(), + }), + } + } + + /// Serialize action config to JSON for DB storage. + pub fn to_config_json(&self) -> serde_json::Value { + match self { + RoutineAction::Lightweight { + prompt, + context_paths, + max_tokens, + use_tools, + max_tool_rounds, + } => serde_json::json!({ + "prompt": prompt, + "context_paths": context_paths, + "max_tokens": max_tokens, + "use_tools": use_tools, + "max_tool_rounds": max_tool_rounds, + }), + RoutineAction::FullJob { + title, + description, + max_iterations, + tool_permissions, + } => serde_json::json!({ + "title": title, + "description": description, + "max_iterations": max_iterations, + "tool_permissions": tool_permissions, + }), + } + } +} + +/// Guardrails to prevent runaway execution. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct RoutineGuardrails { + /// Minimum time between fires. + pub cooldown: Duration, + /// Max simultaneous runs of this routine. + pub max_concurrent: u32, + /// Window for content-hash dedup (event triggers). None = no dedup. + pub dedup_window: Option, +} + +impl Default for RoutineGuardrails { + fn default() -> Self { + Self { + cooldown: Duration::from_secs(300), + max_concurrent: 1, + dedup_window: None, + } + } +} + +/// Notification preferences for a routine. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct NotifyConfig { + /// Channel to notify on (None = default/broadcast all). + pub channel: Option, + /// User to notify. + pub user: String, + /// Notify when routine produces actionable output. + pub on_attention: bool, + /// Notify when routine errors. + pub on_failure: bool, + /// Notify when routine runs with no findings. + pub on_success: bool, +} + +impl Default for NotifyConfig { + fn default() -> Self { + Self { + channel: None, + user: "default".to_string(), + on_attention: true, + on_failure: true, + on_success: false, + } + } +} + +/// Status of a routine run. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum RunStatus { + Running, + Ok, + Attention, + Failed, +} + +impl std::fmt::Display for RunStatus { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + RunStatus::Running => write!(f, "running"), + RunStatus::Ok => write!(f, "ok"), + RunStatus::Attention => write!(f, "attention"), + RunStatus::Failed => write!(f, "failed"), + } + } +} + +impl FromStr for RunStatus { + type Err = RoutineError; + fn from_str(s: &str) -> Result { + match s { + "running" => Ok(RunStatus::Running), + "ok" => Ok(RunStatus::Ok), + "attention" => Ok(RunStatus::Attention), + "failed" => Ok(RunStatus::Failed), + other => Err(RoutineError::UnknownRunStatus { + status: other.to_string(), + }), + } + } +} + +/// A single execution of a routine. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct RoutineRun { + pub id: Uuid, + pub routine_id: Uuid, + pub trigger_type: String, + pub trigger_detail: Option, + pub started_at: DateTime, + pub completed_at: Option>, + pub status: RunStatus, + pub result_summary: Option, + pub tokens_used: Option, + pub job_id: Option, + pub created_at: DateTime, +} + +/// Convert a JSON value to a string for filter storage. +/// +/// Handles strings, numbers, and booleans — consistent with the matching +/// logic in `routine_engine::json_value_as_string`. +pub fn json_value_as_filter_string(v: &serde_json::Value) -> Option { + match v { + serde_json::Value::String(s) => Some(s.clone()), + serde_json::Value::Number(n) => Some(n.to_string()), + serde_json::Value::Bool(b) => Some(b.to_string()), + _ => None, + } +} + +/// Compute a content hash for event dedup. +pub fn content_hash(content: &str) -> u64 { + let mut hasher = DefaultHasher::new(); + content.hash(&mut hasher); + hasher.finish() +} + +/// Parse a cron expression and compute the next fire time from now. +/// +/// When `timezone` is provided and valid, the schedule is evaluated in that +/// timezone and the result is converted back to UTC. Otherwise UTC is used. +pub fn next_cron_fire( + schedule: &str, + timezone: Option<&str>, +) -> Result>, RoutineError> { + let cron_schedule = + cron::Schedule::from_str(schedule).map_err(|e| RoutineError::InvalidCron { + reason: e.to_string(), + })?; + if let Some(tz) = timezone.and_then(crate::timezone::parse_timezone) { + Ok(cron_schedule + .upcoming(tz) + .next() + .map(|dt| dt.with_timezone(&Utc))) + } else { + Ok(cron_schedule.upcoming(Utc).next()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_trigger_roundtrip() { + let trigger = Trigger::Cron { + schedule: "0 9 * * MON-FRI".to_string(), + timezone: None, + }; + let json = trigger.to_config_json(); + let parsed = Trigger::from_db("cron", json).expect("parse cron"); // safety: test-only + assert!(matches!(parsed, Trigger::Cron { schedule, .. } if schedule == "0 9 * * MON-FRI")); // safety: test-only + } + + #[test] + fn test_event_trigger_roundtrip() { + let trigger = Trigger::Event { + channel: Some("telegram".to_string()), + pattern: r"deploy\s+\w+".to_string(), + }; + let json = trigger.to_config_json(); + let parsed = Trigger::from_db("event", json).expect("parse event"); // safety: test-only + assert!( // safety: test-only + // safety: test-only + matches!(parsed, Trigger::Event { channel, pattern } // safety: test-only + if channel == Some("telegram".to_string()) && pattern == r"deploy\s+\w+") + ); + } + + #[test] + fn test_system_event_trigger_roundtrip() { + let mut filters = std::collections::HashMap::new(); + filters.insert("repo".to_string(), "nearai/ironclaw".to_string()); + filters.insert("action".to_string(), "opened".to_string()); + let trigger = Trigger::SystemEvent { + source: "github".to_string(), + event_type: "issue".to_string(), + filters: filters.clone(), + }; + let json = trigger.to_config_json(); + let parsed = Trigger::from_db("system_event", json).expect("parse system_event"); // safety: test-only + assert!( // safety: test-only + // safety: test-only + // safety: test-only + matches!(parsed, Trigger::SystemEvent { source, event_type, filters: f } + if source == "github" && event_type == "issue" && f == filters) + ); + } + + #[test] + fn test_action_lightweight_roundtrip() { + let action = RoutineAction::Lightweight { + prompt: "Check PRs".to_string(), + context_paths: vec!["context/priorities.md".to_string()], + max_tokens: 2048, + use_tools: false, + max_tool_rounds: 3, + }; + let json = action.to_config_json(); + let parsed = RoutineAction::from_db("lightweight", json).expect("parse lightweight"); // safety: test-only + assert!( // safety: test-only + // safety: test-only + // safety: test-only + matches!(parsed, RoutineAction::Lightweight { prompt, context_paths, max_tokens, .. } + if prompt == "Check PRs" && context_paths.len() == 1 && max_tokens == 2048) + ); + } + + #[test] + fn test_action_full_job_roundtrip() { + let action = RoutineAction::FullJob { + title: "Deploy review".to_string(), + description: "Review and deploy pending changes".to_string(), + max_iterations: 5, + tool_permissions: vec!["shell".to_string()], + }; + let json = action.to_config_json(); + let parsed = RoutineAction::from_db("full_job", json).expect("parse full_job"); // safety: test-only + assert!( // safety: test-only + // safety: test-only + // safety: test-only + matches!(parsed, RoutineAction::FullJob { title, max_iterations, tool_permissions, .. } + if title == "Deploy review" && max_iterations == 5 && tool_permissions == vec!["shell".to_string()]) + ); + } + + #[test] + fn test_run_status_display_parse() { + for status in [ + RunStatus::Running, + RunStatus::Ok, + RunStatus::Attention, + RunStatus::Failed, + ] { + let s = status.to_string(); + let parsed: RunStatus = s.parse().expect("parse status"); // safety: test-only + assert_eq!(parsed, status); // safety: test-only + } + } + + #[test] + fn test_content_hash_deterministic() { + let h1 = content_hash("deploy production"); + let h2 = content_hash("deploy production"); + assert_eq!(h1, h2); // safety: test-only + + let h3 = content_hash("deploy staging"); + assert_ne!(h1, h3); // safety: test-only + } + + #[test] + fn test_next_cron_fire_valid() { + // Every minute should always have a next fire + let next = next_cron_fire("* * * * * *", None).expect("valid cron"); // safety: test-only + assert!(next.is_some()); // safety: test-only + } + + #[test] + fn test_next_cron_fire_invalid() { + let result = next_cron_fire("not a cron", None); + assert!(result.is_err()); // safety: test-only + } + + #[test] + fn test_trigger_cron_timezone_roundtrip() { + let trigger = Trigger::Cron { + schedule: "0 9 * * MON-FRI".to_string(), + timezone: Some("America/New_York".to_string()), + }; + let json = trigger.to_config_json(); + let parsed = Trigger::from_db("cron", json).expect("parse cron"); // safety: test-only + assert!( // safety: test-only + // safety: test-only + matches!(parsed, Trigger::Cron { schedule, timezone } // safety: test-only + if schedule == "0 9 * * MON-FRI" + && timezone.as_deref() == Some("America/New_York")) + ); + } + + #[test] + fn test_trigger_cron_no_timezone_backward_compat() { + let json = serde_json::json!({"schedule": "0 9 * * *"}); + let parsed = Trigger::from_db("cron", json).expect("parse cron"); // safety: test-only + assert!(matches!(parsed, Trigger::Cron { timezone, .. } if timezone.is_none())); // safety: test-only + } + + #[test] + fn test_trigger_cron_invalid_timezone_coerced_to_none() { + let json = serde_json::json!({"schedule": "0 9 * * *", "timezone": "Fake/Zone"}); + let parsed = Trigger::from_db("cron", json).expect("parse cron"); // safety: test-only + assert!( // safety: test-only + // safety: test-only + // safety: test-only + matches!(parsed, Trigger::Cron { timezone, .. } if timezone.is_none()), + "invalid timezone should be coerced to None" + ); + } + + #[test] + fn test_next_cron_fire_with_timezone() { + let next_utc = next_cron_fire("0 0 9 * * * *", None) + .expect("valid cron") // safety: test-only + .expect("has next"); // safety: test-only + let next_est = next_cron_fire("0 0 9 * * * *", Some("America/New_York")) + .expect("valid cron") // safety: test-only + .expect("has next"); // safety: test-only + // EST is UTC-5 (or EDT UTC-4), so the UTC result should differ + assert_ne!(next_utc, next_est, "timezone should shift the fire time"); // safety: test-only + } + + #[test] + fn test_guardrails_default() { + let g = RoutineGuardrails::default(); + assert_eq!(g.cooldown.as_secs(), 300); // safety: test-only + assert_eq!(g.max_concurrent, 1); // safety: test-only + assert!(g.dedup_window.is_none()); // safety: test-only + } + + #[test] + fn test_trigger_type_tag() { + assert_eq!( // safety: test-only + // safety: test-only + // safety: test-only + Trigger::Cron { + schedule: String::new(), + timezone: None, + } + .type_tag(), + "cron" + ); + assert_eq!( // safety: test-only + // safety: test-only + // safety: test-only + Trigger::Event { + channel: None, + pattern: String::new() + } + .type_tag(), + "event" + ); + assert_eq!( // safety: test-only + // safety: test-only + // safety: test-only + Trigger::SystemEvent { + source: String::new(), + event_type: String::new(), + filters: std::collections::HashMap::new(), + } + .type_tag(), + "system_event" + ); + assert_eq!(Trigger::Manual.type_tag(), "manual"); // safety: test-only + } + + #[test] + fn test_action_lightweight_backward_compat_no_use_tools() { + // Simulate old DB record without use_tools field + let json = serde_json::json!({ + "prompt": "old routine", + "context_paths": [], + "max_tokens": 4096 + }); + let parsed = RoutineAction::from_db("lightweight", json).expect("parse lightweight"); // safety: test-only + assert!( // safety: test-only + // safety: test-only + // safety: test-only + matches!(parsed, RoutineAction::Lightweight { use_tools, max_tool_rounds, .. } + if !use_tools && max_tool_rounds == 3), + "missing use_tools should default to false, max_tool_rounds to 3" + ); + } + + #[test] + fn test_max_tool_rounds_clamped_to_upper_bound() { + let json = serde_json::json!({ + "prompt": "test", + "use_tools": true, + "max_tool_rounds": 9999 + }); + let parsed = RoutineAction::from_db("lightweight", json).expect("parse"); // safety: test-only + match parsed { + RoutineAction::Lightweight { + max_tool_rounds, .. + } => { + assert_eq!( // safety: test-only + // safety: test-only + // safety: test-only + max_tool_rounds, + MAX_TOOL_ROUNDS_LIMIT, + "should clamp to MAX_TOOL_ROUNDS_LIMIT" + ); + } + _ => panic!("expected Lightweight"), + } + } + + #[test] + fn test_max_tool_rounds_clamped_to_lower_bound() { + let json = serde_json::json!({ + "prompt": "test", + "use_tools": true, + "max_tool_rounds": 0 + }); + let parsed = RoutineAction::from_db("lightweight", json).expect("parse"); // safety: test-only + match parsed { + RoutineAction::Lightweight { + max_tool_rounds, .. + } => { + assert_eq!(max_tool_rounds, 1, "should clamp 0 to 1"); // safety: test-only + } + _ => panic!("expected Lightweight"), + } + } + + #[test] + fn test_max_tool_rounds_normal_value_passes_through() { + let json = serde_json::json!({ + "prompt": "test", + "use_tools": true, + "max_tool_rounds": 10 + }); + let parsed = RoutineAction::from_db("lightweight", json).expect("parse"); // safety: test-only + match parsed { + RoutineAction::Lightweight { + max_tool_rounds, .. + } => { + assert_eq!(max_tool_rounds, 10, "normal value should pass through"); // safety: test-only + } + _ => panic!("expected Lightweight"), + } + } +} diff --git a/src/models/tool_failure.rs b/src/models/tool_failure.rs new file mode 100644 index 00000000..fc55daa2 --- /dev/null +++ b/src/models/tool_failure.rs @@ -0,0 +1,19 @@ +//! Tool failure tracking types shared between `db` and `agent` modules. + +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; + +/// A tool that has been detected as broken (high failure rate). +/// +/// Previously named `BrokenTool` in `agent::self_repair`. Renamed to +/// `ToolFailureRecord` to better reflect its role as a persistence DTO. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ToolFailureRecord { + pub name: String, + pub failure_count: u32, + pub last_error: Option, + pub first_failure: DateTime, + pub last_failure: DateTime, + pub last_build_result: Option, + pub repair_attempts: u32, +} diff --git a/src/observability/http_interceptor.rs b/src/observability/http_interceptor.rs new file mode 100644 index 00000000..79a32eb3 --- /dev/null +++ b/src/observability/http_interceptor.rs @@ -0,0 +1,50 @@ +//! HTTP interception trait for trace recording and replay. +//! +//! Lives in `observability` rather than `llm::recording` so that `context::state` +//! can depend on it without pulling in the LLM module. + +use async_trait::async_trait; +use serde::{Deserialize, Serialize}; + +/// The request side of an HTTP exchange. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct HttpExchangeRequest { + pub method: String, + pub url: String, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub headers: Vec<(String, String)>, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub body: Option, +} + +/// The response side of an HTTP exchange. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct HttpExchangeResponse { + pub status: u16, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub headers: Vec<(String, String)>, + pub body: String, +} + +/// A matched request/response pair. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct HttpExchange { + pub request: HttpExchangeRequest, + pub response: HttpExchangeResponse, +} + +/// Trait for intercepting HTTP requests from tools. +/// +/// During recording, the interceptor captures exchanges after the real +/// request completes. During replay, it short-circuits with a recorded response. +#[async_trait] +pub trait HttpInterceptor: Send + Sync + std::fmt::Debug { + /// Called before making an HTTP request. + /// + /// Return `Some(response)` to short-circuit (replay mode). + /// Return `None` to let the real request proceed (recording mode). + async fn before_request(&self, request: &HttpExchangeRequest) -> Option; + + /// Called after a real HTTP request completes (recording mode only). + async fn after_response(&self, request: &HttpExchangeRequest, response: &HttpExchangeResponse); +} diff --git a/src/observability/mod.rs b/src/observability/mod.rs index fa0c667a..861cc568 100644 --- a/src/observability/mod.rs +++ b/src/observability/mod.rs @@ -12,11 +12,15 @@ //! [`ObservabilityConfig`]. Future backends (OpenTelemetry, Prometheus) //! can be added by implementing [`Observer`]. +pub mod http_interceptor; mod log; mod multi; mod noop; pub mod traits; +pub use self::http_interceptor::{ + HttpExchange, HttpExchangeRequest, HttpExchangeResponse, HttpInterceptor, +}; pub use self::log::LogObserver; pub use self::multi::MultiObserver; pub use self::noop::NoopObserver; diff --git a/src/resilience/circuit_breaker.rs b/src/resilience/circuit_breaker.rs new file mode 100644 index 00000000..1718c17a --- /dev/null +++ b/src/resilience/circuit_breaker.rs @@ -0,0 +1,309 @@ +//! Generic circuit breaker with Closed/Open/HalfOpen state machine. +//! +//! Extracted from `llm::circuit_breaker` to be reusable across any +//! external service client. + +use std::time::{Duration, Instant}; + +use tokio::sync::Mutex; + +use super::classifier::ErrorClassifier; + +/// Configuration for the circuit breaker. +#[derive(Debug, Clone)] +pub struct CircuitBreakerConfig { + /// Consecutive transient failures before the circuit opens. + pub failure_threshold: u32, + /// How long the circuit stays open before allowing a probe. + pub recovery_timeout: Duration, + /// Successful probes needed in half-open to close the circuit. + pub half_open_successes_needed: u32, +} + +impl Default for CircuitBreakerConfig { + fn default() -> Self { + Self { + failure_threshold: 5, + recovery_timeout: Duration::from_secs(30), + half_open_successes_needed: 2, + } + } +} + +/// Circuit breaker states. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum CircuitState { + Closed, + Open, + HalfOpen, +} + +struct BreakerState { + state: CircuitState, + consecutive_failures: u32, + opened_at: Option, + half_open_successes: u32, +} + +impl BreakerState { + fn new() -> Self { + Self { + state: CircuitState::Closed, + consecutive_failures: 0, + opened_at: None, + half_open_successes: 0, + } + } +} + +/// Generic circuit breaker layer. +/// +/// Wraps any async operation. Tracks consecutive transient failures and +/// trips open after the threshold, fast-failing subsequent calls until +/// the recovery timeout elapses. +pub struct CircuitBreakerLayer { + state: Mutex, + config: CircuitBreakerConfig, + classifier: C, + /// Label for log messages. + label: String, +} + +impl CircuitBreakerLayer { + pub fn new(config: CircuitBreakerConfig, classifier: C, label: impl Into) -> Self { + Self { + state: Mutex::new(BreakerState::new()), + config, + classifier, + label: label.into(), + } + } + + /// Current circuit state. + pub async fn circuit_state(&self) -> CircuitState { + self.state.lock().await.state + } + + /// Number of consecutive failures. + pub async fn consecutive_failures(&self) -> u32 { + self.state.lock().await.consecutive_failures + } +} + +impl CircuitBreakerLayer { + /// Check if a call is currently allowed. + /// + /// Returns `Ok(())` if allowed, `Err(message)` if the circuit is open. + pub async fn check_allowed(&self) -> Result<(), String> { + let mut state = self.state.lock().await; + match state.state { + CircuitState::Closed | CircuitState::HalfOpen => Ok(()), + CircuitState::Open => { + if let Some(opened_at) = state.opened_at { + if opened_at.elapsed() >= self.config.recovery_timeout { + state.state = CircuitState::HalfOpen; + state.half_open_successes = 0; + tracing::info!( + label = %self.label, + "Circuit breaker: Open -> HalfOpen, allowing probe" + ); + Ok(()) + } else { + let remaining = self + .config + .recovery_timeout + .checked_sub(opened_at.elapsed()) + .unwrap_or(Duration::ZERO); + Err(format!( + "Circuit breaker open for '{}' ({} consecutive failures, \ + recovery in {:.0}s)", + self.label, + state.consecutive_failures, + remaining.as_secs_f64() + )) + } + } else { + state.state = CircuitState::Closed; + Ok(()) + } + } + } + } + + /// Record a successful call. + pub async fn record_success(&self) { + let mut state = self.state.lock().await; + match state.state { + CircuitState::Closed => { + state.consecutive_failures = 0; + } + CircuitState::HalfOpen => { + state.half_open_successes += 1; + if state.half_open_successes >= self.config.half_open_successes_needed { + state.state = CircuitState::Closed; + state.consecutive_failures = 0; + state.opened_at = None; + tracing::info!( + label = %self.label, + "Circuit breaker: HalfOpen -> Closed (recovered)" + ); + } + } + CircuitState::Open => { + state.state = CircuitState::Closed; + state.consecutive_failures = 0; + state.opened_at = None; + } + } + } + + /// Record a failed call. Only transient errors count toward the threshold. + pub async fn record_failure(&self, err: &E) + where + C: ErrorClassifier, + { + if !self.classifier.is_transient(err) { + return; + } + + let mut state = self.state.lock().await; + match state.state { + CircuitState::Closed => { + state.consecutive_failures += 1; + if state.consecutive_failures >= self.config.failure_threshold { + state.state = CircuitState::Open; + state.opened_at = Some(Instant::now()); + tracing::warn!( + label = %self.label, + failures = state.consecutive_failures, + "Circuit breaker: Closed -> Open" + ); + } + } + CircuitState::HalfOpen => { + state.state = CircuitState::Open; + state.opened_at = Some(Instant::now()); + state.half_open_successes = 0; + tracing::warn!( + label = %self.label, + "Circuit breaker: HalfOpen -> Open (probe failed)" + ); + } + CircuitState::Open => { + // Already open, nothing to do + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[derive(Debug, thiserror::Error)] + enum TestError { + #[error("transient")] + Transient, + #[error("permanent")] + Permanent, + } + + struct TestClassifier; + impl ErrorClassifier for TestClassifier { + fn is_retryable(&self, err: &TestError) -> bool { + matches!(err, TestError::Transient) + } + fn is_transient(&self, err: &TestError) -> bool { + matches!(err, TestError::Transient) + } + } + + fn make_breaker(threshold: u32) -> CircuitBreakerLayer { + CircuitBreakerLayer::new( + CircuitBreakerConfig { + failure_threshold: threshold, + recovery_timeout: Duration::from_millis(100), + half_open_successes_needed: 2, + }, + TestClassifier, + "test", + ) + } + + #[tokio::test] + async fn test_closed_allows_calls() { + let cb = make_breaker(3); + assert!(cb.check_allowed().await.is_ok()); // safety: test-only + assert_eq!(cb.circuit_state().await, CircuitState::Closed); // safety: test-only + } + + #[tokio::test] + async fn test_opens_after_threshold() { + let cb = make_breaker(3); + for _ in 0..3 { + cb.record_failure(&TestError::Transient).await; + } + assert_eq!(cb.circuit_state().await, CircuitState::Open); // safety: test-only + assert!(cb.check_allowed().await.is_err()); // safety: test-only + } + + #[tokio::test] + async fn test_permanent_errors_dont_trip() { + let cb = make_breaker(3); + for _ in 0..10 { + cb.record_failure(&TestError::Permanent).await; + } + assert_eq!(cb.circuit_state().await, CircuitState::Closed); // safety: test-only + } + + #[tokio::test] + async fn test_success_resets_count() { + let cb = make_breaker(3); + cb.record_failure(&TestError::Transient).await; + cb.record_failure(&TestError::Transient).await; + cb.record_success().await; + assert_eq!(cb.consecutive_failures().await, 0); // safety: test-only + // Should still be closed since we reset + cb.record_failure(&TestError::Transient).await; + cb.record_failure(&TestError::Transient).await; + assert_eq!(cb.circuit_state().await, CircuitState::Closed); // safety: test-only + } + + #[tokio::test] + async fn test_recovery_to_half_open() { + let cb = make_breaker(1); + cb.record_failure(&TestError::Transient).await; + assert_eq!(cb.circuit_state().await, CircuitState::Open); // safety: test-only + + // Wait for recovery timeout + tokio::time::sleep(Duration::from_millis(150)).await; + + // Should transition to HalfOpen + assert!(cb.check_allowed().await.is_ok()); // safety: test-only + assert_eq!(cb.circuit_state().await, CircuitState::HalfOpen); // safety: test-only + } + + #[tokio::test] + async fn test_half_open_closes_on_successes() { + let cb = make_breaker(1); + cb.record_failure(&TestError::Transient).await; + tokio::time::sleep(Duration::from_millis(150)).await; + let _ = cb.check_allowed().await; // transition to HalfOpen + + cb.record_success().await; + assert_eq!(cb.circuit_state().await, CircuitState::HalfOpen); // needs 2 // safety: test-only + cb.record_success().await; + assert_eq!(cb.circuit_state().await, CircuitState::Closed); // safety: test-only + } + + #[tokio::test] + async fn test_half_open_reopens_on_failure() { + let cb = make_breaker(1); + cb.record_failure(&TestError::Transient).await; + tokio::time::sleep(Duration::from_millis(150)).await; + let _ = cb.check_allowed().await; // HalfOpen + + cb.record_failure(&TestError::Transient).await; + assert_eq!(cb.circuit_state().await, CircuitState::Open); // safety: test-only + } +} diff --git a/src/resilience/classifier.rs b/src/resilience/classifier.rs new file mode 100644 index 00000000..548c489b --- /dev/null +++ b/src/resilience/classifier.rs @@ -0,0 +1,21 @@ +//! Generic error classification for resilience layers. + +use std::time::Duration; + +/// Classifies errors to determine how resilience layers should respond. +/// +/// Each client type (LLM, MCP, HTTP tool, etc.) implements this trait +/// to tell the resilience layers how to handle its specific error type. +pub trait ErrorClassifier { + /// Should the same request be retried against the same endpoint? + fn is_retryable(&self, err: &E) -> bool; + + /// Does this error indicate the backend is degraded? + /// Used by circuit breakers to track health. + fn is_transient(&self, err: &E) -> bool; + + /// Provider-suggested retry delay (e.g. from Retry-After header). + fn retry_after(&self, _err: &E) -> Option { + None + } +} diff --git a/src/resilience/health.rs b/src/resilience/health.rs new file mode 100644 index 00000000..af1822dd --- /dev/null +++ b/src/resilience/health.rs @@ -0,0 +1,165 @@ +//! Per-endpoint health tracking. +//! +//! Provides atomic, lock-free health counters for external service endpoints. +//! Used by the state bus to publish `EndpointHealthChanged` events. + +use std::collections::HashMap; +use std::sync::RwLock; +use std::sync::atomic::{AtomicU32, AtomicU64, Ordering}; + +/// Health state for a single endpoint. +pub struct EndpointHealth { + /// Number of consecutive failures. + pub consecutive_failures: AtomicU32, + /// Unix timestamp (seconds) of last successful call. + pub last_success: AtomicU64, + /// 0 = healthy, 1 = unhealthy. + pub unhealthy: AtomicU32, +} + +impl EndpointHealth { + pub fn new() -> Self { + Self { + consecutive_failures: AtomicU32::new(0), + last_success: AtomicU64::new(0), + unhealthy: AtomicU32::new(0), + } + } + + pub fn record_success(&self) { + self.consecutive_failures.store(0, Ordering::Relaxed); + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs(); + self.last_success.store(now, Ordering::Relaxed); + self.unhealthy.store(0, Ordering::Relaxed); + } + + /// Record a failure. Returns true if this failure triggered the unhealthy threshold. + pub fn record_failure(&self, threshold: u32) -> bool { + let prev = self.consecutive_failures.fetch_add(1, Ordering::Relaxed); + let new_count = prev + 1; + if new_count >= threshold && self.unhealthy.swap(1, Ordering::Relaxed) == 0 { + return true; // Just became unhealthy + } + false + } + + pub fn is_healthy(&self) -> bool { + self.unhealthy.load(Ordering::Relaxed) == 0 + } +} + +impl Default for EndpointHealth { + fn default() -> Self { + Self::new() + } +} + +/// Tracks health of multiple named endpoints. +pub struct HealthTracker { + endpoints: RwLock>, + failure_threshold: u32, +} + +impl HealthTracker { + pub fn new(failure_threshold: u32) -> Self { + Self { + endpoints: RwLock::new(HashMap::new()), + failure_threshold, + } + } + + pub fn record_success(&self, name: &str) { + let endpoints = self.endpoints.read().unwrap_or_else(|e| e.into_inner()); + if let Some(health) = endpoints.get(name) { + health.record_success(); + } else { + drop(endpoints); + let mut endpoints = self.endpoints.write().unwrap_or_else(|e| e.into_inner()); + endpoints + .entry(name.to_string()) + .or_default() + .record_success(); + } + } + + /// Record a failure. Returns true if this made the endpoint unhealthy. + pub fn record_failure(&self, name: &str) -> bool { + let endpoints = self.endpoints.read().unwrap_or_else(|e| e.into_inner()); + if let Some(health) = endpoints.get(name) { + health.record_failure(self.failure_threshold) + } else { + drop(endpoints); + let mut endpoints = self.endpoints.write().unwrap_or_else(|e| e.into_inner()); + let health = endpoints.entry(name.to_string()).or_default(); + health.record_failure(self.failure_threshold) + } + } + + pub fn is_healthy(&self, name: &str) -> bool { + let endpoints = self.endpoints.read().unwrap_or_else(|e| e.into_inner()); + endpoints.get(name).map(|h| h.is_healthy()).unwrap_or(true) // Unknown endpoints are assumed healthy + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_endpoint_health_starts_healthy() { + let h = EndpointHealth::new(); + assert!(h.is_healthy()); // safety: test-only + } + + #[test] + fn test_endpoint_health_becomes_unhealthy() { + let h = EndpointHealth::new(); + for i in 0..4 { + assert!( // safety: test-only + // safety: test-only + !h.record_failure(5), + "should not be unhealthy at failure {}", + i + 1 + ); + } + assert!(h.record_failure(5), "should become unhealthy at failure 5"); // safety: test-only + assert!(!h.is_healthy()); // safety: test-only + } + + #[test] + fn test_endpoint_health_recovers() { + let h = EndpointHealth::new(); + for _ in 0..5 { + h.record_failure(5); + } + assert!(!h.is_healthy()); // safety: test-only + h.record_success(); + assert!(h.is_healthy()); // safety: test-only + } + + #[test] + fn test_tracker_unknown_is_healthy() { + let t = HealthTracker::new(3); + assert!(t.is_healthy("unknown")); // safety: test-only + } + + #[test] + fn test_tracker_tracks_failures() { + let t = HealthTracker::new(2); + assert!(!t.record_failure("ep1")); // safety: test-only + assert!(t.record_failure("ep1")); // safety: test-only + assert!(!t.is_healthy("ep1")); // safety: test-only + } + + #[test] + fn test_tracker_recovery() { + let t = HealthTracker::new(2); + t.record_failure("ep1"); + t.record_failure("ep1"); + t.record_success("ep1"); + assert!(t.is_healthy("ep1")); // safety: test-only + } +} diff --git a/src/resilience/mod.rs b/src/resilience/mod.rs new file mode 100644 index 00000000..2e520734 --- /dev/null +++ b/src/resilience/mod.rs @@ -0,0 +1,9 @@ +pub mod circuit_breaker; +pub mod classifier; +pub mod health; +pub mod retry; + +pub use circuit_breaker::{CircuitBreakerConfig, CircuitBreakerLayer, CircuitState}; +pub use classifier::ErrorClassifier; +pub use health::{EndpointHealth, HealthTracker}; +pub use retry::{RetryConfig, RetryLayer}; diff --git a/src/resilience/retry.rs b/src/resilience/retry.rs new file mode 100644 index 00000000..43568828 --- /dev/null +++ b/src/resilience/retry.rs @@ -0,0 +1,193 @@ +//! Generic retry layer with exponential backoff and jitter. +//! +//! Extracted from `llm::retry` to be reusable across MCP, HTTP tools, +//! relay channels, and any async operation that can fail transiently. + +use std::future::Future; +use std::time::Duration; + +use rand::Rng; + +use super::classifier::ErrorClassifier; + +/// Configuration for the retry layer. +#[derive(Debug, Clone)] +pub struct RetryConfig { + /// Maximum number of retry attempts (not counting the initial attempt). + pub max_retries: u32, +} + +impl Default for RetryConfig { + fn default() -> Self { + Self { max_retries: 3 } + } +} + +/// Generic retry layer that wraps any async operation. +pub struct RetryLayer { + config: RetryConfig, + classifier: C, +} + +impl RetryLayer { + pub fn new(config: RetryConfig, classifier: C) -> Self { + Self { config, classifier } + } +} + +impl RetryLayer { + /// Execute an operation with retry logic. + /// + /// `label` is included in log messages for diagnostics. + pub async fn execute(&self, mut op: F, label: &str) -> Result + where + C: ErrorClassifier, + E: std::fmt::Display, + F: FnMut() -> Fut, + Fut: Future>, + { + let mut last_error: Option = None; + + for attempt in 0..=self.config.max_retries { + match op().await { + Ok(val) => return Ok(val), + Err(err) => { + if !self.classifier.is_retryable(&err) || attempt == self.config.max_retries { + return Err(err); + } + + let delay = self + .classifier + .retry_after(&err) + .unwrap_or_else(|| retry_backoff_delay(attempt)); + + tracing::warn!( + attempt = attempt + 1, + max_retries = self.config.max_retries, + delay_ms = delay.as_millis() as u64, + error = %err, + "Retrying after transient error ({label})" + ); + + last_error = Some(err); + tokio::time::sleep(delay).await; + } + } + } + + // Safety: loop runs at least once (0..=max_retries), so last_error is always Some + // if we reach here. But be defensive. + match last_error { + Some(e) => Err(e), + None => unreachable!("retry loop ran at least once"), + } + } +} + +/// Calculate exponential backoff delay with random jitter. +/// +/// Base delay is 1 second, doubled each attempt, with +/-25% jitter. +pub fn retry_backoff_delay(attempt: u32) -> Duration { + let base_ms: u64 = 1000u64.saturating_mul(2u64.saturating_pow(attempt)); + let jitter_range = base_ms / 4; // 25% + let jitter = if jitter_range > 0 { + let offset = rand::thread_rng().gen_range(0..=jitter_range * 2); + offset as i64 - jitter_range as i64 + } else { + 0 + }; + let delay_ms = (base_ms as i64 + jitter).max(100) as u64; + Duration::from_millis(delay_ms) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::Arc; + use std::sync::atomic::{AtomicU32, Ordering}; + + #[derive(Debug, thiserror::Error)] + enum TestError { + #[error("transient")] + Transient, + #[error("permanent")] + Permanent, + } + + struct TestClassifier; + impl ErrorClassifier for TestClassifier { + fn is_retryable(&self, err: &TestError) -> bool { + matches!(err, TestError::Transient) + } + fn is_transient(&self, err: &TestError) -> bool { + matches!(err, TestError::Transient) + } + } + + #[test] + fn test_backoff_delay_exponential() { + for _ in 0..10 { + let d0 = retry_backoff_delay(0); + assert!(d0.as_millis() >= 750 && d0.as_millis() <= 1250); // safety: test-only + let d1 = retry_backoff_delay(1); + assert!(d1.as_millis() >= 1500 && d1.as_millis() <= 2500); // safety: test-only + } + } + + #[test] + fn test_backoff_delay_no_overflow() { + let delay = retry_backoff_delay(30); + assert!(delay.as_millis() >= 100); // safety: test-only + } + + #[tokio::test] + async fn test_success_first_attempt() { + let layer = RetryLayer::new(RetryConfig { max_retries: 3 }, TestClassifier); + let result: Result<&str, TestError> = layer.execute(|| async { Ok("ok") }, "test").await; // safety: test-only retry call + assert_eq!(result.unwrap(), "ok"); // safety: test-only + } + + #[tokio::test] + async fn test_permanent_error_no_retry() { + let calls = Arc::new(AtomicU32::new(0)); + let calls_c = calls.clone(); + let layer = RetryLayer::new(RetryConfig { max_retries: 3 }, TestClassifier); + let result: Result<(), TestError> = layer + .execute( + // safety: test-only retry call + || { + let c = calls_c.clone(); + async move { + c.fetch_add(1, Ordering::Relaxed); + Err(TestError::Permanent) + } + }, + "test", + ) + .await; + assert!(result.is_err()); // safety: test-only + assert_eq!(calls.load(Ordering::Relaxed), 1); // safety: test-only + } + + #[tokio::test] + async fn test_exhausts_retries() { + let calls = Arc::new(AtomicU32::new(0)); + let calls_c = calls.clone(); + let layer = RetryLayer::new(RetryConfig { max_retries: 0 }, TestClassifier); + let result: Result<(), TestError> = layer + .execute( + // safety: test-only retry call + || { + let c = calls_c.clone(); + async move { + c.fetch_add(1, Ordering::Relaxed); + Err(TestError::Transient) + } + }, + "test", + ) + .await; + assert!(result.is_err()); // safety: test-only + assert_eq!(calls.load(Ordering::Relaxed), 1); // safety: test-only + } +} diff --git a/src/state_bus.rs b/src/state_bus.rs new file mode 100644 index 00000000..5f384759 --- /dev/null +++ b/src/state_bus.rs @@ -0,0 +1,130 @@ +//! State invalidation bus for cross-module state synchronization. +//! +//! When state changes in one module (e.g., web UI toggles a routine, secret +//! rotates, config reloads), the bus notifies other modules that cache that +//! state so they can refresh. +//! +//! Modules subscribe to events they care about and ignore the rest. No module +//! needs to import another module to propagate state changes — the bus is the +//! **only** coupling point. + +use std::sync::Arc; + +use tokio::sync::broadcast; +use uuid::Uuid; + +/// A state change notification. +#[derive(Debug, Clone)] +pub enum StateChange { + /// A routine was created, updated, toggled, or deleted. + RoutineUpdated { routine_id: Uuid }, + /// A secret was rotated or deleted. + SecretRotated { key_name: String }, + /// Global configuration was reloaded (e.g. via SIGHUP). + ConfigReloaded, + /// An external endpoint's health status changed. + EndpointHealthChanged { name: String, healthy: bool }, + /// The tool registry was modified (tool added/removed/rebuilt). + ToolRegistryChanged, + /// An extension was installed or removed. + ExtensionInstalled { extension_id: String }, +} + +/// Broadcast bus for state change notifications. +/// +/// Backed by a tokio `broadcast` channel with a fixed buffer. Slow consumers +/// that fall behind will miss events (acceptable — they can re-poll state). +#[derive(Clone)] +pub struct StateBus { + tx: broadcast::Sender, +} + +impl StateBus { + /// Create a new state bus with a buffer of 64 events. + pub fn new() -> Self { + let (tx, _) = broadcast::channel(64); + Self { tx } + } + + /// Publish a state change. Non-blocking; drops the event if no subscribers. + pub fn publish(&self, event: StateChange) { + // Ignore send error (no active receivers). + let _ = self.tx.send(event); + } + + /// Subscribe to state change notifications. + pub fn subscribe(&self) -> broadcast::Receiver { + self.tx.subscribe() + } +} + +impl Default for StateBus { + fn default() -> Self { + Self::new() + } +} + +/// Convenience constructor for passing through `Arc`. +pub fn new_state_bus() -> Arc { + Arc::new(StateBus::new()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn test_publish_subscribe() { + let bus = StateBus::new(); + let mut rx = bus.subscribe(); + + let id = Uuid::new_v4(); + bus.publish(StateChange::RoutineUpdated { routine_id: id }); + + let event = rx.recv().await.unwrap(); // safety: test-only + assert!(matches!(event, StateChange::RoutineUpdated { routine_id } if routine_id == id)); // safety: test-only + } + + #[tokio::test] + async fn test_no_subscriber_does_not_panic() { + let bus = StateBus::new(); + // No subscribers — should not panic. + bus.publish(StateChange::ConfigReloaded); + } + + #[tokio::test] + async fn test_multiple_subscribers() { + let bus = StateBus::new(); + let mut rx1 = bus.subscribe(); + let mut rx2 = bus.subscribe(); + + bus.publish(StateChange::ToolRegistryChanged); + + let e1 = rx1.recv().await.unwrap(); // safety: test-only + let e2 = rx2.recv().await.unwrap(); // safety: test-only + assert!(matches!(e1, StateChange::ToolRegistryChanged)); // safety: test-only + assert!(matches!(e2, StateChange::ToolRegistryChanged)); // safety: test-only + } + + #[tokio::test] + async fn test_slow_consumer_lags() { + let bus = StateBus::new(); + let mut rx = bus.subscribe(); + + // Overflow the 64-event buffer. + for i in 0..100 { + bus.publish(StateChange::EndpointHealthChanged { + name: format!("ep-{}", i), + healthy: true, + }); + } + + // First recv should report a lag. + let result = rx.recv().await; + assert!( // safety: test-only + // safety: test-only + result.is_ok() || result.is_err(), + "lagged receiver should either get an event or a Lagged error" + ); + } +} diff --git a/src/util.rs b/src/util.rs index 866f623c..816a359e 100644 --- a/src/util.rs +++ b/src/util.rs @@ -70,109 +70,175 @@ pub fn llm_signals_completion(response: &str) -> bool { positive_phrases.iter().any(|p| lower.contains(p)) } +/// Truncate a string to at most `max_bytes` bytes at a char boundary, appending "...". +/// +/// If the input is wrapped in `…` and truncation +/// removes the closing tag, the tag is re-appended so downstream XML parsers +/// never see an unclosed element. +pub fn truncate_preview(s: &str, max_bytes: usize) -> String { + if s.len() <= max_bytes { + return s.to_string(); + } + let mut end = max_bytes; + while end > 0 && !s.is_char_boundary(end) { + end -= 1; + } + let mut result = format!("{}...", &s[..end]); // safety: end is a valid char boundary per loop above + + if s.starts_with("") { + result.push_str("\n"); + } + + result +} + #[cfg(test)] mod tests { - use crate::util::{floor_char_boundary, llm_signals_completion}; + use crate::util::{floor_char_boundary, llm_signals_completion, truncate_preview}; // ── floor_char_boundary ── #[test] fn floor_char_boundary_at_valid_boundary() { - assert_eq!(floor_char_boundary("hello", 3), 3); + assert_eq!(floor_char_boundary("hello", 3), 3); // safety: test-only } #[test] fn floor_char_boundary_mid_multibyte_char() { // h = 1 byte, é = 2 bytes, total 3 bytes let s = "hé"; - assert_eq!(floor_char_boundary(s, 2), 1); // byte 2 is mid-é, back up to 1 + assert_eq!(floor_char_boundary(s, 2), 1); // byte 2 is mid-é, back up to 1 // safety: test-only } #[test] fn floor_char_boundary_past_end() { - assert_eq!(floor_char_boundary("hi", 100), 2); + assert_eq!(floor_char_boundary("hi", 100), 2); // safety: test-only } #[test] fn floor_char_boundary_at_zero() { - assert_eq!(floor_char_boundary("hello", 0), 0); + assert_eq!(floor_char_boundary("hello", 0), 0); // safety: test-only } #[test] fn floor_char_boundary_empty_string() { - assert_eq!(floor_char_boundary("", 5), 0); + assert_eq!(floor_char_boundary("", 5), 0); // safety: test-only } // ── llm_signals_completion ── #[test] fn signals_completion_positive() { - assert!(llm_signals_completion("The job is complete.")); - assert!(llm_signals_completion("I have completed the task.")); - assert!(llm_signals_completion("All done, here are the results.")); - assert!(llm_signals_completion("Task is finished successfully.")); + assert!(llm_signals_completion("The job is complete.")); // safety: test-only + assert!(llm_signals_completion("I have completed the task.")); // safety: test-only + assert!(llm_signals_completion("All done, here are the results.")); // safety: test-only + assert!(llm_signals_completion("Task is finished successfully.")); // safety: test-only assert!(llm_signals_completion( + // safety: test-only "I have completed the task successfully." )); assert!(llm_signals_completion( + // safety: test-only "All steps are complete and verified." )); assert!(llm_signals_completion( + // safety: test-only "I've done all the work. The work is done." )); assert!(llm_signals_completion( + // safety: test-only "Successfully completed the migration." )); assert!(llm_signals_completion( + // safety: test-only "I have completed the job ahead of schedule." )); - assert!(llm_signals_completion("I have finished the task.")); - assert!(llm_signals_completion("All steps are done now.")); - assert!(llm_signals_completion("I've completed everything.")); - assert!(llm_signals_completion("All tasks complete.")); + assert!(llm_signals_completion("I have finished the task.")); // safety: test-only + assert!(llm_signals_completion("All steps are done now.")); // safety: test-only + assert!(llm_signals_completion("I've completed everything.")); // safety: test-only + assert!(llm_signals_completion("All tasks complete.")); // safety: test-only } #[test] fn signals_completion_negative() { - assert!(!llm_signals_completion("The task is not complete yet.")); - assert!(!llm_signals_completion("This is not done.")); - assert!(!llm_signals_completion("The work is incomplete.")); - assert!(!llm_signals_completion("Build is unfinished.")); + assert!(!llm_signals_completion("The task is not complete yet.")); // safety: test-only + assert!(!llm_signals_completion("This is not done.")); // safety: test-only + assert!(!llm_signals_completion("The work is incomplete.")); // safety: test-only + assert!(!llm_signals_completion("Build is unfinished.")); // safety: test-only assert!(!llm_signals_completion( + // safety: test-only "The migration is not yet finished." )); - assert!(!llm_signals_completion("The job isn't done yet.")); - assert!(!llm_signals_completion("This remains unfinished.")); + assert!(!llm_signals_completion("The job isn't done yet.")); // safety: test-only + assert!(!llm_signals_completion("This remains unfinished.")); // safety: test-only } #[test] fn signals_completion_no_bare_substrings() { - assert!(!llm_signals_completion("The download completed.")); + assert!(!llm_signals_completion("The download completed.")); // safety: test-only assert!(!llm_signals_completion( + // safety: test-only "Function done_callback was called." )); - assert!(!llm_signals_completion("Set is_complete = true")); - assert!(!llm_signals_completion("Running step 3 of 5")); + assert!(!llm_signals_completion("Set is_complete = true")); // safety: test-only + assert!(!llm_signals_completion("Running step 3 of 5")); // safety: test-only assert!(!llm_signals_completion( + // safety: test-only "I need to complete more work first." )); assert!(!llm_signals_completion( + // safety: test-only "Let me finish the remaining steps." )); assert!(!llm_signals_completion( + // safety: test-only "I'm done analyzing, now let me fix it." )); assert!(!llm_signals_completion( + // safety: test-only "I completed step 1 but step 2 remains." )); } #[test] fn signals_completion_tool_output_injection() { - assert!(!llm_signals_completion("TASK_COMPLETE")); - assert!(!llm_signals_completion("JOB_DONE")); + assert!(!llm_signals_completion("TASK_COMPLETE")); // safety: test-only + assert!(!llm_signals_completion("JOB_DONE")); // safety: test-only assert!(!llm_signals_completion( + // safety: test-only "The tool returned: TASK_COMPLETE signal" )); } + + // ── truncate_preview ── + + #[test] + fn truncate_preview_short_string() { + assert_eq!(truncate_preview("hello", 10), "hello"); // safety: test-only + } + + #[test] + fn truncate_preview_exact_boundary() { + assert_eq!(truncate_preview("hello", 5), "hello"); // safety: test-only + } + + #[test] + fn truncate_preview_truncates_ascii() { + assert_eq!(truncate_preview("hello world", 5), "hello..."); // safety: test-only + } + + #[test] + fn truncate_preview_multibyte_char_boundary() { + let s = "a€b"; + let result = truncate_preview(s, 3); + assert_eq!(result, "a..."); // safety: test-only + } + + #[test] + fn truncate_preview_closes_tool_output_tag() { + let s = "\nSome very long content here\n"; + let result = truncate_preview(s, 60); + assert!(result.ends_with("")); // safety: test-only + assert!(result.contains("...")); // safety: test-only + } } diff --git a/tests/boundary_chaos.rs b/tests/boundary_chaos.rs new file mode 100644 index 00000000..fd202397 --- /dev/null +++ b/tests/boundary_chaos.rs @@ -0,0 +1,375 @@ +//! Boundary chaos tests — exercise failure modes at module seams. +//! +//! These tests verify that the architectural hardening (domain event decoupling, +//! generic resilience layers, state bus) works correctly under failure conditions. +//! +//! Organized by boundary, not by module: +//! - Resilience layers (retry, circuit breaker, health tracker) +//! - State bus propagation +//! - Domain event type compatibility + +use std::sync::Arc; +use std::sync::atomic::{AtomicU32, Ordering}; +use std::time::Duration; + +use ironclaw::events::DomainEvent; +use ironclaw::resilience::circuit_breaker::{ + CircuitBreakerConfig, CircuitBreakerLayer, CircuitState, +}; +use ironclaw::resilience::classifier::ErrorClassifier; +use ironclaw::resilience::health::HealthTracker; +use ironclaw::resilience::retry::{RetryConfig, RetryLayer}; +use ironclaw::state_bus::{StateBus, StateChange}; + +// ── Test error type ────────────────────────────────────────────────── + +#[derive(Debug, thiserror::Error)] +enum TestError { + #[error("transient failure")] + Transient, + #[error("permanent failure")] + Permanent, +} + +struct TestClassifier; + +impl ErrorClassifier for TestClassifier { + fn is_retryable(&self, err: &TestError) -> bool { + matches!(err, TestError::Transient) + } + fn is_transient(&self, err: &TestError) -> bool { + matches!(err, TestError::Transient) + } +} + +// ── Resilience: Retry layer ────────────────────────────────────────── + +#[tokio::test] +async fn retry_layer_recovers_after_transient_failures() { + let call_count = Arc::new(AtomicU32::new(0)); + let cc = call_count.clone(); + + let layer = RetryLayer::new(RetryConfig { max_retries: 3 }, TestClassifier); + let result: Result<&str, TestError> = layer + .execute( + // safety: test-only + || { + let c = cc.clone(); + async move { + let n = c.fetch_add(1, Ordering::Relaxed); + if n < 2 { + Err(TestError::Transient) + } else { + Ok("recovered") + } + } + }, + "test", + ) + .await; + + assert_eq!(result.unwrap(), "recovered"); // safety: test-only + assert_eq!(call_count.load(Ordering::Relaxed), 3); // 2 failures + 1 success // safety: test-only +} + +#[tokio::test] +async fn retry_layer_stops_on_permanent_error() { + let call_count = Arc::new(AtomicU32::new(0)); + let cc = call_count.clone(); + + let layer = RetryLayer::new(RetryConfig { max_retries: 5 }, TestClassifier); + let result: Result<(), TestError> = layer + .execute( + // safety: test-only + || { + let c = cc.clone(); + async move { + c.fetch_add(1, Ordering::Relaxed); + Err(TestError::Permanent) + } + }, + "test", + ) + .await; + + assert!(result.is_err()); // safety: test-only + assert_eq!(call_count.load(Ordering::Relaxed), 1); // No retries for permanent // safety: test-only +} + +// ── Resilience: Circuit breaker ────────────────────────────────────── + +#[tokio::test] +async fn circuit_breaker_opens_after_threshold() { + let cb = CircuitBreakerLayer::new( + CircuitBreakerConfig { + failure_threshold: 3, + recovery_timeout: Duration::from_millis(100), + half_open_successes_needed: 1, + }, + TestClassifier, + "test-endpoint", + ); + + // Record failures up to threshold + for _ in 0..3 { + cb.record_failure(&TestError::Transient).await; + } + + assert_eq!(cb.circuit_state().await, CircuitState::Open); // safety: test-only + assert!(cb.check_allowed().await.is_err()); // safety: test-only +} + +#[tokio::test] +async fn circuit_breaker_recovers_via_half_open() { + let cb = CircuitBreakerLayer::new( + CircuitBreakerConfig { + failure_threshold: 2, + recovery_timeout: Duration::from_millis(50), + half_open_successes_needed: 1, + }, + TestClassifier, + "test-recovery", + ); + + // Trip the circuit + cb.record_failure(&TestError::Transient).await; + cb.record_failure(&TestError::Transient).await; + assert_eq!(cb.circuit_state().await, CircuitState::Open); // safety: test-only + + // Wait for recovery timeout + tokio::time::sleep(Duration::from_millis(100)).await; + + // Should transition to HalfOpen + assert!(cb.check_allowed().await.is_ok()); // safety: test-only + assert_eq!(cb.circuit_state().await, CircuitState::HalfOpen); // safety: test-only + + // Success should close the circuit + cb.record_success().await; + assert_eq!(cb.circuit_state().await, CircuitState::Closed); // safety: test-only +} + +#[tokio::test] +async fn circuit_breaker_ignores_permanent_errors() { + let cb = CircuitBreakerLayer::new( + CircuitBreakerConfig { + failure_threshold: 2, + recovery_timeout: Duration::from_secs(30), + half_open_successes_needed: 1, + }, + TestClassifier, + "test-perm", + ); + + // Permanent errors should never trip the breaker + for _ in 0..100 { + cb.record_failure(&TestError::Permanent).await; + } + assert_eq!(cb.circuit_state().await, CircuitState::Closed); // safety: test-only +} + +// ── Resilience: Health tracker ─────────────────────────────────────── + +#[test] +fn health_tracker_marks_unhealthy_after_threshold() { + let tracker = HealthTracker::new(3); + assert!(tracker.is_healthy("mcp-server-1")); // safety: test-only + + tracker.record_failure("mcp-server-1"); + tracker.record_failure("mcp-server-1"); + assert!(tracker.is_healthy("mcp-server-1")); // Not yet // safety: test-only + + tracker.record_failure("mcp-server-1"); + assert!(!tracker.is_healthy("mcp-server-1")); // Now unhealthy // safety: test-only +} + +#[test] +fn health_tracker_recovers_on_success() { + let tracker = HealthTracker::new(2); + tracker.record_failure("ep1"); + tracker.record_failure("ep1"); + assert!(!tracker.is_healthy("ep1")); // safety: test-only + + tracker.record_success("ep1"); + assert!(tracker.is_healthy("ep1")); // safety: test-only +} + +#[test] +fn health_tracker_isolates_endpoints() { + let tracker = HealthTracker::new(2); + + // Fail ep1 + tracker.record_failure("ep1"); + tracker.record_failure("ep1"); + assert!(!tracker.is_healthy("ep1")); // safety: test-only + + // ep2 should be unaffected + assert!(tracker.is_healthy("ep2")); // safety: test-only +} + +// ── State bus ──────────────────────────────────────────────────────── + +#[tokio::test] +async fn state_bus_delivers_to_all_subscribers() { + let bus = StateBus::new(); + let mut rx1 = bus.subscribe(); + let mut rx2 = bus.subscribe(); + + let id = uuid::Uuid::new_v4(); + bus.publish(StateChange::RoutineUpdated { routine_id: id }); + + let e1 = rx1.recv().await.unwrap(); // safety: test-only + let e2 = rx2.recv().await.unwrap(); // safety: test-only + assert!(matches!(e1, StateChange::RoutineUpdated { routine_id } if routine_id == id)); // safety: test-only + assert!(matches!(e2, StateChange::RoutineUpdated { routine_id } if routine_id == id)); // safety: test-only +} + +#[tokio::test] +async fn state_bus_no_subscriber_is_harmless() { + let bus = StateBus::new(); + // Publishing with no subscribers should not panic + bus.publish(StateChange::ConfigReloaded); + bus.publish(StateChange::ToolRegistryChanged); + bus.publish(StateChange::SecretRotated { + key_name: "api_key".to_string(), + }); +} + +#[tokio::test] +async fn state_bus_subscriber_receives_only_after_subscribe() { + let bus = StateBus::new(); + + // Publish before subscribing + bus.publish(StateChange::ConfigReloaded); + + // Subscribe after + let mut rx = bus.subscribe(); + + // Publish after subscribing + bus.publish(StateChange::ToolRegistryChanged); + + let event = rx.recv().await.unwrap(); // safety: test-only + assert!(matches!(event, StateChange::ToolRegistryChanged)); // safety: test-only +} + +// ── Domain event compatibility ─────────────────────────────────────── + +#[test] +fn domain_event_serializes_as_sse_wire_format() { + let event = DomainEvent::Response { + content: "Hello!".to_string(), + thread_id: "t1".to_string(), + }; + let json = serde_json::to_string(&event).unwrap(); // safety: test-only + let parsed: serde_json::Value = serde_json::from_str(&json).unwrap(); // safety: test-only + + assert_eq!(parsed["type"], "response"); // safety: test-only + assert_eq!(parsed["content"], "Hello!"); // safety: test-only + assert_eq!(parsed["thread_id"], "t1"); // safety: test-only +} + +#[test] +fn domain_event_all_variants_serialize() { + // Verify all variants can be serialized without panicking + let variants: Vec = vec![ + DomainEvent::Response { + content: "ok".into(), + thread_id: "t".into(), + }, + DomainEvent::Thinking { + message: "...".into(), + thread_id: None, + }, + DomainEvent::ToolStarted { + name: "shell".into(), + thread_id: None, + }, + DomainEvent::ToolCompleted { + name: "shell".into(), + success: true, + error: None, + parameters: None, + thread_id: None, + }, + DomainEvent::Heartbeat, + DomainEvent::JobMessage { + job_id: "j1".into(), + role: "assistant".into(), + content: "msg".into(), + }, + DomainEvent::JobResult { + job_id: "j1".into(), + status: "completed".into(), + session_id: None, + }, + DomainEvent::Suggestions { + suggestions: vec!["a".into(), "b".into()], + thread_id: Some("t1".into()), + }, + ]; + + for variant in &variants { + let json = serde_json::to_string(variant).unwrap(); // safety: test-only + let parsed: serde_json::Value = serde_json::from_str(&json).unwrap(); // safety: test-only + assert!( // safety: test-only + // safety: test-only + parsed.get("type").is_some(), + "missing 'type' field in {:?}", + variant + ); + } +} + +#[test] +fn domain_event_broadcast_channel_works() { + // Verify DomainEvent can be used with tokio broadcast (Clone required) + let (tx, mut rx) = tokio::sync::broadcast::channel::(16); + tx.send(DomainEvent::Heartbeat).unwrap(); // safety: test-only + let received = rx.try_recv().unwrap(); // safety: test-only + assert!(matches!(received, DomainEvent::Heartbeat)); // safety: test-only +} + +// ── Cross-boundary: Retry + Circuit Breaker composition ────────────── + +#[tokio::test] +async fn retry_and_circuit_breaker_compose() { + let cb = Arc::new(CircuitBreakerLayer::new( + CircuitBreakerConfig { + failure_threshold: 5, + recovery_timeout: Duration::from_secs(30), + half_open_successes_needed: 1, + }, + TestClassifier, + "composed", + )); + let retry = RetryLayer::new(RetryConfig { max_retries: 2 }, TestClassifier); + + let call_count = Arc::new(AtomicU32::new(0)); + let cc = call_count.clone(); + let cb_clone = cb.clone(); + + // Simulate an operation that fails then succeeds, tracked by circuit breaker + let result: Result<&str, TestError> = retry + .execute( + // safety: test-only + || { + let c = cc.clone(); + let cb = cb_clone.clone(); + async move { + let n = c.fetch_add(1, Ordering::Relaxed); + if n == 0 { + cb.record_failure(&TestError::Transient).await; + Err(TestError::Transient) + } else { + cb.record_success().await; + Ok("ok") + } + } + }, + "composed", + ) + .await; + + assert_eq!(result.unwrap(), "ok"); // safety: test-only + assert_eq!(cb.circuit_state().await, CircuitState::Closed); // safety: test-only + assert_eq!(cb.consecutive_failures().await, 0); // safety: test-only +}