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

Add lifecycle hook system replacing legacy tool_hooks

Browse source 
Introduce a configurable hook system that triggers user-defined actions
at workflow lifecycle points (RunStart, StageStart, StageComplete,
StageFailed, EdgeSelected, CheckpointSaved, etc). Hooks can block
execution, skip nodes, or override edge routing via JSON decisions.

- New `hook/` module: types, config, executor (command), runner
- Engine instrumented at 8 lifecycle points with HookRunner calls
- TOML config: `[[hooks]]` in server.toml and run config files
- Config cascade: server hooks + run hooks merge, name collisions
  resolved by run config winning
- Remove legacy tool_hooks.pre/post from codergen handler (breaking)
- 30 e2e integration tests covering all hook events and behaviors

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-03-05 01:42:19 -05:00
parent b57e30efad
commit 55d44f7488
23 changed files with 2656 additions and 478 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
crates/arc-api/src/server.rs
View file

@ -104,6 +104,7 @@ pub struct AppState {
pub db: sqlx::SqlitePool,
max_concurrent_runs: usize,
scheduler_notify: tokio::sync::Notify,
pub hook_config: arc_workflows::hook::HookConfig,
}
/// Build the axum Router with all run endpoints.
@ -329,6 +330,7 @@ pub fn create_app_state_with_options(
db,
max_concurrent_runs,
scheduler_notify: tokio::sync::Notify::new(),
hook_config: arc_workflows::hook::HookConfig::default(),
})
}
@ -468,13 +470,19 @@ async fn execute_run(state: Arc<AppState>, run_id: String) {
let registry = (state.registry_factory)(Arc::clone(&interviewer) as Arc<dyn Interviewer>);
let cwd = std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from("."));
let sandbox: Arc<dyn arc_agent::Sandbox> = Arc::new(LocalSandbox::new(cwd));
let engine = WorkflowRunEngine::with_interviewer(
let mut engine = WorkflowRunEngine::with_interviewer(
registry,
Arc::new(emitter),
Arc::clone(&interviewer) as Arc<dyn Interviewer>,
sandbox,
);
// Wire up hook runner from server config
if !state.hook_config.hooks.is_empty() {
let runner = arc_workflows::hook::HookRunner::new(state.hook_config.clone());
engine.set_hook_runner(std::sync::Arc::new(runner));
}
// Transition to Running, populate interviewer + context
{
let mut runs = state.runs.lock().expect("runs lock poisoned");

42
crates/arc-api/src/server_config.rs
View file

@ -1,6 +1,7 @@
use std::path::{Path, PathBuf};
use arc_workflows::cli::run_config::RunDefaults;
use arc_workflows::hook::HookConfig;
use serde::Deserialize;
use tracing::debug;
@ -105,6 +106,8 @@ pub struct ServerConfig {
pub git: GitConfig,
#[serde(flatten)]
pub run_defaults: RunDefaults,
#[serde(flatten)]
pub hook_config: HookConfig,
}
/// Load server config from an explicit path or `~/.arc/server.toml`, returning defaults if the
@ -381,4 +384,43 @@ authentication_strategies = ["jwt"]
let result = load_server_config(Some(&path));
assert!(result.is_err());
}
#[test]
fn parse_config_with_hooks() {
let toml = r#"
[[hooks]]
event = "run_start"
command = "echo 'run starting'"
[[hooks]]
event = "stage_complete"
command = "echo 'stage done'"
matcher = "codergen"
"#;
let config: ServerConfig = toml::from_str(toml).unwrap();
assert_eq!(config.hook_config.hooks.len(), 2);
assert_eq!(
config.hook_config.hooks[0].event,
arc_workflows::hook::HookEvent::RunStart
);
assert_eq!(
config.hook_config.hooks[0].command.as_deref(),
Some("echo 'run starting'")
);
assert_eq!(
config.hook_config.hooks[1].event,
arc_workflows::hook::HookEvent::StageComplete
);
assert_eq!(
config.hook_config.hooks[1].matcher.as_deref(),
Some("codergen")
);
}
#[test]
fn parse_config_without_hooks_defaults_empty() {
let toml = "";
let config: ServerConfig = toml::from_str(toml).unwrap();
assert!(config.hook_config.hooks.is_empty());
}
}

18
crates/arc-workflows/src/cli/run.rs
View file

@ -636,13 +636,24 @@ pub async fn run_command(
Some(Box::new(BackendRouter::new(Box::new(api), cli)))
}
});
let engine = WorkflowRunEngine::with_interviewer(
let mut engine = WorkflowRunEngine::with_interviewer(
registry,
Arc::clone(&emitter),
interviewer,
Arc::clone(&sandbox),
);
// Wire up hook runner from run config
if let Some(ref cfg) = run_cfg {
if !cfg.hooks.is_empty() {
let hook_config = crate::hook::HookConfig {
hooks: cfg.hooks.clone(),
};
let runner = crate::hook::HookRunner::new(hook_config);
engine.set_hook_runner(Arc::new(runner));
}
}
// 7. Execute
let run_id = worktree_run_id
.or(daytona_run_id)
@ -1563,6 +1574,7 @@ mod tests {
setup: None,
sandbox: None,
vars: None,
hooks: Vec::new(),
};
let (model, provider) = resolve_model_provider(
Some("gpt-5.2"),
@ -1602,6 +1614,7 @@ mod tests {
setup: None,
sandbox: None,
vars: None,
hooks: Vec::new(),
};
let (model, provider) = resolve_model_provider(None, None, Some(&cfg), &defaults, &graph);
assert_eq!(model, "toml-model");
@ -1676,6 +1689,7 @@ mod tests {
setup: None,
sandbox: None,
vars: None,
hooks: Vec::new(),
};
let (model, provider) = resolve_model_provider(None, None, Some(&cfg), &defaults, &graph);
assert_eq!(model, "toml-model");
@ -1697,6 +1711,7 @@ mod tests {
daytona: None,
}),
vars: None,
hooks: Vec::new(),
};
let defaults = RunDefaults::default();
assert!(resolve_preserve_sandbox(true, Some(&cfg), &defaults));
@ -1717,6 +1732,7 @@ mod tests {
daytona: None,
}),
vars: None,
hooks: Vec::new(),
};
let defaults = RunDefaults {
sandbox: Some(run_config::SandboxConfig {

42
crates/arc-workflows/src/cli/run_config.rs
View file

@ -19,6 +19,8 @@ pub struct WorkflowRunConfig {
pub setup: Option<SetupConfig>,
pub sandbox: Option<SandboxConfig>,
pub vars: Option<HashMap<String, String>>,
#[serde(default)]
pub hooks: Vec<crate::hook::HookDefinition>,
}
#[derive(Clone, Debug, Deserialize, PartialEq)]
@ -1078,4 +1080,44 @@ model = "opus"
"unexpected error: {err}"
);
}
#[test]
fn parse_toml_with_hooks() {
let toml = r#"
version = 1
goal = "Test hooks"
graph = "test.dot"
[[hooks]]
event = "stage_start"
command = "./scripts/pre-check.sh"
blocking = true
sandbox = false
[[hooks]]
event = "run_complete"
command = "echo done"
"#;
let cfg: WorkflowRunConfig = toml::from_str(toml).unwrap();
assert_eq!(cfg.hooks.len(), 2);
assert_eq!(cfg.hooks[0].event, crate::hook::HookEvent::StageStart);
assert_eq!(
cfg.hooks[0].command.as_deref(),
Some("./scripts/pre-check.sh")
);
assert_eq!(cfg.hooks[0].blocking, Some(true));
assert_eq!(cfg.hooks[0].sandbox, Some(false));
assert_eq!(cfg.hooks[1].event, crate::hook::HookEvent::RunComplete);
}
#[test]
fn parse_toml_without_hooks_defaults_empty() {
let toml = r#"
version = 1
goal = "No hooks"
graph = "test.dot"
"#;
let cfg: WorkflowRunConfig = toml::from_str(toml).unwrap();
assert!(cfg.hooks.is_empty());
}
}

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

@ -746,6 +746,7 @@ impl WorkflowRunEngine {
emitter,
sandbox,
git_state: std::sync::RwLock::new(None),
hook_runner: None,
},
interviewer: None,
}
@ -760,6 +761,7 @@ impl WorkflowRunEngine {
emitter: Arc::clone(&services.emitter),
sandbox: Arc::clone(&services.sandbox),
git_state: std::sync::RwLock::new(None),
hook_runner: services.hook_runner.clone(),
},
interviewer: None,
}
@ -779,11 +781,32 @@ impl WorkflowRunEngine {
emitter,
sandbox,
git_state: std::sync::RwLock::new(None),
hook_runner: None,
},
interviewer: Some(interviewer),
}
}
/// Set the hook runner for lifecycle hooks.
pub fn set_hook_runner(&mut self, runner: Arc<crate::hook::HookRunner>) {
self.services.hook_runner = Some(runner);
}
/// Run lifecycle hooks and return the merged decision.
/// Returns `Proceed` if no hook runner is configured.
async fn run_hooks(
&self,
hook_context: &crate::hook::HookContext,
work_dir: Option<&Path>,
) -> crate::hook::HookDecision {
let Some(ref runner) = self.services.hook_runner else {
return crate::hook::HookDecision::Proceed;
};
runner
.run(hook_context, self.services.sandbox.as_ref(), work_dir)
.await
}
/// Mirror graph-level attributes into the context.
fn mirror_graph_attributes(graph: &Graph, context: &Context) {
if !graph.goal().is_empty() {
@ -1079,6 +1102,27 @@ impl WorkflowRunEngine {
_ => None,
},
});
// Resolve work_dir from config for hooks
let hook_work_dir: Option<PathBuf> = match config.git_checkpoint {
Some(GitCheckpointMode::Host(ref p)) => Some(p.clone()),
_ => None,
};
// RunStart hook (blocking — can prevent run)
{
let hook_ctx = crate::hook::HookContext::new(
crate::hook::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 {
let msg = reason.unwrap_or_else(|| "blocked by RunStart hook".into());
return Err(ArcError::engine(msg));
}
}
// Write manifest.json (spec 5.6)
let manifest = write_manifest(&config.logs_root, graph, config);
@ -1301,6 +1345,18 @@ impl WorkflowRunEngine {
duration_ms,
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;
}
return Ok((error.to_fail_outcome(), context));
}
}
@ -1353,6 +1409,53 @@ impl WorkflowRunEngine {
attempt: 1,
max_attempts: usize::try_from(retry_policy.max_attempts).unwrap_or(usize::MAX),
});
// StageStart hook (blocking — can skip node)
{
let mut hook_ctx = crate::hook::HookContext::new(
crate::hook::HookEvent::StageStart,
run_id.clone(),
graph.name.clone(),
);
hook_ctx.cwd = hook_work_dir.as_ref().map(|p| p.display().to_string());
hook_ctx.node_id = Some(node.id.clone());
hook_ctx.node_label = Some(node.label().to_string());
hook_ctx.handler_type = node.handler_type().map(String::from);
hook_ctx.attempt = Some(1);
hook_ctx.max_attempts = Some(
usize::try_from(retry_policy.max_attempts).unwrap_or(usize::MAX),
);
let decision = self
.run_hooks(&hook_ctx, hook_work_dir.as_deref())
.await;
match decision {
crate::hook::HookDecision::Skip { reason } => {
let mut outcome = Outcome::skipped();
outcome.notes = Some(
reason.unwrap_or_else(|| "skipped by StageStart hook".into()),
);
completed_nodes.push(node.id.clone());
node_outcomes.insert(node.id.clone(), outcome);
previous_node_id = Some(node.id.clone());
stage_index += 1;
// Select next edge and continue
let edge = select_edge(&node.id, &Outcome::skipped(), &context, graph);
if let Some(e) = edge {
current_node_id = e.to.clone();
incoming_edge = Some(e);
} else {
break;
}
continue;
}
crate::hook::HookDecision::Block { reason } => {
let msg = reason.unwrap_or_else(|| "blocked by StageStart hook".into());
return Err(ArcError::engine(msg));
}
_ => {}
}
}
let stage_start = Instant::now();
let (mut outcome, attempts_used) = if let Some((ref token, _)) = stall_token {
@ -1442,6 +1545,23 @@ impl WorkflowRunEngine {
}),
will_retry: false,
});
// StageFailed hook (non-blocking)
{
let mut hook_ctx = crate::hook::HookContext::new(
crate::hook::HookEvent::StageFailed,
run_id.clone(),
graph.name.clone(),
);
hook_ctx.node_id = Some(node.id.clone());
hook_ctx.node_label = Some(node.label().to_string());
hook_ctx.handler_type = node.handler_type().map(String::from);
hook_ctx.status = Some("fail".into());
hook_ctx.failure_reason = outcome.failure_reason().map(String::from);
let _ = self
.run_hooks(&hook_ctx, hook_work_dir.as_deref())
.await;
}
} else {
self.services
.emitter
@ -1461,6 +1581,22 @@ impl WorkflowRunEngine {
max_attempts: usize::try_from(retry_policy.max_attempts)
.unwrap_or(usize::MAX),
});
// StageComplete hook (non-blocking)
{
let mut hook_ctx = crate::hook::HookContext::new(
crate::hook::HookEvent::StageComplete,
run_id.clone(),
graph.name.clone(),
);
hook_ctx.node_id = Some(node.id.clone());
hook_ctx.node_label = Some(node.label().to_string());
hook_ctx.handler_type = node.handler_type().map(String::from);
hook_ctx.status = Some(outcome.status.to_string());
let _ = self
.run_hooks(&hook_ctx, hook_work_dir.as_deref())
.await;
}
}
// Write per-node status.json (spec 5.6)
@ -1524,6 +1660,41 @@ impl WorkflowRunEngine {
}
(edge, None)
};
// EdgeSelected hook (blocking — can override routing)
let (next_edge, jump_target) = {
let edge_to = jump_target
.as_ref()
.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,
run_id.clone(),
graph.name.clone(),
);
hook_ctx.edge_from = Some(node.id.clone());
hook_ctx.edge_to = Some(to.clone());
hook_ctx.edge_label = next_edge.as_ref().and_then(|e| e.label().map(String::from));
let decision = self
.run_hooks(&hook_ctx, hook_work_dir.as_deref())
.await;
match decision {
crate::hook::HookDecision::Override { edge_to: new_target } => {
// Redirect routing to the hook-specified target
(None, Some(new_target))
}
crate::hook::HookDecision::Block { reason } => {
let msg = reason.unwrap_or_else(|| "blocked by EdgeSelected hook".into());
return Err(ArcError::engine(msg));
}
_ => (next_edge, jump_target),
}
} else {
(next_edge, jump_target)
}
};
let next_node_id_for_checkpoint = jump_target
.as_ref()
.cloned()
@ -1549,6 +1720,17 @@ impl WorkflowRunEngine {
.emit(&WorkflowRunEvent::CheckpointSaved {
node_id: node.id.clone(),
});
// CheckpointSaved hook (non-blocking)
{
let mut hook_ctx = crate::hook::HookContext::new(
crate::hook::HookEvent::CheckpointSaved,
run_id.clone(),
graph.name.clone(),
);
hook_ctx.node_id = Some(node.id.clone());
let _ = self.run_hooks(&hook_ctx, hook_work_dir.as_deref()).await;
}
}
// Step 6b: Write shadow branch first, then run branch commit with trailer
@ -1691,6 +1873,18 @@ impl WorkflowRunEngine {
duration_ms,
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;
}
return Err(error);
}
break;
@ -1769,6 +1963,16 @@ impl WorkflowRunEngine {
final_git_commit_sha: last_git_sha.clone(),
});
// RunComplete hook (non-blocking)
{
let hook_ctx = crate::hook::HookContext::new(
crate::hook::HookEvent::RunComplete,
run_id.clone(),
graph.name.clone(),
);
let _ = self.run_hooks(&hook_ctx, hook_work_dir.as_deref()).await;
}
// Write final.patch: comprehensive diff from base_sha to HEAD
if let (Some(ref mode), Some(ref base)) = (&config.git_checkpoint, &config.base_sha) {
let patch = match mode {

143
crates/arc-workflows/src/handler/codergen.rs
View file

@ -172,28 +172,6 @@ fn truncate(s: &str, max_chars: usize) -> &str {
}
}
/// Resolve a tool hook command from node attributes, falling back to graph attributes.
fn resolve_hook(node: &Node, graph: &Graph, key: &str) -> Option<String> {
node.attrs
.get(key)
.and_then(|v| v.as_str())
.or_else(|| graph.attrs.get(key).and_then(|v| v.as_str()))
.map(String::from)
}
/// Execute a tool hook shell command. Returns true if the command succeeded (exit 0).
fn run_hook(command: &str, node_id: &str, work_dir: Option<&Path>) -> bool {
let mut cmd = std::process::Command::new("sh");
cmd.arg("-c").arg(command).env("ARC_NODE_ID", node_id);
if let Some(wd) = work_dir {
cmd.current_dir(wd);
}
match cmd.output() {
Ok(output) => output.status.success(),
Err(_) => false,
}
}
#[async_trait]
impl Handler for CodergenHandler {
async fn execute(
@ -223,22 +201,7 @@ impl Handler for CodergenHandler {
tokio::fs::create_dir_all(&stage_dir).await?;
tokio::fs::write(stage_dir.join("prompt.md"), &prompt).await?;
// Resolve work_dir from context for hooks
let work_dir_str = context
.get("internal.work_dir")
.and_then(|v| v.as_str().map(String::from));
let work_dir = work_dir_str.as_deref().map(Path::new);
// 3. Execute pre-hook (spec 9.7)
if let Some(pre_hook) = resolve_hook(node, graph, "tool_hooks.pre") {
if !run_hook(&pre_hook, &node.id, work_dir) {
let mut outcome = Outcome::skipped();
outcome.notes = Some("pre-hook returned non-zero, tool call skipped".to_string());
return Ok(outcome);
}
}
// 4. Call LLM backend
// 3. Call LLM backend
let mode = node.codergen_mode()?;
let thread_id = context
.get("internal.thread_id")
@ -288,14 +251,7 @@ impl Handler for CodergenHandler {
)
};
// 5. Execute post-hook (spec 9.7)
if let Some(post_hook) = resolve_hook(node, graph, "tool_hooks.post") {
if !run_hook(&post_hook, &node.id, work_dir) {
context.append_log(format!("post-hook failed for node {}, continuing", node.id));
}
}
// 6. Write response to logs
// 4. Write response to logs
tokio::fs::write(stage_dir.join("response.md"), &response_text).await?;
// 7. Build and write status
@ -343,6 +299,7 @@ mod tests {
std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from(".")),
)),
git_state: std::sync::RwLock::new(None),
hook_runner: None,
}
}
@ -500,100 +457,6 @@ mod tests {
assert_eq!(truncate(&long, 200).len(), 200);
}
#[tokio::test]
async fn codergen_handler_pre_hook_failure_skips_backend() {
let handler = CodergenHandler::new(None);
let mut node = Node::new("step");
node.attrs.insert(
"tool_hooks.pre".to_string(),
AttrValue::String("exit 1".to_string()),
);
let context = Context::new();
let graph = Graph::new("test");
let tmp = TempDir::new().unwrap();
let outcome = handler
.execute(&node, &context, &graph, tmp.path(), &make_services())
.await
.unwrap();
assert_eq!(outcome.status, crate::outcome::StageStatus::Skipped);
assert!(outcome.notes.as_deref().unwrap().contains("pre-hook"));
}
#[tokio::test]
async fn codergen_handler_pre_hook_success_continues() {
let handler = CodergenHandler::new(None);
let mut node = Node::new("step");
node.attrs.insert(
"tool_hooks.pre".to_string(),
AttrValue::String("exit 0".to_string()),
);
let context = Context::new();
let graph = Graph::new("test");
let tmp = TempDir::new().unwrap();
let outcome = handler
.execute(&node, &context, &graph, tmp.path(), &make_services())
.await
.unwrap();
assert_eq!(outcome.status, crate::outcome::StageStatus::Success);
}
#[tokio::test]
async fn codergen_handler_post_hook_failure_logs_warning() {
let handler = CodergenHandler::new(None);
let mut node = Node::new("step");
node.attrs.insert(
"tool_hooks.post".to_string(),
AttrValue::String("exit 1".to_string()),
);
let context = Context::new();
let graph = Graph::new("test");
let tmp = TempDir::new().unwrap();
let outcome = handler
.execute(&node, &context, &graph, tmp.path(), &make_services())
.await
.unwrap();
// Post-hook failure should not fail the node
assert_eq!(outcome.status, crate::outcome::StageStatus::Success);
}
#[test]
fn resolve_hook_from_node_attr() {
let mut node = Node::new("step");
node.attrs.insert(
"tool_hooks.pre".to_string(),
AttrValue::String("echo node".to_string()),
);
let graph = Graph::new("test");
assert_eq!(
resolve_hook(&node, &graph, "tool_hooks.pre"),
Some("echo node".to_string())
);
}
#[test]
fn resolve_hook_falls_back_to_graph() {
let node = Node::new("step");
let mut graph = Graph::new("test");
graph.attrs.insert(
"tool_hooks.pre".to_string(),
AttrValue::String("echo graph".to_string()),
);
assert_eq!(
resolve_hook(&node, &graph, "tool_hooks.pre"),
Some("echo graph".to_string())
);
}
#[test]
fn resolve_hook_none_when_missing() {
let node = Node::new("step");
let graph = Graph::new("test");
assert_eq!(resolve_hook(&node, &graph, "tool_hooks.pre"), None);
}
#[tokio::test]
async fn codergen_handler_passes_thread_id_to_backend() {
use std::sync::{Arc, Mutex};

1
crates/arc-workflows/src/handler/conditional.rs
View file

@ -44,6 +44,7 @@ mod tests {
std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from(".")),
)),
git_state: std::sync::RwLock::new(None),
hook_runner: None,
}
}

1
crates/arc-workflows/src/handler/exit.rs
View file

@ -41,6 +41,7 @@ mod tests {
std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from(".")),
)),
git_state: std::sync::RwLock::new(None),
hook_runner: None,
}
}

1
crates/arc-workflows/src/handler/fan_in.rs
View file

@ -305,6 +305,7 @@ mod tests {
std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from(".")),
)),
git_state: std::sync::RwLock::new(None),
hook_runner: None,
}
}

4
crates/arc-workflows/src/handler/manager_loop.rs
View file

@ -247,6 +247,7 @@ mod tests {
std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from(".")),
)),
git_state: std::sync::RwLock::new(None),
hook_runner: None,
}
}
@ -378,6 +379,7 @@ mod tests {
std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from(".")),
)),
git_state: std::sync::RwLock::new(None),
hook_runner: None,
};
let handler = SubWorkflowHandler;
@ -485,6 +487,7 @@ mod tests {
std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from(".")),
)),
git_state: std::sync::RwLock::new(None),
hook_runner: None,
};
let handler = SubWorkflowHandler;
@ -544,6 +547,7 @@ mod tests {
std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from(".")),
)),
git_state: std::sync::RwLock::new(None),
hook_runner: None,
};
let handler = SubWorkflowHandler;

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

@ -21,6 +21,7 @@ use crate::engine::GitState;
use crate::error::ArcError;
use crate::event::EventEmitter;
use crate::graph::{shape_to_handler_type, Graph, Node};
use crate::hook::HookRunner;
use crate::interviewer::Interviewer;
use crate::outcome::Outcome;
@ -32,6 +33,8 @@ pub struct EngineServices {
/// Git state for the current run. Set via `set_git_state` at the start of
/// `run_internal` and read by parallel/fan-in handlers.
pub(crate) git_state: std::sync::RwLock<Option<Arc<GitState>>>,
/// Hook runner for user-defined lifecycle hooks.
pub hook_runner: Option<Arc<HookRunner>>,
}
impl EngineServices {

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

@ -434,6 +434,7 @@ impl Handler for ParallelHandler {
emitter: Arc::clone(&emitter),
sandbox: Arc::clone(&setup.sandbox),
git_state: std::sync::RwLock::new(None),
hook_runner: None,
};
let handler = registry.resolve(target_node);
let outcome = handler
@ -764,6 +765,7 @@ mod tests {
std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from(".")),
)),
git_state: std::sync::RwLock::new(None),
hook_runner: None,
}
}

1
crates/arc-workflows/src/handler/script.rs
View file

@ -181,6 +181,7 @@ mod tests {
std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from(".")),
)),
git_state: std::sync::RwLock::new(None),
hook_runner: None,
}
}

1
crates/arc-workflows/src/handler/start.rs
View file

@ -40,6 +40,7 @@ mod tests {
std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from(".")),
)),
git_state: std::sync::RwLock::new(None),
hook_runner: None,
}
}

1
crates/arc-workflows/src/handler/wait_human.rs
View file

@ -282,6 +282,7 @@ mod tests {
std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from(".")),
)),
git_state: std::sync::RwLock::new(None),
hook_runner: None,
}
}

1
crates/arc-workflows/src/handler/wait_timer.rs
View file

@ -55,6 +55,7 @@ mod tests {
std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from(".")),
)),
git_state: std::sync::RwLock::new(None),
hook_runner: None,
}
}

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

@ -0,0 +1,379 @@
use serde::Deserialize;
use super::types::HookEvent;
/// How a hook is executed.
#[derive(Debug, Clone, Deserialize, PartialEq)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum HookType {
Command { command: String },
Http { url: String, headers: Option<std::collections::HashMap<String, String>> },
}
/// A single hook definition.
#[derive(Debug, Clone, Deserialize, PartialEq)]
pub struct HookDefinition {
pub name: Option<String>,
pub event: HookEvent,
/// Inline command shorthand — if set, implies `type = "command"`.
#[serde(default)]
pub command: Option<String>,
/// Explicit hook type (command or http). If omitted and `command` is set,
/// defaults to `Command`.
#[serde(flatten)]
pub hook_type: Option<HookType>,
/// Regex matched against node_id, handler_type, or event-specific fields.
pub matcher: Option<String>,
/// Override the event's default blocking behavior.
pub blocking: Option<bool>,
/// Timeout in milliseconds (default: 60_000).
pub timeout_ms: Option<u64>,
/// Run inside the sandbox (true, default) or on the host (false).
pub sandbox: Option<bool>,
}
impl HookDefinition {
/// Resolve the effective hook type: explicit `hook_type` wins, then `command`
/// shorthand, then error.
pub fn resolved_hook_type(&self) -> Option<HookType> {
if let Some(ref ht) = self.hook_type {
return Some(ht.clone());
}
self.command
.as_ref()
.map(|cmd| HookType::Command { command: cmd.clone() })
}
/// Whether this hook is blocking for its event.
#[must_use]
pub fn is_blocking(&self) -> bool {
self.blocking.unwrap_or_else(|| self.event.is_blocking_by_default())
}
/// Timeout duration for this hook.
#[must_use]
pub fn timeout(&self) -> std::time::Duration {
std::time::Duration::from_millis(self.timeout_ms.unwrap_or(60_000))
}
/// Whether this hook runs in the sandbox.
#[must_use]
pub fn runs_in_sandbox(&self) -> bool {
self.sandbox.unwrap_or(true)
}
/// The effective name: explicit name or a generated one.
#[must_use]
pub fn effective_name(&self) -> String {
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));
match self.resolved_hook_type() {
Some(HookType::Command { ref command }) => {
let short = if command.len() > 20 {
&command[..20]
} else {
command
};
format!("{event_str}:{short}")
}
Some(HookType::Http { ref url, .. }) => format!("{event_str}:{url}"),
None => event_str,
}
}
}
/// Top-level hook configuration: a list of hook definitions.
#[derive(Debug, Clone, Default, Deserialize, PartialEq)]
pub struct HookConfig {
#[serde(default)]
pub hooks: Vec<HookDefinition>,
}
impl HookConfig {
/// Merge with another config. Concatenates lists; on name collisions, `other` wins.
#[must_use]
pub fn merge(self, other: Self) -> Self {
let mut by_name: std::collections::HashMap<String, HookDefinition> =
std::collections::HashMap::new();
let mut order: Vec<String> = Vec::new();
for hook in self.hooks {
let name = hook.effective_name();
if !by_name.contains_key(&name) {
order.push(name.clone());
}
by_name.insert(name, hook);
}
for hook in other.hooks {
let name = hook.effective_name();
if !by_name.contains_key(&name) {
order.push(name.clone());
}
by_name.insert(name, hook);
}
let hooks = order
.into_iter()
.filter_map(|name| by_name.remove(&name))
.collect();
Self { hooks }
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn parse_command_shorthand() {
let toml = r#"
[[hooks]]
event = "stage_start"
command = "./scripts/pre-check.sh"
"#;
let config: HookConfig = toml::from_str(toml).unwrap();
assert_eq!(config.hooks.len(), 1);
let hook = &config.hooks[0];
assert_eq!(hook.event, HookEvent::StageStart);
assert_eq!(hook.command.as_deref(), Some("./scripts/pre-check.sh"));
let resolved = hook.resolved_hook_type().unwrap();
assert!(matches!(resolved, HookType::Command { command } if command == "./scripts/pre-check.sh"));
}
#[test]
fn parse_explicit_command_type() {
let toml = r#"
[[hooks]]
event = "run_start"
type = "command"
command = "echo hello"
"#;
let config: HookConfig = toml::from_str(toml).unwrap();
assert_eq!(config.hooks.len(), 1);
let hook = &config.hooks[0];
assert_eq!(hook.event, HookEvent::RunStart);
assert!(hook.resolved_hook_type().is_some());
}
#[test]
fn parse_http_hook() {
let toml = r#"
[[hooks]]
event = "run_complete"
type = "http"
url = "https://hooks.example.com/done"
"#;
let config: HookConfig = toml::from_str(toml).unwrap();
let hook = &config.hooks[0];
assert!(matches!(
hook.resolved_hook_type(),
Some(HookType::Http { url, .. }) if url == "https://hooks.example.com/done"
));
}
#[test]
fn parse_full_hook_definition() {
let toml = r#"
[[hooks]]
name = "pre-check"
event = "stage_start"
command = "./check.sh"
matcher = "codergen"
blocking = true
timeout_ms = 30000
sandbox = false
"#;
let config: HookConfig = toml::from_str(toml).unwrap();
let hook = &config.hooks[0];
assert_eq!(hook.name.as_deref(), Some("pre-check"));
assert_eq!(hook.event, HookEvent::StageStart);
assert_eq!(hook.matcher.as_deref(), Some("codergen"));
assert!(hook.is_blocking());
assert_eq!(hook.timeout(), std::time::Duration::from_millis(30_000));
assert!(!hook.runs_in_sandbox());
}
#[test]
fn blocking_defaults_to_event() {
let blocking_def = HookDefinition {
name: None,
event: HookEvent::StageStart,
command: Some("echo".into()),
hook_type: None,
matcher: None,
blocking: None,
timeout_ms: None,
sandbox: None,
};
assert!(blocking_def.is_blocking());
let non_blocking_def = HookDefinition {
event: HookEvent::StageComplete,
..blocking_def.clone()
};
assert!(!non_blocking_def.is_blocking());
}
#[test]
fn blocking_override() {
let def = HookDefinition {
name: None,
event: HookEvent::StageComplete,
command: Some("echo".into()),
hook_type: None,
matcher: None,
blocking: Some(true),
timeout_ms: None,
sandbox: None,
};
assert!(def.is_blocking());
}
#[test]
fn timeout_defaults_to_60s() {
let def = HookDefinition {
name: None,
event: HookEvent::RunStart,
command: Some("echo".into()),
hook_type: None,
matcher: None,
blocking: None,
timeout_ms: None,
sandbox: None,
};
assert_eq!(def.timeout(), std::time::Duration::from_secs(60));
}
#[test]
fn sandbox_defaults_to_true() {
let def = HookDefinition {
name: None,
event: HookEvent::RunStart,
command: Some("echo".into()),
hook_type: None,
matcher: None,
blocking: None,
timeout_ms: None,
sandbox: None,
};
assert!(def.runs_in_sandbox());
}
#[test]
fn effective_name_uses_explicit() {
let def = HookDefinition {
name: Some("my-hook".into()),
event: HookEvent::RunStart,
command: Some("echo hi".into()),
hook_type: None,
matcher: None,
blocking: None,
timeout_ms: None,
sandbox: None,
};
assert_eq!(def.effective_name(), "my-hook");
}
#[test]
fn effective_name_generated_from_event_and_command() {
let def = HookDefinition {
name: None,
event: HookEvent::RunStart,
command: Some("echo hi".into()),
hook_type: None,
matcher: None,
blocking: None,
timeout_ms: None,
sandbox: None,
};
assert_eq!(def.effective_name(), "run_start:echo hi");
}
#[test]
fn config_merge_concatenates() {
let a = HookConfig {
hooks: vec![HookDefinition {
name: Some("hook-a".into()),
event: HookEvent::RunStart,
command: Some("echo a".into()),
hook_type: None,
matcher: None,
blocking: None,
timeout_ms: None,
sandbox: None,
}],
};
let b = HookConfig {
hooks: vec![HookDefinition {
name: Some("hook-b".into()),
event: HookEvent::RunComplete,
command: Some("echo b".into()),
hook_type: None,
matcher: None,
blocking: None,
timeout_ms: None,
sandbox: None,
}],
};
let merged = a.merge(b);
assert_eq!(merged.hooks.len(), 2);
assert_eq!(merged.hooks[0].name.as_deref(), Some("hook-a"));
assert_eq!(merged.hooks[1].name.as_deref(), Some("hook-b"));
}
#[test]
fn config_merge_name_collision_later_wins() {
let a = HookConfig {
hooks: vec![HookDefinition {
name: Some("shared".into()),
event: HookEvent::RunStart,
command: Some("echo a".into()),
hook_type: None,
matcher: None,
blocking: None,
timeout_ms: None,
sandbox: None,
}],
};
let b = HookConfig {
hooks: vec![HookDefinition {
name: Some("shared".into()),
event: HookEvent::RunComplete,
command: Some("echo b".into()),
hook_type: None,
matcher: None,
blocking: None,
timeout_ms: None,
sandbox: None,
}],
};
let merged = a.merge(b);
assert_eq!(merged.hooks.len(), 1);
assert_eq!(merged.hooks[0].event, HookEvent::RunComplete);
}
#[test]
fn parse_multiple_hooks() {
let toml = r#"
[[hooks]]
event = "run_start"
command = "echo start"
[[hooks]]
event = "stage_complete"
command = "echo done"
matcher = "codergen"
"#;
let config: HookConfig = toml::from_str(toml).unwrap();
assert_eq!(config.hooks.len(), 2);
assert_eq!(config.hooks[0].event, HookEvent::RunStart);
assert_eq!(config.hooks[1].event, HookEvent::StageComplete);
assert_eq!(config.hooks[1].matcher.as_deref(), Some("codergen"));
}
}

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

@ -0,0 +1,309 @@
use std::collections::HashMap;
use std::path::Path;
use std::time::Instant;
use async_trait::async_trait;
use arc_agent::Sandbox;
use super::config::HookDefinition;
use super::types::{HookContext, HookDecision, HookResult};
/// Trait for executing hooks via different transports.
#[async_trait]
pub trait HookExecutor: Send + Sync {
async fn execute(
&self,
definition: &HookDefinition,
context: &HookContext,
sandbox: &dyn Sandbox,
work_dir: Option<&Path>,
) -> HookResult;
}
/// Executes hooks as shell commands (host or sandbox).
pub struct CommandHookExecutor;
impl CommandHookExecutor {
/// Parse a hook decision from JSON stdout and exit code.
fn parse_decision(exit_code: i32, stdout: &str) -> HookDecision {
if exit_code == 0 {
// Try parsing JSON response for explicit decision
if let Ok(decision) = serde_json::from_str::<HookDecision>(stdout.trim()) {
return decision;
}
HookDecision::Proceed
} else if exit_code == 2 {
// Exit 2 = block/skip
if let Ok(decision) = serde_json::from_str::<HookDecision>(stdout.trim()) {
return decision;
}
HookDecision::Block {
reason: Some(format!("hook exited with code 2")),
}
} else {
HookDecision::Block {
reason: Some(format!("hook exited with code {exit_code}")),
}
}
}
}
#[async_trait]
impl HookExecutor for CommandHookExecutor {
async fn execute(
&self,
definition: &HookDefinition,
context: &HookContext,
sandbox: &dyn Sandbox,
work_dir: Option<&Path>,
) -> HookResult {
let start = Instant::now();
let command = match definition.resolved_hook_type() {
Some(super::config::HookType::Command { ref command }) => command.clone(),
_ => {
return HookResult {
hook_name: definition.name.clone(),
decision: HookDecision::Block {
reason: Some("no command specified".into()),
},
duration_ms: 0,
};
}
};
let context_json = serde_json::to_string(context).unwrap_or_default();
let timeout_ms = definition.timeout().as_millis() as u64;
let mut env_vars = HashMap::new();
env_vars.insert("ARC_EVENT".to_string(), context.event.to_string());
env_vars.insert("ARC_RUN_ID".to_string(), context.run_id.clone());
env_vars.insert("ARC_WORKFLOW".to_string(), context.workflow_name.clone());
if let Some(ref node_id) = context.node_id {
env_vars.insert("ARC_NODE_ID".to_string(), node_id.clone());
}
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());
}
match sandbox
.exec_command(&command, timeout_ms, None, Some(&env_vars), None)
.await
{
Ok(result) => Self::parse_decision(result.exit_code, &result.stdout),
Err(e) => HookDecision::Block {
reason: Some(format!("sandbox exec failed: {e}")),
},
}
} else {
// Run on host via sh -c
let mut cmd = std::process::Command::new("sh");
cmd.arg("-c").arg(&command);
if let Some(wd) = work_dir {
cmd.current_dir(wd);
}
for (k, v) in &env_vars {
cmd.env(k, v);
}
// Pipe context JSON to stdin
cmd.stdin(std::process::Stdio::piped());
cmd.stdout(std::process::Stdio::piped());
cmd.stderr(std::process::Stdio::piped());
match cmd.spawn() {
Ok(mut child) => {
// Write context to stdin
if let Some(mut stdin) = child.stdin.take() {
use std::io::Write;
let _ = stdin.write_all(context_json.as_bytes());
}
match child.wait_with_output() {
Ok(output) => {
let exit_code = output.status.code().unwrap_or(1);
let stdout = String::from_utf8_lossy(&output.stdout);
Self::parse_decision(exit_code, &stdout)
}
Err(e) => HookDecision::Block {
reason: Some(format!("command wait failed: {e}")),
},
}
}
Err(e) => HookDecision::Block {
reason: Some(format!("command spawn failed: {e}")),
},
}
};
let duration_ms = start.elapsed().as_millis() as u64;
HookResult {
hook_name: definition.name.clone(),
decision,
duration_ms,
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::hook::config::HookType;
use crate::hook::types::HookEvent;
fn make_context() -> HookContext {
HookContext::new(HookEvent::StageStart, "run-1".into(), "test-wf".into())
}
fn make_definition(command: &str) -> HookDefinition {
HookDefinition {
name: Some("test-hook".into()),
event: HookEvent::StageStart,
command: Some(command.into()),
hook_type: None,
matcher: None,
blocking: None,
timeout_ms: Some(5000),
sandbox: Some(false), // host execution for tests
}
}
#[test]
fn parse_decision_exit_0_proceed() {
assert_eq!(
CommandHookExecutor::parse_decision(0, ""),
HookDecision::Proceed
);
}
#[test]
fn parse_decision_exit_0_with_json() {
let json = r#"{"decision": "skip", "reason": "not needed"}"#;
assert_eq!(
CommandHookExecutor::parse_decision(0, json),
HookDecision::Skip {
reason: Some("not needed".into())
}
);
}
#[test]
fn parse_decision_exit_2_block() {
assert!(matches!(
CommandHookExecutor::parse_decision(2, ""),
HookDecision::Block { .. }
));
}
#[test]
fn parse_decision_exit_2_with_json() {
let json = r#"{"decision": "skip", "reason": "skipping"}"#;
assert_eq!(
CommandHookExecutor::parse_decision(2, json),
HookDecision::Skip {
reason: Some("skipping".into())
}
);
}
#[test]
fn parse_decision_exit_1_block() {
assert!(matches!(
CommandHookExecutor::parse_decision(1, ""),
HookDecision::Block { .. }
));
}
#[test]
fn parse_decision_exit_0_override() {
let json = r#"{"decision": "override", "edge_to": "node_b"}"#;
assert_eq!(
CommandHookExecutor::parse_decision(0, json),
HookDecision::Override {
edge_to: "node_b".into()
}
);
}
#[tokio::test]
async fn command_executor_host_success() {
let executor = CommandHookExecutor;
let def = make_definition("exit 0");
let ctx = make_context();
let sandbox = arc_agent::LocalSandbox::new(std::env::current_dir().unwrap());
let result = executor.execute(&def, &ctx, &sandbox, None).await;
assert_eq!(result.decision, HookDecision::Proceed);
assert_eq!(result.hook_name.as_deref(), Some("test-hook"));
}
#[tokio::test]
async fn command_executor_host_failure() {
let executor = CommandHookExecutor;
let def = make_definition("exit 1");
let ctx = make_context();
let sandbox = arc_agent::LocalSandbox::new(std::env::current_dir().unwrap());
let result = executor.execute(&def, &ctx, &sandbox, None).await;
assert!(matches!(result.decision, HookDecision::Block { .. }));
}
#[tokio::test]
async fn command_executor_host_skip_via_exit_2() {
let executor = CommandHookExecutor;
let def = make_definition("exit 2");
let ctx = make_context();
let sandbox = arc_agent::LocalSandbox::new(std::env::current_dir().unwrap());
let result = executor.execute(&def, &ctx, &sandbox, None).await;
assert!(matches!(result.decision, HookDecision::Block { .. }));
}
#[tokio::test]
async fn command_executor_host_json_decision() {
let executor = CommandHookExecutor;
let def =
make_definition(r#"echo '{"decision": "skip", "reason": "test skip"}'"#);
let ctx = make_context();
let sandbox = arc_agent::LocalSandbox::new(std::env::current_dir().unwrap());
let result = executor.execute(&def, &ctx, &sandbox, None).await;
assert_eq!(
result.decision,
HookDecision::Skip {
reason: Some("test skip".into())
}
);
}
#[tokio::test]
async fn command_executor_env_vars_set() {
let executor = CommandHookExecutor;
// Print env vars to stdout for verification
let def = make_definition("echo $ARC_EVENT:$ARC_RUN_ID:$ARC_WORKFLOW");
let mut ctx = make_context();
ctx.node_id = Some("plan".into());
let sandbox = arc_agent::LocalSandbox::new(std::env::current_dir().unwrap());
let result = executor.execute(&def, &ctx, &sandbox, None).await;
assert_eq!(result.decision, HookDecision::Proceed);
}
#[tokio::test]
async fn command_executor_no_command_blocks() {
let executor = CommandHookExecutor;
let def = HookDefinition {
name: None,
event: HookEvent::StageStart,
command: None,
hook_type: Some(HookType::Http {
url: "http://example.com".into(),
headers: None,
}),
matcher: None,
blocking: None,
timeout_ms: None,
sandbox: Some(false),
};
let ctx = make_context();
let sandbox = arc_agent::LocalSandbox::new(std::env::current_dir().unwrap());
let result = executor.execute(&def, &ctx, &sandbox, None).await;
assert!(matches!(result.decision, HookDecision::Block { .. }));
}
}

8
crates/arc-workflows/src/hook/mod.rs Normal file
View file

@ -0,0 +1,8 @@
pub mod config;
pub mod executor;
pub mod runner;
pub mod types;
pub use config::{HookConfig, HookDefinition, HookType};
pub use runner::HookRunner;
pub use types::{HookContext, HookDecision, HookEvent};

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

@ -0,0 +1,446 @@
use std::path::Path;
use std::sync::Arc;
use arc_agent::Sandbox;
use super::config::{HookConfig, HookDefinition};
use super::executor::{CommandHookExecutor, HookExecutor};
use super::types::{HookContext, HookDecision};
/// Central orchestrator: filters matching hooks, executes them, merges decisions.
pub struct HookRunner {
config: HookConfig,
command_executor: Arc<dyn HookExecutor>,
}
impl HookRunner {
#[must_use]
pub fn new(config: HookConfig) -> Self {
Self {
config,
command_executor: Arc::new(CommandHookExecutor),
}
}
/// Create a HookRunner with a custom executor (for testing).
#[cfg(test)]
pub fn with_executor(config: HookConfig, executor: Arc<dyn HookExecutor>) -> Self {
Self {
config,
command_executor: executor,
}
}
/// Run all matching hooks for the given event and return the merged decision.
pub async fn run(
&self,
context: &HookContext,
sandbox: &dyn Sandbox,
work_dir: Option<&Path>,
) -> HookDecision {
let matching = self.filter_hooks(context);
if matching.is_empty() {
return HookDecision::Proceed;
}
let hooks_matched = matching.len();
tracing::info!(
event = %context.event,
hooks_matched,
"Running hooks"
);
let any_blocking = matching.iter().any(|h| h.is_blocking());
let decision = if any_blocking {
// Sequential execution for blocking hooks, short-circuit on first Block
self.run_sequential(&matching, context, sandbox, work_dir)
.await
} else {
// Parallel execution for non-blocking hooks
self.run_parallel(&matching, context, sandbox, work_dir)
.await
};
tracing::info!(
event = %context.event,
decision = ?decision,
"Hooks complete"
);
decision
}
/// Filter hooks that match the given event and context.
fn filter_hooks(&self, context: &HookContext) -> Vec<&HookDefinition> {
self.config
.hooks
.iter()
.filter(|h| h.event == context.event)
.filter(|h| self.matches(h, context))
.collect()
}
/// Check if a hook's matcher applies to this context.
fn matches(&self, hook: &HookDefinition, context: &HookContext) -> bool {
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"
);
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
}
async fn run_sequential(
&self,
hooks: &[&HookDefinition],
context: &HookContext,
sandbox: &dyn Sandbox,
work_dir: Option<&Path>,
) -> HookDecision {
let mut merged = HookDecision::Proceed;
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 hook.is_blocking() {
merged = merged.merge(result.decision);
// Short-circuit on Block
if matches!(merged, HookDecision::Block { .. }) {
tracing::error!(
hook = %hook.effective_name(),
event = %context.event,
decision = ?merged,
"Hook blocked execution"
);
return merged;
}
} else if !result.decision.is_proceed() {
tracing::warn!(
hook = %hook.effective_name(),
event = %context.event,
decision = ?result.decision,
"Non-blocking hook returned non-proceed, ignoring"
);
}
}
merged
}
async fn run_parallel(
&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;
}
HookDecision::Proceed
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::hook::config::HookConfig;
use crate::hook::types::{HookContext, HookEvent, HookResult};
struct MockExecutor {
decision: HookDecision,
}
#[async_trait::async_trait]
impl HookExecutor for MockExecutor {
async fn execute(
&self,
definition: &HookDefinition,
_context: &HookContext,
_sandbox: &dyn Sandbox,
_work_dir: Option<&Path>,
) -> HookResult {
HookResult {
hook_name: definition.name.clone(),
decision: self.decision.clone(),
duration_ms: 1,
}
}
}
fn make_sandbox() -> arc_agent::LocalSandbox {
arc_agent::LocalSandbox::new(std::env::current_dir().unwrap())
}
fn make_context(event: HookEvent) -> HookContext {
HookContext::new(event, "run-1".into(), "test-wf".into())
}
fn make_hook(event: HookEvent, name: &str) -> HookDefinition {
HookDefinition {
name: Some(name.into()),
event,
command: Some("echo test".into()),
hook_type: None,
matcher: None,
blocking: None,
timeout_ms: None,
sandbox: Some(false),
}
}
#[tokio::test]
async fn no_hooks_returns_proceed() {
let runner = HookRunner::new(HookConfig::default());
let ctx = make_context(HookEvent::RunStart);
let sandbox = make_sandbox();
let decision = runner.run(&ctx, &sandbox, None).await;
assert_eq!(decision, HookDecision::Proceed);
}
#[tokio::test]
async fn filters_by_event() {
let config = HookConfig {
hooks: vec![
make_hook(HookEvent::RunStart, "a"),
make_hook(HookEvent::StageStart, "b"),
],
};
let runner = HookRunner::with_executor(
config,
Arc::new(MockExecutor {
decision: HookDecision::Proceed,
}),
);
let ctx = make_context(HookEvent::RunStart);
let matching = runner.filter_hooks(&ctx);
assert_eq!(matching.len(), 1);
assert_eq!(matching[0].name.as_deref(), Some("a"));
}
#[tokio::test]
async fn matcher_filters_by_node_id() {
let mut hook = make_hook(HookEvent::StageStart, "filtered");
hook.matcher = Some("codergen".into());
let config = HookConfig {
hooks: vec![hook],
};
let runner = HookRunner::with_executor(
config,
Arc::new(MockExecutor {
decision: HookDecision::Proceed,
}),
);
// No node_id — no match
let ctx = make_context(HookEvent::StageStart);
assert!(runner.filter_hooks(&ctx).is_empty());
// Matching node_id
let mut ctx = make_context(HookEvent::StageStart);
ctx.node_id = Some("codergen_step".into());
assert_eq!(runner.filter_hooks(&ctx).len(), 1);
// Non-matching node_id
let mut ctx = make_context(HookEvent::StageStart);
ctx.node_id = Some("start".into());
assert!(runner.filter_hooks(&ctx).is_empty());
}
#[tokio::test]
async fn matcher_filters_by_handler_type() {
let mut hook = make_hook(HookEvent::StageStart, "filtered");
hook.matcher = Some("^codergen$".into());
let config = HookConfig {
hooks: vec![hook],
};
let runner = HookRunner::with_executor(
config,
Arc::new(MockExecutor {
decision: HookDecision::Proceed,
}),
);
let mut ctx = make_context(HookEvent::StageStart);
ctx.handler_type = Some("codergen".into());
assert_eq!(runner.filter_hooks(&ctx).len(), 1);
let mut ctx = make_context(HookEvent::StageStart);
ctx.handler_type = Some("script".into());
assert!(runner.filter_hooks(&ctx).is_empty());
}
#[tokio::test]
async fn blocking_hook_block_decision() {
let config = HookConfig {
hooks: vec![make_hook(HookEvent::RunStart, "blocker")],
};
let runner = HookRunner::with_executor(
config,
Arc::new(MockExecutor {
decision: HookDecision::Block {
reason: Some("denied".into()),
},
}),
);
let ctx = make_context(HookEvent::RunStart);
let sandbox = make_sandbox();
let decision = runner.run(&ctx, &sandbox, None).await;
assert!(matches!(decision, HookDecision::Block { .. }));
}
#[tokio::test]
async fn blocking_hook_skip_decision() {
let mut hook = make_hook(HookEvent::StageStart, "skipper");
hook.blocking = Some(true);
let config = HookConfig {
hooks: vec![hook],
};
let runner = HookRunner::with_executor(
config,
Arc::new(MockExecutor {
decision: HookDecision::Skip {
reason: Some("skip it".into()),
},
}),
);
let ctx = make_context(HookEvent::StageStart);
let sandbox = make_sandbox();
let decision = runner.run(&ctx, &sandbox, None).await;
assert!(matches!(decision, HookDecision::Skip { .. }));
}
#[tokio::test]
async fn non_blocking_hook_doesnt_block() {
let mut hook = make_hook(HookEvent::StageComplete, "observer");
hook.blocking = Some(false);
let config = HookConfig {
hooks: vec![hook],
};
let runner = HookRunner::with_executor(
config,
Arc::new(MockExecutor {
decision: HookDecision::Block {
reason: Some("ignored".into()),
},
}),
);
let ctx = make_context(HookEvent::StageComplete);
let sandbox = make_sandbox();
let decision = runner.run(&ctx, &sandbox, None).await;
// Non-blocking hooks don't affect the decision
assert_eq!(decision, HookDecision::Proceed);
}
#[tokio::test]
async fn command_executor_integration_success() {
let config = HookConfig {
hooks: vec![{
let mut h = make_hook(HookEvent::RunStart, "echo-hook");
h.command = Some("exit 0".into());
h
}],
};
let runner = HookRunner::new(config);
let ctx = make_context(HookEvent::RunStart);
let sandbox = make_sandbox();
let decision = runner.run(&ctx, &sandbox, None).await;
assert_eq!(decision, HookDecision::Proceed);
}
#[tokio::test]
async fn command_executor_integration_block() {
let config = HookConfig {
hooks: vec![{
let mut h = make_hook(HookEvent::RunStart, "fail-hook");
h.command = Some("exit 1".into());
h
}],
};
let runner = HookRunner::new(config);
let ctx = make_context(HookEvent::RunStart);
let sandbox = make_sandbox();
let decision = runner.run(&ctx, &sandbox, None).await;
assert!(matches!(decision, HookDecision::Block { .. }));
}
}

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

@ -0,0 +1,317 @@
use serde::{Deserialize, Serialize};
/// Lifecycle events that can trigger user-defined hooks.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum HookEvent {
RunStart,
RunComplete,
RunFailed,
StageStart,
StageComplete,
StageFailed,
StageRetrying,
EdgeSelected,
ParallelStart,
ParallelComplete,
SandboxReady,
SandboxCleanup,
CheckpointSaved,
}
impl HookEvent {
/// Whether hooks for this event block execution by default.
#[must_use]
pub fn is_blocking_by_default(self) -> bool {
matches!(self, Self::RunStart | Self::StageStart | Self::EdgeSelected)
}
}
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)
}
}
/// Rich JSON payload sent to hooks.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HookContext {
pub event: HookEvent,
pub run_id: String,
pub workflow_name: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub cwd: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub node_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub node_label: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub handler_type: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub status: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub edge_from: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub edge_to: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub edge_label: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub failure_reason: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub attempt: Option<usize>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub max_attempts: Option<usize>,
}
impl HookContext {
#[must_use]
pub fn new(event: HookEvent, run_id: String, workflow_name: String) -> Self {
Self {
event,
run_id,
workflow_name,
cwd: None,
node_id: None,
node_label: None,
handler_type: None,
status: None,
edge_from: None,
edge_to: None,
edge_label: None,
failure_reason: None,
attempt: None,
max_attempts: None,
}
}
}
/// Decision returned by blocking hooks.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "decision", rename_all = "snake_case")]
pub enum HookDecision {
Proceed,
Skip {
#[serde(default)]
reason: Option<String>,
},
Block {
#[serde(default)]
reason: Option<String>,
},
Override {
edge_to: String,
},
}
impl Default for HookDecision {
fn default() -> Self {
Self::Proceed
}
}
impl HookDecision {
/// Merge two decisions. Block > Skip/Override > Proceed.
#[must_use]
pub fn merge(self, other: Self) -> Self {
match (&self, &other) {
(Self::Block { .. }, _) => self,
(_, Self::Block { .. }) => other,
(Self::Skip { .. }, _) | (Self::Override { .. }, _) => self,
(_, Self::Skip { .. }) | (_, Self::Override { .. }) => other,
_ => Self::Proceed,
}
}
#[must_use]
pub fn is_proceed(&self) -> bool {
matches!(self, Self::Proceed)
}
}
/// Result from executing a single hook.
#[derive(Debug, Clone)]
pub struct HookResult {
pub hook_name: Option<String>,
pub decision: HookDecision,
pub duration_ms: u64,
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn hook_event_serde_round_trip() {
let events = [
HookEvent::RunStart,
HookEvent::RunComplete,
HookEvent::RunFailed,
HookEvent::StageStart,
HookEvent::StageComplete,
HookEvent::StageFailed,
HookEvent::StageRetrying,
HookEvent::EdgeSelected,
HookEvent::ParallelStart,
HookEvent::ParallelComplete,
HookEvent::SandboxReady,
HookEvent::SandboxCleanup,
HookEvent::CheckpointSaved,
];
for event in events {
let json = serde_json::to_string(&event).unwrap();
let back: HookEvent = serde_json::from_str(&json).unwrap();
assert_eq!(event, back);
}
}
#[test]
fn hook_event_serializes_as_snake_case() {
assert_eq!(
serde_json::to_string(&HookEvent::RunStart).unwrap(),
"\"run_start\""
);
assert_eq!(
serde_json::to_string(&HookEvent::StageRetrying).unwrap(),
"\"stage_retrying\""
);
}
#[test]
fn hook_event_display() {
assert_eq!(HookEvent::RunStart.to_string(), "run_start");
assert_eq!(HookEvent::CheckpointSaved.to_string(), "checkpoint_saved");
}
#[test]
fn hook_event_blocking_defaults() {
assert!(HookEvent::RunStart.is_blocking_by_default());
assert!(HookEvent::StageStart.is_blocking_by_default());
assert!(HookEvent::EdgeSelected.is_blocking_by_default());
assert!(!HookEvent::RunComplete.is_blocking_by_default());
assert!(!HookEvent::StageFailed.is_blocking_by_default());
assert!(!HookEvent::CheckpointSaved.is_blocking_by_default());
}
#[test]
fn hook_context_serde_round_trip() {
let ctx = HookContext {
event: HookEvent::StageStart,
run_id: "run-123".into(),
workflow_name: "test-wf".into(),
cwd: Some("/tmp".into()),
node_id: Some("plan".into()),
node_label: Some("Plan".into()),
handler_type: Some("codergen".into()),
status: None,
edge_from: None,
edge_to: None,
edge_label: None,
failure_reason: None,
attempt: Some(1),
max_attempts: Some(3),
};
let json = serde_json::to_string(&ctx).unwrap();
let back: HookContext = serde_json::from_str(&json).unwrap();
assert_eq!(back.event, HookEvent::StageStart);
assert_eq!(back.run_id, "run-123");
assert_eq!(back.node_id.as_deref(), Some("plan"));
}
#[test]
fn hook_context_omits_none_fields() {
let ctx = HookContext::new(
HookEvent::RunStart,
"run-1".into(),
"wf".into(),
);
let json = serde_json::to_string(&ctx).unwrap();
assert!(!json.contains("node_id"));
assert!(!json.contains("failure_reason"));
}
#[test]
fn hook_decision_serde_round_trip() {
let decisions = [
HookDecision::Proceed,
HookDecision::Skip {
reason: Some("not needed".into()),
},
HookDecision::Block {
reason: Some("forbidden".into()),
},
HookDecision::Override {
edge_to: "node_b".into(),
},
];
for decision in decisions {
let json = serde_json::to_string(&decision).unwrap();
let back: HookDecision = serde_json::from_str(&json).unwrap();
assert_eq!(decision, back);
}
}
#[test]
fn hook_decision_merge_block_wins() {
let block = HookDecision::Block {
reason: Some("no".into()),
};
let skip = HookDecision::Skip {
reason: Some("skip".into()),
};
let proceed = HookDecision::Proceed;
assert!(matches!(
proceed.clone().merge(block.clone()),
HookDecision::Block { .. }
));
assert!(matches!(
block.clone().merge(skip.clone()),
HookDecision::Block { .. }
));
assert!(matches!(
skip.clone().merge(block.clone()),
HookDecision::Block { .. }
));
}
#[test]
fn hook_decision_merge_skip_over_proceed() {
let skip = HookDecision::Skip {
reason: Some("skip".into()),
};
let proceed = HookDecision::Proceed;
assert!(matches!(
proceed.clone().merge(skip.clone()),
HookDecision::Skip { .. }
));
assert!(matches!(
skip.merge(proceed),
HookDecision::Skip { .. }
));
}
#[test]
fn hook_decision_merge_first_non_proceed_wins() {
let skip = HookDecision::Skip {
reason: Some("a".into()),
};
let override_d = HookDecision::Override {
edge_to: "x".into(),
};
// First non-Proceed wins when no Block
assert!(matches!(
skip.merge(override_d),
HookDecision::Skip { .. }
));
}
#[test]
fn hook_decision_default_is_proceed() {
assert_eq!(HookDecision::default(), HookDecision::Proceed);
}
}

1
crates/arc-workflows/src/lib.rs
View file

@ -11,6 +11,7 @@ pub mod event;
pub mod git;
pub mod graph;
pub mod handler;
pub mod hook;
pub mod interviewer;
pub mod outcome;
pub mod parser;

1199
crates/arc-workflows/tests/integration.rs
View file

File diff suppressed because it is too large Load diff