// src/agent/hooks/dispatch.rs //! //! HookRegistry 调度方法——所有 run_* 生命周期事件分发。 use std::time::Duration; use futures_util::future::join_all; use tokio::time::timeout as tokio_timeout; use tracing::{info, warn}; use super::{ AgentHook, AsyncAgentHook, BlockingError, HookEvent, PermissionDeniedContext, PermissionRequestAction, PermissionRequestContext, PostCompactContext, PostToolUseAction, PostToolUseContext, PostToolUseFailureContext, PostToolUseResult, PreCompactContext, PreToolUseAction, PreToolUseContext, PreToolUseResult, SessionStartContext, SessionStopContext, StepCompleteContext, SubagentStartContext, SubagentStopContext, TaggedContext, DEFAULT_HOOK_TIMEOUT, }; use serde_json; impl super::HookRegistry { // ── 便捷调用方法 ── /// 调度匹配指定事件的异步 hooks(fire-and-forget,5s dispatch 超时) async fn dispatch_async_hooks_for( &self, event: HookEvent, _session_id: &str, _tool_name: &str, _tool_args: &serde_json::Value, post_ctx: Option<&PostToolUseContext>, failure_ctx: Option<&PostToolUseFailureContext>, ) { let async_hooks: Vec<&dyn AsyncAgentHook> = self .async_event_index .get(&event) .map(|indices| { indices .iter() .filter_map(|&idx| self.async_hooks.get(idx).map(|b| b.as_ref())) .filter(|_hook| { // Async hooks can declare filtering too via name convention; // for now, all matching the event fire. true }) .collect() }) .unwrap_or_default(); if async_hooks.is_empty() { return; } let timeout_dur = Duration::from_secs(5); // async dispatch timeout // Build futures — clone contexts so futures own their data and satisfy 'static. match event { HookEvent::PostToolUse if post_ctx.is_some() => { let ctx = post_ctx.unwrap().clone(); for hook in async_hooks { let ctx = ctx.clone(); let _ = tokio_timeout(timeout_dur, hook.on_post_tool_use_async(ctx)).await; } } HookEvent::PostToolUseFailure if failure_ctx.is_some() => { let ctx = failure_ctx.unwrap().clone(); for hook in async_hooks { let ctx = ctx.clone(); let _ = tokio_timeout(timeout_dur, hook.on_post_tool_use_failure_async(ctx)).await; } } _ => {} } } /// 调用所有 on_session_start hooks(并行执行,含 session hooks) pub async fn run_on_session_start(&self, ctx: &SessionStartContext) { let hooks = self.collect_hooks_for(HookEvent::OnSessionStart, &ctx.session_id); if hooks.is_empty() { return; } let ctx_clone = ctx.clone(); let futures: Vec<_> = hooks .iter() .map(|hook| { let ctx = ctx_clone.clone(); let timeout_dur = hook.timeout().unwrap_or(DEFAULT_HOOK_TIMEOUT); async move { let _ = tokio_timeout(timeout_dur, hook.on_session_start(&ctx)).await; } }) .collect(); join_all(futures).await; } /// 调用所有 pre_tool_use hooks(并行执行,各自有独立超时)。 /// /// 所有订阅了 PreToolUse 的 hooks(含 session hooks)并行运行,每个 hook 包装在 /// `tokio::time::timeout` 中。完成后聚合所有结果: /// - Block → 收集到 blocking_errors /// - MutateInput → 累积 additional_contexts 并更新 final_args /// - PermissionRequired → 记录 /// - 超时 → 记录 warning,视为 non-blocking error pub async fn run_pre_tool_use(&self, ctx: &PreToolUseContext) -> PreToolUseResult { let hooks = self.collect_tool_hooks_for( HookEvent::PreToolUse, &ctx.session_id, &ctx.tool_name, &ctx.tool_args, ); // Fast path: 单 hook 或空 → 顺序执行 if hooks.len() <= 1 { return self.run_pre_tool_use_sequential(ctx).await; } let ctx_clone = ctx.clone(); let futures: Vec<_> = hooks .iter() .map(|hook| { let ctx = ctx_clone.clone(); let timeout_dur = hook.timeout().unwrap_or(DEFAULT_HOOK_TIMEOUT); let hook_name = hook.name().to_string(); async move { ( hook_name, tokio_timeout(timeout_dur, hook.pre_tool_use(&ctx)).await, ) } }) .collect(); // 并行执行所有 hooks let results = join_all(futures).await; // 聚合结果 let mut additional_contexts: Vec = Vec::new(); let mut tagged_contexts: Vec = Vec::new(); let mut final_args = ctx.tool_args.clone(); let mut final_action = PreToolUseAction::Continue; let mut blocking_errors: Vec = Vec::new(); for (hook_name, result) in results { match result { Ok(action) => { match &action { PreToolUseAction::Block { reason } => { warn!( "[Hooks] {hook_name} 阻止了工具 {} 的执行: {reason}", ctx.tool_name ); blocking_errors.push(BlockingError::new(hook_name, reason.clone())); } PreToolUseAction::MutateInput { updated_args, additional_context, } => { info!( "[Hooks] {hook_name} 修改了工具 {} 的输入参数", ctx.tool_name ); final_args = updated_args.clone(); if let Some(ctx_str) = additional_context { additional_contexts.push(ctx_str.clone()); tagged_contexts.push(TaggedContext::new( &hook_name, HookEvent::PreToolUse, ctx_str, )); } } PreToolUseAction::PermissionRequired { .. } => { info!( "[Hooks] {hook_name} 请求了工具 {} 的权限检查", ctx.tool_name ); } PreToolUseAction::Continue => {} } if !matches!(action, PreToolUseAction::Continue) { final_action = action; } } Err(_elapsed) => { warn!("[Hooks] {hook_name} 超时(PreToolUse),跳过其反馈"); } } } let additional_context = if additional_contexts.is_empty() { None } else { Some(additional_contexts.join("\n")) }; PreToolUseResult { action: final_action, additional_contexts, additional_context, blocking_errors, final_args, tagged_contexts, } } /// 顺序执行 pre_tool_use hooks(含 session hooks) async fn run_pre_tool_use_sequential(&self, ctx: &PreToolUseContext) -> PreToolUseResult { let mut additional_contexts: Vec = Vec::new(); let mut tagged_contexts: Vec = Vec::new(); let mut final_args = ctx.tool_args.clone(); let mut final_action = PreToolUseAction::Continue; let mut blocking_errors: Vec = Vec::new(); // 全局 hooks(通过 collect_tool_hooks_for 应用 match_filter 过滤) let hooks = self.collect_tool_hooks_for( HookEvent::PreToolUse, &ctx.session_id, &ctx.tool_name, &ctx.tool_args, ); for hook in &hooks { let action = hook.pre_tool_use(ctx).await; Self::process_pre_tool_action( &action, hook.name(), &ctx.tool_name, &mut blocking_errors, &mut final_args, &mut additional_contexts, &mut tagged_contexts, &mut final_action, ); } let additional_context = if additional_contexts.is_empty() { None } else { Some(additional_contexts.join("\n")) }; PreToolUseResult { action: final_action, additional_contexts, additional_context, blocking_errors, final_args, tagged_contexts, } } /// 调用所有 post_tool_use hooks(并行执行,含 session hooks,各自有独立超时)。 pub async fn run_post_tool_use(&self, ctx: &PostToolUseContext) -> PostToolUseResult { // 同时调度异步 hooks(fire-and-forget) self.dispatch_async_hooks_for( HookEvent::PostToolUse, &ctx.session_id, &ctx.tool_name, &ctx.tool_args, Some(ctx), None, ) .await; let hooks = self.collect_tool_hooks_for( HookEvent::PostToolUse, &ctx.session_id, &ctx.tool_name, &ctx.tool_args, ); // Fast path: 单 hook 或空 → 顺序执行 if hooks.len() <= 1 { return self.run_post_tool_use_sequential(ctx).await; } let ctx_clone = ctx.clone(); let futures: Vec<_> = hooks .iter() .map(|hook| { let ctx = ctx_clone.clone(); let timeout_dur = hook.timeout().unwrap_or(DEFAULT_HOOK_TIMEOUT); let hook_name = hook.name().to_string(); async move { ( hook_name, tokio_timeout(timeout_dur, hook.post_tool_use(&ctx)).await, ) } }) .collect(); let results = join_all(futures).await; let mut final_content = ctx.output_content.clone(); let mut additional_contexts: Vec = Vec::new(); let mut tagged_contexts: Vec = Vec::new(); let mut warnings: Vec = Vec::new(); let mut post_permission_requests: Vec<(String, String)> = Vec::new(); let mut metadata: std::collections::HashMap = std::collections::HashMap::new(); for (hook_name, result) in results { match result { Ok(action) => match &action { PostToolUseAction::MutateOutput { updated_content: ref uc, additional_context: ref ac, } => { info!("[Hooks] {hook_name} 修改了工具 {} 的输出", ctx.tool_name); final_content = uc.clone(); if let Some(ctx_str) = ac { additional_contexts.push(ctx_str.clone()); tagged_contexts.push(TaggedContext::new( &hook_name, HookEvent::PostToolUse, ctx_str, )); } } PostToolUseAction::Warning { message, truncate_output, } => { warn!( "[Hooks] {hook_name} 发出了对 {} 的警告: {message}", ctx.tool_name ); warnings.push(message.clone()); if *truncate_output { final_content = final_content.chars().take(1000).collect::() + "\n\n[输出已由 Hook 截断]"; } } PostToolUseAction::PermissionRequired { permission, tool_name, } => { info!("[Hooks] {hook_name} 事后请求工具 {tool_name} 的权限: {permission}"); post_permission_requests.push((tool_name.clone(), permission.clone())); } PostToolUseAction::Metadata { key, value } => { metadata.insert(key.clone(), value.clone()); } PostToolUseAction::Continue => {} }, Err(_elapsed) => { warn!("[Hooks] {hook_name} 超时(PostToolUse),跳过其反馈"); } } } PostToolUseResult { final_content, additional_contexts, tagged_contexts, warnings, post_permission_requests, metadata, } } /// 顺序执行 post_tool_use hooks(含 session hooks) async fn run_post_tool_use_sequential(&self, ctx: &PostToolUseContext) -> PostToolUseResult { let mut final_content = ctx.output_content.clone(); let mut additional_contexts: Vec = Vec::new(); let mut tagged_contexts: Vec = Vec::new(); let mut warnings: Vec = Vec::new(); let mut post_permission_requests: Vec<(String, String)> = Vec::new(); let mut metadata: std::collections::HashMap = std::collections::HashMap::new(); let hooks = self.collect_tool_hooks_for( HookEvent::PostToolUse, &ctx.session_id, &ctx.tool_name, &ctx.tool_args, ); for hook in &hooks { let action = hook.post_tool_use(ctx).await; match action { PostToolUseAction::MutateOutput { updated_content, additional_context, } => { final_content = updated_content; if let Some(ctx_str) = additional_context { additional_contexts.push(ctx_str.clone()); tagged_contexts.push(TaggedContext::new( hook.name(), HookEvent::PostToolUse, ctx_str, )); } } PostToolUseAction::Warning { message, truncate_output, } => { warnings.push(message); if truncate_output { final_content = final_content.chars().take(1000).collect::() + "\n\n[输出已由 Hook 截断]"; } } PostToolUseAction::PermissionRequired { permission, tool_name, } => { post_permission_requests.push((tool_name, permission)); } PostToolUseAction::Metadata { key, value } => { metadata.insert(key, value.clone()); } PostToolUseAction::Continue => {} } } PostToolUseResult { final_content, additional_contexts, tagged_contexts, warnings, post_permission_requests, metadata, } } /// 调用所有 post_tool_use_failure hooks(并行执行,含 session hooks) pub async fn run_on_post_tool_use_failure( &self, ctx: &PostToolUseFailureContext, ) -> PostToolUseResult { // 同时调度异步 hooks(fire-and-forget) self.dispatch_async_hooks_for( HookEvent::PostToolUseFailure, &ctx.session_id, &ctx.tool_name, &ctx.tool_args, None, Some(ctx), ) .await; let hooks = self.collect_tool_hooks_for( HookEvent::PostToolUseFailure, &ctx.session_id, &ctx.tool_name, &ctx.tool_args, ); // Fast path: 单 hook 或空 → 顺序执行 if hooks.len() <= 1 { let mut final_content = String::new(); let mut additional_contexts: Vec = Vec::new(); let mut tagged_contexts: Vec = Vec::new(); for hook in &hooks { let action = hook.on_post_tool_use_failure(ctx).await; match action { PostToolUseAction::MutateOutput { updated_content, additional_context, } => { final_content = updated_content; if let Some(ctx_str) = additional_context { additional_contexts.push(ctx_str.clone()); tagged_contexts.push(TaggedContext::new( hook.name(), HookEvent::PostToolUseFailure, ctx_str, )); } } PostToolUseAction::Warning { message, truncate_output, } => { if truncate_output { final_content = final_content.chars().take(1000).collect(); } tagged_contexts.push(TaggedContext::new( hook.name(), HookEvent::PostToolUseFailure, format!("Warning: {message}"), )); } PostToolUseAction::PermissionRequired { .. } | PostToolUseAction::Metadata { .. } | PostToolUseAction::Continue => {} } } return PostToolUseResult { final_content, additional_contexts, tagged_contexts, warnings: Vec::new(), post_permission_requests: Vec::new(), metadata: std::collections::HashMap::new(), }; } let ctx_clone = ctx.clone(); let futures: Vec<_> = hooks .iter() .map(|hook| { let ctx = ctx_clone.clone(); let timeout_dur = hook.timeout().unwrap_or(DEFAULT_HOOK_TIMEOUT); let hook_name = hook.name().to_string(); async move { ( hook_name, tokio_timeout(timeout_dur, hook.on_post_tool_use_failure(&ctx)).await, ) } }) .collect(); let results = join_all(futures).await; let mut final_content = String::new(); let mut additional_contexts: Vec = Vec::new(); let mut tagged_contexts: Vec = Vec::new(); for (hook_name, result) in results { match result { Ok(action) => match action { PostToolUseAction::MutateOutput { updated_content, additional_context, } => { final_content = updated_content; if let Some(ctx_str) = additional_context { additional_contexts.push(ctx_str.clone()); tagged_contexts.push(TaggedContext::new( &hook_name, HookEvent::PostToolUseFailure, ctx_str, )); } } PostToolUseAction::Warning { message, truncate_output, } => { if truncate_output { final_content = final_content.chars().take(1000).collect(); } tagged_contexts.push(TaggedContext::new( &hook_name, HookEvent::PostToolUseFailure, format!("Warning: {message}"), )); } PostToolUseAction::PermissionRequired { .. } | PostToolUseAction::Metadata { .. } | PostToolUseAction::Continue => {} }, Err(_elapsed) => { // 超时已在警告中体现,跳过该 hook 的反馈 } } } PostToolUseResult { final_content, additional_contexts, tagged_contexts, warnings: Vec::new(), post_permission_requests: Vec::new(), metadata: std::collections::HashMap::new(), } } /// 调用所有 on_step_complete hooks(并行执行,含 session hooks) pub async fn run_on_step_complete(&self, ctx: &StepCompleteContext) { let hooks = self.collect_hooks_for(HookEvent::OnStepComplete, &ctx.session_id); if hooks.is_empty() { return; } let ctx_clone = ctx.clone(); let futures: Vec<_> = hooks .iter() .map(|hook| { let ctx = ctx_clone.clone(); let timeout_dur = hook.timeout().unwrap_or(DEFAULT_HOOK_TIMEOUT); async move { let _ = tokio_timeout(timeout_dur, hook.on_step_complete(&ctx)).await; } }) .collect(); join_all(futures).await; } /// 调用所有 on_session_stop hooks(顺序执行 + 自动清理 session hooks) pub async fn run_on_session_stop(&self, ctx: &SessionStopContext<'_>) { // 全局 hooks for &idx in self.indices_for(HookEvent::OnSessionStop) { if let Some(hook) = self.hooks.get(idx) { let _ = tokio_timeout( hook.timeout().unwrap_or(DEFAULT_HOOK_TIMEOUT), hook.on_session_stop(ctx), ) .await; } } // Session hooks(Stop 事件也通知它们) for hook in self.get_session_hooks(&ctx.session_id) { let _ = tokio_timeout( hook.timeout().unwrap_or(DEFAULT_HOOK_TIMEOUT), hook.on_session_stop(ctx), ) .await; } // 注意:clear_session_hooks 需要 &mut self,但此方法是 &self。 // 调用方应在 run_on_session_stop 返回后主动调用 clear_session_hooks。 } /// 调用所有 on_subagent_start hooks(并行执行,含 session hooks) pub async fn run_on_subagent_start(&self, ctx: &SubagentStartContext) { let hooks = self.collect_hooks_for(HookEvent::OnSubagentStart, &ctx.parent_session_id); if hooks.is_empty() { return; } let ctx_clone = ctx.clone(); let futures: Vec<_> = hooks .iter() .map(|hook| { let ctx = ctx_clone.clone(); let timeout_dur = hook.timeout().unwrap_or(DEFAULT_HOOK_TIMEOUT); async move { let _ = tokio_timeout(timeout_dur, hook.on_subagent_start(&ctx)).await; } }) .collect(); join_all(futures).await; } /// 调用所有 on_subagent_stop hooks(并行执行,含 session hooks) pub async fn run_on_subagent_stop(&self, ctx: &SubagentStopContext) { let hooks = self.collect_hooks_for(HookEvent::OnSubagentStop, &ctx.parent_session_id); if hooks.is_empty() { return; } let ctx_clone = ctx.clone(); let futures: Vec<_> = hooks .iter() .map(|hook| { let ctx = ctx_clone.clone(); let timeout_dur = hook.timeout().unwrap_or(DEFAULT_HOOK_TIMEOUT); async move { let _ = tokio_timeout(timeout_dur, hook.on_subagent_stop(&ctx)).await; } }) .collect(); join_all(futures).await; } /// 调用所有 on_pre_compact hooks(并行执行,含 session hooks) pub async fn run_on_pre_compact(&self, ctx: &PreCompactContext) { let hooks = self.collect_hooks_for(HookEvent::OnPreCompact, &ctx.session_id); if hooks.is_empty() { return; } let ctx_clone = ctx.clone(); let futures: Vec<_> = hooks .iter() .map(|hook| { let ctx = ctx_clone.clone(); let timeout_dur = hook.timeout().unwrap_or(DEFAULT_HOOK_TIMEOUT); async move { let _ = tokio_timeout(timeout_dur, hook.on_pre_compact(&ctx)).await; } }) .collect(); join_all(futures).await; } /// 调用所有 on_post_compact hooks(并行执行,含 session hooks) pub async fn run_on_post_compact(&self, ctx: &PostCompactContext) { let hooks = self.collect_hooks_for(HookEvent::OnPostCompact, &ctx.session_id); if hooks.is_empty() { return; } let ctx_clone = ctx.clone(); let futures: Vec<_> = hooks .iter() .map(|hook| { let ctx = ctx_clone.clone(); let timeout_dur = hook.timeout().unwrap_or(DEFAULT_HOOK_TIMEOUT); async move { let _ = tokio_timeout(timeout_dur, hook.on_post_compact(&ctx)).await; } }) .collect(); join_all(futures).await; } /// 分发 PermissionRequest 事件。 /// /// 并行调用所有匹配的 hook。每个 hook 返回 `PermissionRequestAction`: /// - `Continue` → 保持当前决策 /// - `Override` → 覆盖决策(第一个 Override 生效,后续不再检查) /// - `InjectContext` → 注入上下文但不改变决策 /// /// 返回 (final_action, additional_contexts)。 pub async fn run_on_permission_request( &self, ctx: &PermissionRequestContext, ) -> (PermissionRequestAction, Vec) { let hooks = self.collect_hooks_for(HookEvent::PermissionRequest, &ctx.session_id); if hooks.is_empty() { return (PermissionRequestAction::Continue, Vec::new()); } let mut final_action = PermissionRequestAction::Continue; let mut injected_contexts: Vec = Vec::new(); let mut overridden = false; let futures: Vec<_> = hooks .iter() .map(|hook| { let hook: &dyn AgentHook = *hook; let ctx = ctx.clone(); let timeout_dur = DEFAULT_HOOK_TIMEOUT; async move { let result = tokio_timeout(timeout_dur, hook.on_permission_request(&ctx)).await; match result { Ok(action) => Some((hook.name().to_string(), action)), Err(_) => { warn!("[HookDispatch] PermissionRequest hook {} 超时", hook.name()); None } } } }) .collect(); let results: Vec> = join_all(futures).await; for result in results.into_iter().flatten() { let (hook_name, action) = result; match action { PermissionRequestAction::Continue => { // 默认,不改变 } PermissionRequestAction::Override(decision) if !overridden => { overridden = true; final_action = PermissionRequestAction::Override(decision); info!( "[HookDispatch] PermissionRequest hook {} 覆盖决策", hook_name ); } PermissionRequestAction::Override(_) => { warn!( "[HookDispatch] PermissionRequest hook {}: 覆盖被忽略(已有其他 hook 先覆盖)", hook_name ); } PermissionRequestAction::InjectContext { ref content } => { injected_contexts.push(content.clone()); } } } (final_action, injected_contexts) } /// 分发 PermissionDenied 事件(fire-and-forget,仅用于审计日志)。 /// /// 并行调用所有匹配的 hook。返回值不影响执行流程。 pub async fn run_on_permission_denied(&self, ctx: &PermissionDeniedContext) { let hooks = self.collect_hooks_for(HookEvent::PermissionDenied, &ctx.session_id); if hooks.is_empty() { return; } let futures: Vec<_> = hooks .iter() .map(|hook| { let hook: &dyn AgentHook = *hook; let ctx = ctx.clone(); let timeout_dur = DEFAULT_HOOK_TIMEOUT; async move { let _ = tokio_timeout(timeout_dur, hook.on_permission_denied(&ctx)).await; } }) .collect(); join_all(futures).await; } }