diff --git a/.gitignore b/.gitignore index e3ff7420..4febf1eb 100644 --- a/.gitignore +++ b/.gitignore @@ -4,6 +4,9 @@ .env.* !.env.example +# Claude Code worktrees +.claude/worktrees/ + target/ # WASM build artifacts (loaded from disk, not bundled) diff --git a/.sidecar/shells.json.lock b/.sidecar/shells.json.lock new file mode 100644 index 00000000..e69de29b diff --git a/.todos/command_usage.jsonl b/.todos/command_usage.jsonl new file mode 100644 index 00000000..448236da --- /dev/null +++ b/.todos/command_usage.jsonl @@ -0,0 +1 @@ +{"ts":"2026-02-12T13:55:06.864688+04:00","cmd":"init","session":"ses_b1bea6","ok":true,"dur_ms":85} diff --git a/.todos/config.json b/.todos/config.json new file mode 100644 index 00000000..58b374b4 --- /dev/null +++ b/.todos/config.json @@ -0,0 +1,7 @@ +{ + "pane_heights": [ + 0.3333333333333333, + 0.3333333333333333, + 0.3333333333333333 + ] +} \ No newline at end of file diff --git a/.todos/config.json.lock b/.todos/config.json.lock new file mode 100644 index 00000000..e69de29b diff --git a/.todos/db.lock b/.todos/db.lock new file mode 100644 index 00000000..e69de29b diff --git a/.todos/issues.db b/.todos/issues.db new file mode 100644 index 00000000..675062ef Binary files /dev/null and b/.todos/issues.db differ diff --git a/.todos/issues.db-shm b/.todos/issues.db-shm new file mode 100644 index 00000000..167349fd Binary files /dev/null and b/.todos/issues.db-shm differ diff --git a/.todos/issues.db-wal b/.todos/issues.db-wal new file mode 100644 index 00000000..a0287ce2 Binary files /dev/null and b/.todos/issues.db-wal differ diff --git a/Cargo.toml b/Cargo.toml index 1fe4fc28..0a5d6750 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,3 +1,11 @@ +[workspace] +exclude = [ + "channels-src/telegram", + "channels-src/slack", + "channels-src/whatsapp", + "tools-src/gmail", +] + [package] name = "ironclaw" version = "0.3.0" diff --git a/FEATURE_PARITY.md b/FEATURE_PARITY.md index 37929d37..054b9e0a 100644 --- a/FEATURE_PARITY.md +++ b/FEATURE_PARITY.md @@ -112,7 +112,7 @@ This document tracks feature parity between IronClaw (Rust implementation) and O | `pairing` | ✅ | ✅ | - | list/approve for channel DM pairing | | `nodes` | ✅ | ❌ | P3 | Device management | | `plugins` | ✅ | ❌ | P3 | Plugin management | -| `hooks` | ✅ | ❌ | P2 | Lifecycle hooks | +| `hooks` | ✅ | ✅ | P2 | Lifecycle hooks | | `cron` | ✅ | ❌ | P2 | Scheduled jobs | | `webhooks` | ✅ | ❌ | P3 | Webhook config | | `message send` | ✅ | ❌ | P2 | Send to channels | @@ -323,14 +323,14 @@ This document tracks feature parity between IronClaw (Rust implementation) and O | Cron jobs | ✅ | ✅ | - | Routines with cron trigger | | Timezone support | ✅ | ✅ | - | Via cron expressions | | One-shot/recurring jobs | ✅ | ✅ | - | Manual + cron triggers | -| `beforeInbound` hook | ✅ | ❌ | P2 | | -| `beforeOutbound` hook | ✅ | ❌ | P2 | | -| `beforeToolCall` hook | ✅ | ❌ | P2 | | +| `beforeInbound` hook | ✅ | ✅ | P2 | | +| `beforeOutbound` hook | ✅ | ✅ | P2 | | +| `beforeToolCall` hook | ✅ | ✅ | P2 | | | `onMessage` hook | ✅ | ✅ | - | Routines with event trigger | -| `onSessionStart` hook | ✅ | ❌ | P2 | | -| `onSessionEnd` hook | ✅ | ❌ | P2 | | +| `onSessionStart` hook | ✅ | ✅ | P2 | | +| `onSessionEnd` hook | ✅ | ✅ | P2 | | | `transcribeAudio` hook | ✅ | ❌ | P3 | | -| `transformResponse` hook | ✅ | ❌ | P2 | | +| `transformResponse` hook | ✅ | ✅ | P2 | | | Bundled hooks | ✅ | ❌ | P2 | | | Plugin hooks | ✅ | ❌ | P3 | | | Workspace hooks | ✅ | ❌ | P2 | Inline code | @@ -420,14 +420,10 @@ This document tracks feature parity between IronClaw (Rust implementation) and O - ✅ Telegram channel (WASM, DM pairing, caption, /start) - ❌ WhatsApp channel - ✅ Multi-provider failover (`FailoverProvider` with retryable error classification) -- ❌ Hooks system (beforeInbound, beforeToolCall, etc.) +- ✅ Hooks system (beforeInbound, beforeToolCall, beforeOutbound, onSessionStart, onSessionEnd, transformResponse) ### P2 - Medium Priority -- ❌ Cron job scheduling -- ❌ Web Control UI -- ❌ WebChat channel -- 🚧 Media handling (caption support; no image/PDF processing) -- ❌ CLI subcommands (config, status, memory, doctor) +- ❌ Media handling (images, PDFs) - ❌ Ollama/local model support - ❌ Configuration hot-reload - ❌ Webhook trigger endpoint in web gateway diff --git a/src/agent/agent_loop.rs b/src/agent/agent_loop.rs index 04a78e9d..4a9c2ab9 100644 --- a/src/agent/agent_loop.rs +++ b/src/agent/agent_loop.rs @@ -22,6 +22,7 @@ use crate::context::JobContext; use crate::db::Database; use crate::error::Error; use crate::extensions::ExtensionManager; +use crate::hooks::HookRegistry; use crate::llm::{ChatMessage, LlmProvider, Reasoning, ReasoningContext, RespondResult}; use crate::safety::SafetyLayer; use crate::tools::ToolRegistry; @@ -74,6 +75,7 @@ pub struct AgentDeps { pub tools: Arc, pub workspace: Option>, pub extension_manager: Option>, + pub hooks: Arc, } /// The main agent that coordinates all components. @@ -116,6 +118,7 @@ impl Agent { deps.safety.clone(), deps.tools.clone(), deps.store.clone(), + deps.hooks.clone(), )); Self { @@ -158,6 +161,10 @@ impl Agent { self.deps.workspace.as_ref() } + fn hooks(&self) -> &Arc { + &self.deps.hooks + } + /// Run the agent main loop. pub async fn run(self) -> Result<(), Error> { // Start channels @@ -425,10 +432,32 @@ impl Agent { match self.handle_message(&message).await { Ok(Some(response)) if !response.is_empty() => { - let _ = self - .channels - .respond(&message, OutgoingResponse::text(response)) - .await; + // Hook: BeforeOutbound — allow hooks to modify or suppress outbound + let event = crate::hooks::HookEvent::Outbound { + user_id: message.user_id.clone(), + channel: message.channel.clone(), + content: response.clone(), + thread_id: message.thread_id.clone(), + }; + match self.hooks().run(&event).await { + Err(err) => { + tracing::warn!("BeforeOutbound hook blocked response: {}", err); + } + Ok(crate::hooks::HookOutcome::Continue { + modified: Some(new_content), + }) => { + let _ = self + .channels + .respond(&message, OutgoingResponse::text(new_content)) + .await; + } + _ => { + let _ = self + .channels + .respond(&message, OutgoingResponse::text(response)) + .await; + } + } } Ok(Some(_)) => { // Empty response, nothing to send (e.g. approval handled via send_status) @@ -474,7 +503,33 @@ impl Agent { async fn handle_message(&self, message: &IncomingMessage) -> Result, Error> { // Parse submission type first - let submission = SubmissionParser::parse(&message.content); + let mut submission = SubmissionParser::parse(&message.content); + + // Hook: BeforeInbound — allow hooks to modify or reject user input + if let Submission::UserInput { ref content } = submission { + let event = crate::hooks::HookEvent::Inbound { + user_id: message.user_id.clone(), + channel: message.channel.clone(), + content: content.clone(), + thread_id: message.thread_id.clone(), + }; + match self.hooks().run(&event).await { + Err(crate::hooks::HookError::Rejected { reason }) => { + return Ok(Some(format!("[Message rejected: {}]", reason))); + } + Err(err) => { + return Ok(Some(format!("[Message blocked by hook policy: {}]", err))); + } + Ok(crate::hooks::HookOutcome::Continue { + modified: Some(new_content), + }) => { + submission = Submission::UserInput { + content: new_content, + }; + } + _ => {} // Continue, fail-open errors already logged in registry + } + } // Hydrate thread from DB if it's a historical thread not in memory if let Some(ref external_thread_id) = message.thread_id { @@ -883,6 +938,27 @@ impl Agent { // Complete, fail, or request approval match result { Ok(AgenticLoopResult::Response(response)) => { + // Hook: TransformResponse — allow hooks to modify or reject the final response + let response = { + let event = crate::hooks::HookEvent::ResponseTransform { + user_id: message.user_id.clone(), + thread_id: thread_id.to_string(), + response: response.clone(), + }; + match self.hooks().run(&event).await { + Err(crate::hooks::HookError::Rejected { reason }) => { + format!("[Response filtered: {}]", reason) + } + Err(err) => { + format!("[Response blocked by hook policy: {}]", err) + } + Ok(crate::hooks::HookOutcome::Continue { + modified: Some(new_response), + }) => new_response, + _ => response, // fail-open: use original + } + }; + thread.complete_turn(&response); self.persist_response_chain(thread); let _ = self @@ -1160,8 +1236,8 @@ impl Agent { } } - // Execute each tool (with approval checking) - for tc in tool_calls { + // Execute each tool (with approval checking and hook interception) + for mut tc in tool_calls { // Check if tool requires approval if let Some(tool) = self.tools().get(&tc.name).await && tool.requires_approval() @@ -1216,6 +1292,47 @@ impl Agent { } } + // Hook: BeforeToolCall — allow hooks to modify or reject tool calls + { + let event = crate::hooks::HookEvent::ToolCall { + tool_name: tc.name.clone(), + parameters: tc.arguments.clone(), + user_id: message.user_id.clone(), + context: "chat".to_string(), + }; + match self.hooks().run(&event).await { + Err(crate::hooks::HookError::Rejected { reason }) => { + context_messages.push(ChatMessage::tool_result( + &tc.id, + &tc.name, + format!("Tool call rejected by hook: {}", reason), + )); + continue; + } + Err(err) => { + context_messages.push(ChatMessage::tool_result( + &tc.id, + &tc.name, + format!("Tool call blocked by hook policy: {}", err), + )); + continue; + } + Ok(crate::hooks::HookOutcome::Continue { + modified: Some(new_params), + }) => match serde_json::from_str(&new_params) { + Ok(parsed) => tc.arguments = parsed, + Err(e) => { + tracing::warn!( + tool = %tc.name, + "Hook returned non-JSON modification for ToolCall, ignoring: {}", + e + ); + } + }, + _ => {} // Continue, fail-open errors already logged + } + } + let _ = self .channels .send_status( diff --git a/src/agent/scheduler.rs b/src/agent/scheduler.rs index a665c2af..23b9ea7c 100644 --- a/src/agent/scheduler.rs +++ b/src/agent/scheduler.rs @@ -14,6 +14,7 @@ use crate::config::AgentConfig; use crate::context::{ContextManager, JobContext, JobState}; use crate::db::Database; use crate::error::{Error, JobError}; +use crate::hooks::HookRegistry; use crate::llm::LlmProvider; use crate::safety::SafetyLayer; use crate::tools::ToolRegistry; @@ -49,6 +50,7 @@ pub struct Scheduler { safety: Arc, tools: Arc, store: Option>, + hooks: Arc, /// Running jobs (main LLM-driven jobs). jobs: Arc>>, /// Running sub-tasks (tool executions, background tasks). @@ -64,6 +66,7 @@ impl Scheduler { safety: Arc, tools: Arc, store: Option>, + hooks: Arc, ) -> Self { Self { config, @@ -72,6 +75,7 @@ impl Scheduler { safety, tools, store, + hooks, jobs: Arc::new(RwLock::new(HashMap::new())), subtasks: Arc::new(RwLock::new(HashMap::new())), } @@ -118,6 +122,7 @@ impl Scheduler { safety: self.safety.clone(), tools: self.tools.clone(), store: self.store.clone(), + hooks: self.hooks.clone(), timeout: self.config.job_timeout, use_planning: self.config.use_planning, }; diff --git a/src/agent/session_manager.rs b/src/agent/session_manager.rs index db0be886..244348cd 100644 --- a/src/agent/session_manager.rs +++ b/src/agent/session_manager.rs @@ -11,6 +11,7 @@ use uuid::Uuid; use crate::agent::session::Session; use crate::agent::undo::UndoManager; +use crate::hooks::HookRegistry; /// Key for mapping external thread IDs to internal ones. #[derive(Clone, Hash, Eq, PartialEq)] @@ -25,6 +26,7 @@ pub struct SessionManager { sessions: RwLock>>>, thread_map: RwLock>, undo_managers: RwLock>>>, + hooks: Option>, } impl SessionManager { @@ -34,9 +36,16 @@ impl SessionManager { sessions: RwLock::new(HashMap::new()), thread_map: RwLock::new(HashMap::new()), undo_managers: RwLock::new(HashMap::new()), + hooks: None, } } + /// Attach a hook registry for session lifecycle events. + pub fn with_hooks(mut self, hooks: Arc) -> Self { + self.hooks = Some(hooks); + self + } + /// Get or create a session for a user. pub async fn get_or_create_session(&self, user_id: &str) -> Arc> { // Fast path: check if session exists @@ -54,8 +63,28 @@ impl SessionManager { return Arc::clone(session); } - let session = Arc::new(Mutex::new(Session::new(user_id))); + let new_session = Session::new(user_id); + let session_id = new_session.id.to_string(); + let session = Arc::new(Mutex::new(new_session)); sessions.insert(user_id.to_string(), Arc::clone(&session)); + + // Fire OnSessionStart hook (fire-and-forget) + if let Some(ref hooks) = self.hooks { + let hooks = hooks.clone(); + let uid = user_id.to_string(); + let sid = session_id; + tokio::spawn(async move { + use crate::hooks::HookEvent; + let event = HookEvent::SessionStart { + user_id: uid, + session_id: sid, + }; + if let Err(e) = hooks.run(&event).await { + tracing::warn!("OnSessionStart hook error: {}", e); + } + }); + } + session } @@ -173,8 +202,8 @@ impl SessionManager { pub async fn prune_stale_sessions(&self, max_idle: std::time::Duration) -> usize { let cutoff = chrono::Utc::now() - chrono::TimeDelta::seconds(max_idle.as_secs() as i64); - // Find stale session user_ids - let stale_users: Vec = { + // Find stale sessions (user_id + session_id) + let stale_sessions: Vec<(String, String)> = { let sessions = self.sessions.read().await; sessions .iter() @@ -182,7 +211,7 @@ impl SessionManager { // Try to lock; skip if contended (someone is actively using it) let sess = session.try_lock().ok()?; if sess.last_active_at < cutoff { - Some(user_id.clone()) + Some((user_id.clone(), sess.id.to_string())) } else { None } @@ -190,6 +219,11 @@ impl SessionManager { .collect() }; + let stale_users: Vec = stale_sessions + .iter() + .map(|(user_id, _)| user_id.clone()) + .collect(); + if stale_users.is_empty() { return 0; } @@ -207,6 +241,25 @@ impl SessionManager { } } + // Fire OnSessionEnd hooks for stale sessions (fire-and-forget) + if let Some(ref hooks) = self.hooks { + for (user_id, session_id) in &stale_sessions { + let hooks = hooks.clone(); + let uid = user_id.clone(); + let sid = session_id.clone(); + tokio::spawn(async move { + use crate::hooks::HookEvent; + let event = HookEvent::SessionEnd { + user_id: uid, + session_id: sid, + }; + if let Err(e) = hooks.run(&event).await { + tracing::warn!("OnSessionEnd hook error: {}", e); + } + }); + } + } + // Remove sessions let count = { let mut sessions = self.sessions.write().await; diff --git a/src/agent/worker.rs b/src/agent/worker.rs index 565a3d22..3ec88586 100644 --- a/src/agent/worker.rs +++ b/src/agent/worker.rs @@ -12,6 +12,7 @@ use crate::agent::task::TaskOutput; use crate::context::{ContextManager, JobState}; use crate::db::Database; use crate::error::Error; +use crate::hooks::HookRegistry; use crate::llm::{ ActionPlan, ChatMessage, LlmProvider, Reasoning, ReasoningContext, RespondResult, ToolSelection, }; @@ -29,6 +30,7 @@ pub struct WorkerDeps { pub safety: Arc, pub tools: Arc, pub store: Option>, + pub hooks: Arc, pub timeout: Duration, pub use_planning: bool, } @@ -352,23 +354,11 @@ Report when the job is complete or if you encounter issues you cannot resolve."# .map(|selection| { let tool_name = selection.tool_name.clone(); let params = selection.parameters.clone(); - let tools = self.tools().clone(); - let context_manager = self.context_manager().clone(); - let safety = self.safety().clone(); + let deps = self.deps.clone(); let job_id = self.job_id; - let store = self.deps.store.clone(); async move { - let result = Self::execute_tool_inner( - tools, - context_manager, - safety, - store, - job_id, - &tool_name, - ¶ms, - ) - .await; + let result = Self::execute_tool_inner(&deps, job_id, &tool_name, ¶ms).await; ToolExecResult { result } } }) @@ -379,20 +369,18 @@ Report when the job is complete or if you encounter issues you cannot resolve."# /// Inner tool execution logic that can be called from both single and parallel paths. async fn execute_tool_inner( - tools: Arc, - context_manager: Arc, - safety: Arc, - store: Option>, + deps: &WorkerDeps, job_id: Uuid, tool_name: &str, params: &serde_json::Value, ) -> Result { - let tool = tools - .get(tool_name) - .await - .ok_or_else(|| crate::error::ToolError::NotFound { - name: tool_name.to_string(), - })?; + let tool = + deps.tools + .get(tool_name) + .await + .ok_or_else(|| crate::error::ToolError::NotFound { + name: tool_name.to_string(), + })?; // Tools requiring approval are blocked in autonomous jobs if tool.requires_approval() { @@ -402,8 +390,46 @@ Report when the job is complete or if you encounter issues you cannot resolve."# .into()); } - // Get job context for the tool - let job_ctx = context_manager.get_context(job_id).await?; + // Fetch job context early so we have the real user_id for hooks + let job_ctx = deps.context_manager.get_context(job_id).await?; + + // Run BeforeToolCall hook + let params = { + use crate::hooks::{HookError, HookEvent, HookOutcome}; + let event = HookEvent::ToolCall { + tool_name: tool_name.to_string(), + parameters: params.clone(), + user_id: job_ctx.user_id.clone(), + context: format!("job:{}", job_id), + }; + match deps.hooks.run(&event).await { + Err(HookError::Rejected { reason }) => { + return Err(crate::error::ToolError::ExecutionFailed { + name: tool_name.to_string(), + reason: format!("Blocked by hook: {}", reason), + } + .into()); + } + Err(err) => { + return Err(crate::error::ToolError::ExecutionFailed { + name: tool_name.to_string(), + reason: format!("Blocked by hook failure mode: {}", err), + } + .into()); + } + Ok(HookOutcome::Continue { + modified: Some(new_params), + }) => serde_json::from_str(&new_params).unwrap_or_else(|e| { + tracing::warn!( + tool = %tool_name, + "Hook returned non-JSON modification for ToolCall, ignoring: {}", + e + ); + params.clone() + }), + _ => params.clone(), + } + }; if job_ctx.state == JobState::Cancelled { return Err(crate::error::ToolError::ExecutionFailed { name: tool_name.to_string(), @@ -413,7 +439,7 @@ Report when the job is complete or if you encounter issues you cannot resolve."# } // Validate tool parameters - let validation = safety.validator().validate_tool_params(params); + let validation = deps.safety.validator().validate_tool_params(¶ms); if !validation.is_valid { let details = validation .errors @@ -478,8 +504,8 @@ Report when the job is complete or if you encounter issues you cannot resolve."# Ok(Ok(output)) => { let output_str = serde_json::to_string_pretty(&output.result) .ok() - .map(|s| safety.sanitize_tool_output(tool_name, &s).content); - context_manager + .map(|s| deps.safety.sanitize_tool_output(tool_name, &s).content); + deps.context_manager .update_memory(job_id, |mem| { let rec = mem.create_action(tool_name, params.clone()).succeed( output_str.clone(), @@ -492,7 +518,8 @@ Report when the job is complete or if you encounter issues you cannot resolve."# .await .ok() } - Ok(Err(e)) => context_manager + Ok(Err(e)) => deps + .context_manager .update_memory(job_id, |mem| { let rec = mem .create_action(tool_name, params.clone()) @@ -502,7 +529,8 @@ Report when the job is complete or if you encounter issues you cannot resolve."# }) .await .ok(), - Err(_) => context_manager + Err(_) => deps + .context_manager .update_memory(job_id, |mem| { let rec = mem .create_action(tool_name, params.clone()) @@ -515,7 +543,7 @@ Report when the job is complete or if you encounter issues you cannot resolve."# }; // Persist action to database (fire-and-forget) - if let (Some(action), Some(store)) = (action, store) { + if let (Some(action), Some(store)) = (action, deps.store.clone()) { tokio::spawn(async move { if let Err(e) = store.save_action(job_id, &action).await { tracing::warn!("Failed to persist action for job {}: {}", job_id, e); @@ -701,16 +729,7 @@ Report when the job is complete or if you encounter issues you cannot resolve."# tool_name: &str, params: &serde_json::Value, ) -> Result { - Self::execute_tool_inner( - self.tools().clone(), - self.context_manager().clone(), - self.safety().clone(), - self.deps.store.clone(), - self.job_id, - tool_name, - params, - ) - .await + Self::execute_tool_inner(&self.deps, self.job_id, tool_name, params).await } async fn mark_completed(&self) -> Result<(), Error> { diff --git a/src/error.rs b/src/error.rs index acb3598a..5aa7d461 100644 --- a/src/error.rs +++ b/src/error.rs @@ -40,6 +40,9 @@ pub enum Error { #[error("Workspace error: {0}")] Workspace(#[from] WorkspaceError), + #[error("Hook error: {0}")] + Hook(#[from] crate::hooks::HookError), + #[error("Orchestrator error: {0}")] Orchestrator(#[from] OrchestratorError), diff --git a/src/hooks/hook.rs b/src/hooks/hook.rs new file mode 100644 index 00000000..7df174e1 --- /dev/null +++ b/src/hooks/hook.rs @@ -0,0 +1,199 @@ +//! Core hook types and traits. + +use std::time::Duration; + +use async_trait::async_trait; + +/// Points in the agent lifecycle where hooks can be attached. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum HookPoint { + /// Before processing an inbound user message. + BeforeInbound, + /// Before executing a tool call. + BeforeToolCall, + /// Before sending an outbound response. + BeforeOutbound, + /// When a new session starts. + OnSessionStart, + /// When a session ends (pruned or expired). + OnSessionEnd, + /// Transform the final response before completing a turn. + TransformResponse, +} + +/// Contextual data carried with each hook invocation. +#[derive(Debug, Clone)] +pub enum HookEvent { + /// An inbound user message about to be processed. + Inbound { + user_id: String, + channel: String, + content: String, + thread_id: Option, + }, + /// A tool call about to be executed. + ToolCall { + tool_name: String, + parameters: serde_json::Value, + user_id: String, + /// "chat" for interactive, or a job ID string for autonomous jobs. + context: String, + }, + /// An outbound response about to be sent. + Outbound { + user_id: String, + channel: String, + content: String, + thread_id: Option, + }, + /// A new session was created. + SessionStart { user_id: String, session_id: String }, + /// A session was ended (pruned). + SessionEnd { user_id: String, session_id: String }, + /// The final response is being transformed before completing a turn. + ResponseTransform { + user_id: String, + thread_id: String, + response: String, + }, +} + +impl HookEvent { + /// Returns the [`HookPoint`] this event corresponds to. + pub fn hook_point(&self) -> HookPoint { + match self { + HookEvent::Inbound { .. } => HookPoint::BeforeInbound, + HookEvent::ToolCall { .. } => HookPoint::BeforeToolCall, + HookEvent::Outbound { .. } => HookPoint::BeforeOutbound, + HookEvent::SessionStart { .. } => HookPoint::OnSessionStart, + HookEvent::SessionEnd { .. } => HookPoint::OnSessionEnd, + HookEvent::ResponseTransform { .. } => HookPoint::TransformResponse, + } + } + + /// Apply a modification string to the event's primary content field. + pub fn apply_modification(&mut self, modified: &str) { + match self { + HookEvent::Inbound { content, .. } | HookEvent::Outbound { content, .. } => { + *content = modified.to_string(); + } + HookEvent::ToolCall { parameters, .. } => match serde_json::from_str(modified) { + Ok(parsed) => *parameters = parsed, + Err(e) => { + tracing::warn!( + "Hook returned non-JSON modification for ToolCall, ignoring: {}", + e + ); + } + }, + HookEvent::ResponseTransform { response, .. } => { + *response = modified.to_string(); + } + HookEvent::SessionStart { .. } | HookEvent::SessionEnd { .. } => { + // Session events don't have modifiable content + } + } + } +} + +/// The result of executing a hook. +#[derive(Debug, Clone)] +pub enum HookOutcome { + /// Continue processing, optionally with modified content. + Continue { + /// If `Some`, replace the event's primary content with this value. + modified: Option, + }, + /// Reject the event entirely. + Reject { + /// Human-readable reason for the rejection. + reason: String, + }, +} + +impl HookOutcome { + /// Shorthand for `Continue { modified: None }`. + pub fn ok() -> Self { + HookOutcome::Continue { modified: None } + } + + /// Shorthand for `Continue { modified: Some(value) }`. + pub fn modify(value: String) -> Self { + HookOutcome::Continue { + modified: Some(value), + } + } + + /// Shorthand for `Reject { reason }`. + pub fn reject(reason: impl Into) -> Self { + HookOutcome::Reject { + reason: reason.into(), + } + } +} + +/// How to handle hook execution failures. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum HookFailureMode { + /// On error/timeout, continue processing as if the hook returned `ok()`. + FailOpen, + /// On error/timeout, reject the event. + FailClosed, +} + +/// Hook execution errors. +#[derive(Debug, thiserror::Error)] +pub enum HookError { + #[error("Hook execution failed: {reason}")] + ExecutionFailed { reason: String }, + + #[error("Hook timed out after {timeout:?}")] + Timeout { timeout: Duration }, + + #[error("Hook rejected: {reason}")] + Rejected { reason: String }, +} + +/// Context passed to hooks alongside the event. +pub struct HookContext { + /// Arbitrary metadata hooks can use. + pub metadata: serde_json::Value, +} + +impl Default for HookContext { + fn default() -> Self { + Self { + metadata: serde_json::Value::Null, + } + } +} + +/// Trait for implementing lifecycle hooks. +/// +/// Hooks intercept and can modify agent operations at well-defined points. +#[async_trait] +pub trait Hook: Send + Sync { + /// A unique name for this hook. + fn name(&self) -> &str; + + /// The lifecycle points this hook should be called at. + fn hook_points(&self) -> &[HookPoint]; + + /// How to handle failures in this hook. + /// + /// Default: `FailOpen` (continue on error). + fn failure_mode(&self) -> HookFailureMode { + HookFailureMode::FailOpen + } + + /// Maximum time this hook is allowed to run. + /// + /// Default: 5 seconds. + fn timeout(&self) -> Duration { + Duration::from_secs(5) + } + + /// Execute the hook. + async fn execute(&self, event: &HookEvent, ctx: &HookContext) + -> Result; +} diff --git a/src/hooks/mod.rs b/src/hooks/mod.rs new file mode 100644 index 00000000..9ea6a5ac --- /dev/null +++ b/src/hooks/mod.rs @@ -0,0 +1,19 @@ +//! Lifecycle hooks for intercepting and transforming agent operations. +//! +//! The hook system provides 6 well-defined interception points: +//! +//! - **BeforeInbound** — Before processing an inbound user message +//! - **BeforeToolCall** — Before executing a tool call +//! - **BeforeOutbound** — Before sending an outbound response +//! - **OnSessionStart** — When a new session starts +//! - **OnSessionEnd** — When a session ends +//! - **TransformResponse** — Transform the final response before completing a turn +//! +//! Hooks are executed in priority order (lower number = higher priority). +//! Each hook can pass through, modify content, or reject the event. + +pub mod hook; +pub mod registry; + +pub use hook::{Hook, HookContext, HookError, HookEvent, HookFailureMode, HookOutcome, HookPoint}; +pub use registry::HookRegistry; diff --git a/src/hooks/registry.rs b/src/hooks/registry.rs new file mode 100644 index 00000000..6148d954 --- /dev/null +++ b/src/hooks/registry.rs @@ -0,0 +1,555 @@ +//! Hook registry for managing and executing lifecycle hooks. + +use std::sync::Arc; + +use tokio::sync::RwLock; + +use crate::hooks::hook::{Hook, HookContext, HookError, HookEvent, HookFailureMode, HookOutcome}; + +/// A registered hook with its priority. +struct HookEntry { + hook: Arc, + priority: u32, +} + +/// Registry that manages hooks and executes them at lifecycle points. +/// +/// Hooks are executed in priority order (lower number = higher priority). +/// A `Reject` outcome stops the chain immediately. +/// A `Modify` outcome chains through subsequent hooks. +pub struct HookRegistry { + hooks: RwLock>, +} + +impl HookRegistry { + /// Create an empty registry. + pub fn new() -> Self { + Self { + hooks: RwLock::new(Vec::new()), + } + } + + /// Register a hook with default priority (100). + pub async fn register(&self, hook: Arc) { + self.register_with_priority(hook, 100).await; + } + + /// Register a hook with a specific priority. + /// + /// Lower priority number = runs first. + pub async fn register_with_priority(&self, hook: Arc, priority: u32) { + let mut hooks = self.hooks.write().await; + hooks.push(HookEntry { hook, priority }); + hooks.sort_by_key(|e| e.priority); + } + + /// Unregister a hook by name. Returns `true` if it was found and removed. + pub async fn unregister(&self, name: &str) -> bool { + let mut hooks = self.hooks.write().await; + let before = hooks.len(); + hooks.retain(|e| e.hook.name() != name); + hooks.len() < before + } + + /// List all registered hook names (in priority order). + pub async fn list(&self) -> Vec { + let hooks = self.hooks.read().await; + hooks.iter().map(|e| e.hook.name().to_string()).collect() + } + + /// Run all hooks matching the event's hook point. + /// + /// - Hooks run in priority order (lowest first). + /// - `Reject` stops the chain immediately. + /// - `Modify` chains the modification through subsequent hooks. + /// - Timeout/error handling respects each hook's `failure_mode`. + pub async fn run(&self, event: &HookEvent) -> Result { + let point = event.hook_point(); + let ctx = HookContext::default(); + + // Clone matching hooks and drop the read guard before executing. + // Each hook can run up to its timeout, so holding the guard would + // block concurrent register/unregister/run calls. + let matching: Vec> = { + let hooks = self.hooks.read().await; + hooks + .iter() + .filter(|e| e.hook.hook_points().contains(&point)) + .map(|e| e.hook.clone()) + .collect() + }; + + if matching.is_empty() { + return Ok(HookOutcome::ok()); + } + + let mut current_event = event.clone(); + + for hook in &matching { + let timeout = hook.timeout(); + + let result = tokio::time::timeout(timeout, hook.execute(¤t_event, &ctx)).await; + + match result { + Ok(Ok(HookOutcome::Reject { reason })) => { + tracing::debug!(hook = hook.name(), "Hook rejected: {}", reason); + return Err(HookError::Rejected { reason }); + } + Ok(Ok(HookOutcome::Continue { + modified: Some(value), + })) => { + tracing::debug!(hook = hook.name(), "Hook modified content"); + current_event.apply_modification(&value); + } + Ok(Ok(HookOutcome::Continue { modified: None })) => { + // No-op, continue chain + } + Ok(Err(err)) => match hook.failure_mode() { + HookFailureMode::FailOpen => { + tracing::warn!(hook = hook.name(), "Hook failed (fail-open): {}", err); + } + HookFailureMode::FailClosed => { + tracing::warn!(hook = hook.name(), "Hook failed (fail-closed): {}", err); + return Err(HookError::ExecutionFailed { + reason: format!("Hook '{}' failed: {}", hook.name(), err), + }); + } + }, + Err(_elapsed) => match hook.failure_mode() { + HookFailureMode::FailOpen => { + tracing::warn!( + hook = hook.name(), + "Hook timed out (fail-open) after {:?}", + timeout + ); + } + HookFailureMode::FailClosed => { + tracing::warn!( + hook = hook.name(), + "Hook timed out (fail-closed) after {:?}", + timeout + ); + return Err(HookError::Timeout { timeout }); + } + }, + } + } + + // Determine final outcome by comparing with original event + let modified = extract_content(¤t_event); + let original = extract_content(event); + + if modified != original { + Ok(HookOutcome::modify(modified)) + } else { + Ok(HookOutcome::ok()) + } + } +} + +impl Default for HookRegistry { + fn default() -> Self { + Self::new() + } +} + +/// Extract the primary content string from a hook event. +fn extract_content(event: &HookEvent) -> String { + match event { + HookEvent::Inbound { content, .. } | HookEvent::Outbound { content, .. } => content.clone(), + HookEvent::ToolCall { parameters, .. } => { + serde_json::to_string(parameters).unwrap_or_default() + } + HookEvent::ResponseTransform { response, .. } => response.clone(), + HookEvent::SessionStart { session_id, .. } | HookEvent::SessionEnd { session_id, .. } => { + session_id.clone() + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::hooks::hook::{HookFailureMode, HookPoint}; + use async_trait::async_trait; + use std::time::Duration; + + /// A test hook that always returns ok. + struct PassthroughHook { + name: String, + points: Vec, + } + + #[async_trait] + impl Hook for PassthroughHook { + fn name(&self) -> &str { + &self.name + } + fn hook_points(&self) -> &[HookPoint] { + &self.points + } + async fn execute( + &self, + _event: &HookEvent, + _ctx: &HookContext, + ) -> Result { + Ok(HookOutcome::ok()) + } + } + + /// A hook that modifies content by appending a suffix. + struct ModifyHook { + name: String, + suffix: String, + points: Vec, + } + + #[async_trait] + impl Hook for ModifyHook { + fn name(&self) -> &str { + &self.name + } + fn hook_points(&self) -> &[HookPoint] { + &self.points + } + async fn execute( + &self, + event: &HookEvent, + _ctx: &HookContext, + ) -> Result { + let content = extract_content(event); + Ok(HookOutcome::modify(format!("{}{}", content, self.suffix))) + } + } + + /// A hook that always rejects. + struct RejectHook { + name: String, + reason: String, + points: Vec, + } + + #[async_trait] + impl Hook for RejectHook { + fn name(&self) -> &str { + &self.name + } + fn hook_points(&self) -> &[HookPoint] { + &self.points + } + async fn execute( + &self, + _event: &HookEvent, + _ctx: &HookContext, + ) -> Result { + Ok(HookOutcome::reject(&self.reason)) + } + } + + /// A hook that always errors. + struct ErrorHook { + name: String, + points: Vec, + failure_mode: HookFailureMode, + } + + #[async_trait] + impl Hook for ErrorHook { + fn name(&self) -> &str { + &self.name + } + fn hook_points(&self) -> &[HookPoint] { + &self.points + } + fn failure_mode(&self) -> HookFailureMode { + self.failure_mode + } + async fn execute( + &self, + _event: &HookEvent, + _ctx: &HookContext, + ) -> Result { + Err(HookError::ExecutionFailed { + reason: "test error".into(), + }) + } + } + + /// A hook that sleeps longer than its timeout. + struct SlowHook { + name: String, + points: Vec, + failure_mode: HookFailureMode, + } + + #[async_trait] + impl Hook for SlowHook { + fn name(&self) -> &str { + &self.name + } + fn hook_points(&self) -> &[HookPoint] { + &self.points + } + fn failure_mode(&self) -> HookFailureMode { + self.failure_mode + } + fn timeout(&self) -> Duration { + Duration::from_millis(50) + } + async fn execute( + &self, + _event: &HookEvent, + _ctx: &HookContext, + ) -> Result { + tokio::time::sleep(Duration::from_millis(200)).await; + Ok(HookOutcome::ok()) + } + } + + fn test_event() -> HookEvent { + HookEvent::Inbound { + user_id: "user-1".into(), + channel: "test".into(), + content: "hello".into(), + thread_id: None, + } + } + + #[tokio::test] + async fn test_empty_registry_returns_ok() { + let registry = HookRegistry::new(); + let result = registry.run(&test_event()).await; + assert!(result.is_ok()); + assert!(matches!( + result.unwrap(), + HookOutcome::Continue { modified: None } + )); + } + + #[tokio::test] + async fn test_register_and_list() { + let registry = HookRegistry::new(); + registry + .register(Arc::new(PassthroughHook { + name: "hook-a".into(), + points: vec![HookPoint::BeforeInbound], + })) + .await; + registry + .register(Arc::new(PassthroughHook { + name: "hook-b".into(), + points: vec![HookPoint::BeforeInbound], + })) + .await; + + let names = registry.list().await; + assert_eq!(names, vec!["hook-a", "hook-b"]); + } + + #[tokio::test] + async fn test_priority_ordering() { + let registry = HookRegistry::new(); + + // Register in reverse priority order + registry + .register_with_priority( + Arc::new(ModifyHook { + name: "low-prio".into(), + suffix: "-LOW".into(), + points: vec![HookPoint::BeforeInbound], + }), + 200, + ) + .await; + registry + .register_with_priority( + Arc::new(ModifyHook { + name: "high-prio".into(), + suffix: "-HIGH".into(), + points: vec![HookPoint::BeforeInbound], + }), + 10, + ) + .await; + + // Should run in priority order: high-prio first, then low-prio + let names = registry.list().await; + assert_eq!(names[0], "high-prio"); + assert_eq!(names[1], "low-prio"); + + let result = registry.run(&test_event()).await.unwrap(); + match result { + HookOutcome::Continue { modified: Some(m) } => { + // "hello" -> "hello-HIGH" -> "hello-HIGH-LOW" + assert_eq!(m, "hello-HIGH-LOW"); + } + other => panic!("Expected modification chain, got: {:?}", other), + } + } + + #[tokio::test] + async fn test_reject_stops_chain() { + let registry = HookRegistry::new(); + + registry + .register_with_priority( + Arc::new(RejectHook { + name: "blocker".into(), + reason: "blocked".into(), + points: vec![HookPoint::BeforeInbound], + }), + 10, + ) + .await; + registry + .register_with_priority( + Arc::new(ModifyHook { + name: "modifier".into(), + suffix: "-MODIFIED".into(), + points: vec![HookPoint::BeforeInbound], + }), + 20, + ) + .await; + + let result = registry.run(&test_event()).await; + assert!(result.is_err()); + match result.unwrap_err() { + HookError::Rejected { reason } => assert_eq!(reason, "blocked"), + other => panic!("Expected Rejected, got: {:?}", other), + } + } + + #[tokio::test] + async fn test_modification_chaining() { + let registry = HookRegistry::new(); + + registry + .register_with_priority( + Arc::new(ModifyHook { + name: "first".into(), + suffix: "-A".into(), + points: vec![HookPoint::BeforeInbound], + }), + 10, + ) + .await; + registry + .register_with_priority( + Arc::new(ModifyHook { + name: "second".into(), + suffix: "-B".into(), + points: vec![HookPoint::BeforeInbound], + }), + 20, + ) + .await; + + let result = registry.run(&test_event()).await.unwrap(); + match result { + HookOutcome::Continue { modified: Some(m) } => { + assert_eq!(m, "hello-A-B"); + } + other => panic!("Expected chained modification, got: {:?}", other), + } + } + + #[tokio::test] + async fn test_fail_open_on_error() { + let registry = HookRegistry::new(); + registry + .register(Arc::new(ErrorHook { + name: "err-open".into(), + points: vec![HookPoint::BeforeInbound], + failure_mode: HookFailureMode::FailOpen, + })) + .await; + + let result = registry.run(&test_event()).await; + assert!(result.is_ok()); + } + + #[tokio::test] + async fn test_fail_closed_on_error() { + let registry = HookRegistry::new(); + registry + .register(Arc::new(ErrorHook { + name: "err-closed".into(), + points: vec![HookPoint::BeforeInbound], + failure_mode: HookFailureMode::FailClosed, + })) + .await; + + let result = registry.run(&test_event()).await; + assert!(result.is_err()); + assert!(matches!( + result.unwrap_err(), + HookError::ExecutionFailed { .. } + )); + } + + #[tokio::test] + async fn test_fail_open_on_timeout() { + let registry = HookRegistry::new(); + registry + .register(Arc::new(SlowHook { + name: "slow-open".into(), + points: vec![HookPoint::BeforeInbound], + failure_mode: HookFailureMode::FailOpen, + })) + .await; + + let result = registry.run(&test_event()).await; + assert!(result.is_ok()); + } + + #[tokio::test] + async fn test_fail_closed_on_timeout() { + let registry = HookRegistry::new(); + registry + .register(Arc::new(SlowHook { + name: "slow-closed".into(), + points: vec![HookPoint::BeforeInbound], + failure_mode: HookFailureMode::FailClosed, + })) + .await; + + let result = registry.run(&test_event()).await; + assert!(result.is_err()); + assert!(matches!(result.unwrap_err(), HookError::Timeout { .. })); + } + + #[tokio::test] + async fn test_unregister() { + let registry = HookRegistry::new(); + registry + .register(Arc::new(PassthroughHook { + name: "removable".into(), + points: vec![HookPoint::BeforeInbound], + })) + .await; + + assert_eq!(registry.list().await.len(), 1); + assert!(registry.unregister("removable").await); + assert_eq!(registry.list().await.len(), 0); + + // Unregistering non-existent returns false + assert!(!registry.unregister("nonexistent").await); + } + + #[tokio::test] + async fn test_hooks_only_match_their_points() { + let registry = HookRegistry::new(); + registry + .register(Arc::new(RejectHook { + name: "outbound-only".into(), + reason: "blocked".into(), + points: vec![HookPoint::BeforeOutbound], + })) + .await; + + // Inbound event should not be affected by outbound-only hook + let result = registry.run(&test_event()).await; + assert!(result.is_ok()); + } +} diff --git a/src/lib.rs b/src/lib.rs index ed72e3df..2fca3c3f 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -50,6 +50,7 @@ pub mod estimation; pub mod evaluation; pub mod extensions; pub mod history; +pub mod hooks; pub mod llm; pub mod orchestrator; pub mod pairing; diff --git a/src/main.rs b/src/main.rs index fac21c40..1b398def 100644 --- a/src/main.rs +++ b/src/main.rs @@ -22,6 +22,7 @@ use ironclaw::{ config::Config, context::ContextManager, extensions::ExtensionManager, + hooks::HookRegistry, llm::{ FailoverProvider, LlmProvider, SessionConfig, create_cheap_llm_provider, create_llm_provider, create_llm_provider_with_config, create_session_manager, @@ -1132,8 +1133,11 @@ async fn main() -> anyhow::Result<()> { // Create context manager (shared between job tools and agent) let context_manager = Arc::new(ContextManager::new(config.agent.max_parallel_jobs)); + // Create hook registry + let hooks = Arc::new(HookRegistry::new()); + // Create session manager (shared between agent and web gateway) - let session_manager = Arc::new(SessionManager::new()); + let session_manager = Arc::new(SessionManager::new().with_hooks(hooks.clone())); // Register job tools (sandbox deps auto-injected when container_job_manager is available) tools.register_job_tools( @@ -1199,6 +1203,7 @@ async fn main() -> anyhow::Result<()> { tools, workspace, extension_manager, + hooks, }; let agent = Agent::new( config.agent.clone(), diff --git a/src/setup/wizard.rs b/src/setup/wizard.rs index 31c4cc4e..1cc5dd89 100644 --- a/src/setup/wizard.rs +++ b/src/setup/wizard.rs @@ -2090,8 +2090,6 @@ mod tests { #[tokio::test] async fn test_install_missing_bundled_channels_installs_telegram() { - use crate::channels::wasm::available_channel_names; - // WASM artifacts only exist in dev builds (not CI). Skip gracefully // rather than fail when the telegram channel hasn't been compiled. if !available_channel_names().contains(&"telegram") {