diff --git a/src/client/common.rs b/src/client/common.rs index 7a223cb..535ab44 100644 --- a/src/client/common.rs +++ b/src/client/common.rs @@ -2,6 +2,7 @@ use super::*; use super::access_token::{distrust_access_token, get_access_token}; use crate::config::{RenderMode, paths}; +use crate::hooks::{self, HookEvent}; use crate::{ config::{AppConfig, Input, RequestContext}, function::{FunctionDeclaration, ToolCall, ToolResult, eval_tool_calls}, @@ -20,7 +21,7 @@ use reqwest::{Client as ReqwestClient, RequestBuilder}; use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; use std::sync::LazyLock; -use std::time::Duration; +use std::time::{Duration, Instant}; use tokio::sync::mpsc::unbounded_channel; pub const MODELS_YAML: &str = include_str!("../../models.yaml"); @@ -546,6 +547,54 @@ pub async fn create_openai_compatible_client_config( Ok(Some((model, clients))) } +/// Base extras every `llm.request.*` event carries. The three transport +/// functions below are the single seam family for these events: they cover +/// every provider and every caller, so the `Client` trait and per-provider +/// code stay uninstrumented. +fn llm_request_extras(client: &dyn Client) -> Vec<(&'static str, String)> { + vec![ + ("COYOTE_LLM_PROVIDER", client.name().to_string()), + ("COYOTE_LLM_MODEL", client.model().id()), + ] +} + +fn llm_completed_extras(client: &dyn Client, started_at: Instant) -> Vec<(&'static str, String)> { + let mut extras = llm_request_extras(client); + extras.push(( + "COYOTE_LLM_DURATION_MS", + started_at.elapsed().as_millis().to_string(), + )); + extras +} + +/// `aborted` marks a user interrupt; otherwise the error is `api_error` when +/// it bottoms out in an [`ApiStatusError`] (which also yields +/// `COYOTE_LLM_STATUS`), else `other`. Abort paths that produce no error +/// object report `Aborted.`. +fn llm_failed_extras( + client: &dyn Client, + err: Option<&anyhow::Error>, + aborted: bool, +) -> Vec<(&'static str, String)> { + let mut extras = llm_request_extras(client); + if aborted { + extras.push(("COYOTE_LLM_ERROR_KIND", "aborted".to_string())); + } else if let Some(api_err) = err.and_then(|err| err.downcast_ref::()) { + extras.push(("COYOTE_LLM_ERROR_KIND", "api_error".to_string())); + extras.push(("COYOTE_LLM_STATUS", api_err.status.to_string())); + } else { + extras.push(("COYOTE_LLM_ERROR_KIND", "other".to_string())); + } + extras.push(( + "COYOTE_LLM_ERROR", + // `{:#}` renders the whole context chain, so the root cause survives + // the `context(...)` wrapping the client trait defaults apply. + err.map(|err| format!("{err:#}")) + .unwrap_or_else(|| "Aborted.".to_string()), + )); + extras +} + pub async fn call_chat_completions( input: &Input, print: bool, @@ -557,15 +606,28 @@ pub async fn call_chat_completions( let is_child_agent = ctx.current_depth > 0; let suppress_spinner = is_child_agent || ctx.render_mode == RenderMode::Silent; let spinner_message = if suppress_spinner { "" } else { "Generating" }; + hooks::fire( + HookEvent::LlmRequestStarted, + ctx, + &llm_request_extras(client), + None, + ); + let request_started_at = Instant::now(); let ret = abortable_run_with_spinner( client.chat_completions(input.clone()), spinner_message, - abort_signal, + abort_signal.clone(), ) .await; match ret { Ok(ret) => { + hooks::fire( + HookEvent::LlmRequestCompleted, + ctx, + &llm_completed_extras(client, request_started_at), + None, + ); let ChatCompletionsOutput { mut text, tool_calls, @@ -583,7 +645,15 @@ pub async fn call_chat_completions( ctx.record_token_usage(usage, client.model()); finish_completion(ctx, text, tool_calls, thinking).await } - Err(err) => Err(err), + Err(err) => { + hooks::fire( + HookEvent::LlmRequestFailed, + ctx, + &llm_failed_extras(client, Some(&err), abort_signal.aborted()), + None, + ); + Err(err) + } } } @@ -600,6 +670,13 @@ pub async fn call_chat_completions_streaming( handler.set_silent(true); } + hooks::fire( + HookEvent::LlmRequestStarted, + ctx, + &llm_request_extras(client), + None, + ); + let request_started_at = Instant::now(); let (send_ret, render_ret) = tokio::join!( client.chat_completions_streaming(input, &mut handler), render_stream(rx, client.app_config(), abort_signal.clone(), silent), @@ -615,12 +692,32 @@ pub async fn call_chat_completions_streaming( ctx.record_token_usage(usage, client.model()); if aborted_ctrld { + hooks::fire( + HookEvent::LlmRequestFailed, + ctx, + &llm_failed_extras(client, None, true), + None, + ); bail!("Aborted."); } - render_ret?; + if let Err(err) = render_ret { + hooks::fire( + HookEvent::LlmRequestFailed, + ctx, + &llm_failed_extras(client, Some(&err), false), + None, + ); + return Err(err); + } if aborted_ctrlc { + hooks::fire( + HookEvent::LlmRequestFailed, + ctx, + &llm_failed_extras(client, None, true), + None, + ); if !ctx.working_mode.is_repl() || ctx.session.is_none() { bail!("Aborted."); } @@ -644,6 +741,12 @@ pub async fn call_chat_completions_streaming( match send_ret { Ok(_) => { + hooks::fire( + HookEvent::LlmRequestCompleted, + ctx, + &llm_completed_extras(client, request_started_at), + None, + ); if !silent && !text.is_empty() && !text.ends_with('\n') { println!(); } @@ -653,6 +756,12 @@ pub async fn call_chat_completions_streaming( if !silent && !text.is_empty() { println!(); } + hooks::fire( + HookEvent::LlmRequestFailed, + ctx, + &llm_failed_extras(client, Some(&err), false), + None, + ); Err(err) } } @@ -675,10 +784,36 @@ pub async fn call_chat_completions_streaming_quiet( ctx: &mut RequestContext, abort_signal: AbortSignal, ) -> Result<(String, Vec)> { + hooks::fire( + HookEvent::LlmRequestStarted, + ctx, + &llm_request_extras(client), + None, + ); + let request_started_at = Instant::now(); if client.model().no_stream() { let output = tokio::select! { - ret = client.chat_completions(input.clone()) => ret?, - _ = wait_abort_signal(&abort_signal) => bail!("Aborted."), + ret = client.chat_completions(input.clone()) => match ret { + Ok(output) => output, + Err(err) => { + hooks::fire( + HookEvent::LlmRequestFailed, + ctx, + &llm_failed_extras(client, Some(&err), false), + None, + ); + return Err(err); + } + }, + _ = wait_abort_signal(&abort_signal) => { + hooks::fire( + HookEvent::LlmRequestFailed, + ctx, + &llm_failed_extras(client, None, true), + None, + ); + bail!("Aborted.") + } }; let ChatCompletionsOutput { text, @@ -688,8 +823,20 @@ pub async fn call_chat_completions_streaming_quiet( } = output; ctx.record_token_usage(usage, client.model()); if abort_signal.aborted() { + hooks::fire( + HookEvent::LlmRequestFailed, + ctx, + &llm_failed_extras(client, None, true), + None, + ); bail!("Aborted."); } + hooks::fire( + HookEvent::LlmRequestCompleted, + ctx, + &llm_completed_extras(client, request_started_at), + None, + ); return finish_completion(ctx, text, tool_calls, thinking).await; } @@ -705,10 +852,30 @@ pub async fn call_chat_completions_streaming_quiet( ctx.record_token_usage(usage, client.model()); if abort_signal.aborted() { + hooks::fire( + HookEvent::LlmRequestFailed, + ctx, + &llm_failed_extras(client, None, true), + None, + ); bail!("Aborted."); } - send_ret?; + if let Err(err) = send_ret { + hooks::fire( + HookEvent::LlmRequestFailed, + ctx, + &llm_failed_extras(client, Some(&err), false), + None, + ); + return Err(err); + } + hooks::fire( + HookEvent::LlmRequestCompleted, + ctx, + &llm_completed_extras(client, request_started_at), + None, + ); finish_completion(ctx, text, tool_calls, thinking).await } @@ -1524,4 +1691,331 @@ mod tests { assert_eq!(client.streaming_calls.load(Ordering::SeqCst), 1); assert_paced_reply(&text, &results); } + + /// Fake provider whose every request fails with an [`ApiStatusError`], + /// wrapped in the same context chain the real trait defaults produce. + struct FailingStatusClient { + config: AppConfig, + model: Model, + status: u16, + } + + impl FailingStatusClient { + fn new(status: u16) -> Self { + Self { + config: AppConfig::default(), + model: Model::default(), + status, + } + } + + fn error(&self) -> anyhow::Error { + anyhow::Error::new(ApiStatusError { + status: self.status, + message: format!("error (status: {})", self.status), + }) + } + } + + #[async_trait::async_trait] + impl Client for FailingStatusClient { + fn app_config(&self) -> &AppConfig { + &self.config + } + + fn extra_config(&self) -> Option<&ExtraConfig> { + None + } + + fn patch_config(&self) -> Option<&RequestPatch> { + None + } + + fn name(&self) -> &str { + "failing-status" + } + + fn model(&self) -> &Model { + &self.model + } + + async fn chat_completions_inner( + &self, + _client: &ReqwestClient, + _data: ChatCompletionsData, + ) -> Result { + Err(self.error()) + } + + async fn chat_completions_streaming_inner( + &self, + _client: &ReqwestClient, + _handler: &mut SseHandler, + _data: ChatCompletionsData, + ) -> Result<()> { + Err(self.error()) + } + } + + fn llm_hooks_map(marker: &str) -> crate::hooks::HooksMap { + [ + "llm.request.started", + "llm.request.completed", + "llm.request.failed", + ] + .into_iter() + .map(|event| { + ( + event.to_string(), + vec![crate::hooks::HookDef { + name: marker.to_string(), + command: "true".to_string(), + }], + ) + }) + .collect() + } + + fn llm_hooked_ctx(marker: &str) -> RequestContext { + let mut app = AppState::test_default(); + app.config = Arc::new(AppConfig { + hooks: llm_hooks_map(marker), + ..Default::default() + }); + let mut ctx = RequestContext::new(Arc::new(app), WorkingMode::Cmd); + ctx.render_mode = RenderMode::Silent; + ctx + } + + fn llm_captures(marker: &str) -> Vec { + crate::hooks::test_sink::snapshot() + .into_iter() + .filter(|capture| capture.hook_name == marker) + .collect() + } + + #[tokio::test(start_paused = true)] + #[serial_test::serial] + async fn plain_transport_fires_llm_request_started_and_completed() { + let marker = "llm-hooks-plain-ok-x7x"; + let mut ctx = llm_hooked_ctx(marker); + let input = Input::from_str(&ctx, "hi", None).unwrap(); + let mut client = paced_client(false); + client.nonstream_replies = true; + let _guard = crate::hooks::test_sink::install(); + + call_chat_completions( + &input, + false, + false, + &client, + &mut ctx, + create_abort_signal(), + ) + .await + .unwrap(); + + let captures = llm_captures(marker); + assert_eq!(captures.len(), 2, "{captures:?}"); + assert_eq!(captures[0].event, HookEvent::LlmRequestStarted); + assert_eq!( + captures[0] + .envs + .get("COYOTE_LLM_PROVIDER") + .map(String::as_str), + Some("paced-stream") + ); + assert!(captures[0].envs.contains_key("COYOTE_LLM_MODEL")); + assert_eq!(captures[1].event, HookEvent::LlmRequestCompleted); + assert!(captures[1].envs.contains_key("COYOTE_LLM_DURATION_MS")); + assert!(!captures[1].envs.contains_key("COYOTE_LLM_ERROR_KIND")); + } + + #[tokio::test] + #[serial_test::serial] + async fn plain_transport_fires_llm_request_failed_with_api_error_kind() { + let marker = "llm-hooks-plain-api-err-x7x"; + let mut ctx = llm_hooked_ctx(marker); + let input = Input::from_str(&ctx, "hi", None).unwrap(); + let client = FailingStatusClient::new(429); + let _guard = crate::hooks::test_sink::install(); + + let err = call_chat_completions( + &input, + false, + false, + &client, + &mut ctx, + create_abort_signal(), + ) + .await + .unwrap_err(); + + assert!( + err.downcast_ref::().is_some(), + "hooks must not change the error: {err:?}" + ); + let captures = llm_captures(marker); + assert_eq!(captures.len(), 2, "{captures:?}"); + assert_eq!(captures[0].event, HookEvent::LlmRequestStarted); + let failed = &captures[1]; + assert_eq!(failed.event, HookEvent::LlmRequestFailed); + assert_eq!( + failed.envs.get("COYOTE_LLM_ERROR_KIND").map(String::as_str), + Some("api_error") + ); + assert_eq!( + failed.envs.get("COYOTE_LLM_STATUS").map(String::as_str), + Some("429") + ); + assert!( + failed + .envs + .get("COYOTE_LLM_ERROR") + .unwrap() + .contains("error (status: 429)") + ); + assert!(!failed.envs.contains_key("COYOTE_LLM_DURATION_MS")); + } + + #[tokio::test(start_paused = true)] + #[serial_test::serial] + async fn streaming_transport_fires_llm_request_started_and_completed() { + let marker = "llm-hooks-stream-ok-x7x"; + let mut ctx = llm_hooked_ctx(marker); + let input = Input::from_str(&ctx, "hi", None).unwrap(); + let client = paced_client(false); + let _guard = crate::hooks::test_sink::install(); + + call_chat_completions_streaming(&input, &client, &mut ctx, create_abort_signal()) + .await + .unwrap(); + + let captures = llm_captures(marker); + assert_eq!(captures.len(), 2, "{captures:?}"); + assert_eq!(captures[0].event, HookEvent::LlmRequestStarted); + assert_eq!(captures[1].event, HookEvent::LlmRequestCompleted); + assert!(captures[1].envs.contains_key("COYOTE_LLM_DURATION_MS")); + } + + #[tokio::test(start_paused = true)] + #[serial_test::serial] + async fn streaming_transport_abort_fires_llm_request_failed_aborted() { + let marker = "llm-hooks-stream-abort-x7x"; + let mut ctx = llm_hooked_ctx(marker); + let input = Input::from_str(&ctx, "hi", None).unwrap(); + let client = paced_client(true); + let abort = create_abort_signal(); + let trigger = abort.clone(); + tokio::spawn(async move { + tokio::time::sleep(PACED_GAP * PACED_DELTAS as u32 + Duration::from_secs(30)).await; + trigger.set_ctrlc(); + }); + let _guard = crate::hooks::test_sink::install(); + + let err = call_chat_completions_streaming(&input, &client, &mut ctx, abort) + .await + .expect_err("an aborted stream must not succeed"); + + assert_eq!(err.to_string(), "Aborted."); + let captures = llm_captures(marker); + assert_eq!(captures.len(), 2, "{captures:?}"); + let failed = &captures[1]; + assert_eq!(failed.event, HookEvent::LlmRequestFailed); + assert_eq!( + failed.envs.get("COYOTE_LLM_ERROR_KIND").map(String::as_str), + Some("aborted") + ); + assert!(!failed.envs.contains_key("COYOTE_LLM_STATUS")); + } + + #[tokio::test(start_paused = true)] + #[serial_test::serial] + async fn quiet_transport_fires_llm_request_started_and_completed() { + let marker = "llm-hooks-quiet-ok-x7x"; + let mut ctx = llm_hooked_ctx(marker); + let input = Input::from_str(&ctx, "hi", None).unwrap(); + let client = paced_client(false); + let _guard = crate::hooks::test_sink::install(); + + call_chat_completions_streaming_quiet(&input, &client, &mut ctx, create_abort_signal()) + .await + .unwrap(); + + let captures = llm_captures(marker); + assert_eq!(captures.len(), 2, "{captures:?}"); + assert_eq!(captures[0].event, HookEvent::LlmRequestStarted); + assert_eq!( + captures[0] + .envs + .get("COYOTE_LLM_PROVIDER") + .map(String::as_str), + Some("paced-stream") + ); + assert_eq!(captures[1].event, HookEvent::LlmRequestCompleted); + assert!(captures[1].envs.contains_key("COYOTE_LLM_DURATION_MS")); + } + + #[tokio::test(start_paused = true)] + #[serial_test::serial] + async fn quiet_transport_abort_fires_llm_request_failed_aborted() { + let marker = "llm-hooks-quiet-abort-x7x"; + let mut ctx = llm_hooked_ctx(marker); + let input = Input::from_str(&ctx, "hi", None).unwrap(); + let client = paced_client(true); + let abort = create_abort_signal(); + let trigger = abort.clone(); + tokio::spawn(async move { + tokio::time::sleep(PACED_GAP * PACED_DELTAS as u32 + Duration::from_secs(30)).await; + trigger.set_ctrlc(); + }); + let _guard = crate::hooks::test_sink::install(); + + let err = call_chat_completions_streaming_quiet(&input, &client, &mut ctx, abort) + .await + .expect_err("an aborted stream must not succeed"); + + assert_eq!(err.to_string(), "Aborted."); + let captures = llm_captures(marker); + assert_eq!(captures.len(), 2, "{captures:?}"); + let failed = &captures[1]; + assert_eq!(failed.event, HookEvent::LlmRequestFailed); + assert_eq!( + failed.envs.get("COYOTE_LLM_ERROR_KIND").map(String::as_str), + Some("aborted") + ); + assert_eq!( + failed.envs.get("COYOTE_LLM_ERROR").map(String::as_str), + Some("Aborted.") + ); + } + + #[tokio::test] + #[serial_test::serial] + async fn quiet_transport_fires_llm_request_failed_with_api_error_kind() { + let marker = "llm-hooks-quiet-api-err-x7x"; + let mut ctx = llm_hooked_ctx(marker); + let input = Input::from_str(&ctx, "hi", None).unwrap(); + let client = FailingStatusClient::new(500); + let _guard = crate::hooks::test_sink::install(); + + let err = + call_chat_completions_streaming_quiet(&input, &client, &mut ctx, create_abort_signal()) + .await + .unwrap_err(); + + assert!(err.downcast_ref::().is_some()); + let captures = llm_captures(marker); + assert_eq!(captures.len(), 2, "{captures:?}"); + let failed = &captures[1]; + assert_eq!(failed.event, HookEvent::LlmRequestFailed); + assert_eq!( + failed.envs.get("COYOTE_LLM_ERROR_KIND").map(String::as_str), + Some("api_error") + ); + assert_eq!( + failed.envs.get("COYOTE_LLM_STATUS").map(String::as_str), + Some("500") + ); + } } diff --git a/src/function/jobs.rs b/src/function/jobs.rs index e8e490f..7aaadf2 100644 --- a/src/function/jobs.rs +++ b/src/function/jobs.rs @@ -9,6 +9,7 @@ use crate::config::{ McpRuntime, RequestContext, effective_max_concurrent_jobs, jobs_enabled, paths, }; use crate::graph; +use crate::hooks::{self, HookEvent, ResolvedHook}; use crate::mcp::{ MCP_DESCRIBE_META_FUNCTION_NAME_PREFIX, MCP_INVOKE_META_FUNCTION_NAME_PREFIX, MCP_PROMPT_META_FUNCTION_NAME_PREFIX, MCP_READ_META_FUNCTION_NAME_PREFIX, @@ -21,6 +22,7 @@ use crate::utils::{ }; use anyhow::{Context, Result, anyhow, bail}; +use chrono::{SecondsFormat, Utc}; use indexmap::IndexMap; use parking_lot::{Mutex, RwLock}; use serde_json::{Value, json}; @@ -53,6 +55,75 @@ pub fn is_agent_task(supervisor: Option<&Arc>>, id: &str) -> || supervisor.is_some_and(|sup| sup.read().has_agent(id)) } +/// Terminal-event hooks for one job, resolved on the caller's context before +/// the task is spawned (`build_env_snapshot` precedent) so the detached +/// closure owns everything it fires with. Session and agent names are +/// snapshotted because the context is gone by fire time. +struct JobHookSnapshot { + completed: Vec, + failed: Vec, + session_name: Option, + agent_name: Option, + job_id: String, + tool: String, +} + +impl JobHookSnapshot { + fn resolve(ctx: &RequestContext, job_id: &str, tool: &str) -> Self { + Self { + completed: ctx.resolved_hooks(HookEvent::JobCompleted), + failed: ctx.resolved_hooks(HookEvent::JobFailed), + session_name: ctx + .session + .as_ref() + .map(|session| session.name().to_string()), + agent_name: ctx.agent.as_ref().map(|agent| agent.name().to_string()), + job_id: job_id.to_string(), + tool: tool.to_string(), + } + } + + /// Fires `job.completed` or `job.failed`. `fire_resolved` callers supply + /// their own base envs, so this mirrors the engine's assembly; the event + /// name and timestamp are computed here so they reflect the actual + /// outcome rather than the moment the snapshot was taken. + fn fire_terminal(self, success: bool, error: Option) { + let (event, resolved) = if success { + (HookEvent::JobCompleted, self.completed) + } else { + (HookEvent::JobFailed, self.failed) + }; + if resolved.is_empty() { + return; + } + let mut base_envs = vec![ + ("COYOTE_EVENT".to_string(), event.as_str().to_string()), + ( + "COYOTE_CONFIG_DIR".to_string(), + paths::config_dir().display().to_string(), + ), + ( + "COYOTE_EVENT_TIMESTAMP".to_string(), + Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true), + ), + ]; + if let Some(name) = self.session_name { + base_envs.push(("COYOTE_SESSION_ID".to_string(), name)); + } + if let Some(name) = self.agent_name { + base_envs.push(("COYOTE_AGENT_NAME".to_string(), name)); + } + let mut extras = vec![ + ("COYOTE_JOB_ID", self.job_id), + ("COYOTE_TOOL_NAME", self.tool), + ]; + if let Some(error) = error { + extras.push(("COYOTE_JOB_ERROR", error)); + } + hooks::fire_resolved(event, resolved, base_envs, &extras, None); + } +} + pub struct RingBuf { buf: Vec, capacity: usize, @@ -470,15 +541,27 @@ async fn handle_start(ctx: &mut RequestContext, args: &Value) -> Result { let task_notifications = Arc::clone(&ctx.notification_queue); let notify_id = job_id.clone(); let notify_tool = tool.clone(); + let hook_snapshot = JobHookSnapshot::resolve(ctx, &job_id, &tool); + hooks::fire( + HookEvent::JobStarted, + ctx, + &[ + ("COYOTE_JOB_ID", job_id.clone()), + ("COYOTE_TOOL_NAME", tool.clone()), + ], + None, + ); tokio::spawn(async move { let result = run_mcp_job(job_ctx, server, inner_tool, inner_args).await; let success = result.is_ok(); + let error = result.as_ref().err().map(|err| err.to_string()); task_state.lock().status = if success { JobStatus::Completed } else { JobStatus::Failed }; task_notifications.push(job_notification(¬ify_id, ¬ify_tool, success)); + hook_snapshot.fire_terminal(success, error); result }) } else { @@ -488,9 +571,26 @@ async fn handle_start(ctx: &mut RequestContext, args: &Value) -> Result { let task_notifications = Arc::clone(&ctx.notification_queue); let notify_id = job_id.clone(); let notify_tool = tool.clone(); + let hook_snapshot = JobHookSnapshot::resolve(ctx, &job_id, &tool); + hooks::fire( + HookEvent::JobStarted, + ctx, + &[ + ("COYOTE_JOB_ID", job_id.clone()), + ("COYOTE_TOOL_NAME", tool.clone()), + ], + None, + ); tokio::spawn(async move { let result = run_process_job(snapshot, Arc::clone(&task_state), task_buf).await; let success = matches!(&result, Ok(job_result) if job_result.exit_code == Some(0)); + let error = (!success).then(|| match &result { + Err(err) => err.to_string(), + Ok(job_result) => match job_result.exit_code { + Some(code) => format!("exited with code {code}"), + None => "terminated without an exit code".to_string(), + }, + }); let mut job_state = task_state.lock(); job_state.pgid = None; job_state.status = if success { @@ -502,6 +602,7 @@ async fn handle_start(ctx: &mut RequestContext, args: &Value) -> Result { drop(job_state); task_notifications.push(job_notification(¬ify_id, ¬ify_tool, success)); + hook_snapshot.fire_terminal(success, error); result }) }; @@ -3119,4 +3220,190 @@ mod tests { ); assert!(ctx.supervisor.is_none(), "no job may be spawned"); } + + fn job_hooks_map(marker: &str) -> crate::hooks::HooksMap { + ["job.started", "job.completed", "job.failed"] + .into_iter() + .map(|event| { + ( + event.to_string(), + vec![crate::hooks::HookDef { + name: marker.to_string(), + command: "true".to_string(), + }], + ) + }) + .collect() + } + + fn job_hooked_ctx(marker: &str) -> RequestContext { + let app = app_state_with_config(|config| { + config.hooks = job_hooks_map(marker); + }); + RequestContext::new(app, WorkingMode::Cmd) + } + + fn job_captures(marker: &str) -> Vec { + crate::hooks::test_sink::snapshot() + .into_iter() + .filter(|capture| capture.hook_name == marker) + .collect() + } + + #[cfg(unix)] + #[test] + #[serial_test::serial] + fn job_hooks_fire_started_and_completed() { + run_async(async { + let marker = "job-hooks-ok-x7x"; + let mut ctx = job_hooked_ctx(marker); + ctx.declared_function_names.insert("echo".into()); + let _guard = crate::hooks::test_sink::install(); + + let started = handle_start(&mut ctx, &json!({"tool": "echo", "arguments": {}})) + .await + .unwrap(); + assert_eq!(started["status"], "ok"); + let job_id = started["job_id"].as_str().unwrap().to_string(); + + // `job.started` is dispatched before `handle_start` returns. + let captures = job_captures(marker); + assert_eq!(captures.len(), 1, "{captures:?}"); + assert_eq!(captures[0].event, HookEvent::JobStarted); + assert_eq!( + captures[0].envs.get("COYOTE_JOB_ID").map(String::as_str), + Some(job_id.as_str()) + ); + assert_eq!( + captures[0].envs.get("COYOTE_TOOL_NAME").map(String::as_str), + Some("echo") + ); + + let collected = handle_collect(&ctx, &json!({"id": job_id})).await.unwrap(); + assert_eq!(collected["status"], "completed"); + + let captures = job_captures(marker); + assert_eq!(captures.len(), 2, "{captures:?}"); + let completed = &captures[1]; + assert_eq!(completed.event, HookEvent::JobCompleted); + assert_eq!( + completed.envs.get("COYOTE_EVENT").map(String::as_str), + Some("job.completed") + ); + assert_eq!( + completed.envs.get("COYOTE_JOB_ID").map(String::as_str), + Some(job_id.as_str()) + ); + assert_eq!( + completed.envs.get("COYOTE_TOOL_NAME").map(String::as_str), + Some("echo") + ); + assert!(completed.envs.contains_key("COYOTE_CONFIG_DIR")); + assert!(completed.envs.contains_key("COYOTE_EVENT_TIMESTAMP")); + assert!(!completed.envs.contains_key("COYOTE_SESSION_ID")); + assert!(!completed.envs.contains_key("COYOTE_JOB_ERROR")); + }); + } + + #[cfg(unix)] + #[test] + #[serial_test::serial] + fn job_hooks_fire_failed_with_job_error() { + run_async(async { + let marker = "job-hooks-fail-x7x"; + let mut ctx = job_hooked_ctx(marker); + ctx.declared_function_names.insert("false".into()); + let _guard = crate::hooks::test_sink::install(); + + let started = handle_start(&mut ctx, &json!({"tool": "false", "arguments": {}})) + .await + .unwrap(); + let job_id = started["job_id"].as_str().unwrap().to_string(); + + let collected = handle_collect(&ctx, &json!({"id": job_id})).await.unwrap(); + assert_eq!(collected["status"], "failed"); + + let captures = job_captures(marker); + assert_eq!(captures.len(), 2, "{captures:?}"); + assert_eq!(captures[0].event, HookEvent::JobStarted); + let failed = &captures[1]; + assert_eq!(failed.event, HookEvent::JobFailed); + assert_eq!( + failed.envs.get("COYOTE_EVENT").map(String::as_str), + Some("job.failed") + ); + assert_eq!( + failed.envs.get("COYOTE_JOB_ID").map(String::as_str), + Some(job_id.as_str()) + ); + assert_eq!( + failed.envs.get("COYOTE_JOB_ERROR").map(String::as_str), + Some("exited with code 1") + ); + }); + } + + #[test] + #[serial_test::serial] + fn job_hooks_fire_for_mcp_jobs() { + run_async(async { + let marker = "job-hooks-mcp-x7x"; + let mut ctx = job_hooked_ctx(marker); + ctx.declared_function_names + .insert("mcp_invoke_fixture".into()); + let fixture = FixtureServer::default(); + let (mut runtime, _server) = fixture_runtime(fixture).await; + let mut filter = ToolFilter::default(); + filter.push_layer(LayerSource::Global, &[]); + runtime.tool_filters.insert("fixture".to_string(), filter); + ctx.tool_scope.mcp_runtime = runtime; + let _guard = crate::hooks::test_sink::install(); + + let started = handle_start( + &mut ctx, + &json!({"tool": "mcp_invoke_fixture", "arguments": {"tool": "dup"}}), + ) + .await + .unwrap(); + let job_id = started["job_id"].as_str().unwrap().to_string(); + + let collected = handle_collect(&ctx, &json!({"id": job_id})).await.unwrap(); + assert_eq!(collected["status"], "failed"); + + let captures = job_captures(marker); + assert_eq!(captures.len(), 2, "{captures:?}"); + assert_eq!(captures[0].event, HookEvent::JobStarted); + assert_eq!( + captures[0].envs.get("COYOTE_TOOL_NAME").map(String::as_str), + Some("mcp_invoke_fixture") + ); + let failed = &captures[1]; + assert_eq!(failed.event, HookEvent::JobFailed); + assert!( + failed + .envs + .get("COYOTE_JOB_ERROR") + .unwrap() + .contains("dup not found"), + "{failed:?}" + ); + }); + } + + #[test] + #[serial_test::serial] + fn rejected_job_start_fires_no_hooks() { + let marker = "job-hooks-rejected-x7x"; + let mut ctx = job_hooked_ctx(marker); + let _guard = crate::hooks::test_sink::install(); + + let result = run_async(handle_start( + &mut ctx, + &json!({"tool": "undeclared_tool", "arguments": {}}), + )) + .unwrap(); + + assert_eq!(result["status"], "error"); + assert!(job_captures(marker).is_empty()); + } } diff --git a/src/function/mod.rs b/src/function/mod.rs index c2b73dc..3d5ecd0 100644 --- a/src/function/mod.rs +++ b/src/function/mod.rs @@ -17,6 +17,7 @@ use crate::{ use crate::config::ensure_parent_exists; use crate::config::paths; +use crate::hooks::{self, HookEvent}; use crate::mcp::{ MCP_DESCRIBE_META_FUNCTION_NAME_PREFIX, MCP_INVOKE_META_FUNCTION_NAME_PREFIX, MCP_META_FUNCTION_PREFIXES, MCP_PROMPT_META_FUNCTION_NAME_PREFIX, @@ -1538,7 +1539,29 @@ impl ToolCall { Ok(result) } + /// Evaluates the call, firing `tool.*` hooks around it: `tool.started` + /// once the name and arguments are resolved, then exactly one of + /// `tool.completed` / `tool.failed`. A failure before resolution + /// (unknown tool, malformed arguments) fires `tool.failed` alone. Hooks + /// are pure observers — the returned value is exactly what the tool + /// produced. pub async fn eval(&self, ctx: &mut RequestContext) -> Result { + let result = self.eval_inner(ctx).await; + if let Err(err) = &result { + hooks::fire( + HookEvent::ToolFailed, + ctx, + &[ + ("COYOTE_TOOL_NAME", self.name.clone()), + ("COYOTE_TOOL_ERROR", err.to_string()), + ], + None, + ); + } + result + } + + async fn eval_inner(&self, ctx: &mut RequestContext) -> Result { let agent = ctx.agent.clone(); let functions = ctx.tool_scope.functions.clone(); let current_depth = ctx.current_depth; @@ -1566,7 +1589,8 @@ impl ToolCall { ); }; - cmd_args.push(json_data.to_string()); + let args_json = json_data.to_string(); + cmd_args.push(args_json.clone()); if (*IS_STDOUT_TERMINAL || *SHOW_TOOL_CALLS) && current_depth == 0 @@ -1575,6 +1599,14 @@ impl ToolCall { println!("{}", format_call_log(&cmd_name, &cmd_args, &json_data)); } + hooks::fire( + HookEvent::ToolStarted, + ctx, + &[("COYOTE_TOOL_NAME", self.name.clone())], + Some(args_json.clone()), + ); + let tool_started_at = Instant::now(); + let output = match cmd_name.as_str() { _ if cmd_name.starts_with(MCP_SEARCH_META_FUNCTION_NAME_PREFIX) => { Self::search_mcp_tools(ctx, &cmd_name, &json_data) @@ -1699,6 +1731,38 @@ impl ToolCall { }, }; + match output.get("tool_call_error") { + Some(error) => { + let error = error + .as_str() + .map(str::to_string) + .unwrap_or_else(|| error.to_string()); + hooks::fire( + HookEvent::ToolFailed, + ctx, + &[ + ("COYOTE_TOOL_NAME", self.name.clone()), + ("COYOTE_TOOL_ERROR", error), + ], + Some(args_json), + ); + } + None => { + hooks::fire( + HookEvent::ToolCompleted, + ctx, + &[ + ("COYOTE_TOOL_NAME", self.name.clone()), + ( + "COYOTE_TOOL_DURATION_MS", + tool_started_at.elapsed().as_millis().to_string(), + ), + ], + Some(args_json), + ); + } + } + Ok(output) } @@ -5243,4 +5307,241 @@ mod tests { assert!(err.starts_with(expected), "{name}: {err}"); } } + + fn tool_hooks_map(marker: &str) -> crate::hooks::HooksMap { + ["tool.started", "tool.completed", "tool.failed"] + .into_iter() + .map(|event| { + ( + event.to_string(), + vec![crate::hooks::HookDef { + name: marker.to_string(), + command: "true".to_string(), + }], + ) + }) + .collect() + } + + fn ctx_with_tool_hooks(marker: &str, update: impl FnOnce(&mut AppConfig)) -> RequestContext { + let mut app = AppState::test_default(); + let mut config = AppConfig { + hooks: tool_hooks_map(marker), + ..Default::default() + }; + update(&mut config); + app.config = Arc::new(config); + RequestContext::new(Arc::new(app), WorkingMode::Cmd) + } + + fn tool_captures(marker: &str) -> Vec { + crate::hooks::test_sink::snapshot() + .into_iter() + .filter(|capture| capture.hook_name == marker) + .collect() + } + + #[test] + #[serial] + fn eval_fires_started_and_completed_for_builtin_tools() { + use crate::hooks::HookEvent; + let marker = "tool-hooks-builtin-ok-x7x"; + let mut ctx = ctx_with_tool_hooks(marker, |config| { + config.function_calling_support = true; + config.auto_continue = true; + }); + ctx.tool_scope.functions.append_todo_functions(); + let _guard = crate::hooks::test_sink::install(); + + let output = + run_async(call_with_args("todo__init", json!({"goal": "g"})).eval(&mut ctx)).unwrap(); + + assert_eq!(output["status"], "ok"); + let captures = tool_captures(marker); + assert_eq!(captures.len(), 2, "{captures:?}"); + assert_eq!(captures[0].event, HookEvent::ToolStarted); + assert_eq!( + captures[0].envs.get("COYOTE_TOOL_NAME").map(String::as_str), + Some("todo__init") + ); + assert_eq!(captures[0].payload.as_deref(), Some(r#"{"goal":"g"}"#)); + assert_eq!(captures[0].cwd, paths::config_dir()); + assert_eq!(captures[1].event, HookEvent::ToolCompleted); + assert!(captures[1].envs.contains_key("COYOTE_TOOL_DURATION_MS")); + assert_eq!(captures[1].payload.as_deref(), Some(r#"{"goal":"g"}"#)); + } + + #[test] + #[serial] + fn eval_fires_started_and_completed_for_mcp_meta_tools() { + use crate::hooks::HookEvent; + let marker = "tool-hooks-mcp-ok-x7x"; + run_async(async { + let mut result = CallToolResult::success(vec![ContentBlock::text("hi")]); + result.structured_content = Some(json!({"ok": true})); + let fixture = FixtureServer { + tool_result: Some(result), + ..Default::default() + }; + let (runtime, _server) = fixture_runtime(fixture).await; + let mut ctx = ctx_with_tool_hooks(marker, |_| {}); + ctx.tool_scope.mcp_runtime = runtime; + ctx.tool_scope + .functions + .append_mcp_meta_functions(vec![tools_only("fixture")]); + let _guard = crate::hooks::test_sink::install(); + + let output = call_with_args("mcp_invoke_fixture", json!({"tool": "dup"})) + .eval(&mut ctx) + .await + .unwrap(); + + assert_eq!(output["isError"], false); + let captures = tool_captures(marker); + assert_eq!(captures.len(), 2, "{captures:?}"); + assert_eq!(captures[0].event, HookEvent::ToolStarted); + assert_eq!( + captures[0].envs.get("COYOTE_TOOL_NAME").map(String::as_str), + Some("mcp_invoke_fixture") + ); + assert_eq!(captures[0].payload.as_deref(), Some(r#"{"tool":"dup"}"#)); + assert_eq!(captures[1].event, HookEvent::ToolCompleted); + }); + } + + #[cfg(unix)] + #[test] + #[serial] + fn eval_fires_started_and_completed_for_external_command_tools() { + use crate::hooks::HookEvent; + let marker = "tool-hooks-external-ok-x7x"; + let mut ctx = ctx_with_tool_hooks(marker, |_| {}); + ctx.tool_scope + .functions + .append_declaration(FunctionDeclaration { + name: "echo".to_string(), + description: String::new(), + parameters: JsonSchema::default(), + agent: false, + }); + let _guard = crate::hooks::test_sink::install(); + + let output = run_async(call_with_args("echo", json!({"x": 1})).eval(&mut ctx)).unwrap(); + + assert_eq!(output, Value::Null); + let captures = tool_captures(marker); + assert_eq!(captures.len(), 2, "{captures:?}"); + assert_eq!(captures[0].event, HookEvent::ToolStarted); + assert_eq!( + captures[0].envs.get("COYOTE_TOOL_NAME").map(String::as_str), + Some("echo") + ); + assert_eq!(captures[1].event, HookEvent::ToolCompleted); + } + + #[cfg(unix)] + #[test] + #[serial] + fn eval_fires_failed_for_external_command_nonzero_exit() { + use crate::hooks::HookEvent; + let marker = "tool-hooks-external-fail-x7x"; + let mut ctx = ctx_with_tool_hooks(marker, |_| {}); + ctx.tool_scope + .functions + .append_declaration(FunctionDeclaration { + name: "false".to_string(), + description: String::new(), + parameters: JsonSchema::default(), + agent: false, + }); + let _guard = crate::hooks::test_sink::install(); + + let output = run_async(call_with_args("false", json!({})).eval(&mut ctx)).unwrap(); + + assert!(output["tool_call_error"].is_string()); + let captures = tool_captures(marker); + assert_eq!(captures.len(), 2, "{captures:?}"); + assert_eq!(captures[0].event, HookEvent::ToolStarted); + assert_eq!(captures[1].event, HookEvent::ToolFailed); + assert!( + captures[1] + .envs + .get("COYOTE_TOOL_ERROR") + .unwrap() + .contains("exited with code 1") + ); + assert!(!captures[1].envs.contains_key("COYOTE_TOOL_DURATION_MS")); + } + + #[test] + #[serial] + fn eval_fires_failed_on_error_shaped_output_with_identical_result() { + use crate::hooks::HookEvent; + let marker = "tool-hooks-error-shape-x7x"; + // Default config leaves todo tools disabled, so the handler bails + // and eval shapes the failure as a `tool_call_error` output. + let mut hooked = ctx_with_tool_hooks(marker, |_| {}); + hooked.tool_scope.functions.append_todo_functions(); + let mut plain = RequestContext::new(Arc::new(AppState::test_default()), WorkingMode::Cmd); + plain.tool_scope.functions.append_todo_functions(); + let call = call_with_args("todo__done", json!({})); + let _guard = crate::hooks::test_sink::install(); + + let hooked_output = run_async(call.eval(&mut hooked)).unwrap(); + let plain_output = run_async(call.eval(&mut plain)).unwrap(); + + assert_eq!( + hooked_output.to_string(), + plain_output.to_string(), + "hooks must not change the tool result" + ); + let error = hooked_output["tool_call_error"].as_str().unwrap(); + assert!(error.starts_with("Todo tool failed:"), "{error}"); + let captures = tool_captures(marker); + assert_eq!(captures.len(), 2, "{captures:?}"); + assert_eq!(captures[0].event, HookEvent::ToolStarted); + assert_eq!(captures[1].event, HookEvent::ToolFailed); + assert_eq!( + captures[1] + .envs + .get("COYOTE_TOOL_ERROR") + .map(String::as_str), + Some(error) + ); + } + + #[test] + #[serial] + fn eval_err_fires_failed_without_started_and_identical_error() { + use crate::hooks::HookEvent; + let marker = "tool-hooks-eval-err-x7x"; + let mut hooked = ctx_with_tool_hooks(marker, |_| {}); + let mut plain = RequestContext::new(Arc::new(AppState::test_default()), WorkingMode::Cmd); + let call = call_with_args("nonexistent_tool", json!({})); + let _guard = crate::hooks::test_sink::install(); + + let hooked_err = run_async(call.eval(&mut hooked)).unwrap_err(); + let plain_err = run_async(call.eval(&mut plain)).unwrap_err(); + + assert_eq!( + hooked_err.to_string(), + plain_err.to_string(), + "hooks must not change the error" + ); + let captures = tool_captures(marker); + assert_eq!(captures.len(), 1, "{captures:?}"); + assert_eq!(captures[0].event, HookEvent::ToolFailed); + assert_eq!( + captures[0].envs.get("COYOTE_TOOL_NAME").map(String::as_str), + Some("nonexistent_tool") + ); + assert!( + captures[0] + .envs + .get("COYOTE_TOOL_ERROR") + .unwrap() + .contains("Unexpected call") + ); + assert!(captures[0].payload.is_none()); + } } diff --git a/src/hooks.rs b/src/hooks.rs index 3f1d7b1..23f3c52 100644 --- a/src/hooks.rs +++ b/src/hooks.rs @@ -248,7 +248,7 @@ pub fn fire_resolved( .collect(); #[cfg(test)] - if test_sink::record(event, &hook, &envs) { + if test_sink::record(event, &hook, &envs, payload.as_deref()) { continue; } @@ -496,6 +496,11 @@ pub(crate) mod test_sink { pub event: HookEvent, pub hook_name: String, pub envs: HashMap, + /// The tool-arguments JSON as handed to dispatch; `None` for events + /// that carry no payload. + pub payload: Option, + /// The working directory the hook would have spawned in. + pub cwd: std::path::PathBuf, } /// Suppresses process spawning and records every dispatch synchronously @@ -518,7 +523,12 @@ pub(crate) mod test_sink { } } - pub(super) fn record(event: HookEvent, hook: &ResolvedHook, envs: &[(String, String)]) -> bool { + pub(super) fn record( + event: HookEvent, + hook: &ResolvedHook, + envs: &[(String, String)], + payload: Option<&str>, + ) -> bool { if INSTALLED.load(Ordering::SeqCst) == 0 { return false; } @@ -529,6 +539,8 @@ pub(crate) mod test_sink { event, hook_name: hook.name.clone(), envs: envs.iter().cloned().collect(), + payload: payload.map(str::to_string), + cwd: hook.cwd.clone(), }); true } @@ -901,7 +913,7 @@ mod tests { command: "true".to_string(), cwd: env::temp_dir(), }; - assert!(test_sink::record(HookEvent::TurnFailed, &hook, &[])); + assert!(test_sink::record(HookEvent::TurnFailed, &hook, &[], None)); let captures = test_sink::drain(); assert!(