diff --git a/CLAUDE.md b/CLAUDE.md index e18a165d..59f478ba 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -117,3 +117,4 @@ Tools declare a `PermissionLevel` (ReadOnly / Supervised / Dangerous). The runti ## Principles - Architecture must conform to SOLID, GRASP, and YAGNI; files should stay under 200 lines; balance cohesion and SRP — split by reason to change, not by line count. +- Names must be specific and descriptive — files, modules, functions, and variables should say exactly what they do. Avoid vague names like `common`, `helpers`, `utils`, `misc`, `edge_test`, `manager`, `handler`, `data`, `info`, `process`. diff --git a/crates/loopal-acp/src/translate/mod.rs b/crates/loopal-acp/src/translate/mod.rs index d5b1fdfa..ef497eda 100644 --- a/crates/loopal-acp/src/translate/mod.rs +++ b/crates/loopal-acp/src/translate/mod.rs @@ -112,7 +112,8 @@ pub fn translate_event(payload: &AgentEventPayload, session_id: &str) -> Option< | AgentEventPayload::ServerToolResult { .. } | AgentEventPayload::RetryCleared | AgentEventPayload::SubAgentSpawned { .. } - | AgentEventPayload::AutoModeDecision { .. } => None, + | AgentEventPayload::AutoModeDecision { .. } + | AgentEventPayload::TurnCompleted { .. } => None, } } diff --git a/crates/loopal-agent-client/src/bridge_handlers.rs b/crates/loopal-agent-client/src/bridge_handlers.rs index 318875cd..9555f220 100644 --- a/crates/loopal-agent-client/src/bridge_handlers.rs +++ b/crates/loopal-agent-client/src/bridge_handlers.rs @@ -26,6 +26,9 @@ pub(crate) async fn handle_permission( let tool_id = params["tool_call_id"].as_str().unwrap_or("").to_string(); let event = AgentEvent { agent_name: None, + event_id: loopal_protocol::event_id::next_event_id(), + turn_id: loopal_protocol::event_id::current_turn_id(), + correlation_id: loopal_protocol::event_id::current_correlation_id(), payload: loopal_protocol::AgentEventPayload::ToolPermissionRequest { id: tool_id, name: tool_name.clone(), @@ -60,6 +63,9 @@ pub(crate) async fn handle_question( if let Ok(questions) = parsed { let event = AgentEvent { agent_name: None, + event_id: loopal_protocol::event_id::next_event_id(), + turn_id: 0, + correlation_id: 0, payload: loopal_protocol::AgentEventPayload::UserQuestionRequest { id: "ipc".into(), questions, diff --git a/crates/loopal-agent-server/src/agent_setup.rs b/crates/loopal-agent-server/src/agent_setup.rs index 92914960..a1f071cd 100644 --- a/crates/loopal-agent-server/src/agent_setup.rs +++ b/crates/loopal-agent-server/src/agent_setup.rs @@ -103,10 +103,14 @@ pub fn build_with_frontend( ); let auto_classifier = if permission_mode == loopal_tool_api::PermissionMode::Auto { - Some(Arc::new(loopal_auto_mode::AutoClassifier::new( - config.instructions.clone(), - cwd.to_string_lossy().into_owned(), - ))) + Some(Arc::new( + loopal_auto_mode::AutoClassifier::new_with_thresholds( + config.instructions.clone(), + cwd.to_string_lossy().into_owned(), + config.settings.harness.cb_max_consecutive_denials, + config.settings.harness.cb_max_total_denials, + ), + )) } else { None }; @@ -197,6 +201,8 @@ pub fn build_with_frontend( memory_channel, scheduled_rx: Some(scheduled_rx), auto_classifier, + harness: config.settings.harness.clone(), + rewake_rx: None, // TODO: wire from AsyncHookStore when async hooks are configured }; Ok(params) } diff --git a/crates/loopal-agent-server/src/hub_emitter.rs b/crates/loopal-agent-server/src/hub_emitter.rs index da26818e..c1af0a73 100644 --- a/crates/loopal-agent-server/src/hub_emitter.rs +++ b/crates/loopal-agent-server/src/hub_emitter.rs @@ -25,6 +25,9 @@ impl EventEmitter for HubEventEmitter { }; let event = AgentEvent { agent_name: self.agent_name.clone(), + event_id: loopal_protocol::event_id::next_event_id(), + turn_id: loopal_protocol::event_id::current_turn_id(), + correlation_id: loopal_protocol::event_id::current_correlation_id(), payload, }; let params = serde_json::to_value(&event) diff --git a/crates/loopal-agent-server/src/hub_frontend.rs b/crates/loopal-agent-server/src/hub_frontend.rs index ef599a31..549d77db 100644 --- a/crates/loopal-agent-server/src/hub_frontend.rs +++ b/crates/loopal-agent-server/src/hub_frontend.rs @@ -61,6 +61,9 @@ impl AgentFrontend for HubFrontend { async fn emit(&self, payload: AgentEventPayload) -> Result<()> { let event = AgentEvent { agent_name: self.agent_name.clone(), + event_id: loopal_protocol::event_id::next_event_id(), + turn_id: loopal_protocol::event_id::current_turn_id(), + correlation_id: loopal_protocol::event_id::current_correlation_id(), payload, }; let params = serde_json::to_value(&event) @@ -174,6 +177,9 @@ impl AgentFrontend for HubFrontend { fn try_emit(&self, payload: AgentEventPayload) -> bool { let event = AgentEvent { agent_name: self.agent_name.clone(), + event_id: loopal_protocol::event_id::next_event_id(), + turn_id: loopal_protocol::event_id::current_turn_id(), + correlation_id: loopal_protocol::event_id::current_correlation_id(), payload, }; let Ok(params) = serde_json::to_value(&event) else { diff --git a/crates/loopal-agent-server/src/ipc_emitter.rs b/crates/loopal-agent-server/src/ipc_emitter.rs index f7865549..f7cdd5fe 100644 --- a/crates/loopal-agent-server/src/ipc_emitter.rs +++ b/crates/loopal-agent-server/src/ipc_emitter.rs @@ -21,6 +21,9 @@ impl EventEmitter for IpcEventEmitter { async fn emit(&self, payload: AgentEventPayload) -> Result<()> { let event = AgentEvent { agent_name: self.agent_name.clone(), + event_id: loopal_protocol::event_id::next_event_id(), + turn_id: loopal_protocol::event_id::current_turn_id(), + correlation_id: loopal_protocol::event_id::current_correlation_id(), payload, }; let params = serde_json::to_value(&event) diff --git a/crates/loopal-agent-server/src/ipc_frontend.rs b/crates/loopal-agent-server/src/ipc_frontend.rs index 9d87ad64..e6fafcbe 100644 --- a/crates/loopal-agent-server/src/ipc_frontend.rs +++ b/crates/loopal-agent-server/src/ipc_frontend.rs @@ -44,6 +44,9 @@ impl AgentFrontend for IpcFrontend { async fn emit(&self, payload: AgentEventPayload) -> Result<()> { let event = AgentEvent { agent_name: self.agent_name.clone(), + event_id: loopal_protocol::event_id::next_event_id(), + turn_id: loopal_protocol::event_id::current_turn_id(), + correlation_id: loopal_protocol::event_id::current_correlation_id(), payload, }; let params = serde_json::to_value(&event) @@ -180,6 +183,9 @@ impl AgentFrontend for IpcFrontend { fn try_emit(&self, payload: AgentEventPayload) -> bool { let event = AgentEvent { agent_name: self.agent_name.clone(), + event_id: loopal_protocol::event_id::next_event_id(), + turn_id: loopal_protocol::event_id::current_turn_id(), + correlation_id: loopal_protocol::event_id::current_correlation_id(), payload, }; let Ok(params) = serde_json::to_value(&event) else { diff --git a/crates/loopal-agent-server/tests/suite/params_test.rs b/crates/loopal-agent-server/tests/suite/params_test.rs index 559943cc..a0bb81ac 100644 --- a/crates/loopal-agent-server/tests/suite/params_test.rs +++ b/crates/loopal-agent-server/tests/suite/params_test.rs @@ -67,6 +67,9 @@ async fn event_forwarder_delivers_sub_agent_events() { // Send a sub-agent event through the channel let event = loopal_protocol::AgentEvent { agent_name: Some("sub-1".into()), + event_id: 0, + turn_id: 0, + correlation_id: 0, payload: loopal_protocol::AgentEventPayload::Stream { text: "from sub-agent".into(), }, diff --git a/crates/loopal-auto-mode/src/circuit_breaker.rs b/crates/loopal-auto-mode/src/circuit_breaker.rs index e8b57fa7..7066f3cd 100644 --- a/crates/loopal-auto-mode/src/circuit_breaker.rs +++ b/crates/loopal-auto-mode/src/circuit_breaker.rs @@ -12,6 +12,8 @@ const MAX_TOTAL_DENIALS: u32 = 20; /// are exceeded, preventing runaway classification loops. pub struct CircuitBreaker { inner: Mutex, + max_consecutive: u32, + max_total: u32, } struct Inner { @@ -25,12 +27,19 @@ struct Inner { impl CircuitBreaker { pub fn new() -> Self { + Self::with_thresholds(MAX_CONSECUTIVE_DENIALS, MAX_TOTAL_DENIALS) + } + + /// Create with custom thresholds (from HarnessConfig). + pub fn with_thresholds(max_consecutive: u32, max_total: u32) -> Self { Self { inner: Mutex::new(Inner { consecutive: HashMap::new(), total_denials: 0, degraded: false, }), + max_consecutive, + max_total, } } @@ -39,9 +48,9 @@ impl CircuitBreaker { let mut inner = self.inner.lock().unwrap(); let count = inner.consecutive.entry(tool_name.to_string()).or_insert(0); *count += 1; - let consecutive_exceeded = *count >= MAX_CONSECUTIVE_DENIALS; + let consecutive_exceeded = *count >= self.max_consecutive; inner.total_denials += 1; - if consecutive_exceeded || inner.total_denials >= MAX_TOTAL_DENIALS { + if consecutive_exceeded || inner.total_denials >= self.max_total { inner.degraded = true; } } diff --git a/crates/loopal-auto-mode/src/classifier.rs b/crates/loopal-auto-mode/src/classifier.rs index edd00846..019f72c0 100644 --- a/crates/loopal-auto-mode/src/classifier.rs +++ b/crates/loopal-auto-mode/src/classifier.rs @@ -41,6 +41,21 @@ impl AutoClassifier { } } + /// Create with custom circuit breaker thresholds (from HarnessConfig). + pub fn new_with_thresholds( + instructions: String, + cwd: String, + max_consecutive: u32, + max_total: u32, + ) -> Self { + Self { + circuit_breaker: CircuitBreaker::with_thresholds(max_consecutive, max_total), + cache: ClassifierCache::new(), + instructions, + cwd, + } + } + /// Whether the circuit breaker has tripped (too many denials/errors). pub fn is_degraded(&self) -> bool { self.circuit_breaker.is_degraded() diff --git a/crates/loopal-config/src/harness.rs b/crates/loopal-config/src/harness.rs new file mode 100644 index 00000000..9a833a1d --- /dev/null +++ b/crates/loopal-config/src/harness.rs @@ -0,0 +1,35 @@ +use serde::{Deserialize, Serialize}; + +/// Control-loop parameters for the agent harness. +/// +/// All fields have sensible defaults matching the previous hardcoded values. +/// Override via `settings.json` under the `"harness"` key. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(default)] +pub struct HarnessConfig { + /// Loop detection: warn after this many repeated tool calls (default: 3). + pub loop_warn_threshold: u32, + /// Loop detection: abort turn after this many repeats (default: 5). + pub loop_abort_threshold: u32, + /// Auto-mode circuit breaker: max consecutive denials per tool (default: 3). + pub cb_max_consecutive_denials: u32, + /// Auto-mode circuit breaker: max total denials per session (default: 20). + pub cb_max_total_denials: u32, + /// Max automatic continuations when LLM hits max_tokens (default: 3). + pub max_auto_continuations: u32, + /// Max Stop hook feedback rounds before forcing exit (default: 2). + pub max_stop_feedback: u32, +} + +impl Default for HarnessConfig { + fn default() -> Self { + Self { + loop_warn_threshold: 3, + loop_abort_threshold: 5, + cb_max_consecutive_denials: 3, + cb_max_total_denials: 20, + max_auto_continuations: 3, + max_stop_feedback: 2, + } + } +} diff --git a/crates/loopal-config/src/hook.rs b/crates/loopal-config/src/hook.rs index a16a7115..63419378 100644 --- a/crates/loopal-config/src/hook.rs +++ b/crates/loopal-config/src/hook.rs @@ -1,7 +1,9 @@ +use std::collections::HashMap; + use serde::{Deserialize, Serialize}; -/// Hook event types -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +/// Hook event types — lifecycle points where hooks can intercept. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum HookEvent { /// Before tool execution @@ -12,35 +14,77 @@ pub enum HookEvent { PreRequest, /// After user submits input PostInput, + /// When a session starts + SessionStart, + /// When a session ends + SessionEnd, + /// Right before the agent concludes its response (exit-gate) + Stop, + /// Before conversation compaction + PreCompact, } -/// Hook configuration +/// Hook executor type — determines how the hook runs. +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum HookType { + /// Shell command (default, backward compatible). + #[default] + Command, + /// HTTP POST to a webhook URL. + Http, + /// LLM prompt hook (lightweight classifier call). + Prompt, +} + +/// Hook configuration. +/// +/// Backward compatible: `{"event": "pre_tool_use", "command": "echo hi"}` +/// still works (type defaults to Command, url/prompt/headers ignored). #[derive(Debug, Clone, Serialize, Deserialize)] pub struct HookConfig { - /// Event that triggers this hook pub event: HookEvent, - /// Shell command to execute + /// Executor type (default: command). + #[serde(default, rename = "type")] + pub hook_type: HookType, + /// Shell command (Command type). Ignored for Http/Prompt types (leave empty). + #[serde(default)] pub command: String, - /// Optional: only trigger for specific tool names + /// Webhook URL (required for Http type). + #[serde(default)] + pub url: Option, + /// HTTP headers (Http type only). + #[serde(default)] + pub headers: HashMap, + /// LLM prompt (required for Prompt type). + #[serde(default)] + pub prompt: Option, + /// LLM model override (Prompt type only). + #[serde(default)] + pub model: Option, + /// Legacy tool_filter (use `condition` instead). #[serde(default)] pub tool_filter: Option>, - /// Timeout in milliseconds (default: 10000) + /// Condition expression: "Bash(git push*)", "Write(*.rs)", "*" + #[serde(default, rename = "if")] + pub condition: Option, + /// Timeout in milliseconds (default: 10000). #[serde(default = "default_hook_timeout")] pub timeout_ms: u64, + /// Deduplication ID across config layers. + #[serde(default)] + pub id: Option, } fn default_hook_timeout() -> u64 { 10_000 } -/// Result from hook execution +/// Result from hook execution (legacy, used by runner.rs). #[derive(Debug, Clone)] pub struct HookResult { - /// Exit code (0 = success) pub exit_code: i32, - /// Stdout output pub stdout: String, - /// Stderr output pub stderr: String, } diff --git a/crates/loopal-config/src/hook_condition.rs b/crates/loopal-config/src/hook_condition.rs new file mode 100644 index 00000000..7b8ff7c7 --- /dev/null +++ b/crates/loopal-config/src/hook_condition.rs @@ -0,0 +1,131 @@ +//! Condition expression for hook matching. +//! +//! Replaces simple `tool_filter: ["Bash"]` with expressive patterns: +//! - `"Bash(git push*)"` — tool name + argument glob +//! - `"Write(*.rs)"` — file path glob +//! - `"Bash|Write"` — OR multiple tools +//! - `"*"` — match everything + +/// Parsed condition for hook matching. +/// +/// Stored as a raw string in config; parsed on demand for matching. +/// This avoids upfront compilation cost for hooks that may never fire. +pub fn matches_condition(condition: &str, tool_name: &str, tool_input: &serde_json::Value) -> bool { + if condition == "*" { + return true; + } + + // OR syntax: "Bash|Write|Edit" + if condition.contains('|') && !condition.contains('(') { + return condition.split('|').any(|part| part.trim() == tool_name); + } + + // Tool(glob) syntax: "Bash(git push*)" + if let Some(paren_start) = condition.find('(') { + let name_part = &condition[..paren_start]; + if name_part != tool_name { + return false; + } + let glob_part = condition[paren_start + 1..] + .strip_suffix(')') + .unwrap_or(&condition[paren_start + 1..]); + let primary_arg = extract_primary_arg(tool_name, tool_input); + return glob_match(glob_part, &primary_arg); + } + + // Plain tool name: "Bash" + condition == tool_name +} + +/// Extract the "primary argument" from tool input for glob matching. +fn extract_primary_arg(tool_name: &str, input: &serde_json::Value) -> String { + let field = match tool_name { + "Bash" => "command", + "Write" | "Edit" | "MultiEdit" | "Read" => "file_path", + "ApplyPatch" => "patch", + "Grep" | "Glob" => "pattern", + "Fetch" => "url", + "WebSearch" => "query", + "Ls" => "path", + _ => return input.to_string(), + }; + input + .get(field) + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string() +} + +/// Simple glob matching: `*` matches any substring, `?` matches one char. +fn glob_match(pattern: &str, text: &str) -> bool { + let p: Vec = pattern.chars().collect(); + let t: Vec = text.chars().collect(); + glob_match_inner(&p, &t, 0, 0) +} + +fn glob_match_inner(pattern: &[char], text: &[char], pi: usize, ti: usize) -> bool { + if pi == pattern.len() { + return ti == text.len(); + } + if pattern[pi] == '*' { + // '*' matches zero or more characters + for skip in ti..=text.len() { + if glob_match_inner(pattern, text, pi + 1, skip) { + return true; + } + } + return false; + } + if ti == text.len() { + return false; + } + if pattern[pi] == '?' || pattern[pi] == text[ti] { + return glob_match_inner(pattern, text, pi + 1, ti + 1); + } + false +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + #[test] + fn wildcard_matches_all() { + assert!(matches_condition("*", "Bash", &json!({}))); + assert!(matches_condition("*", "Write", &json!({}))); + } + + #[test] + fn plain_tool_name() { + assert!(matches_condition("Bash", "Bash", &json!({}))); + assert!(!matches_condition("Bash", "Write", &json!({}))); + } + + #[test] + fn or_syntax() { + assert!(matches_condition("Bash|Write", "Bash", &json!({}))); + assert!(matches_condition("Bash|Write", "Write", &json!({}))); + assert!(!matches_condition("Bash|Write", "Read", &json!({}))); + } + + #[test] + fn tool_with_glob() { + let input = json!({"command": "git push origin main"}); + assert!(matches_condition("Bash(git push*)", "Bash", &input)); + assert!(!matches_condition("Bash(git pull*)", "Bash", &input)); + } + + #[test] + fn file_path_glob() { + let input = json!({"file_path": "src/main.rs"}); + assert!(matches_condition("Write(*.rs)", "Write", &input)); + assert!(!matches_condition("Write(*.ts)", "Write", &input)); + } + + #[test] + fn wrong_tool_with_glob() { + let input = json!({"command": "git push"}); + assert!(!matches_condition("Write(git*)", "Bash", &input)); + } +} diff --git a/crates/loopal-config/src/lib.rs b/crates/loopal-config/src/lib.rs index 5bb164a7..7134d5f9 100644 --- a/crates/loopal-config/src/lib.rs +++ b/crates/loopal-config/src/lib.rs @@ -1,4 +1,6 @@ +pub mod harness; pub mod hook; +pub mod hook_condition; pub mod housekeeping; pub mod layer; pub mod loader; @@ -12,7 +14,8 @@ pub mod settings; pub mod skills; mod validate; -pub use hook::{HookConfig, HookEvent, HookResult}; +pub use harness::HarnessConfig; +pub use hook::{HookConfig, HookEvent, HookResult, HookType}; pub use layer::{ConfigLayer, LayerSource}; pub use locations::*; pub use pipeline::load_config; diff --git a/crates/loopal-config/src/resolver.rs b/crates/loopal-config/src/resolver.rs index 6646d356..a52d63ba 100644 --- a/crates/loopal-config/src/resolver.rs +++ b/crates/loopal-config/src/resolver.rs @@ -72,8 +72,18 @@ impl ConfigResolver { ); } - // Hooks: append all, preserving order + // Hooks: dedup by id (higher layer wins), append others. for config in layer.hooks { + if let Some(ref id) = config.id { + // Same id across layers: higher-priority layer replaces. + if let Some(pos) = hooks.iter().position(|h| h.config.id.as_ref() == Some(id)) { + hooks[pos] = HookEntry { + config, + source: layer.source.clone(), + }; + continue; + } + } hooks.push(HookEntry { config, source: layer.source.clone(), diff --git a/crates/loopal-config/src/settings.rs b/crates/loopal-config/src/settings.rs index ecbdd983..ebaa7576 100644 --- a/crates/loopal-config/src/settings.rs +++ b/crates/loopal-config/src/settings.rs @@ -3,6 +3,7 @@ use std::collections::HashMap; use indexmap::IndexMap; use serde::{Deserialize, Serialize}; +use crate::harness::HarnessConfig; use crate::hook::HookConfig; use crate::sandbox::SandboxConfig; use loopal_provider_api::{ModelOverride, TaskType, ThinkingConfig}; @@ -52,6 +53,10 @@ pub struct Settings { /// Auto-memory configuration #[serde(default)] pub memory: MemoryConfig, + + /// Harness control parameters — configurable thresholds for the agent control loop. + #[serde(default)] + pub harness: HarnessConfig, } impl Default for Settings { @@ -68,6 +73,7 @@ impl Default for Settings { sandbox: SandboxConfig::default(), thinking: ThinkingConfig::default(), memory: MemoryConfig::default(), + harness: HarnessConfig::default(), } } } diff --git a/crates/loopal-config/tests/suite.rs b/crates/loopal-config/tests/suite.rs index 9837ca74..41f90231 100644 --- a/crates/loopal-config/tests/suite.rs +++ b/crates/loopal-config/tests/suite.rs @@ -17,6 +17,8 @@ mod locations_test; mod plugin_test; #[path = "suite/resolver_edge_test.rs"] mod resolver_edge_test; +#[path = "suite/resolver_hooks_test.rs"] +mod resolver_hooks_test; #[path = "suite/resolver_test.rs"] mod resolver_test; #[path = "suite/settings_routing_test.rs"] diff --git a/crates/loopal-config/tests/suite/hook_test.rs b/crates/loopal-config/tests/suite/hook_test.rs index 32a8d8e7..60fd7df2 100644 --- a/crates/loopal-config/tests/suite/hook_test.rs +++ b/crates/loopal-config/tests/suite/hook_test.rs @@ -135,6 +135,13 @@ fn test_hook_config_serde_roundtrip() { command: "cargo fmt".to_string(), tool_filter: Some(vec!["Write".to_string()]), timeout_ms: 5000, + hook_type: Default::default(), + url: None, + headers: Default::default(), + prompt: None, + model: None, + condition: None, + id: None, }; let json = serde_json::to_string(&config).unwrap(); let deserialized: HookConfig = serde_json::from_str(&json).unwrap(); diff --git a/crates/loopal-config/tests/suite/resolver_edge_test.rs b/crates/loopal-config/tests/suite/resolver_edge_test.rs index d63d6358..7908dd7f 100644 --- a/crates/loopal-config/tests/suite/resolver_edge_test.rs +++ b/crates/loopal-config/tests/suite/resolver_edge_test.rs @@ -1,6 +1,5 @@ use std::collections::HashMap; -use loopal_config::hook::{HookConfig, HookEvent}; use loopal_config::layer::{ConfigLayer, LayerSource}; use loopal_config::resolver::ConfigResolver; use loopal_config::settings::McpServerConfig; @@ -55,27 +54,6 @@ fn test_resolve_mcp_written_back_to_settings() { assert_eq!(command, "test-cmd"); } -#[test] -fn test_resolve_hooks_written_back_to_settings() { - let mut resolver = ConfigResolver::new(); - let hook = HookConfig { - event: HookEvent::PreToolUse, - command: "echo test".into(), - tool_filter: None, - timeout_ms: 10_000, - }; - let mut layer = ConfigLayer { - source: LayerSource::Global, - ..Default::default() - }; - layer.hooks = vec![hook]; - resolver.add_layer(layer); - let config = resolver.resolve().unwrap(); - assert_eq!(config.hooks.len(), 1); - assert_eq!(config.settings.hooks.len(), 1); - assert_eq!(config.settings.hooks[0].command, "echo test"); -} - #[test] fn test_resolve_layers_tracked() { let mut resolver = ConfigResolver::new(); diff --git a/crates/loopal-config/tests/suite/resolver_hooks_test.rs b/crates/loopal-config/tests/suite/resolver_hooks_test.rs new file mode 100644 index 00000000..88f5066e --- /dev/null +++ b/crates/loopal-config/tests/suite/resolver_hooks_test.rs @@ -0,0 +1,106 @@ +//! Resolver tests specific to hook merge/dedup semantics. + +use loopal_config::hook::{HookConfig, HookEvent}; +use loopal_config::layer::{ConfigLayer, LayerSource}; +use loopal_config::resolver::ConfigResolver; + +fn make_hook(event: HookEvent, command: &str, id: Option<&str>) -> HookConfig { + HookConfig { + event, + command: command.into(), + tool_filter: None, + timeout_ms: 10_000, + hook_type: Default::default(), + url: None, + headers: Default::default(), + prompt: None, + model: None, + condition: None, + id: id.map(String::from), + } +} + +#[test] +fn test_resolve_hooks_append_all() { + let mut resolver = ConfigResolver::new(); + + let hook1 = make_hook(HookEvent::PreToolUse, "echo global", None); + let hook2 = make_hook(HookEvent::PostToolUse, "echo project", None); + + let mut layer1 = ConfigLayer { + source: LayerSource::Global, + ..Default::default() + }; + layer1.hooks = vec![hook1]; + let mut layer2 = ConfigLayer { + source: LayerSource::Project, + ..Default::default() + }; + layer2.hooks = vec![hook2]; + + resolver.add_layer(layer1); + resolver.add_layer(layer2); + + let config = resolver.resolve().unwrap(); + assert_eq!(config.hooks.len(), 2); + assert_eq!(config.hooks[0].config.command, "echo global"); + assert_eq!(config.hooks[1].config.command, "echo project"); +} + +#[test] +fn test_resolve_hooks_dedup_by_id_higher_priority_wins() { + let mut resolver = ConfigResolver::new(); + + let mut global = ConfigLayer { + source: LayerSource::Global, + ..Default::default() + }; + global.hooks = vec![ + make_hook(HookEvent::PreToolUse, "echo global-lint", Some("lint")), + make_hook(HookEvent::PostToolUse, "echo global-log", None), + ]; + + let mut project = ConfigLayer { + source: LayerSource::Project, + ..Default::default() + }; + project.hooks = vec![make_hook( + HookEvent::PreToolUse, + "echo project-lint", + Some("lint"), + )]; + + resolver.add_layer(global); + resolver.add_layer(project); + + let config = resolver.resolve().unwrap(); + // id="lint" deduped: project wins over global. + // id=None hook appended. + assert_eq!(config.hooks.len(), 2); + let lint = config + .hooks + .iter() + .find(|h| h.config.id.as_deref() == Some("lint")) + .unwrap(); + assert_eq!(lint.config.command, "echo project-lint"); + assert_eq!(lint.source, LayerSource::Project); + // No-id hook preserved. + let log = config.hooks.iter().find(|h| h.config.id.is_none()).unwrap(); + assert_eq!(log.config.command, "echo global-log"); +} + +#[test] +fn test_resolve_hooks_written_back_to_settings() { + let mut resolver = ConfigResolver::new(); + let hook = make_hook(HookEvent::PreToolUse, "echo test", None); + let mut layer = ConfigLayer { + source: LayerSource::Global, + ..Default::default() + }; + layer.hooks = vec![hook]; + resolver.add_layer(layer); + let config = resolver.resolve().unwrap(); + assert_eq!(config.hooks.len(), 1); + assert_eq!(config.settings.hooks.len(), 1); + assert_eq!(config.settings.hooks[0].command, "echo test"); +} diff --git a/crates/loopal-config/tests/suite/resolver_test.rs b/crates/loopal-config/tests/suite/resolver_test.rs index caaebbab..04a46d9c 100644 --- a/crates/loopal-config/tests/suite/resolver_test.rs +++ b/crates/loopal-config/tests/suite/resolver_test.rs @@ -1,6 +1,5 @@ use std::collections::HashMap; -use loopal_config::hook::{HookConfig, HookEvent}; use loopal_config::layer::{ConfigLayer, LayerSource}; use loopal_config::resolver::ConfigResolver; use loopal_config::settings::McpServerConfig; @@ -152,43 +151,6 @@ fn test_resolve_skills_override_by_name() { assert_eq!(config.skills["/commit"].source, LayerSource::Project); } -#[test] -fn test_resolve_hooks_append_all() { - let mut resolver = ConfigResolver::new(); - - let hook1 = HookConfig { - event: HookEvent::PreToolUse, - command: "echo global".into(), - tool_filter: None, - timeout_ms: 10_000, - }; - let hook2 = HookConfig { - event: HookEvent::PostToolUse, - command: "echo project".into(), - tool_filter: None, - timeout_ms: 5_000, - }; - - let mut layer1 = ConfigLayer { - source: LayerSource::Global, - ..Default::default() - }; - layer1.hooks = vec![hook1]; - let mut layer2 = ConfigLayer { - source: LayerSource::Project, - ..Default::default() - }; - layer2.hooks = vec![hook2]; - - resolver.add_layer(layer1); - resolver.add_layer(layer2); - - let config = resolver.resolve().unwrap(); - assert_eq!(config.hooks.len(), 2); - assert_eq!(config.hooks[0].config.command, "echo global"); - assert_eq!(config.hooks[1].config.command, "echo project"); -} - #[test] fn test_resolve_instructions_concatenated() { let mut resolver = ConfigResolver::new(); diff --git a/crates/loopal-hooks/BUILD.bazel b/crates/loopal-hooks/BUILD.bazel index 265d4e49..bc48a4fe 100644 --- a/crates/loopal-hooks/BUILD.bazel +++ b/crates/loopal-hooks/BUILD.bazel @@ -8,12 +8,20 @@ rust_library( deps = [ "//crates/loopal-config", "//crates/loopal-error", + "//crates/loopal-message", + "//crates/loopal-protocol", + "//crates/loopal-provider-api", + "@crates//:futures", + "@crates//:reqwest", "@crates//:serde", "@crates//:serde_json", "@crates//:thiserror", "@crates//:tokio", "@crates//:tracing", ], + proc_macro_deps = [ + "@crates//:async-trait", + ], ) rust_test( @@ -23,8 +31,16 @@ rust_test( edition = "2024", deps = [ ":loopal-hooks", + "//crates/loopal-config", + "//crates/loopal-error", + "//crates/loopal-kernel", + "//crates/loopal-provider-api", + "//crates/loopal-test-support", + "@crates//:futures", "@crates//:serde_json", "@crates//:tokio", - "//crates/loopal-config", + ], + proc_macro_deps = [ + "@crates//:async-trait", ], ) diff --git a/crates/loopal-hooks/src/async_store.rs b/crates/loopal-hooks/src/async_store.rs new file mode 100644 index 00000000..9f65f4e0 --- /dev/null +++ b/crates/loopal-hooks/src/async_store.rs @@ -0,0 +1,74 @@ +//! Async hook store — manages background hook tasks and rewake signaling. +//! +//! When an async hook completes with `rewake: true`, it sends an Envelope +//! to the agent loop via `rewake_tx`, waking the idle agent. This mirrors +//! the scheduler's `ScheduledTrigger → Envelope → select!` pattern. + +use std::sync::{Arc, Mutex}; + +use loopal_protocol::{Envelope, MessageSource, UserContent}; +use tokio::sync::mpsc; +use tokio::task::JoinHandle; + +use crate::executor::ExecutorFactory; +use crate::output::interpret_output; + +/// Tracks in-flight async hook tasks. Analogous to `BackgroundTaskStore`. +pub struct AsyncHookStore { + rewake_tx: mpsc::Sender, + tasks: Mutex>>, +} + +impl AsyncHookStore { + pub fn new(rewake_tx: mpsc::Sender) -> Self { + Self { + rewake_tx, + tasks: Mutex::new(Vec::new()), + } + } + + /// Spawn an async hook task. The task runs in the background; + /// if exit code 2 or `rewake: true`, it sends an Envelope to wake the agent. + pub fn spawn( + &self, + config: &loopal_config::HookConfig, + input: serde_json::Value, + factory: &Arc, + ) { + let Some(executor) = factory.create(config) else { + return; // Misconfigured hook, already logged by factory. + }; + let tx = self.rewake_tx.clone(); + let handle = tokio::spawn(async move { + match executor.execute(input).await { + Ok(raw) => { + let output = interpret_output(&raw); + if output.rewake { + let content = output.additional_context.unwrap_or_default(); + let env = Envelope::new( + MessageSource::System("hook".into()), + "self", + UserContent::text_only(content), + ); + let _ = tx.send(env).await; + } + } + Err(e) => { + tracing::warn!(error = %e, "async hook failed"); + } + } + }); + if let Ok(mut tasks) = self.tasks.lock() { + // Housekeeping: remove completed tasks to prevent unbounded growth. + tasks.retain(|h| !h.is_finished()); + tasks.push(handle); + } + } + + /// Clean up completed tasks (housekeeping, not required for correctness). + pub fn cleanup_completed(&self) { + if let Ok(mut tasks) = self.tasks.lock() { + tasks.retain(|h| !h.is_finished()); + } + } +} diff --git a/crates/loopal-hooks/src/executor.rs b/crates/loopal-hooks/src/executor.rs new file mode 100644 index 00000000..171e8ebe --- /dev/null +++ b/crates/loopal-hooks/src/executor.rs @@ -0,0 +1,42 @@ +//! Hook executor abstraction — the polymorphism point for execution strategies. +//! +//! New executor types (Command, Http, Prompt) implement `HookExecutor` +//! without modifying existing code (OCP). Callers depend on the trait, +//! not concrete types (DIP). + +use loopal_error::HookError; + +/// Raw output from any hook executor, before interpretation. +/// +/// Maps directly to the exit-code protocol: +/// - exit 0: success +/// - exit 2: feedback injection / rewake +/// - other: non-blocking error +#[derive(Debug, Clone)] +pub struct RawHookOutput { + /// Process exit code (or HTTP-status-derived equivalent). + pub exit_code: i32, + /// Primary output (stdout for command, body for HTTP, LLM text for prompt). + pub stdout: String, + /// Diagnostic output (stderr for command; empty for HTTP/prompt). + pub stderr: String, +} + +/// Trait for hook execution strategies. +/// +/// Each implementation handles one transport (shell, HTTP, LLM prompt). +/// The `HookService` dispatches to the correct executor via `ExecutorFactory`. +#[async_trait::async_trait] +pub trait HookExecutor: Send + Sync { + /// Execute the hook with the given JSON input payload. + async fn execute(&self, input: serde_json::Value) -> Result; +} + +/// Factory for creating executors from hook configuration. +/// +/// Implemented by Kernel (GRASP Creator: Kernel owns Provider for PromptExecutor). +pub trait ExecutorFactory: Send + Sync { + /// Create a boxed executor for the given hook configuration. + /// Returns None if the config is invalid (missing fields, unavailable provider). + fn create(&self, config: &loopal_config::HookConfig) -> Option>; +} diff --git a/crates/loopal-hooks/src/executor_command.rs b/crates/loopal-hooks/src/executor_command.rs new file mode 100644 index 00000000..dfca5a6e --- /dev/null +++ b/crates/loopal-hooks/src/executor_command.rs @@ -0,0 +1,62 @@ +//! Command hook executor — spawns a shell subprocess. +//! +//! Refactored from `runner.rs`. Implements `HookExecutor` trait (OCP). + +use std::time::Duration; + +use loopal_error::HookError; +use tokio::io::AsyncWriteExt; +use tokio::process::Command; +use tracing::debug; + +use crate::executor::{HookExecutor, RawHookOutput}; + +/// Executes a hook by spawning `sh -c ` and piping JSON to stdin. +pub struct CommandExecutor { + pub command: String, + pub timeout: Duration, +} + +#[async_trait::async_trait] +impl HookExecutor for CommandExecutor { + async fn execute(&self, input: serde_json::Value) -> Result { + debug!(command = %self.command, "running command hook"); + + let mut child = Command::new("sh") + .arg("-c") + .arg(&self.command) + .stdin(std::process::Stdio::piped()) + .stdout(std::process::Stdio::piped()) + .stderr(std::process::Stdio::piped()) + .spawn() + .map_err(|e| HookError::ExecutionFailed(e.to_string()))?; + + // Write JSON to stdin — ignore BrokenPipe (child may exit without reading). + if let Some(mut stdin) = child.stdin.take() { + let data = serde_json::to_vec(&input) + .map_err(|e| HookError::ExecutionFailed(e.to_string()))?; + if let Err(e) = stdin.write_all(&data).await + && e.kind() != std::io::ErrorKind::BrokenPipe + { + return Err(HookError::ExecutionFailed(e.to_string())); + } + drop(stdin); + } + + let output = tokio::time::timeout(self.timeout, child.wait_with_output()) + .await + .map_err(|_| { + HookError::Timeout(format!( + "hook timed out after {}ms", + self.timeout.as_millis() + )) + })? + .map_err(|e| HookError::ExecutionFailed(e.to_string()))?; + + Ok(RawHookOutput { + exit_code: output.status.code().unwrap_or(-1), + stdout: String::from_utf8_lossy(&output.stdout).into_owned(), + stderr: String::from_utf8_lossy(&output.stderr).into_owned(), + }) + } +} diff --git a/crates/loopal-hooks/src/executor_http.rs b/crates/loopal-hooks/src/executor_http.rs new file mode 100644 index 00000000..60432d2f --- /dev/null +++ b/crates/loopal-hooks/src/executor_http.rs @@ -0,0 +1,58 @@ +//! HTTP hook executor — POSTs JSON to a webhook endpoint. +//! +//! Uses reqwest to send hook input as JSON body. HTTP status maps to exit code: +//! - 2xx → exit 0 (success), response body as stdout +//! - 4xx/5xx → exit 1 (error), response body as stderr + +use std::time::Duration; + +use loopal_error::HookError; + +use crate::executor::{HookExecutor, RawHookOutput}; + +/// Executes a hook by POSTing JSON to a URL and interpreting the response. +pub struct HttpExecutor { + pub url: String, + pub headers: std::collections::HashMap, + pub timeout: Duration, +} + +#[async_trait::async_trait] +impl HookExecutor for HttpExecutor { + async fn execute(&self, input: serde_json::Value) -> Result { + let client = reqwest::Client::builder() + .timeout(self.timeout) + .build() + .map_err(|e| HookError::ExecutionFailed(e.to_string()))?; + + let mut req = client.post(&self.url).json(&input); + for (key, value) in &self.headers { + req = req.header(key.as_str(), value.as_str()); + } + + let response = req + .send() + .await + .map_err(|e| HookError::ExecutionFailed(format!("HTTP request failed: {e}")))?; + + let status = response.status(); + let body = response + .text() + .await + .map_err(|e| HookError::ExecutionFailed(format!("HTTP body read failed: {e}")))?; + + if status.is_success() { + Ok(RawHookOutput { + exit_code: 0, + stdout: body, + stderr: String::new(), + }) + } else { + Ok(RawHookOutput { + exit_code: 1, + stdout: String::new(), + stderr: body, + }) + } + } +} diff --git a/crates/loopal-hooks/src/executor_prompt.rs b/crates/loopal-hooks/src/executor_prompt.rs new file mode 100644 index 00000000..b337ea97 --- /dev/null +++ b/crates/loopal-hooks/src/executor_prompt.rs @@ -0,0 +1,83 @@ +//! Prompt hook executor — uses a lightweight LLM call as hook logic. +//! +//! Reuses the same pattern as `loopal-auto-mode/src/llm_call.rs`: +//! small max_tokens, temperature 0, no tools. The hook's `prompt` field +//! becomes the system prompt; the hook input JSON becomes the user message. + +use std::sync::Arc; +use std::time::Duration; + +use futures::StreamExt; +use loopal_error::HookError; +use loopal_provider_api::{ChatParams, Provider, StreamChunk}; + +use crate::executor::{HookExecutor, RawHookOutput}; + +/// Executes a hook by calling an LLM with a prompt. +pub struct PromptExecutor { + pub system_prompt: String, + pub model: String, + pub provider: Arc, + pub timeout: Duration, + pub max_tokens: u32, +} + +#[async_trait::async_trait] +impl HookExecutor for PromptExecutor { + async fn execute(&self, input: serde_json::Value) -> Result { + let user_msg = serde_json::to_string_pretty(&input).unwrap_or_else(|_| input.to_string()); + + let params = ChatParams { + model: self.model.clone(), + messages: vec![loopal_message::Message::user(&user_msg)], + system_prompt: self.system_prompt.clone(), + tools: vec![], + max_tokens: self.max_tokens, + temperature: Some(0.0), + thinking: None, + debug_dump_dir: None, + }; + + let text = tokio::time::timeout(self.timeout, self.stream_text(¶ms)) + .await + .map_err(|_| { + HookError::Timeout(format!( + "prompt hook timed out after {}ms", + self.timeout.as_millis() + )) + })??; + + // Try to extract exit_code from JSON response, default to 0. + let exit_code = serde_json::from_str::(&text) + .ok() + .and_then(|v| v.get("exit_code").and_then(|c| c.as_i64())) + .unwrap_or(0) as i32; + + Ok(RawHookOutput { + exit_code, + stdout: text, + stderr: String::new(), + }) + } +} + +impl PromptExecutor { + async fn stream_text(&self, params: &ChatParams) -> Result { + let mut stream = self + .provider + .stream_chat(params) + .await + .map_err(|e| HookError::ExecutionFailed(e.to_string()))?; + + let mut text = String::new(); + while let Some(chunk) = stream.next().await { + match chunk { + Ok(StreamChunk::Text { text: t }) => text.push_str(&t), + Ok(StreamChunk::Done { .. }) => break, + Err(e) => return Err(HookError::ExecutionFailed(e.to_string())), + _ => {} // ignore thinking, usage, etc. + } + } + Ok(text) + } +} diff --git a/crates/loopal-hooks/src/input.rs b/crates/loopal-hooks/src/input.rs new file mode 100644 index 00000000..43bcf30b --- /dev/null +++ b/crates/loopal-hooks/src/input.rs @@ -0,0 +1,66 @@ +//! Hook input construction — builds the JSON payload for each event. +//! +//! Information Expert: this module knows what data each event needs, +//! so the payload construction logic lives here. + +use loopal_config::HookEvent; +use serde_json::json; + +/// Contextual data available to hooks. Not all fields are populated for +/// every event — `build_hook_input` selects what's relevant. +#[derive(Debug, Default)] +pub struct HookContext<'a> { + // Tool events + pub tool_name: Option<&'a str>, + pub tool_input: Option<&'a serde_json::Value>, + pub tool_output: Option<&'a str>, + pub is_error: Option, + // Session events + pub session_id: Option<&'a str>, + pub cwd: Option<&'a str>, + // Stop event + pub stop_reason: Option<&'a str>, +} + +/// Build the JSON payload sent to a hook's stdin based on the event type. +pub fn build_hook_input(event: HookEvent, ctx: &HookContext<'_>) -> serde_json::Value { + match event { + HookEvent::PreToolUse => json!({ + "event": "pre_tool_use", + "tool_name": ctx.tool_name, + "tool_input": ctx.tool_input, + }), + HookEvent::PostToolUse => json!({ + "event": "post_tool_use", + "tool_name": ctx.tool_name, + "tool_input": ctx.tool_input, + "tool_output": ctx.tool_output, + "is_error": ctx.is_error, + }), + HookEvent::PreRequest => json!({ + "event": "pre_request", + "session_id": ctx.session_id, + }), + HookEvent::PostInput => json!({ + "event": "post_input", + "session_id": ctx.session_id, + }), + HookEvent::SessionStart => json!({ + "event": "session_start", + "session_id": ctx.session_id, + "cwd": ctx.cwd, + }), + HookEvent::SessionEnd => json!({ + "event": "session_end", + "session_id": ctx.session_id, + }), + HookEvent::Stop => json!({ + "event": "stop", + "reason": ctx.stop_reason, + }), + HookEvent::PreCompact => json!({ + "event": "pre_compact", + "session_id": ctx.session_id, + }), + } +} diff --git a/crates/loopal-hooks/src/lib.rs b/crates/loopal-hooks/src/lib.rs index 14817cfe..aaf24d72 100644 --- a/crates/loopal-hooks/src/lib.rs +++ b/crates/loopal-hooks/src/lib.rs @@ -1,5 +1,20 @@ +pub mod async_store; +pub mod executor; +pub mod executor_command; +pub mod executor_http; +pub mod executor_prompt; +pub mod input; +pub mod output; pub mod registry; pub mod runner; +pub mod service; +pub use executor::{ExecutorFactory, HookExecutor, RawHookOutput}; +pub use executor_command::CommandExecutor; +pub use executor_http::HttpExecutor; +pub use executor_prompt::PromptExecutor; +pub use input::{HookContext, build_hook_input}; +pub use output::{HookOutput, PermissionOverride, interpret_output, interpret_pre_tool_output}; pub use registry::HookRegistry; pub use runner::run_hook; +pub use service::HookService; diff --git a/crates/loopal-hooks/src/output.rs b/crates/loopal-hooks/src/output.rs new file mode 100644 index 00000000..6937ba63 --- /dev/null +++ b/crates/loopal-hooks/src/output.rs @@ -0,0 +1,205 @@ +//! Structured hook output — typed interpretation of `RawHookOutput`. +//! +//! The `interpret_output` function converts raw exit-code + stdout into +//! a typed `HookOutput`. Consumers read only the fields relevant to +//! their event (ISP). Backward compatible with exit-code 0/2 protocol. + +use serde::Deserialize; + +use crate::executor::RawHookOutput; + +/// Interpreted hook output. Each field is optional — consumers only +/// read what their event requires (ISP). +#[derive(Debug, Clone, Default, Deserialize)] +#[serde(default)] +pub struct HookOutput { + /// PreToolUse: override permission decision. + pub permission: Option, + /// Text to inject into the conversation (tool result or LLM context). + pub additional_context: Option, + /// PreToolUse: replace tool input parameters. + pub updated_input: Option, + /// Signal to wake an idle agent (used by asyncRewake hooks). + #[serde(default)] + pub rewake: bool, + /// Suppress the default behavior for this event. + #[serde(default)] + pub suppress: bool, +} + +/// Permission override returned by PreToolUse hooks. +#[derive(Debug, Clone, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum PermissionOverride { + Allow, + Deny { reason: String }, +} + +/// Interpret raw executor output into typed `HookOutput`. +/// +/// **Strategy:** Try JSON parse first (structured output). On parse failure, +/// fall back to the exit-code protocol (backward compatible). +/// +/// Information Expert: this function knows both the JSON schema and the +/// exit-code protocol, so interpretation logic lives here. +pub fn interpret_output(raw: &RawHookOutput) -> HookOutput { + // Structured path: try to parse stdout as JSON + if !raw.stdout.is_empty() + && let Ok(parsed) = serde_json::from_str::(&raw.stdout) + { + return parsed; + } + // Fallback: exit-code protocol + match raw.exit_code { + 0 => HookOutput::default(), + 2 => { + let text = if raw.stdout.is_empty() { + raw.stderr.clone() + } else { + raw.stdout.clone() + }; + HookOutput { + additional_context: if text.is_empty() { None } else { Some(text) }, + rewake: true, + ..Default::default() + } + } + _ => HookOutput::default(), // non-zero non-2: logged by caller, no action + } +} + +/// Interpret raw output for PreToolUse hooks (backward-compat: non-zero = deny). +/// +/// PreToolUse hooks historically treated any non-zero exit code as rejection. +/// This variant preserves that behavior while adding structured JSON support. +pub fn interpret_pre_tool_output(raw: &RawHookOutput) -> HookOutput { + // Structured path: try JSON first + if !raw.stdout.is_empty() + && let Ok(parsed) = serde_json::from_str::(&raw.stdout) + { + return parsed; + } + match raw.exit_code { + 0 => HookOutput::default(), + _ => HookOutput { + permission: Some(PermissionOverride::Deny { + reason: if raw.stderr.is_empty() { + format!("hook exited with code {}", raw.exit_code) + } else { + raw.stderr.trim().to_string() + }, + }), + ..Default::default() + }, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn json_output_parsed() { + let raw = RawHookOutput { + exit_code: 0, + stdout: r#"{"additional_context":"lint passed","rewake":false}"#.into(), + stderr: String::new(), + }; + let out = interpret_output(&raw); + assert_eq!(out.additional_context.as_deref(), Some("lint passed")); + assert!(!out.rewake); + } + + #[test] + fn exit_code_2_fallback() { + let raw = RawHookOutput { + exit_code: 2, + stdout: "error: type mismatch".into(), + stderr: String::new(), + }; + let out = interpret_output(&raw); + assert_eq!( + out.additional_context.as_deref(), + Some("error: type mismatch") + ); + assert!(out.rewake); + } + + #[test] + fn exit_code_0_noop() { + let raw = RawHookOutput { + exit_code: 0, + stdout: "not json".into(), + stderr: String::new(), + }; + let out = interpret_output(&raw); + assert!(out.additional_context.is_none()); + assert!(out.permission.is_none()); + } + + #[test] + fn exit_code_1_noop() { + let raw = RawHookOutput { + exit_code: 1, + stdout: String::new(), + stderr: "hook crashed".into(), + }; + let out = interpret_output(&raw); + assert!(out.additional_context.is_none()); + } + + // ── interpret_pre_tool_output tests ───────────────────── + + #[test] + fn pre_tool_nonzero_exit_denies() { + let raw = RawHookOutput { + exit_code: 1, + stdout: String::new(), + stderr: "denied by hook".into(), + }; + let out = interpret_pre_tool_output(&raw); + assert!(matches!( + out.permission, + Some(PermissionOverride::Deny { .. }) + )); + if let Some(PermissionOverride::Deny { reason }) = out.permission { + assert!(reason.contains("denied by hook")); + } + } + + #[test] + fn pre_tool_exit_0_allows() { + let raw = RawHookOutput { + exit_code: 0, + stdout: String::new(), + stderr: String::new(), + }; + let out = interpret_pre_tool_output(&raw); + assert!(out.permission.is_none()); + } + + #[test] + fn pre_tool_json_structured_deny() { + let raw = RawHookOutput { + exit_code: 0, + stdout: r#"{"permission":{"deny":{"reason":"policy violation"}}}"#.into(), + stderr: String::new(), + }; + let out = interpret_pre_tool_output(&raw); + assert!(matches!( + out.permission, + Some(PermissionOverride::Deny { .. }) + )); + } + + #[test] + fn pre_tool_json_structured_allow() { + let raw = RawHookOutput { + exit_code: 0, + stdout: r#"{"permission":"allow"}"#.into(), + stderr: String::new(), + }; + let out = interpret_pre_tool_output(&raw); + assert!(matches!(out.permission, Some(PermissionOverride::Allow))); + } +} diff --git a/crates/loopal-hooks/src/registry.rs b/crates/loopal-hooks/src/registry.rs index 679580a7..1b0fe75f 100644 --- a/crates/loopal-hooks/src/registry.rs +++ b/crates/loopal-hooks/src/registry.rs @@ -1,3 +1,4 @@ +use loopal_config::hook_condition::matches_condition; use loopal_config::{HookConfig, HookEvent}; /// Registry holding hook configurations and matching logic. @@ -10,23 +11,41 @@ impl HookRegistry { Self { hooks } } - /// Return hooks matching the given event and optional tool name. - pub fn match_hooks(&self, event: HookEvent, tool_name: Option<&str>) -> Vec<&HookConfig> { + /// Return hooks matching the given event, optional tool name, and optional input. + /// + /// Matching priority: `condition` field > `tool_filter` field > match all. + /// Pass `tool_input` for condition expressions with globs (e.g. `"Bash(git push*)"`) + /// to work correctly. + pub fn match_hooks( + &self, + event: HookEvent, + tool_name: Option<&str>, + tool_input: Option<&serde_json::Value>, + ) -> Vec<&HookConfig> { + let null = serde_json::Value::Null; + let input = tool_input.unwrap_or(&null); self.hooks .iter() .filter(|hook| { if hook.event != event { return false; } - // If hook has a tool_filter, the tool_name must match one of them + // Priority 1: condition expression (new) + if let Some(ref cond) = hook.condition { + return match tool_name { + Some(name) => matches_condition(cond, name, input), + None => cond == "*", + }; + } + // Priority 2: legacy tool_filter if let Some(ref filters) = hook.tool_filter { - match tool_name { + return match tool_name { Some(name) => filters.iter().any(|f| f == name), None => false, - } - } else { - true + }; } + // No filter: match all + true }) .collect() } diff --git a/crates/loopal-hooks/src/service.rs b/crates/loopal-hooks/src/service.rs new file mode 100644 index 00000000..471e8189 --- /dev/null +++ b/crates/loopal-hooks/src/service.rs @@ -0,0 +1,72 @@ +//! Hook service — single entry point for hook orchestration. +//! +//! SRP: coordinates matching, execution, and output interpretation. +//! Consumers call `run_hooks()` and get back typed `HookOutput`s. + +use std::sync::Arc; + +use loopal_config::HookEvent; +use tracing::warn; + +use crate::executor::ExecutorFactory; +use crate::input::{HookContext, build_hook_input}; +use crate::output::{HookOutput, interpret_output, interpret_pre_tool_output}; +use crate::registry::HookRegistry; + +/// Central hook orchestration service. +/// +/// Replaces the previous pattern of "registry.match + for loop + run_hook" +/// scattered across call sites. Now there's one entry point. +pub struct HookService { + registry: HookRegistry, + factory: Arc, +} + +impl HookService { + pub fn new(registry: HookRegistry, factory: Arc) -> Self { + Self { registry, factory } + } + + /// Run all matching hooks for an event. Returns aggregated outputs. + /// + /// Async hooks will be handled in a future phase (AsyncHookStore). + /// Currently all hooks execute synchronously in sequence. + pub async fn run_hooks(&self, event: HookEvent, context: &HookContext<'_>) -> Vec { + let matched = self + .registry + .match_hooks(event, context.tool_name, context.tool_input); + if matched.is_empty() { + return Vec::new(); + } + + let input = build_hook_input(event, context); + let mut outputs = Vec::new(); + + for config in matched { + let Some(executor) = self.factory.create(config) else { + continue; // Misconfigured hook, already logged by factory. + }; + match executor.execute(input.clone()).await { + Ok(raw) => { + // PreToolUse: non-zero exit = deny (backward compat) + let out = if event == HookEvent::PreToolUse { + interpret_pre_tool_output(&raw) + } else { + interpret_output(&raw) + }; + outputs.push(out); + } + Err(e) => { + warn!(event = ?event, error = %e, "hook execution failed"); + } + } + } + outputs + } + + /// Access the underlying registry (for backward-compat call sites + /// that still need direct matching). + pub fn registry(&self) -> &HookRegistry { + &self.registry + } +} diff --git a/crates/loopal-hooks/tests/suite.rs b/crates/loopal-hooks/tests/suite.rs index 980966ba..5bb5b6a0 100644 --- a/crates/loopal-hooks/tests/suite.rs +++ b/crates/loopal-hooks/tests/suite.rs @@ -1,4 +1,10 @@ // Single test binary — includes all test modules +#[path = "suite/executor_factory_test.rs"] +mod executor_factory_test; +#[path = "suite/executor_http_test.rs"] +mod executor_http_test; +#[path = "suite/executor_prompt_test.rs"] +mod executor_prompt_test; #[path = "suite/registry_test.rs"] mod registry_test; #[path = "suite/runner_test.rs"] diff --git a/crates/loopal-hooks/tests/suite/executor_factory_test.rs b/crates/loopal-hooks/tests/suite/executor_factory_test.rs new file mode 100644 index 00000000..236b3241 --- /dev/null +++ b/crates/loopal-hooks/tests/suite/executor_factory_test.rs @@ -0,0 +1,75 @@ +//! Tests for DefaultExecutorFactory dispatch logic. + +use std::sync::Arc; + +use loopal_config::{HookConfig, HookEvent, HookType}; +use loopal_hooks::executor::ExecutorFactory; +use loopal_kernel::hook_factory::DefaultExecutorFactory; + +fn make_config(hook_type: HookType) -> HookConfig { + HookConfig { + event: HookEvent::PostToolUse, + hook_type, + command: "echo test".into(), + url: Some("http://localhost:9999".into()), + headers: Default::default(), + prompt: Some("test prompt".into()), + model: None, + tool_filter: None, + condition: None, + timeout_ms: 5000, + id: None, + } +} + +#[tokio::test] +async fn test_factory_creates_command_executor() { + let factory = DefaultExecutorFactory::new(None); + let executor = factory.create(&make_config(HookType::Command)).unwrap(); + let result = executor.execute(serde_json::json!({})).await; + assert!(result.is_ok()); + assert_eq!(result.unwrap().exit_code, 0); +} + +#[tokio::test] +async fn test_factory_creates_http_executor() { + let factory = DefaultExecutorFactory::new(None); + let config = make_config(HookType::Http); + let executor = factory.create(&config).unwrap(); + let result = executor.execute(serde_json::json!({})).await; + // Connection to localhost:9999 should fail (either refused or timeout) + assert!(result.is_err() || result.as_ref().is_ok_and(|r| r.exit_code != 0)); +} + +#[tokio::test] +async fn test_factory_prompt_without_provider_returns_none() { + let factory = DefaultExecutorFactory::new(None); + assert!(factory.create(&make_config(HookType::Prompt)).is_none()); +} + +#[tokio::test] +async fn test_factory_prompt_with_provider_creates_executor() { + use loopal_provider_api::StreamChunk; + use loopal_test_support::mock_provider::MockProvider; + + let chunks = vec![ + Ok(StreamChunk::Text { text: "ok".into() }), + Ok(StreamChunk::Done { + stop_reason: loopal_provider_api::StopReason::EndTurn, + }), + ]; + let provider: Arc = Arc::new(MockProvider::new(chunks)); + let factory = DefaultExecutorFactory::new(Some(provider)); + let executor = factory.create(&make_config(HookType::Prompt)).unwrap(); + let result = executor.execute(serde_json::json!({})).await; + assert!(result.is_ok()); + assert_eq!(result.unwrap().exit_code, 0); +} + +#[tokio::test] +async fn test_factory_http_missing_url_returns_none() { + let factory = DefaultExecutorFactory::new(None); + let mut config = make_config(HookType::Http); + config.url = None; + assert!(factory.create(&config).is_none()); +} diff --git a/crates/loopal-hooks/tests/suite/executor_http_test.rs b/crates/loopal-hooks/tests/suite/executor_http_test.rs new file mode 100644 index 00000000..bfd911ca --- /dev/null +++ b/crates/loopal-hooks/tests/suite/executor_http_test.rs @@ -0,0 +1,73 @@ +//! Tests for HttpExecutor using a local TCP server. + +use std::time::Duration; + +use loopal_hooks::executor::HookExecutor; +use loopal_hooks::executor_http::HttpExecutor; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpListener; + +/// Spawn a minimal HTTP server that returns the given status and body. +async fn spawn_http_server(status: u16, body: &str) -> String { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let body = body.to_string(); + tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.unwrap(); + let mut buf = vec![0u8; 4096]; + let _ = stream.read(&mut buf).await; // consume request + let response = format!( + "HTTP/1.1 {status} OK\r\nContent-Length: {}\r\n\r\n{body}", + body.len() + ); + let _ = stream.write_all(response.as_bytes()).await; + }); + format!("http://{addr}") +} + +#[tokio::test] +async fn test_http_executor_success_200() { + let url = spawn_http_server(200, r#"{"additional_context":"ok"}"#).await; + let exec = HttpExecutor { + url, + headers: Default::default(), + timeout: Duration::from_secs(5), + }; + let result = exec.execute(serde_json::json!({"tool": "Write"})).await; + let output = result.unwrap(); + assert_eq!(output.exit_code, 0); + assert!(output.stdout.contains("additional_context")); +} + +#[tokio::test] +async fn test_http_executor_error_500() { + let url = spawn_http_server(500, "internal error").await; + let exec = HttpExecutor { + url, + headers: Default::default(), + timeout: Duration::from_secs(5), + }; + let result = exec.execute(serde_json::json!({})).await; + let output = result.unwrap(); + assert_eq!(output.exit_code, 1); + assert!(output.stderr.contains("internal error")); +} + +#[tokio::test] +async fn test_http_executor_timeout() { + // Server never responds + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + let (mut _stream, _) = listener.accept().await.unwrap(); + tokio::time::sleep(Duration::from_secs(60)).await; // hang + }); + + let exec = HttpExecutor { + url: format!("http://{addr}"), + headers: Default::default(), + timeout: Duration::from_millis(100), // very short + }; + let result = exec.execute(serde_json::json!({})).await; + assert!(result.is_err()); +} diff --git a/crates/loopal-hooks/tests/suite/executor_prompt_test.rs b/crates/loopal-hooks/tests/suite/executor_prompt_test.rs new file mode 100644 index 00000000..25472e36 --- /dev/null +++ b/crates/loopal-hooks/tests/suite/executor_prompt_test.rs @@ -0,0 +1,53 @@ +//! Tests for PromptExecutor using MockProvider. + +use std::sync::Arc; +use std::time::Duration; + +use loopal_hooks::executor::HookExecutor; +use loopal_hooks::executor_prompt::PromptExecutor; +use loopal_provider_api::StreamChunk; +use loopal_test_support::mock_provider::MockProvider; + +fn make_prompt_executor(response_text: &str) -> PromptExecutor { + let chunks = vec![ + Ok(StreamChunk::Text { + text: response_text.into(), + }), + Ok(StreamChunk::Done { + stop_reason: loopal_provider_api::StopReason::EndTurn, + }), + ]; + let provider = Arc::new(MockProvider::new(chunks)); + PromptExecutor { + system_prompt: "You are a hook.".into(), + model: "test-model".into(), + provider, + timeout: Duration::from_secs(5), + max_tokens: 256, + } +} + +#[tokio::test] +async fn test_prompt_executor_success() { + let exec = make_prompt_executor("all good"); + let result = exec.execute(serde_json::json!({"tool": "Write"})).await; + let output = result.unwrap(); + assert_eq!(output.exit_code, 0); + assert!(output.stdout.contains("all good")); +} + +#[tokio::test] +async fn test_prompt_executor_json_exit_code() { + let exec = make_prompt_executor(r#"{"exit_code": 2, "reason": "blocked"}"#); + let result = exec.execute(serde_json::json!({})).await; + let output = result.unwrap(); + assert_eq!(output.exit_code, 2); +} + +#[tokio::test] +async fn test_prompt_executor_no_exit_code_defaults_to_0() { + let exec = make_prompt_executor(r#"{"feedback": "looks fine"}"#); + let result = exec.execute(serde_json::json!({})).await; + let output = result.unwrap(); + assert_eq!(output.exit_code, 0); +} diff --git a/crates/loopal-hooks/tests/suite/registry_test.rs b/crates/loopal-hooks/tests/suite/registry_test.rs index 49cff541..ed25dd47 100644 --- a/crates/loopal-hooks/tests/suite/registry_test.rs +++ b/crates/loopal-hooks/tests/suite/registry_test.rs @@ -7,6 +7,13 @@ fn make_hook(event: HookEvent, tool_filter: Option>) -> HookConfig { command: "echo test".into(), tool_filter, timeout_ms: 10_000, + hook_type: Default::default(), + url: None, + headers: Default::default(), + prompt: None, + model: None, + condition: None, + id: None, } } @@ -16,7 +23,7 @@ fn test_match_by_event() { make_hook(HookEvent::PreToolUse, None), make_hook(HookEvent::PostToolUse, None), ]); - let matched = reg.match_hooks(HookEvent::PreToolUse, None); + let matched = reg.match_hooks(HookEvent::PreToolUse, None, None); assert_eq!(matched.len(), 1); assert_eq!(matched[0].event, HookEvent::PreToolUse); } @@ -28,27 +35,32 @@ fn test_match_with_tool_filter() { Some(vec!["bash".into(), "write".into()]), )]); assert_eq!( - reg.match_hooks(HookEvent::PreToolUse, Some("bash")).len(), + reg.match_hooks(HookEvent::PreToolUse, Some("bash"), None) + .len(), 1 ); assert_eq!( - reg.match_hooks(HookEvent::PreToolUse, Some("read")).len(), + reg.match_hooks(HookEvent::PreToolUse, Some("read"), None) + .len(), 0 ); - assert_eq!(reg.match_hooks(HookEvent::PreToolUse, None).len(), 0); + assert_eq!(reg.match_hooks(HookEvent::PreToolUse, None, None).len(), 0); } #[test] fn test_no_match_wrong_event() { let reg = HookRegistry::new(vec![make_hook(HookEvent::PreToolUse, None)]); - assert!(reg.match_hooks(HookEvent::PostToolUse, None).is_empty()); + assert!( + reg.match_hooks(HookEvent::PostToolUse, None, None) + .is_empty() + ); } #[test] fn test_no_filter_matches_any_tool() { let reg = HookRegistry::new(vec![make_hook(HookEvent::PreToolUse, None)]); assert_eq!( - reg.match_hooks(HookEvent::PreToolUse, Some("anything")) + reg.match_hooks(HookEvent::PreToolUse, Some("anything"), None) .len(), 1 ); @@ -61,7 +73,7 @@ fn test_match_hooks_returns_all_matching_for_event() { make_hook(HookEvent::PreToolUse, None), make_hook(HookEvent::PostToolUse, None), ]); - let matched = reg.match_hooks(HookEvent::PreToolUse, None); + let matched = reg.match_hooks(HookEvent::PreToolUse, None, None); assert_eq!(matched.len(), 2); } @@ -75,28 +87,31 @@ fn test_match_hooks_with_tool_filter_only_matches_specified_tools() { ), ]); // "bash" matches only the first hook - let matched = reg.match_hooks(HookEvent::PreToolUse, Some("bash")); + let matched = reg.match_hooks(HookEvent::PreToolUse, Some("bash"), None); assert_eq!(matched.len(), 1); // "write" matches only the second hook - let matched = reg.match_hooks(HookEvent::PreToolUse, Some("write")); + let matched = reg.match_hooks(HookEvent::PreToolUse, Some("write"), None); assert_eq!(matched.len(), 1); // "edit" matches only the second hook - let matched = reg.match_hooks(HookEvent::PreToolUse, Some("edit")); + let matched = reg.match_hooks(HookEvent::PreToolUse, Some("edit"), None); assert_eq!(matched.len(), 1); // "unknown" matches neither - let matched = reg.match_hooks(HookEvent::PreToolUse, Some("unknown")); + let matched = reg.match_hooks(HookEvent::PreToolUse, Some("unknown"), None); assert_eq!(matched.len(), 0); } #[test] fn test_empty_registry_returns_empty() { let reg = HookRegistry::new(vec![]); - assert!(reg.match_hooks(HookEvent::PreToolUse, None).is_empty()); assert!( - reg.match_hooks(HookEvent::PreToolUse, Some("bash")) + reg.match_hooks(HookEvent::PreToolUse, None, None) + .is_empty() + ); + assert!( + reg.match_hooks(HookEvent::PreToolUse, Some("bash"), None) .is_empty() ); } @@ -109,14 +124,142 @@ fn test_mixed_filtered_and_unfiltered_hooks() { ]); // With tool_name "bash", both should match - let matched = reg.match_hooks(HookEvent::PreToolUse, Some("bash")); + let matched = reg.match_hooks(HookEvent::PreToolUse, Some("bash"), None); assert_eq!(matched.len(), 2); // With tool_name "read", only the unfiltered one matches - let matched = reg.match_hooks(HookEvent::PreToolUse, Some("read")); + let matched = reg.match_hooks(HookEvent::PreToolUse, Some("read"), None); assert_eq!(matched.len(), 1); // With None tool_name, only the unfiltered one matches (filtered requires a tool name) - let matched = reg.match_hooks(HookEvent::PreToolUse, None); + let matched = reg.match_hooks(HookEvent::PreToolUse, None, None); assert_eq!(matched.len(), 1); } + +// ── Condition expression integration tests ────────────────── + +fn make_condition_hook(event: HookEvent, condition: &str) -> HookConfig { + HookConfig { + event, + command: "echo test".into(), + tool_filter: None, + timeout_ms: 10_000, + condition: Some(condition.into()), + id: None, + hook_type: Default::default(), + url: None, + headers: Default::default(), + prompt: None, + model: None, + } +} + +#[test] +fn test_condition_wildcard_matches_any_tool() { + let reg = HookRegistry::new(vec![make_condition_hook(HookEvent::PreToolUse, "*")]); + assert_eq!( + reg.match_hooks(HookEvent::PreToolUse, Some("Bash"), None) + .len(), + 1 + ); + assert_eq!( + reg.match_hooks(HookEvent::PreToolUse, Some("Write"), None) + .len(), + 1 + ); +} + +#[test] +fn test_condition_exact_tool_name() { + let reg = HookRegistry::new(vec![make_condition_hook(HookEvent::PreToolUse, "Bash")]); + assert_eq!( + reg.match_hooks(HookEvent::PreToolUse, Some("Bash"), None) + .len(), + 1 + ); + assert_eq!( + reg.match_hooks(HookEvent::PreToolUse, Some("Write"), None) + .len(), + 0 + ); +} + +#[test] +fn test_condition_or_syntax() { + let reg = HookRegistry::new(vec![make_condition_hook( + HookEvent::PreToolUse, + "Bash|Write", + )]); + assert_eq!( + reg.match_hooks(HookEvent::PreToolUse, Some("Bash"), None) + .len(), + 1 + ); + assert_eq!( + reg.match_hooks(HookEvent::PreToolUse, Some("Write"), None) + .len(), + 1 + ); + assert_eq!( + reg.match_hooks(HookEvent::PreToolUse, Some("Read"), None) + .len(), + 0 + ); +} + +#[test] +fn test_condition_overrides_tool_filter() { + let mut hook = make_condition_hook(HookEvent::PreToolUse, "Read"); + hook.tool_filter = Some(vec!["Bash".into()]); + let reg = HookRegistry::new(vec![hook]); + assert_eq!( + reg.match_hooks(HookEvent::PreToolUse, Some("Bash"), None) + .len(), + 0 + ); + assert_eq!( + reg.match_hooks(HookEvent::PreToolUse, Some("Read"), None) + .len(), + 1 + ); +} + +#[test] +fn test_condition_glob_matches_with_tool_input() { + let reg = HookRegistry::new(vec![make_condition_hook( + HookEvent::PreToolUse, + "Bash(git push*)", + )]); + let push = serde_json::json!({"command": "git push origin main"}); + let pull = serde_json::json!({"command": "git pull origin main"}); + assert_eq!( + reg.match_hooks(HookEvent::PreToolUse, Some("Bash"), Some(&push)) + .len(), + 1 + ); + assert_eq!( + reg.match_hooks(HookEvent::PreToolUse, Some("Bash"), Some(&pull)) + .len(), + 0 + ); +} + +#[test] +fn test_condition_file_glob_matches_with_tool_input() { + let reg = HookRegistry::new(vec![make_condition_hook( + HookEvent::PreToolUse, + "Write(*.rs)", + )]); + let rs = serde_json::json!({"file_path": "src/main.rs"}); + let ts = serde_json::json!({"file_path": "src/main.ts"}); + assert_eq!( + reg.match_hooks(HookEvent::PreToolUse, Some("Write"), Some(&rs)) + .len(), + 1 + ); + assert_eq!( + reg.match_hooks(HookEvent::PreToolUse, Some("Write"), Some(&ts)) + .len(), + 0 + ); +} diff --git a/crates/loopal-hooks/tests/suite/runner_test.rs b/crates/loopal-hooks/tests/suite/runner_test.rs index 4586ddaa..28d7eab5 100644 --- a/crates/loopal-hooks/tests/suite/runner_test.rs +++ b/crates/loopal-hooks/tests/suite/runner_test.rs @@ -7,6 +7,13 @@ fn make_hook(command: &str, timeout_ms: u64) -> HookConfig { command: command.to_string(), tool_filter: None, timeout_ms, + hook_type: Default::default(), + url: None, + headers: Default::default(), + prompt: None, + model: None, + condition: None, + id: None, } } @@ -197,6 +204,13 @@ async fn test_run_hook_post_tool_use_event() { command: "echo post-hook".to_string(), tool_filter: None, timeout_ms: 5000, + hook_type: Default::default(), + url: None, + headers: Default::default(), + prompt: None, + model: None, + condition: None, + id: None, }; let data = serde_json::json!({"tool_name": "Read", "result": "ok"}); let result = run_hook(&hook, data) diff --git a/crates/loopal-ipc/tests/suite/e2e_ipc_edge_test.rs b/crates/loopal-ipc/tests/suite/e2e_ipc_edge_test.rs index b741d2ce..949bef02 100644 --- a/crates/loopal-ipc/tests/suite/e2e_ipc_edge_test.rs +++ b/crates/loopal-ipc/tests/suite/e2e_ipc_edge_test.rs @@ -40,6 +40,9 @@ async fn e2e_streaming_events_ordered() { for i in 0..5 { let event = AgentEvent { agent_name: None, + event_id: 0, + turn_id: 0, + correlation_id: 0, payload: AgentEventPayload::Stream { text: format!("chunk-{i}"), }, @@ -149,6 +152,9 @@ async fn e2e_bridge_stops_on_incoming_close() { let event = AgentEvent { agent_name: None, + event_id: 0, + turn_id: 0, + correlation_id: 0, payload: AgentEventPayload::Stream { text: "one".into() }, }; fwd_tx diff --git a/crates/loopal-ipc/tests/suite/e2e_ipc_test.rs b/crates/loopal-ipc/tests/suite/e2e_ipc_test.rs index a43aebc0..8c460c97 100644 --- a/crates/loopal-ipc/tests/suite/e2e_ipc_test.rs +++ b/crates/loopal-ipc/tests/suite/e2e_ipc_test.rs @@ -54,6 +54,9 @@ async fn e2e_message_then_event_roundtrip() { let _ = sc.respond(id, serde_json::json!({"ok": true})).await; let event = AgentEvent { agent_name: None, + event_id: 0, + turn_id: 0, + correlation_id: 0, payload: AgentEventPayload::Stream { text: "reply".into(), }, diff --git a/crates/loopal-ipc/tests/suite/regression_test.rs b/crates/loopal-ipc/tests/suite/regression_test.rs index c31a2f16..744cd140 100644 --- a/crates/loopal-ipc/tests/suite/regression_test.rs +++ b/crates/loopal-ipc/tests/suite/regression_test.rs @@ -51,6 +51,9 @@ async fn bridge_survives_malformed_event_notification() { // Send valid event after let valid = loopal_protocol::AgentEvent { agent_name: None, + event_id: 0, + turn_id: 0, + correlation_id: 0, payload: AgentEventPayload::AwaitingInput, }; server_conn @@ -124,6 +127,9 @@ async fn client_recv_survives_malformed_event() { let valid = loopal_protocol::AgentEvent { agent_name: Some("test".into()), + event_id: 0, + turn_id: 0, + correlation_id: 0, payload: AgentEventPayload::Finished, }; server_conn diff --git a/crates/loopal-kernel/src/hook_factory.rs b/crates/loopal-kernel/src/hook_factory.rs new file mode 100644 index 00000000..d54cb2a4 --- /dev/null +++ b/crates/loopal-kernel/src/hook_factory.rs @@ -0,0 +1,70 @@ +//! Hook executor factory — creates `HookExecutor` from `HookConfig`. +//! +//! GRASP Creator: Kernel implements this because constructing a +//! `PromptExecutor` requires Provider access that only Kernel has. + +use std::sync::Arc; +use std::time::Duration; + +use loopal_config::{HookConfig, HookType}; +use loopal_hooks::executor::{ExecutorFactory, HookExecutor}; +use loopal_hooks::executor_command::CommandExecutor; +use loopal_hooks::executor_http::HttpExecutor; +use loopal_hooks::executor_prompt::PromptExecutor; +use loopal_provider_api::Provider; +use tracing::error; + +/// Factory that dispatches to Command, Http, or Prompt executors. +pub struct DefaultExecutorFactory { + /// Provider for PromptExecutor (resolved at Kernel construction). + provider: Option>, +} + +impl DefaultExecutorFactory { + pub fn new(provider: Option>) -> Self { + Self { provider } + } +} + +impl ExecutorFactory for DefaultExecutorFactory { + fn create(&self, config: &HookConfig) -> Option> { + let timeout = Duration::from_millis(config.timeout_ms); + match config.hook_type { + HookType::Command => Some(Box::new(CommandExecutor { + command: config.command.clone(), + timeout, + })), + HookType::Http => { + let Some(ref url) = config.url else { + error!("Http hook missing required `url` field, skipping"); + return None; + }; + if url.is_empty() { + error!("Http hook has empty `url` field, skipping"); + return None; + } + Some(Box::new(HttpExecutor { + url: url.clone(), + headers: config.headers.clone(), + timeout, + })) + } + HookType::Prompt => { + let Some(ref provider) = self.provider else { + error!("Prompt hook requires a Provider but none available, skipping"); + return None; + }; + Some(Box::new(PromptExecutor { + system_prompt: config.prompt.clone().unwrap_or_default(), + model: config + .model + .clone() + .unwrap_or_else(|| "claude-haiku-4-5-20251001".into()), + provider: provider.clone(), + timeout, + max_tokens: 256, + })) + } + } + } +} diff --git a/crates/loopal-kernel/src/kernel.rs b/crates/loopal-kernel/src/kernel.rs index 917216d0..890c583a 100644 --- a/crates/loopal-kernel/src/kernel.rs +++ b/crates/loopal-kernel/src/kernel.rs @@ -3,7 +3,7 @@ use std::sync::Arc; use loopal_config::HookEvent; use loopal_config::Settings; use loopal_error::Result; -use loopal_hooks::HookRegistry; +use loopal_hooks::{HookRegistry, HookService}; use loopal_mcp::types::{McpPrompt, McpResource}; use loopal_mcp::{McpManager, McpToolAdapter}; use loopal_provider::ProviderRegistry; @@ -12,6 +12,7 @@ use loopal_tools::ToolRegistry; use tokio::sync::RwLock; use tracing::{info, warn}; +use crate::hook_factory::DefaultExecutorFactory; use crate::provider_registry; use loopal_tool_background::BackgroundTaskStore; @@ -19,7 +20,7 @@ use loopal_tool_background::BackgroundTaskStore; pub struct Kernel { tool_registry: ToolRegistry, provider_registry: ProviderRegistry, - hook_registry: HookRegistry, + hook_service: HookService, mcp_manager: Arc>, /// MCP server instructions cached at start_mcp() time. mcp_instructions: Vec<(String, String)>, @@ -41,6 +42,8 @@ impl Kernel { provider_registry::register_providers(&settings, &mut provider_registry); let hook_registry = HookRegistry::new(settings.hooks.clone()); + let factory = Arc::new(DefaultExecutorFactory::new(None)); + let hook_service = HookService::new(hook_registry, factory); let mcp_manager = Arc::new(RwLock::new(McpManager::new())); info!("kernel initialized"); @@ -48,7 +51,7 @@ impl Kernel { Ok(Self { tool_registry, provider_registry, - hook_registry, + hook_service, mcp_manager, mcp_instructions: Vec::new(), mcp_resources: Vec::new(), @@ -171,7 +174,14 @@ impl Kernel { event: HookEvent, tool_name: Option<&str>, ) -> Vec<&loopal_config::HookConfig> { - self.hook_registry.match_hooks(event, tool_name) + self.hook_service + .registry() + .match_hooks(event, tool_name, None) + } + + /// Access the hook service for structured hook execution. + pub fn hook_service(&self) -> &HookService { + &self.hook_service } /// Get the shared MCP manager for server instructions and other queries. @@ -179,17 +189,17 @@ impl Kernel { &self.mcp_manager } - /// Get MCP server instructions cached from the initialize handshake. + /// MCP server instructions cached from the initialize handshake. pub fn mcp_instructions(&self) -> &[(String, String)] { &self.mcp_instructions } - /// Get MCP resources cached at startup. + /// MCP resources cached at startup. pub fn mcp_resources(&self) -> &[(String, McpResource)] { &self.mcp_resources } - /// Get MCP prompts cached at startup. + /// MCP prompts cached at startup. pub fn mcp_prompts(&self) -> &[(String, McpPrompt)] { &self.mcp_prompts } diff --git a/crates/loopal-kernel/src/lib.rs b/crates/loopal-kernel/src/lib.rs index 50cfa487..c74bbc52 100644 --- a/crates/loopal-kernel/src/lib.rs +++ b/crates/loopal-kernel/src/lib.rs @@ -1,3 +1,4 @@ +pub mod hook_factory; pub mod kernel; pub mod provider_registry; pub mod sampling; diff --git a/crates/loopal-protocol/src/event.rs b/crates/loopal-protocol/src/event.rs index bd380c0c..bd7a8158 100644 --- a/crates/loopal-protocol/src/event.rs +++ b/crates/loopal-protocol/src/event.rs @@ -1,14 +1,25 @@ use serde::{Deserialize, Serialize}; -use crate::question::Question; +use crate::event_id::{current_correlation_id, current_turn_id, next_event_id}; +use crate::event_payload::AgentEventPayload; -/// Complete event with agent identity, transported via event channel to consumers. +/// Complete event with agent identity and causality tracking, +/// transported via event channel to consumers. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct AgentEvent { /// Agent that produced this event. Hub fills this in for all agents /// (root = `Some("main")`, sub-agent = `Some("name")`). /// `None` only in the agent process before Hub injection. pub agent_name: Option, + /// Monotonically increasing process-unique ID (0 = unset). + #[serde(default)] + pub event_id: u64, + /// Turn that produced this event (0 = outside a turn). + #[serde(default)] + pub turn_id: u32, + /// Groups related events (e.g. parallel tool batch). 0 = ungrouped. + #[serde(default)] + pub correlation_id: u64, pub payload: AgentEventPayload, } @@ -17,6 +28,9 @@ impl AgentEvent { pub fn root(payload: AgentEventPayload) -> Self { Self { agent_name: None, + event_id: next_event_id(), + turn_id: current_turn_id(), + correlation_id: current_correlation_id(), payload, } } @@ -25,175 +39,10 @@ impl AgentEvent { pub fn named(name: impl Into, payload: AgentEventPayload) -> Self { Self { agent_name: Some(name.into()), + event_id: next_event_id(), + turn_id: current_turn_id(), + correlation_id: current_correlation_id(), payload, } } } - -/// Event payload. Runner/LLM/Tools only construct this enum. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub enum AgentEventPayload { - /// Streaming text chunk from LLM - Stream { text: String }, - - /// Streaming thinking/reasoning chunk from LLM - ThinkingStream { text: String }, - - /// Thinking phase completed - ThinkingComplete { token_count: u32 }, - - /// LLM is calling a tool - ToolCall { - id: String, - name: String, - input: serde_json::Value, - }, - - /// Tool execution completed - ToolResult { - id: String, - name: String, - result: String, - is_error: bool, - /// Wall-clock execution time in milliseconds (filled by runtime). - #[serde(default, skip_serializing_if = "Option::is_none")] - duration_ms: Option, - /// Structured data from the tool (e.g. bytes_written for Write). - #[serde(default, skip_serializing_if = "Option::is_none")] - metadata: Option, - }, - - /// Periodic progress update for long-running tools (e.g. Bash). - ToolProgress { - id: String, - name: String, - /// Latest output tail or status message. - output_tail: String, - /// Elapsed time in milliseconds since tool started. - elapsed_ms: u64, - }, - - /// Marks the start of a parallel tool batch (3+ tools executing concurrently). - ToolBatchStart { tool_ids: Vec }, - - /// Tool requires user permission - ToolPermissionRequest { - id: String, - name: String, - input: serde_json::Value, - }, - - /// Error occurred - Error { message: String }, - - /// Transient retry error — not persisted in message history. - RetryError { - message: String, - attempt: u32, - max_attempts: u32, - }, - - /// Retry succeeded or cancelled — signal retry resolution. - RetryCleared, - - /// Agent is waiting for user input - AwaitingInput, - - /// LLM output truncated by max_tokens; auto-continuing. - AutoContinuation { - continuation: u32, - max_continuations: u32, - }, - - /// Token usage update - TokenUsage { - input_tokens: u32, - output_tokens: u32, - context_window: u32, - cache_creation_input_tokens: u32, - cache_read_input_tokens: u32, - thinking_tokens: u32, - }, - - /// Mode changed - ModeChanged { mode: String }, - - /// Agent loop started - Started, - - /// Agent loop finished - Finished, - - /// A message was routed through the MessageRouter (Observation Plane). - /// - /// Emitted automatically by `MessageRouter::route()` for every envelope - /// delivered, providing transparent inter-agent communication visibility. - MessageRouted { - source: String, - target: String, - content_preview: String, - }, - - /// Tool is requesting user to answer questions. - UserQuestionRequest { - id: String, - questions: Vec, - }, - - /// Conversation was rewound; remaining_turns is the count after truncation. - Rewound { remaining_turns: usize }, - - /// Conversation was compacted; old messages removed to reduce context. - Compacted { - kept: usize, - removed: usize, - tokens_before: u32, - tokens_after: u32, - /// "smart" (LLM summarization) or "emergency" (blind truncation). - strategy: String, - }, - - /// Agent work was interrupted (cancel signal or new message while busy). - Interrupted, - - /// Files modified during the completed turn. - TurnDiffSummary { modified_files: Vec }, - - /// Server-side tool invoked (e.g. web_search). Observational — not user-initiated. - ServerToolUse { - id: String, - name: String, - input: serde_json::Value, - }, - - /// Server-side tool result received. Observational — not user-initiated. - ServerToolResult { - tool_use_id: String, - content: serde_json::Value, - }, - - /// A sub-agent was spawned by Hub. - SubAgentSpawned { - name: String, - agent_id: String, - /// Parent agent name (None for root-spawned agents). - #[serde(default, skip_serializing_if = "Option::is_none")] - parent: Option, - /// Model used by the spawned agent. - #[serde(default, skip_serializing_if = "Option::is_none")] - model: Option, - /// Session ID of the sub-agent's persistent session storage. - #[serde(default, skip_serializing_if = "Option::is_none")] - session_id: Option, - }, - - /// Auto-mode classifier made a permission decision. - AutoModeDecision { - tool_name: String, - decision: String, - reason: String, - /// Classification wall-clock time in milliseconds (0 for cached results). - #[serde(default)] - duration_ms: u64, - }, -} diff --git a/crates/loopal-protocol/src/event_id.rs b/crates/loopal-protocol/src/event_id.rs new file mode 100644 index 00000000..27a076fe --- /dev/null +++ b/crates/loopal-protocol/src/event_id.rs @@ -0,0 +1,38 @@ +//! Monotonically increasing event ID generator + turn/correlation tracking. +//! +//! Used to assign unique IDs to `AgentEvent` instances for causality tracking. +//! IDs start at 1; 0 is reserved for "unset" (e.g. events from older producers). +//! +//! Turn ID and correlation ID are set by the runtime during execution and read +//! by event emitters to stamp outgoing events without changing the emit() trait. + +use std::sync::atomic::{AtomicU32, AtomicU64, Ordering}; + +static NEXT_EVENT_ID: AtomicU64 = AtomicU64::new(1); +static CURRENT_TURN_ID: AtomicU32 = AtomicU32::new(0); +static CURRENT_CORRELATION_ID: AtomicU64 = AtomicU64::new(0); + +/// Generate the next unique event ID (monotonically increasing, never 0). +pub fn next_event_id() -> u64 { + NEXT_EVENT_ID.fetch_add(1, Ordering::Relaxed) +} + +/// Set the current turn ID (called by runtime at turn boundaries). +pub fn set_current_turn_id(id: u32) { + CURRENT_TURN_ID.store(id, Ordering::Relaxed); +} + +/// Get the current turn ID (called by event emitters). +pub fn current_turn_id() -> u32 { + CURRENT_TURN_ID.load(Ordering::Relaxed) +} + +/// Set the current correlation ID (called by runtime for tool batches). +pub fn set_current_correlation_id(id: u64) { + CURRENT_CORRELATION_ID.store(id, Ordering::Relaxed); +} + +/// Get the current correlation ID (called by event emitters). +pub fn current_correlation_id() -> u64 { + CURRENT_CORRELATION_ID.load(Ordering::Relaxed) +} diff --git a/crates/loopal-protocol/src/event_payload.rs b/crates/loopal-protocol/src/event_payload.rs new file mode 100644 index 00000000..17d0b48e --- /dev/null +++ b/crates/loopal-protocol/src/event_payload.rs @@ -0,0 +1,180 @@ +use serde::{Deserialize, Serialize}; + +use crate::question::Question; + +/// Event payload. Runner/LLM/Tools only construct this enum. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub enum AgentEventPayload { + /// Streaming text chunk from LLM + Stream { text: String }, + + /// Streaming thinking/reasoning chunk from LLM + ThinkingStream { text: String }, + + /// Thinking phase completed + ThinkingComplete { token_count: u32 }, + + /// LLM is calling a tool + ToolCall { + id: String, + name: String, + input: serde_json::Value, + }, + + /// Tool execution completed + ToolResult { + id: String, + name: String, + result: String, + is_error: bool, + /// Wall-clock execution time in milliseconds (filled by runtime). + #[serde(default, skip_serializing_if = "Option::is_none")] + duration_ms: Option, + /// Structured data from the tool (e.g. bytes_written for Write). + #[serde(default, skip_serializing_if = "Option::is_none")] + metadata: Option, + }, + + /// Periodic progress update for long-running tools (e.g. Bash). + ToolProgress { + id: String, + name: String, + /// Latest output tail or status message. + output_tail: String, + /// Elapsed time in milliseconds since tool started. + elapsed_ms: u64, + }, + + /// Marks the start of a parallel tool batch (3+ tools executing concurrently). + ToolBatchStart { tool_ids: Vec }, + + /// Tool requires user permission + ToolPermissionRequest { + id: String, + name: String, + input: serde_json::Value, + }, + + /// Error occurred + Error { message: String }, + + /// Transient retry error — not persisted in message history. + RetryError { + message: String, + attempt: u32, + max_attempts: u32, + }, + + /// Retry succeeded or cancelled — signal retry resolution. + RetryCleared, + + /// Agent is waiting for user input + AwaitingInput, + + /// LLM output truncated by max_tokens; auto-continuing. + AutoContinuation { + continuation: u32, + max_continuations: u32, + }, + + /// Token usage update + TokenUsage { + input_tokens: u32, + output_tokens: u32, + context_window: u32, + cache_creation_input_tokens: u32, + cache_read_input_tokens: u32, + thinking_tokens: u32, + }, + + /// Mode changed + ModeChanged { mode: String }, + + /// Agent loop started + Started, + + /// Agent loop finished + Finished, + + /// Inter-agent message routed through MessageRouter (Observation Plane). + MessageRouted { + source: String, + target: String, + content_preview: String, + }, + + /// Tool is requesting user to answer questions. + UserQuestionRequest { + id: String, + questions: Vec, + }, + + /// Conversation was rewound; remaining_turns is the count after truncation. + Rewound { remaining_turns: usize }, + + /// Conversation was compacted; old messages removed to reduce context. + Compacted { + kept: usize, + removed: usize, + tokens_before: u32, + tokens_after: u32, + /// "smart" (LLM summarization) or "emergency" (blind truncation). + strategy: String, + }, + + /// Agent work was interrupted (cancel signal or new message while busy). + Interrupted, + + /// Files modified during the completed turn. + TurnDiffSummary { modified_files: Vec }, + + /// Server-side tool invoked (e.g. web_search). Observational. + ServerToolUse { + id: String, + name: String, + input: serde_json::Value, + }, + + /// Server-side tool result received. Observational. + ServerToolResult { + tool_use_id: String, + content: serde_json::Value, + }, + + /// A sub-agent was spawned by Hub. + SubAgentSpawned { + name: String, + agent_id: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + parent: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + model: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + session_id: Option, + }, + + /// Auto-mode classifier made a permission decision. + AutoModeDecision { + tool_name: String, + decision: String, + reason: String, + #[serde(default)] + duration_ms: u64, + }, + + /// Aggregated metrics emitted at the end of each turn. + TurnCompleted { + turn_id: u32, + duration_ms: u64, + llm_calls: u32, + tool_calls_requested: u32, + tool_calls_approved: u32, + tool_calls_denied: u32, + tool_errors: u32, + auto_continuations: u32, + warnings_injected: u32, + tokens_in: u32, + tokens_out: u32, + modified_files: Vec, + }, +} diff --git a/crates/loopal-protocol/src/lib.rs b/crates/loopal-protocol/src/lib.rs index c8b1fa0f..2dcc6b34 100644 --- a/crates/loopal-protocol/src/lib.rs +++ b/crates/loopal-protocol/src/lib.rs @@ -5,6 +5,8 @@ pub mod command; pub mod control; pub mod envelope; pub mod event; +pub mod event_id; +pub mod event_payload; pub mod interrupt; pub mod projected; pub mod projection; @@ -17,7 +19,8 @@ pub use bg_task::{BgTaskSnapshot, BgTaskStatus}; pub use command::AgentMode; pub use control::ControlCommand; pub use envelope::{Envelope, MessageSource}; -pub use event::{AgentEvent, AgentEventPayload}; +pub use event::AgentEvent; +pub use event_payload::AgentEventPayload; pub use interrupt::InterruptSignal; pub use projected::{ProjectedMessage, ProjectedToolCall}; pub use projection::project_messages; diff --git a/crates/loopal-runtime/src/agent_loop/compaction.rs b/crates/loopal-runtime/src/agent_loop/compaction.rs index 7d8d8d6b..3d15050f 100644 --- a/crates/loopal-runtime/src/agent_loop/compaction.rs +++ b/crates/loopal-runtime/src/agent_loop/compaction.rs @@ -24,6 +24,17 @@ impl AgentLoopRunner { return Ok(()); } + // PreCompact hook: notify before compaction. + crate::fire_hooks::fire_hooks( + &self.params.deps.kernel, + loopal_config::HookEvent::PreCompact, + &loopal_hooks::HookContext { + session_id: Some(&self.params.session.id), + ..Default::default() + }, + ) + .await; + let msg_tokens = self.params.store.current_tokens(); let budget = self.params.store.budget().clone(); diff --git a/crates/loopal-runtime/src/agent_loop/input.rs b/crates/loopal-runtime/src/agent_loop/input.rs index af09bd9c..01046d65 100644 --- a/crates/loopal-runtime/src/agent_loop/input.rs +++ b/crates/loopal-runtime/src/agent_loop/input.rs @@ -1,5 +1,5 @@ -//! Agent input handling — wait for user input, scheduler triggers, and -//! Hub-injected notifications (e.g. sub-agent completion). +//! Agent input handling — wait for user input, scheduler triggers, +//! hook rewake signals, and Hub-injected notifications. use crate::agent_input::AgentInput; use loopal_error::Result; @@ -9,12 +9,10 @@ use tracing::{error, info}; use super::WaitResult; use super::message_build::build_user_message; use super::runner::AgentLoopRunner; +use crate::fire_hooks::fire_hooks; impl AgentLoopRunner { /// Wait for input from any source. Returns None if all channels closed. - /// - /// Does NOT emit AwaitingInput — that's handled by `run_loop`'s - /// state machine via `transition(WaitingForInput)`. pub async fn wait_for_input(&mut self) -> Result> { let stale = self.interrupt.take(); if stale { @@ -22,39 +20,68 @@ impl AgentLoopRunner { } info!("awaiting input"); loop { - // Select between frontend input and scheduler triggers. - // Hub-injected notifications (sub-agent completion) arrive via - // frontend.recv_input() through the IPC → input_tx path. - let input = if let Some(ref mut rx) = self.trigger_rx { - tokio::select! { - input = self.params.deps.frontend.recv_input() => input, - envelope = rx.recv() => { - if let Some(env) = envelope { - return Ok(Some(self.ingest_message(&env))); - } - info!("scheduler channel closed"); - self.trigger_rx = None; - continue; - } - } - } else { - self.params.deps.frontend.recv_input().await - }; + let input = self.select_input().await; match input { - Some(AgentInput::Message(env)) => { - return Ok(Some(self.ingest_message(&env))); + SelectResult::AgentInput(Some(AgentInput::Message(env))) => { + let result = self.ingest_message(&env); + fire_hooks( + &self.params.deps.kernel, + loopal_config::HookEvent::PostInput, + &loopal_hooks::HookContext { + session_id: Some(&self.params.session.id), + ..Default::default() + }, + ) + .await; + return Ok(Some(result)); } - Some(AgentInput::Control(ctrl)) => { + SelectResult::AgentInput(Some(AgentInput::Control(ctrl))) => { self.handle_control(ctrl).await?; } - None => { + SelectResult::AgentInput(None) => { info!("input channel closed, ending agent loop"); return Ok(None); } + SelectResult::Envelope(env) => { + return Ok(Some(self.ingest_message(&env))); + } + SelectResult::ChannelClosed => continue, } } } + /// Multiplex frontend, scheduler, and hook rewake channels. + async fn select_input(&mut self) -> SelectResult { + match (&mut self.trigger_rx, &mut self.rewake_rx) { + (Some(sched), Some(rewake)) => tokio::select! { + input = self.params.deps.frontend.recv_input() => SelectResult::AgentInput(input), + env = sched.recv() => match env { + Some(e) => SelectResult::Envelope(e), + None => { self.trigger_rx = None; SelectResult::ChannelClosed } + }, + env = rewake.recv() => match env { + Some(e) => SelectResult::Envelope(e), + None => { self.rewake_rx = None; SelectResult::ChannelClosed } + }, + }, + (Some(sched), None) => tokio::select! { + input = self.params.deps.frontend.recv_input() => SelectResult::AgentInput(input), + env = sched.recv() => match env { + Some(e) => SelectResult::Envelope(e), + None => { self.trigger_rx = None; SelectResult::ChannelClosed } + }, + }, + (None, Some(rewake)) => tokio::select! { + input = self.params.deps.frontend.recv_input() => SelectResult::AgentInput(input), + env = rewake.recv() => match env { + Some(e) => SelectResult::Envelope(e), + None => { self.rewake_rx = None; SelectResult::ChannelClosed } + }, + }, + (None, None) => SelectResult::AgentInput(self.params.deps.frontend.recv_input().await), + } + } + /// Accept a message envelope: persist (if not ephemeral) and push to store. pub(super) fn ingest_message(&mut self, env: &Envelope) -> WaitResult { let mut user_msg = build_user_message(env); @@ -75,17 +102,25 @@ impl AgentLoopRunner { WaitResult::MessageAdded } - /// Non-blocking drain of all pending input (frontend + scheduler). - /// Returns immediately with whatever messages are queued. Used by Task - /// agents to check if there's more work before deciding to exit. + /// Non-blocking drain of all pending input (frontend + scheduler + rewake). pub(super) async fn drain_pending_input(&mut self) -> Vec { let mut pending = self.params.deps.frontend.drain_pending().await; - // Also drain scheduler triggers. if let Some(ref mut rx) = self.trigger_rx { while let Ok(env) = rx.try_recv() { pending.push(env); } } + if let Some(ref mut rx) = self.rewake_rx { + while let Ok(env) = rx.try_recv() { + pending.push(env); + } + } pending } } + +enum SelectResult { + AgentInput(Option), + Envelope(Envelope), + ChannelClosed, +} diff --git a/crates/loopal-runtime/src/agent_loop/llm.rs b/crates/loopal-runtime/src/agent_loop/llm.rs index b7cad959..94658c1e 100644 --- a/crates/loopal-runtime/src/agent_loop/llm.rs +++ b/crates/loopal-runtime/src/agent_loop/llm.rs @@ -29,6 +29,18 @@ impl AgentLoopRunner { .deps .kernel .resolve_provider(self.params.config.model())?; + + // PreRequest hook: notify before LLM call. + crate::fire_hooks::fire_hooks( + &self.params.deps.kernel, + loopal_config::HookEvent::PreRequest, + &loopal_hooks::HookContext { + session_id: Some(&self.params.session.id), + ..Default::default() + }, + ) + .await; + let llm_start = Instant::now(); info!( model = %self.params.config.model(), messages = messages.len(), diff --git a/crates/loopal-runtime/src/agent_loop/loop_detector.rs b/crates/loopal-runtime/src/agent_loop/loop_detector.rs index b9f65d3e..5fd5f143 100644 --- a/crates/loopal-runtime/src/agent_loop/loop_detector.rs +++ b/crates/loopal-runtime/src/agent_loop/loop_detector.rs @@ -16,15 +16,31 @@ const ABORT_THRESHOLD: u32 = 5; const SIGNATURE_INPUT_LIMIT: usize = 200; /// Tracks tool call signatures and their cumulative occurrence count. -#[derive(Default)] pub struct LoopDetector { /// (signature → cumulative count across the turn) signatures: HashMap, + warn_threshold: u32, + abort_threshold: u32, +} + +impl Default for LoopDetector { + fn default() -> Self { + Self::new() + } } impl LoopDetector { pub fn new() -> Self { - Self::default() + Self::with_thresholds(WARN_THRESHOLD, ABORT_THRESHOLD) + } + + /// Create a detector with custom thresholds (from HarnessConfig). + pub fn with_thresholds(warn: u32, abort: u32) -> Self { + Self { + signatures: HashMap::new(), + warn_threshold: warn, + abort_threshold: abort, + } } } @@ -41,14 +57,14 @@ impl TurnObserver for LoopDetector { let count = self.signatures.entry(sig).or_insert(0); *count += 1; - if *count >= ABORT_THRESHOLD { + if *count >= self.abort_threshold { tracing::warn!(tool = name, count, "loop detected, aborting turn"); return ObserverAction::AbortTurn(format!( "Loop detected: tool '{name}' called {count} cumulative times \ with similar arguments. Aborting to prevent waste.", )); } - if *count >= WARN_THRESHOLD { + if *count >= self.warn_threshold { tracing::warn!(tool = name, count, "possible loop detected"); worst = ObserverAction::InjectWarning(format!( "[WARNING: Tool '{name}' has been called {count} times with similar \ diff --git a/crates/loopal-runtime/src/agent_loop/mod.rs b/crates/loopal-runtime/src/agent_loop/mod.rs index bb216dbd..15f8885f 100644 --- a/crates/loopal-runtime/src/agent_loop/mod.rs +++ b/crates/loopal-runtime/src/agent_loop/mod.rs @@ -30,12 +30,14 @@ pub(crate) mod tools_plan; mod tools_resolve; pub mod turn_context; mod turn_exec; +pub(crate) mod turn_metrics; pub mod turn_observer; use std::collections::HashSet; use std::sync::Arc; use crate::frontend::traits::AgentFrontend; +use loopal_config::HarnessConfig; use loopal_context::ContextStore; use loopal_error::{AgentOutput, Result}; use loopal_kernel::Kernel; @@ -52,9 +54,6 @@ use finished_guard::FinishedGuard; pub use runner::AgentLoopRunner; -/// Maximum number of automatic continuations when LLM hits max_tokens. -pub(crate) const MAX_AUTO_CONTINUATIONS: u32 = 3; - // ── Sub-structs ──────────────────────────────────────────────────── /// Agent lifecycle mode — determines idle behavior after turn completion. @@ -162,6 +161,10 @@ pub struct AgentLoopParams { pub scheduled_rx: Option>, /// Auto-mode LLM classifier (active when permission_mode == Auto). pub auto_classifier: Option>, + /// Harness control parameters (loop thresholds, continuations, etc.). + pub harness: HarnessConfig, + /// Receive end for async hook rewake messages (exit code 2 from background hooks). + pub rewake_rx: Option>, } /// Public wrapper — constructs default observers and runs the loop. @@ -169,8 +172,12 @@ pub struct AgentLoopParams { /// A `FinishedGuard` ensures `Finished` is always emitted — even on panic. pub async fn agent_loop(params: AgentLoopParams) -> Result { let mut guard = FinishedGuard::new(params.deps.frontend.clone()); + let h = ¶ms.harness; let observers: Vec> = vec![ - Box::new(loop_detector::LoopDetector::new()), + Box::new(loop_detector::LoopDetector::with_thresholds( + h.loop_warn_threshold, + h.loop_abort_threshold, + )), Box::new(diff_tracker::DiffTracker::new(params.deps.frontend.clone())), ]; let mut runner = AgentLoopRunner::new(params); diff --git a/crates/loopal-runtime/src/agent_loop/runner.rs b/crates/loopal-runtime/src/agent_loop/runner.rs index 1bf6ce00..fc54b432 100644 --- a/crates/loopal-runtime/src/agent_loop/runner.rs +++ b/crates/loopal-runtime/src/agent_loop/runner.rs @@ -11,6 +11,7 @@ use super::token_accumulator::TokenAccumulator; use super::turn_context::TurnContext; use super::turn_observer::TurnObserver; use super::{AgentLoopParams, TurnOutput}; +use crate::fire_hooks::fire_hooks; use crate::plan_file::PlanFile; /// Encapsulates the agent loop state and behavior. @@ -25,6 +26,8 @@ pub struct AgentLoopRunner { pub observers: Vec>, /// Scheduler message receiver — consumed in `wait_for_input()`. pub trigger_rx: Option>, + /// Async hook rewake channel — background hooks send Envelopes here. + pub rewake_rx: Option>, /// Explicit agent state — source of truth, propagated via events to Session layer. pub status: AgentStatus, /// Plan file for the current session (created lazily on first plan mode entry). @@ -51,6 +54,7 @@ impl AgentLoopRunner { let interrupt = params.interrupt.signal.clone(); let interrupt_tx = params.interrupt.tx.clone(); let trigger_rx = params.scheduled_rx.take(); + let rewake_rx = params.rewake_rx.take(); let plan_file = PlanFile::new(std::path::Path::new(¶ms.session.cwd)); Self { params, @@ -62,6 +66,7 @@ impl AgentLoopRunner { interrupt_tx, observers: Vec::new(), trigger_rx, + rewake_rx, status: AgentStatus::Starting, plan_file, } @@ -77,13 +82,16 @@ impl AgentLoopRunner { /// Actual run logic, executed inside the `agent` span. async fn run_instrumented(&mut self) -> Result { info!(model = %self.params.config.model(), "agent loop started"); - // Started is a one-time lifecycle event (not a status transition). - // Status moves to Running via transition() before each turn. self.transition(AgentStatus::Running).await?; self.emit(AgentEventPayload::Started).await?; + self.fire_session_hook(loopal_config::HookEvent::SessionStart) + .await; let result = self.run_loop().await; + self.fire_session_hook(loopal_config::HookEvent::SessionEnd) + .await; + if let Err(ref e) = result { let _ = self.transition_error(e.to_string()).await; } @@ -92,10 +100,24 @@ impl AgentLoopRunner { result } + /// Fire a session-level hook (SessionStart, SessionEnd). + async fn fire_session_hook(&self, event: loopal_config::HookEvent) { + fire_hooks( + &self.params.deps.kernel, + event, + &loopal_hooks::HookContext { + session_id: Some(&self.params.session.id), + cwd: Some(&self.params.session.cwd), + ..Default::default() + }, + ) + .await; + } + /// One complete turn: LLM → [tools → LLM]* → returns when no tool calls. - /// - /// Wraps `execute_turn_inner` with observer on_turn_start/on_turn_end. + /// Emits `TurnCompleted` event with aggregated metrics at the end. pub(super) async fn execute_turn(&mut self, turn_ctx: &mut TurnContext) -> Result { + loopal_protocol::event_id::set_current_turn_id(turn_ctx.turn_id); // global turn context for obs in &mut self.observers { obs.on_turn_start(turn_ctx); } @@ -103,6 +125,47 @@ impl AgentLoopRunner { for obs in &mut self.observers { obs.on_turn_end(turn_ctx); } + + // Finalize and emit turn telemetry. + turn_ctx.metrics.warnings_injected = turn_ctx.pending_warnings.len() as u32; + turn_ctx.metrics.tokens_in = self.tokens.input; + turn_ctx.metrics.tokens_out = self.tokens.output; + let m = &turn_ctx.metrics; + let files: Vec = turn_ctx.modified_files.iter().cloned().collect(); + let duration_ms = turn_ctx.started_at.elapsed().as_millis() as u64; + info!( + turn = turn_ctx.turn_id, + duration_ms, + llm = m.llm_calls, + tools = m.tool_calls_requested, + ok = m.tool_calls_approved, + denied = m.tool_calls_denied, + errs = m.tool_errors, + tok_in = m.tokens_in, + tok_out = m.tokens_out, + "turn completed" + ); + + let _ = self + .emit(AgentEventPayload::TurnCompleted { + turn_id: turn_ctx.turn_id, + duration_ms, + llm_calls: m.llm_calls, + tool_calls_requested: m.tool_calls_requested, + tool_calls_approved: m.tool_calls_approved, + tool_calls_denied: m.tool_calls_denied, + tool_errors: m.tool_errors, + auto_continuations: m.auto_continuations, + warnings_injected: m.warnings_injected, + tokens_in: m.tokens_in, + tokens_out: m.tokens_out, + modified_files: files, + }) + .await; + + // Reset turn context — events outside turns carry turn_id/correlation_id = 0. + loopal_protocol::event_id::set_current_turn_id(0); + loopal_protocol::event_id::set_current_correlation_id(0); result } @@ -111,15 +174,7 @@ impl AgentLoopRunner { self.params.deps.frontend.emit(payload).await } - /// Transition to a new agent status and emit the corresponding event. - /// - /// **This is the ONLY way to change agent status.** Every status change - /// goes through this method, ensuring SSOT and deterministic event emission. - /// - /// Skips emission when already in the target status (idempotent) to prevent - /// duplicate idle events — e.g. `emit_interrupted()` already transitions to - /// WaitingForInput, so the subsequent `transition(WaitingForInput)` at the - /// top of the loop must NOT emit a second AwaitingInput. + /// Transition to a new agent status. Skips if already in target (idempotent). pub(super) async fn transition(&mut self, new_status: AgentStatus) -> Result<()> { if self.status == new_status { return Ok(()); @@ -141,7 +196,6 @@ impl AgentLoopRunner { } /// Recalculate context budget from current model config. - /// /// Called after model switch so the compaction thresholds match the new model. pub(super) fn recalculate_budget(&mut self) { let tool_defs = self.params.deps.kernel.tool_definitions(); diff --git a/crates/loopal-runtime/src/agent_loop/tools.rs b/crates/loopal-runtime/src/agent_loop/tools.rs index a34bea35..d856e7d4 100644 --- a/crates/loopal-runtime/src/agent_loop/tools.rs +++ b/crates/loopal-runtime/src/agent_loop/tools.rs @@ -9,6 +9,7 @@ use super::question_parse::{format_answers, parse_questions}; use super::runner::AgentLoopRunner; use super::tool_exec::execute_approved_tools; use super::tools_inject::success_block; +use super::turn_metrics::ToolExecStats; use crate::mode::AgentMode; use crate::plan_file::wrap_plan_reminder; @@ -16,13 +17,16 @@ use loopal_error::Result; impl AgentLoopRunner { /// Execute tool calls: intercept → precheck → permission → parallel execution. + /// + /// Returns [`ToolExecStats`] for turn-level metrics aggregation. pub async fn execute_tools( &mut self, tool_uses: Vec<(String, String, serde_json::Value)>, cancel: &TurnCancel, - ) -> Result<()> { + ) -> Result { if cancel.is_cancelled() { - return self.emit_all_interrupted(&tool_uses).await; + self.emit_all_interrupted(&tool_uses).await?; + return Ok(ToolExecStats::default()); } // Phase 0: Intercept special tools (EnterPlanMode, ExitPlanMode, AskUser) @@ -41,6 +45,12 @@ impl AgentLoopRunner { "check_tools done" ); + let mut stats = ToolExecStats { + approved: check.approved.len() as u32, + denied: check.denied.len() as u32, + errors: 0, + }; + // Phase 2: Parallel execution let mut indexed_results: Vec<(usize, ContentBlock)> = Vec::new(); indexed_results.extend(intercepted); @@ -50,6 +60,9 @@ impl AgentLoopRunner { if check.approved.len() >= 3 { let tool_ids: Vec = check.approved.iter().map(|(id, _, _)| id.clone()).collect(); + // Set correlation ID so all events in this batch are grouped. + let batch_id = loopal_protocol::event_id::next_event_id(); + loopal_protocol::event_id::set_current_correlation_id(batch_id); self.emit(AgentEventPayload::ToolBatchStart { tool_ids }) .await?; } @@ -67,6 +80,8 @@ impl AgentLoopRunner { ) .await; indexed_results.extend(parallel); + // Reset correlation ID after batch completes. + loopal_protocol::event_id::set_current_correlation_id(0); } // Plan mode: wrap non-intercepted tool results with system-reminder. @@ -86,7 +101,15 @@ impl AgentLoopRunner { } } - self.finalize_tool_results(indexed_results) + // Count execution errors from result blocks + for (_, block) in &indexed_results { + if let ContentBlock::ToolResult { is_error: true, .. } = block { + stats.errors += 1; + } + } + + self.finalize_tool_results(indexed_results)?; + Ok(stats) } /// Phase 0: intercept EnterPlanMode, ExitPlanMode, AskUser. diff --git a/crates/loopal-runtime/src/agent_loop/turn_context.rs b/crates/loopal-runtime/src/agent_loop/turn_context.rs index 23f42f1b..ef3a3ab5 100644 --- a/crates/loopal-runtime/src/agent_loop/turn_context.rs +++ b/crates/loopal-runtime/src/agent_loop/turn_context.rs @@ -2,12 +2,13 @@ //! //! Created at the start of each turn in `run_loop`, passed to //! `execute_turn`, and consumed at turn end. Holds data that -//! observers accumulate during a turn (e.g. file diffs). +//! observers accumulate during a turn (e.g. file diffs, metrics). use std::collections::BTreeSet; use std::time::Instant; use super::cancel::TurnCancel; +use super::turn_metrics::TurnMetrics; /// Mutable context for a single turn (LLM → [tools → LLM]* → done). pub struct TurnContext { @@ -20,6 +21,8 @@ pub struct TurnContext { /// to the tool results message. Must NOT be pushed as a separate User /// message — that breaks tool_use/tool_result pairing after normalization. pub pending_warnings: Vec, + /// Aggregated telemetry counters for this turn. + pub metrics: TurnMetrics, } impl TurnContext { @@ -30,6 +33,7 @@ impl TurnContext { started_at: Instant::now(), modified_files: BTreeSet::new(), pending_warnings: Vec::new(), + metrics: TurnMetrics::default(), } } } diff --git a/crates/loopal-runtime/src/agent_loop/turn_exec.rs b/crates/loopal-runtime/src/agent_loop/turn_exec.rs index c2537dd6..fe13868d 100644 --- a/crates/loopal-runtime/src/agent_loop/turn_exec.rs +++ b/crates/loopal-runtime/src/agent_loop/turn_exec.rs @@ -1,16 +1,14 @@ //! Inner turn execution loop and observer dispatch. -//! -//! Split from runner.rs to keep files under 200 lines. use loopal_error::Result; use loopal_protocol::AgentEventPayload; use loopal_provider_api::StopReason; use tracing::{debug, info, warn}; +use super::TurnOutput; use super::runner::AgentLoopRunner; use super::turn_context::TurnContext; use super::turn_observer::ObserverAction; -use super::{MAX_AUTO_CONTINUATIONS, TurnOutput}; impl AgentLoopRunner { /// Inner loop: LLM → [tools → LLM]* → done. @@ -20,6 +18,8 @@ impl AgentLoopRunner { ) -> Result { let mut last_text = String::new(); let mut continuation_count: u32 = 0; + let mut stop_feedback_count: u32 = 0; + let max_stop_feedback = self.params.harness.max_stop_feedback; loop { if turn_ctx.cancel.is_cancelled() { info!("turn cancelled before LLM call"); @@ -30,9 +30,10 @@ impl AgentLoopRunner { self.check_and_compact().await?; // Prepare context for LLM (clone + strip old thinking) let working = self.params.store.prepare_for_llm(); + turn_ctx.metrics.llm_calls += 1; let result = self.stream_llm_with(&working, &turn_ctx.cancel).await?; - // Determine tool list for recording. MaxTokens+tools = truncated args. + // MaxTokens + tool calls = truncated args, discard tools. let truncated = result.stop_reason == StopReason::MaxTokens && !result.tool_uses.is_empty(); if truncated { @@ -57,11 +58,12 @@ impl AgentLoopRunner { if !result.assistant_text.is_empty() { last_text.clone_from(&result.assistant_text); } - if continuation_count < MAX_AUTO_CONTINUATIONS { + if continuation_count < self.params.harness.max_auto_continuations { continuation_count += 1; + turn_ctx.metrics.auto_continuations = continuation_count; self.emit(AgentEventPayload::AutoContinuation { continuation: continuation_count, - max_continuations: MAX_AUTO_CONTINUATIONS, + max_continuations: self.params.harness.max_auto_continuations, }) .await?; continue; @@ -88,11 +90,12 @@ impl AgentLoopRunner { } if result.tool_uses.is_empty() && result.stop_reason == StopReason::MaxTokens { - if continuation_count < MAX_AUTO_CONTINUATIONS { + if continuation_count < self.params.harness.max_auto_continuations { continuation_count += 1; + turn_ctx.metrics.auto_continuations = continuation_count; self.emit(AgentEventPayload::AutoContinuation { continuation: continuation_count, - max_continuations: MAX_AUTO_CONTINUATIONS, + max_continuations: self.params.harness.max_auto_continuations, }) .await?; continue; @@ -101,15 +104,37 @@ impl AgentLoopRunner { } if result.tool_uses.is_empty() { - // Use last_text (already updated at line 79-81 if current - // assistant_text is non-empty). This preserves the last - // meaningful text when the final LLM response is empty — - // e.g. ephemeral sub-agents whose last call returns no - // visible content still propagate their earlier output. + // Stop hook: if any hook provides feedback, continue (bounded). + if stop_feedback_count < max_stop_feedback { + let stop_outputs = self + .params + .deps + .kernel + .hook_service() + .run_hooks( + loopal_config::HookEvent::Stop, + &loopal_hooks::HookContext { + stop_reason: Some("end_turn"), + ..Default::default() + }, + ) + .await; + let feedback: Vec<&str> = stop_outputs + .iter() + .filter_map(|o| o.additional_context.as_deref()) + .collect(); + if !feedback.is_empty() { + stop_feedback_count += 1; + self.params + .store + .append_warnings_to_last_user(vec![feedback.join("\n")]); + continue; + } + } return Ok(TurnOutput { output: last_text }); } - // Observer: on_before_tools + // Observer: on_before_tools — may inject warnings or abort. debug!(tool_count = result.tool_uses.len(), "pre-tool phase"); if self.run_before_tools(turn_ctx, &result.tool_uses).await? { return Ok(TurnOutput { output: last_text }); @@ -126,18 +151,18 @@ impl AgentLoopRunner { "tool exec start" ); let cancel = &turn_ctx.cancel; - self.execute_tools(result.tool_uses.clone(), cancel).await?; + turn_ctx.metrics.tool_calls_requested += result.tool_uses.len() as u32; + let stats = self.execute_tools(result.tool_uses.clone(), cancel).await?; + turn_ctx.metrics.tool_calls_approved += stats.approved; + turn_ctx.metrics.tool_calls_denied += stats.denied; + turn_ctx.metrics.tool_errors += stats.errors; info!("tool exec complete"); - // Append observer warnings (e.g. loop detector) AFTER tool results. - // They must come after ToolResult blocks — inserting them before - // breaks tool_use/tool_result pairing when normalize_messages merges - // consecutive same-role User messages. + // Append observer warnings after tool results (must follow ToolResult blocks). let warnings = std::mem::take(&mut turn_ctx.pending_warnings); self.params.store.append_warnings_to_last_user(warnings); self.inject_pending_messages().await; - // Observer: on_after_tools with results from the last message let result_blocks = self .params @@ -154,7 +179,7 @@ impl AgentLoopRunner { } } - /// Run before-tools observers. Returns true if the turn should abort. + /// Run before-tools observers. Returns `true` if the turn should abort. pub(super) async fn run_before_tools( &mut self, turn_ctx: &mut TurnContext, @@ -164,10 +189,6 @@ impl AgentLoopRunner { match obs.on_before_tools(turn_ctx, tool_uses) { ObserverAction::Continue => {} ObserverAction::InjectWarning(msg) => { - // Store in context — appended to tool results message later. - // Pushing a separate User(Text) message here would break - // tool_use/tool_result pairing after normalize_messages merges - // consecutive same-role messages. turn_ctx.pending_warnings.push(msg); } ObserverAction::AbortTurn(reason) => { diff --git a/crates/loopal-runtime/src/agent_loop/turn_metrics.rs b/crates/loopal-runtime/src/agent_loop/turn_metrics.rs new file mode 100644 index 00000000..6049e15b --- /dev/null +++ b/crates/loopal-runtime/src/agent_loop/turn_metrics.rs @@ -0,0 +1,42 @@ +//! Per-turn metrics aggregated during the turn lifecycle. +//! +//! `TurnMetrics` is a counter-based telemetry struct accumulated +//! in `TurnContext` and emitted as `TurnCompleted` event at turn end. + +/// Aggregated metrics for a single agent turn. +/// +/// Counters are incremented at various execution points: +/// - `llm_calls`: each `stream_llm_with` invocation +/// - `tool_calls_*`: from `ToolExecStats` returned by `execute_tools` +/// - `auto_continuations`: from inner loop continuation count +/// - `warnings_injected`: from `pending_warnings.len()` +/// - `tokens_*`: from `TokenAccumulator` delta +#[derive(Debug, Default, Clone)] +pub struct TurnMetrics { + /// LLM streaming calls made during this turn. + pub llm_calls: u32, + /// Tools requested by LLM (total across all LLM iterations in the turn). + pub tool_calls_requested: u32, + /// Tools that passed permission checks and were executed. + pub tool_calls_approved: u32, + /// Tools denied by sandbox/permission/plan-mode. + pub tool_calls_denied: u32, + /// Tools whose execution returned is_error=true or Err. + pub tool_errors: u32, + /// MaxTokens auto-continuations triggered. + pub auto_continuations: u32, + /// Warnings injected by observers (e.g. loop detector). + pub warnings_injected: u32, + /// Input tokens consumed during this turn. + pub tokens_in: u32, + /// Output tokens produced during this turn. + pub tokens_out: u32, +} + +/// Summary returned by `execute_tools` for metrics aggregation. +#[derive(Debug, Default)] +pub struct ToolExecStats { + pub approved: u32, + pub denied: u32, + pub errors: u32, +} diff --git a/crates/loopal-runtime/src/fire_hooks.rs b/crates/loopal-runtime/src/fire_hooks.rs new file mode 100644 index 00000000..21556a78 --- /dev/null +++ b/crates/loopal-runtime/src/fire_hooks.rs @@ -0,0 +1,17 @@ +//! Thin convenience wrapper for firing hooks from the runtime. +//! +//! Avoids repeating `kernel.hook_service().run_hooks(...)` + error +//! handling at every call site. Outputs are intentionally discarded +//! for events that are observation-only (SessionStart/End, PreCompact, etc.). + +use loopal_config::HookEvent; +use loopal_hooks::HookContext; +use loopal_kernel::Kernel; + +/// Fire all matching hooks for an event, discarding outputs. +/// +/// Used for observation-only events (SessionStart, SessionEnd, PreCompact, etc.) +/// where the hooks cannot influence the control flow. +pub async fn fire_hooks(kernel: &Kernel, event: HookEvent, ctx: &HookContext<'_>) { + let _ = kernel.hook_service().run_hooks(event, ctx).await; +} diff --git a/crates/loopal-runtime/src/frontend/emitter.rs b/crates/loopal-runtime/src/frontend/emitter.rs index 2e815ca1..6bc0a174 100644 --- a/crates/loopal-runtime/src/frontend/emitter.rs +++ b/crates/loopal-runtime/src/frontend/emitter.rs @@ -26,6 +26,9 @@ impl EventEmitter for ChannelEventEmitter { async fn emit(&self, payload: AgentEventPayload) -> Result<()> { let event = AgentEvent { agent_name: self.agent_name.clone(), + event_id: loopal_protocol::event_id::next_event_id(), + turn_id: loopal_protocol::event_id::current_turn_id(), + correlation_id: loopal_protocol::event_id::current_correlation_id(), payload, }; self.tx diff --git a/crates/loopal-runtime/src/frontend/relay_permission.rs b/crates/loopal-runtime/src/frontend/relay_permission.rs index fcde8565..7955cddc 100644 --- a/crates/loopal-runtime/src/frontend/relay_permission.rs +++ b/crates/loopal-runtime/src/frontend/relay_permission.rs @@ -30,6 +30,9 @@ impl PermissionHandler for RelayPermissionHandler { async fn decide(&self, id: &str, name: &str, input: &serde_json::Value) -> PermissionDecision { let event = AgentEvent { agent_name: None, + event_id: loopal_protocol::event_id::next_event_id(), + turn_id: loopal_protocol::event_id::current_turn_id(), + correlation_id: loopal_protocol::event_id::current_correlation_id(), payload: AgentEventPayload::ToolPermissionRequest { id: id.to_string(), name: name.to_string(), diff --git a/crates/loopal-runtime/src/frontend/unified.rs b/crates/loopal-runtime/src/frontend/unified.rs index 6013c194..2742045a 100644 --- a/crates/loopal-runtime/src/frontend/unified.rs +++ b/crates/loopal-runtime/src/frontend/unified.rs @@ -66,6 +66,9 @@ impl AgentFrontend for UnifiedFrontend { async fn emit(&self, payload: AgentEventPayload) -> Result<()> { let event = AgentEvent { agent_name: self.agent_name.clone(), + event_id: loopal_protocol::event_id::next_event_id(), + turn_id: loopal_protocol::event_id::current_turn_id(), + correlation_id: loopal_protocol::event_id::current_correlation_id(), payload, }; if self.agent_name.is_some() { @@ -133,6 +136,9 @@ impl AgentFrontend for UnifiedFrontend { fn try_emit(&self, payload: AgentEventPayload) -> bool { let event = AgentEvent { agent_name: self.agent_name.clone(), + event_id: loopal_protocol::event_id::next_event_id(), + turn_id: loopal_protocol::event_id::current_turn_id(), + correlation_id: loopal_protocol::event_id::current_correlation_id(), payload, }; self.event_tx.try_send(event).is_ok() diff --git a/crates/loopal-runtime/src/lib.rs b/crates/loopal-runtime/src/lib.rs index 0bda8670..61e80686 100644 --- a/crates/loopal-runtime/src/lib.rs +++ b/crates/loopal-runtime/src/lib.rs @@ -1,5 +1,6 @@ pub mod agent_input; pub mod agent_loop; +pub mod fire_hooks; pub mod frontend; pub mod mode; pub mod permission; diff --git a/crates/loopal-runtime/src/tool_pipeline.rs b/crates/loopal-runtime/src/tool_pipeline.rs index 8091070a..c62d2c27 100644 --- a/crates/loopal-runtime/src/tool_pipeline.rs +++ b/crates/loopal-runtime/src/tool_pipeline.rs @@ -2,7 +2,7 @@ use std::time::Instant; use loopal_config::HookEvent; use loopal_error::{LoopalError, Result}; -use loopal_hooks::run_hook; +use loopal_hooks::{HookContext, HookOutput, PermissionOverride}; use loopal_kernel::Kernel; use loopal_tool_api::{ToolContext, ToolResult, handle_overflow}; use serde_json::Value; @@ -14,7 +14,10 @@ const MAX_RESULT_LINES: usize = 2000; const MAX_RESULT_BYTES: usize = 100_000; /// Execute a tool through the full pipeline: -/// pre-hooks -> execute -> overflow-to-file -> post-hooks. +/// pre-hooks → execute → overflow-to-file → post-hooks. +/// +/// Pre-hooks can reject (PermissionOverride::Deny) or modify input (updated_input). +/// Post-hooks can inject feedback (additional_context) into the tool result. pub async fn execute_tool( kernel: &Kernel, name: &str, @@ -26,37 +29,43 @@ pub async fn execute_tool( .get_tool(name) .ok_or_else(|| LoopalError::Tool(loopal_error::ToolError::NotFound(name.to_string())))?; - // Run pre-hooks - let pre_hooks = kernel.get_hooks(HookEvent::PreToolUse, Some(name)); - for hook_config in &pre_hooks { - let hook_data = serde_json::json!({ - "tool_name": name, - "tool_input": input, - }); - match run_hook(hook_config, hook_data).await { - Ok(result) => { - if !result.is_success() { - warn!( - tool = name, - exit_code = result.exit_code, - "pre-hook rejected" - ); - return Ok(ToolResult::error(format!( - "Pre-hook rejected: {}", - result.stderr.trim() - ))); - } - } - Err(e) => { - warn!(tool = name, error = %e, "pre-hook failed"); - return Ok(ToolResult::error(format!("Pre-hook error: {e}"))); + // ── Pre-hooks ────────────────────────────────────────────── + let pre_outputs = kernel + .hook_service() + .run_hooks( + HookEvent::PreToolUse, + &HookContext { + tool_name: Some(name), + tool_input: Some(&input), + ..Default::default() + }, + ) + .await; + + // Check for rejections and input modifications. + let mut effective_input = input; + let mut input_updated = false; + for out in &pre_outputs { + if let Some(PermissionOverride::Deny { ref reason }) = out.permission { + warn!(tool = name, %reason, "pre-hook rejected"); + return Ok(ToolResult::error(format!("Pre-hook rejected: {reason}"))); + } + if let Some(ref updated) = out.updated_input { + if input_updated { + warn!( + tool = name, + "multiple pre-hooks modified input, later override wins" + ); } + effective_input = updated.clone(); + input_updated = true; } } + // ── Execute ──────────────────────────────────────────────── debug!(tool = name, "executing tool"); let start = Instant::now(); - let result = tool.execute(input.clone(), ctx).await?; + let result = tool.execute(effective_input.clone(), ctx).await?; let duration = start.elapsed(); info!( tool = name, @@ -66,7 +75,7 @@ pub async fn execute_tool( "tool pipeline exec" ); - // Overflow-to-file: save large outputs to disk, return preview + path. + // ── Overflow-to-file ─────────────────────────────────────── let overflow = handle_overflow(&result.content, MAX_RESULT_LINES, MAX_RESULT_BYTES, name); let result = if overflow.overflowed { warn!( @@ -83,18 +92,37 @@ pub async fn execute_tool( result }; - let post_hooks = kernel.get_hooks(HookEvent::PostToolUse, Some(name)); - for hook_config in &post_hooks { - let hook_data = serde_json::json!({ - "tool_name": name, - "tool_input": input, - "tool_output": result.content, - "is_error": result.is_error, - }); - if let Err(e) = run_hook(hook_config, hook_data).await { - warn!(tool = name, error = %e, "post-hook failed"); - } - } + // ── Post-hooks ───────────────────────────────────────────── + let post_outputs = kernel + .hook_service() + .run_hooks( + HookEvent::PostToolUse, + &HookContext { + tool_name: Some(name), + tool_input: Some(&effective_input), + tool_output: Some(&result.content), + is_error: Some(result.is_error), + ..Default::default() + }, + ) + .await; + let result = append_post_hook_feedback(result, &post_outputs); Ok(result) } + +/// Collect `additional_context` from post-hook outputs and append to tool result. +fn append_post_hook_feedback(mut result: ToolResult, outputs: &[HookOutput]) -> ToolResult { + let feedback: Vec<&str> = outputs + .iter() + .filter_map(|o| o.additional_context.as_deref()) + .collect(); + + if feedback.is_empty() { + return result; + } + + result.content.push_str("\n\n[POST-HOOK FEEDBACK]\n"); + result.content.push_str(&feedback.join("\n---\n")); + result +} diff --git a/crates/loopal-runtime/tests/agent_loop/auto_mode_degradation_test.rs b/crates/loopal-runtime/tests/agent_loop/auto_mode_degradation_test.rs index 0c65e6fe..283d7a72 100644 --- a/crates/loopal-runtime/tests/agent_loop/auto_mode_degradation_test.rs +++ b/crates/loopal-runtime/tests/agent_loop/auto_mode_degradation_test.rs @@ -115,6 +115,8 @@ async fn no_provider_denies_gracefully() { memory_channel: None, scheduled_rx: None, auto_classifier: Some(classifier), + harness: loopal_config::HarnessConfig::default(), + rewake_rx: None, }; let mut runner = AgentLoopRunner::new(params); diff --git a/crates/loopal-runtime/tests/agent_loop/auto_mode_helpers.rs b/crates/loopal-runtime/tests/agent_loop/auto_mode_helpers.rs index a55ba10f..b1ed2bb7 100644 --- a/crates/loopal-runtime/tests/agent_loop/auto_mode_helpers.rs +++ b/crates/loopal-runtime/tests/agent_loop/auto_mode_helpers.rs @@ -130,6 +130,8 @@ pub fn make_auto_runner_with_setup( memory_channel: None, scheduled_rx: None, auto_classifier: Some(classifier), + harness: loopal_config::HarnessConfig::default(), + rewake_rx: None, }; (AgentLoopRunner::new(params), event_rx) } diff --git a/crates/loopal-runtime/tests/agent_loop/drain_pending_test.rs b/crates/loopal-runtime/tests/agent_loop/drain_pending_test.rs index 746d56eb..9a370f17 100644 --- a/crates/loopal-runtime/tests/agent_loop/drain_pending_test.rs +++ b/crates/loopal-runtime/tests/agent_loop/drain_pending_test.rs @@ -138,6 +138,8 @@ async fn test_subagent_drains_pending_before_exit() { memory_channel: None, scheduled_rx: None, auto_classifier: None, + harness: loopal_config::HarnessConfig::default(), + rewake_rx: None, }; // Drain events in background so channels don't block diff --git a/crates/loopal-runtime/tests/agent_loop/input_edge_test.rs b/crates/loopal-runtime/tests/agent_loop/input_edge_test.rs index 9de05528..6045906e 100644 --- a/crates/loopal-runtime/tests/agent_loop/input_edge_test.rs +++ b/crates/loopal-runtime/tests/agent_loop/input_edge_test.rs @@ -56,6 +56,8 @@ fn test_model_info_defaults_for_unknown_model() { memory_channel: None, scheduled_rx: None, auto_classifier: None, + harness: loopal_config::HarnessConfig::default(), + rewake_rx: None, }; let runner = AgentLoopRunner::new(params); diff --git a/crates/loopal-runtime/tests/agent_loop/integration_test.rs b/crates/loopal-runtime/tests/agent_loop/integration_test.rs index e25ba761..a79ad7b0 100644 --- a/crates/loopal-runtime/tests/agent_loop/integration_test.rs +++ b/crates/loopal-runtime/tests/agent_loop/integration_test.rs @@ -61,6 +61,8 @@ async fn test_agent_loop_immediate_channel_close() { memory_channel: None, scheduled_rx: None, auto_classifier: None, + harness: loopal_config::HarnessConfig::default(), + rewake_rx: None, }; // Drop senders to close channels diff --git a/crates/loopal-runtime/tests/agent_loop/mock_provider.rs b/crates/loopal-runtime/tests/agent_loop/mock_provider.rs index 3070223c..1327f6ac 100644 --- a/crates/loopal-runtime/tests/agent_loop/mock_provider.rs +++ b/crates/loopal-runtime/tests/agent_loop/mock_provider.rs @@ -45,6 +45,8 @@ fn build_params( memory_channel: None, scheduled_rx: None, auto_classifier: None, + harness: loopal_config::HarnessConfig::default(), + rewake_rx: None, } } diff --git a/crates/loopal-runtime/tests/agent_loop/mod.rs b/crates/loopal-runtime/tests/agent_loop/mod.rs index 61c779a9..569f31be 100644 --- a/crates/loopal-runtime/tests/agent_loop/mod.rs +++ b/crates/loopal-runtime/tests/agent_loop/mod.rs @@ -95,6 +95,8 @@ pub fn make_runner() -> (AgentLoopRunner, mpsc::Receiver) { memory_channel: None, scheduled_rx: None, auto_classifier: None, + harness: loopal_config::HarnessConfig::default(), + rewake_rx: None, }; (AgentLoopRunner::new(params), event_rx) } @@ -139,6 +141,8 @@ pub fn make_runner_with_channels() -> ( memory_channel: None, scheduled_rx: None, auto_classifier: None, + harness: loopal_config::HarnessConfig::default(), + rewake_rx: None, }; ( AgentLoopRunner::new(params), diff --git a/crates/loopal-runtime/tests/agent_loop/model_routing_test.rs b/crates/loopal-runtime/tests/agent_loop/model_routing_test.rs index 9815e244..073d0104 100644 --- a/crates/loopal-runtime/tests/agent_loop/model_routing_test.rs +++ b/crates/loopal-runtime/tests/agent_loop/model_routing_test.rs @@ -54,6 +54,8 @@ fn make_runner_with_routing( memory_channel: None, scheduled_rx: None, auto_classifier: None, + harness: loopal_config::HarnessConfig::default(), + rewake_rx: None, }; (AgentLoopRunner::new(params), event_rx) } @@ -138,6 +140,8 @@ fn test_model_routing_default_override_via_config_model() { memory_channel: None, scheduled_rx: None, auto_classifier: None, + harness: loopal_config::HarnessConfig::default(), + rewake_rx: None, }; let (runner, _rx) = (AgentLoopRunner::new(params), event_rx); diff --git a/crates/loopal-runtime/tests/agent_loop/permission_test_ext.rs b/crates/loopal-runtime/tests/agent_loop/permission_test_ext.rs index 65f0dd2f..ae8dabd3 100644 --- a/crates/loopal-runtime/tests/agent_loop/permission_test_ext.rs +++ b/crates/loopal-runtime/tests/agent_loop/permission_test_ext.rs @@ -133,6 +133,8 @@ async fn test_check_permission_channel_closed_denies() { memory_channel: None, scheduled_rx: None, auto_classifier: None, + harness: loopal_config::HarnessConfig::default(), + rewake_rx: None, }; let runner = AgentLoopRunner::new(params); diff --git a/crates/loopal-runtime/tests/agent_loop/turn_completion_test.rs b/crates/loopal-runtime/tests/agent_loop/turn_completion_test.rs index d4adfe97..b5686b87 100644 --- a/crates/loopal-runtime/tests/agent_loop/turn_completion_test.rs +++ b/crates/loopal-runtime/tests/agent_loop/turn_completion_test.rs @@ -102,6 +102,8 @@ pub(crate) fn make_multi_runner( memory_channel: None, scheduled_rx: None, auto_classifier: None, + harness: loopal_config::HarnessConfig::default(), + rewake_rx: None, }; (AgentLoopRunner::new(params), event_rx) } diff --git a/crates/loopal-runtime/tests/suite/tool_pipeline_hooks_test.rs b/crates/loopal-runtime/tests/suite/tool_pipeline_hooks_test.rs index 2aa6de4f..8eb64ac8 100644 --- a/crates/loopal-runtime/tests/suite/tool_pipeline_hooks_test.rs +++ b/crates/loopal-runtime/tests/suite/tool_pipeline_hooks_test.rs @@ -38,6 +38,13 @@ async fn test_passing_pre_hook() { command: "echo ok".to_string(), tool_filter: None, timeout_ms: 5000, + hook_type: Default::default(), + url: None, + headers: Default::default(), + prompt: None, + model: None, + condition: None, + id: None, }]); let (path, ctx) = temp_file("tool_pre_hook_pass.txt", "pre-hook pass content"); let result = execute_tool( @@ -61,6 +68,13 @@ async fn test_failing_pre_hook() { command: "echo 'denied by hook' >&2; exit 1".to_string(), tool_filter: None, timeout_ms: 5000, + hook_type: Default::default(), + url: None, + headers: Default::default(), + prompt: None, + model: None, + condition: None, + id: None, }]); let (path, ctx) = temp_file("tool_pre_hook_fail.txt", "should not read this"); let result = execute_tool( @@ -84,6 +98,13 @@ async fn test_post_hook_failure_ignored() { command: "exit 1".to_string(), tool_filter: None, timeout_ms: 5000, + hook_type: Default::default(), + url: None, + headers: Default::default(), + prompt: None, + model: None, + condition: None, + id: None, }]); let (path, ctx) = temp_file("tool_post_hook_fail.txt", "post hook test content"); let result = execute_tool( @@ -107,6 +128,13 @@ async fn test_filtered_pre_hook_not_matching() { command: "exit 1".to_string(), tool_filter: Some(vec!["Bash".to_string()]), timeout_ms: 5000, + hook_type: Default::default(), + url: None, + headers: Default::default(), + prompt: None, + model: None, + condition: None, + id: None, }]); let (path, ctx) = temp_file("tool_filtered_hook.txt", "filtered hook content"); let result = execute_tool( @@ -130,12 +158,26 @@ async fn test_both_pre_and_post_hooks() { command: "echo pre-hook-ok".to_string(), tool_filter: None, timeout_ms: 5000, + hook_type: Default::default(), + url: None, + headers: Default::default(), + prompt: None, + model: None, + condition: None, + id: None, }, HookConfig { event: HookEvent::PostToolUse, command: "echo post-hook-ok".to_string(), tool_filter: None, timeout_ms: 5000, + hook_type: Default::default(), + url: None, + headers: Default::default(), + prompt: None, + model: None, + condition: None, + id: None, }, ]); let (path, ctx) = temp_file("tool_both_hooks.txt", "both hooks content"); diff --git a/crates/loopal-session/src/agent_handler.rs b/crates/loopal-session/src/agent_handler.rs index 7dc01887..76925e1c 100644 --- a/crates/loopal-session/src/agent_handler.rs +++ b/crates/loopal-session/src/agent_handler.rs @@ -192,7 +192,8 @@ pub(crate) fn apply_agent_event( } AgentEventPayload::SubAgentSpawned { .. } | AgentEventPayload::MessageRouted { .. } - | AgentEventPayload::TurnDiffSummary { .. } => {} + | AgentEventPayload::TurnDiffSummary { .. } + | AgentEventPayload::TurnCompleted { .. } => {} AgentEventPayload::AutoModeDecision { tool_name, decision, diff --git a/crates/loopal-test-support/src/wiring.rs b/crates/loopal-test-support/src/wiring.rs index 46fc1fa1..67c18e1b 100644 --- a/crates/loopal-test-support/src/wiring.rs +++ b/crates/loopal-test-support/src/wiring.rs @@ -158,6 +158,8 @@ pub(crate) async fn wire(builder: HarnessBuilder) -> (SpawnedHarness, AgentLoopR memory_channel: None, scheduled_rx: Some(scheduled_rx), auto_classifier: None, + harness: loopal_config::HarnessConfig::default(), + rewake_rx: None, }; let harness = SpawnedHarness { diff --git a/crates/loopal-tui/tests/suite/e2e_hooks_test.rs b/crates/loopal-tui/tests/suite/e2e_hooks_test.rs index f8b904fa..27be2c2b 100644 --- a/crates/loopal-tui/tests/suite/e2e_hooks_test.rs +++ b/crates/loopal-tui/tests/suite/e2e_hooks_test.rs @@ -37,6 +37,13 @@ async fn test_pre_tool_hook_executes() { command: script.to_str().unwrap().to_string(), tool_filter: None, timeout_ms: 5000, + hook_type: Default::default(), + url: None, + headers: Default::default(), + prompt: None, + model: None, + condition: None, + id: None, }]; let calls = vec![ @@ -84,6 +91,13 @@ async fn test_hook_failure_blocks_tool() { command: script.to_str().unwrap().to_string(), tool_filter: None, timeout_ms: 5000, + hook_type: Default::default(), + url: None, + headers: Default::default(), + prompt: None, + model: None, + condition: None, + id: None, }]; let calls = vec![ @@ -117,6 +131,13 @@ async fn test_post_hook_output_captured() { command: script.to_str().unwrap().to_string(), tool_filter: None, timeout_ms: 5000, + hook_type: Default::default(), + url: None, + headers: Default::default(), + prompt: None, + model: None, + condition: None, + id: None, }]; let calls = vec![