scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-05 08:10:39 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 10

Simplify hook system after code review

Browse source 
- Replace JSON-based HookEvent::Display with simple match arms
- Use floor_char_boundary for Unicode-safe truncation in effective_name
- Cache compiled regexes in HookRunner instead of recompiling per check
- Simplify run_non_blocking (was run_parallel) to plain sequential loop
- Propagate hook_runner through parallel handler branch services
- Use unique temp file paths for sandbox hook context (avoid collisions)
- Add hook imports to engine.rs, reducing verbose crate:🪝: paths
- Extract duplicate RunFailed hook code into run_failed_hook helper

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-03-05 01:49:53 -05:00
parent 55d44f7488
commit fd70d41fca
6 changed files with 130 additions and 135 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

86
crates/arc-workflows/src/engine.rs
View file

@ -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<crate::hook::HookRunner>) {
pub fn set_hook_runner(&mut self, runner: Arc<HookRunner>) {
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(),
);

3
crates/arc-workflows/src/handler/parallel.rs
View file

@ -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

11
crates/arc-workflows/src/hook/config.rs
View file

@ -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}"),

14
crates/arc-workflows/src/hook/executor.rs
View file

@ -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)

131
crates/arc-workflows/src/hook/runner.rs
View file

@ -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<dyn HookExecutor>,
/// Pre-compiled regexes keyed by matcher pattern string.
compiled_matchers: HashMap<String, regex::Regex>,
}
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<dyn HookExecutor>) -> Self {
let compiled_matchers = Self::compile_matchers(&config);
Self {
config,
command_executor: executor,
compiled_matchers,
}
}
fn compile_matchers(config: &HookConfig) -> HashMap<String, regex::Regex> {
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
}

20
crates/arc-workflows/src/hook/types.rs
View file

@ -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",
})
}
}