diff --git a/crates/arc-workflows/src/engine.rs b/crates/arc-workflows/src/engine.rs index f2614d951..65575220e 100644 --- a/crates/arc-workflows/src/engine.rs +++ b/crates/arc-workflows/src/engine.rs @@ -22,6 +22,7 @@ use crate::error::{ArcError, FailureClass, FailureSignature, Result}; use crate::event::{EventEmitter, WorkflowRunEvent}; use crate::graph::{Edge, Graph, Node}; use crate::handler::{EngineServices, HandlerRegistry}; +use crate::hook::{HookContext, HookDecision, HookEvent, HookRunner}; use crate::interviewer::Interviewer; use crate::outcome::{Outcome, StageStatus}; use crate::preamble::build_preamble; @@ -788,7 +789,7 @@ impl WorkflowRunEngine { } /// Set the hook runner for lifecycle hooks. - pub fn set_hook_runner(&mut self, runner: Arc) { + pub fn set_hook_runner(&mut self, runner: Arc) { self.services.hook_runner = Some(runner); } @@ -796,17 +797,34 @@ impl WorkflowRunEngine { /// Returns `Proceed` if no hook runner is configured. async fn run_hooks( &self, - hook_context: &crate::hook::HookContext, + hook_context: &HookContext, work_dir: Option<&Path>, - ) -> crate::hook::HookDecision { + ) -> HookDecision { let Some(ref runner) = self.services.hook_runner else { - return crate::hook::HookDecision::Proceed; + return HookDecision::Proceed; }; runner .run(hook_context, self.services.sandbox.as_ref(), work_dir) .await } + /// Fire a non-blocking RunFailed hook. + async fn run_failed_hook( + &self, + run_id: &str, + workflow_name: &str, + error: &ArcError, + work_dir: Option<&Path>, + ) { + let mut hook_ctx = HookContext::new( + HookEvent::RunFailed, + run_id.to_string(), + workflow_name.to_string(), + ); + hook_ctx.failure_reason = Some(error.to_string()); + let _ = self.run_hooks(&hook_ctx, work_dir).await; + } + /// Mirror graph-level attributes into the context. fn mirror_graph_attributes(graph: &Graph, context: &Context) { if !graph.goal().is_empty() { @@ -1111,13 +1129,13 @@ impl WorkflowRunEngine { // RunStart hook (blocking — can prevent run) { - let hook_ctx = crate::hook::HookContext::new( - crate::hook::HookEvent::RunStart, + let hook_ctx = HookContext::new( + HookEvent::RunStart, run_id.clone(), graph.name.clone(), ); let decision = self.run_hooks(&hook_ctx, hook_work_dir.as_deref()).await; - if let crate::hook::HookDecision::Block { reason } = decision { + if let HookDecision::Block { reason } = decision { let msg = reason.unwrap_or_else(|| "blocked by RunStart hook".into()); return Err(ArcError::engine(msg)); } @@ -1346,16 +1364,7 @@ impl WorkflowRunEngine { git_commit_sha: last_git_sha.clone(), }); - // RunFailed hook (non-blocking) - { - let mut hook_ctx = crate::hook::HookContext::new( - crate::hook::HookEvent::RunFailed, - run_id.clone(), - graph.name.clone(), - ); - hook_ctx.failure_reason = Some(error.to_string()); - let _ = self.run_hooks(&hook_ctx, hook_work_dir.as_deref()).await; - } + self.run_failed_hook(&run_id, &graph.name, &error, hook_work_dir.as_deref()).await; return Ok((error.to_fail_outcome(), context)); } @@ -1412,8 +1421,8 @@ impl WorkflowRunEngine { // StageStart hook (blocking — can skip node) { - let mut hook_ctx = crate::hook::HookContext::new( - crate::hook::HookEvent::StageStart, + let mut hook_ctx = HookContext::new( + HookEvent::StageStart, run_id.clone(), graph.name.clone(), ); @@ -1429,7 +1438,7 @@ impl WorkflowRunEngine { .run_hooks(&hook_ctx, hook_work_dir.as_deref()) .await; match decision { - crate::hook::HookDecision::Skip { reason } => { + HookDecision::Skip { reason } => { let mut outcome = Outcome::skipped(); outcome.notes = Some( reason.unwrap_or_else(|| "skipped by StageStart hook".into()), @@ -1448,7 +1457,7 @@ impl WorkflowRunEngine { } continue; } - crate::hook::HookDecision::Block { reason } => { + HookDecision::Block { reason } => { let msg = reason.unwrap_or_else(|| "blocked by StageStart hook".into()); return Err(ArcError::engine(msg)); } @@ -1548,8 +1557,8 @@ impl WorkflowRunEngine { // StageFailed hook (non-blocking) { - let mut hook_ctx = crate::hook::HookContext::new( - crate::hook::HookEvent::StageFailed, + let mut hook_ctx = HookContext::new( + HookEvent::StageFailed, run_id.clone(), graph.name.clone(), ); @@ -1584,8 +1593,8 @@ impl WorkflowRunEngine { // StageComplete hook (non-blocking) { - let mut hook_ctx = crate::hook::HookContext::new( - crate::hook::HookEvent::StageComplete, + let mut hook_ctx = HookContext::new( + HookEvent::StageComplete, run_id.clone(), graph.name.clone(), ); @@ -1668,8 +1677,8 @@ impl WorkflowRunEngine { .cloned() .or_else(|| next_edge.as_ref().map(|e| e.to.clone())); if let Some(ref to) = edge_to { - let mut hook_ctx = crate::hook::HookContext::new( - crate::hook::HookEvent::EdgeSelected, + let mut hook_ctx = HookContext::new( + HookEvent::EdgeSelected, run_id.clone(), graph.name.clone(), ); @@ -1680,11 +1689,11 @@ impl WorkflowRunEngine { .run_hooks(&hook_ctx, hook_work_dir.as_deref()) .await; match decision { - crate::hook::HookDecision::Override { edge_to: new_target } => { + HookDecision::Override { edge_to: new_target } => { // Redirect routing to the hook-specified target (None, Some(new_target)) } - crate::hook::HookDecision::Block { reason } => { + HookDecision::Block { reason } => { let msg = reason.unwrap_or_else(|| "blocked by EdgeSelected hook".into()); return Err(ArcError::engine(msg)); } @@ -1723,8 +1732,8 @@ impl WorkflowRunEngine { // CheckpointSaved hook (non-blocking) { - let mut hook_ctx = crate::hook::HookContext::new( - crate::hook::HookEvent::CheckpointSaved, + let mut hook_ctx = HookContext::new( + HookEvent::CheckpointSaved, run_id.clone(), graph.name.clone(), ); @@ -1874,16 +1883,7 @@ impl WorkflowRunEngine { git_commit_sha: last_git_sha.clone(), }); - // RunFailed hook (non-blocking) - { - let mut hook_ctx = crate::hook::HookContext::new( - crate::hook::HookEvent::RunFailed, - run_id.clone(), - graph.name.clone(), - ); - hook_ctx.failure_reason = Some(error.to_string()); - let _ = self.run_hooks(&hook_ctx, hook_work_dir.as_deref()).await; - } + self.run_failed_hook(&run_id, &graph.name, &error, hook_work_dir.as_deref()).await; return Err(error); } @@ -1965,8 +1965,8 @@ impl WorkflowRunEngine { // RunComplete hook (non-blocking) { - let hook_ctx = crate::hook::HookContext::new( - crate::hook::HookEvent::RunComplete, + let hook_ctx = HookContext::new( + HookEvent::RunComplete, run_id.clone(), graph.name.clone(), ); diff --git a/crates/arc-workflows/src/handler/parallel.rs b/crates/arc-workflows/src/handler/parallel.rs index ffd2d6c6a..e4ea3b21c 100644 --- a/crates/arc-workflows/src/handler/parallel.rs +++ b/crates/arc-workflows/src/handler/parallel.rs @@ -392,6 +392,7 @@ impl Handler for ParallelHandler { for setup in branch_setups { let registry = Arc::clone(&services.registry); let emitter = Arc::clone(&services.emitter); + let hook_runner = services.hook_runner.clone(); let graph = graph.clone(); let logs_root = logs_root.to_path_buf(); let sem = Arc::clone(&semaphore); @@ -434,7 +435,7 @@ impl Handler for ParallelHandler { emitter: Arc::clone(&emitter), sandbox: Arc::clone(&setup.sandbox), git_state: std::sync::RwLock::new(None), - hook_runner: None, + hook_runner: hook_runner.clone(), }; let handler = registry.resolve(target_node); let outcome = handler diff --git a/crates/arc-workflows/src/hook/config.rs b/crates/arc-workflows/src/hook/config.rs index 8bf108558..bf0c3acab 100644 --- a/crates/arc-workflows/src/hook/config.rs +++ b/crates/arc-workflows/src/hook/config.rs @@ -68,17 +68,10 @@ impl HookDefinition { if let Some(ref n) = self.name { return n.clone(); } - let event_str = serde_json::to_value(&self.event) - .ok() - .and_then(|v| v.as_str().map(String::from)) - .unwrap_or_else(|| format!("{:?}", self.event)); + let event_str = self.event.to_string(); match self.resolved_hook_type() { Some(HookType::Command { ref command }) => { - let short = if command.len() > 20 { - &command[..20] - } else { - command - }; + let short = &command[..arc_agent::floor_char_boundary(command, 20)]; format!("{event_str}:{short}") } Some(HookType::Http { ref url, .. }) => format!("{event_str}:{url}"), diff --git a/crates/arc-workflows/src/hook/executor.rs b/crates/arc-workflows/src/hook/executor.rs index c386c5812..2c9827c7d 100644 --- a/crates/arc-workflows/src/hook/executor.rs +++ b/crates/arc-workflows/src/hook/executor.rs @@ -84,10 +84,16 @@ impl HookExecutor for CommandHookExecutor { } let decision = if definition.runs_in_sandbox() { - // Write context to temp file, pass path as env var - let ctx_path = "/tmp/arc-hook-context.json"; - if sandbox.write_file(ctx_path, &context_json).await.is_ok() { - env_vars.insert("ARC_HOOK_CONTEXT".to_string(), ctx_path.to_string()); + // Write context to a unique temp file, pass path as env var + let ctx_path = format!( + "/tmp/arc-hook-context-{}.json", + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_nanos() + ); + if sandbox.write_file(&ctx_path, &context_json).await.is_ok() { + env_vars.insert("ARC_HOOK_CONTEXT".to_string(), ctx_path.clone()); } match sandbox .exec_command(&command, timeout_ms, None, Some(&env_vars), None) diff --git a/crates/arc-workflows/src/hook/runner.rs b/crates/arc-workflows/src/hook/runner.rs index 30b1bc4d5..b3e27fe3b 100644 --- a/crates/arc-workflows/src/hook/runner.rs +++ b/crates/arc-workflows/src/hook/runner.rs @@ -1,3 +1,4 @@ +use std::collections::HashMap; use std::path::Path; use std::sync::Arc; @@ -11,26 +12,46 @@ use super::types::{HookContext, HookDecision}; pub struct HookRunner { config: HookConfig, command_executor: Arc, + /// Pre-compiled regexes keyed by matcher pattern string. + compiled_matchers: HashMap, } impl HookRunner { #[must_use] pub fn new(config: HookConfig) -> Self { + let compiled_matchers = Self::compile_matchers(&config); Self { config, command_executor: Arc::new(CommandHookExecutor), + compiled_matchers, } } /// Create a HookRunner with a custom executor (for testing). #[cfg(test)] pub fn with_executor(config: HookConfig, executor: Arc) -> Self { + let compiled_matchers = Self::compile_matchers(&config); Self { config, command_executor: executor, + compiled_matchers, } } + fn compile_matchers(config: &HookConfig) -> HashMap { + let mut map = HashMap::new(); + for hook in &config.hooks { + if let Some(ref pattern) = hook.matcher { + if !map.contains_key(pattern) { + if let Ok(re) = regex::Regex::new(pattern) { + map.insert(pattern.clone(), re); + } + } + } + } + map + } + /// Run all matching hooks for the given event and return the merged decision. pub async fn run( &self, @@ -57,8 +78,8 @@ impl HookRunner { self.run_sequential(&matching, context, sandbox, work_dir) .await } else { - // Parallel execution for non-blocking hooks - self.run_parallel(&matching, context, sandbox, work_dir) + // Non-blocking: run all, ignore decisions + self.run_non_blocking(&matching, context, sandbox, work_dir) .await }; @@ -86,36 +107,18 @@ impl HookRunner { let Some(ref pattern) = hook.matcher else { return true; }; - let Ok(re) = regex::Regex::new(pattern) else { - tracing::warn!( - hook = %hook.effective_name(), - pattern, - "Invalid hook matcher regex" - ); + let Some(re) = self.compiled_matchers.get(pattern) else { + // Pattern failed to compile during construction — already warned return false; }; - // Match against node_id, handler_type, edge_to, edge_from - if let Some(ref node_id) = context.node_id { - if re.is_match(node_id) { - return true; - } - } - if let Some(ref handler_type) = context.handler_type { - if re.is_match(handler_type) { - return true; - } - } - if let Some(ref edge_to) = context.edge_to { - if re.is_match(edge_to) { - return true; - } - } - if let Some(ref edge_from) = context.edge_from { - if re.is_match(edge_from) { - return true; - } - } - false + [ + context.node_id.as_deref(), + context.handler_type.as_deref(), + context.edge_to.as_deref(), + context.edge_from.as_deref(), + ] + .iter() + .any(|field| field.is_some_and(|v| re.is_match(v))) } async fn run_sequential( @@ -167,55 +170,37 @@ impl HookRunner { merged } - async fn run_parallel( + async fn run_non_blocking( &self, hooks: &[&HookDefinition], context: &HookContext, sandbox: &dyn Sandbox, work_dir: Option<&Path>, ) -> HookDecision { - let futures: Vec<_> = hooks - .iter() - .map(|hook| { - let executor = Arc::clone(&self.command_executor); - let hook_clone = (*hook).clone(); - let ctx_clone = context.clone(); - let wd = work_dir.map(|p| p.to_path_buf()); - async move { - tracing::debug!( - hook = %hook_clone.effective_name(), - event = %ctx_clone.event, - "Executing hook" - ); - let result = executor - .execute(&hook_clone, &ctx_clone, sandbox, wd.as_deref()) - .await; - tracing::debug!( - hook = %hook_clone.effective_name(), - duration_ms = result.duration_ms, - decision = ?result.decision, - "Hook complete" - ); - if !result.decision.is_proceed() { - tracing::warn!( - hook = %hook_clone.effective_name(), - event = %ctx_clone.event, - decision = ?result.decision, - "Non-blocking hook failed, continuing" - ); - } - result - } - }) - .collect(); - - // We can't easily use join_all with a reference to sandbox since Sandbox - // is not necessarily Send-safe for concurrent borrows. Run sequentially - // but log as non-blocking (don't short-circuit). - for future in futures { - let result = future.await; - // Non-blocking hooks: log but don't merge decisions - let _ = result; + for hook in hooks { + tracing::debug!( + hook = %hook.effective_name(), + event = %context.event, + "Executing hook" + ); + let result = self + .command_executor + .execute(hook, context, sandbox, work_dir) + .await; + tracing::debug!( + hook = %hook.effective_name(), + duration_ms = result.duration_ms, + decision = ?result.decision, + "Hook complete" + ); + if !result.decision.is_proceed() { + tracing::warn!( + hook = %hook.effective_name(), + event = %context.event, + decision = ?result.decision, + "Non-blocking hook failed, continuing" + ); + } } HookDecision::Proceed } diff --git a/crates/arc-workflows/src/hook/types.rs b/crates/arc-workflows/src/hook/types.rs index b7a6e879b..0239193d9 100644 --- a/crates/arc-workflows/src/hook/types.rs +++ b/crates/arc-workflows/src/hook/types.rs @@ -29,11 +29,21 @@ impl HookEvent { impl std::fmt::Display for HookEvent { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - let s = serde_json::to_value(self) - .ok() - .and_then(|v| v.as_str().map(String::from)) - .unwrap_or_else(|| format!("{self:?}")); - f.write_str(&s) + f.write_str(match self { + Self::RunStart => "run_start", + Self::RunComplete => "run_complete", + Self::RunFailed => "run_failed", + Self::StageStart => "stage_start", + Self::StageComplete => "stage_complete", + Self::StageFailed => "stage_failed", + Self::StageRetrying => "stage_retrying", + Self::EdgeSelected => "edge_selected", + Self::ParallelStart => "parallel_start", + Self::ParallelComplete => "parallel_complete", + Self::SandboxReady => "sandbox_ready", + Self::SandboxCleanup => "sandbox_cleanup", + Self::CheckpointSaved => "checkpoint_saved", + }) } }