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

feat: Add lifecycle hooks system with 6 interception points (#18)

Browse Source 
* feat: Add lifecycle hooks system with 6 interception points

Implement extensible hook infrastructure for intercepting and transforming
agent operations at well-defined points in the lifecycle:

- BeforeInbound: intercept/modify/reject incoming user messages
- BeforeToolCall: intercept/modify/reject tool executions (chat + job)
- BeforeOutbound: intercept/modify/suppress outgoing responses
- TransformResponse: transform final response before completing a turn
- OnSessionStart: fire-and-forget notification on new session creation
- OnSessionEnd: fire-and-forget notification on session pruning

Hooks execute in priority order with modification chaining, reject
short-circuits, configurable failure modes (FailOpen/FailClosed),
and per-hook timeouts. Empty registry is zero-cost (all hooks pass
through immediately).

Co-Authored-By: Claude Opus 4.6 <[email protected]>

* fix: enforce hook fail-closed semantics

* Merge upstream/main into feat/hooks-system-clean

Resolve merge conflicts:
- FEATURE_PARITY.md: Keep both upstream cron/routines status and hooks status
- src/error.rs: Keep both Hook and Orchestrator/Worker error variants

Co-Authored-By: Claude Opus 4.6 <[email protected]>

* fix: resolve CI test failures in pairing store and wizard

- Fix pairing store truncate bug: record_failed_approve used
  .truncate(true) which wiped the file before reading, causing rate
  limiting to never accumulate past 1 attempt. Changed to
  .truncate(false) to preserve existing data.

- Fix wizard test: skip test_install_missing_bundled_channels when
  telegram WASM artifact specifically isn't available, not just when
  all channels are empty (whatsapp may exist without telegram).

- Add workspace exclude for subcrate directories to prevent cargo
  from discovering them as workspace members during builds.

Co-Authored-By: Claude Opus 4.6 <[email protected]>

* fix: address PR #18 review comments

- Remove duplicate maybe_hydrate_thread call (rebase artifact)
- Fix RwLock held across async hook execution in HookRegistry::run()
- Add tracing::warn for silent JSON parse failures in hook modifications
- Refactor execute_tool_inner to accept &WorkerDeps instead of 8 Arc params
- Use real user_id from JobContext instead of job_id UUID in BeforeToolCall hook

Co-Authored-By: Claude Opus 4.6 <[email protected]>

* fix: cargo fmt + remove tracked worktree breaking CI

- Apply rustfmt formatting (method chain line breaks, match arm style)
- Remove .claude/worktrees/ from git tracking (caused submodule error in CI)
- Add .claude/worktrees/ to .gitignore

Co-Authored-By: Claude Opus 4.6 <[email protected]>

---------

Co-authored-by: Firat Sertgoz <[email protected]>
Co-authored-by: Claude Opus 4.6 <[email protected]>
This commit is contained in:
firat.sertgoz
2026-02-17 05:40:05 +00:00
committed by GitHub
co-authored by Firat Sertgoz Claude Opus 4.6
parent 5e44185e48
commit 7c553b0973
22 changed files with 1058 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+3
View File
@@ -4,6 +4,9 @@
.env.*
!.env.example
# Claude Code worktrees
.claude/worktrees/
target/
# WASM build artifacts (loaded from disk, not bundled)
.sidecar/shells.json.lock
View File
.todos/command_usage.jsonl
+1
View File
@@ -0,0 +1 @@
{"ts":"2026-02-12T13:55:06.864688+04:00","cmd":"init","session":"ses_b1bea6","ok":true,"dur_ms":85}
.todos/config.json
+7
View File
@@ -0,0 +1,7 @@
{
"pane_heights": [
0.3333333333333333,
0.3333333333333333,
0.3333333333333333
]
}
.todos/config.json.lock
View File
.todos/db.lock
View File
.todos/issues.db
BIN
View File
Binary file not shown.
.todos/issues.db-shm
BIN
View File
Binary file not shown.
.todos/issues.db-wal
BIN
View File
Binary file not shown.
Cargo.toml
+8
View File
@@ -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"
FEATURE_PARITY.md
+9 -13
View File
@@ -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
src/agent/agent_loop.rs
+124 -7
View File
@@ -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<ToolRegistry>,
pub workspace: Option<Arc<Workspace>>,
pub extension_manager: Option<Arc<ExtensionManager>>,
pub hooks: Arc<HookRegistry>,
}
/// 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<HookRegistry> {
&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<Option<String>, 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(
src/agent/scheduler.rs
+5
View File
@@ -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<SafetyLayer>,
tools: Arc<ToolRegistry>,
store: Option<Arc<dyn Database>>,
hooks: Arc<HookRegistry>,
/// Running jobs (main LLM-driven jobs).
jobs: Arc<RwLock<HashMap<Uuid, ScheduledJob>>>,
/// Running sub-tasks (tool executions, background tasks).
@@ -64,6 +66,7 @@ impl Scheduler {
safety: Arc<SafetyLayer>,
tools: Arc<ToolRegistry>,
store: Option<Arc<dyn Database>>,
hooks: Arc<HookRegistry>,
) -> 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,
};
src/agent/session_manager.rs
+57 -4
View File
@@ -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<HashMap<String, Arc<Mutex<Session>>>>,
thread_map: RwLock<HashMap<ThreadKey, Uuid>>,
undo_managers: RwLock<HashMap<Uuid, Arc<Mutex<UndoManager>>>>,
hooks: Option<Arc<HookRegistry>>,
}
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<HookRegistry>) -> 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<Mutex<Session>> {
// 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<String> = {
// 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<String> = 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;
src/agent/worker.rs
+61 -42
View File
@@ -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<SafetyLayer>,
pub tools: Arc<ToolRegistry>,
pub store: Option<Arc<dyn Database>>,
pub hooks: Arc<HookRegistry>,
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,
&params,
)
.await;
let result = Self::execute_tool_inner(&deps, job_id, &tool_name, &params).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<ToolRegistry>,
context_manager: Arc<ContextManager>,
safety: Arc<SafetyLayer>,
store: Option<Arc<dyn Database>>,
deps: &WorkerDeps,
job_id: Uuid,
tool_name: &str,
params: &serde_json::Value,
) -> Result<String, Error> {
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(&params);
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<String, Error> {
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> {
src/error.rs
+3
View File
@@ -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),
src/hooks/hook.rs
+199
View File
@@ -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<String>,
},
/// 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<String>,
},
/// 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<String>,
},
/// 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<String>) -> 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<HookOutcome, HookError>;
}
src/hooks/mod.rs
+19
View File
@@ -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;
src/hooks/registry.rs
+555
View File
@@ -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<dyn Hook>,
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<Vec<HookEntry>>,
}
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<dyn Hook>) {
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<dyn Hook>, 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<String> {
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<HookOutcome, HookError> {
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<Arc<dyn Hook>> = {
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(&current_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(&current_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<HookPoint>,
}
#[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<HookOutcome, HookError> {
Ok(HookOutcome::ok())
}
}
/// A hook that modifies content by appending a suffix.
struct ModifyHook {
name: String,
suffix: String,
points: Vec<HookPoint>,
}
#[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<HookOutcome, HookError> {
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<HookPoint>,
}
#[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<HookOutcome, HookError> {
Ok(HookOutcome::reject(&self.reason))
}
}
/// A hook that always errors.
struct ErrorHook {
name: String,
points: Vec<HookPoint>,
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<HookOutcome, HookError> {
Err(HookError::ExecutionFailed {
reason: "test error".into(),
})
}
}
/// A hook that sleeps longer than its timeout.
struct SlowHook {
name: String,
points: Vec<HookPoint>,
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<HookOutcome, HookError> {
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());
}
}
src/lib.rs
+1
View File
@@ -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;
src/main.rs
+6 -1
View File
@@ -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(),
src/setup/wizard.rs
-2
View File
@@ -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") {