From 8710737964e865f515e06bd460c1493e1f49894c Mon Sep 17 00:00:00 2001 From: Alex Clarke Date: Tue, 15 Sep 2026 20:17:59 -0600 Subject: [PATCH] feat(TASK-035): wire turn/session/compression seams + top-level agent bracket MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Turn bracket (repl::TurnBracket): turn.started on entry and exactly one of turn.{completed,interrupted,failed} on exit, with the interrupted-vs-failed discrimination (abort_signal.aborted_ctrlc()) in one place. Wired at the three outer boundaries: ask(), start_directive, and shell_execute — never at the per-LLM-round before/after_chat_completion seams. Session seams: session.started fires for newly created sessions only (use_session's new_from_ctx paths; resume fires nothing), session.ended in exit_session while the session is still attached, session.compressed on the success path of both compress_session call sites (auto-compression and the .compress command). Session-held role hooks: Session snapshots the role's HooksMap on set_role, clears it on clear_role/sync_agent, and re-reads it from the role file on load; resolved_hooks gains a session leg so role hooks keep firing when the role lives in the session (fresh set_role and resumed sessions alike). Top-level agent bracket: agent.started/completed/failed once per agent RUN — headless --agent dispatch and the interactive agent-context lifetime (enter via .agent or startup; exit via switch-away or REPL teardown), never per message. COYOTE_AGENT_ID is synthesized as top--. Exit-path seams call hooks::drain_pending with a 2s bound so exit hooks finish spawning before process teardown; interactive mid-session events never drain. --- src/config/request_context.rs | 431 ++++++++++++++++++++++++++++++++++ src/config/session.rs | 33 +++ src/hooks.rs | 16 +- src/main.rs | 82 +++++-- src/repl/mod.rs | 248 ++++++++++++++++++- 5 files changed, 778 insertions(+), 32 deletions(-) diff --git a/src/config/request_context.rs b/src/config/request_context.rs index 921efa6..16daa82 100644 --- a/src/config/request_context.rs +++ b/src/config/request_context.rs @@ -50,6 +50,7 @@ use super::memory::{ DEFAULT_MEMORY_CAP_WITH_TOOLS, DEFAULT_MEMORY_CAP_WITHOUT_TOOLS, MemoryStore, WorkspaceMemory, }; use crate::graph; +use crate::hooks::{self, HookEvent}; use anyhow::{Context, Error, Result, bail}; use colored::Colorize; use gman::providers::SupportedProvider; @@ -59,6 +60,7 @@ use inquire::{Confirm, MultiSelect, Text, list_option::ListOption, validator::Va use log::warn; use parking_lot::RwLock; use prompts::DEFAULT_SKILL_INSTRUCTIONS; +use rand::distr::{Alphanumeric, SampleString}; use std::collections::{BTreeSet, HashMap, HashSet}; use std::fs::{File, OpenOptions, read_dir, read_to_string, remove_dir_all, remove_file}; use std::io::Write; @@ -354,6 +356,13 @@ pub struct RequestContext { pub session_abort: Option, + /// Synthesized `COYOTE_AGENT_ID` for a top-level agent run (headless + /// `--agent` dispatch or an interactive agent context). Scheme: + /// `top--<8-char alphanumeric nonce>`; spawned sub-agents + /// carry supervisor-issued ids instead. `Some` exactly while the + /// top-level `agent.*` bracket is open. + pub top_level_agent_id: Option, + pub render_mode: RenderMode, } @@ -398,6 +407,7 @@ impl RequestContext { auto_continue_paused: None, pending_prefill: None, session_abort: None, + top_level_agent_id: None, render_mode: RenderMode::default(), } } @@ -469,6 +479,7 @@ impl RequestContext { auto_continue_paused: None, pending_prefill: None, session_abort: None, + top_level_agent_id: None, render_mode: RenderMode::default(), }; ctx.refresh_mcp_tool_filters(); @@ -529,6 +540,9 @@ impl RequestContext { auto_continue_paused: self.auto_continue_paused.clone(), pending_prefill: None, session_abort: self.session_abort.clone(), + // Branches never own the top-level `agent.*` bracket; only the + // root flow that opened it may fire its terminal event. + top_level_agent_id: None, render_mode: self.render_mode, } } @@ -585,6 +599,7 @@ impl RequestContext { auto_continue_paused: None, pending_prefill: None, session_abort: None, + top_level_agent_id: None, render_mode: parent.render_mode, } } @@ -1054,6 +1069,12 @@ impl RequestContext { } pub fn exit_session(&mut self) -> Result<()> { + // Fired while the session is still attached so the hook sees + // COYOTE_SESSION_ID and session-held role hooks still resolve; no + // session means no event. + if self.session.is_some() { + hooks::fire(HookEvent::SessionEnded, self, &[], None); + } if let Some(mut session) = self.session.take() { let sessions_dir = self.sessions_dir(); session.exit(&sessions_dir, self.working_mode.is_repl())?; @@ -4613,6 +4634,7 @@ impl RequestContext { ); } let mut session; + let mut created_new_session = false; match session_name { None | Some(TEMP_SESSION_NAME) => { let session_file = self.session_file(TEMP_SESSION_NAME); @@ -4622,11 +4644,13 @@ impl RequestContext { })?; } session = Some(Session::new_from_ctx(self, app, TEMP_SESSION_NAME)?); + created_new_session = true; } Some(name) => { let session_path = self.session_file(name); if !session_path.exists() { session = Some(Session::new_from_ctx(self, app, name)?); + created_new_session = true; } else { session = Some(Session::load_from_ctx(self, app, name, &session_path)?); } @@ -4709,6 +4733,11 @@ impl RequestContext { } self.refresh_mcp_tool_filters(); self.init_agent_session_variables(new_session)?; + // NEW sessions only, and only once the session is fully engaged on + // the context; resuming an existing session fires nothing. + if created_new_session { + hooks::fire(HookEvent::SessionStarted, self, &[], None); + } Ok(()) } @@ -4874,6 +4903,9 @@ impl RequestContext { tool_tracker, }; + // Interactive switch-away closes the top-level agent bracket while + // the agent is still attached, so agent-scoped hooks resolve. + self.top_level_agent_finished(None); if self.agent.take().is_some() { if let Some(supervisor) = self.supervisor.clone() { supervisor.read().cancel_recursive(); @@ -4901,6 +4933,60 @@ impl RequestContext { Ok(()) } + /// Opens the top-level `agent.*` bracket: fires `agent.started` once per + /// agent-context lifetime when the ROOT context runs with a loaded agent + /// (headless `--agent` dispatch, or entering an agent interactively). + /// Spawned sub-agents are bracketed at their spawn seam instead, never + /// here. Synthesizes and stores `COYOTE_AGENT_ID` under the + /// `top--<8-char alphanumeric nonce>` scheme; calling this + /// again while the bracket is open, or without an agent, is a no-op — + /// per-message granularity belongs to `turn.*`, not `agent.*`. + pub fn top_level_agent_started(&mut self) { + if self.top_level_agent_id.is_some() { + return; + } + let Some(agent) = self.agent.as_ref() else { + return; + }; + let nonce = Alphanumeric.sample_string(&mut rand::rng(), 8); + let id = format!("top-{}-{nonce}", agent.name()); + self.top_level_agent_id = Some(id.clone()); + hooks::fire( + HookEvent::AgentStarted, + self, + &[("COYOTE_AGENT_ID", id)], + None, + ); + } + + /// Closes the top-level `agent.*` bracket: fires `agent.completed`, or + /// `agent.failed` carrying `COYOTE_AGENT_ERROR` when the run ended in an + /// error. No-op unless [`Self::top_level_agent_started`] opened the + /// bracket; the stored id is consumed so exactly one terminal event + /// fires per lifetime. + pub fn top_level_agent_finished(&mut self, error: Option<&Error>) { + let Some(id) = self.top_level_agent_id.take() else { + return; + }; + match error { + None => hooks::fire( + HookEvent::AgentCompleted, + self, + &[("COYOTE_AGENT_ID", id)], + None, + ), + Some(err) => hooks::fire( + HookEvent::AgentFailed, + self, + &[ + ("COYOTE_AGENT_ID", id), + ("COYOTE_AGENT_ERROR", format!("{err:#}")), + ], + None, + ), + } + } + pub async fn edit_role(&mut self, app: &AppConfig, abort_signal: AbortSignal) -> Result<()> { let role_name; if let Some(session) = self.session.as_ref() { @@ -5421,6 +5507,7 @@ mod tests { use crate::config::tool_scope::test_fixtures::{FixtureServer, fixture_runtime}; use crate::function::jobs::RingBuf; use crate::function::{ToolCall, skill}; + use crate::hooks::{HookDef, HooksMap, test_sink}; use crate::mcp::{McpServer, McpServerFeatures, McpServersConfig, McpTransportType}; use crate::supervisor::{ AgentExitStatus, AgentHandle, AgentResult, JobHandle, JobResult, JobState, JobStatus, @@ -5428,6 +5515,7 @@ mod tests { use crate::utils; use crate::utils::get_env_name; use crate::vault::Vault; + use anyhow::anyhow; use rmcp::model::PromptArgument; use serde_json::json; use serial_test::serial; @@ -8230,6 +8318,349 @@ mod tests { assert!(ctx.session.is_none()); } + fn hooks_map_of(events: &[&str], marker: &str) -> HooksMap { + let mut map = HooksMap::default(); + for event in events { + map.insert( + event.to_string(), + vec![HookDef { + name: format!("{marker}-{event}"), + command: "true".to_string(), + }], + ); + } + map + } + + fn marker_captures(marker: &str) -> Vec { + test_sink::snapshot() + .into_iter() + .filter(|capture| capture.hook_name.starts_with(marker)) + .collect() + } + + /// Writes an agent whose config carries agent-scoped hooks for every + /// `agent.*` event, named `-`. + fn seed_agent_with_hooks(agent_name: &str, marker: &str) { + let agent_dir = paths::agent_data_dir(agent_name); + create_dir_all(&agent_dir).unwrap(); + let mut config = format!("name: {agent_name}\ninstructions: hi\nhooks:\n"); + for event in ["agent.started", "agent.completed", "agent.failed"] { + config.push_str(&format!( + " {event}:\n - name: {marker}-{event}\n command: 'true'\n" + )); + } + write(agent_dir.join("config.yaml"), config).unwrap(); + } + + fn unique_name(prefix: &str) -> String { + format!( + "{prefix}_{}", + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos() + ) + } + + #[test] + #[serial] + fn use_session_fires_session_started_for_new_sessions_only() { + let _guard = TestConfigDirGuard::new(); + let sessions_dir = paths::local_dir("sessions"); + create_dir_all(&sessions_dir).unwrap(); + let _sink = test_sink::install(); + let marker = "sess-started-q8n"; + + let mut ctx = create_test_ctx(); + ctx.update_app_config(|app| app.hooks = hooks_map_of(&["session.started"], marker)); + let app = ctx.app.config.clone(); + let abort = utils::create_abort_signal(); + run_async(ctx.use_session(&app, Some("hook-fresh"), abort.clone())).unwrap(); + + let captures = marker_captures(marker); + assert_eq!(captures.len(), 1, "a new session fires exactly one event"); + assert_eq!( + captures[0] + .envs + .get("COYOTE_SESSION_ID") + .map(String::as_str), + Some("hook-fresh") + ); + + // Resuming a persisted session fires nothing. + write_paused_todo_session(&ctx, "hook-resume"); + let mut resumed_ctx = create_test_ctx(); + resumed_ctx.update_app_config(|app| app.hooks = hooks_map_of(&["session.started"], marker)); + let resumed_app = resumed_ctx.app.config.clone(); + run_async(resumed_ctx.use_session(&resumed_app, Some("hook-resume"), abort)).unwrap(); + assert_eq!( + marker_captures(marker).len(), + 1, + "resume must not fire session.started" + ); + } + + #[test] + #[serial] + fn exit_session_fires_session_ended_only_with_a_session() { + let _guard = TestConfigDirGuard::new(); + let _sink = test_sink::install(); + let marker = "sess-ended-w5c"; + + let mut ctx = create_test_ctx(); + ctx.update_app_config(|app| app.hooks = hooks_map_of(&["session.ended"], marker)); + + // Sessionless (headless) exit path: nothing fires. + ctx.exit_session().unwrap(); + assert!(marker_captures(marker).is_empty()); + + ctx.session = Some(Session::default()); + ctx.exit_session().unwrap(); + ctx.exit_session().unwrap(); + assert_eq!( + marker_captures(marker).len(), + 1, + "session.ended fires exactly once, only while a session is attached" + ); + } + + #[test] + #[serial] + fn chat_completion_seams_fire_no_turn_events_in_child_contexts() { + let _guard = TestConfigDirGuard::new(); + let _sink = test_sink::install(); + let marker = "child-turn-b4h"; + + let mut ctx = create_test_ctx(); + ctx.update_app_config(|app| { + app.hooks = hooks_map_of( + &[ + "turn.started", + "turn.completed", + "turn.interrupted", + "turn.failed", + ], + marker, + ) + }); + let app = ctx.app.config.clone(); + + let input = Input::from_str(&ctx, "hello", None).unwrap(); + ctx.before_chat_completion(&input).unwrap(); + ctx.after_chat_completion(&app, &input, "output", &[]) + .unwrap(); + + assert!( + marker_captures(marker).is_empty(), + "per-round chat-completion seams must never fire turn events" + ); + } + + #[test] + #[serial] + fn top_level_agent_bracket_fires_started_and_completed_once() { + let _guard = TestConfigDirGuard::new(); + let _sink = test_sink::install(); + let marker = "agent-top-ok-m2s"; + 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)).unwrap(); + + ctx.top_level_agent_started(); + // Per agent RUN, never per message: re-entering while open is a no-op. + ctx.top_level_agent_started(); + ctx.top_level_agent_finished(None); + ctx.top_level_agent_finished(None); + + let captures = marker_captures(marker); + let started: Vec<_> = captures + .iter() + .filter(|capture| capture.hook_name == format!("{marker}-agent.started")) + .collect(); + let completed: Vec<_> = captures + .iter() + .filter(|capture| capture.hook_name == format!("{marker}-agent.completed")) + .collect(); + assert_eq!(started.len(), 1); + assert_eq!(completed.len(), 1); + assert!( + captures + .iter() + .all(|capture| capture.hook_name != format!("{marker}-agent.failed")) + ); + let id = started[0] + .envs + .get("COYOTE_AGENT_ID") + .expect("agent id env"); + assert!( + id.starts_with(&format!("top-{agent_name}-")), + "synthesized id must follow the top-- scheme: {id}" + ); + assert_eq!(completed[0].envs.get("COYOTE_AGENT_ID"), Some(id)); + assert_eq!( + started[0].envs.get("COYOTE_AGENT_NAME").map(String::as_str), + Some(agent_name.as_str()) + ); + } + + #[test] + #[serial] + fn top_level_agent_failure_fires_agent_failed_with_error() { + let _guard = TestConfigDirGuard::new(); + let _sink = test_sink::install(); + let marker = "agent-top-err-f7d"; + 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)).unwrap(); + + ctx.top_level_agent_started(); + let err = anyhow!("dispatch exploded"); + ctx.top_level_agent_finished(Some(&err)); + + let captures = marker_captures(marker); + let failed: Vec<_> = captures + .iter() + .filter(|capture| capture.hook_name == format!("{marker}-agent.failed")) + .collect(); + assert_eq!(failed.len(), 1); + assert!( + captures + .iter() + .all(|capture| capture.hook_name != format!("{marker}-agent.completed")) + ); + assert_eq!( + failed[0].envs.get("COYOTE_AGENT_ERROR").map(String::as_str), + Some("dispatch exploded") + ); + assert!(failed[0].envs.contains_key("COYOTE_AGENT_ID")); + } + + #[test] + #[serial] + fn top_level_agent_helpers_noop_without_agent() { + let _guard = TestConfigDirGuard::new(); + let _sink = test_sink::install(); + let marker = "agent-top-role-only-c9k"; + + let mut ctx = create_test_ctx(); + ctx.update_app_config(|app| { + app.hooks = hooks_map_of( + &["agent.started", "agent.completed", "agent.failed"], + marker, + ) + }); + ctx.role = Some(Role::new("plain", "prompt")); + + ctx.top_level_agent_started(); + ctx.top_level_agent_finished(None); + + assert!( + marker_captures(marker).is_empty(), + "a role-only context must fire no agent events" + ); + assert!(ctx.top_level_agent_id.is_none()); + } + + #[test] + #[serial] + fn exit_agent_closes_top_level_agent_bracket() { + let _guard = TestConfigDirGuard::new(); + let _sink = test_sink::install(); + let marker = "agent-top-exit-v3p"; + 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)).unwrap(); + ctx.top_level_agent_started(); + + ctx.exit_agent(&app).unwrap(); + // A later exit seam (REPL teardown) must not double-fire. + ctx.top_level_agent_finished(None); + + let captures = marker_captures(marker); + assert_eq!( + captures + .iter() + .filter(|capture| capture.hook_name == format!("{marker}-agent.completed")) + .count(), + 1, + "switch-away fires the terminal event exactly once" + ); + } + + #[test] + #[serial] + fn resolved_hooks_resolves_role_hooks_held_by_fresh_session() { + let _guard = TestConfigDirGuard::new(); + let sessions_dir = paths::local_dir("sessions"); + create_dir_all(&sessions_dir).unwrap(); + + let mut ctx = create_test_ctx(); + let app = ctx.app.config.clone(); + let abort = utils::create_abort_signal(); + run_async(ctx.use_session(&app, Some("role-holder"), abort)).unwrap(); + + let role_content = "---\nhooks:\n turn.completed:\n - name: session-role-probe\n command: role-cmd\n---\nPrompt"; + ctx.use_role_obj(Role::new("hooked-role", role_content)) + .unwrap(); + assert!( + ctx.role.is_none(), + "use_role_obj must move the role into the active session" + ); + + let resolved = ctx.resolved_hooks(HookEvent::TurnCompleted); + assert_eq!(resolved.len(), 1); + assert_eq!(resolved[0].name, "session-role-probe"); + assert_eq!(resolved[0].command, "role-cmd"); + assert_eq!(resolved[0].cwd, paths::roles_dir()); + } + + #[test] + #[serial] + fn resolved_hooks_resolves_role_hooks_after_session_resume() { + let _guard = TestConfigDirGuard::new(); + let sessions_dir = paths::local_dir("sessions"); + create_dir_all(&sessions_dir).unwrap(); + let roles_dir = paths::roles_dir(); + create_dir_all(&roles_dir).unwrap(); + write( + roles_dir.join("resume-hooked.md"), + "---\nhooks:\n session.ended:\n - name: resume-role-probe\n command: role-cmd\n---\nYou are hooked.", + ) + .unwrap(); + + let seed_ctx = create_test_ctx(); + let session_path = seed_ctx.session_file("resume-role"); + ensure_parent_exists(&session_path).unwrap(); + write( + &session_path, + "model: test-seeded:test-chat\nrole_name: resume-hooked\nmessages: []\n", + ) + .unwrap(); + + let mut ctx = create_test_ctx(); + let app = ctx.app.config.clone(); + let abort = utils::create_abort_signal(); + run_async(ctx.use_session(&app, Some("resume-role"), abort)).unwrap(); + assert!(ctx.role.is_none()); + + let resolved = ctx.resolved_hooks(HookEvent::SessionEnded); + assert_eq!(resolved.len(), 1, "resumed session must restore role hooks"); + assert_eq!(resolved[0].name, "resume-role-probe"); + } + fn write_paused_todo_session(ctx: &RequestContext, name: &str) { let session_path = ctx.session_file(name); ensure_parent_exists(&session_path).unwrap(); diff --git a/src/config/session.rs b/src/config/session.rs index 6f541b9..6a0e244 100644 --- a/src/config/session.rs +++ b/src/config/session.rs @@ -3,6 +3,7 @@ use super::todo::TodoList; use super::*; use crate::client::{Message, MessageContent, MessageRole, TokenUsage}; +use crate::hooks::HooksMap; use crate::render::MarkdownRender; use anyhow::{Context, Result, bail}; @@ -113,6 +114,8 @@ pub struct Session { #[serde(skip)] role_prompt: String, #[serde(skip)] + role_hooks: Option, + #[serde(skip)] name: String, #[serde(skip)] path: Option, @@ -193,6 +196,7 @@ impl Session { && let Ok(role) = ctx.retrieve_role(app, role_name) { session.role_prompt = role.prompt().to_string(); + session.role_hooks = role.hooks().cloned(); } session.update_tokens(); @@ -254,6 +258,13 @@ impl Session { self.role_name.as_deref() } + /// Hooks of the role this session holds. `set_role` snapshots them + /// because only the frontmatter-stripped role prompt is persisted; + /// `load_from_ctx` restores them by re-reading the role file on resume. + pub fn role_hooks(&self) -> Option<&HooksMap> { + self.role_hooks.as_ref() + } + pub fn dirty(&self) -> bool { self.dirty } @@ -561,6 +572,7 @@ impl Session { self.model = role.model().clone(); self.role_name = convert_option_string(role.name()); self.role_prompt = role.prompt().to_string(); + self.role_hooks = role.hooks().cloned(); self.dirty = true; self.update_tokens(); } @@ -568,11 +580,13 @@ impl Session { pub fn clear_role(&mut self) { self.role_name = None; self.role_prompt.clear(); + self.role_hooks = None; } pub fn sync_agent(&mut self, agent: &Agent) { self.role_name = None; self.role_prompt = agent.interpolated_instructions(); + self.role_hooks = None; self.agent_variables = agent.variables().clone(); self.agent_instructions = self.role_prompt.clone(); if let Some(threshold) = agent.compression_threshold() { @@ -1111,6 +1125,25 @@ mod tests { assert!(!session.dirty()); } + #[test] + fn set_role_captures_and_clear_role_drops_role_hooks() { + let content = "---\nhooks:\n turn.completed:\n - name: role-probe\n command: role-cmd\n---\nPrompt"; + let mut session = Session::default(); + + session.set_role(Role::new("hooked", content)); + let hooks = session.role_hooks().expect("set_role must snapshot hooks"); + assert_eq!(hooks["turn.completed"][0].name, "role-probe"); + + session.clear_role(); + assert!(session.role_hooks().is_none()); + + session.set_role(Role::new("plain", "No frontmatter")); + assert!( + session.role_hooks().is_none(), + "a role without hooks must not leave a stale snapshot" + ); + } + fn push_interrupted_turn(session: &mut Session) { session.messages.push(Message::new( MessageRole::User, diff --git a/src/hooks.rs b/src/hooks.rs index e8e082d..f7cfee3 100644 --- a/src/hooks.rs +++ b/src/hooks.rs @@ -97,9 +97,10 @@ impl RequestContext { /// (whitelisted through `global_hooks` when an agent is active), then /// role, then agent. Duplicate names across scopes all fire. /// - /// Role hooks resolve only while the role is held directly on the context; - /// a role held by an active session is not consulted, because sessions do - /// not carry hooks. Resolution through sessions is a call-site concern. + /// Role hooks resolve from the role held directly on the context or, + /// when the active role was moved into (or restored by) a session, from + /// the hooks snapshot the session captured via `Session::set_role` / + /// `Session::load_from_ctx`. #[allow(dead_code)] pub fn resolved_hooks(&self, event: HookEvent) -> Vec { resolve_hooks( @@ -108,7 +109,14 @@ impl RequestContext { self.agent .as_ref() .map(|agent| (agent.global_hooks(), agent.name())), - self.role.as_ref().and_then(|role| role.hooks()), + self.role + .as_ref() + .and_then(|role| role.hooks()) + .or_else(|| { + self.session + .as_ref() + .and_then(|session| session.role_hooks()) + }), self.agent .as_ref() .map(|agent| (agent.hooks(), agent.name())), diff --git a/src/main.rs b/src/main.rs index 9ac333a..1c197a9 100644 --- a/src/main.rs +++ b/src/main.rs @@ -23,7 +23,8 @@ extern crate log; use crate::cli::Cli; use crate::client::{ - ModelType, call_chat_completions, call_chat_completions_streaming, catalog, list_models, oauth, + Client, ModelType, call_chat_completions, call_chat_completions_streaming, catalog, + list_models, oauth, }; use crate::config::instructions::WORKSPACE_INSTRUCTIONS_FILE_NAME; use crate::config::{ @@ -36,7 +37,7 @@ use crate::config::{memory, paths}; use crate::function::agents::{GuardrailAction, check_pending_tasks_guardrail}; use crate::mcp::McpServersConfig; use crate::render::{prompt_theme, render_error}; -use crate::repl::Repl; +use crate::repl::{EXIT_HOOK_DRAIN_TIMEOUT, Repl, TurnBracket}; use crate::utils::*; use crate::vault::{Vault, interpolate_secrets}; use anyhow::{Context, Result, anyhow, bail}; @@ -387,6 +388,7 @@ async fn run( let app = Arc::clone(&ctx.app.config); ctx.use_agent(app.as_ref(), agent, session, abort_signal.clone()) .await?; + ctx.top_level_agent_started(); } else { let app: Arc = Arc::clone(&ctx.app.config); if let Some(prompt) = &cli.temp_role { @@ -523,9 +525,14 @@ async fn run( return Ok(()); } if cli.execute && !is_repl { - let input = create_input(&ctx, text, &cli.file, abort_signal.clone()).await?; - shell_execute(&mut ctx, &SHELL, input, abort_signal.clone()).await?; - return Ok(()); + let result = async { + let input = create_input(&ctx, text, &cli.file, abort_signal.clone()).await?; + shell_execute(&mut ctx, &SHELL, input, abort_signal.clone()).await + } + .await; + ctx.top_level_agent_finished(result.as_ref().err()); + hooks::drain_pending(EXIT_HOOK_DRAIN_TIMEOUT).await; + return result; } { @@ -545,9 +552,15 @@ async fn run( match is_repl { false => { - let mut input = create_input(&ctx, text, &cli.file, abort_signal.clone()).await?; - input.use_embeddings(abort_signal.clone()).await?; - start_directive(&mut ctx, input, cli.code, abort_signal).await + let result = async { + let mut input = create_input(&ctx, text, &cli.file, abort_signal.clone()).await?; + input.use_embeddings(abort_signal.clone()).await?; + start_directive(&mut ctx, input, cli.code, abort_signal.clone()).await + } + .await; + ctx.top_level_agent_finished(result.as_ref().err()); + hooks::drain_pending(EXIT_HOOK_DRAIN_TIMEOUT).await; + result } true => { if !*IS_STDOUT_TERMINAL { @@ -564,6 +577,17 @@ async fn start_directive( input: Input, code_mode: bool, abort_signal: AbortSignal, +) -> Result<()> { + let bracket = TurnBracket::enter(ctx); + let result = start_directive_inner(ctx, input, code_mode, abort_signal.clone()).await; + bracket.finish(ctx, &abort_signal, result) +} + +async fn start_directive_inner( + ctx: &mut RequestContext, + input: Input, + code_mode: bool, + abort_signal: AbortSignal, ) -> Result<()> { let app: Arc = Arc::clone(&ctx.app.config); ctx.session_abort = Some(abort_signal.clone()); @@ -639,22 +663,10 @@ async fn shell_execute( abort_signal: AbortSignal, ) -> Result<()> { let app: Arc = Arc::clone(&ctx.app.config); - let client = input.create_client()?; - ctx.before_chat_completion(&input)?; - let (eval_str, _) = call_chat_completions( - &input, - false, - true, - client.as_ref(), - ctx, - abort_signal.clone(), - ) - .await?; - - ctx.after_chat_completion(app.as_ref(), &input, &eval_str, &[])?; - if eval_str.is_empty() { - bail!("No command generated"); - } + // Each revision round is its own turn: the recursive 'r' call re-enters. + let bracket = TurnBracket::enter(ctx); + let result = shell_execute_turn(ctx, &input, abort_signal.clone()).await; + let (client, eval_str) = bracket.finish(ctx, &abort_signal, result)?; if app.dry_run { app.print_markdown(&eval_str)?; return Ok(()); @@ -680,6 +692,7 @@ async fn shell_execute( if code == 0 && app.save_shell_history { let _ = append_to_shell_history(&shell.name, &eval_str, code); } + hooks::drain_pending(EXIT_HOOK_DRAIN_TIMEOUT).await; process::exit(code); } 'r' => { @@ -727,6 +740,27 @@ async fn shell_execute( Ok(()) } +/// The LLM portion of one shell-execute turn — everything the turn bracket +/// measures. The interactive follow-up menu (execute/revise/describe/copy) +/// runs after the turn has already completed. +async fn shell_execute_turn( + ctx: &mut RequestContext, + input: &Input, + abort_signal: AbortSignal, +) -> Result<(Box, String)> { + let app: Arc = Arc::clone(&ctx.app.config); + let client = input.create_client()?; + ctx.before_chat_completion(input)?; + let (eval_str, _) = + call_chat_completions(input, false, true, client.as_ref(), ctx, abort_signal).await?; + + ctx.after_chat_completion(app.as_ref(), input, &eval_str, &[])?; + if eval_str.is_empty() { + bail!("No command generated"); + } + Ok((client, eval_str)) +} + async fn create_input( ctx: &RequestContext, text: Option, diff --git a/src/repl/mod.rs b/src/repl/mod.rs index f0b41a2..9c45845 100644 --- a/src/repl/mod.rs +++ b/src/repl/mod.rs @@ -17,6 +17,7 @@ use crate::config::{ }; 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::utils::{ AbortSignal, SHELL, abortable_run_with_spinner, create_abort_signal, dimmed_text, @@ -41,11 +42,58 @@ use reedline::{ use reedline::{MenuBuilder, Signal}; use std::collections::HashMap; use std::sync::LazyLock; +use std::time::Duration; use std::{env, process, sync::Arc}; use tokio::task; const MENU_NAME: &str = "completion_menu"; +/// Bounded wait applied at process-exit seams (REPL teardown, headless and +/// shell-execute exits) so already-fired exit hooks — `session.ended`, the +/// final `turn.*` / `agent.*` events — finish spawning before the process +/// tears down. Never used mid-session. +pub(crate) const EXIT_HOOK_DRAIN_TIMEOUT: Duration = Duration::from_secs(2); + +/// Brackets one user-visible turn at an outer boundary (`ask`, +/// `start_directive`, `shell_execute`): `enter` fires `turn.started`, and +/// `finish` fires exactly one of `turn.completed` / `turn.interrupted` / +/// `turn.failed`. The interrupted-vs-failed discrimination lives here and +/// nowhere else: a ctrl-c abort wins over the result, then an error means +/// failed (carrying `COYOTE_ERROR`), otherwise completed. Tool rounds, +/// auto-continue, and compression retries run inside one bracket and never +/// re-fire turn events. +pub(crate) struct TurnBracket(()); + +impl TurnBracket { + #[must_use] + pub(crate) fn enter(ctx: &RequestContext) -> Self { + hooks::fire(HookEvent::TurnStarted, ctx, &[], None); + Self(()) + } + + pub(crate) fn finish( + self, + ctx: &RequestContext, + abort_signal: &AbortSignal, + result: Result, + ) -> Result { + if abort_signal.aborted_ctrlc() { + hooks::fire(HookEvent::TurnInterrupted, ctx, &[], None); + } else { + match &result { + Ok(_) => hooks::fire(HookEvent::TurnCompleted, ctx, &[], None), + Err(err) => hooks::fire( + HookEvent::TurnFailed, + ctx, + &[("COYOTE_ERROR", format!("{err:#}"))], + None, + ), + } + } + result + } +} + pub const DEFAULT_CONTINUATION_PROMPT: &str = indoc! {" [SYSTEM REMINDER - TODO CONTINUATION] You have incomplete tasks. Rules: @@ -482,8 +530,12 @@ Type ".help" for additional help. supervisor.read().cancel_recursive(); } - self.ctx.write().exit_session()?; - Ok(()) + let exit_result = self.ctx.write().exit_session(); + self.ctx + .write() + .top_level_agent_finished(exit_result.as_ref().err()); + hooks::drain_pending(EXIT_HOOK_DRAIN_TIMEOUT).await; + exit_result } fn create_editor(ctx: Arc>, app: &AppConfig) -> Result { @@ -983,6 +1035,7 @@ pub async fn run_repl_command( let app = Arc::clone(&ctx.app.config); ctx.use_agent(app.as_ref(), agent_name, session_name, abort_signal.clone()) .await?; + ctx.top_level_agent_started(); if let Some(session) = &ctx.session { let (compressed, active) = replay::snapshot(session); replay::render(app.as_ref(), &compressed, &active)?; @@ -1090,6 +1143,7 @@ pub async fn run_repl_command( abort_signal.clone(), ) .await?; + hooks::fire(HookEvent::SessionCompressed, ctx, &[], None); println!("✓ Successfully compressed the session."); } _ => { @@ -1433,6 +1487,17 @@ async fn ask( abort_signal: AbortSignal, input: Input, with_embeddings: bool, +) -> Result<()> { + let bracket = TurnBracket::enter(ctx); + let result = ask_inner(ctx, abort_signal.clone(), input, with_embeddings).await; + bracket.finish(ctx, &abort_signal, result) +} + +async fn ask_inner( + ctx: &mut RequestContext, + abort_signal: AbortSignal, + input: Input, + with_embeddings: bool, ) -> Result<()> { let mut input = input; let mut with_embeddings = with_embeddings; @@ -1579,8 +1644,9 @@ async fn ask( }; eprintln!("\n📢 {}", color.italic().paint("Compressing the session."),); - if let Err(err) = ctx.compress_session().await { - warn!("Failed to compress the session: {err}"); + match ctx.compress_session().await { + Ok(()) => hooks::fire(HookEvent::SessionCompressed, ctx, &[], None), + Err(err) => warn!("Failed to compress the session: {err}"), } if let Some(session) = ctx.session.as_mut() { session.set_compressing(false); @@ -1939,6 +2005,180 @@ pub fn split_args_text(line: &str, is_win: bool) -> (Vec, &str) { #[cfg(test)] mod tests { use super::*; + use crate::config::{AppState, Session, WorkingMode}; + use crate::hooks::{HookDef, HooksMap, test_sink}; + use anyhow::anyhow; + use serial_test::serial; + + const TURN_EVENTS: [&str; 4] = [ + "turn.started", + "turn.completed", + "turn.interrupted", + "turn.failed", + ]; + + fn ctx_with_hooks(events: &[&str], marker: &str) -> RequestContext { + let mut hooks_map = HooksMap::default(); + for event in events { + hooks_map.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: hooks_map, + ..Default::default() + }); + RequestContext::new(Arc::new(app), WorkingMode::Repl) + } + + /// Sink captures for `marker`, ordered (started, completed, interrupted, failed). + fn turn_counts(marker: &str) -> (usize, usize, usize, usize) { + let captures = test_sink::snapshot(); + let count = |event: &str| { + captures + .iter() + .filter(|capture| capture.hook_name == format!("{marker}-{event}")) + .count() + }; + ( + count("turn.started"), + count("turn.completed"), + count("turn.interrupted"), + count("turn.failed"), + ) + } + + #[tokio::test] + #[serial] + async fn turn_bracket_ok_final_fires_started_and_completed() { + let _sink = test_sink::install(); + let marker = "tb-ok-x7q"; + let ctx = ctx_with_hooks(&TURN_EVENTS, marker); + let abort_signal = create_abort_signal(); + + let bracket = TurnBracket::enter(&ctx); + let result: Result<()> = async { Ok(()) }.await; + bracket.finish(&ctx, &abort_signal, result).unwrap(); + + assert_eq!(turn_counts(marker), (1, 1, 0, 0)); + } + + #[tokio::test] + #[serial] + async fn turn_bracket_tool_rounds_then_final_fire_one_turn() { + let _sink = test_sink::install(); + let marker = "tb-rounds-p2m"; + let ctx = ctx_with_hooks(&TURN_EVENTS, marker); + let abort_signal = create_abort_signal(); + + let bracket = TurnBracket::enter(&ctx); + let result: Result<()> = async { + // Simulates a turn that loops through tool rounds before the + // final round; rounds happen inside the bracket and never re-fire. + let mut rounds = 0; + loop { + rounds += 1; + if rounds == 3 { + break Ok(()); + } + } + } + .await; + bracket.finish(&ctx, &abort_signal, result).unwrap(); + + assert_eq!(turn_counts(marker), (1, 1, 0, 0)); + } + + #[tokio::test] + #[serial] + async fn turn_bracket_error_fires_failed_with_error_env() { + let _sink = test_sink::install(); + let marker = "tb-err-k4d"; + let ctx = ctx_with_hooks(&TURN_EVENTS, marker); + let abort_signal = create_abort_signal(); + + let bracket = TurnBracket::enter(&ctx); + let result: Result<()> = async { Err(anyhow!("turn exploded")) }.await; + let err = bracket.finish(&ctx, &abort_signal, result).unwrap_err(); + assert_eq!(err.to_string(), "turn exploded"); + + assert_eq!(turn_counts(marker), (1, 0, 0, 1)); + let captures = test_sink::snapshot(); + let failed = captures + .iter() + .find(|capture| capture.hook_name == format!("{marker}-turn.failed")) + .expect("turn.failed capture"); + assert_eq!( + failed.envs.get("COYOTE_ERROR").map(String::as_str), + Some("turn exploded") + ); + } + + #[tokio::test] + #[serial] + async fn turn_bracket_ctrlc_error_fires_interrupted_not_failed() { + let _sink = test_sink::install(); + let marker = "tb-ctrlc-err-j6w"; + let ctx = ctx_with_hooks(&TURN_EVENTS, marker); + let abort_signal = create_abort_signal(); + abort_signal.set_ctrlc(); + + let bracket = TurnBracket::enter(&ctx); + let result: Result<()> = async { Err(anyhow!("Aborted!")) }.await; + assert!(bracket.finish(&ctx, &abort_signal, result).is_err()); + + assert_eq!(turn_counts(marker), (1, 0, 1, 0)); + } + + #[tokio::test] + #[serial] + async fn turn_bracket_ctrlc_with_ok_exit_fires_interrupted() { + let _sink = test_sink::install(); + let marker = "tb-ctrlc-ok-r3v"; + let ctx = ctx_with_hooks(&TURN_EVENTS, marker); + let abort_signal = create_abort_signal(); + abort_signal.set_ctrlc(); + + let bracket = TurnBracket::enter(&ctx); + let result: Result<()> = async { Ok(()) }.await; + bracket.finish(&ctx, &abort_signal, result).unwrap(); + + assert_eq!(turn_counts(marker), (1, 0, 1, 0)); + } + + #[tokio::test] + #[serial] + async fn compress_command_failure_fires_no_session_compressed() { + let _sink = test_sink::install(); + let marker = "compress-fail-z9t"; + let mut ctx = ctx_with_hooks(&["session.compressed"], marker); + ctx.session = Some(Session::default()); + let abort_signal = create_abort_signal(); + + // Boxed: `run_repl_command`'s state machine is far larger than a test + // thread's stack. + let result = Box::pin(run_repl_command( + &mut ctx, + abort_signal, + ".compress session", + )) + .await; + + assert!(result.is_err(), "compressing an empty session must fail"); + assert_eq!( + test_sink::snapshot() + .iter() + .filter(|capture| capture.hook_name.starts_with(marker)) + .count(), + 0, + "a failed compression must fire no session.compressed hook" + ); + } #[test] fn test_process_command_line() {