#[cfg(target_arch = "wasm32")]
mod tokio {
pub use tokio_with_wasm::alias::*;
}
inventory::submit! {
meerkat_skills::SkillRegistration {
id: "hook-authoring",
name: "Hook Authoring",
description: "Writing hooks for the 8 hook points, execution modes, and decision semantics",
scope: meerkat_core::skills::SkillScope::Builtin,
requires_capabilities: &["hooks"],
body: include_str!("../skills/hook-authoring/SKILL.md"),
extensions: &[],
}
}
inventory::submit! {
meerkat_capabilities::CapabilityRegistration {
id: meerkat_capabilities::CapabilityId::Hooks,
description: "8 hook points, 3 runtimes (in-process/command/HTTP), observe/allow/deny semantics",
scope: meerkat_capabilities::CapabilityScope::Universal,
requires_feature: None,
prerequisites: &[],
status_resolver: None,
}
}
use chrono::Utc;
use futures::StreamExt;
use meerkat_core::config::HookInProcessRuntimeConfig;
use meerkat_core::time_compat::Duration;
use meerkat_core::{
HookCapability, HookDecision, HookEngine, HookEngineError, HookEntryConfig, HookExecutionMode,
HookExecutionReport, HookId, HookInvocation, HookOutcome, HookPatch, HookPatchEnvelope,
HookRevision, HookRunOverrides, HookRuntimeKind, HooksConfig, SessionId,
};
use serde::{Deserialize, Serialize};
use std::collections::{HashMap, HashSet};
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::sync::OnceLock;
use std::sync::atomic::{AtomicU64, Ordering};
#[cfg(not(target_arch = "wasm32"))]
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt};
#[cfg(not(target_arch = "wasm32"))]
use tokio::process::Command;
use tokio::sync::{Mutex, Semaphore};
use tokio::time::timeout;
pub use meerkat_core::config::HookInProcessHandlerId as InProcessHookHandlerId;
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
#[serde(rename_all = "snake_case")]
pub struct RuntimeHookResponse {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub decision: Option<HookDecision>,
#[serde(default)]
pub patches: Vec<HookPatch>,
}
#[derive(Debug, Clone, Deserialize)]
struct CommandRuntimeConfig {
command: String,
#[serde(default)]
args: Vec<String>,
#[serde(default)]
env: HashMap<String, String>,
}
#[derive(Debug, Clone, Deserialize)]
struct HttpRuntimeConfig {
url: String,
#[serde(default = "default_http_method")]
method: String,
#[serde(default)]
headers: HashMap<String, String>,
}
fn default_http_method() -> String {
"POST".to_string()
}
#[cfg(not(target_arch = "wasm32"))]
type HandlerFuture = Pin<Box<dyn Future<Output = Result<RuntimeHookResponse, String>> + Send>>;
#[cfg(target_arch = "wasm32")]
type HandlerFuture = Pin<Box<dyn Future<Output = Result<RuntimeHookResponse, String>>>>;
pub type InProcessHookHandler = Arc<dyn Fn(HookInvocation) -> HandlerFuture + Send + Sync>;
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct HookPatchDeliveryTarget {
session_id: SessionId,
}
impl HookPatchDeliveryTarget {
pub fn for_session(session_id: SessionId) -> Self {
Self { session_id }
}
pub fn session_id(&self) -> &SessionId {
&self.session_id
}
}
#[derive(Debug, Clone)]
struct DeliveredHookPatch {
target: HookPatchDeliveryTarget,
envelope: HookPatchEnvelope,
}
#[derive(Debug, Clone)]
struct LocalHookPatchDelivery {
queue: Arc<Mutex<Vec<DeliveredHookPatch>>>,
}
impl LocalHookPatchDelivery {
fn new() -> Self {
Self {
queue: Arc::new(Mutex::new(Vec::new())),
}
}
async fn publish(&self, target: HookPatchDeliveryTarget, patches: Vec<HookPatchEnvelope>) {
if patches.is_empty() {
return;
}
let mut guard = self.queue.lock().await;
let mut patches = patches;
guard.reserve(patches.len());
let Some(last) = patches.pop() else {
return;
};
for envelope in patches {
guard.push(DeliveredHookPatch {
target: target.clone(),
envelope,
});
}
guard.push(DeliveredHookPatch {
target,
envelope: last,
});
}
async fn drain(&self, target: &HookPatchDeliveryTarget) -> Vec<HookPatchEnvelope> {
let mut guard = self.queue.lock().await;
let queued = guard.len();
let mut remaining = Vec::with_capacity(queued);
let mut drained = Vec::new();
for published in guard.drain(..) {
if &published.target == target {
drained.push(published.envelope);
} else {
remaining.push(published);
}
}
*guard = remaining;
drained
}
}
#[derive(Clone)]
pub struct DefaultHookEngine {
config: HooksConfig,
base_entries: Arc<Vec<HookEntryConfig>>,
base_validation_error: Option<String>,
http_client: Arc<OnceLock<reqwest::Client>>,
in_process_handlers:
Arc<std::sync::RwLock<HashMap<InProcessHookHandlerId, InProcessHookHandler>>>,
background_patch_delivery: LocalHookPatchDelivery,
background_slots: Arc<Semaphore>,
revision: Arc<AtomicU64>,
}
impl DefaultHookEngine {
pub fn new(config: HooksConfig) -> Self {
let base_validation_error = config
.entries
.iter()
.find_map(|entry| Self::validate_entry(entry).err())
.map(|err| err.to_string());
let background_max_concurrency = config.background_max_concurrency.max(1);
Self {
base_entries: Arc::new(config.entries.clone()),
config,
base_validation_error,
http_client: Arc::new(OnceLock::new()),
in_process_handlers: Arc::new(std::sync::RwLock::new(HashMap::new())),
background_patch_delivery: LocalHookPatchDelivery::new(),
background_slots: Arc::new(Semaphore::new(background_max_concurrency)),
revision: Arc::new(AtomicU64::new(1)),
}
}
pub fn with_in_process_handler(
self,
name: impl Into<InProcessHookHandlerId>,
handler: InProcessHookHandler,
) -> Self {
let next = self;
let name = name.into();
match next.in_process_handlers.write() {
Ok(mut map) => {
map.insert(name, handler);
}
Err(err) => {
tracing::warn!("failed to register in-process hook handler: {}", err);
}
}
next
}
pub async fn register_in_process_handler(
&self,
name: impl Into<InProcessHookHandlerId>,
handler: InProcessHookHandler,
) {
match self.in_process_handlers.write() {
Ok(mut map) => {
map.insert(name.into(), handler);
}
Err(err) => {
tracing::warn!("failed to register in-process hook handler: {}", err);
}
}
}
fn next_revision(&self) -> HookRevision {
HookRevision(self.revision.fetch_add(1, Ordering::SeqCst))
}
#[cfg(not(target_arch = "wasm32"))]
async fn read_stream_limited<R>(
mut stream: R,
byte_limit: usize,
stream_name: &str,
) -> Result<Vec<u8>, String>
where
R: AsyncRead + Unpin,
{
let mut out = Vec::with_capacity(byte_limit.min(8 * 1024));
let mut chunk = [0u8; 8 * 1024];
loop {
let read = stream
.read(&mut chunk)
.await
.map_err(|err| format!("failed reading {stream_name}: {err}"))?;
if read == 0 {
break;
}
out.extend_from_slice(&chunk[..read]);
if out.len() > byte_limit {
return Err(format!(
"{} exceeds max size: {} > {}",
stream_name,
out.len(),
byte_limit
));
}
}
Ok(out)
}
fn effective_entries<'a>(
&'a self,
overrides: Option<&'a HookRunOverrides>,
) -> Result<std::borrow::Cow<'a, [HookEntryConfig]>, HookEngineError> {
if let Some(reason) = &self.base_validation_error {
return Err(HookEngineError::InvalidConfiguration(reason.clone()));
}
if let Some(overrides) = overrides {
if overrides.disable.is_empty() && overrides.entries.is_empty() {
return Ok(std::borrow::Cow::Borrowed(self.base_entries.as_slice()));
}
let mut entries = Vec::with_capacity(self.base_entries.len() + overrides.entries.len());
if overrides.disable.is_empty() {
entries.extend(self.base_entries.iter().cloned());
} else {
let disabled: HashSet<HookId> = overrides.disable.iter().cloned().collect();
entries.extend(
self.base_entries
.iter()
.filter(|entry| !disabled.contains(&entry.id))
.cloned(),
);
}
entries.extend(overrides.entries.clone());
for entry in &entries {
Self::validate_entry(entry)?;
}
return Ok(std::borrow::Cow::Owned(entries));
}
Ok(std::borrow::Cow::Borrowed(self.base_entries.as_slice()))
}
async fn drain_published_patches_for_session(
&self,
session_id: &SessionId,
) -> Vec<HookPatchEnvelope> {
let target = HookPatchDeliveryTarget::for_session(session_id.clone());
self.background_patch_delivery.drain(&target).await
}
async fn publish_patches(
&self,
target: HookPatchDeliveryTarget,
patches: Vec<HookPatchEnvelope>,
) {
self.background_patch_delivery
.publish(target, patches)
.await;
}
fn validate_entry(entry: &HookEntryConfig) -> Result<(), HookEngineError> {
if entry.id.0.trim().is_empty() {
return Err(HookEngineError::InvalidConfiguration(
"hook id cannot be empty".to_string(),
));
}
if entry.mode == HookExecutionMode::Background
&& entry.point.is_pre()
&& entry.capability != HookCapability::Observe
{
return Err(HookEngineError::InvalidConfiguration(format!(
"pre_* background hooks must be observe-only: {}",
entry.id
)));
}
Ok(())
}
async fn execute_one(
&self,
entry: HookEntryConfig,
registration_index: usize,
invocation: HookInvocation,
) -> Result<HookOutcome, HookEngineError> {
let start = meerkat_core::time_compat::Instant::now();
let timeout_ms = entry.timeout_ms.unwrap_or(self.config.default_timeout_ms);
let mut outcome = HookOutcome {
hook_id: entry.id.clone(),
point: entry.point,
priority: entry.priority,
registration_index,
decision: None,
patches: Vec::new(),
published_patches: Vec::new(),
error: None,
duration_ms: None,
};
let runtime_result = timeout(
Duration::from_millis(timeout_ms),
self.invoke_runtime(&entry, invocation.clone()),
)
.await;
let response = match runtime_result {
Err(_) => {
return Err(HookEngineError::Timeout {
hook_id: entry.id.clone(),
timeout_ms,
});
}
Ok(response) => response?,
};
outcome.decision = runtime_decision_with_configured_hook_id(response.decision, &entry.id);
outcome.patches = response.patches;
if entry.mode == HookExecutionMode::Background {
if entry.point.is_pre() {
if !outcome.patches.is_empty()
|| matches!(outcome.decision, Some(HookDecision::Deny { .. }))
{
outcome.error = Some("pre_* background hooks are observe-only".to_string());
}
outcome.patches.clear();
outcome.decision = None;
} else if entry.point.is_post() {
let mut envelopes = Vec::new();
for patch in outcome.patches.drain(..) {
envelopes.push(HookPatchEnvelope {
revision: self.next_revision(),
hook_id: entry.id.clone(),
point: entry.point,
patch,
published_at: Utc::now(),
});
}
outcome.published_patches = envelopes;
outcome.decision = None;
}
}
outcome.duration_ms = Some(start.elapsed().as_millis() as u64);
Ok(outcome)
}
async fn invoke_runtime(
&self,
entry: &HookEntryConfig,
invocation: HookInvocation,
) -> Result<RuntimeHookResponse, HookEngineError> {
let runtime_config =
entry
.runtime
.config_value()
.map_err(|err| HookEngineError::ExecutionFailed {
hook_id: entry.id.clone(),
reason: format!("invalid runtime config payload: {err}"),
})?;
match entry.runtime.kind {
HookRuntimeKind::InProcess => {
let cfg: HookInProcessRuntimeConfig = entry
.runtime
.in_process_config()
.map_err(|err| HookEngineError::ExecutionFailed {
hook_id: entry.id.clone(),
reason: format!("invalid in_process runtime config: {err}"),
})?
.ok_or_else(|| HookEngineError::ExecutionFailed {
hook_id: entry.id.clone(),
reason: "runtime kind is not in_process".to_string(),
})?;
let handler = {
let handlers = self.in_process_handlers.read().map_err(|err| {
HookEngineError::ExecutionFailed {
hook_id: entry.id.clone(),
reason: format!("in-process handler lock poisoned: {err}"),
}
})?;
handlers.get(&cfg.handler).cloned().ok_or_else(|| {
HookEngineError::ExecutionFailed {
hook_id: entry.id.clone(),
reason: format!("in-process handler '{}' not registered", cfg.handler),
}
})?
};
(handler)(invocation)
.await
.map_err(|reason| HookEngineError::ExecutionFailed {
hook_id: entry.id.clone(),
reason,
})
}
HookRuntimeKind::Command => {
let cfg: CommandRuntimeConfig =
serde_json::from_value(runtime_config).map_err(|err| {
HookEngineError::ExecutionFailed {
hook_id: entry.id.clone(),
reason: format!("invalid command runtime config: {err}"),
}
})?;
self.invoke_command_runtime(entry, &cfg.command, &cfg.args, &cfg.env, invocation)
.await
}
HookRuntimeKind::Http => {
let cfg: HttpRuntimeConfig =
serde_json::from_value(runtime_config).map_err(|err| {
HookEngineError::ExecutionFailed {
hook_id: entry.id.clone(),
reason: format!("invalid http runtime config: {err}"),
}
})?;
self.invoke_http_runtime(entry, &cfg.url, &cfg.method, &cfg.headers, invocation)
.await
}
}
}
#[cfg(not(target_arch = "wasm32"))]
async fn invoke_command_runtime(
&self,
entry: &HookEntryConfig,
command: &str,
args: &[String],
env: &HashMap<String, String>,
invocation: HookInvocation,
) -> Result<RuntimeHookResponse, HookEngineError> {
let mut child = Command::new(command)
.args(args)
.envs(env)
.stdin(std::process::Stdio::piped())
.stdout(std::process::Stdio::piped())
.stderr(std::process::Stdio::piped())
.spawn()
.map_err(|err| HookEngineError::ExecutionFailed {
hook_id: entry.id.clone(),
reason: format!("failed to spawn command hook: {err}"),
})?;
let payload =
serde_json::to_vec(&invocation).map_err(|err| HookEngineError::ExecutionFailed {
hook_id: entry.id.clone(),
reason: format!("failed to encode invocation payload: {err}"),
})?;
if payload.len() > self.config.payload_max_bytes {
return Err(HookEngineError::ExecutionFailed {
hook_id: entry.id.clone(),
reason: format!(
"hook payload exceeds max size: {} > {}",
payload.len(),
self.config.payload_max_bytes
),
});
}
if let Some(mut stdin) = child.stdin.take() {
stdin
.write_all(&payload)
.await
.map_err(|err| HookEngineError::ExecutionFailed {
hook_id: entry.id.clone(),
reason: format!("failed to write command hook stdin: {err}"),
})?;
}
let stdout = child
.stdout
.take()
.ok_or_else(|| HookEngineError::ExecutionFailed {
hook_id: entry.id.clone(),
reason: "command hook stdout pipe unavailable".to_string(),
})?;
let stderr = child
.stderr
.take()
.ok_or_else(|| HookEngineError::ExecutionFailed {
hook_id: entry.id.clone(),
reason: "command hook stderr pipe unavailable".to_string(),
})?;
let max_output_bytes = self.config.payload_max_bytes;
let stdout_task = tokio::spawn(Self::read_stream_limited(
stdout,
max_output_bytes,
"command stdout",
));
let stderr_task = tokio::spawn(Self::read_stream_limited(
stderr,
max_output_bytes,
"command stderr",
));
let status = child
.wait()
.await
.map_err(|err| HookEngineError::ExecutionFailed {
hook_id: entry.id.clone(),
reason: format!("failed waiting for command hook: {err}"),
})?;
let stdout = stdout_task
.await
.map_err(|err| HookEngineError::ExecutionFailed {
hook_id: entry.id.clone(),
reason: format!("command stdout task join failed: {err}"),
})?
.map_err(|reason| HookEngineError::ExecutionFailed {
hook_id: entry.id.clone(),
reason,
})?;
let stderr = stderr_task
.await
.map_err(|err| HookEngineError::ExecutionFailed {
hook_id: entry.id.clone(),
reason: format!("command stderr task join failed: {err}"),
})?
.map_err(|reason| HookEngineError::ExecutionFailed {
hook_id: entry.id.clone(),
reason,
})?;
if !status.success() {
let stderr = String::from_utf8_lossy(&stderr).to_string();
return Err(HookEngineError::ExecutionFailed {
hook_id: entry.id.clone(),
reason: format!("command hook exited with {status}: {stderr}"),
});
}
serde_json::from_slice::<RuntimeHookResponse>(&stdout).map_err(|err| {
HookEngineError::ExecutionFailed {
hook_id: entry.id.clone(),
reason: format!("invalid command hook response: {err}"),
}
})
}
#[cfg(target_arch = "wasm32")]
async fn invoke_command_runtime(
&self,
entry: &HookEntryConfig,
_command: &str,
_args: &[String],
_env: &HashMap<String, String>,
_invocation: HookInvocation,
) -> Result<RuntimeHookResponse, HookEngineError> {
Err(HookEngineError::ExecutionFailed {
hook_id: entry.id.clone(),
reason: "command hooks are not supported on wasm32".to_string(),
})
}
async fn invoke_http_runtime(
&self,
entry: &HookEntryConfig,
url: &str,
method: &str,
headers: &HashMap<String, String>,
invocation: HookInvocation,
) -> Result<RuntimeHookResponse, HookEngineError> {
let payload =
serde_json::to_vec(&invocation).map_err(|err| HookEngineError::ExecutionFailed {
hook_id: entry.id.clone(),
reason: format!("failed to encode invocation payload: {err}"),
})?;
if payload.len() > self.config.payload_max_bytes {
return Err(HookEngineError::ExecutionFailed {
hook_id: entry.id.clone(),
reason: format!(
"hook payload exceeds max size: {} > {}",
payload.len(),
self.config.payload_max_bytes
),
});
}
let method = reqwest::Method::from_bytes(method.as_bytes()).map_err(|err| {
HookEngineError::ExecutionFailed {
hook_id: entry.id.clone(),
reason: format!("invalid HTTP method '{method}': {err}"),
}
})?;
let http_client = self.http_client.get_or_init(reqwest::Client::new);
let mut req = http_client
.request(method, url)
.header("content-type", "application/json")
.body(payload);
for (name, value) in headers {
req = req.header(name, value);
}
let response = req
.send()
.await
.map_err(|err| HookEngineError::ExecutionFailed {
hook_id: entry.id.clone(),
reason: format!("HTTP hook request failed: {err}"),
})?;
let status = response.status();
let mut bytes = Vec::new();
let mut stream = response.bytes_stream();
while let Some(chunk) = stream.next().await {
let chunk = chunk.map_err(|err| HookEngineError::ExecutionFailed {
hook_id: entry.id.clone(),
reason: format!("failed reading HTTP hook response body: {err}"),
})?;
bytes.extend_from_slice(&chunk);
if bytes.len() > self.config.payload_max_bytes {
return Err(HookEngineError::ExecutionFailed {
hook_id: entry.id.clone(),
reason: format!(
"HTTP hook response exceeds max size: {} > {}",
bytes.len(),
self.config.payload_max_bytes
),
});
}
}
if !status.is_success() {
let body = String::from_utf8_lossy(&bytes);
return Err(HookEngineError::ExecutionFailed {
hook_id: entry.id.clone(),
reason: format!("HTTP hook returned {status}: {body}"),
});
}
serde_json::from_slice::<RuntimeHookResponse>(&bytes).map_err(|err| {
HookEngineError::ExecutionFailed {
hook_id: entry.id.clone(),
reason: format!("invalid HTTP hook response: {err}"),
}
})
}
}
fn runtime_decision_with_configured_hook_id(
decision: Option<HookDecision>,
hook_id: &HookId,
) -> Option<HookDecision> {
match decision {
Some(HookDecision::Deny {
reason_code,
message,
payload,
..
}) => Some(HookDecision::Deny {
hook_id: hook_id.clone(),
reason_code,
message,
payload,
}),
decision => decision,
}
}
#[cfg_attr(target_arch = "wasm32", async_trait::async_trait(?Send))]
#[cfg_attr(not(target_arch = "wasm32"), async_trait::async_trait)]
impl HookEngine for DefaultHookEngine {
fn matching_hooks(
&self,
invocation: &HookInvocation,
overrides: Option<&HookRunOverrides>,
) -> Result<Vec<HookId>, HookEngineError> {
let entries = self.effective_entries(overrides)?;
Ok(entries
.iter()
.filter(|entry| entry.enabled && entry.point == invocation.point)
.map(|entry| entry.id.clone())
.collect())
}
async fn execute(
&self,
invocation: HookInvocation,
overrides: Option<&HookRunOverrides>,
) -> Result<HookExecutionReport, HookEngineError> {
let mut foreground = Vec::new();
let mut background = Vec::new();
let entries = self.effective_entries(overrides)?;
for (registration_index, entry) in entries
.iter()
.filter(|entry| entry.enabled && entry.point == invocation.point)
.cloned()
.enumerate()
{
if entry.mode == HookExecutionMode::Background {
background.push((registration_index, entry));
} else {
foreground.push((registration_index, entry));
}
}
if foreground.is_empty() && background.is_empty() {
return Ok(HookExecutionReport::empty());
}
foreground.sort_by(|(a_idx, a_entry), (b_idx, b_entry)| {
a_entry
.priority
.cmp(&b_entry.priority)
.then_with(|| a_idx.cmp(b_idx))
});
let mut merged = HookExecutionReport::empty();
for (registration_index, entry) in foreground {
let outcome = self
.execute_one(entry, registration_index, invocation.clone())
.await?;
if let Some(decision) = outcome.decision.clone() {
match &decision {
HookDecision::Deny { .. } => {
merged.decision = Some(decision);
}
HookDecision::Allow => {
if !matches!(merged.decision, Some(HookDecision::Deny { .. })) {
merged.decision = Some(HookDecision::Allow);
}
}
}
}
merged.patches.extend(outcome.patches.clone());
let should_stop = matches!(outcome.decision, Some(HookDecision::Deny { .. }));
merged.outcomes.push(outcome);
if should_stop {
break;
}
}
if !matches!(merged.decision, Some(HookDecision::Deny { .. })) {
for (registration_index, entry) in background {
let permit = match self.background_slots.clone().try_acquire_owned() {
Ok(permit) => permit,
Err(_) => {
tracing::warn!("background hook queue full, skipping {}", entry.id);
continue;
}
};
let engine = self.clone();
let invocation_cloned = invocation.clone();
tokio::spawn(async move {
let _permit = permit;
let delivery_target =
HookPatchDeliveryTarget::for_session(invocation_cloned.session_id.clone());
let outcome = match engine
.execute_one(entry, registration_index, invocation_cloned)
.await
{
Ok(outcome) => outcome,
Err(error) => {
tracing::warn!(
error = %error,
"background hook execution failed"
);
return;
}
};
if let Some(error) = &outcome.error {
tracing::warn!(
hook_id = %outcome.hook_id,
point = ?outcome.point,
error = %error,
"background hook execution produced an error"
);
}
if !outcome.published_patches.is_empty() {
engine
.publish_patches(delivery_target, outcome.published_patches)
.await;
}
});
}
}
Ok(merged)
}
async fn drain_published_patches(
&self,
session_id: &SessionId,
) -> Result<Vec<HookPatchEnvelope>, HookEngineError> {
Ok(self.drain_published_patches_for_session(session_id).await)
}
}
#[cfg(test)]
#[allow(
clippy::expect_used,
clippy::field_reassign_with_default,
clippy::panic,
clippy::unwrap_used
)]
mod tests {
use super::*;
use meerkat_core::{
HookFailurePolicy, HookLlmRequest, HookPoint, HookReasonCode, HookRuntimeConfig,
HookRuntimeKind, SessionId,
};
use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering};
fn static_handler(response: RuntimeHookResponse) -> InProcessHookHandler {
Arc::new(move |_invocation| {
let response = response.clone();
Box::pin(async move { Ok(response) })
})
}
fn delayed_handler(delay_ms: u64, response: RuntimeHookResponse) -> InProcessHookHandler {
Arc::new(move |_invocation| {
let response = response.clone();
Box::pin(async move {
tokio::time::sleep(Duration::from_millis(delay_ms)).await;
Ok(response)
})
})
}
fn runtime_in_process(name: &str) -> HookRuntimeConfig {
HookRuntimeConfig::in_process(name).unwrap_or_default()
}
#[test]
fn in_process_handler_id_preserves_string_wire_shape() {
let id: InProcessHookHandlerId = serde_json::from_value(serde_json::json!("handler-a"))
.expect("handler id should deserialize from string");
assert_eq!(id.as_str(), "handler-a");
assert_eq!(
serde_json::to_value(&id).expect("handler id should serialize"),
serde_json::json!("handler-a")
);
}
#[tokio::test]
async fn deterministic_merge_by_priority_then_registration() {
let mut config = HooksConfig::default();
config.entries = vec![
HookEntryConfig {
id: HookId::new("hook-a"),
point: HookPoint::PreLlmRequest,
priority: 10,
runtime: runtime_in_process("a"),
capability: HookCapability::Observe,
..Default::default()
},
HookEntryConfig {
id: HookId::new("hook-b"),
point: HookPoint::PreLlmRequest,
priority: 5,
runtime: runtime_in_process("b"),
capability: HookCapability::Guardrail,
..Default::default()
},
];
let engine = DefaultHookEngine::new(config);
engine
.register_in_process_handler(
"a",
static_handler(RuntimeHookResponse {
decision: None,
patches: Vec::new(),
}),
)
.await;
engine
.register_in_process_handler(
"b",
static_handler(RuntimeHookResponse {
decision: Some(HookDecision::Allow),
patches: Vec::new(),
}),
)
.await;
let report = engine
.execute(
HookInvocation {
point: HookPoint::PreLlmRequest,
session_id: SessionId::new(),
turn_number: Some(1),
prompt_input: None,
prompt: None,
error_report: None,
error_class: None,
error: None,
llm_request: Some(HookLlmRequest {
max_tokens: 256,
temperature: None,
provider_params: None,
message_count: 1,
}),
llm_response: None,
tool_call: None,
tool_result: None,
},
None,
)
.await
.unwrap();
let hook_ids: Vec<_> = report
.outcomes
.iter()
.map(|outcome| &outcome.hook_id)
.collect();
assert_eq!(
hook_ids,
vec![&HookId::new("hook-b"), &HookId::new("hook-a")]
);
assert!(matches!(report.decision, Some(HookDecision::Allow)));
assert!(report.patches.is_empty());
}
#[tokio::test]
async fn background_hooks_are_observation_only_and_publish_no_patches() {
let mut config = HooksConfig::default();
config.entries = vec![HookEntryConfig {
id: HookId::new("hook-post-bg"),
point: HookPoint::PostToolExecution,
mode: HookExecutionMode::Background,
capability: HookCapability::Observe,
runtime: runtime_in_process("post-bg"),
..Default::default()
}];
let engine = DefaultHookEngine::new(config);
engine
.register_in_process_handler(
"post-bg",
static_handler(RuntimeHookResponse {
decision: None,
patches: Vec::new(),
}),
)
.await;
let session_id = SessionId::new();
let report = engine
.execute(
HookInvocation {
point: HookPoint::PostToolExecution,
session_id: session_id.clone(),
turn_number: Some(1),
prompt_input: None,
prompt: None,
error_report: None,
error_class: None,
error: None,
llm_request: None,
llm_response: None,
tool_call: None,
tool_result: None,
},
None,
)
.await
.unwrap();
tokio::time::sleep(Duration::from_millis(10)).await;
assert!(report.patches.is_empty());
assert!(report.published_patches.is_empty());
assert!(
engine
.drain_published_patches(&session_id)
.await
.unwrap()
.is_empty()
);
}
#[tokio::test]
async fn legacy_semantic_patch_payload_from_command_runtime_fails_closed() {
let mut config = HooksConfig::default();
config.entries = vec![HookEntryConfig {
id: HookId::new("legacy-patch-command"),
point: HookPoint::PreLlmRequest,
capability: HookCapability::Rewrite,
runtime: HookRuntimeConfig::new(
HookRuntimeKind::Command,
Some(serde_json::json!({
"command": "sh",
"args": [
"-c",
"cat >/dev/null; printf '%s' '{\"patches\":[{\"patch_type\":\"llm_request\",\"max_tokens\":1}]}'"
],
"env": {}
})),
)
.unwrap_or_default(),
..Default::default()
}];
let engine = DefaultHookEngine::new(config);
let err = engine
.execute(
HookInvocation {
point: HookPoint::PreLlmRequest,
session_id: SessionId::new(),
turn_number: Some(1),
prompt_input: None,
prompt: None,
error_report: None,
error_class: None,
error: None,
llm_request: Some(HookLlmRequest {
max_tokens: 256,
temperature: None,
provider_params: None,
message_count: 1,
}),
llm_response: None,
tool_call: None,
tool_result: None,
},
None,
)
.await
.expect_err("legacy semantic patch payloads must fail closed");
assert!(matches!(
err,
HookEngineError::ExecutionFailed { ref hook_id, .. }
if hook_id == &HookId::new("legacy-patch-command")
));
}
#[tokio::test]
async fn deterministic_merge_ignores_completion_order() {
let mut config = HooksConfig::default();
config.entries = vec![
HookEntryConfig {
id: HookId::new("slow-low-priority"),
point: HookPoint::PreLlmRequest,
priority: 100,
runtime: runtime_in_process("slow"),
capability: HookCapability::Observe,
..Default::default()
},
HookEntryConfig {
id: HookId::new("fast-high-priority"),
point: HookPoint::PreLlmRequest,
priority: 1,
runtime: runtime_in_process("fast"),
capability: HookCapability::Observe,
..Default::default()
},
];
let engine = DefaultHookEngine::new(config);
engine
.register_in_process_handler(
"slow",
delayed_handler(
100,
RuntimeHookResponse {
decision: None,
patches: Vec::new(),
},
),
)
.await;
engine
.register_in_process_handler(
"fast",
delayed_handler(
1,
RuntimeHookResponse {
decision: None,
patches: Vec::new(),
},
),
)
.await;
let report = engine
.execute(
HookInvocation {
point: HookPoint::PreLlmRequest,
session_id: SessionId::new(),
turn_number: Some(1),
prompt_input: None,
prompt: None,
error_report: None,
error_class: None,
error: None,
llm_request: Some(HookLlmRequest {
max_tokens: 256,
temperature: None,
provider_params: None,
message_count: 1,
}),
llm_response: None,
tool_call: None,
tool_result: None,
},
None,
)
.await
.unwrap();
let hook_ids: Vec<_> = report
.outcomes
.iter()
.map(|outcome| &outcome.hook_id)
.collect();
assert_eq!(
hook_ids,
vec![
&HookId::new("fast-high-priority"),
&HookId::new("slow-low-priority")
]
);
assert!(report.patches.is_empty());
}
#[tokio::test]
async fn observe_runtime_error_returns_typed_engine_failure() {
let mut config = HooksConfig::default();
config.entries = vec![HookEntryConfig {
id: HookId::new("observe-missing-handler"),
point: HookPoint::PreToolExecution,
capability: HookCapability::Observe,
runtime: runtime_in_process("missing"),
..Default::default()
}];
let engine = DefaultHookEngine::new(config);
let err = engine
.execute(
HookInvocation {
point: HookPoint::PreToolExecution,
session_id: SessionId::new(),
turn_number: Some(1),
prompt_input: None,
prompt: None,
error_report: None,
error_class: None,
error: None,
llm_request: None,
llm_response: None,
tool_call: None,
tool_result: None,
},
None,
)
.await
.expect_err("hook runtime errors must fail closed through typed engine errors");
assert!(matches!(
err,
HookEngineError::ExecutionFailed { ref hook_id, .. }
if hook_id == &HookId::new("observe-missing-handler")
));
}
#[tokio::test]
async fn rewrite_runtime_error_returns_typed_engine_failure() {
let mut config = HooksConfig::default();
config.entries = vec![HookEntryConfig {
id: HookId::new("rewrite-missing-handler"),
point: HookPoint::PreToolExecution,
capability: HookCapability::Rewrite,
runtime: runtime_in_process("missing"),
..Default::default()
}];
let engine = DefaultHookEngine::new(config);
let err = engine
.execute(
HookInvocation {
point: HookPoint::PreToolExecution,
session_id: SessionId::new(),
turn_number: Some(1),
prompt_input: None,
prompt: None,
error_report: None,
error_class: None,
error: None,
llm_request: None,
llm_response: None,
tool_call: None,
tool_result: None,
},
None,
)
.await
.expect_err("hook runtime errors must not be converted into hook-local denials");
assert!(matches!(
err,
HookEngineError::ExecutionFailed { ref hook_id, .. }
if hook_id == &HookId::new("rewrite-missing-handler")
));
}
#[tokio::test]
async fn deny_short_circuits_lower_priority_hooks() {
let mut config = HooksConfig::default();
config.entries = vec![
HookEntryConfig {
id: HookId::new("guardrail-deny"),
point: HookPoint::PreToolExecution,
priority: 1,
capability: HookCapability::Guardrail,
runtime: runtime_in_process("guardrail"),
..Default::default()
},
HookEntryConfig {
id: HookId::new("observer-late"),
point: HookPoint::PreToolExecution,
priority: 100,
capability: HookCapability::Observe,
runtime: runtime_in_process("observer"),
..Default::default()
},
];
let observed_runs = Arc::new(AtomicUsize::new(0));
let observer_runs = observed_runs.clone();
let engine = DefaultHookEngine::new(config);
engine
.register_in_process_handler(
"guardrail",
static_handler(RuntimeHookResponse {
decision: Some(HookDecision::deny(
HookId::new("guardrail-deny"),
HookReasonCode::PolicyViolation,
"deny",
None,
)),
patches: Vec::new(),
}),
)
.await;
engine
.register_in_process_handler(
"observer",
Arc::new(move |_invocation| {
let observer_runs = observer_runs.clone();
Box::pin(async move {
observer_runs.fetch_add(1, AtomicOrdering::SeqCst);
Ok(RuntimeHookResponse {
decision: Some(HookDecision::Allow),
patches: Vec::new(),
})
})
}),
)
.await;
let report = engine
.execute(
HookInvocation {
point: HookPoint::PreToolExecution,
session_id: SessionId::new(),
turn_number: Some(1),
prompt_input: None,
prompt: None,
error_report: None,
error_class: None,
error: None,
llm_request: None,
llm_response: None,
tool_call: None,
tool_result: None,
},
None,
)
.await
.unwrap();
assert!(matches!(report.decision, Some(HookDecision::Deny { .. })));
assert_eq!(report.outcomes.len(), 1);
assert_eq!(observed_runs.load(AtomicOrdering::SeqCst), 0);
}
#[tokio::test]
async fn deny_decision_uses_configured_hook_id_over_runtime_payload() {
let configured_hook_id = HookId::new("configured-guardrail");
let returned_hook_id = HookId::new("runtime-payload-identity");
let mut config = HooksConfig::default();
config.entries = vec![HookEntryConfig {
id: configured_hook_id.clone(),
point: HookPoint::PreToolExecution,
capability: HookCapability::Guardrail,
runtime: runtime_in_process("guardrail"),
..Default::default()
}];
let engine = DefaultHookEngine::new(config);
engine
.register_in_process_handler(
"guardrail",
static_handler(RuntimeHookResponse {
decision: Some(HookDecision::deny(
returned_hook_id,
HookReasonCode::PolicyViolation,
"deny",
None,
)),
patches: Vec::new(),
}),
)
.await;
let report = engine
.execute(
HookInvocation {
point: HookPoint::PreToolExecution,
session_id: SessionId::new(),
turn_number: Some(1),
prompt_input: None,
prompt: None,
error_report: None,
error_class: None,
error: None,
llm_request: None,
llm_response: None,
tool_call: None,
tool_result: None,
},
None,
)
.await
.unwrap();
match report.decision {
Some(HookDecision::Deny { hook_id, .. }) => {
assert_eq!(hook_id, configured_hook_id);
}
decision => panic!("expected deny decision, got {decision:?}"),
}
assert_eq!(report.outcomes.len(), 1);
assert_eq!(report.outcomes[0].hook_id, configured_hook_id);
match &report.outcomes[0].decision {
Some(HookDecision::Deny { hook_id, .. }) => {
assert_eq!(hook_id, &configured_hook_id);
}
decision => panic!("expected outcome deny decision, got {decision:?}"),
}
}
#[tokio::test]
async fn run_started_background_guardrail_is_rejected() {
let mut config = HooksConfig::default();
config.entries = vec![HookEntryConfig {
id: HookId::new("run-start-bg-guardrail"),
point: HookPoint::RunStarted,
mode: HookExecutionMode::Background,
capability: HookCapability::Guardrail,
runtime: runtime_in_process("guardrail"),
..Default::default()
}];
let engine = DefaultHookEngine::new(config);
let err = engine
.execute(
HookInvocation {
point: HookPoint::RunStarted,
session_id: SessionId::new(),
turn_number: Some(0),
prompt_input: None,
prompt: Some("hi".to_string()),
error_report: None,
error_class: None,
error: None,
llm_request: None,
llm_response: None,
tool_call: None,
tool_result: None,
},
None,
)
.await
.expect_err("invalid background pre hook must be rejected");
assert!(matches!(err, HookEngineError::InvalidConfiguration(_)));
}
#[tokio::test]
async fn fail_open_override_cannot_convert_runtime_error_to_success() {
let mut config = HooksConfig::default();
config.entries = vec![HookEntryConfig {
id: HookId::new("rewrite-explicit-open"),
point: HookPoint::PreToolExecution,
capability: HookCapability::Rewrite,
failure_policy: Some(HookFailurePolicy::FailOpen),
runtime: runtime_in_process("missing"),
..Default::default()
}];
let engine = DefaultHookEngine::new(config);
let err = engine
.execute(
HookInvocation {
point: HookPoint::PreToolExecution,
session_id: SessionId::new(),
turn_number: Some(1),
prompt_input: None,
prompt: None,
error_report: None,
error_class: None,
error: None,
llm_request: None,
llm_response: None,
tool_call: None,
tool_result: None,
},
None,
)
.await
.expect_err("legacy fail_open override must not make runtime errors warning-only");
assert!(matches!(
err,
HookEngineError::ExecutionFailed { ref hook_id, .. }
if hook_id == &HookId::new("rewrite-explicit-open")
));
}
#[tokio::test]
async fn background_failure_policy_is_inert_compatibility_config() {
let mut config = HooksConfig::default();
config.entries = vec![HookEntryConfig {
id: HookId::new("invalid-background-fail-closed"),
point: HookPoint::PostToolExecution,
mode: HookExecutionMode::Background,
capability: HookCapability::Observe,
failure_policy: Some(HookFailurePolicy::FailClosed),
runtime: runtime_in_process("bg"),
..Default::default()
}];
let engine = DefaultHookEngine::new(config);
let report = engine
.execute(
HookInvocation {
point: HookPoint::PostToolExecution,
session_id: SessionId::new(),
turn_number: Some(1),
prompt_input: None,
prompt: None,
error_report: None,
error_class: None,
error: None,
llm_request: None,
llm_response: None,
tool_call: None,
tool_result: None,
},
None,
)
.await
.expect("legacy failure_policy must not be an admission authority");
assert!(report.decision.is_none());
}
#[tokio::test]
async fn command_runtime_hook_executes() {
let mut config = HooksConfig::default();
config.entries = vec![HookEntryConfig {
id: HookId::new("command-hook"),
point: HookPoint::PreToolExecution,
runtime: HookRuntimeConfig::new(
HookRuntimeKind::Command,
Some(serde_json::json!({
"command": "sh",
"args": ["-c", "cat >/dev/null; printf '{\"patches\":[]}'"],
"env": {}
})),
)
.unwrap_or_default(),
..Default::default()
}];
let engine = DefaultHookEngine::new(config);
let report = engine
.execute(
HookInvocation {
point: HookPoint::PreToolExecution,
session_id: SessionId::new(),
turn_number: Some(1),
prompt_input: None,
prompt: None,
error_report: None,
error_class: None,
error: None,
llm_request: None,
llm_response: None,
tool_call: None,
tool_result: None,
},
None,
)
.await
.unwrap();
assert_eq!(report.outcomes.len(), 1);
assert!(
report.outcomes[0].error.is_none(),
"command runtime error: {:?}",
report.outcomes[0].error
);
}
#[tokio::test]
#[ignore = "integration-real: binds TCP port and makes real HTTP request"]
async fn http_runtime_hook_executes() {
use tokio::io::{AsyncReadExt, AsyncWriteExt};
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move {
loop {
let Ok((mut socket, _)) = listener.accept().await else {
break;
};
tokio::spawn(async move {
let mut buf = vec![0_u8; 4096];
let _ = socket.read(&mut buf).await;
let body = r#"{"patches":[]}"#;
let response = format!(
"HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{}",
body.len(),
body
);
let _ = socket.write_all(response.as_bytes()).await;
});
}
});
let mut config = HooksConfig::default();
config.default_timeout_ms = 30_000;
config.entries = vec![HookEntryConfig {
id: HookId::new("http-hook"),
point: HookPoint::PreToolExecution,
runtime: HookRuntimeConfig::new(
HookRuntimeKind::Http,
Some(serde_json::json!({
"url": format!("http://{}/hook", addr),
"method": "POST",
"headers": {}
})),
)
.unwrap_or_default(),
..Default::default()
}];
let engine = DefaultHookEngine::new(config);
let report = engine
.execute(
HookInvocation {
point: HookPoint::PreToolExecution,
session_id: SessionId::new(),
turn_number: Some(1),
prompt_input: None,
prompt: None,
error_report: None,
error_class: None,
error: None,
llm_request: None,
llm_response: None,
tool_call: None,
tool_result: None,
},
None,
)
.await
.unwrap();
assert_eq!(report.outcomes.len(), 1);
assert!(
report.outcomes[0].error.is_none(),
"http runtime error: {:?}",
report.outcomes[0].error
);
}
}