gitops/optimclaw
1
0
Fork 0
mirror of https://github.com/outbackdingo/optimclaw.git synced 2026-08-25 14:53:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: decouple modules, add resilience middleware and state bus [skip-regression-check]

Browse Source 
Break circular dependencies between agent, db, channels, and context
modules by extracting shared domain types to neutral locations:

- Extract routine types to src/models/routine.rs
- Extract ToolFailureRecord to src/models/tool_failure.rs
- Move SseEvent to src/events.rs as DomainEvent
- Move HttpInterceptor to src/observability/
- Move truncate_preview to src/util.rs

Add generic resilience middleware (src/resilience/):
- ErrorClassifier, RetryLayer, CircuitBreakerLayer, HealthTracker

Add state invalidation bus (src/state_bus.rs)
Add boundary chaos tests (tests/boundary_chaos.rs)

Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>
This commit is contained in:
Illia Polosukhin
2026-03-14 21:17:25 -07:00
co-authored by Claude Opus 4.6
parent 757d24bd90
commit b04d14b114
27 changed files with 2398 additions and 1061 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/agent/agent_loop.rs
+1 -1
View File
@@ -74,7 +74,7 @@ pub struct AgentDeps {
/// Cost enforcement guardrails (daily budget, hourly rate limits).
pub cost_guard: Arc<crate::agent::cost_guard::CostGuard>,
/// SSE broadcast sender for live job event streaming to the web gateway.
pub sse_tx: Option<tokio::sync::broadcast::Sender<crate::channels::web::types::SseEvent>>,
pub sse_tx: Option<tokio::sync::broadcast::Sender<crate::events::DomainEvent>>,
/// HTTP interceptor for trace recording/replay.
pub http_interceptor: Option<Arc<dyn crate::llm::recording::HttpInterceptor>>,
/// Audio transcription middleware for voice messages.
src/agent/job_monitor.rs
+1 -1
View File
@@ -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.
src/agent/routine.rs
+5 -809
View File
@@ -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<DateTime<Utc>>,
pub next_fire_at: Option<DateTime<Utc>>,
pub run_count: u64,
pub consecutive_failures: u32,
pub state: serde_json::Value,
pub created_at: DateTime<Utc>,
pub updated_at: DateTime<Utc>,
}
/// 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<String>,
},
/// Fire when a channel message matches a pattern.
Event {
/// Optional channel filter (e.g. "telegram", "slack").
channel: Option<String>,
/// 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<String, String>,
},
/// 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<Self, RoutineError> {
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<String>,
/// 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<String>,
},
}
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<String>`.
pub fn parse_tool_permissions(value: &serde_json::Value) -> Vec<String> {
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<Self, RoutineError> {
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<Duration>,
}
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<String>,
/// 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<Self, Self::Err> {
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<String>,
pub started_at: DateTime<Utc>,
pub completed_at: Option<DateTime<Utc>>,
pub status: RunStatus,
pub result_summary: Option<String>,
pub tokens_used: Option<i32>,
pub job_id: Option<Uuid>,
pub created_at: DateTime<Utc>,
}
/// 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<String> {
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<Option<DateTime<Utc>>, 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::*;
src/agent/scheduler.rs
+1 -1
View File
@@ -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;
src/agent/self_repair.rs
+5 -11
View File
@@ -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<String>,
pub first_failure: DateTime<Utc>,
pub last_failure: DateTime<Utc>,
pub last_build_result: Option<serde_json::Value>,
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)]
src/agent/session.rs
+1 -1
View File
@@ -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)]
src/agent/thread_ops.rs
+1 -1
View File
@@ -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.";
src/channels/web/types.rs
+3 -143
View File
@@ -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<String>,
},
#[serde(rename = "tool_started")]
ToolStarted {
name: String,
#[serde(skip_serializing_if = "Option::is_none")]
thread_id: Option<String>,
},
#[serde(rename = "tool_completed")]
ToolCompleted {
name: String,
success: bool,
#[serde(skip_serializing_if = "Option::is_none")]
error: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
parameters: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
thread_id: Option<String>,
},
#[serde(rename = "tool_result")]
ToolResult {
name: String,
preview: String,
#[serde(skip_serializing_if = "Option::is_none")]
thread_id: Option<String>,
},
#[serde(rename = "stream_chunk")]
StreamChunk {
content: String,
#[serde(skip_serializing_if = "Option::is_none")]
thread_id: Option<String>,
},
#[serde(rename = "status")]
Status {
message: String,
#[serde(skip_serializing_if = "Option::is_none")]
thread_id: Option<String>,
},
#[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<String>,
},
#[serde(rename = "auth_required")]
AuthRequired {
extension_name: String,
#[serde(skip_serializing_if = "Option::is_none")]
instructions: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
auth_url: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
setup_url: Option<String>,
},
#[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<String>,
},
#[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<String>,
},
/// An image was generated by a tool.
#[serde(rename = "image_generated")]
ImageGenerated {
data_url: String,
#[serde(skip_serializing_if = "Option::is_none")]
path: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
thread_id: Option<String>,
},
/// Suggested follow-up messages for the user.
#[serde(rename = "suggestions")]
Suggestions {
suggestions: Vec<String>,
#[serde(skip_serializing_if = "Option::is_none")]
thread_id: Option<String>,
},
/// 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<String>,
},
}
/// 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 ---
src/channels/web/util.rs
+3 -21
View File
@@ -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 `<tool_output …>…</tool_output>` 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 <tool_output> if truncation cut through the closing tag.
if s.starts_with("<tool_output") && !result.ends_with("</tool_output>") {
result.push_str("\n</tool_output>");
}
result
crate::util::truncate_preview(s, max_bytes)
}
/// Build TurnInfo pairs from flat DB messages (user/tool_calls/assistant triples).
src/context/state.rs
+1 -1
View File
@@ -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)]
src/db/mod.rs
+6 -3
View File
@@ -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<Vec<BrokenTool>, DatabaseError>;
async fn get_broken_tools(
&self,
threshold: i32,
) -> Result<Vec<ToolFailureRecord>, DatabaseError>;
async fn mark_tool_repaired(&self, tool_name: &str) -> Result<(), DatabaseError>;
async fn increment_repair_attempts(&self, tool_name: &str) -> Result<(), DatabaseError>;
}
src/events.rs
+159
View File
@@ -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<String>,
},
#[serde(rename = "tool_started")]
ToolStarted {
name: String,
#[serde(skip_serializing_if = "Option::is_none")]
thread_id: Option<String>,
},
#[serde(rename = "tool_completed")]
ToolCompleted {
name: String,
success: bool,
#[serde(skip_serializing_if = "Option::is_none")]
error: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
parameters: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
thread_id: Option<String>,
},
#[serde(rename = "tool_result")]
ToolResult {
name: String,
preview: String,
#[serde(skip_serializing_if = "Option::is_none")]
thread_id: Option<String>,
},
#[serde(rename = "stream_chunk")]
StreamChunk {
content: String,
#[serde(skip_serializing_if = "Option::is_none")]
thread_id: Option<String>,
},
#[serde(rename = "status")]
Status {
message: String,
#[serde(skip_serializing_if = "Option::is_none")]
thread_id: Option<String>,
},
#[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<String>,
},
#[serde(rename = "auth_required")]
AuthRequired {
extension_name: String,
#[serde(skip_serializing_if = "Option::is_none")]
instructions: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
auth_url: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
setup_url: Option<String>,
},
#[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<String>,
},
#[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<String>,
},
/// An image was generated by a tool.
#[serde(rename = "image_generated")]
ImageGenerated {
data_url: String,
#[serde(skip_serializing_if = "Option::is_none")]
path: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
thread_id: Option<String>,
},
/// Suggested follow-up messages for the user.
#[serde(rename = "suggestions")]
Suggestions {
suggestions: Vec<String>,
#[serde(skip_serializing_if = "Option::is_none")]
thread_id: Option<String>,
},
/// 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<String>,
},
}
src/lib.rs
+4
View File
@@ -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;
src/llm/recording.rs
+5 -43
View File
@@ -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<String>,
}
/// 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<HttpExchangeResponse>;
/// 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)]
src/models/mod.rs
+13
View File
@@ -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;
src/models/routine.rs
+823
View File
@@ -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<DateTime<Utc>>,
pub next_fire_at: Option<DateTime<Utc>>,
pub run_count: u64,
pub consecutive_failures: u32,
pub state: serde_json::Value,
pub created_at: DateTime<Utc>,
pub updated_at: DateTime<Utc>,
}
/// 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<String>,
},
/// Fire when a channel message matches a pattern.
Event {
/// Optional channel filter (e.g. "telegram", "slack").
channel: Option<String>,
/// 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<String, String>,
},
/// 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<Self, RoutineError> {
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<String>,
/// 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<String>,
},
}
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<String>`.
pub fn parse_tool_permissions(value: &serde_json::Value) -> Vec<String> {
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<Self, RoutineError> {
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<Duration>,
}
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<String>,
/// 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<Self, Self::Err> {
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<String>,
pub started_at: DateTime<Utc>,
pub completed_at: Option<DateTime<Utc>>,
pub status: RunStatus,
pub result_summary: Option<String>,
pub tokens_used: Option<i32>,
pub job_id: Option<Uuid>,
pub created_at: DateTime<Utc>,
}
/// 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<String> {
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<Option<DateTime<Utc>>, 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"),
}
}
}
src/models/tool_failure.rs
+19
View File
@@ -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<String>,
pub first_failure: DateTime<Utc>,
pub last_failure: DateTime<Utc>,
pub last_build_result: Option<serde_json::Value>,
pub repair_attempts: u32,
}
src/observability/http_interceptor.rs
+50
View File
@@ -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<String>,
}
/// 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<HttpExchangeResponse>;
/// Called after a real HTTP request completes (recording mode only).
async fn after_response(&self, request: &HttpExchangeRequest, response: &HttpExchangeResponse);
}
src/observability/mod.rs
+4
View File
@@ -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;
src/resilience/circuit_breaker.rs
+309
View File
@@ -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<Instant>,
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<C> {
state: Mutex<BreakerState>,
config: CircuitBreakerConfig,
classifier: C,
/// Label for log messages.
label: String,
}
impl<C> CircuitBreakerLayer<C> {
pub fn new(config: CircuitBreakerConfig, classifier: C, label: impl Into<String>) -> 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<C> CircuitBreakerLayer<C> {
/// 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<E>(&self, err: &E)
where
C: ErrorClassifier<E>,
{
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<TestError> 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<TestClassifier> {
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
}
}
src/resilience/classifier.rs
+21
View File
@@ -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<E> {
/// 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<Duration> {
None
}
}
src/resilience/health.rs
+165
View File
@@ -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<HashMap<String, EndpointHealth>>,
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
}
}
src/resilience/mod.rs
+9
View File
@@ -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};
src/resilience/retry.rs
+193
View File
@@ -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<C> {
config: RetryConfig,
classifier: C,
}
impl<C> RetryLayer<C> {
pub fn new(config: RetryConfig, classifier: C) -> Self {
Self { config, classifier }
}
}
impl<C> RetryLayer<C> {
/// Execute an operation with retry logic.
///
/// `label` is included in log messages for diagnostics.
pub async fn execute<T, E, F, Fut>(&self, mut op: F, label: &str) -> Result<T, E>
where
C: ErrorClassifier<E>,
E: std::fmt::Display,
F: FnMut() -> Fut,
Fut: Future<Output = Result<T, E>>,
{
let mut last_error: Option<E> = 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<TestError> 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
}
}
src/state_bus.rs
+130
View File
@@ -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<StateChange>,
}
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<StateChange> {
self.tx.subscribe()
}
}
impl Default for StateBus {
fn default() -> Self {
Self::new()
}
}
/// Convenience constructor for passing through `Arc`.
pub fn new_state_bus() -> Arc<StateBus> {
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"
);
}
}
src/util.rs
+91 -25
View File
@@ -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 `<tool_output …>…</tool_output>` 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("<tool_output") && !result.ends_with("</tool_output>") {
result.push_str("\n</tool_output>");
}
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 = "<tool_output name=\"search\" sanitized=\"true\">\nSome very long content here\n</tool_output>";
let result = truncate_preview(s, 60);
assert!(result.ends_with("</tool_output>")); // safety: test-only
assert!(result.contains("...")); // safety: test-only
}
}
tests/boundary_chaos.rs
+375
View File
@@ -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<TestError> 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<DomainEvent> = 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::<DomainEvent>(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
}