feat(TASK-049): classify idle-prompt ctrl-c + ctrl-d as a completed exit

Ctrl-C at the reedline prompt no longer latches the abort signal
unconditionally: latch_prompt_interrupt only marks an interruption when
the supervisor had live agents/jobs to cancel (new
Supervisor::has_active_tasks), so the advertised exit path (ctrl-c at an
idle prompt, then ctrl-d) fires agent.completed at teardown instead of
agent.interrupted. Mid-turn ctrl-c still latches directly via
wait_user_interrupt/poll_abort_signal, preserving the
interrupted-then-exit classification.

Also: probe-lane tests for escalation.answered single-fire,
escalation question env truncation, agent.failed inverse gate, and
hooks probe Scenario I (session.started vs session.resumed); comment
notes on escalation.answered semantics and the agent__cancel blind
spot.

scripts/usage-probe-hooks.sh

@@ -25,6 +25,10 @@
 #      path refused) and notify.sh's no-notifier/no-tty fallback strips
 #      ANSI/OSC control bytes; the hooks-dir .builtin-manifest is never
 #      installed executable.
+#   I) `--session <name>` discriminates the session events end-to-end: a
+#      fresh named session fires exactly one session.started (and no
+#      session.resumed); resuming a persisted session fires exactly one
+#      session.resumed (and no session.started).
 #
 # Not covered here (see src/function/mod.rs tests for tool.* + payload-file
 # coverage instead): a live tool.started firing requires a real LLM round
@@ -458,5 +462,38 @@ grep -Fq '[coyote] turn.completed tool=evil]0;own[31mred' "$NOTIFY_OUT" \
   || fail "ANSI/OSC escape bytes leaked through the notify.sh tty fallback"
 echo "PASS: installed hook scripts honor XDG default, 0600, UTC timestamps, symlink refusal, secret filter, and ANSI-stripped notify fallback"
 
+echo "== Scenario I: session.started vs session.resumed via --session =="
+CFG_I="$WORKDIR/i"
+mkdir -p "$CFG_I"
+SESS_LOG="$WORKDIR/i-sess.log"
+write_dryrun_config "$CFG_I" "hooks:" \
+  "  session.started:" \
+  "    - name: probe-sess" \
+  "      command: \"echo STARTED >> $SESS_LOG\"" \
+  "  session.resumed:" \
+  "    - name: probe-sess" \
+  "      command: \"echo RESUMED >> $SESS_LOG\""
+
+# A fresh named session (no file on disk yet) fires session.started only.
+COYOTE_CONFIG_DIR="$CFG_I" "$BIN" --session probe-fresh --no-stream "say exactly: hi" > /dev/null
+wait_for "session.started log" test -s "$SESS_LOG"
+[ "$(grep -c '^STARTED$' "$SESS_LOG" 2>/dev/null || echo 0)" = "1" ] \
+  || fail "expected exactly one session.started for a fresh session, got: $(cat "$SESS_LOG")"
+grep -q '^RESUMED$' "$SESS_LOG" \
+  && fail "a fresh named session must not fire session.resumed: $(cat "$SESS_LOG")"
+
+# A persisted session on disk: resuming it fires session.resumed, and never
+# session.started (the inverse gate).
+mkdir -p "$CFG_I/sessions"
+printf 'model: dryrun:dry-model\nmessages: []\n' > "$CFG_I/sessions/probe-resume.yaml"
+COYOTE_CONFIG_DIR="$CFG_I" "$BIN" --session probe-resume --no-stream "say exactly: hi" > /dev/null
+scenario_i_resumed() { grep -q '^RESUMED$' "$SESS_LOG"; }
+wait_for "session.resumed log" scenario_i_resumed
+[ "$(grep -c '^RESUMED$' "$SESS_LOG" 2>/dev/null || echo 0)" = "1" ] \
+  || fail "expected exactly one session.resumed for a resumed session, got: $(cat "$SESS_LOG")"
+[ "$(grep -c '^STARTED$' "$SESS_LOG" 2>/dev/null || echo 0)" = "1" ] \
+  || fail "resuming a session must not fire session.started: $(cat "$SESS_LOG")"
+echo "PASS: fresh session fires session.started only; resume fires session.resumed only"
+
 echo
 echo "ALL SCENARIOS PASSED"

src/config/request_context.rs

@@ -8619,6 +8619,50 @@ mod tests {
         }));
     }
 
+    /// The REPL's advertised exit path -- ctrl-c at an idle prompt (which
+    /// prints "To exit, press Ctrl+D"), then ctrl-d -- is a deliberate exit:
+    /// with no active tasks to cancel the prompt latch stays clear, so
+    /// teardown fires agent.completed, never agent.interrupted.
+    #[test]
+    #[serial]
+    fn idle_prompt_ctrlc_then_ctrld_exit_fires_agent_completed() {
+        let _guard = TestConfigDirGuard::new();
+        let _sink = test_sink::install();
+        let marker = "agent-top-idle-cc-w2b";
+        let agent_name = unique_name("hook_agent");
+        seed_agent_with_hooks(&agent_name, marker);
+
+        let mut ctx = create_test_ctx();
+        let app = ctx.app.config.clone();
+        let abort = utils::create_abort_signal();
+        run_async(ctx.use_agent(&app, &agent_name, None, abort.clone())).unwrap();
+        ctx.top_level_agent_started();
+
+        // Ctrl-c at the idle prompt: nothing to cancel, so no latch.
+        let supervisor = Arc::new(RwLock::new(Supervisor::new(4, 3)));
+        crate::repl::latch_prompt_interrupt(&abort, &supervisor);
+        assert!(!abort.aborted_ctrlc());
+
+        // Ctrl-d exits; teardown sees only the ctrl-d flag.
+        abort.set_ctrld();
+        ctx.top_level_agent_finished(None, Some(&abort));
+
+        let captures = marker_captures(marker);
+        assert_eq!(
+            captures
+                .iter()
+                .filter(|capture| capture.hook_name == format!("{marker}-agent.completed"))
+                .count(),
+            1
+        );
+        assert!(
+            captures
+                .iter()
+                .all(|capture| capture.hook_name != format!("{marker}-agent.interrupted")),
+            "a deliberate exit is never an interruption"
+        );
+    }
+
     #[test]
     #[serial]
     fn top_level_agent_helpers_noop_without_agent() {

src/function/agents.rs

@@ -1782,7 +1782,11 @@ fn handle_reply_escalation(ctx: &mut RequestContext, args: &Value) -> Result<Val
             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.
+            // describe the same agent for one escalation id. "Answered"
+            // means a parent submitted an answer, not that the asker
+            // received it: the reply_tx.send failure is ignored by design,
+            // so the event also fires for an escalation whose asker already
+            // timed out or was cancelled.
             hooks::fire(
                 HookEvent::EscalationAnswered,
                 ctx,
@@ -3399,6 +3403,79 @@ mod tests {
         assert!(result.is_err());
     }
 
+    /// `escalation.answered` belongs to the successful take -> reply arm
+    /// alone: exactly once per escalation, and never from the failure arms
+    /// (unknown id, already-taken id).
+    #[test]
+    #[serial]
+    fn reply_escalation_fires_answered_once_and_never_on_failure_arms() {
+        let _sink = hooks::test_sink::install();
+        let marker = "t049_answered_once";
+        let mut hooks_map = crate::hooks::HooksMap::default();
+        hooks_map.insert(
+            "escalation.answered".to_string(),
+            vec![crate::hooks::HookDef {
+                name: marker.to_string(),
+                command: "true".to_string(),
+            }],
+        );
+        let mut app = AppState::test_default();
+        app.config = Arc::new(AppConfig {
+            hooks: hooks_map,
+            ..Default::default()
+        });
+        let mut ctx = RequestContext::new(Arc::new(app), WorkingMode::Cmd);
+        let queue = Arc::new(EscalationQueue::new());
+        let (tx, rx) = tokio::sync::oneshot::channel();
+        queue.submit(EscalationRequest {
+            id: "esc_once".into(),
+            from_agent_id: "a1".into(),
+            from_agent_name: "explore".into(),
+            question: "[input] What do?".into(),
+            options: None,
+            reply_tx: tx,
+        });
+        ctx.escalation_queue = Some(queue);
+
+        // Unknown id: an error result, and no event.
+        let miss = handle_reply_escalation(
+            &mut ctx,
+            &json!({"escalation_id": "missing", "reply": "nope"}),
+        )
+        .unwrap();
+        assert_eq!(miss["status"], "error");
+        assert!(
+            hook_captures_named(marker).is_empty(),
+            "a failed reply must not fire escalation.answered"
+        );
+
+        // The successful take -> reply arm fires it exactly once.
+        let ok = handle_reply_escalation(
+            &mut ctx,
+            &json!({"escalation_id": "esc_once", "reply": "do X"}),
+        )
+        .unwrap();
+        assert_eq!(ok["status"], "ok");
+        assert_eq!(rx.blocking_recv().unwrap(), "do X");
+        let answered = hook_captures_named(marker);
+        assert_eq!(answered.len(), 1, "{answered:?}");
+        assert_eq!(answered[0].envs["COYOTE_ESCALATION_ID"], "esc_once");
+        assert_eq!(answered[0].envs["COYOTE_ESCALATION_REPLY"], "do X");
+
+        // Replying again to the same id hits the already-taken arm: still one.
+        let again = handle_reply_escalation(
+            &mut ctx,
+            &json!({"escalation_id": "esc_once", "reply": "again"}),
+        )
+        .unwrap();
+        assert_eq!(again["status"], "error");
+        assert_eq!(
+            hook_captures_named(marker).len(),
+            1,
+            "a second reply must not re-fire escalation.answered"
+        );
+    }
+
     #[test]
     fn handle_task_create_simple() {
         let mut ctx = ctx_with_supervisor(4, 3);

src/function/user_interaction.rs

@@ -504,6 +504,54 @@ mod tests {
         assert!(!raised[0].envs.contains_key("COYOTE_ESCALATION_REPLY"));
     }
 
+    /// `COYOTE_ESCALATION_QUESTION` rides through the standard env
+    /// truncation: a huge question is trimmed to 2048 bytes on the raised
+    /// seam, never passed through whole.
+    #[tokio::test]
+    #[serial]
+    async fn escalation_question_env_is_truncated_to_2048_bytes() {
+        let _sink = test_sink::install();
+        let marker = "esc-trunc-p9j";
+
+        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-trunc".to_string());
+
+        let raise_args = json!({ "question": "q".repeat(3000) });
+        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 request = None;
+            for _ in 0..1000 {
+                if let Some(entry) = queue.pending_summary().first() {
+                    let id = entry["escalation_id"].as_str().unwrap().to_string();
+                    request = queue.take(&id);
+                    break;
+                }
+                tokio::task::yield_now().await;
+            }
+            let request = request.expect("escalation must reach the queue");
+            let _ = request.reply_tx.send("ok".to_string());
+        };
+        let (raised_result, ()) = tokio::join!(raise, reply);
+        assert_eq!(raised_result.unwrap()["answer"], "ok");
+
+        let raised = marker_captures(marker, "escalation.raised");
+        assert_eq!(raised.len(), 1, "{raised:?}");
+        let question = raised[0]
+            .envs
+            .get("COYOTE_ESCALATION_QUESTION")
+            .expect("QUESTION env must be present on escalation.raised");
+        assert_eq!(
+            question.len(),
+            2048,
+            "QUESTION env must be 2048-byte truncated"
+        );
+        assert!(question.starts_with("[input] qqq"));
+    }
+
     #[tokio::test(start_paused = true)]
     async fn zero_timeout_waits_indefinitely_for_the_reply() {
         let (tx, rx) = oneshot::channel();

src/graph/agent.rs

@@ -53,7 +53,11 @@ impl AgentNodeExecutor {
             // The child's own abort signal dies inside `run`; the session
             // signal is what a user interrupt fires (the executor bridges it
             // onto the graph abort), so it is the only honest
-            // interrupted-vs-failed discriminator at this seam.
+            // interrupted-vs-failed discriminator at this seam. Blind spot:
+            // agent__cancel sets the child handle's signal, which never
+            // reaches parent_ctx.session_abort, so a node failing during
+            // such a cancellation stays agent.failed while the outer bracket
+            // reports agent.interrupted (narrow window, accepted).
             if parent_ctx
                 .session_abort
                 .as_ref()
@@ -496,6 +500,37 @@ mod tests {
         );
     }
 
+    /// The inverse gate: a failure whose session signal never fired stays
+    /// `agent.failed` with its error env — the presence of a signal alone
+    /// must not reclassify a real failure as an interruption.
+    #[tokio::test]
+    #[serial]
+    async fn execute_failure_without_fired_session_abort_fires_agent_failed() {
+        let _sink = test_sink::install();
+        let marker = "graph-agent-fail-inv-x7q";
+        let mut ctx = ctx_with_agent_hooks(marker);
+        // Past max depth the node fails before touching agent config on disk.
+        ctx.current_depth = default_max_agent_depth();
+        ctx.session_abort = Some(create_abort_signal());
+        let node = node_with("hi", None);
+        let mut state = manager_with(&[]);
+
+        AgentNodeExecutor::execute("test_node", &node, &mut state, &mut ctx, false)
+            .await
+            .expect_err("agent past max depth should fail");
+
+        let captures: Vec<_> = test_sink::snapshot()
+            .into_iter()
+            .filter(|capture| capture.hook_name.starts_with(marker))
+            .collect();
+        assert_eq!(captures.len(), 1, "{captures:?}");
+        assert_eq!(captures[0].hook_name, format!("{marker}-agent.failed"));
+        assert!(
+            captures[0].envs.contains_key("COYOTE_AGENT_ERROR"),
+            "a real failure keeps its error env"
+        );
+    }
+
     fn retryable_node(max_attempts: u32) -> AgentNode {
         let mut node = node_with("hi", None);
         node.max_attempts = max_attempts;

src/repl/mod.rs

@@ -19,6 +19,7 @@ use crate::config::{AssetCategory, paths};
 use crate::function::agents::{GuardrailAction, check_pending_tasks_guardrail};
 use crate::hooks::{self, HookEvent};
 use crate::render::render_error;
+use crate::supervisor::Supervisor;
 use crate::utils::{
     AbortSignal, SHELL, abortable_run_with_spinner, create_abort_signal, dimmed_text,
     drain_stale_tty_input, run_command, set_text, temp_file,
@@ -512,9 +513,8 @@ Type ".help" for additional help.
                     }
                 }
                 Ok(Signal::CtrlC) => {
-                    self.abort_signal.set_ctrlc();
                     if let Some(supervisor) = self.ctx.read().supervisor.clone() {
-                        supervisor.read().cancel_recursive();
+                        latch_prompt_interrupt(&self.abort_signal, &supervisor);
                     }
                     println!("(To exit, press Ctrl+D or enter \".exit\")\n");
                 }
@@ -1715,6 +1715,21 @@ fn reset_continuation(ctx: &mut RequestContext) {
     ctx.reset_continuation_count();
 }
 
+/// Ctrl-C at the prompt: cancel background work, but only latch the abort
+/// signal when there was live work to cancel. An idle-prompt ctrl-c is a
+/// step on the advertised exit path ("To exit, press Ctrl+D"), not an
+/// interruption, so teardown after the following ctrl-d classifies the run
+/// as completed. Mid-turn ctrl-c never passes through here -- it latches
+/// the signal directly (wait_user_interrupt / poll_abort_signal), so an
+/// interrupted turn followed by ctrl-d still reports agent.interrupted.
+pub(crate) fn latch_prompt_interrupt(abort_signal: &AbortSignal, supervisor: &RwLock<Supervisor>) {
+    let supervisor = supervisor.read();
+    if supervisor.has_active_tasks() {
+        abort_signal.set_ctrlc();
+    }
+    supervisor.cancel_recursive();
+}
+
 fn pause_banner_text(ctx: &RequestContext) -> Option<String> {
     let reason = ctx.auto_continue_paused.as_deref()?;
     if !ctx.todo_list.has_incomplete() {
@@ -2898,4 +2913,63 @@ mod tests {
 
         assert_eq!(pause_banner_text(&ctx), None);
     }
+
+    #[test]
+    fn latch_prompt_interrupt_stays_clear_at_idle_prompt() {
+        let abort = create_abort_signal();
+        let supervisor = RwLock::new(Supervisor::new(4, 3));
+
+        latch_prompt_interrupt(&abort, &supervisor);
+
+        assert!(
+            !abort.aborted_ctrlc(),
+            "ctrl-c with nothing to cancel must not latch as an interruption"
+        );
+    }
+
+    #[test]
+    fn latch_prompt_interrupt_latches_when_cancelling_active_children() {
+        use crate::supervisor::mailbox::Inbox;
+        use crate::supervisor::{AgentExitStatus, AgentHandle, AgentResult};
+
+        // Keep the runtime alive so the spawned task is never polled and the
+        // child counts as running.
+        let rt = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .unwrap();
+        let join_handle = rt.spawn(async {
+            Ok(AgentResult {
+                id: "done".into(),
+                agent_name: "test".into(),
+                output: "result".into(),
+                exit_status: AgentExitStatus::Completed,
+            })
+        });
+        let child_signal = create_abort_signal();
+        let mut sup = Supervisor::new(4, 3);
+        sup.register(AgentHandle {
+            id: "a1".to_string(),
+            agent_name: "explore".to_string(),
+            depth: 1,
+            inbox: Arc::new(Inbox::new()),
+            abort_signal: child_signal.clone(),
+            join_handle,
+            child_supervisor: None,
+        })
+        .unwrap();
+        let supervisor = RwLock::new(sup);
+        let abort = create_abort_signal();
+
+        latch_prompt_interrupt(&abort, &supervisor);
+
+        assert!(
+            abort.aborted_ctrlc(),
+            "cancelling live work is an interruption"
+        );
+        assert!(
+            child_signal.aborted_ctrlc(),
+            "the child must still be cancelled"
+        );
+    }
 }

src/supervisor/mod.rs

@@ -190,6 +190,12 @@ impl Supervisor {
             .count()
     }
 
+    /// True when any registered agent or job is still running -- i.e. a
+    /// cancellation right now would actually interrupt in-flight work.
+    pub fn has_active_tasks(&self) -> bool {
+        self.effective_active_count() > 0 || self.active_job_count() > 0
+    }
+
     pub fn max_concurrent(&self) -> usize {
         self.max_concurrent
     }
@@ -449,6 +455,50 @@ mod tests {
         assert!(result.unwrap_err().to_string().contains("at capacity"));
     }
 
+    #[test]
+    fn has_active_tasks_false_when_empty_or_all_finished() {
+        let mut sup = Supervisor::new(4, 3);
+        assert!(!sup.has_active_tasks());
+        // make_handle drops its runtime, aborting the task, so the
+        // registered agent is already finished.
+        sup.register(make_handle("a1", "explore", 1)).unwrap();
+        assert!(!sup.has_active_tasks());
+    }
+
+    #[test]
+    fn has_active_tasks_true_with_running_agent() {
+        // Keep the runtime alive so the task is never polled and the agent
+        // counts as running.
+        let rt = Builder::new_current_thread().enable_all().build().unwrap();
+        let join_handle = rt.spawn(async {
+            Ok::<AgentResult, Error>(AgentResult {
+                id: "done".into(),
+                agent_name: "test".into(),
+                output: "result".into(),
+                exit_status: AgentExitStatus::Completed,
+            })
+        });
+        let running_handle = AgentHandle {
+            id: "a1".to_string(),
+            agent_name: "explore".to_string(),
+            depth: 1,
+            inbox: Arc::new(Inbox::new()),
+            abort_signal: create_abort_signal(),
+            join_handle,
+            child_supervisor: None,
+        };
+        let mut sup = Supervisor::new(4, 3);
+        sup.register(running_handle).unwrap();
+        assert!(sup.has_active_tasks());
+    }
+
+    #[test]
+    fn has_active_tasks_true_with_running_job() {
+        let mut sup = Supervisor::new(4, 3).with_max_concurrent_jobs(1);
+        sup.register(make_job("j1", create_abort_signal())).unwrap();
+        assert!(sup.has_active_tasks());
+    }
+
     #[test]
     fn supervisor_register_rejects_exceeding_depth() {
         let mut sup = Supervisor::new(4, 2);