diff --git a/src/function/agents.rs b/src/function/agents.rs index 3baafaa..4e05b38 100644 --- a/src/function/agents.rs +++ b/src/function/agents.rs @@ -4,8 +4,9 @@ use crate::client::{Model, ModelType, call_chat_completions}; use crate::config::{ Agent, AgentVariable, AgentVariables, AppState, Input, RequestContext, Role, RoleLike, default_max_agent_depth, effective_max_concurrent_jobs, jobs_enabled, - list_agents_with_descriptions, load_agent_variables, + list_agents_with_descriptions, load_agent_variables, paths, }; +use crate::hooks::{self, HookEvent, ResolvedHook}; use crate::supervisor::mailbox::{Envelope, EnvelopePayload, Inbox, PeerRegistry, graph_agent_id}; use crate::supervisor::notification::agent_notification; use crate::supervisor::{AgentExitStatus, AgentHandle, AgentResult, Supervisor, TaskKind}; @@ -14,7 +15,7 @@ use crate::utils::{AbortSignal, create_abort_signal, wait_abort_signal, wait_use use crate::graph; use crate::repl::DEFAULT_CONTINUATION_PROMPT; use anyhow::{Context, Result, anyhow, bail}; -use chrono::Utc; +use chrono::{SecondsFormat, Utc}; use indexmap::IndexMap; use log::{debug, warn}; use parking_lot::RwLock; @@ -996,6 +997,86 @@ fn sync_agent_functions_to_ctx(ctx: &mut RequestContext) -> Result<()> { Ok(()) } +/// Mirrors the base-env contract `hooks::fire` assembles from a live context, +/// for detached call sites that dispatch pre-resolved hook snapshots after +/// the context they were resolved from is gone. The names are captured by +/// the caller when it takes the snapshot; the timestamp is taken here, at +/// fire time, so it reflects when the event actually happened. +pub(crate) fn hook_base_envs( + event: HookEvent, + session_name: Option<&str>, + agent_name: Option<&str>, +) -> Vec<(String, String)> { + let mut envs = vec![ + ("COYOTE_EVENT".to_string(), event.as_str().to_string()), + ( + "COYOTE_CONFIG_DIR".to_string(), + paths::config_dir().display().to_string(), + ), + ( + "COYOTE_EVENT_TIMESTAMP".to_string(), + Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true), + ), + ]; + if let Some(name) = session_name { + envs.push(("COYOTE_SESSION_ID".to_string(), name.to_string())); + } + if let Some(name) = agent_name { + envs.push(("COYOTE_AGENT_NAME".to_string(), name.to_string())); + } + envs +} + +/// Pre-resolved `agent.completed`/`agent.failed` snapshots for a spawned +/// agent's result arms, captured from the child context before the spawn +/// task takes ownership of it. Resolving against the child keeps global +/// hooks gated by the child's own whitelist, the same gate every other +/// event fired from the child's context goes through. +struct SpawnResultHooks { + completed: Vec, + failed: Vec, + session_name: Option, + agent_id: String, + agent_name: String, +} + +impl SpawnResultHooks { + fn resolve(child_ctx: &RequestContext, agent_id: &str, agent_name: &str) -> Self { + Self { + completed: child_ctx.resolved_hooks(HookEvent::AgentCompleted), + failed: child_ctx.resolved_hooks(HookEvent::AgentFailed), + session_name: child_ctx + .session + .as_ref() + .map(|session| session.name().to_string()), + agent_id: agent_id.to_string(), + agent_name: agent_name.to_string(), + } + } + + /// Fires `agent.completed`, or `agent.failed` when `error` is given. + fn fire(self, error: Option<&str>) { + let mut extras = vec![ + ("COYOTE_AGENT_ID", self.agent_id), + ("COYOTE_AGENT_NAME", self.agent_name.clone()), + ]; + let (event, resolved) = match error { + None => (HookEvent::AgentCompleted, self.completed), + Some(error) => { + extras.push(("COYOTE_AGENT_ERROR", error.to_string())); + (HookEvent::AgentFailed, self.failed) + } + }; + hooks::fire_resolved( + event, + resolved, + hook_base_envs(event, self.session_name.as_deref(), Some(&self.agent_name)), + &extras, + None, + ); + } +} + async fn handle_spawn(ctx: &mut RequestContext, args: &Value) -> Result { let agent_name = args .get("agent") @@ -1149,6 +1230,19 @@ async fn handle_spawn(ctx: &mut RequestContext, args: &Value) -> Result { debug!("Spawning child agent '{agent_name}' as '{agent_id}'"); + hooks::fire( + HookEvent::AgentStarted, + &child_ctx, + &[ + ("COYOTE_AGENT_ID", agent_id.clone()), + ("COYOTE_AGENT_NAME", agent_name.clone()), + ], + None, + ); + // Resolved before the spawn: the result arms outlive the child context + // the snapshots come from. + let result_hooks = SpawnResultHooks::resolve(&child_ctx, &agent_id, &agent_name); + let spawn_agent_id = agent_id.clone(); let spawn_agent_name = agent_name.clone(); let spawn_abort = child_abort.clone(); @@ -1173,6 +1267,10 @@ async fn handle_spawn(ctx: &mut RequestContext, args: &Value) -> Result { }, }; let success = agent_result.exit_status == AgentExitStatus::Completed; + match &agent_result.exit_status { + AgentExitStatus::Completed => result_hooks.fire(None), + AgentExitStatus::Failed(e) => result_hooks.fire(Some(e)), + } spawn_notifications.push(agent_notification( &agent_result.id, &agent_result.agent_name, @@ -2154,6 +2252,155 @@ mod tests { .unwrap(); } + /// A graph agent whose graph.yaml wires `agent.*` hooks at its top + /// level. With `failing`, the start node references a script that does + /// not exist, so the child run fails at runtime — after the spawn seam + /// has already fired `agent.started`. + fn write_graph_agent_with_hooks(agent_name: &str, marker: &str, failing: bool) { + let agent_dir = paths::agent_data_dir(agent_name); + create_dir_all(&agent_dir).unwrap(); + let (start, nodes) = if failing { + ( + "work", + " work:\n type: script\n script: missing.sh\n next: done\n done:\n type: end\n output: done\n", + ) + } else { + ("done", " done:\n type: end\n output: done\n") + }; + write( + agent_dir.join("graph.yaml"), + format!( + "name: {agent_name}\n\ + start: {start}\n\ + settings:\n\ + \x20 validate_before_run: false\n\ + hooks:\n\ + \x20 agent.started:\n\ + \x20 - name: {marker}_started\n\ + \x20 command: \"true\"\n\ + \x20 agent.completed:\n\ + \x20 - name: {marker}_completed\n\ + \x20 command: \"true\"\n\ + \x20 agent.failed:\n\ + \x20 - name: {marker}_failed\n\ + \x20 command: \"true\"\n\ + nodes:\n\ + {nodes}" + ), + ) + .unwrap(); + } + + fn hook_captures_named(name: &str) -> Vec { + hooks::test_sink::snapshot() + .into_iter() + .filter(|capture| capture.hook_name == name) + .collect() + } + + /// Spawns an agent through the supervisor seam and awaits its spawn + /// task, returning the spawn id. The result arms fire their hooks + /// before that task returns, so the sink is complete afterwards. + fn spawn_and_collect(agent_name: &str) -> String { + run_async(async { + let mut ctx = ctx_with_supervisor(4, 3); + let spawned = handle_spawn(&mut ctx, &json!({"agent": agent_name, "prompt": "hi"})) + .await + .unwrap(); + assert_eq!(spawned["status"], "ok", "{spawned}"); + let id = spawned["id"].as_str().unwrap().to_string(); + let handle = ctx + .supervisor + .as_ref() + .unwrap() + .write() + .take(&id) + .expect("spawned agent must be registered"); + handle + .join_handle + .await + .expect("spawn task must not panic") + .expect("spawn task must return an AgentResult"); + id + }) + } + + #[test] + #[serial] + fn spawned_agent_success_fires_started_and_completed_with_matching_id() { + let _guard = TestConfigDirGuard::new(); + let _sink = hooks::test_sink::install(); + let agent_name = unique_agent_name("test_hook_spawn_ok"); + write_graph_agent_with_hooks(&agent_name, "t037_spawn_ok", false); + + let id = spawn_and_collect(&agent_name); + + let started = hook_captures_named("t037_spawn_ok_started"); + assert_eq!(started.len(), 1, "{started:?}"); + assert_eq!(started[0].envs["COYOTE_EVENT"], "agent.started"); + assert_eq!(started[0].envs["COYOTE_AGENT_ID"], id); + assert_eq!(started[0].envs["COYOTE_AGENT_NAME"], agent_name); + + let completed = hook_captures_named("t037_spawn_ok_completed"); + assert_eq!(completed.len(), 1, "{completed:?}"); + assert_eq!(completed[0].envs["COYOTE_EVENT"], "agent.completed"); + assert_eq!(completed[0].envs["COYOTE_AGENT_ID"], id); + assert_eq!(completed[0].envs["COYOTE_AGENT_NAME"], agent_name); + + assert!(hook_captures_named("t037_spawn_ok_failed").is_empty()); + } + + #[test] + #[serial] + fn spawned_agent_failure_fires_started_and_failed_with_matching_id() { + let _guard = TestConfigDirGuard::new(); + let _sink = hooks::test_sink::install(); + let agent_name = unique_agent_name("test_hook_spawn_fail"); + write_graph_agent_with_hooks(&agent_name, "t037_spawn_fail", true); + + let id = spawn_and_collect(&agent_name); + + let started = hook_captures_named("t037_spawn_fail_started"); + assert_eq!(started.len(), 1, "{started:?}"); + assert_eq!(started[0].envs["COYOTE_AGENT_ID"], id); + + let failed = hook_captures_named("t037_spawn_fail_failed"); + assert_eq!(failed.len(), 1, "{failed:?}"); + assert_eq!(failed[0].envs["COYOTE_EVENT"], "agent.failed"); + assert_eq!(failed[0].envs["COYOTE_AGENT_ID"], id); + assert!( + !failed[0].envs["COYOTE_AGENT_ERROR"].is_empty(), + "a failed spawn must carry its error" + ); + + assert!(hook_captures_named("t037_spawn_fail_completed").is_empty()); + } + + #[test] + fn hook_base_envs_carries_optional_names_only_when_present() { + let envs = hook_base_envs(HookEvent::AgentCompleted, None, None); + let keys: Vec<&str> = envs.iter().map(|(key, _)| key.as_str()).collect(); + assert!(keys.contains(&"COYOTE_EVENT")); + assert!(keys.contains(&"COYOTE_CONFIG_DIR")); + assert!(keys.contains(&"COYOTE_EVENT_TIMESTAMP")); + assert!(!keys.contains(&"COYOTE_SESSION_ID")); + assert!(!keys.contains(&"COYOTE_AGENT_NAME")); + assert!( + envs.iter() + .any(|(key, value)| key == "COYOTE_EVENT" && value == "agent.completed") + ); + + let envs = hook_base_envs(HookEvent::AgentFailed, Some("sess"), Some("bot")); + assert!( + envs.iter() + .any(|(key, value)| key == "COYOTE_SESSION_ID" && value == "sess") + ); + assert!( + envs.iter() + .any(|(key, value)| key == "COYOTE_AGENT_NAME" && value == "bot") + ); + } + fn run_agent_for_graph_error( agent_name: &str, graph_inputs: Option>, diff --git a/src/graph/agent.rs b/src/graph/agent.rs index 86ded61..e6b2e55 100644 --- a/src/graph/agent.rs +++ b/src/graph/agent.rs @@ -6,6 +6,7 @@ use super::types::AgentNode; use super::wall_clock; use crate::config::RequestContext; use crate::function::agents::run_agent_for_graph; +use crate::hooks::{self, HookEvent}; use crate::supervisor::mailbox::{Inbox, PeerRegistry}; use crate::utils::create_abort_signal; use anyhow::{Context, Error, Result, anyhow}; @@ -35,6 +36,14 @@ impl AgentNodeExecutor { parent_ctx: &mut RequestContext, retire_peer_on_return: bool, ) -> Result { + // The teammate identity doubles as the agent's id for observers; the + // run consumes it off the ctx, so it is captured up front. Nodes + // without one are identified by their node id. + let agent_hook_id = parent_ctx + .peer_assignment + .as_ref() + .map(|(id, _)| id.clone()) + .unwrap_or_else(|| node_id.to_string()); let result = run( node_id, node, @@ -43,6 +52,21 @@ impl AgentNodeExecutor { retire_peer_on_return, ) .await; + // Fired at the point of failure, before a declared fallback converts + // it into a successful route: observers must see the agent's failure + // even when the node itself continues. + if let Err(e) = &result { + hooks::fire( + HookEvent::AgentFailed, + parent_ctx, + &[ + ("COYOTE_AGENT_ID", agent_hook_id), + ("COYOTE_AGENT_NAME", node.agent.clone()), + ("COYOTE_AGENT_ERROR", format!("{e:#}")), + ], + None, + ); + } outcome_from(node_id, node, state_manager, result) } } diff --git a/src/graph/executor.rs b/src/graph/executor.rs index fe537d2..1f380af 100644 --- a/src/graph/executor.rs +++ b/src/graph/executor.rs @@ -1,6 +1,6 @@ use super::agent::{AgentExecutionOutcome, AgentNodeExecutor}; use super::llm::{LlmExecutionOutcome, LlmNodeExecutor}; -use super::logging::{GraphLogger, narrate_node_complete, narrate_node_failed}; +use super::logging::{GraphLogger, narrate_node_complete, narrate_node_failed, node_type_label}; use super::map::MapNodeExecutor; use super::rag::RagNodeExecutor; use super::script::ScriptExecutor; @@ -11,6 +11,8 @@ use super::user_interaction::{ApprovalNodeExecutor, InputNodeExecutor}; use super::validator::{AgentValidationContext, GraphValidator}; use super::wall_clock; use crate::config::{AgentVariable, AgentVariables, RenderMode, RequestContext}; +use crate::function::agents::hook_base_envs; +use crate::hooks::{self, HookEvent, ResolvedHook}; use crate::supervisor::mailbox::{Inbox, PeerAssignment, PeerRegistry, graph_agent_id}; use crate::utils::{AbortSignal, wait_abort_signal, wait_user_interrupt}; use anyhow::{Context, Result, anyhow, bail}; @@ -234,6 +236,16 @@ impl GraphExecutor { })? .clone(); logger.node_start(&node, in_super_step); + let node_hooks = NodeHookEmitter::resolve(ctx, node_id, node_type_label(&node)); + hooks::fire( + HookEvent::GraphNodeStarted, + ctx, + &[ + ("COYOTE_NODE_ID", node_id.clone()), + ("COYOTE_NODE_TYPE", node_hooks.node_type.to_string()), + ], + None, + ); let branch_state = state.fork_for_branch_state(); let mut branch_ctx = ctx.fork_for_branch(); let mut peer_id: Option = None; @@ -281,6 +293,7 @@ impl GraphExecutor { "aborted", in_super_step, ); + node_hooks.failed(Duration::default(), "aborted"); return ( current.clone(), branch_state, @@ -315,6 +328,7 @@ impl GraphExecutor { route.as_deref(), in_super_step, ); + node_hooks.completed(elapsed, route.as_deref()); } Ok(StepResult::End(_)) => { narrate_node_complete( @@ -324,6 +338,7 @@ impl GraphExecutor { Some("END"), in_super_step, ); + node_hooks.completed(elapsed, Some("END")); } Err(e) => { narrate_node_failed( @@ -333,6 +348,7 @@ impl GraphExecutor { &e.to_string(), in_super_step, ); + node_hooks.failed(elapsed, &e.to_string()); } } (current, state, result, elapsed) @@ -498,6 +514,73 @@ fn provision_frontier_peers( (Some(registry), assignments) } +/// Pre-resolved `graph.node.completed`/`graph.node.failed` snapshots plus +/// the metadata one branch task needs to fire them. Captured per branch +/// before the branch is spawned: every parallel super-step branch owns its +/// emitter, so node metadata never crosses between concurrent tasks and no +/// task borrows the context it was forked from. Base envs are rebuilt at +/// fire time so the timestamp reflects when the node actually finished. +struct NodeHookEmitter { + completed: Vec, + failed: Vec, + session_name: Option, + agent_name: Option, + node_id: String, + node_type: &'static str, +} + +impl NodeHookEmitter { + fn resolve(ctx: &RequestContext, node_id: &str, node_type: &'static str) -> Self { + Self { + completed: ctx.resolved_hooks(HookEvent::GraphNodeCompleted), + failed: ctx.resolved_hooks(HookEvent::GraphNodeFailed), + session_name: ctx + .session + .as_ref() + .map(|session| session.name().to_string()), + agent_name: ctx.agent.as_ref().map(|agent| agent.name().to_string()), + node_id: node_id.to_string(), + node_type, + } + } + + fn extras(&self, elapsed: Duration) -> Vec<(&'static str, String)> { + vec![ + ("COYOTE_NODE_ID", self.node_id.clone()), + ("COYOTE_NODE_TYPE", self.node_type.to_string()), + ("COYOTE_NODE_DURATION_MS", elapsed.as_millis().to_string()), + ] + } + + fn dispatch(&self, event: HookEvent, resolved: &[ResolvedHook], extras: &[(&str, String)]) { + hooks::fire_resolved( + event, + resolved.to_vec(), + hook_base_envs( + event, + self.session_name.as_deref(), + self.agent_name.as_deref(), + ), + extras, + None, + ); + } + + fn completed(&self, elapsed: Duration, route: Option<&str>) { + let mut extras = self.extras(elapsed); + if let Some(route) = route { + extras.push(("COYOTE_NODE_ROUTE", route.to_string())); + } + self.dispatch(HookEvent::GraphNodeCompleted, &self.completed, &extras); + } + + fn failed(&self, elapsed: Duration, error: &str) { + let mut extras = self.extras(elapsed); + extras.push(("COYOTE_NODE_ERROR", error.to_string())); + self.dispatch(HookEvent::GraphNodeFailed, &self.failed, &extras); + } +} + pub(super) struct TaskCancelGuard(Vec); impl TaskCancelGuard { @@ -1050,9 +1133,10 @@ mod tests { mod integration_tests { use super::*; use crate::config::paths; - use crate::config::{AppState, WorkingMode}; + use crate::config::{Agent, AgentConfig, AppConfig, AppState, WorkingMode}; #[cfg(unix)] use crate::function::jobs::RingBuf; + use crate::hooks::HookDef; #[cfg(unix)] use crate::supervisor::{JobHandle, JobResult, JobState, JobStatus, Supervisor, notification}; use crate::utils::{create_abort_signal, get_env_name, temp_file}; @@ -1068,6 +1152,33 @@ mod integration_tests { which::which(name).is_ok() } + fn hooks_config(entries: &[(&str, &str)]) -> crate::hooks::HooksMap { + let mut map = crate::hooks::HooksMap::new(); + for (event, name) in entries { + map.entry((*event).to_string()).or_default().push(HookDef { + name: (*name).to_string(), + command: "true".to_string(), + }); + } + map + } + + fn ctx_with_global_hooks(hooks: crate::hooks::HooksMap) -> RequestContext { + let mut app = AppState::test_default(); + app.config = Arc::new(AppConfig { + hooks, + ..Default::default() + }); + RequestContext::new(Arc::new(app), WorkingMode::Cmd) + } + + fn hook_captures_named(name: &str) -> Vec { + crate::hooks::test_sink::snapshot() + .into_iter() + .filter(|capture| capture.hook_name == name) + .collect() + } + struct TestWorkspace { dir: PathBuf, } @@ -1210,6 +1321,250 @@ nodes: assert_eq!(result, "held|held"); } + /// An agent node that fails but declares a fallback must surface the + /// failure to observers as `agent.failed`, fired at the point of + /// failure, even though the fallback then converts it into a normal + /// node completion that routes to the fallback target — so the same + /// node also yields `graph.node.completed`, never `graph.node.failed`. + #[tokio::test] + async fn agent_node_failure_with_fallback_fires_agent_failed_and_node_completed() { + let _sink = crate::hooks::test_sink::install(); + let mut ctx = ctx_with_global_hooks(hooks_config(&[ + ("agent.failed", "t037_fb_agent_failed"), + ("graph.node.completed", "t037_fb_node_completed"), + ("graph.node.failed", "t037_fb_node_failed"), + ])); + // Forces the agent node to fail before it ever loads the (absent) + // agent from disk, keeping the test hermetic. + ctx.current_depth = crate::config::default_max_agent_depth(); + + let yaml = r#" +name: t +settings: + validate_before_run: false +start: worker +nodes: + worker: + type: agent + agent: missing_agent + prompt: p + fallback: recover + next: done + recover: + type: end + output: recovered + done: + type: end + output: done +"#; + let graph: Graph = serde_yaml::from_str(yaml).unwrap(); + let abort = create_abort_signal(); + let result = GraphExecutor::new(graph, ".") + .execute(&mut ctx, abort) + .await + .unwrap_or_else(|e| panic!("executor failed: {e:#}")); + assert_eq!(result, "recovered"); + + let agent_failed = hook_captures_named("t037_fb_agent_failed"); + assert_eq!(agent_failed.len(), 1, "{agent_failed:?}"); + assert_eq!(agent_failed[0].envs["COYOTE_EVENT"], "agent.failed"); + assert_eq!(agent_failed[0].envs["COYOTE_AGENT_ID"], "worker"); + assert_eq!(agent_failed[0].envs["COYOTE_AGENT_NAME"], "missing_agent"); + assert!( + agent_failed[0].envs["COYOTE_AGENT_ERROR"].contains("Max agent depth exceeded"), + "{:?}", + agent_failed[0].envs + ); + + let node_completed = hook_captures_named("t037_fb_node_completed"); + let worker_completed: Vec<_> = node_completed + .iter() + .filter(|capture| capture.envs["COYOTE_NODE_ID"] == "worker") + .collect(); + assert_eq!(worker_completed.len(), 1, "{node_completed:?}"); + assert_eq!(worker_completed[0].envs["COYOTE_NODE_TYPE"], "agent"); + assert_eq!(worker_completed[0].envs["COYOTE_NODE_ROUTE"], "recover"); + + let node_failed = hook_captures_named("t037_fb_node_failed"); + assert!( + node_failed + .iter() + .all(|capture| capture.envs["COYOTE_NODE_ID"] != "worker"), + "a node that continues via fallback must not fire graph.node.failed: {node_failed:?}" + ); + } + + /// Two branches running in the same super-step each carry their own + /// node metadata, captured task-locally before the branch was spawned; + /// the distinct routes prove one branch's metadata never leaks into a + /// sibling's events. + #[tokio::test] + async fn parallel_super_step_branches_fire_per_node_metadata() { + if !cmd_available("bash") { + eprintln!("skipping: bash not available"); + return; + } + let _sink = crate::hooks::test_sink::install(); + let ws = TestWorkspace::new(); + for script in [ + "dispatcher.sh", + "worker_a.sh", + "worker_b.sh", + "sink_a.sh", + "sink_b.sh", + ] { + ws.write_script(script, "#!/bin/bash\necho '{}'\n"); + } + let mut ctx = ctx_with_global_hooks(hooks_config(&[ + ("graph.node.started", "t037_par_started"), + ("graph.node.completed", "t037_par_completed"), + ])); + + let yaml = r#" +name: t +start: dispatcher +nodes: + dispatcher: + type: script + script: dispatcher.sh + state_updates: {} + next: [worker_a, worker_b] + worker_a: + type: script + script: worker_a.sh + state_updates: {} + next: sink_a + worker_b: + type: script + script: worker_b.sh + state_updates: {} + next: sink_b + sink_a: + type: script + script: sink_a.sh + state_updates: {} + next: join + sink_b: + type: script + script: sink_b.sh + state_updates: {} + next: join + join: + type: end + output: done +"#; + let graph: Graph = serde_yaml::from_str(yaml).unwrap(); + let abort = create_abort_signal(); + let result = GraphExecutor::new(graph, &ws.dir) + .execute(&mut ctx, abort) + .await + .unwrap_or_else(|e| panic!("executor failed: {e:#}")); + assert_eq!(result, "done"); + + let started = hook_captures_named("t037_par_started"); + for node_id in [ + "dispatcher", + "worker_a", + "worker_b", + "sink_a", + "sink_b", + "join", + ] { + assert_eq!( + started + .iter() + .filter(|capture| capture.envs["COYOTE_NODE_ID"] == node_id) + .count(), + 1, + "exactly one graph.node.started for '{node_id}': {started:?}" + ); + } + + let completed = hook_captures_named("t037_par_completed"); + let completed_route = |node_id: &str| -> String { + let matches: Vec<_> = completed + .iter() + .filter(|capture| capture.envs["COYOTE_NODE_ID"] == node_id) + .collect(); + assert_eq!( + matches.len(), + 1, + "exactly one graph.node.completed for '{node_id}': {completed:?}" + ); + let expected_type = if node_id == "join" { "end" } else { "script" }; + assert_eq!(matches[0].envs["COYOTE_NODE_TYPE"], expected_type); + let duration = &matches[0].envs["COYOTE_NODE_DURATION_MS"]; + let _: u128 = duration + .parse() + .unwrap_or_else(|_| panic!("non-numeric duration for '{node_id}': {duration}")); + matches[0].envs["COYOTE_NODE_ROUTE"].clone() + }; + assert_eq!(completed_route("dispatcher"), "worker_a, worker_b"); + assert_eq!(completed_route("worker_a"), "sink_a"); + assert_eq!(completed_route("worker_b"), "sink_b"); + assert_eq!(completed_route("sink_a"), "join"); + assert_eq!(completed_route("sink_b"), "join"); + assert_eq!(completed_route("join"), "END"); + } + + /// A graph agent's hooks come from graph.yaml's top level: they resolve + /// through the agent scope end-to-end, while a global hook on the same + /// event stays gated off because the agent whitelisted nothing. + #[tokio::test] + async fn graph_agent_hooks_resolve_from_graph_yaml_top_level() { + let _sink = crate::hooks::test_sink::install(); + let mut ctx = + ctx_with_global_hooks(hooks_config(&[("graph.node.completed", "t037_gy_global")])); + + let yaml = r#" +name: hooked_graph +settings: + validate_before_run: false +start: done +hooks: + graph.node.started: + - name: t037_gy_started + command: "true" + graph.node.completed: + - name: t037_gy_completed + command: "true" +nodes: + done: + type: end + output: ok +"#; + let graph: Graph = serde_yaml::from_str(yaml).unwrap(); + ctx.agent = Some(Agent::test_new(AgentConfig::from_graph( + "hooked_graph", + &graph, + ))); + + let abort = create_abort_signal(); + let result = GraphExecutor::new(graph, ".") + .execute(&mut ctx, abort) + .await + .unwrap_or_else(|e| panic!("executor failed: {e:#}")); + assert_eq!(result, "ok"); + + let started = hook_captures_named("t037_gy_started"); + assert_eq!(started.len(), 1, "{started:?}"); + assert_eq!(started[0].envs["COYOTE_EVENT"], "graph.node.started"); + assert_eq!(started[0].envs["COYOTE_NODE_ID"], "done"); + assert_eq!(started[0].envs["COYOTE_NODE_TYPE"], "end"); + assert_eq!(started[0].envs["COYOTE_AGENT_NAME"], "hooked_graph"); + + let completed = hook_captures_named("t037_gy_completed"); + assert_eq!(completed.len(), 1, "{completed:?}"); + assert_eq!(completed[0].envs["COYOTE_NODE_ID"], "done"); + assert_eq!(completed[0].envs["COYOTE_NODE_ROUTE"], "END"); + assert_eq!(completed[0].envs["COYOTE_AGENT_NAME"], "hooked_graph"); + + assert!( + hook_captures_named("t037_gy_global").is_empty(), + "an agent context must not run global hooks it did not whitelist" + ); + } + /// Declared variables are seeded into state at run start, so the agent /// node's strict prompt interpolation of `{{foo}}` succeeds instead of /// bailing with "not found in state". `ctx.agent` is None here, so each