feat(TASK-049): add escalation.raised/answered, session.resumed, and agent.interrupted hook events

- escalation.raised fires in the child ctx after queue submit; escalation.answered
  fires in the parent ctx from the taken request, so both agree on origin envs
- escalation.* resolves global hooks ungated: the per-agent global_hooks whitelist
  never applies to these root-level events
- session.resumed fires on the use_session resume leg, inverse of session.started
- agent.interrupted replaces agent.completed/failed when the abort signal was
  ctrl-c'd: top-level bracket (now abort-aware), spawned-subagent result arm
  (agent__cancel / REPL ctrl-c), and graph agent nodes via the session signal;
  exit_agent and the shell-exit path stay non-interrupted by passing no signal
- document the deliberate non-fire paths (direct prompts, escalation timeout and
  parent-drop, graph approval nodes)

src/function/agents.rs

@@ -997,14 +997,15 @@ fn sync_agent_functions_to_ctx(ctx: &mut RequestContext) -> Result<()> {
     Ok(())
 }
 
-/// 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.
+/// Pre-resolved `agent.completed`/`agent.interrupted`/`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<ResolvedHook>,
     failed: Vec<ResolvedHook>,
+    interrupted: Vec<ResolvedHook>,
     session_name: Option<String>,
     agent_id: String,
     agent_name: String,
@@ -1015,6 +1016,7 @@ impl SpawnResultHooks {
         Self {
             completed: child_ctx.resolved_hooks(HookEvent::AgentCompleted),
             failed: child_ctx.resolved_hooks(HookEvent::AgentFailed),
+            interrupted: child_ctx.resolved_hooks(HookEvent::AgentInterrupted),
             session_name: child_ctx
                 .session
                 .as_ref()
@@ -1024,14 +1026,20 @@ impl SpawnResultHooks {
         }
     }
 
-    /// Fires `agent.completed`, or `agent.failed` when `error` is given.
-    fn fire(self, error: Option<&str>) {
+    /// Fires `agent.completed`, or — for a failed child — `agent.interrupted`
+    /// when its abort signal was ctrl-c'd (REPL ctrl-c or `agent__cancel`),
+    /// else `agent.failed`. The signal, never the error text, decides:
+    /// cancellation is not a failure, so no error env rides along.
+    fn fire(self, error: Option<&str>, child_abort: &AbortSignal) {
         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(_) if child_abort.aborted_ctrlc() => {
+                (HookEvent::AgentInterrupted, self.interrupted)
+            }
             Some(error) => {
                 extras.push(("COYOTE_AGENT_ERROR", error.to_string()));
                 (HookEvent::AgentFailed, self.failed)
@@ -1216,6 +1224,7 @@ async fn handle_spawn(ctx: &mut RequestContext, args: &Value) -> Result<Value> {
     let spawn_agent_id = agent_id.clone();
     let spawn_agent_name = agent_name.clone();
     let spawn_abort = child_abort.clone();
+    let hook_abort = child_abort.clone();
     let spawn_notifications = Arc::clone(&ctx.notification_queue);
     let child_supervisor = child_ctx.supervisor.clone();
 
@@ -1238,8 +1247,8 @@ async fn handle_spawn(ctx: &mut RequestContext, args: &Value) -> Result<Value> {
         };
         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)),
+            AgentExitStatus::Completed => result_hooks.fire(None, &hook_abort),
+            AgentExitStatus::Failed(e) => result_hooks.fire(Some(e), &hook_abort),
         }
         spawn_notifications.push(agent_notification(
             &agent_result.id,
@@ -1771,6 +1780,21 @@ fn handle_reply_escalation(ctx: &mut RequestContext, args: &Value) -> Result<Val
             let from_agent = request.from_agent_name.clone();
             let question = request.question.clone();
             let _ = request.reply_tx.send(reply.to_string());
+            // Origin envs come from the taken request, not this (replying)
+            // context, so `escalation.answered` and `escalation.raised`
+            // describe the same agent for one escalation id.
+            hooks::fire(
+                HookEvent::EscalationAnswered,
+                ctx,
+                &[
+                    ("COYOTE_ESCALATION_ID", escalation_id.to_string()),
+                    ("COYOTE_ESCALATION_FROM_AGENT_ID", request.from_agent_id),
+                    ("COYOTE_ESCALATION_FROM_AGENT_NAME", from_agent.clone()),
+                    ("COYOTE_ESCALATION_QUESTION", question.clone()),
+                    ("COYOTE_ESCALATION_REPLY", reply.to_string()),
+                ],
+                None,
+            );
             Ok(json!({
                 "status": "ok",
                 "message": format!("Reply sent to agent '{from_agent}' for escalation '{escalation_id}'"),
@@ -2251,6 +2275,9 @@ mod tests {
                  \x20 agent.completed:\n\
                  \x20   - name: {marker}_completed\n\
                  \x20     command: \"true\"\n\
+                 \x20 agent.interrupted:\n\
+                 \x20   - name: {marker}_interrupted\n\
+                 \x20     command: \"true\"\n\
                  \x20 agent.failed:\n\
                  \x20   - name: {marker}_failed\n\
                  \x20     command: \"true\"\n\
@@ -2346,6 +2373,54 @@ mod tests {
         assert!(hook_captures_named("t037_spawn_fail_completed").is_empty());
     }
 
+    #[test]
+    #[serial]
+    fn cancelled_spawned_agent_fires_interrupted_not_failed() {
+        let _guard = TestConfigDirGuard::new();
+        let _sink = hooks::test_sink::install();
+        let agent_name = unique_agent_name("test_hook_spawn_cancel");
+        write_graph_agent_with_hooks(&agent_name, "t049_spawn_cancel", true);
+
+        let id = 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");
+            // The `agent__cancel` signal, fired before the spawn task runs:
+            // on this current-thread runtime the task only progresses once
+            // awaited, so its failure arm observes the ctrl-c deterministically.
+            handle.abort_signal.set_ctrlc();
+            handle
+                .join_handle
+                .await
+                .expect("spawn task must not panic")
+                .expect("spawn task must return an AgentResult");
+            id
+        });
+
+        let interrupted = hook_captures_named("t049_spawn_cancel_interrupted");
+        assert_eq!(interrupted.len(), 1, "{interrupted:?}");
+        assert_eq!(interrupted[0].envs["COYOTE_EVENT"], "agent.interrupted");
+        assert_eq!(interrupted[0].envs["COYOTE_AGENT_ID"], id);
+        assert_eq!(interrupted[0].envs["COYOTE_AGENT_NAME"], agent_name);
+        assert!(
+            !interrupted[0].envs.contains_key("COYOTE_AGENT_ERROR"),
+            "an interruption is not a failure and carries no error env"
+        );
+
+        assert!(hook_captures_named("t049_spawn_cancel_failed").is_empty());
+        assert!(hook_captures_named("t049_spawn_cancel_completed").is_empty());
+    }
+
     #[test]
     fn agent_event_base_envs_carry_optional_names_only_when_present() {
         let envs = hooks::base_envs_parts(HookEvent::AgentCompleted, None, None);

src/function/user_interaction.rs

@@ -1,5 +1,6 @@
 use super::{FunctionDeclaration, JsonSchema};
 use crate::config::RequestContext;
+use crate::hooks::{self, HookEvent};
 use crate::supervisor::escalation::{EscalationRequest, new_escalation_id};
 use crate::utils::{ACP_SERVER, HEADLESS, queue_acp_permission};
 
@@ -179,6 +180,8 @@ fn handle_headless(action: &str, args: &Value) -> Value {
     })
 }
 
+// The root agent prompting its own user is a direct interaction, not an
+// escalation: no escalation event fires on this path.
 fn handle_direct(action: &str, args: &Value) -> Result<Value> {
     match action {
         "select" => handle_direct_ask(args),
@@ -278,22 +281,37 @@ async fn handle_escalated(ctx: &RequestContext, action: &str, args: &Value) -> R
     let escalation_id = new_escalation_id();
     let (tx, rx) = oneshot::channel();
 
+    let question = format!("[{action}] {question}");
     let request = EscalationRequest {
         id: escalation_id.clone(),
-        from_agent_id,
+        from_agent_id: from_agent_id.clone(),
         from_agent_name: from_agent_name.clone(),
-        question: format!("[{action}] {question}"),
+        question: question.clone(),
         options,
         reply_tx: tx,
     };
 
     root_queue.submit(request);
 
+    hooks::fire(
+        HookEvent::EscalationRaised,
+        ctx,
+        &[
+            ("COYOTE_ESCALATION_ID", escalation_id),
+            ("COYOTE_ESCALATION_FROM_AGENT_ID", from_agent_id),
+            ("COYOTE_ESCALATION_FROM_AGENT_NAME", from_agent_name),
+            ("COYOTE_ESCALATION_QUESTION", question),
+        ],
+        None,
+    );
+
     await_escalation_reply(rx, timeout_secs).await
 }
 
 /// Waits for the parent's reply. `timeout_secs == 0` waits indefinitely; a
 /// dropped sender (the parent cancelled the request) resolves either way.
+/// Both fallback arms resolve without touching the escalation queue, so
+/// neither fires an escalation event: the request was never answered.
 async fn await_escalation_reply(rx: oneshot::Receiver<String>, timeout_secs: u64) -> Result<Value> {
     let reply = if timeout_secs == 0 {
         rx.await
@@ -346,8 +364,14 @@ fn parse_options(args: &Value) -> Result<Vec<String>> {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::config::{AppConfig, AppState, WorkingMode};
+    use crate::function::agents::handle_agent_tool;
+    use crate::hooks::{HookDef, HooksMap, test_sink};
+    use crate::supervisor::escalation::EscalationQueue;
+    use serial_test::serial;
     use std::fs;
     use std::path::{Path, PathBuf};
+    use std::sync::Arc;
 
     #[test]
     fn headless_select_returns_structured_json() {
@@ -374,6 +398,112 @@ mod tests {
         assert_eq!(DEFAULT_ESCALATION_TIMEOUT_SECS, 0);
     }
 
+    fn ctx_with_escalation_hooks(marker: &str) -> RequestContext {
+        let mut hooks = HooksMap::default();
+        for event in ["escalation.raised", "escalation.answered"] {
+            hooks.insert(
+                event.to_string(),
+                vec![HookDef {
+                    name: format!("{marker}-{event}"),
+                    command: "true".to_string(),
+                }],
+            );
+        }
+        let mut app = AppState::test_default();
+        app.config = Arc::new(AppConfig {
+            hooks,
+            ..Default::default()
+        });
+        RequestContext::new(Arc::new(app), WorkingMode::Cmd)
+    }
+
+    fn marker_captures(marker: &str, event: &str) -> Vec<test_sink::Capture> {
+        test_sink::snapshot()
+            .into_iter()
+            .filter(|capture| capture.hook_name == format!("{marker}-{event}"))
+            .collect()
+    }
+
+    /// `escalation.raised` (child seam) and `escalation.answered` (parent
+    /// seam) must describe the same origin agent for one escalation id: the
+    /// answered envs come from the taken queue request, never from the
+    /// replying context.
+    #[tokio::test]
+    #[serial]
+    async fn escalation_raised_and_answered_agree_on_origin_envs() {
+        let _sink = test_sink::install();
+        let marker = "esc-parity-k4w";
+
+        let queue = Arc::new(EscalationQueue::new());
+        let mut child_ctx = ctx_with_escalation_hooks(marker);
+        child_ctx.escalation_queue = Some(Arc::clone(&queue));
+        child_ctx.self_agent_id = Some("esc-child-1".to_string());
+        let mut parent_ctx = ctx_with_escalation_hooks(marker);
+        parent_ctx.escalation_queue = Some(Arc::clone(&queue));
+
+        let raise_args = json!({ "question": "which db?" });
+        let raise = handle_escalated(&child_ctx, "input", &raise_args);
+        let reply = async {
+            // Bounded poll: on the current-thread runtime the raise future
+            // submits before its first await, so this resolves immediately.
+            let mut escalation_id = None;
+            for _ in 0..1000 {
+                if let Some(entry) = queue.pending_summary().first() {
+                    escalation_id = entry["escalation_id"].as_str().map(str::to_string);
+                    break;
+                }
+                tokio::task::yield_now().await;
+            }
+            let escalation_id = escalation_id.expect("escalation must reach the queue");
+            handle_agent_tool(
+                &mut parent_ctx,
+                "agent__reply_escalation",
+                &json!({ "escalation_id": escalation_id, "reply": "postgres" }),
+            )
+            .await
+            .unwrap()
+        };
+        let (raised_result, replied) = tokio::join!(raise, reply);
+        assert_eq!(raised_result.unwrap()["answer"], "postgres");
+        assert_eq!(replied["status"], "ok");
+
+        let raised = marker_captures(marker, "escalation.raised");
+        let answered = marker_captures(marker, "escalation.answered");
+        assert_eq!(raised.len(), 1, "{raised:?}");
+        assert_eq!(answered.len(), 1, "{answered:?}");
+        for key in [
+            "COYOTE_ESCALATION_ID",
+            "COYOTE_ESCALATION_FROM_AGENT_ID",
+            "COYOTE_ESCALATION_FROM_AGENT_NAME",
+            "COYOTE_ESCALATION_QUESTION",
+        ] {
+            assert!(raised[0].envs.contains_key(key), "{key} missing on raised");
+            assert_eq!(raised[0].envs.get(key), answered[0].envs.get(key), "{key}");
+        }
+        assert_eq!(
+            raised[0]
+                .envs
+                .get("COYOTE_ESCALATION_FROM_AGENT_ID")
+                .map(String::as_str),
+            Some("esc-child-1")
+        );
+        assert_eq!(
+            raised[0]
+                .envs
+                .get("COYOTE_ESCALATION_QUESTION")
+                .map(String::as_str),
+            Some("[input] which db?")
+        );
+        assert_eq!(
+            answered[0]
+                .envs
+                .get("COYOTE_ESCALATION_REPLY")
+                .map(String::as_str),
+            Some("postgres")
+        );
+        assert!(!raised[0].envs.contains_key("COYOTE_ESCALATION_REPLY"));
+    }
+
     #[tokio::test(start_paused = true)]
     async fn zero_timeout_waits_indefinitely_for_the_reply() {
         let (tx, rx) = oneshot::channel();