feat(TASK-036): fire tool, LLM, and job hook events at their seams

Wire the hook engine into the three runtime seams:

- tool.*: ToolCall::eval fires tool.started once the name and arguments
  are resolved, then exactly one of tool.completed/tool.failed,
  discriminated on both the Err path and tool_call_error-shaped outputs.
  The arguments JSON is handed to the engine as the payload; eval itself
  performs no payload-file I/O.
- llm.request.*: the three transport free functions
  (call_chat_completions, call_chat_completions_streaming, and the quiet
  graph-node variant) fire started/completed/failed for every provider
  and caller, classifying failures as aborted, api_error (with
  COYOTE_LLM_STATUS), or other.
- job.*: handle_job_tool fires job.started before either spawn and the
  terminal event inside the detached closures via a pre-resolved
  fire_resolved snapshot, so job__start's return is never delayed.

The test sink now captures the payload and cwd of each dispatch
(sink-only change), and seam tests pin event cadence, env contracts, and
byte-identical tool results and errors.
This commit is contained in:
Dark-Alex-17 committed 2026-09-16 13:44:35 -06:00
1 parent 0c2a68ecc4
commit 2731e98ed1
4 files changed
+1105 -11

No files matched your search

+501 -7
View File
@@ -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::<ApiStatusError>()) {
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<ToolResult>)> {
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<ChatCompletionsOutput> {
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::Capture> {
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::<ApiStatusError>().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::<ApiStatusError>().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")
);
}
}