From a3b71a6889b9c994d527531ad2de6a810d7b2955 Mon Sep 17 00:00:00 2001 From: Alex Clarke Date: Wed, 16 Sep 2026 19:55:52 -0600 Subject: [PATCH] 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 | 37 +++++++++++++++ src/config/request_context.rs | 44 ++++++++++++++++++ src/function/agents.rs | 79 +++++++++++++++++++++++++++++++- src/function/user_interaction.rs | 48 +++++++++++++++++++ src/graph/agent.rs | 37 ++++++++++++++- src/repl/mod.rs | 78 ++++++++++++++++++++++++++++++- src/supervisor/mod.rs | 50 ++++++++++++++++++++ 7 files changed, 369 insertions(+), 4 deletions(-) diff --git a/scripts/usage-probe-hooks.sh b/scripts/usage-probe-hooks.sh index 2cd99be..90326e1 100755 --- a/scripts/usage-probe-hooks.sh +++ b/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 ` 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" diff --git a/src/config/request_context.rs b/src/config/request_context.rs index 7756496..b7566e2 100644 --- a/src/config/request_context.rs +++ b/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() { diff --git a/src/function/agents.rs b/src/function/agents.rs index 10847f4..4ff0bc6 100644 --- a/src/function/agents.rs +++ b/src/function/agents.rs @@ -1782,7 +1782,11 @@ fn handle_reply_escalation(ctx: &mut RequestContext, args: &Value) -> Result 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); diff --git a/src/function/user_interaction.rs b/src/function/user_interaction.rs index c22fc8c..6281fd1 100644 --- a/src/function/user_interaction.rs +++ b/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(); diff --git a/src/graph/agent.rs b/src/graph/agent.rs index b2f22b9..39cdb7e 100644 --- a/src/graph/agent.rs +++ b/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; diff --git a/src/repl/mod.rs b/src/repl/mod.rs index b0e85b6..69a0ec3 100644 --- a/src/repl/mod.rs +++ b/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) { + let supervisor = supervisor.read(); + if supervisor.has_active_tasks() { + abort_signal.set_ctrlc(); + } + supervisor.cancel_recursive(); +} + fn pause_banner_text(ctx: &RequestContext) -> Option { 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" + ); + } } diff --git a/src/supervisor/mod.rs b/src/supervisor/mod.rs index 044635c..f80eb4e 100644 --- a/src/supervisor/mod.rs +++ b/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 { + 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);