ai_agent/utils/hooks/
async_hook_registry.rs

1// Source: ~/claudecode/openclaudecode/src/utils/hooks/AsyncHookRegistry.ts
2#![allow(dead_code)]
3#![allow(static_mut_refs)]
4
5use std::collections::HashMap;
6use std::sync::{Arc, Mutex};
7use tokio::time::{Duration, interval};
8
9/// Represents a hook event type
10#[derive(Debug, Clone, PartialEq, Eq)]
11pub enum HookEvent {
12    SessionStart,
13    Setup,
14    PreToolUse,
15    PostToolUse,
16    PostToolUseFailure,
17    PermissionDenied,
18    Notification,
19    UserPromptSubmit,
20    SessionEnd,
21    Stop,
22    StopFailure,
23    SubagentStart,
24    SubagentStop,
25    PreCompact,
26    PostCompact,
27    PermissionRequest,
28    TeammateIdle,
29    TaskCreated,
30    TaskCompleted,
31    Elicitation,
32    ElicitationResult,
33    ConfigChange,
34    InstructionsLoaded,
35    WorktreeCreate,
36    WorktreeRemove,
37    CwdChanged,
38    FileChanged,
39    StatusLine,
40    FileSuggestion,
41    Custom(String),
42}
43
44impl HookEvent {
45    pub fn as_str(&self) -> &str {
46        match self {
47            HookEvent::SessionStart => "SessionStart",
48            HookEvent::Setup => "Setup",
49            HookEvent::PreToolUse => "PreToolUse",
50            HookEvent::PostToolUse => "PostToolUse",
51            HookEvent::PostToolUseFailure => "PostToolUseFailure",
52            HookEvent::PermissionDenied => "PermissionDenied",
53            HookEvent::Notification => "Notification",
54            HookEvent::UserPromptSubmit => "UserPromptSubmit",
55            HookEvent::SessionEnd => "SessionEnd",
56            HookEvent::Stop => "Stop",
57            HookEvent::StopFailure => "StopFailure",
58            HookEvent::SubagentStart => "SubagentStart",
59            HookEvent::SubagentStop => "SubagentStop",
60            HookEvent::PreCompact => "PreCompact",
61            HookEvent::PostCompact => "PostCompact",
62            HookEvent::PermissionRequest => "PermissionRequest",
63            HookEvent::TeammateIdle => "TeammateIdle",
64            HookEvent::TaskCreated => "TaskCreated",
65            HookEvent::TaskCompleted => "TaskCompleted",
66            HookEvent::Elicitation => "Elicitation",
67            HookEvent::ElicitationResult => "ElicitationResult",
68            HookEvent::ConfigChange => "ConfigChange",
69            HookEvent::InstructionsLoaded => "InstructionsLoaded",
70            HookEvent::WorktreeCreate => "WorktreeCreate",
71            HookEvent::WorktreeRemove => "WorktreeRemove",
72            HookEvent::CwdChanged => "CwdChanged",
73            HookEvent::FileChanged => "FileChanged",
74            HookEvent::StatusLine => "StatusLine",
75            HookEvent::FileSuggestion => "FileSuggestion",
76            HookEvent::Custom(s) => s,
77        }
78    }
79}
80
81/// JSON output from an async hook
82#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
83#[serde(rename_all = "camelCase")]
84pub struct AsyncHookJsonOutput {
85    /// Timeout in seconds (0 means default)
86    #[serde(skip_serializing_if = "Option::is_none")]
87    pub async_timeout: Option<u64>,
88}
89
90/// JSON output from a sync hook
91#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
92pub struct SyncHookJsonOutput {
93    #[serde(flatten)]
94    pub extra: HashMap<String, serde_json::Value>,
95}
96
97/// Output from a shell command task
98#[derive(Clone)]
99pub struct TaskOutput {
100    stdout: Arc<Mutex<String>>,
101    stderr: Arc<Mutex<String>>,
102}
103
104impl TaskOutput {
105    pub fn new() -> Self {
106        Self {
107            stdout: Arc::new(Mutex::new(String::new())),
108            stderr: Arc::new(Mutex::new(String::new())),
109        }
110    }
111
112    pub async fn get_stdout(&self) -> String {
113        self.stdout.lock().unwrap().clone()
114    }
115
116    pub fn get_stderr(&self) -> String {
117        self.stderr.lock().unwrap().clone()
118    }
119
120    pub fn append_stdout(&self, data: &str) {
121        self.stdout.lock().unwrap().push_str(data);
122    }
123
124    pub fn append_stderr(&self, data: &str) {
125        self.stderr.lock().unwrap().push_str(data);
126    }
127}
128
129/// A shell command being executed
130pub struct ShellCommand {
131    pub status: ShellCommandStatus,
132    pub task_output: TaskOutput,
133    pub pid: Option<u32>,
134}
135
136impl ShellCommand {
137    pub fn cleanup(&self) {
138        if let Some(pid) = self.pid {
139            unsafe {
140                libc::kill(pid as i32, libc::SIGTERM);
141            }
142        }
143    }
144
145    pub fn kill(&mut self) {
146        self.status = ShellCommandStatus::Killed;
147        self.cleanup();
148    }
149}
150
151#[derive(Debug, Clone, PartialEq)]
152pub enum ShellCommandStatus {
153    Running,
154    Completed,
155    Killed,
156}
157
158/// A pending async hook in the registry
159pub struct PendingAsyncHook {
160    pub process_id: String,
161    pub hook_id: String,
162    pub hook_name: String,
163    pub hook_event: HookEvent,
164    pub tool_name: Option<String>,
165    pub plugin_id: Option<String>,
166    pub start_time: std::time::SystemTime,
167    pub timeout_ms: u64,
168    pub command: String,
169    pub response_attachment_sent: bool,
170    pub shell_command: Option<ShellCommand>,
171    pub progress_task_id: Option<u64>, // Simple ID for tracking (simplified)
172}
173
174/// Global registry state for pending async hooks
175struct AsyncHookRegistryState {
176    pending_hooks: HashMap<String, PendingAsyncHook>,
177}
178
179lazy_static::lazy_static! {
180    static ref ASYNC_HOOK_REGISTRY: Arc<Mutex<AsyncHookRegistryState>> = Arc::new(Mutex::new(
181        AsyncHookRegistryState {
182            pending_hooks: HashMap::new(),
183        }
184    ));
185}
186
187/// Parameters for registering a pending async hook
188pub struct RegisterPendingAsyncHookParams {
189    pub process_id: String,
190    pub hook_id: String,
191    pub async_response: AsyncHookJsonOutput,
192    pub hook_name: String,
193    pub hook_event: HookEvent,
194    pub command: String,
195    pub shell_command: ShellCommand,
196    pub tool_name: Option<String>,
197    pub plugin_id: Option<String>,
198}
199
200/// Register a pending async hook
201pub fn register_pending_async_hook(params: RegisterPendingAsyncHookParams) {
202    let timeout = params.async_response.async_timeout.unwrap_or(15) * 1000; // Default 15s, convert to ms
203
204    log_for_debugging(&format!(
205        "Hooks: Registering async hook {} ({}) with timeout {}ms",
206        params.process_id, params.hook_name, timeout
207    ));
208
209    let hook_id = params.hook_id.clone();
210    let hook_name = params.hook_name.clone();
211    let hook_event = params.hook_event.clone();
212    let process_id = params.process_id.clone();
213    let shell_task_output = params.shell_command.task_output.clone();
214
215    // Create progress interval that polls shell command output
216    let _progress_handle = start_hook_progress_interval(HookProgressParams {
217        hook_id: params.hook_id.clone(),
218        hook_name: params.hook_name.clone(),
219        hook_event: params.hook_event.clone(),
220        get_output: Arc::new(move || {
221            let task_output = shell_task_output.clone();
222            Box::pin(async move {
223                let stdout = task_output.get_stdout().await;
224                let stderr = task_output.get_stderr();
225                let output = format!("{}{}", stdout, stderr);
226                HookOutput {
227                    stdout,
228                    stderr,
229                    output,
230                }
231            })
232        }),
233        interval_ms: None,
234    });
235
236    let pending_hook = PendingAsyncHook {
237        process_id: params.process_id.clone(),
238        hook_id: params.hook_id,
239        hook_name: params.hook_name,
240        hook_event: params.hook_event,
241        tool_name: params.tool_name,
242        plugin_id: params.plugin_id,
243        start_time: std::time::SystemTime::now(),
244        timeout_ms: timeout,
245        command: params.command,
246        response_attachment_sent: false,
247        shell_command: Some(params.shell_command),
248        progress_task_id: None,
249    };
250
251    let mut registry = ASYNC_HOOK_REGISTRY.lock().unwrap();
252    registry
253        .pending_hooks
254        .insert(params.process_id, pending_hook);
255}
256
257/// Get all pending async hooks that haven't sent their response attachment
258pub fn get_pending_async_hooks() -> Vec<Arc<Mutex<PendingAsyncHook>>> {
259    let registry = ASYNC_HOOK_REGISTRY.lock().unwrap();
260    registry
261        .pending_hooks
262        .values()
263        .filter(|hook| !hook.response_attachment_sent)
264        .map(|hook| {
265            Arc::new(Mutex::new(PendingAsyncHook {
266                process_id: hook.process_id.clone(),
267                hook_id: hook.hook_id.clone(),
268                hook_name: hook.hook_name.clone(),
269                hook_event: hook.hook_event.clone(),
270                tool_name: hook.tool_name.clone(),
271                plugin_id: hook.plugin_id.clone(),
272                start_time: hook.start_time,
273                timeout_ms: hook.timeout_ms,
274                command: hook.command.clone(),
275                response_attachment_sent: hook.response_attachment_sent,
276                shell_command: None, // Can't clone shell command
277                progress_task_id: None,
278            }))
279        })
280        .collect()
281}
282
283pub struct HookProgressParams {
284    pub hook_id: String,
285    pub hook_name: String,
286    pub hook_event: HookEvent,
287    pub get_output: Arc<
288        dyn Fn() -> std::pin::Pin<Box<dyn std::future::Future<Output = HookOutput> + Send>>
289            + Send
290            + Sync,
291    >,
292    pub interval_ms: Option<u64>,
293}
294
295pub struct HookOutput {
296    pub stdout: String,
297    pub stderr: String,
298    pub output: String,
299}
300
301const MAX_PENDING_EVENTS: usize = 100;
302
303static mut EVENT_HANDLER: Option<Box<dyn Fn(HookExecutionEvent) + Send + Sync>> = None;
304static mut PENDING_EVENTS: Vec<HookExecutionEvent> = Vec::new();
305static mut ALL_HOOK_EVENTS_ENABLED: bool = false;
306
307// Always emitted hook events regardless of includeHookEvents option
308const ALWAYS_EMITTED_HOOK_EVENTS: [&str; 2] = ["SessionStart", "Setup"];
309
310#[derive(Debug, Clone)]
311pub enum HookExecutionEvent {
312    Started {
313        hook_id: String,
314        hook_name: String,
315        hook_event: String,
316    },
317    Progress {
318        hook_id: String,
319        hook_name: String,
320        hook_event: String,
321        stdout: String,
322        stderr: String,
323        output: String,
324    },
325    Response {
326        hook_id: String,
327        hook_name: String,
328        hook_event: String,
329        output: String,
330        stdout: String,
331        stderr: String,
332        exit_code: Option<i32>,
333        outcome: HookOutcome,
334    },
335}
336
337#[derive(Debug, Clone)]
338pub enum HookOutcome {
339    Success,
340    Error,
341    Cancelled,
342}
343
344fn emit_hook_event(event: HookExecutionEvent) {
345    unsafe {
346        if let Some(ref handler) = EVENT_HANDLER {
347            handler(event);
348        } else {
349            PENDING_EVENTS.push(event);
350            if PENDING_EVENTS.len() > MAX_PENDING_EVENTS {
351                PENDING_EVENTS.remove(0);
352            }
353        }
354    }
355}
356
357fn should_emit(hook_event: &str) -> bool {
358    if ALWAYS_EMITTED_HOOK_EVENTS.contains(&hook_event) {
359        return true;
360    }
361    unsafe { ALL_HOOK_EVENTS_ENABLED }
362}
363
364/// Register a handler for hook execution events
365pub fn register_hook_event_handler(handler: Option<Box<dyn Fn(HookExecutionEvent) + Send + Sync>>) {
366    unsafe {
367        let old_handler = EVENT_HANDLER.take();
368        EVENT_HANDLER = handler;
369
370        // If we have a new handler and pending events, deliver them
371        if let Some(ref handler) = EVENT_HANDLER {
372            let events: Vec<HookExecutionEvent> = PENDING_EVENTS.drain(..).collect();
373            for event in events {
374                handler(event);
375            }
376        } else {
377            // Restore old handler if any
378            if let Some(old) = old_handler {
379                EVENT_HANDLER = Some(old);
380            }
381        }
382    }
383}
384
385/// Emit hook started event
386pub fn emit_hook_started(hook_id: &str, hook_name: &str, hook_event: &str) {
387    if !should_emit(hook_event) {
388        return;
389    }
390    emit_hook_event(HookExecutionEvent::Started {
391        hook_id: hook_id.to_string(),
392        hook_name: hook_name.to_string(),
393        hook_event: hook_event.to_string(),
394    });
395}
396
397/// Emit hook progress event
398pub fn emit_hook_progress(params: HookProgressParams) {
399    if !should_emit(params.hook_event.as_str()) {
400        return;
401    }
402    emit_hook_event(HookExecutionEvent::Progress {
403        hook_id: params.hook_id,
404        hook_name: params.hook_name,
405        hook_event: params.hook_event.as_str().to_string(),
406        stdout: String::new(),
407        stderr: String::new(),
408        output: String::new(),
409    });
410}
411
412/// Start a progress interval that periodically emits hook progress events.
413/// Returns a JoinHandle that can be aborted to stop the interval.
414pub fn start_hook_progress_interval(params: HookProgressParams) -> tokio::task::JoinHandle<()> {
415    if !should_emit(params.hook_event.as_str()) {
416        return tokio::spawn(async {});
417    }
418
419    let interval_ms = params.interval_ms.unwrap_or(1000);
420    let hook_id = params.hook_id.clone();
421    let hook_name = params.hook_name.clone();
422    let hook_event = params.hook_event.clone();
423    let get_output = params.get_output;
424
425    // Spawn tokio task for progress polling
426    tokio::spawn(async move {
427        let mut last_emitted_output = String::new();
428        let mut interval = interval(Duration::from_millis(interval_ms));
429
430        loop {
431            interval.tick().await;
432            let output = get_output().await;
433            if output.output == last_emitted_output {
434                continue;
435            }
436            last_emitted_output = output.output.clone();
437
438            emit_hook_event(HookExecutionEvent::Progress {
439                hook_id: hook_id.clone(),
440                hook_name: hook_name.clone(),
441                hook_event: hook_event.as_str().to_string(),
442                stdout: output.stdout,
443                stderr: output.stderr,
444                output: output.output,
445            });
446        }
447    })
448}
449
450/// Emit hook response event
451pub fn emit_hook_response(data: HookResponseData) {
452    // Always log full hook output to debug log for verbose mode debugging
453    let output_to_log =
454        if !data.stdout.is_empty() || !data.stderr.is_empty() || !data.output.is_empty() {
455            if !data.stdout.is_empty() {
456                Some(&data.stdout)
457            } else if !data.stderr.is_empty() {
458                Some(&data.stderr)
459            } else {
460                Some(&data.output)
461            }
462        } else {
463            None
464        };
465
466    if let Some(output) = output_to_log {
467        log_for_debugging(&format!(
468            "Hook {} ({}) {:?}:\n{}",
469            data.hook_name, data.hook_event, data.outcome, output
470        ));
471    }
472
473    if !should_emit(&data.hook_event) {
474        return;
475    }
476
477    emit_hook_event(HookExecutionEvent::Response {
478        hook_id: data.hook_id,
479        hook_name: data.hook_name,
480        hook_event: data.hook_event,
481        output: data.output,
482        stdout: data.stdout,
483        stderr: data.stderr,
484        exit_code: data.exit_code,
485        outcome: data.outcome,
486    });
487}
488
489pub struct HookResponseData {
490    pub hook_id: String,
491    pub hook_name: String,
492    pub hook_event: String,
493    pub output: String,
494    pub stdout: String,
495    pub stderr: String,
496    pub exit_code: Option<i32>,
497    pub outcome: HookOutcome,
498}
499
500/// Enable emission of all hook event types (beyond SessionStart and Setup)
501pub fn set_all_hook_events_enabled(enabled: bool) {
502    unsafe {
503        ALL_HOOK_EVENTS_ENABLED = enabled;
504    }
505}
506
507/// Clear hook event state
508pub fn clear_hook_event_state() {
509    unsafe {
510        EVENT_HANDLER = None;
511        PENDING_EVENTS.clear();
512        ALL_HOOK_EVENTS_ENABLED = false;
513    }
514}
515
516/// Finalize a hook after completion
517async fn finalize_hook(_hook: &PendingAsyncHook, exit_code: i32, outcome: HookOutcome) {
518    // Note: progress_task_id cannot be called through a shared reference
519    // since it's a FnOnce. In practice, the caller would have already stopped it.
520
521    let stdout = if let Some(shell_cmd) = &_hook.shell_command {
522        shell_cmd.task_output.get_stdout().await
523    } else {
524        String::new()
525    };
526    let stderr = _hook
527        .shell_command
528        .as_ref()
529        .map_or(String::new(), |s| s.task_output.get_stderr());
530
531    if let Some(shell_cmd) = &_hook.shell_command {
532        shell_cmd.cleanup();
533    }
534
535    emit_hook_response(HookResponseData {
536        hook_id: _hook.hook_id.clone(),
537        hook_name: _hook.hook_name.clone(),
538        hook_event: _hook.hook_event.as_str().to_string(),
539        output: format!("{}{}", stdout, stderr),
540        stdout,
541        stderr,
542        exit_code: Some(exit_code),
543        outcome,
544    });
545}
546
547/// Response from check_for_async_hook_responses
548pub struct AsyncHookResponse {
549    pub process_id: String,
550    pub response: SyncHookJsonOutput,
551    pub hook_name: String,
552    pub hook_event: HookEvent,
553    pub tool_name: Option<String>,
554    pub plugin_id: Option<String>,
555    pub stdout: String,
556    pub stderr: String,
557    pub exit_code: Option<i32>,
558}
559
560/// Check for completed async hook responses
561pub async fn check_for_async_hook_responses() -> Vec<AsyncHookResponse> {
562    let mut responses: Vec<AsyncHookResponse> = Vec::new();
563
564    let pending_count;
565    let hooks_snapshot;
566    {
567        let registry = ASYNC_HOOK_REGISTRY.lock().unwrap();
568        pending_count = registry.pending_hooks.len();
569        hooks_snapshot = registry
570            .pending_hooks
571            .values()
572            .map(|h| h.process_id.clone())
573            .collect::<Vec<_>>();
574    }
575
576    log_for_debugging(&format!(
577        "Hooks: Found {} total hooks in registry",
578        pending_count
579    ));
580
581    let mut process_ids_to_remove: Vec<String> = Vec::new();
582    let mut session_start_completed = false;
583
584    for process_id in hooks_snapshot {
585        let hook_result = {
586            let mut registry = ASYNC_HOOK_REGISTRY.lock().unwrap();
587            let hook = match registry.pending_hooks.get_mut(&process_id) {
588                Some(h) => h,
589                None => continue,
590            };
591
592            if !hook.shell_command.is_some() {
593                log_for_debugging(&format!(
594                    "Hooks: Hook {} has no shell command, removing from registry",
595                    process_id
596                ));
597                hook.progress_task_id = None;
598
599                process_ids_to_remove.push(process_id.clone());
600                continue;
601            }
602
603            let shell_cmd = hook.shell_command.as_ref().unwrap();
604            if shell_cmd.status == ShellCommandStatus::Killed {
605                log_for_debugging(&format!(
606                    "Hooks: Hook {} is killed, removing from registry",
607                    process_id
608                ));
609                hook.progress_task_id = None;
610
611                shell_cmd.cleanup();
612                process_ids_to_remove.push(process_id.clone());
613                continue;
614            }
615
616            if shell_cmd.status != ShellCommandStatus::Completed {
617                continue;
618            }
619
620            if hook.response_attachment_sent {
621                log_for_debugging(&format!(
622                    "Hooks: Skipping hook {} - already delivered/sent",
623                    process_id
624                ));
625                hook.progress_task_id = None;
626
627                process_ids_to_remove.push(process_id.clone());
628                continue;
629            }
630
631            let stdout = shell_cmd.task_output.get_stdout().await;
632            if stdout.trim().is_empty() {
633                log_for_debugging(&format!("Hooks: Skipping hook {} - no stdout", process_id));
634                hook.progress_task_id = None;
635
636                process_ids_to_remove.push(process_id.clone());
637                continue;
638            }
639
640            let lines: Vec<&str> = stdout.lines().collect();
641            log_for_debugging(&format!(
642                "Hooks: Processing {} lines of stdout for {}",
643                lines.len(),
644                process_id
645            ));
646
647            let exit_code = 0; // Would come from shell command result
648
649            let mut response = SyncHookJsonOutput {
650                extra: HashMap::new(),
651            };
652            for line in &lines {
653                if line.trim().starts_with('{') {
654                    log_for_debugging(&format!(
655                        "Hooks: Found JSON line: {}...",
656                        &line.trim().chars().take(100).collect::<String>()
657                    ));
658                    if let Ok(parsed) = serde_json::from_str::<serde_json::Value>(line.trim()) {
659                        if !parsed
660                            .as_object()
661                            .map_or(false, |obj| obj.contains_key("async"))
662                        {
663                            log_for_debugging(&format!(
664                                "Hooks: Found sync response from {}: {}",
665                                process_id,
666                                serde_json::to_string(&parsed).unwrap_or_default()
667                            ));
668                            if let Some(obj) = parsed.as_object() {
669                                response.extra =
670                                    obj.iter().map(|(k, v)| (k.clone(), v.clone())).collect();
671                            }
672                            break;
673                        }
674                    }
675                }
676            }
677
678            hook.response_attachment_sent = true;
679            let is_session_start = hook.hook_event == HookEvent::SessionStart;
680
681            // Finalize (spawn to avoid blocking)
682            let hook_clone = PendingAsyncHook {
683                process_id: hook.process_id.clone(),
684                hook_id: hook.hook_id.clone(),
685                hook_name: hook.hook_name.clone(),
686                hook_event: hook.hook_event.clone(),
687                tool_name: hook.tool_name.clone(),
688                plugin_id: hook.plugin_id.clone(),
689                start_time: hook.start_time,
690                timeout_ms: hook.timeout_ms,
691                command: hook.command.clone(),
692                response_attachment_sent: true,
693                shell_command: None,
694                progress_task_id: None,
695            };
696            tokio::spawn(async move {
697                finalize_hook(&hook_clone, exit_code, HookOutcome::Success).await;
698            });
699
700            process_ids_to_remove.push(process_id.clone());
701            session_start_completed = session_start_completed || is_session_start;
702
703            Some((
704                process_id.clone(),
705                response,
706                hook.hook_name.clone(),
707                hook.hook_event.clone(),
708                hook.tool_name.clone(),
709                hook.plugin_id.clone(),
710                stdout,
711                shell_cmd.task_output.get_stderr(),
712                Some(exit_code),
713            ))
714        };
715
716        if let Some((
717            process_id,
718            response,
719            hook_name,
720            hook_event,
721            tool_name,
722            plugin_id,
723            stdout,
724            stderr,
725            exit_code,
726        )) = hook_result
727        {
728            responses.push(AsyncHookResponse {
729                process_id,
730                response,
731                hook_name,
732                hook_event,
733                tool_name,
734                plugin_id,
735                stdout,
736                stderr,
737                exit_code,
738            });
739        }
740    }
741
742    // Remove processed hooks
743    {
744        let mut registry = ASYNC_HOOK_REGISTRY.lock().unwrap();
745        for process_id in process_ids_to_remove {
746            registry.pending_hooks.remove(&process_id);
747        }
748    }
749
750    if session_start_completed {
751        log_for_debugging("Invalidating session env cache after SessionStart hook completed");
752        invalidate_session_env_cache();
753    }
754
755    log_for_debugging(&format!(
756        "Hooks: checkForNewResponses returning {} responses",
757        responses.len()
758    ));
759
760    responses
761}
762
763/// Remove delivered async hooks from the registry
764pub fn remove_delivered_async_hooks(process_ids: &[String]) {
765    let mut registry = ASYNC_HOOK_REGISTRY.lock().unwrap();
766    for process_id in process_ids {
767        if let Some(hook) = registry.pending_hooks.get(process_id) {
768            if hook.response_attachment_sent {
769                log_for_debugging(&format!("Hooks: Removing delivered hook {}", process_id));
770                // Note: can't call progress_task_id on borrowed ref
771            }
772        }
773        registry.pending_hooks.remove(process_id);
774    }
775}
776
777/// Finalize all pending async hooks (e.g., on shutdown)
778pub async fn finalize_pending_async_hooks() {
779    let hooks_snapshot;
780    {
781        let registry = ASYNC_HOOK_REGISTRY.lock().unwrap();
782        hooks_snapshot = registry
783            .pending_hooks
784            .values()
785            .map(|h| h.process_id.clone())
786            .collect::<Vec<_>>();
787    }
788
789    let mut futures = Vec::new();
790    for process_id in hooks_snapshot {
791        let mut registry = ASYNC_HOOK_REGISTRY.lock().unwrap();
792        if let Some(hook) = registry.pending_hooks.remove(&process_id) {
793            let exit_code;
794            let outcome;
795
796            if let Some(ref shell_cmd) = hook.shell_command {
797                if shell_cmd.status == ShellCommandStatus::Completed {
798                    exit_code = 0;
799                    outcome = HookOutcome::Success;
800                } else {
801                    if shell_cmd.status != ShellCommandStatus::Killed {
802                        // Can't mutate through ref
803                    }
804                    exit_code = 1;
805                    outcome = HookOutcome::Cancelled;
806                }
807            } else {
808                exit_code = 1;
809                outcome = HookOutcome::Cancelled;
810            }
811
812            futures.push(tokio::spawn(async move {
813                finalize_hook(&hook, exit_code, outcome).await;
814            }));
815        }
816    }
817
818    // Wait for all finalize tasks
819    for f in futures {
820        let _ = f.await;
821    }
822}
823
824/// Clear all async hooks (test utility)
825pub fn clear_all_async_hooks() {
826    let mut registry = ASYNC_HOOK_REGISTRY.lock().unwrap();
827    for hook in registry.pending_hooks.values() {
828        // Can't call progress_task_id through &ref
829    }
830    registry.pending_hooks.clear();
831}
832
833/// Log for debugging (simplified)
834fn log_for_debugging(msg: &str) {
835    log::debug!("{}", msg);
836}
837
838/// Invalidate session env cache (simplified)
839fn invalidate_session_env_cache() {
840    log::debug!("Invalidating session env cache");
841}
842
843/// JSON parse helper
844fn json_parse(s: &str) -> Result<serde_json::Value, serde_json::Error> {
845    serde_json::from_str(s)
846}
847
848/// JSON stringify helper
849fn json_stringify(value: &serde_json::Value) -> String {
850    serde_json::to_string(value).unwrap_or_default()
851}
ai_agent/utils/hooks/async_hook_registry.rs

ai_agent/utils/hooks/
async_hook_registry.rs
